@@ -192,17 +192,40 @@ async m = do
192-- StreamT
192-- StreamT
193------------------------------------------------------------------------------
193------------------------------------------------------------------------------
195−-- | The 'Monad' instance of 'StreamT' combines two streams serially in the
195+-- | The 'Monad' instance of 'StreamT' runs the /monadic continuation/ for each
196−-- most straightforward manner, just like the nested 'for' loops in imperative
196+-- element of the stream, serially.
197−-- paradigm. It moves to the next element in the first stream only after the
197+--
198−-- first element has been combined with every element in the second stream. In
198+-- @
199−-- other words, 'StreamT' nests loops in a depth first manner.
199+-- main = 'runStreamT' $ do
200+-- x <- return 1 \<\> return 2
208+-- 'StreamT' nests streams serially in a depth first manner.
211+-- main = 'runStreamT' $ do
212+-- x <- return 1 \<\> return 2
213+-- y <- return 3 \<\> return 4
214+-- liftIO $ print (x, y)
223+-- This behavior is exactly like a list transformer. We call the monadic code
224+-- being run for each element of the stream a monadic continuation. In
225+-- imperative paradigm we can think of this composition as nested @for@ loops
226+-- and the monadic continuation is the body of the loop. The loop iterates for
227+-- all elements of the stream.
201−-- Note that 'StreamT' style composition works well operationally even when the
202−-- streams being composed are infinite. However, if the second stream is
203−-- infinite we will never come back to the second iteration of the first
206newtype StreamT m a = StreamT {getStreamT :: Stream m a}
229newtype StreamT m a = StreamT {getStreamT :: Stream m a}
207 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
230 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
@@ -302,13 +325,23 @@ instance (Monad m, Floating a) => Floating (StreamT m a) where
302-- InterleavedT
325-- InterleavedT
303------------------------------------------------------------------------------
326------------------------------------------------------------------------------
305−-- | The 'Monad' instance of 'InterleavedT' fairly interleaves the iterations
328+-- | Like 'StreamT' but different in nesting behavior. It fairly interleaves
306−-- of the inner and the outer loop, nesting loops in a breadth first manner.
329+-- the iterations of the inner and the outer loop, nesting loops in a breadth
308−-- Note that 'InterleavedT' style composition cannot work well operationally on
332+--
309−-- infinite streams. Since the next iteration of the outer loop is started even
333+-- @
310−-- before the first one has finsihed, it will keep accumulating space resulting
334+-- main = 'runInterleavedT' $ do
311−-- in a continuous increase in space consumption.
335+-- x <- return 1 \<\> return 2
336+-- y <- return 3 \<\> return 4
337+-- liftIO $ print (x, y)
313newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
346newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
314 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
347 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
@@ -401,16 +434,27 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where
402------------------------------------------------------------------------------
435------------------------------------------------------------------------------
404−-- | 'AsyncT' is the concurrent equivalent of 'StreamT'. The type 'AsyncT'
437+-- | Like 'StreamT' but /may/ run each iteration concurrently using demand
405−-- follows the same composition scheme as 'StreamT' except that it can run
406−-- multiple iterations concurrently. More concurrent iterations are started
438+-- driven concurrency. More concurrent iterations are started only if the
407−-- only if the previous iterations are not able to produce enough output for
439+-- previous iterations are not able to produce enough output for the consumer.
410−-- Just like 'StreamT', 'AsyncT' style composition works well operationally
441+-- @
411−-- for infinite streams. However, if the second stream is infinite we will
442+-- import "Streamly"
412−-- never be able to start the second iteration of the first stream.
443+-- import Control.Concurrent
445+-- main = 'runAsyncT' $ do
446+-- n <- return 3 \<\> return 2 \<\> return 1
448+-- threadDelay (n * 1000000)
449+-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
452+-- ThreadId 40: Delay 1
453+-- ThreadId 39: Delay 2
454+-- ThreadId 38: Delay 3
457+-- All iterations may run in the same thread if they do not block.
414newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
458newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
415 deriving (Semigroup, Monoid, MonadTrans)
459 deriving (Semigroup, Monoid, MonadTrans)
@@ -514,17 +558,27 @@ instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where
514-- ParallelT
558-- ParallelT
515------------------------------------------------------------------------------
559------------------------------------------------------------------------------
517−-- | 'ParallelT' is the concurrent equivalent of 'InterleavedT'. It follows the
561+-- | Like 'StreamT' but runs /all/ iterations fairly concurrently using a round
518−-- same composition scheme as 'InterleavedT' i.e. fairly interleaving the inner
562+-- robin scheduling.
519−-- and outer loops in a round robin fashion except that it can run multiple
563+--
520−-- iterations concurrently.
564+-- @
566+-- import Control.Concurrent
522−-- Note that, just like 'InterleavedT', 'ParallelT' style composition cannot
568+-- main = 'runParallelT' $ do
523−-- work well operationally on infinite streams. Because earlier iterations do
569+-- n <- return 3 \<\> return 2 \<\> return 1
524−-- not finish before the new ones are started, it will keep on accumulating the
570+-- liftIO $ do
525−-- elements of the stream resulting in continuous increase in space
571+-- threadDelay (n * 1000000)
526−-- consumption.
572+-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
575+-- ThreadId 40: Delay 1
576+-- ThreadId 39: Delay 2
577+-- ThreadId 38: Delay 3
580+-- Unlike 'AsyncT' all iterations are guaranteed to run fairly concurrently,
528newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
582newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
529 deriving (Semigroup, Monoid, MonadTrans)
583 deriving (Semigroup, Monoid, MonadTrans)
@@ -627,8 +681,22 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
627 yield1 a (Just ra) = merge a ra
681 yield1 a (Just ra) = merge a ra
628 (runStream mx) Nothing stp yield1
682 (runStream mx) Nothing stp yield1
630−-- | 'ZipStream' zips serially i.e. it produces one element from each stream in
684+-- | 'ZipStream' zips serially i.e. it produces one element from each stream
631−-- a serial manner and then zips the two elements.
685+-- serially and then zips the two elements. Note, for convenience we have used
686+-- the 'zipping' combinator in the following example instead of using a type
690+-- main = (toList . 'zipping' $ (,) \<$\> s1 \<*\> s2) >>= print
691+-- where s1 = pure 1 <> pure 2
692+-- s2 = pure 3 <> pure 4
698+-- This applicative operation can be seen as the zipping equivalent of
699+-- interleaving with '<=>'.
632newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
700newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
633 deriving (Semigroup, Monoid)
701 deriving (Semigroup, Monoid)
@@ -702,8 +770,20 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
702 mb <- async m2
770 mb <- async m2
703 (runStream (toStream (zipWith f ma mb))) Nothing stp yld
771 (runStream (toStream (zipWith f ma mb))) Nothing stp yld
705−-- | 'ZipAsync' zips in parallel, it produces one element from each stream
773+-- | Like 'ZipStream' but zips in parallel, it generates both the elements to
706−-- concurrently and then zips the two.
774+-- be zipped concurrently.
777+-- main = (toList . 'zippingAsync' $ (,) \<$\> s1 \<*\> s2) >>= print
778+-- where s1 = pure 1 <> pure 2
779+-- s2 = pure 3 <> pure 4
785+-- This applicative operation can be seen as the zipping equivalent of
786+-- parallel composition with '<|>'.
707newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
787newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
708 deriving (Semigroup, Monoid)
788 deriving (Semigroup, Monoid)
@@ -800,27 +880,27 @@ zippingAsync x = x
800-- Running Streams, convenience functions specialized to types
880-- Running Streams, convenience functions specialized to types
801-------------------------------------------------------------------------------
881-------------------------------------------------------------------------------
803−-- | Run a 'StreamT' computation. Same as @runStreaming . serially@.
883+-- | Same as @runStreaming . serially@.
804runStreamT :: Monad m => StreamT m a -> m ()
884runStreamT :: Monad m => StreamT m a -> m ()
805runStreamT = runStreaming
885runStreamT = runStreaming
807−-- | Run an 'InterleavedT' computation. Same as @runStreaming . interleaving@.
887+-- | Same as @runStreaming . interleaving@.
808runInterleavedT :: Monad m => InterleavedT m a -> m ()
888runInterleavedT :: Monad m => InterleavedT m a -> m ()
809runInterleavedT = runStreaming
889runInterleavedT = runStreaming
811−-- | Run an 'AsyncT' computation. Same as @runStreaming . asyncly@.
891+-- | Same as @runStreaming . asyncly@.
812runAsyncT :: Monad m => AsyncT m a -> m ()
892runAsyncT :: Monad m => AsyncT m a -> m ()
813runAsyncT = runStreaming
893runAsyncT = runStreaming
815−-- | Run a 'ParallelT' computation. Same as @runStreaming . parallely@.
895+-- | Same as @runStreaming . parallely@.
816runParallelT :: Monad m => ParallelT m a -> m ()
896runParallelT :: Monad m => ParallelT m a -> m ()
817runParallelT = runStreaming
897runParallelT = runStreaming
819−-- | Run a 'ZipStream' computation. Same as @runStreaming . zipping@.
899+-- | Same as @runStreaming . zipping@.
820runZipStream :: Monad m => ZipStream m a -> m ()
900runZipStream :: Monad m => ZipStream m a -> m ()
821runZipStream = runStreaming
901runZipStream = runStreaming
823−-- | Run a 'ZipAsync' computation. Same as @runStreaming . zippingAsync@.
903+-- | Same as @runStreaming . zippingAsync@.
824runZipAsync :: Monad m => ZipAsync m a -> m ()
904runZipAsync :: Monad m => ZipAsync m a -> m ()
825runZipAsync = runStreaming
905runZipAsync = runStreaming
@@ -831,21 +911,40 @@ runZipAsync = runStreaming
831infixr 5 <=>
911infixr 5 <=>
833-- | Sequential interleaved composition, in contrast to '<>' this operator
913-- | Sequential interleaved composition, in contrast to '<>' this operator
834−-- fairly interleaves the two streams instead of appending them; yielding one
914+-- fairly interleaves two streams instead of appending them; yielding one
835−-- element from each stream alternately. Unlike '<>' it cannot be used to fold
915+-- element from each stream alternately.
836−-- an infinite container of streams. This operator corresponds to the
916+--
837−-- 'InterleavedT' type for product style composition.
917+-- @
918+-- main = ('toList' . 'serially' $ (return 1 <> return 2) \<=\> (return 3 <> return 4)) >>= print
924+-- This operator corresponds to the 'InterleavedT' style. Unlike '<>', this
925+-- operator cannot be used to fold infinite containers since that might
926+-- accumulate too many partially drained streams. To be clear, it can combine
927+-- infinite streams but not infinite number of streams.
838{-# INLINE (<=>) #-}
928{-# INLINE (<=>) #-}
839(<=>) :: Streaming t => t m a -> t m a -> t m a
929(<=>) :: Streaming t => t m a -> t m a -> t m a
840m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
930m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
842−-- | Parallel interleaved composition, in contrast to '<|>' this operator
932+-- | Demand driven concurrent composition. In contrast to '<|>' this operator
843−-- "merges" streams in a left biased manner rather than fairly interleaving
933+-- concurrently "merges" streams in a left biased manner rather than fairly
844−-- them. It keeps yielding from the stream on the left as long as it can. If
934+-- interleaving them. It keeps yielding from the stream on the left as long as
845−-- the left stream blocks or cannot keep up with the pace of the consumer it
935+-- it can. If the left stream blocks or cannot keep up with the pace of the
846−-- can yield from the stream on the right in parallel. Unlike '<|>' it can be
936+-- consumer it can concurrently yield from the stream on the right in parallel.
939+-- main = ('toList' . 'serially' $ (return 1 <> return 2) \<| (return 3 <> return 4)) >>= print
847−-- used to fold infinite containers of streams. This operator corresponds to
945+-- Unlike '<|>' it can be used to fold infinite containers of streams. This
848−-- the 'AsyncT' type for product style composition.
946+-- operator corresponds to the 'AsyncT' type for product style composition.
849{-# INLINE (<|) #-}
948{-# INLINE (<|) #-}
850(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
949(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
851m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
950m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)