git.iko.soy/streamly
on d2aaef0 help sign in
streamly/src/Streamly/Internal/Data/Stream/ZipAsync.hs
133 lines 4.7 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.ZipAsync
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
9-- Portability : GHC
10--
11-- To run examples in this module:
12--
13-- >>> import qualified Streamly.Prelude as Stream
14--
15module Streamly.Internal.Data.Stream.ZipAsync
16 ( ZipAsyncM(..)
17 , ZipAsync
18 , consMZipAsync
19 , zipAsyncWithK
20 , zipAsyncWithMK
21 )
22where
24#if __GLASGOW_HASKELL__ < 808
25import Data.Semigroup (Semigroup(..))
26#endif
27import Streamly.Internal.Control.Concurrent (MonadAsync)
28import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
30import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
31import qualified Streamly.Internal.Data.Stream.StreamK as K
32import qualified Streamly.Internal.Data.Stream.StreamD as D
33import qualified Streamly.Internal.Data.Stream.Serial as Serial
34import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
35import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
36import qualified Streamly.Internal.Data.Stream.Type as Stream
37import Streamly.Internal.Data.SVar
39import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
41#include "Instances.hs"
43-- $setup
44-- >>> import qualified Streamly.Prelude as Stream
45-- >>> import Control.Concurrent (threadDelay)
46-- >>> :{
47-- delay n = do
48-- threadDelay (n * 1000000) -- sleep for n seconds
49-- putStrLn (show n ++ " sec") -- print "n sec"
50-- return n -- IO Int
51-- :}
53------------------------------------------------------------------------------
54-- Parallel Zipping
55------------------------------------------------------------------------------
57-- | Like 'zipAsyncWith' but with a monadic zipping function.
58--
59-- @since 0.4.0
60{-# INLINE zipAsyncWithMK #-}
61zipAsyncWithMK :: MonadAsync m
62 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
63zipAsyncWithMK f m1 m2 = K.mkStream $ \st yld sng stp -> do
64 sv <- newParallelVar StopNone (adaptState st)
65 SVar.toSVarParallel (adaptState st) sv $ D.fromStreamK m2
66 K.foldStream st yld sng stp
67 $ K.zipWithM f m1 (Stream.toStreamK (SVar.fromSVar sv))
69-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
70-- along with the change of behvaior to end the stream concurrently.
71--
72-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
73-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
74-- style. The maximum number of elements of each stream evaluated in advance
75-- can be controlled by 'maxBuffer'.
76--
77-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
78-- ends while we are still evaluating stream @a@ and waiting for a result then
79-- stream will not end until after the evaluation of stream @a@ finishes. This
80-- behavior can potentially be changed in future to end the stream immediately
81-- as soon as any of the stream end is detected.
82--
83-- @since 0.1.0
84{-# INLINE zipAsyncWithK #-}
85zipAsyncWithK :: MonadAsync m
86 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
87zipAsyncWithK f = zipAsyncWithMK (\a b -> return (f a b))
89------------------------------------------------------------------------------
90-- Parallely Zipping Streams
91------------------------------------------------------------------------------
92--
93-- | For 'ZipAsyncM' streams:
94--
95-- @
96-- (<>) = 'Streamly.Prelude.serial'
97-- (<*>) = 'Streamly.Prelude.serial.zipAsyncWith' id
98-- @
99--
100-- Applicative evaluates the streams being zipped concurrently, the following
101-- would take half the time that it would take in serial zipping:
102--
103-- >>> s = Stream.fromFoldableM $ Prelude.map delay [1, 1, 1]
104-- >>> Stream.toList $ Stream.fromZipAsync $ (,) <$> s <*> s
105-- ...
106-- [(1,1),(1,1),(1,1)]
107--
108-- /Since: 0.2.0 ("Streamly")/
109--
110-- @since 0.8.0
111newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
112 deriving (Semigroup, Monoid)
114-- | An IO stream whose applicative instance zips streams wAsyncly.
115--
116-- /Since: 0.2.0 ("Streamly")/
117--
118-- @since 0.8.0
119type ZipAsync = ZipAsyncM IO
121consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
122consMZipAsync m (ZipAsyncM r) = ZipAsyncM $ K.consM m r
124instance Monad m => Functor (ZipAsyncM m) where
125 {-# INLINE fmap #-}
126 fmap f (ZipAsyncM m) =
127 ZipAsyncM $ Stream.toStreamK $ fmap f (Stream.fromStreamK m)
129instance MonadAsync m => Applicative (ZipAsyncM m) where
130 pure = ZipAsyncM . Stream.toStreamK . Serial.repeat
132 {-# INLINE (<*>) #-}
133 ZipAsyncM m1 <*> ZipAsyncM m2 = ZipAsyncM $ zipAsyncWithK id m1 m2