1{-# LANGUAGE UndecidableInstances #-}
2{-# OPTIONS_GHC -fno-warn-deprecations #-}
7-- Module : Streamly.Internal.Data.Stream.Async
8-- Copyright : (c) 2017 Composewell Technologies
11-- Maintainer : streamly@composewell.com
12-- Stability : experimental
15-- To run examples in this module:
17-- >>> import qualified Streamly.Prelude as Stream
18-- >>> import Control.Concurrent (threadDelay)
21-- threadDelay (n * 1000000) -- sleep for n seconds
22-- putStrLn (show n ++ " sec") -- print "n sec"
26module Streamly.Internal.Data.Stream.Async {-# DEPRECATED "Please use \"Streamly.Internal.Data.Stream.Concurrent\" from streamly package instead." #-}
42import Control.Concurrent (myThreadId)
43import Control.Monad.Base (MonadBase(..), liftBaseDefault)
44import Control.Monad.Catch (MonadThrow, throwM)
45import Control.Concurrent.MVar (newEmptyMVar)
46-- import Control.Monad.Error.Class (MonadError(..))
47import Control.Monad.IO.Class (MonadIO(..))
48import Control.Monad.Reader.Class (MonadReader(..))
49import Control.Monad.State.Class (MonadState(..))
50import Control.Monad.Trans.Class (MonadTrans(lift))
51import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)
52import Data.IORef (IORef, newIORef, readIORef)
53import Data.Maybe (fromJust)
55import Prelude hiding (map)
56import qualified Data.Set as S
58import Streamly.Internal.Control.Concurrent
59 (MonadRunInIO, MonadAsync, askRunInIO, restoreM)
60import Streamly.Internal.Data.Atomics
61 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
62import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
63import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
65import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
66 (foldStreamShared, mkStream, foldStream, fromEffect
67 , nil, concatMapWith, fromPure, bindWith)
68import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
69 (Stream(..), Step(..), mapM, toStreamK, fromStreamK)
70import qualified Streamly.Internal.Data.Stream.Serial as Stream (toStreamK)
72import Streamly.Internal.Data.SVar
74#include "Instances.hs"
77-- >>> :set -fno-warn-deprecations
78-- >>> import qualified Streamly.Prelude as Stream
79-- >>> import Control.Concurrent (threadDelay)
82-- threadDelay (n * 1000000) -- sleep for n seconds
83-- putStrLn (show n ++ " sec") -- print "n sec"
88{-# INLINABLE withLocal #-}
89withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
91 K.mkStream $ \st yld sng stp ->
92 let single = local f . sng
93 yieldk a r = local f $ yld a (withLocal f r)
94 in K.foldStream st yieldk single (local f stp) m
96-------------------------------------------------------------------------------
98-------------------------------------------------------------------------------
100-- Note: For purely right associated expressions this queue should have at most
101-- one element. It grows to more than one when we have left associcated
102-- expressions. Large left associated compositions can grow this to a
104{-# INLINE enqueueLIFO #-}
106 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
107enqueueLIFO sv q m = do
108 atomicModifyIORefCAS_ q $ \ms -> m : ms
111data WorkerStatus = Continue | Suspend
113{-# INLINE workLoopLIFO #-}
116 => IORef [(RunInIO m, Stream m a)]
121workLoopLIFO q st sv winfo = run
125 stop = liftIO $ sendStop sv winfo
130 Just (RunInIO runin, m) -> do
131 -- XXX when we finish we need to send the monadic state back to
132 -- the parent so that the state can be merged back. We capture
133 -- and return the state in the stop continuation.
135 -- Instead of using the run function we can just restore the
136 -- monad state here. That way it can work easily for
137 -- distributed case as well.
138 r <- liftIO $ runin $
139 K.foldStreamShared st yieldk single (return Continue) m
146 res <- liftIO $ sendYield sv winfo (ChildYield a)
147 return $ if res then Continue else Suspend
150 res <- liftIO $ sendYield sv winfo (ChildYield a)
152 then K.foldStreamShared st yieldk single (return Continue) r
154 runInIO <- askRunInIO
155 liftIO $ enqueueLIFO sv q (runInIO, r)
158 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
160 x : xs -> (xs, Just x)
162-- We duplicate workLoop for yield limit and no limit cases because it has
163-- around 40% performance overhead in the worst case.
165-- XXX we can pass yinfo directly as an argument here so that we do not have to
166-- make a check every time.
167{-# INLINE workLoopLIFOLimited #-}
169 :: forall m a. MonadRunInIO m
170 => IORef [(RunInIO m, Stream m a)]
175workLoopLIFOLimited q st sv winfo = run
179 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
180 stop = liftIO $ sendStop sv winfo
185 Just (RunInIO runin, m) -> do
186 -- XXX This is just a best effort minimization of concurrency
187 -- to the yield limit. If the stream is made of concurrent
188 -- streams we do not reserve the yield limit in the constituent
189 -- streams before executing the action. This can be done
190 -- though, by sharing the yield limit ref with downstream
191 -- actions via state passing. Just a todo.
192 yieldLimitOk <- liftIO $ decrementYieldLimit sv
195 r <- liftIO $ runin $
196 K.foldStreamShared st yieldk single incrContinue m
201 -- Avoid any side effects, undo the yield limit decrement if we
202 -- never yielded anything.
204 enqueueLIFO sv q (RunInIO runin, m)
205 incrementYieldLimit sv
209 res <- liftIO $ sendYield sv winfo (ChildYield a)
210 return $ if res then Continue else Suspend
212 -- XXX can we pass on the yield limit downstream to limit the concurrency
213 -- of constituent streams.
215 res <- liftIO $ sendYield sv winfo (ChildYield a)
216 yieldLimitOk <- liftIO $ decrementYieldLimit sv
217 if res && yieldLimitOk
218 then K.foldStreamShared st yieldk single incrContinue r
220 runInIO <- askRunInIO
221 liftIO $ incrementYieldLimit sv
222 liftIO $ enqueueLIFO sv q (runInIO, r)
225 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
227 x : xs -> (xs, Just x)
229-------------------------------------------------------------------------------
231-------------------------------------------------------------------------------
233-- XXX we can use the Ahead style sequence/heap mechanism to make the best
234-- effort to always try to finish the streams on the left side of an expression
235-- first as long as possible.
237{-# INLINE enqueueFIFO #-}
240 -> LinkedQueue (RunInIO m, t m a)
241 -> (RunInIO m, t m a)
243enqueueFIFO sv q m = do
247-- XXX we can remove sv as it is derivable from st
249{-# INLINE workLoopFIFO #-}
252 => LinkedQueue (RunInIO m, Stream m a)
257workLoopFIFO q st sv winfo = run
261 stop = liftIO $ sendStop sv winfo
263 work <- liftIO $ tryPopR q
266 Just (RunInIO runin, m) -> do
267 r <- liftIO $ runin $
268 K.foldStreamShared st yieldk single (return Continue) m
275 res <- liftIO $ sendYield sv winfo (ChildYield a)
276 return $ if res then Continue else Suspend
278 -- XXX in general we would like to yield "n" elements from a single stream
279 -- before moving on to the next. Single element granularity could be too
280 -- expensive in certain cases. Similarly, we can use time limit for
283 res <- liftIO $ sendYield sv winfo (ChildYield a)
284 runInIO <- askRunInIO
285 liftIO $ enqueueFIFO sv q (runInIO, r)
286 return $ if res then Continue else Suspend
288{-# INLINE workLoopFIFOLimited #-}
290 :: forall m a. MonadRunInIO m
291 => LinkedQueue (RunInIO m, Stream m a)
296workLoopFIFOLimited q st sv winfo = run
300 stop = liftIO $ sendStop sv winfo
301 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
303 work <- liftIO $ tryPopR q
306 Just (RunInIO runin, m) -> do
307 yieldLimitOk <- liftIO $ decrementYieldLimit sv
310 r <- liftIO $ runin $
311 K.foldStreamShared st yieldk single incrContinue m
317 enqueueFIFO sv q (RunInIO runin, m)
318 incrementYieldLimit sv
322 res <- liftIO $ sendYield sv winfo (ChildYield a)
323 return $ if res then Continue else Suspend
326 res <- liftIO $ sendYield sv winfo (ChildYield a)
327 runInIO <- askRunInIO
328 liftIO $ enqueueFIFO sv q (runInIO, r)
329 yieldLimitOk <- liftIO $ decrementYieldLimit sv
330 if res && yieldLimitOk
333 incrementYieldLimit sv
336-------------------------------------------------------------------------------
338-- This code belongs in SVar.hs but is kept here for perf reasons
339-------------------------------------------------------------------------------
341-- XXX we have this function in this file because passing runStreamLIFO as a
342-- function argument to this function results in a perf degradation of more
343-- than 10%. Need to investigate what the root cause is.
344-- Interestingly, the same thing does not make any difference for Ahead.
345getLifoSVar :: forall m a. MonadAsync m
346 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
347getLifoSVar st mrun = do
348 outQ <- newIORef ([], 0)
349 outQMv <- newEmptyMVar
351 wfw <- newIORef False
352 running <- newIORef S.empty
353 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
354 yl <- case getYieldLimit st of
355 Nothing -> return Nothing
356 Just x -> Just <$> newIORef x
357 rateInfo <- getYieldRateInfo st
359 stats <- newSVarStats
362 let isWorkFinished _ = null <$> readIORef q
364 let isWorkFinishedLimited sv = do
366 case remainingWork sv of
370 Nothing -> return False
371 qEmpty <- null <$> readIORef q
372 return $ qEmpty || yieldsDone
374 let getSVar :: SVar Stream m a
375 -> (SVar Stream m a -> m [ChildEvent a])
376 -> (SVar Stream m a -> m Bool)
377 -> (SVar Stream m a -> IO Bool)
378 -> (IORef [(RunInIO m, Stream m a)]
384 getSVar sv readOutput postProc workDone wloop = SVar
386 , outputQueueFromConsumer = undefined
388 , maxBufferLimit = getMaxBuffer st
389 , pushBufferSpace = undefined
390 , pushBufferPolicy = undefined
391 , pushBufferMVar = undefined
392 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
393 , yieldRateInfo = rateInfo
394 , outputDoorBell = outQMv
395 , outputDoorBellFromConsumer = undefined
396 , readOutputQ = readOutput sv
397 , postProcess = postProc sv
398 , workerThreads = running
399 , workLoop = wloop q st{streamVar = Just sv} sv
400 , enqueue = enqueueLIFO sv q
401 , isWorkDone = workDone sv
402 , isQueueDone = workDone sv
404 , svarStyle = AsyncVar
405 , svarStopStyle = StopNone
406 , svarStopBy = undefined
408 , workerCount = active
409 , accountThread = delThread sv
410 , workerStopMVar = undefined
412 , svarInspectMode = getInspectMode st
414 , aheadWorkQueue = undefined
415 , outputHeap = undefined
420 case getStreamRate st of
422 case getYieldLimit st of
423 Nothing -> getSVar sv readOutputQBounded
427 Just _ -> getSVar sv readOutputQBounded
429 isWorkFinishedLimited
432 case getYieldLimit st of
433 Nothing -> getSVar sv readOutputQPaced
437 Just _ -> getSVar sv readOutputQPaced
439 isWorkFinishedLimited
443getFifoSVar :: forall m a. MonadAsync m
444 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
445getFifoSVar st mrun = do
446 outQ <- newIORef ([], 0)
447 outQMv <- newEmptyMVar
449 wfw <- newIORef False
450 running <- newIORef S.empty
452 yl <- case getYieldLimit st of
453 Nothing -> return Nothing
454 Just x -> Just <$> newIORef x
455 rateInfo <- getYieldRateInfo st
457 stats <- newSVarStats
460 let isWorkFinished _ = nullQ q
461 let isWorkFinishedLimited sv = do
463 case remainingWork sv of
467 Nothing -> return False
469 return $ qEmpty || yieldsDone
471 let getSVar :: SVar Stream m a
472 -> (SVar Stream m a -> m [ChildEvent a])
473 -> (SVar Stream m a -> m Bool)
474 -> (SVar Stream m a -> IO Bool)
475 -> (LinkedQueue (RunInIO m, Stream m a)
481 getSVar sv readOutput postProc workDone wloop = SVar
483 , outputQueueFromConsumer = undefined
485 , maxBufferLimit = getMaxBuffer st
486 , pushBufferSpace = undefined
487 , pushBufferPolicy = undefined
488 , pushBufferMVar = undefined
489 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
490 , yieldRateInfo = rateInfo
491 , outputDoorBell = outQMv
492 , outputDoorBellFromConsumer = undefined
493 , readOutputQ = readOutput sv
494 , postProcess = postProc sv
495 , workerThreads = running
496 , workLoop = wloop q st{streamVar = Just sv} sv
497 , enqueue = enqueueFIFO sv q
498 , isWorkDone = workDone sv
499 , isQueueDone = workDone sv
501 , svarStyle = WAsyncVar
502 , svarStopStyle = StopNone
503 , svarStopBy = undefined
505 , workerCount = active
506 , accountThread = delThread sv
507 , workerStopMVar = undefined
509 , svarInspectMode = getInspectMode st
511 , aheadWorkQueue = undefined
512 , outputHeap = undefined
517 case getStreamRate st of
519 case getYieldLimit st of
520 Nothing -> getSVar sv readOutputQBounded
524 Just _ -> getSVar sv readOutputQBounded
526 isWorkFinishedLimited
529 case getYieldLimit st of
530 Nothing -> getSVar sv readOutputQPaced
534 Just _ -> getSVar sv readOutputQPaced
536 isWorkFinishedLimited
540{-# INLINABLE newAsyncVar #-}
541newAsyncVar :: MonadAsync m
542 => State Stream m a -> Stream m a -> m (SVar Stream m a)
545 sv <- liftIO $ getLifoSVar st mrun
548-- | Generate a stream asynchronously to keep it buffered, lazily consume
553{-# INLINABLE mkAsyncK #-}
554mkAsyncK :: MonadAsync m => Stream m a -> Stream m a
555mkAsyncK m = K.mkStream $ \st yld sng stp -> do
556 sv <- newAsyncVar (adaptState st) m
557 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
560-- This is slightly faster than the CPS version above
562{-# INLINE_NORMAL mkAsyncD #-}
563mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
564mkAsyncD m = D.Stream step Nothing
567 step gst Nothing = do
568 sv <- newAsyncVar gst (D.toStreamK m)
569 return $ D.Skip $ Just $ fromSVarD sv
571 step gst (Just (D.UnStream step1 st)) = do
574 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
575 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
578-- XXX We can pass captureMonadState instead of using MonadAsync
579-- | Create a new SVar and enqueue one stream computation on it.
580{-# INLINABLE newWAsyncVar #-}
581newWAsyncVar :: MonadAsync m
582 => State Stream m a -> Stream m a -> m (SVar Stream m a)
583newWAsyncVar st m = do
585 sv <- liftIO $ getFifoSVar st mrun
586 -- XXX Use just Stream and IO in all the functions below
587 -- XXX pass mrun instead of calling captureMonadState again inside it
590------------------------------------------------------------------------------
591-- Running streams concurrently
592------------------------------------------------------------------------------
594-- Concurrency rate control.
596-- Our objective is to create more threads on demand if the consumer is running
597-- faster than us. As soon as we encounter a concurrent composition we create a
598-- push pull pair of threads. We use an SVar for communication between the
599-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
600-- The producer creates more threads if the SVar drains and becomes empty, that
601-- is the consumer is running faster.
603-- XXX Note 1: This mechanism can be problematic if the initial production
604-- latency is high, we may end up creating too many threads. So we need some
605-- way to monitor and use the latency as well. Having a limit on the dispatches
606-- (programmer controlled) may also help.
608-- TBD Note 2: We may want to run computations at the lower level of the
609-- composition tree serially even when they are composed using a parallel
610-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
611-- place of 'wAsync'. If we find that an SVar immediately above a computation
612-- gets drained empty we can switch to parallelizing the computation. For that
613-- we can use a state flag to fork the rest of the computation at any point of
614-- time inside the Monad bind operation if the consumer is running at a faster
617-- TBD Note 3: the binary operation ('parallel') composition allows us to
618-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
619-- chunksizes we will need to compose the parallel actions using a data
620-- constructor (A Free container) instead so that we can divide it in chunks of
621-- arbitrary size before dispatching. If the stream is composed of
622-- hierarchically composed grains of different sizes then we can always switch
623-- to a desired granularity depending on the consumer speed.
625-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
626-- into just the number of CPUs.
628-- | Join two computations on the currently running 'SVar' queue for concurrent
629-- execution. When we are using parallel composition, an SVar is passed around
630-- as a state variable. We try to schedule a new parallel computation on the
631-- SVar passed to us. The first time, when no SVar exists, a new SVar is
632-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
633-- already scheduled on the SVar is further evaluated. For example, when (a
634-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
635-- the current scheduler queue.
637-- The 'SVarStyle' required by the current composition context is passed as one
638-- of the parameters. If the scheduling and composition style of the new
639-- computation being scheduled is different than the style of the current SVar,
640-- then we create a new SVar and schedule it on that. The newly created SVar
641-- joins as one of the computations on the current SVar queue.
643-- Cases when we need to switch to a new SVar:
645-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
646-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
647-- because of the scheduling policy change.
648-- * if we 'adapt' a stream of type 'async' to a stream of type
649-- 'Parallel', we create a new SVar at the transitioning bind.
650-- * When the stream is switching from disjunctive composition to conjunctive
651-- composition and vice-versa we create a new SVar to isolate the scheduling
654forkSVarAsync :: MonadAsync m
655 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
656forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do
658 AsyncVar -> newAsyncVar st (concurrently m1 m2)
659 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
660 _ -> error "illegal svar type"
661 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
663 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
664 runInIO <- askRunInIO
665 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
666 K.foldStreamShared st yld sng stp ma
668{-# INLINE joinStreamVarAsync #-}
669joinStreamVarAsync :: MonadAsync m
670 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
671joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->
673 Just sv | svarStyle sv == style -> do
674 runInIO <- askRunInIO
675 liftIO $ enqueue sv (runInIO, m2)
676 K.foldStreamShared st yld sng stp m1
677 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
679------------------------------------------------------------------------------
680-- Semigroup and Monoid style compositions for parallel actions
681------------------------------------------------------------------------------
684asyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
685asyncK = joinStreamVarAsync AsyncVar
687-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
688-- huge difference in the performance of consM in IsStream instance even we
689-- have a SPECIALIZE in the instance.
691-- | XXX we can implement it more efficienty by directly implementing instead
692-- of combining streams using async.
693{-# INLINE consMAsync #-}
694{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
695consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
696consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r
698------------------------------------------------------------------------------
700------------------------------------------------------------------------------
702-- | For 'AsyncT' streams:
705-- (<>) = 'Streamly.Prelude.async'
706-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
709-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
710-- executed concurrently a la the 'async' combinator, producing results and
711-- side effects of iterations out of order:
714-- Stream.toList $ Stream.fromAsync $ do
715-- x <- Stream.fromList [2,1] -- foreach x in stream
716-- Stream.fromEffect $ delay x
722-- Nested monad binds behave like nested @for@ loops with nested iterations
723-- executed concurrently, a la the 'async' combinator:
726-- Stream.toList $ Stream.fromAsync $ do
727-- x <- Stream.fromList [1,2] -- foreach x in stream
728-- y <- Stream.fromList [2,4] -- foreach y in stream
729-- Stream.fromEffect $ delay (x + y)
737-- The behavior can be explained as follows. All the iterations corresponding
738-- to the element @1@ in the first stream constitute one output stream and all
739-- the iterations corresponding to @2@ constitute another output stream and
740-- these two output streams are merged using 'async'.
742-- /Since: 0.1.0 ("Streamly")/
745newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
747instance MonadTrans AsyncT where
749 lift = AsyncT . K.fromEffect
751-- | A demand driven left biased parallely composing IO stream of elements of
752-- type @a@. See 'AsyncT' documentation for more details.
754-- /Since: 0.2.0 ("Streamly")/
757type Async = AsyncT IO
759------------------------------------------------------------------------------
761------------------------------------------------------------------------------
763-- Monomorphically typed version of "async" for better performance of Semigroup
766{-# SPECIALIZE append :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
767append :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
768append (AsyncT m1) (AsyncT m2) = AsyncT $ asyncK m1 m2
770instance MonadAsync m => Semigroup (AsyncT m a) where
773------------------------------------------------------------------------------
775------------------------------------------------------------------------------
777instance MonadAsync m => Monoid (AsyncT m a) where
778 mempty = AsyncT K.nil
781------------------------------------------------------------------------------
783------------------------------------------------------------------------------
785{-# INLINE apAsync #-}
786{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
787apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
788apAsync (AsyncT m1) (AsyncT m2) =
789 let f x1 = K.concatMapWith asyncK (K.fromPure . x1) m2
790 in AsyncT $ K.concatMapWith asyncK f m1
792instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
794 pure = AsyncT . K.fromPure
799------------------------------------------------------------------------------
801------------------------------------------------------------------------------
803-- GHC: if we change the implementation of bindWith with arguments in a
804-- different order we see a significant performance degradation (~2x).
805{-# INLINE bindAsync #-}
806{-# SPECIALIZE bindAsync ::
807 AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
808bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
809bindAsync (AsyncT m) f = AsyncT $ K.bindWith asyncK m (getAsyncT . f)
811-- GHC: if we specify arguments in the definition of (>>=) we see a significant
812-- performance degradation (~2x).
813instance MonadAsync m => Monad (AsyncT m) where
817------------------------------------------------------------------------------
819------------------------------------------------------------------------------
821MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
823------------------------------------------------------------------------------
825------------------------------------------------------------------------------
827{-# INLINE wAsyncK #-}
828wAsyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
829wAsyncK = joinStreamVarAsync WAsyncVar
831-- | XXX we can implement it more efficienty by directly implementing instead
832-- of combining streams using wAsync.
833{-# INLINE consMWAsync #-}
834{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
835consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
836consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r
838-- | For 'WAsyncT' streams:
841-- (<>) = 'Streamly.Prelude.wAsync'
842-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
845-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
846-- executed concurrently a la the 'wAsync' combinator, producing results and
847-- side effects of iterations out of order:
850-- Stream.toList $ Stream.fromWAsync $ do
851-- x <- Stream.fromList [2,1] -- foreach x in stream
852-- Stream.fromEffect $ delay x
858-- Nested monad binds behave like nested @for@ loops with nested iterations
859-- executed concurrently, a la the 'wAsync' combinator:
862-- Stream.toList $ Stream.fromWAsync $ do
863-- x <- Stream.fromList [1,2] -- foreach x in stream
864-- y <- Stream.fromList [2,4] -- foreach y in stream
865-- Stream.fromEffect $ delay (x + y)
873-- The behavior can be explained as follows. All the iterations corresponding
874-- to the element @1@ in the first stream constitute one 'WAsyncT' output
875-- stream and all the iterations corresponding to @2@ constitute another
876-- 'WAsyncT' output stream and these two output streams are merged using
879-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
880-- contrast to the depth wise scheduling behavior of 'AsyncT'.
882-- /Since: 0.2.0 ("Streamly")/
886-- XXX This documentation is redundant, need to be cleaned up/removed.
889-- import qualified "Streamly.Prelude" as S
890-- import Control.Concurrent
892-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
898-- For this example, we are using @maxThreads 1@ so that concurrent thread
899-- scheduling does not affect the results and make them unpredictable. Let's
900-- now take a more general example:
903-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
906-- [1,4,2,7,5,3,8,6,9]
909-- This is how the execution of the above stream proceeds:
911-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
912-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
913-- queue is represented by the rightmost item.
914-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
915-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
916-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
917-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
918-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
919-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
920-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
921-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
922-- 4. Note that the scheduler queue expands by one more stream component in
923-- every pass because one more @<>@ is broken down into two components. At this
924-- point there are no more @<>@ operations to be broken down further and the
925-- queue has reached its maximum size. Now these streams are scheduled in
926-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
928-- As we see above, in a right associated expression composed with @<>@, only
929-- one @<>@ operation is broken down into two components in one execution,
930-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
931-- scheduler passes to expand the whole expression. By the time @n-th@
932-- component is added to the scheduler queue, the first component would have
933-- received @n@ scheduler passes.
935-- Since all streams get interleaved, this operation is not suitable for
936-- folding an infinite lazy container of infinite size streams. However, if
937-- the streams are small, the streams on the left may get finished before more
938-- streams are added to the scheduler queue from the right side of the
939-- expression, so it may be possible to fold an infinite lazy container of
940-- streams. For example, if the streams are of size @n@ then at most @n@
941-- streams would be in the scheduler queue at a time.
943-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
944-- therefore the output of 'WAsyncT' even with a single thread of execution is
945-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
946-- its scheduling behavior.
948-- Any exceptions generated by a constituent stream are propagated to the
949-- output stream. The output and exceptions from a single stream are guaranteed
950-- to arrive in the same order in the resulting stream as they were generated
951-- in the input stream. However, the relative ordering of elements from
952-- different streams in the resulting stream can vary depending on scheduling
953-- and generation delays.
955-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
956-- concurrently using a round robin scheduling.
959-- main = S.'drain' . S.'fromWAsync' $ do
960-- n <- return 3 \<\> return 2 \<\> return 1
962-- threadDelay (n * 1000000)
963-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
966-- ThreadId 40: Delay 1
967-- ThreadId 39: Delay 2
968-- ThreadId 38: Delay 3
971newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
973instance MonadTrans WAsyncT where
975 lift = WAsyncT . K.fromEffect
977-- | A round robin parallely composing IO stream of elements of type @a@.
978-- See 'WAsyncT' documentation for more details.
980-- /Since: 0.2.0 ("Streamly")/
983type WAsync = WAsyncT IO
985------------------------------------------------------------------------------
987------------------------------------------------------------------------------
989{-# INLINE wAppend #-}
990{-# SPECIALIZE wAppend :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
991wAppend :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
992wAppend (WAsyncT m1) (WAsyncT m2) = WAsyncT $ wAsyncK m1 m2
994instance MonadAsync m => Semigroup (WAsyncT m a) where
997------------------------------------------------------------------------------
999------------------------------------------------------------------------------
1001instance MonadAsync m => Monoid (WAsyncT m a) where
1002 mempty = WAsyncT K.nil
1005------------------------------------------------------------------------------
1007------------------------------------------------------------------------------
1009{-# INLINE apWAsync #-}
1010{-# SPECIALIZE apWAsync ::
1011 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
1012apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
1013apWAsync (WAsyncT m1) (WAsyncT m2) =
1014 let f x1 = K.concatMapWith wAsyncK (K.fromPure . x1) m2
1015 in WAsyncT $ K.concatMapWith wAsyncK f m1
1017-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1018-- performance degradation (~2x).
1019instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1020 pure = WAsyncT . K.fromPure
1023------------------------------------------------------------------------------
1025------------------------------------------------------------------------------
1027-- GHC: if we change the implementation of bindWith with arguments in a
1028-- different order we see a significant performance degradation (~2x).
1029{-# INLINE bindWAsync #-}
1030{-# SPECIALIZE bindWAsync ::
1031 WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1032bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1033bindWAsync (WAsyncT m) f = WAsyncT $ K.bindWith wAsyncK m (getWAsyncT . f)
1035-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1036-- performance degradation (~2x).
1037instance MonadAsync m => Monad (WAsyncT m) where
1041------------------------------------------------------------------------------
1043------------------------------------------------------------------------------
1045MONAD_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