1{-# LANGUAGE UndecidableInstances #-}
6-- Module : Streamly.Internal.Data.Stream.Async
7-- Copyright : (c) 2017 Composewell Technologies
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
14-- To run examples in this module:
16-- >>> import qualified Streamly.Prelude as Stream
17-- >>> import Control.Concurrent (threadDelay)
20-- threadDelay (n * 1000000) -- sleep for n seconds
21-- putStrLn (show n ++ " sec") -- print "n sec"
25module Streamly.Internal.Data.Stream.Async
41import Control.Concurrent (myThreadId)
42import Control.Monad.Base (MonadBase(..), liftBaseDefault)
43import Control.Monad.Catch (MonadThrow, throwM)
44import Control.Concurrent.MVar (newEmptyMVar)
45-- import Control.Monad.Error.Class (MonadError(..))
46import Control.Monad.IO.Class (MonadIO(..))
47import Control.Monad.Reader.Class (MonadReader(..))
48import Control.Monad.State.Class (MonadState(..))
49import Control.Monad.Trans.Class (MonadTrans(lift))
50import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)
51import Data.IORef (IORef, newIORef, readIORef)
52import Data.Maybe (fromJust)
53#if __GLASGOW_HASKELL__ < 808
54import Data.Semigroup (Semigroup(..))
57import Prelude hiding (map)
58import qualified Data.Set as S
60import Streamly.Internal.Control.Concurrent
61 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO, restoreM)
62import Streamly.Internal.Data.Atomics
63 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
64import Streamly.Internal.Data.Stream.Serial (SerialT (..))
65import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
66import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
68import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
69import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
71import Streamly.Internal.Data.SVar.Type
72import Streamly.Internal.Data.SVar
74#include "Instances.hs"
77-- >>> import qualified Streamly.Prelude as Stream
78-- >>> import Control.Concurrent (threadDelay)
81-- threadDelay (n * 1000000) -- sleep for n seconds
82-- putStrLn (show n ++ " sec") -- print "n sec"
87-------------------------------------------------------------------------------
89-------------------------------------------------------------------------------
91-- Note: For purely right associated expressions this queue should have at most
92-- one element. It grows to more than one when we have left associcated
93-- expressions. Large left associated compositions can grow this to a
95{-# INLINE enqueueLIFO #-}
97 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
98enqueueLIFO sv q m = do
99 atomicModifyIORefCAS_ q $ \ms -> m : ms
102data WorkerStatus = Continue | Suspend
104{-# INLINE workLoopLIFO #-}
107 => IORef [(RunInIO m, Stream m a)]
112workLoopLIFO q st sv winfo = run
116 stop = liftIO $ sendStop sv winfo
121 Just (RunInIO runin, m) -> do
122 -- XXX when we finish we need to send the monadic state back to
123 -- the parent so that the state can be merged back. We capture
124 -- and return the state in the stop continuation.
126 -- Instead of using the run function we can just restore the
127 -- monad state here. That way it can work easily for
128 -- distributed case as well.
129 r <- liftIO $ runin $
130 K.foldStreamShared st yieldk single (return Continue) m
137 res <- liftIO $ sendYield sv winfo (ChildYield a)
138 return $ if res then Continue else Suspend
141 res <- liftIO $ sendYield sv winfo (ChildYield a)
143 then K.foldStreamShared st yieldk single (return Continue) r
145 runInIO <- askRunInIO
146 liftIO $ enqueueLIFO sv q (runInIO, r)
149 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
151 x : xs -> (xs, Just x)
153-- We duplicate workLoop for yield limit and no limit cases because it has
154-- around 40% performance overhead in the worst case.
156-- XXX we can pass yinfo directly as an argument here so that we do not have to
157-- make a check every time.
158{-# INLINE workLoopLIFOLimited #-}
160 :: forall m a. MonadRunInIO m
161 => IORef [(RunInIO m, Stream m a)]
166workLoopLIFOLimited q st sv winfo = run
170 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
171 stop = liftIO $ sendStop sv winfo
176 Just (RunInIO runin, m) -> do
177 -- XXX This is just a best effort minimization of concurrency
178 -- to the yield limit. If the stream is made of concurrent
179 -- streams we do not reserve the yield limit in the constituent
180 -- streams before executing the action. This can be done
181 -- though, by sharing the yield limit ref with downstream
182 -- actions via state passing. Just a todo.
183 yieldLimitOk <- liftIO $ decrementYieldLimit sv
186 r <- liftIO $ runin $
187 K.foldStreamShared st yieldk single incrContinue m
192 -- Avoid any side effects, undo the yield limit decrement if we
193 -- never yielded anything.
195 enqueueLIFO sv q (RunInIO runin, m)
196 incrementYieldLimit sv
200 res <- liftIO $ sendYield sv winfo (ChildYield a)
201 return $ if res then Continue else Suspend
203 -- XXX can we pass on the yield limit downstream to limit the concurrency
204 -- of constituent streams.
206 res <- liftIO $ sendYield sv winfo (ChildYield a)
207 yieldLimitOk <- liftIO $ decrementYieldLimit sv
208 if res && yieldLimitOk
209 then K.foldStreamShared st yieldk single incrContinue r
211 runInIO <- askRunInIO
212 liftIO $ incrementYieldLimit sv
213 liftIO $ enqueueLIFO sv q (runInIO, r)
216 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
218 x : xs -> (xs, Just x)
220-------------------------------------------------------------------------------
222-------------------------------------------------------------------------------
224-- XXX we can use the Ahead style sequence/heap mechanism to make the best
225-- effort to always try to finish the streams on the left side of an expression
226-- first as long as possible.
228{-# INLINE enqueueFIFO #-}
231 -> LinkedQueue (RunInIO m, t m a)
232 -> (RunInIO m, t m a)
234enqueueFIFO sv q m = do
238-- XXX we can remove sv as it is derivable from st
240{-# INLINE workLoopFIFO #-}
243 => LinkedQueue (RunInIO m, Stream m a)
248workLoopFIFO q st sv winfo = run
252 stop = liftIO $ sendStop sv winfo
254 work <- liftIO $ tryPopR q
257 Just (RunInIO runin, m) -> do
258 r <- liftIO $ runin $
259 K.foldStreamShared st yieldk single (return Continue) m
266 res <- liftIO $ sendYield sv winfo (ChildYield a)
267 return $ if res then Continue else Suspend
269 -- XXX in general we would like to yield "n" elements from a single stream
270 -- before moving on to the next. Single element granularity could be too
271 -- expensive in certain cases. Similarly, we can use time limit for
274 res <- liftIO $ sendYield sv winfo (ChildYield a)
275 runInIO <- askRunInIO
276 liftIO $ enqueueFIFO sv q (runInIO, r)
277 return $ if res then Continue else Suspend
279{-# INLINE workLoopFIFOLimited #-}
281 :: forall m a. MonadRunInIO m
282 => LinkedQueue (RunInIO m, Stream m a)
287workLoopFIFOLimited q st sv winfo = run
291 stop = liftIO $ sendStop sv winfo
292 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
294 work <- liftIO $ tryPopR q
297 Just (RunInIO runin, m) -> do
298 yieldLimitOk <- liftIO $ decrementYieldLimit sv
301 r <- liftIO $ runin $
302 K.foldStreamShared st yieldk single incrContinue m
308 enqueueFIFO sv q (RunInIO runin, m)
309 incrementYieldLimit sv
313 res <- liftIO $ sendYield sv winfo (ChildYield a)
314 return $ if res then Continue else Suspend
317 res <- liftIO $ sendYield sv winfo (ChildYield a)
318 runInIO <- askRunInIO
319 liftIO $ enqueueFIFO sv q (runInIO, r)
320 yieldLimitOk <- liftIO $ decrementYieldLimit sv
321 if res && yieldLimitOk
324 incrementYieldLimit sv
327-------------------------------------------------------------------------------
329-- This code belongs in SVar.hs but is kept here for perf reasons
330-------------------------------------------------------------------------------
332-- XXX we have this function in this file because passing runStreamLIFO as a
333-- function argument to this function results in a perf degradation of more
334-- than 10%. Need to investigate what the root cause is.
335-- Interestingly, the same thing does not make any difference for Ahead.
336getLifoSVar :: forall m a. MonadAsync m
337 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
338getLifoSVar st mrun = do
339 outQ <- newIORef ([], 0)
340 outQMv <- newEmptyMVar
342 wfw <- newIORef False
343 running <- newIORef S.empty
344 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
345 yl <- case getYieldLimit st of
346 Nothing -> return Nothing
347 Just x -> Just <$> newIORef x
348 rateInfo <- getYieldRateInfo st
350 stats <- newSVarStats
353 let isWorkFinished _ = null <$> readIORef q
355 let isWorkFinishedLimited sv = do
357 case remainingWork sv of
361 Nothing -> return False
362 qEmpty <- null <$> readIORef q
363 return $ qEmpty || yieldsDone
365 let getSVar :: SVar Stream m a
366 -> (SVar Stream m a -> m [ChildEvent a])
367 -> (SVar Stream m a -> m Bool)
368 -> (SVar Stream m a -> IO Bool)
369 -> (IORef [(RunInIO m, Stream m a)]
375 getSVar sv readOutput postProc workDone wloop = SVar
377 , outputQueueFromConsumer = undefined
379 , maxBufferLimit = getMaxBuffer st
380 , pushBufferSpace = undefined
381 , pushBufferPolicy = undefined
382 , pushBufferMVar = undefined
383 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
384 , yieldRateInfo = rateInfo
385 , outputDoorBell = outQMv
386 , outputDoorBellFromConsumer = undefined
387 , readOutputQ = readOutput sv
388 , postProcess = postProc sv
389 , workerThreads = running
390 , workLoop = wloop q st{streamVar = Just sv} sv
391 , enqueue = enqueueLIFO sv q
392 , isWorkDone = workDone sv
393 , isQueueDone = workDone sv
395 , svarStyle = AsyncVar
396 , svarStopStyle = StopNone
397 , svarStopBy = undefined
399 , workerCount = active
400 , accountThread = delThread sv
401 , workerStopMVar = undefined
403 , svarInspectMode = getInspectMode st
405 , aheadWorkQueue = undefined
406 , outputHeap = undefined
411 case getStreamRate st of
413 case getYieldLimit st of
414 Nothing -> getSVar sv readOutputQBounded
418 Just _ -> getSVar sv readOutputQBounded
420 isWorkFinishedLimited
423 case getYieldLimit st of
424 Nothing -> getSVar sv readOutputQPaced
428 Just _ -> getSVar sv readOutputQPaced
430 isWorkFinishedLimited
434getFifoSVar :: forall m a. MonadAsync m
435 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
436getFifoSVar st mrun = do
437 outQ <- newIORef ([], 0)
438 outQMv <- newEmptyMVar
440 wfw <- newIORef False
441 running <- newIORef S.empty
443 yl <- case getYieldLimit st of
444 Nothing -> return Nothing
445 Just x -> Just <$> newIORef x
446 rateInfo <- getYieldRateInfo st
448 stats <- newSVarStats
451 let isWorkFinished _ = nullQ q
452 let isWorkFinishedLimited sv = do
454 case remainingWork sv of
458 Nothing -> return False
460 return $ qEmpty || yieldsDone
462 let getSVar :: SVar Stream m a
463 -> (SVar Stream m a -> m [ChildEvent a])
464 -> (SVar Stream m a -> m Bool)
465 -> (SVar Stream m a -> IO Bool)
466 -> (LinkedQueue (RunInIO m, Stream m a)
472 getSVar sv readOutput postProc workDone wloop = SVar
474 , outputQueueFromConsumer = undefined
476 , maxBufferLimit = getMaxBuffer st
477 , pushBufferSpace = undefined
478 , pushBufferPolicy = undefined
479 , pushBufferMVar = undefined
480 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
481 , yieldRateInfo = rateInfo
482 , outputDoorBell = outQMv
483 , outputDoorBellFromConsumer = undefined
484 , readOutputQ = readOutput sv
485 , postProcess = postProc sv
486 , workerThreads = running
487 , workLoop = wloop q st{streamVar = Just sv} sv
488 , enqueue = enqueueFIFO sv q
489 , isWorkDone = workDone sv
490 , isQueueDone = workDone sv
492 , svarStyle = WAsyncVar
493 , svarStopStyle = StopNone
494 , svarStopBy = undefined
496 , workerCount = active
497 , accountThread = delThread sv
498 , workerStopMVar = undefined
500 , svarInspectMode = getInspectMode st
502 , aheadWorkQueue = undefined
503 , outputHeap = undefined
508 case getStreamRate st of
510 case getYieldLimit st of
511 Nothing -> getSVar sv readOutputQBounded
515 Just _ -> getSVar sv readOutputQBounded
517 isWorkFinishedLimited
520 case getYieldLimit st of
521 Nothing -> getSVar sv readOutputQPaced
525 Just _ -> getSVar sv readOutputQPaced
527 isWorkFinishedLimited
531{-# INLINABLE newAsyncVar #-}
532newAsyncVar :: MonadAsync m
533 => State Stream m a -> Stream m a -> m (SVar Stream m a)
536 sv <- liftIO $ getLifoSVar st mrun
539-- | Generate a stream asynchronously to keep it buffered, lazily consume
544{-# INLINABLE mkAsyncK #-}
545mkAsyncK :: MonadAsync m => Stream m a -> Stream m a
546mkAsyncK m = K.mkStream $ \st yld sng stp -> do
547 sv <- newAsyncVar (adaptState st) m
548 K.foldStream st yld sng stp $ getSerialT $ fromSVar sv
551-- This is slightly faster than the CPS version above
553{-# INLINE_NORMAL mkAsyncD #-}
554mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
555mkAsyncD m = D.Stream step Nothing
558 step gst Nothing = do
559 sv <- newAsyncVar gst (D.toStreamK m)
560 return $ D.Skip $ Just $ fromSVarD sv
562 step gst (Just (D.UnStream step1 st)) = do
565 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
566 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
569-- XXX We can pass captureMonadState instead of using MonadAsync
570-- | Create a new SVar and enqueue one stream computation on it.
571{-# INLINABLE newWAsyncVar #-}
572newWAsyncVar :: MonadAsync m
573 => State Stream m a -> Stream m a -> m (SVar Stream m a)
574newWAsyncVar st m = do
576 sv <- liftIO $ getFifoSVar st mrun
577 -- XXX Use just Stream and IO in all the functions below
578 -- XXX pass mrun instead of calling captureMonadState again inside it
581------------------------------------------------------------------------------
582-- Running streams concurrently
583------------------------------------------------------------------------------
585-- Concurrency rate control.
587-- Our objective is to create more threads on demand if the consumer is running
588-- faster than us. As soon as we encounter a concurrent composition we create a
589-- push pull pair of threads. We use an SVar for communication between the
590-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
591-- The producer creates more threads if the SVar drains and becomes empty, that
592-- is the consumer is running faster.
594-- XXX Note 1: This mechanism can be problematic if the initial production
595-- latency is high, we may end up creating too many threads. So we need some
596-- way to monitor and use the latency as well. Having a limit on the dispatches
597-- (programmer controlled) may also help.
599-- TBD Note 2: We may want to run computations at the lower level of the
600-- composition tree serially even when they are composed using a parallel
601-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
602-- place of 'wAsync'. If we find that an SVar immediately above a computation
603-- gets drained empty we can switch to parallelizing the computation. For that
604-- we can use a state flag to fork the rest of the computation at any point of
605-- time inside the Monad bind operation if the consumer is running at a faster
608-- TBD Note 3: the binary operation ('parallel') composition allows us to
609-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
610-- chunksizes we will need to compose the parallel actions using a data
611-- constructor (A Free container) instead so that we can divide it in chunks of
612-- arbitrary size before dispatching. If the stream is composed of
613-- hierarchically composed grains of different sizes then we can always switch
614-- to a desired granularity depending on the consumer speed.
616-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
617-- into just the number of CPUs.
619-- | Join two computations on the currently running 'SVar' queue for concurrent
620-- execution. When we are using parallel composition, an SVar is passed around
621-- as a state variable. We try to schedule a new parallel computation on the
622-- SVar passed to us. The first time, when no SVar exists, a new SVar is
623-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
624-- already scheduled on the SVar is further evaluated. For example, when (a
625-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
626-- the current scheduler queue.
628-- The 'SVarStyle' required by the current composition context is passed as one
629-- of the parameters. If the scheduling and composition style of the new
630-- computation being scheduled is different than the style of the current SVar,
631-- then we create a new SVar and schedule it on that. The newly created SVar
632-- joins as one of the computations on the current SVar queue.
634-- Cases when we need to switch to a new SVar:
636-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
637-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
638-- because of the scheduling policy change.
639-- * if we 'adapt' a stream of type 'async' to a stream of type
640-- 'Parallel', we create a new SVar at the transitioning bind.
641-- * When the stream is switching from disjunctive composition to conjunctive
642-- composition and vice-versa we create a new SVar to isolate the scheduling
645forkSVarAsync :: MonadAsync m
646 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
647forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do
649 AsyncVar -> newAsyncVar st (concurrently m1 m2)
650 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
651 _ -> error "illegal svar type"
652 K.foldStream st yld sng stp $ getSerialT $ fromSVar sv
654 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
655 runInIO <- askRunInIO
656 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
657 K.foldStreamShared st yld sng stp ma
659{-# INLINE joinStreamVarAsync #-}
660joinStreamVarAsync :: MonadAsync m
661 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
662joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->
664 Just sv | svarStyle sv == style -> do
665 runInIO <- askRunInIO
666 liftIO $ enqueue sv (runInIO, m2)
667 K.foldStreamShared st yld sng stp m1
668 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
670------------------------------------------------------------------------------
671-- Semigroup and Monoid style compositions for parallel actions
672------------------------------------------------------------------------------
675asyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
676asyncK = joinStreamVarAsync AsyncVar
678-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
679-- huge difference in the performance of consM in IsStream instance even we
680-- have a SPECIALIZE in the instance.
682-- | XXX we can implement it more efficienty by directly implementing instead
683-- of combining streams using async.
684{-# INLINE consMAsync #-}
685{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
686consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
687consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r
689------------------------------------------------------------------------------
691------------------------------------------------------------------------------
693-- | For 'AsyncT' streams:
696-- (<>) = 'Streamly.Prelude.async'
697-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
700-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
701-- executed concurrently a la the 'async' combinator, producing results and
702-- side effects of iterations out of order:
705-- Stream.toList $ Stream.fromAsync $ do
706-- x <- Stream.fromList [2,1] -- foreach x in stream
707-- Stream.fromEffect $ delay x
713-- Nested monad binds behave like nested @for@ loops with nested iterations
714-- executed concurrently, a la the 'async' combinator:
717-- Stream.toList $ Stream.fromAsync $ do
718-- x <- Stream.fromList [1,2] -- foreach x in stream
719-- y <- Stream.fromList [2,4] -- foreach y in stream
720-- Stream.fromEffect $ delay (x + y)
728-- The behavior can be explained as follows. All the iterations corresponding
729-- to the element @1@ in the first stream constitute one output stream and all
730-- the iterations corresponding to @2@ constitute another output stream and
731-- these two output streams are merged using 'async'.
733-- /Since: 0.1.0 ("Streamly")/
736newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
737 deriving (MonadTrans)
739-- | A demand driven left biased parallely composing IO stream of elements of
740-- type @a@. See 'AsyncT' documentation for more details.
742-- /Since: 0.2.0 ("Streamly")/
745type Async = AsyncT IO
747------------------------------------------------------------------------------
749------------------------------------------------------------------------------
751-- Monomorphically typed version of "async" for better performance of Semigroup
754{-# SPECIALIZE append :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
755append :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
756append (AsyncT m1) (AsyncT m2) = AsyncT $ asyncK m1 m2
758instance MonadAsync m => Semigroup (AsyncT m a) where
761------------------------------------------------------------------------------
763------------------------------------------------------------------------------
765instance MonadAsync m => Monoid (AsyncT m a) where
766 mempty = AsyncT K.nil
769------------------------------------------------------------------------------
771------------------------------------------------------------------------------
773{-# INLINE apAsync #-}
774{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
775apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
776apAsync (AsyncT m1) (AsyncT m2) =
777 let f x1 = K.concatMapWith asyncK (pure . x1) m2
778 in AsyncT $ K.concatMapWith asyncK f m1
780instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
782 pure = AsyncT . K.fromPure
787------------------------------------------------------------------------------
789------------------------------------------------------------------------------
791-- GHC: if we change the implementation of bindWith with arguments in a
792-- different order we see a significant performance degradation (~2x).
793{-# INLINE bindAsync #-}
794{-# SPECIALIZE bindAsync ::
795 AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
796bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
797bindAsync (AsyncT m) f = AsyncT $ K.bindWith asyncK m (getAsyncT . f)
799-- GHC: if we specify arguments in the definition of (>>=) we see a significant
800-- performance degradation (~2x).
801instance MonadAsync m => Monad (AsyncT m) where
805------------------------------------------------------------------------------
807------------------------------------------------------------------------------
809MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
811------------------------------------------------------------------------------
813------------------------------------------------------------------------------
815{-# INLINE wAsyncK #-}
816wAsyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
817wAsyncK = joinStreamVarAsync WAsyncVar
819-- | XXX we can implement it more efficienty by directly implementing instead
820-- of combining streams using wAsync.
821{-# INLINE consMWAsync #-}
822{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
823consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
824consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r
826-- | For 'WAsyncT' streams:
829-- (<>) = 'Streamly.Prelude.wAsync'
830-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
833-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
834-- executed concurrently a la the 'wAsync' combinator, producing results and
835-- side effects of iterations out of order:
838-- Stream.toList $ Stream.fromWAsync $ do
839-- x <- Stream.fromList [2,1] -- foreach x in stream
840-- Stream.fromEffect $ delay x
846-- Nested monad binds behave like nested @for@ loops with nested iterations
847-- executed concurrently, a la the 'wAsync' combinator:
850-- Stream.toList $ Stream.fromWAsync $ do
851-- x <- Stream.fromList [1,2] -- foreach x in stream
852-- y <- Stream.fromList [2,4] -- foreach y in stream
853-- Stream.fromEffect $ delay (x + y)
861-- The behavior can be explained as follows. All the iterations corresponding
862-- to the element @1@ in the first stream constitute one 'WAsyncT' output
863-- stream and all the iterations corresponding to @2@ constitute another
864-- 'WAsyncT' output stream and these two output streams are merged using
867-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
868-- contrast to the depth wise scheduling behavior of 'AsyncT'.
870-- /Since: 0.2.0 ("Streamly")/
874-- XXX This documentation is redundant, need to be cleaned up/removed.
877-- import qualified "Streamly.Prelude" as S
878-- import Control.Concurrent
880-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
886-- For this example, we are using @maxThreads 1@ so that concurrent thread
887-- scheduling does not affect the results and make them unpredictable. Let's
888-- now take a more general example:
891-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
894-- [1,4,2,7,5,3,8,6,9]
897-- This is how the execution of the above stream proceeds:
899-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
900-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
901-- queue is represented by the rightmost item.
902-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
903-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
904-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
905-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
906-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
907-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
908-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
909-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
910-- 4. Note that the scheduler queue expands by one more stream component in
911-- every pass because one more @<>@ is broken down into two components. At this
912-- point there are no more @<>@ operations to be broken down further and the
913-- queue has reached its maximum size. Now these streams are scheduled in
914-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
916-- As we see above, in a right associated expression composed with @<>@, only
917-- one @<>@ operation is broken down into two components in one execution,
918-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
919-- scheduler passes to expand the whole expression. By the time @n-th@
920-- component is added to the scheduler queue, the first component would have
921-- received @n@ scheduler passes.
923-- Since all streams get interleaved, this operation is not suitable for
924-- folding an infinite lazy container of infinite size streams. However, if
925-- the streams are small, the streams on the left may get finished before more
926-- streams are added to the scheduler queue from the right side of the
927-- expression, so it may be possible to fold an infinite lazy container of
928-- streams. For example, if the streams are of size @n@ then at most @n@
929-- streams would be in the scheduler queue at a time.
931-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
932-- therefore the output of 'WAsyncT' even with a single thread of execution is
933-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
934-- its scheduling behavior.
936-- Any exceptions generated by a constituent stream are propagated to the
937-- output stream. The output and exceptions from a single stream are guaranteed
938-- to arrive in the same order in the resulting stream as they were generated
939-- in the input stream. However, the relative ordering of elements from
940-- different streams in the resulting stream can vary depending on scheduling
941-- and generation delays.
943-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
944-- concurrently using a round robin scheduling.
947-- main = S.'drain' . S.'fromWAsync' $ do
948-- n <- return 3 \<\> return 2 \<\> return 1
950-- threadDelay (n * 1000000)
951-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
954-- ThreadId 40: Delay 1
955-- ThreadId 39: Delay 2
956-- ThreadId 38: Delay 3
959newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
960 deriving (MonadTrans)
962-- | A round robin parallely composing IO stream of elements of type @a@.
963-- See 'WAsyncT' documentation for more details.
965-- /Since: 0.2.0 ("Streamly")/
968type WAsync = WAsyncT IO
970------------------------------------------------------------------------------
972------------------------------------------------------------------------------
974{-# INLINE wAppend #-}
975{-# SPECIALIZE wAppend :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
976wAppend :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
977wAppend (WAsyncT m1) (WAsyncT m2) = WAsyncT $ wAsyncK m1 m2
979instance MonadAsync m => Semigroup (WAsyncT m a) where
982------------------------------------------------------------------------------
984------------------------------------------------------------------------------
986instance MonadAsync m => Monoid (WAsyncT m a) where
987 mempty = WAsyncT K.nil
990------------------------------------------------------------------------------
992------------------------------------------------------------------------------
994{-# INLINE apWAsync #-}
995{-# SPECIALIZE apWAsync ::
996 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
997apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
998apWAsync (WAsyncT m1) (WAsyncT m2) =
999 let f x1 = K.concatMapWith wAsyncK (pure . x1) m2
1000 in WAsyncT $ K.concatMapWith wAsyncK f m1
1002-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1003-- performance degradation (~2x).
1004instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1005 pure = WAsyncT . K.fromPure
1008------------------------------------------------------------------------------
1010------------------------------------------------------------------------------
1012-- GHC: if we change the implementation of bindWith with arguments in a
1013-- different order we see a significant performance degradation (~2x).
1014{-# INLINE bindWAsync #-}
1015{-# SPECIALIZE bindWAsync ::
1016 WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1017bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1018bindWAsync (WAsyncT m) f = WAsyncT $ K.bindWith wAsyncK m (getWAsyncT . f)
1020-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1021-- performance degradation (~2x).
1022instance MonadAsync m => Monad (WAsyncT m) where
1026------------------------------------------------------------------------------
1028------------------------------------------------------------------------------
1030MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list