@@ -634,9 +634,55 @@ joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
635infixr 6 `async`
635infixr 6 `async`
637−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
637+-- | Merges two streams, both the streams may be evaluated concurrently,
638−-- Merges two streams possibly concurrently, preferring the
638+-- outputs from both are used as they arrive:
639−-- elements from the left one when available.
639+--
640+-- >>> stream1 = Stream.yieldM (delay 4)
641+-- >>> stream2 = Stream.yieldM (delay 2)
642+-- >>> Stream.toList $ stream1 `async` stream2
645+-- Multiple streams can be combined. With enough threads, all of them can be
646+-- scheduled simultaneously:
648+-- >>> stream3 = Stream.yieldM (delay 1)
649+-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
652+-- With 2 threads, only two can be scheduled at a time, when one of those
653+-- finishes, the third one gets scheduled:
655+-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
658+-- With a single thread, it becomes serial:
660+-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
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.
667+-- In the following example, both the streams are scheduled for concurrent
668+-- evaluation but each individual stream is evaluated serially:
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]
675+-- If total threads are 2, the third stream is scheduled only after one of the
676+-- first two has finished:
678+-- >>> stream3 = Stream.fromListM $ map delay [2,2] -- SerialT IO Int
679+-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
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.
641-- /Since: 0.2.0 ("Streamly")/
687-- /Since: 0.2.0 ("Streamly")/
@@ -668,51 +714,45 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
669------------------------------------------------------------------------------
715------------------------------------------------------------------------------
671−-- | The 'Semigroup' operation (@<>@) for 'AsyncT' merges two streams
717+-- | For 'AsyncT' 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.
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.
683−-- import qualified "Streamly.Prelude" as S
684−-- import Control.Concurrent
686−-- main = (S.toList . S.'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
689−-- [1,2,3,4]
720+-- (<>) = 'Streamly.Prelude.async'
721+-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
692−-- Any exceptions generated by a constituent stream are propagated to the
724+-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
693−-- output stream. The output and exceptions from a single stream are guaranteed
725+-- executed concurrently a la the 'async' combinator, producing results and
694−-- to arrive in the same order in the resulting stream as they were generated
726+-- side effects of iterations out of order:
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.
699−-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
728+-- >>> :{
700−-- concurrently based on demand. More concurrent iterations are started only
729+-- Stream.toList $ Stream.asyncly $ do
701−-- if the previous iterations are not able to produce enough output for the
730+-- x <- Stream.fromList [2,1] -- foreach x in stream
702−-- consumer.
731+-- Stream.yieldM $ delay x
704−-- @
737+-- Nested monad binds behave like nested @for@ loops with nested iterations
705−-- main = S.'drain' . S.'asyncly' $ do
738+-- executed concurrently, a la the 'async' combinator:
706−-- n <- return 3 \<\> return 2 \<\> return 1
739+--
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
707−-- S.yieldM $ do
744+-- Stream.yieldM $ delay (x + y)
708−-- threadDelay (n * 1000000)
745+-- :}
709−-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
746+-- 3 sec
712−-- ThreadId 40: Delay 1
749+-- 6 sec
713−-- ThreadId 39: Delay 2
750+-- [3,4,5,6]
714−-- ThreadId 38: Delay 3
751+--
715−-- @
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'.
717-- /Since: 0.1.0 ("Streamly")/
757-- /Since: 0.1.0 ("Streamly")/
@@ -817,28 +857,119 @@ consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
818infixr 6 `wAsync`
858infixr 6 `wAsync`
820−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
860+-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
821−-- Merges two streams concurrently choosing elements from both fairly.
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
867+-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
868+-- starts behaving like 'wSerial'.
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
875+-- >>> Stream.toList $ Stream.wAsyncly $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
878+-- With two threads available, and combining three streams:
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]
884+-- >>> Stream.toList $ Stream.wAsyncly $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
885+-- [1,4,2,7,5,3,8,6,9]
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.
890+-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
891+-- but the exact scheduling is slightly different in both cases.
823-- @since 0.8.0
893-- @since 0.8.0
825-- /Since: 0.2.0 ("Streamly")/
895-- /Since: 0.2.0 ("Streamly")/
897+-- Scheduling details:
899+-- This is how the execution of the above example proceeds:
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]@.
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.
826{-# INLINE wAsync #-}
925{-# INLINE wAsync #-}
827wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
926wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
828wAsync = joinStreamVarAsync WAsyncVar
927wAsync = joinStreamVarAsync WAsyncVar
830−-- | 'WAsyncT' is similar to 'WSerialT' but with concurrent execution.
929+-- | For 'WAsyncT' streams:
831−-- The 'Semigroup' operation (@<>@) for 'WAsyncT' merges two streams
930+--
832−-- concurrently interleaving the actions from both the streams. In @s1
931+-- @
833−-- <> s2 <> s3 ...@, the individual actions from streams @s1@, @s2@ and @s3@
932+-- (<>) = 'Streamly.Prelude.wAsync'
834−-- are scheduled for execution in a round-robin fashion. Multiple scheduled
933+-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
936+-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
835−-- actions may be executed concurrently, the results from concurrent executions
937+-- executed concurrently a la the 'wAsync' combinator, producing results and
836−-- are consumed in the order in which they become available.
938+-- side effects of iterations out of order:
941+-- Stream.toList $ Stream.wAsyncly $ do
942+-- x <- Stream.fromList [2,1] -- foreach x in stream
943+-- Stream.yieldM $ delay x
949+-- Nested monad binds behave like nested @for@ loops with nested iterations
950+-- executed concurrently, a la the 'wAsync' combinator:
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)
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
839-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
970-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
840-- contrast to the depth wise scheduling behavior of 'AsyncT'.
971-- contrast to the depth wise scheduling behavior of 'AsyncT'.
843-- import qualified "Streamly.Prelude" as S
974-- import qualified "Streamly.Prelude" as S
844-- import Control.Concurrent
975-- import Control.Concurrent