@@ -192,17 +192,40 @@ async m = do
192192-- StreamT
193193------------------------------------------------------------------------------
194194
195−-- | The 'Monad' instance of 'StreamT' combines two streams serially in the
196−-- most straightforward manner, just like the nested 'for' loops in imperative
197−-- paradigm. It moves to the next element in the first stream only after the
198−-- first element has been combined with every element in the second stream. In
199−-- other words, 'StreamT' nests loops in a depth first manner.
195+-- | The 'Monad' instance of 'StreamT' runs the /monadic continuation/ for each
196+-- element of the stream, serially.
197+--
198+-- @
199+-- main = 'runStreamT' $ do
200+-- x <- return 1 \<\> return 2
201+-- liftIO $ print x
202+-- @
203+-- @
204+-- "1"
205+-- "2"
206+-- @
207+--
208+-- 'StreamT' nests streams serially in a depth first manner.
209+--
210+-- @
211+-- main = 'runStreamT' $ do
212+-- x <- return 1 \<\> return 2
213+-- y <- return 3 \<\> return 4
214+-- liftIO $ print (x, y)
215+-- @
216+-- @
217+-- (1,3)
218+-- (1,4)
219+-- (2,3)
220+-- (2,4)
221+-- @
222+--
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.
200228--
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
204−-- stream.
205−
206229newtype StreamT m a = StreamT {getStreamT :: Stream m a}
207230 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
208231
@@ -302,13 +325,23 @@ instance (Monad m, Floating a) => Floating (StreamT m a) where
302325-- InterleavedT
303326------------------------------------------------------------------------------
304327
305−-- | The 'Monad' instance of 'InterleavedT' fairly interleaves the iterations
306−-- of the inner and the outer loop, nesting loops in a breadth first manner.
328+-- | Like 'StreamT' but different in nesting behavior. It fairly interleaves
329+-- the iterations of the inner and the outer loop, nesting loops in a breadth
330+-- first manner.
307331--
308−-- Note that 'InterleavedT' style composition cannot work well operationally on
309−-- infinite streams. Since the next iteration of the outer loop is started even
310−-- before the first one has finsihed, it will keep accumulating space resulting
311−-- in a continuous increase in space consumption.
332+--
333+-- @
334+-- main = 'runInterleavedT' $ do
335+-- x <- return 1 \<\> return 2
336+-- y <- return 3 \<\> return 4
337+-- liftIO $ print (x, y)
338+-- @
339+-- @
340+-- (1,3)
341+-- (2,3)
342+-- (1,4)
343+-- (2,4)
344+-- @
312345--
313346newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
314347 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
@@ -401,16 +434,27 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where
401434-- AsyncT
402435------------------------------------------------------------------------------
403436
404−-- | 'AsyncT' is the concurrent equivalent of 'StreamT'. The type 'AsyncT'
405−-- follows the same composition scheme as 'StreamT' except that it can run
406−-- multiple iterations concurrently. More concurrent iterations are started
407−-- only if the previous iterations are not able to produce enough output for
408−-- the consumer.
437+-- | Like 'StreamT' but /may/ run each iteration concurrently using demand
438+-- driven concurrency. More concurrent iterations are started only if the
439+-- previous iterations are not able to produce enough output for the consumer.
409440--
410−-- Just like 'StreamT', 'AsyncT' style composition works well operationally
411−-- for infinite streams. However, if the second stream is infinite we will
412−-- never be able to start the second iteration of the first stream.
413−
441+-- @
442+-- import "Streamly"
443+-- import Control.Concurrent
444+--
445+-- main = 'runAsyncT' $ do
446+-- n <- return 3 \<\> return 2 \<\> return 1
447+-- liftIO $ do
448+-- threadDelay (n * 1000000)
449+-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
450+-- @
451+-- @
452+-- ThreadId 40: Delay 1
453+-- ThreadId 39: Delay 2
454+-- ThreadId 38: Delay 3
455+-- @
456+--
457+-- All iterations may run in the same thread if they do not block.
414458newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
415459 deriving (Semigroup, Monoid, MonadTrans)
416460
@@ -514,17 +558,27 @@ instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where
514558-- ParallelT
515559------------------------------------------------------------------------------
516560
517−-- | 'ParallelT' is the concurrent equivalent of 'InterleavedT'. It follows the
518−-- same composition scheme as 'InterleavedT' i.e. fairly interleaving the inner
519−-- and outer loops in a round robin fashion except that it can run multiple
520−-- iterations concurrently.
561+-- | Like 'StreamT' but runs /all/ iterations fairly concurrently using a round
562+-- robin scheduling.
563+--
564+-- @
565+-- import "Streamly"
566+-- import Control.Concurrent
521567--
522−-- Note that, just like 'InterleavedT', 'ParallelT' style composition cannot
523−-- work well operationally on infinite streams. Because earlier iterations do
524−-- not finish before the new ones are started, it will keep on accumulating the
525−-- elements of the stream resulting in continuous increase in space
526−-- consumption.
568+-- main = 'runParallelT' $ do
569+-- n <- return 3 \<\> return 2 \<\> return 1
570+-- liftIO $ do
571+-- threadDelay (n * 1000000)
572+-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
573+-- @
574+-- @
575+-- ThreadId 40: Delay 1
576+-- ThreadId 39: Delay 2
577+-- ThreadId 38: Delay 3
578+-- @
527579--
580+-- Unlike 'AsyncT' all iterations are guaranteed to run fairly concurrently,
581+-- unconditionally.
528582newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
529583 deriving (Semigroup, Monoid, MonadTrans)
530584
@@ -627,8 +681,22 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
627681 yield1 a (Just ra) = merge a ra
628682 (runStream mx) Nothing stp yield1
629683
630−-- | 'ZipStream' zips serially i.e. it produces one element from each stream in
631−-- a serial manner and then zips the two elements.
684+-- | 'ZipStream' zips serially i.e. it produces one element from each stream
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
687+-- annotation.
688+--
689+-- @
690+-- main = (toList . 'zipping' $ (,) \<$\> s1 \<*\> s2) >>= print
691+-- where s1 = pure 1 <> pure 2
692+-- s2 = pure 3 <> pure 4
693+-- @
694+-- @
695+-- [(1,3),(2,4)]
696+-- @
697+--
698+-- This applicative operation can be seen as the zipping equivalent of
699+-- interleaving with '<=>'.
632700newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
633701 deriving (Semigroup, Monoid)
634702
@@ -702,8 +770,20 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
702770 mb <- async m2
703771 (runStream (toStream (zipWith f ma mb))) Nothing stp yld
704772
705−-- | 'ZipAsync' zips in parallel, it produces one element from each stream
706−-- concurrently and then zips the two.
773+-- | Like 'ZipStream' but zips in parallel, it generates both the elements to
774+-- be zipped concurrently.
775+--
776+-- @
777+-- main = (toList . 'zippingAsync' $ (,) \<$\> s1 \<*\> s2) >>= print
778+-- where s1 = pure 1 <> pure 2
779+-- s2 = pure 3 <> pure 4
780+-- @
781+-- @
782+-- [(1,3),(2,4)]
783+-- @
784+--
785+-- This applicative operation can be seen as the zipping equivalent of
786+-- parallel composition with '<|>'.
707787newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
708788 deriving (Semigroup, Monoid)
709789
@@ -800,27 +880,27 @@ zippingAsync x = x
800880-- Running Streams, convenience functions specialized to types
801881-------------------------------------------------------------------------------
802882
803−-- | Run a 'StreamT' computation. Same as @runStreaming . serially@.
883+-- | Same as @runStreaming . serially@.
804884runStreamT :: Monad m => StreamT m a -> m ()
805885runStreamT = runStreaming
806886
807−-- | Run an 'InterleavedT' computation. Same as @runStreaming . interleaving@.
887+-- | Same as @runStreaming . interleaving@.
808888runInterleavedT :: Monad m => InterleavedT m a -> m ()
809889runInterleavedT = runStreaming
810890
811−-- | Run an 'AsyncT' computation. Same as @runStreaming . asyncly@.
891+-- | Same as @runStreaming . asyncly@.
812892runAsyncT :: Monad m => AsyncT m a -> m ()
813893runAsyncT = runStreaming
814894
815−-- | Run a 'ParallelT' computation. Same as @runStreaming . parallely@.
895+-- | Same as @runStreaming . parallely@.
816896runParallelT :: Monad m => ParallelT m a -> m ()
817897runParallelT = runStreaming
818898
819−-- | Run a 'ZipStream' computation. Same as @runStreaming . zipping@.
899+-- | Same as @runStreaming . zipping@.
820900runZipStream :: Monad m => ZipStream m a -> m ()
821901runZipStream = runStreaming
822902
823−-- | Run a 'ZipAsync' computation. Same as @runStreaming . zippingAsync@.
903+-- | Same as @runStreaming . zippingAsync@.
824904runZipAsync :: Monad m => ZipAsync m a -> m ()
825905runZipAsync = runStreaming
826906
@@ -831,21 +911,40 @@ runZipAsync = runStreaming
831911infixr 5 <=>
832912
833913-- | Sequential interleaved composition, in contrast to '<>' this operator
834−-- fairly interleaves the two streams instead of appending them; yielding one
835−-- element from each stream alternately. Unlike '<>' it cannot be used to fold
836−-- an infinite container of streams. This operator corresponds to the
837−-- 'InterleavedT' type for product style composition.
914+-- fairly interleaves two streams instead of appending them; yielding one
915+-- element from each stream alternately.
916+--
917+-- @
918+-- main = ('toList' . 'serially' $ (return 1 <> return 2) \<=\> (return 3 <> return 4)) >>= print
919+-- @
920+-- @
921+-- [1,3,2,4]
922+-- @
923+--
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.
838928{-# INLINE (<=>) #-}
839929(<=>) :: Streaming t => t m a -> t m a -> t m a
840930m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
841931
842−-- | Parallel interleaved composition, in contrast to '<|>' this operator
843−-- "merges" streams in a left biased manner rather than fairly interleaving
844−-- them. It keeps yielding from the stream on the left as long as it can. If
845−-- the left stream blocks or cannot keep up with the pace of the consumer it
846−-- can yield from the stream on the right in parallel. Unlike '<|>' it can be
847−-- used to fold infinite containers of streams. This operator corresponds to
848−-- the 'AsyncT' type for product style composition.
932+-- | Demand driven concurrent composition. In contrast to '<|>' this operator
933+-- concurrently "merges" streams in a left biased manner rather than fairly
934+-- interleaving them. It keeps yielding from the stream on the left as long as
935+-- it can. If the left stream blocks or cannot keep up with the pace of the
936+-- consumer it can concurrently yield from the stream on the right in parallel.
937+--
938+-- @
939+-- main = ('toList' . 'serially' $ (return 1 <> return 2) \<| (return 3 <> return 4)) >>= print
940+-- @
941+-- @
942+-- [1,2,3,4]
943+-- @
944+--
945+-- Unlike '<|>' it can be used to fold infinite containers of streams. This
946+-- operator corresponds to the 'AsyncT' type for product style composition.
947+--
849948{-# INLINE (<|) #-}
850949(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
851950m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)