@@ -634,9 +634,55 @@ joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
634634
635635infixr 6 `async`
636636
637−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
638−-- Merges two streams possibly concurrently, preferring the
639−-- elements from the left one when available.
637+-- | Merges two streams, both the streams may be evaluated concurrently,
638+-- outputs from both are used as they arrive:
639+--
640+-- >>> stream1 = Stream.yieldM (delay 4)
641+-- >>> stream2 = Stream.yieldM (delay 2)
642+-- >>> Stream.toList $ stream1 `async` stream2
643+-- [4,2]
644+--
645+-- Multiple streams can be combined. With enough threads, all of them can be
646+-- scheduled simultaneously:
647+--
648+-- >>> stream3 = Stream.yieldM (delay 1)
649+-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
650+-- [1,2,4]
651+--
652+-- With 2 threads, only two can be scheduled at a time, when one of those
653+-- finishes, the third one gets scheduled:
654+--
655+-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
656+-- [2,1,4]
657+--
658+-- With a single thread, it becomes serial:
659+--
660+-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
661+-- [4,2,1]
662+--
663+-- Only streams are scheduled for async evaluation, how actions within a
664+-- stream are evaluated depends on the stream type. If it is a concurrent
665+-- stream they will be evaluated concurrently.
666+--
667+-- In the following example, both the streams are scheduled for concurrent
668+-- evaluation but each individual stream is evaluated serially:
669+--
670+-- >>> stream1 = Stream.fromListM $ map delay [3,3] -- SerialT IO Int
671+-- >>> stream2 = Stream.fromListM $ map delay [1,1] -- SerialT IO Int
672+-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
673+-- [1,1,3,3]
674+--
675+-- If total threads are 2, the third stream is scheduled only after one of the
676+-- first two has finished:
677+--
678+-- >>> stream3 = Stream.fromListM $ map delay [2,2] -- SerialT IO Int
679+-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
680+-- [1,1,3,2,3,2]
681+--
682+-- Thus 'async' goes deep in first few streams rather than going wide in all
683+-- streams. It prefers to evaluate the leftmost streams as much as possible.
684+-- Because of this behavior, 'async' can be safely used to fold an infinite
685+-- lazy container of streams.
640686--
641687-- /Since: 0.2.0 ("Streamly")/
642688--
@@ -668,51 +714,45 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
668714-- AsyncT
669715------------------------------------------------------------------------------
670716
671−-- | The 'Semigroup' operation (@<>@) for 'AsyncT' merges two streams
672−-- concurrently with priority given to the first stream. In @s1 <> s2 <> s3
673−-- ...@ the streams s1, s2 and s3 are scheduled for execution in that order.
674−-- Multiple scheduled streams may be executed concurrently and the elements
675−-- generated by them are served to the consumer as and when they become
676−-- available. This behavior is similar to the scheduling and execution behavior
677−-- of actions in a single async stream.
678−--
679−-- Since only a finite number of streams are executed concurrently, this
680−-- operation can be used to fold an infinite lazy container of streams.
681−--
682−-- @
683−-- import qualified "Streamly.Prelude" as S
684−-- import Control.Concurrent
717+-- | For 'AsyncT' streams:
685718--
686−-- main = (S.toList . S.'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
687−-- @
688719-- @
689−-- [1,2,3,4]
720+-- (<>) = 'Streamly.Prelude.async'
721+-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
690722-- @
691723--
692−-- Any exceptions generated by a constituent stream are propagated to the
693−-- output stream. The output and exceptions from a single stream are guaranteed
694−-- to arrive in the same order in the resulting stream as they were generated
695−-- in the input stream. However, the relative ordering of elements from
696−-- different streams in the resulting stream can vary depending on scheduling
697−-- and generation delays.
724+-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
725+-- executed concurrently a la the 'async' combinator, producing results and
726+-- side effects of iterations out of order:
698727--
699−-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
700−-- concurrently based on demand. More concurrent iterations are started only
701−-- if the previous iterations are not able to produce enough output for the
702−-- consumer.
728+-- >>> :{
729+-- Stream.toList $ Stream.asyncly $ do
730+-- x <- Stream.fromList [2,1] -- foreach x in stream
731+-- Stream.yieldM $ delay x
732+-- :}
733+-- 1 sec
734+-- 2 sec
735+-- [1,2]
703736--
704−-- @
705−-- main = S.'drain' . S.'asyncly' $ do
706−-- n <- return 3 \<\> return 2 \<\> return 1
707−-- S.yieldM $ do
708−-- threadDelay (n * 1000000)
709−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
710−-- @
711−-- @
712−-- ThreadId 40: Delay 1
713−-- ThreadId 39: Delay 2
714−-- ThreadId 38: Delay 3
715−-- @
737+-- Nested monad binds behave like nested @for@ loops with nested iterations
738+-- executed concurrently, a la the 'async' combinator:
739+--
740+-- >>> :{
741+-- Stream.toList $ Stream.asyncly $ do
742+-- x <- Stream.fromList [1,2] -- foreach x in stream
743+-- y <- Stream.fromList [2,4] -- foreach y in stream
744+-- Stream.yieldM $ delay (x + y)
745+-- :}
746+-- 3 sec
747+-- 4 sec
748+-- 5 sec
749+-- 6 sec
750+-- [3,4,5,6]
751+--
752+-- The behavior can be explained as follows. All the iterations corresponding
753+-- to the element @1@ in the first stream constitute one output stream and all
754+-- the iterations corresponding to @2@ constitute another output stream and
755+-- these two output streams are merged using 'async'.
716756--
717757-- /Since: 0.1.0 ("Streamly")/
718758--
@@ -817,28 +857,119 @@ consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
817857
818858infixr 6 `wAsync`
819859
820−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
821−-- Merges two streams concurrently choosing elements from both fairly.
860+-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
861+-- singleton stream behavior. For multi-element streams, while 'async' is left
862+-- biased i.e. it tries to evaluate the left side stream as much as possible,
863+-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
864+-- deep while 'wAsync' goes wide. However, outputs are always used as they
865+-- arrive.
866+--
867+-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
868+-- starts behaving like 'wSerial'.
869+--
870+-- >>> stream1 = Stream.fromList [1,2,3]
871+-- >>> stream2 = Stream.fromList [4,5,6]
872+-- >>> Stream.toList $ Stream.asyncly $ Stream.maxThreads 1 $ stream1 `async` stream2
873+-- [1,2,3,4,5,6]
874+--
875+-- >>> Stream.toList $ Stream.wAsyncly $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
876+-- [1,4,2,5,3,6]
877+--
878+-- With two threads available, and combining three streams:
879+--
880+-- >>> stream3 = Stream.fromList [7,8,9]
881+-- >>> Stream.toList $ Stream.asyncly $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
882+-- [1,2,3,4,5,6,7,8,9]
883+--
884+-- >>> Stream.toList $ Stream.wAsyncly $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
885+-- [1,4,2,7,5,3,8,6,9]
886+--
887+-- This operation cannot be used to fold an infinite lazy container of streams,
888+-- because it schedules all the streams in a round robin manner.
889+--
890+-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
891+-- but the exact scheduling is slightly different in both cases.
822892--
823893-- @since 0.8.0
824894--
825895-- /Since: 0.2.0 ("Streamly")/
896+
897+-- Scheduling details:
898+--
899+-- This is how the execution of the above example proceeds:
900+--
901+-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
902+-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
903+-- queue is represented by the rightmost item.
904+-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
905+-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
906+-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
907+-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
908+-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
909+-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
910+-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
911+-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
912+-- 4. Note that the scheduler queue expands by one more stream component in
913+-- every pass because one more @<>@ is broken down into two components. At this
914+-- point there are no more @<>@ operations to be broken down further and the
915+-- queue has reached its maximum size. Now these streams are scheduled in
916+-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
917+--
918+-- As we see above, in a right associated expression composed with @<>@, only
919+-- one @<>@ operation is broken down into two components in one execution,
920+-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
921+-- scheduler passes to expand the whole expression. By the time @n-th@
922+-- component is added to the scheduler queue, the first component would have
923+-- received @n@ scheduler passes.
924+--
826925{-# INLINE wAsync #-}
827926wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
828927wAsync = joinStreamVarAsync WAsyncVar
829928
830−-- | 'WAsyncT' is similar to 'WSerialT' but with concurrent execution.
831−-- The 'Semigroup' operation (@<>@) for 'WAsyncT' merges two streams
832−-- concurrently interleaving the actions from both the streams. In @s1
833−-- <> s2 <> s3 ...@, the individual actions from streams @s1@, @s2@ and @s3@
834−-- are scheduled for execution in a round-robin fashion. Multiple scheduled
835−-- actions may be executed concurrently, the results from concurrent executions
836−-- are consumed in the order in which they become available.
929+-- | For 'WAsyncT' streams:
930+--
931+-- @
932+-- (<>) = 'Streamly.Prelude.wAsync'
933+-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
934+-- @
935+--
936+-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
937+-- executed concurrently a la the 'wAsync' combinator, producing results and
938+-- side effects of iterations out of order:
837939--
940+-- >>> :{
941+-- Stream.toList $ Stream.wAsyncly $ do
942+-- x <- Stream.fromList [2,1] -- foreach x in stream
943+-- Stream.yieldM $ delay x
944+-- :}
945+-- 1 sec
946+-- 2 sec
947+-- [1,2]
948+--
949+-- Nested monad binds behave like nested @for@ loops with nested iterations
950+-- executed concurrently, a la the 'wAsync' combinator:
951+--
952+-- >>> :{
953+-- Stream.toList $ Stream.wAsyncly $ do
954+-- x <- Stream.fromList [1,2] -- foreach x in stream
955+-- y <- Stream.fromList [2,4] -- foreach y in stream
956+-- Stream.yieldM $ delay (x + y)
957+-- :}
958+-- 3 sec
959+-- 4 sec
960+-- 5 sec
961+-- 6 sec
962+-- [3,4,5,6]
963+--
964+-- The behavior can be explained as follows. All the iterations corresponding
965+-- to the element @1@ in the first stream constitute one 'WAsyncT' output
966+-- stream and all the iterations corresponding to @2@ constitute another
967+-- 'WAsyncT' output stream and these two output streams are merged using
968+-- 'wAsync'.
838969--
839970-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
840971-- contrast to the depth wise scheduling behavior of 'AsyncT'.
841−--
972+
842973-- @
843974-- import qualified "Streamly.Prelude" as S
844975-- import Control.Concurrent