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 {-# DEPRECATED "Please use \"Streamly.Internal.Data.Stream.Concurrent\" instead." #-}
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)
54import Prelude hiding (map)
55import qualified Data.Set as S
57import Streamly.Internal.Control.Concurrent
58 (MonadRunInIO, MonadAsync, askRunInIO, restoreM)
59import Streamly.Internal.Data.Atomics
60 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
61import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
62import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
64import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
65 (foldStreamShared, mkStream, foldStream, fromEffect
66 , nil, concatMapWith, fromPure, bindWith)
67import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
68 (Stream(..), Step(..), mapM, toStreamK, fromStreamK)
69import qualified Streamly.Internal.Data.Stream as Stream (toStreamK)
71import Streamly.Internal.Data.SVar
73#include "Instances.hs"
76-- >>> :set -fno-warn-deprecations
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{-# INLINABLE withLocal #-}
88withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
90 K.mkStream $ \st yld sng stp ->
91 let single = local f . sng
92 yieldk a r = local f $ yld a (withLocal f r)
93 in K.foldStream st yieldk single (local f stp) m
95-------------------------------------------------------------------------------
97-------------------------------------------------------------------------------
99-- Note: For purely right associated expressions this queue should have at most
100-- one element. It grows to more than one when we have left associcated
101-- expressions. Large left associated compositions can grow this to a
103{-# INLINE enqueueLIFO #-}
105 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
106enqueueLIFO sv q m = do
107 atomicModifyIORefCAS_ q $ \ms -> m : ms
110data WorkerStatus = Continue | Suspend
112{-# INLINE workLoopLIFO #-}
115 => IORef [(RunInIO m, Stream m a)]
120workLoopLIFO q st sv winfo = run
124 stop = liftIO $ sendStop sv winfo
129 Just (RunInIO runin, m) -> do
130 -- XXX when we finish we need to send the monadic state back to
131 -- the parent so that the state can be merged back. We capture
132 -- and return the state in the stop continuation.
134 -- Instead of using the run function we can just restore the
135 -- monad state here. That way it can work easily for
136 -- distributed case as well.
137 r <- liftIO $ runin $
138 K.foldStreamShared st yieldk single (return Continue) m
145 res <- liftIO $ sendYield sv winfo (ChildYield a)
146 return $ if res then Continue else Suspend
149 res <- liftIO $ sendYield sv winfo (ChildYield a)
151 then K.foldStreamShared st yieldk single (return Continue) r
153 runInIO <- askRunInIO
154 liftIO $ enqueueLIFO sv q (runInIO, r)
157 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
159 x : xs -> (xs, Just x)
161-- We duplicate workLoop for yield limit and no limit cases because it has
162-- around 40% performance overhead in the worst case.
164-- XXX we can pass yinfo directly as an argument here so that we do not have to
165-- make a check every time.
166{-# INLINE workLoopLIFOLimited #-}
168 :: forall m a. MonadRunInIO m
169 => IORef [(RunInIO m, Stream m a)]
174workLoopLIFOLimited q st sv winfo = run
178 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
179 stop = liftIO $ sendStop sv winfo
184 Just (RunInIO runin, m) -> do
185 -- XXX This is just a best effort minimization of concurrency
186 -- to the yield limit. If the stream is made of concurrent
187 -- streams we do not reserve the yield limit in the constituent
188 -- streams before executing the action. This can be done
189 -- though, by sharing the yield limit ref with downstream
190 -- actions via state passing. Just a todo.
191 yieldLimitOk <- liftIO $ decrementYieldLimit sv
194 r <- liftIO $ runin $
195 K.foldStreamShared st yieldk single incrContinue m
200 -- Avoid any side effects, undo the yield limit decrement if we
201 -- never yielded anything.
203 enqueueLIFO sv q (RunInIO runin, m)
204 incrementYieldLimit sv
208 res <- liftIO $ sendYield sv winfo (ChildYield a)
209 return $ if res then Continue else Suspend
211 -- XXX can we pass on the yield limit downstream to limit the concurrency
212 -- of constituent streams.
214 res <- liftIO $ sendYield sv winfo (ChildYield a)
215 yieldLimitOk <- liftIO $ decrementYieldLimit sv
216 if res && yieldLimitOk
217 then K.foldStreamShared st yieldk single incrContinue r
219 runInIO <- askRunInIO
220 liftIO $ incrementYieldLimit sv
221 liftIO $ enqueueLIFO sv q (runInIO, r)
224 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
226 x : xs -> (xs, Just x)
228-------------------------------------------------------------------------------
230-------------------------------------------------------------------------------
232-- XXX we can use the Ahead style sequence/heap mechanism to make the best
233-- effort to always try to finish the streams on the left side of an expression
234-- first as long as possible.
236{-# INLINE enqueueFIFO #-}
239 -> LinkedQueue (RunInIO m, t m a)
240 -> (RunInIO m, t m a)
242enqueueFIFO sv q m = do
246-- XXX we can remove sv as it is derivable from st
248{-# INLINE workLoopFIFO #-}
251 => LinkedQueue (RunInIO m, Stream m a)
256workLoopFIFO q st sv winfo = run
260 stop = liftIO $ sendStop sv winfo
262 work <- liftIO $ tryPopR q
265 Just (RunInIO runin, m) -> do
266 r <- liftIO $ runin $
267 K.foldStreamShared st yieldk single (return Continue) m
274 res <- liftIO $ sendYield sv winfo (ChildYield a)
275 return $ if res then Continue else Suspend
277 -- XXX in general we would like to yield "n" elements from a single stream
278 -- before moving on to the next. Single element granularity could be too
279 -- expensive in certain cases. Similarly, we can use time limit for
282 res <- liftIO $ sendYield sv winfo (ChildYield a)
283 runInIO <- askRunInIO
284 liftIO $ enqueueFIFO sv q (runInIO, r)
285 return $ if res then Continue else Suspend
287{-# INLINE workLoopFIFOLimited #-}
289 :: forall m a. MonadRunInIO m
290 => LinkedQueue (RunInIO m, Stream m a)
295workLoopFIFOLimited q st sv winfo = run
299 stop = liftIO $ sendStop sv winfo
300 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
302 work <- liftIO $ tryPopR q
305 Just (RunInIO runin, m) -> do
306 yieldLimitOk <- liftIO $ decrementYieldLimit sv
309 r <- liftIO $ runin $
310 K.foldStreamShared st yieldk single incrContinue m
316 enqueueFIFO sv q (RunInIO runin, m)
317 incrementYieldLimit sv
321 res <- liftIO $ sendYield sv winfo (ChildYield a)
322 return $ if res then Continue else Suspend
325 res <- liftIO $ sendYield sv winfo (ChildYield a)
326 runInIO <- askRunInIO
327 liftIO $ enqueueFIFO sv q (runInIO, r)
328 yieldLimitOk <- liftIO $ decrementYieldLimit sv
329 if res && yieldLimitOk
332 incrementYieldLimit sv
335-------------------------------------------------------------------------------
337-- This code belongs in SVar.hs but is kept here for perf reasons
338-------------------------------------------------------------------------------
340-- XXX we have this function in this file because passing runStreamLIFO as a
341-- function argument to this function results in a perf degradation of more
342-- than 10%. Need to investigate what the root cause is.
343-- Interestingly, the same thing does not make any difference for Ahead.
344getLifoSVar :: forall m a. MonadAsync m
345 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
346getLifoSVar st mrun = do
347 outQ <- newIORef ([], 0)
348 outQMv <- newEmptyMVar
350 wfw <- newIORef False
351 running <- newIORef S.empty
352 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
353 yl <- case getYieldLimit st of
354 Nothing -> return Nothing
355 Just x -> Just <$> newIORef x
356 rateInfo <- getYieldRateInfo st
358 stats <- newSVarStats
361 let isWorkFinished _ = null <$> readIORef q
363 let isWorkFinishedLimited sv = do
365 case remainingWork sv of
369 Nothing -> return False
370 qEmpty <- null <$> readIORef q
371 return $ qEmpty || yieldsDone
373 let getSVar :: SVar Stream m a
374 -> (SVar Stream m a -> m [ChildEvent a])
375 -> (SVar Stream m a -> m Bool)
376 -> (SVar Stream m a -> IO Bool)
377 -> (IORef [(RunInIO m, Stream m a)]
383 getSVar sv readOutput postProc workDone wloop = SVar
385 , outputQueueFromConsumer = undefined
387 , maxBufferLimit = getMaxBuffer st
388 , pushBufferSpace = undefined
389 , pushBufferPolicy = undefined
390 , pushBufferMVar = undefined
391 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
392 , yieldRateInfo = rateInfo
393 , outputDoorBell = outQMv
394 , outputDoorBellFromConsumer = undefined
395 , readOutputQ = readOutput sv
396 , postProcess = postProc sv
397 , workerThreads = running
398 , workLoop = wloop q st{streamVar = Just sv} sv
399 , enqueue = enqueueLIFO sv q
400 , isWorkDone = workDone sv
401 , isQueueDone = workDone sv
403 , svarStyle = AsyncVar
404 , svarStopStyle = StopNone
405 , svarStopBy = undefined
407 , workerCount = active
408 , accountThread = delThread sv
409 , workerStopMVar = undefined
411 , svarInspectMode = getInspectMode st
413 , aheadWorkQueue = undefined
414 , outputHeap = undefined
419 case getStreamRate st of
421 case getYieldLimit st of
422 Nothing -> getSVar sv readOutputQBounded
426 Just _ -> getSVar sv readOutputQBounded
428 isWorkFinishedLimited
431 case getYieldLimit st of
432 Nothing -> getSVar sv readOutputQPaced
436 Just _ -> getSVar sv readOutputQPaced
438 isWorkFinishedLimited
442getFifoSVar :: forall m a. MonadAsync m
443 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
444getFifoSVar st mrun = do
445 outQ <- newIORef ([], 0)
446 outQMv <- newEmptyMVar
448 wfw <- newIORef False
449 running <- newIORef S.empty
451 yl <- case getYieldLimit st of
452 Nothing -> return Nothing
453 Just x -> Just <$> newIORef x
454 rateInfo <- getYieldRateInfo st
456 stats <- newSVarStats
459 let isWorkFinished _ = nullQ q
460 let isWorkFinishedLimited sv = do
462 case remainingWork sv of
466 Nothing -> return False
468 return $ qEmpty || yieldsDone
470 let getSVar :: SVar Stream m a
471 -> (SVar Stream m a -> m [ChildEvent a])
472 -> (SVar Stream m a -> m Bool)
473 -> (SVar Stream m a -> IO Bool)
474 -> (LinkedQueue (RunInIO m, Stream m a)
480 getSVar sv readOutput postProc workDone wloop = SVar
482 , outputQueueFromConsumer = undefined
484 , maxBufferLimit = getMaxBuffer st
485 , pushBufferSpace = undefined
486 , pushBufferPolicy = undefined
487 , pushBufferMVar = undefined
488 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
489 , yieldRateInfo = rateInfo
490 , outputDoorBell = outQMv
491 , outputDoorBellFromConsumer = undefined
492 , readOutputQ = readOutput sv
493 , postProcess = postProc sv
494 , workerThreads = running
495 , workLoop = wloop q st{streamVar = Just sv} sv
496 , enqueue = enqueueFIFO sv q
497 , isWorkDone = workDone sv
498 , isQueueDone = workDone sv
500 , svarStyle = WAsyncVar
501 , svarStopStyle = StopNone
502 , svarStopBy = undefined
504 , workerCount = active
505 , accountThread = delThread sv
506 , workerStopMVar = undefined
508 , svarInspectMode = getInspectMode st
510 , aheadWorkQueue = undefined
511 , outputHeap = undefined
516 case getStreamRate st of
518 case getYieldLimit st of
519 Nothing -> getSVar sv readOutputQBounded
523 Just _ -> getSVar sv readOutputQBounded
525 isWorkFinishedLimited
528 case getYieldLimit st of
529 Nothing -> getSVar sv readOutputQPaced
533 Just _ -> getSVar sv readOutputQPaced
535 isWorkFinishedLimited
539{-# INLINABLE newAsyncVar #-}
540newAsyncVar :: MonadAsync m
541 => State Stream m a -> Stream m a -> m (SVar Stream m a)
544 sv <- liftIO $ getLifoSVar st mrun
547-- | Generate a stream asynchronously to keep it buffered, lazily consume
552{-# INLINABLE mkAsyncK #-}
553mkAsyncK :: MonadAsync m => Stream m a -> Stream m a
554mkAsyncK m = K.mkStream $ \st yld sng stp -> do
555 sv <- newAsyncVar (adaptState st) m
556 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
559-- This is slightly faster than the CPS version above
561{-# INLINE_NORMAL mkAsyncD #-}
562mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
563mkAsyncD m = D.Stream step Nothing
566 step gst Nothing = do
567 sv <- newAsyncVar gst (D.toStreamK m)
568 return $ D.Skip $ Just $ fromSVarD sv
570 step gst (Just (D.UnStream step1 st)) = do
573 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
574 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
577-- XXX We can pass captureMonadState instead of using MonadAsync
578-- | Create a new SVar and enqueue one stream computation on it.
579{-# INLINABLE newWAsyncVar #-}
580newWAsyncVar :: MonadAsync m
581 => State Stream m a -> Stream m a -> m (SVar Stream m a)
582newWAsyncVar st m = do
584 sv <- liftIO $ getFifoSVar st mrun
585 -- XXX Use just Stream and IO in all the functions below
586 -- XXX pass mrun instead of calling captureMonadState again inside it
589------------------------------------------------------------------------------
590-- Running streams concurrently
591------------------------------------------------------------------------------
593-- Concurrency rate control.
595-- Our objective is to create more threads on demand if the consumer is running
596-- faster than us. As soon as we encounter a concurrent composition we create a
597-- push pull pair of threads. We use an SVar for communication between the
598-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
599-- The producer creates more threads if the SVar drains and becomes empty, that
600-- is the consumer is running faster.
602-- XXX Note 1: This mechanism can be problematic if the initial production
603-- latency is high, we may end up creating too many threads. So we need some
604-- way to monitor and use the latency as well. Having a limit on the dispatches
605-- (programmer controlled) may also help.
607-- TBD Note 2: We may want to run computations at the lower level of the
608-- composition tree serially even when they are composed using a parallel
609-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
610-- place of 'wAsync'. If we find that an SVar immediately above a computation
611-- gets drained empty we can switch to parallelizing the computation. For that
612-- we can use a state flag to fork the rest of the computation at any point of
613-- time inside the Monad bind operation if the consumer is running at a faster
616-- TBD Note 3: the binary operation ('parallel') composition allows us to
617-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
618-- chunksizes we will need to compose the parallel actions using a data
619-- constructor (A Free container) instead so that we can divide it in chunks of
620-- arbitrary size before dispatching. If the stream is composed of
621-- hierarchically composed grains of different sizes then we can always switch
622-- to a desired granularity depending on the consumer speed.
624-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
625-- into just the number of CPUs.
627-- | Join two computations on the currently running 'SVar' queue for concurrent
628-- execution. When we are using parallel composition, an SVar is passed around
629-- as a state variable. We try to schedule a new parallel computation on the
630-- SVar passed to us. The first time, when no SVar exists, a new SVar is
631-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
632-- already scheduled on the SVar is further evaluated. For example, when (a
633-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
634-- the current scheduler queue.
636-- The 'SVarStyle' required by the current composition context is passed as one
637-- of the parameters. If the scheduling and composition style of the new
638-- computation being scheduled is different than the style of the current SVar,
639-- then we create a new SVar and schedule it on that. The newly created SVar
640-- joins as one of the computations on the current SVar queue.
642-- Cases when we need to switch to a new SVar:
644-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
645-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
646-- because of the scheduling policy change.
647-- * if we 'adapt' a stream of type 'async' to a stream of type
648-- 'Parallel', we create a new SVar at the transitioning bind.
649-- * When the stream is switching from disjunctive composition to conjunctive
650-- composition and vice-versa we create a new SVar to isolate the scheduling
653forkSVarAsync :: MonadAsync m
654 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
655forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do
657 AsyncVar -> newAsyncVar st (concurrently m1 m2)
658 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
659 _ -> error "illegal svar type"
660 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
662 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
663 runInIO <- askRunInIO
664 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
665 K.foldStreamShared st yld sng stp ma
667{-# INLINE joinStreamVarAsync #-}
668joinStreamVarAsync :: MonadAsync m
669 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
670joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->
672 Just sv | svarStyle sv == style -> do
673 runInIO <- askRunInIO
674 liftIO $ enqueue sv (runInIO, m2)
675 K.foldStreamShared st yld sng stp m1
676 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
678------------------------------------------------------------------------------
679-- Semigroup and Monoid style compositions for parallel actions
680------------------------------------------------------------------------------
683asyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
684asyncK = joinStreamVarAsync AsyncVar
686-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
687-- huge difference in the performance of consM in IsStream instance even we
688-- have a SPECIALIZE in the instance.
690-- | XXX we can implement it more efficienty by directly implementing instead
691-- of combining streams using async.
692{-# INLINE consMAsync #-}
693{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
694consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
695consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r
697------------------------------------------------------------------------------
699------------------------------------------------------------------------------
701-- | For 'AsyncT' streams:
704-- (<>) = 'Streamly.Prelude.async'
705-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
708-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
709-- executed concurrently a la the 'async' combinator, producing results and
710-- side effects of iterations out of order:
713-- Stream.toList $ Stream.fromAsync $ do
714-- x <- Stream.fromList [2,1] -- foreach x in stream
715-- Stream.fromEffect $ delay x
721-- Nested monad binds behave like nested @for@ loops with nested iterations
722-- executed concurrently, a la the 'async' combinator:
725-- Stream.toList $ Stream.fromAsync $ do
726-- x <- Stream.fromList [1,2] -- foreach x in stream
727-- y <- Stream.fromList [2,4] -- foreach y in stream
728-- Stream.fromEffect $ delay (x + y)
736-- The behavior can be explained as follows. All the iterations corresponding
737-- to the element @1@ in the first stream constitute one output stream and all
738-- the iterations corresponding to @2@ constitute another output stream and
739-- these two output streams are merged using 'async'.
741-- /Since: 0.1.0 ("Streamly")/
744newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
746instance MonadTrans AsyncT where
748 lift = AsyncT . K.fromEffect
750-- | A demand driven left biased parallely composing IO stream of elements of
751-- type @a@. See 'AsyncT' documentation for more details.
753-- /Since: 0.2.0 ("Streamly")/
756type Async = AsyncT IO
758------------------------------------------------------------------------------
760------------------------------------------------------------------------------
762-- Monomorphically typed version of "async" for better performance of Semigroup
765{-# SPECIALIZE append :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
766append :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
767append (AsyncT m1) (AsyncT m2) = AsyncT $ asyncK m1 m2
769instance MonadAsync m => Semigroup (AsyncT m a) where
772------------------------------------------------------------------------------
774------------------------------------------------------------------------------
776instance MonadAsync m => Monoid (AsyncT m a) where
777 mempty = AsyncT K.nil
780------------------------------------------------------------------------------
782------------------------------------------------------------------------------
784{-# INLINE apAsync #-}
785{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
786apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
787apAsync (AsyncT m1) (AsyncT m2) =
788 let f x1 = K.concatMapWith asyncK (K.fromPure . x1) m2
789 in AsyncT $ K.concatMapWith asyncK f m1
791instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
793 pure = AsyncT . K.fromPure
798------------------------------------------------------------------------------
800------------------------------------------------------------------------------
802-- GHC: if we change the implementation of bindWith with arguments in a
803-- different order we see a significant performance degradation (~2x).
804{-# INLINE bindAsync #-}
805{-# SPECIALIZE bindAsync ::
806 AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
807bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
808bindAsync (AsyncT m) f = AsyncT $ K.bindWith asyncK m (getAsyncT . f)
810-- GHC: if we specify arguments in the definition of (>>=) we see a significant
811-- performance degradation (~2x).
812instance MonadAsync m => Monad (AsyncT m) where
816------------------------------------------------------------------------------
818------------------------------------------------------------------------------
820MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
822------------------------------------------------------------------------------
824------------------------------------------------------------------------------
826{-# INLINE wAsyncK #-}
827wAsyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
828wAsyncK = joinStreamVarAsync WAsyncVar
830-- | XXX we can implement it more efficienty by directly implementing instead
831-- of combining streams using wAsync.
832{-# INLINE consMWAsync #-}
833{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
834consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
835consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r
837-- | For 'WAsyncT' streams:
840-- (<>) = 'Streamly.Prelude.wAsync'
841-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
844-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
845-- executed concurrently a la the 'wAsync' combinator, producing results and
846-- side effects of iterations out of order:
849-- Stream.toList $ Stream.fromWAsync $ do
850-- x <- Stream.fromList [2,1] -- foreach x in stream
851-- Stream.fromEffect $ delay x
857-- Nested monad binds behave like nested @for@ loops with nested iterations
858-- executed concurrently, a la the 'wAsync' combinator:
861-- Stream.toList $ Stream.fromWAsync $ do
862-- x <- Stream.fromList [1,2] -- foreach x in stream
863-- y <- Stream.fromList [2,4] -- foreach y in stream
864-- Stream.fromEffect $ delay (x + y)
872-- The behavior can be explained as follows. All the iterations corresponding
873-- to the element @1@ in the first stream constitute one 'WAsyncT' output
874-- stream and all the iterations corresponding to @2@ constitute another
875-- 'WAsyncT' output stream and these two output streams are merged using
878-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
879-- contrast to the depth wise scheduling behavior of 'AsyncT'.
881-- /Since: 0.2.0 ("Streamly")/
885-- XXX This documentation is redundant, need to be cleaned up/removed.
888-- import qualified "Streamly.Prelude" as S
889-- import Control.Concurrent
891-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
897-- For this example, we are using @maxThreads 1@ so that concurrent thread
898-- scheduling does not affect the results and make them unpredictable. Let's
899-- now take a more general example:
902-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
905-- [1,4,2,7,5,3,8,6,9]
908-- This is how the execution of the above stream proceeds:
910-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
911-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
912-- queue is represented by the rightmost item.
913-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
914-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
915-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
916-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
917-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
918-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
919-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
920-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
921-- 4. Note that the scheduler queue expands by one more stream component in
922-- every pass because one more @<>@ is broken down into two components. At this
923-- point there are no more @<>@ operations to be broken down further and the
924-- queue has reached its maximum size. Now these streams are scheduled in
925-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
927-- As we see above, in a right associated expression composed with @<>@, only
928-- one @<>@ operation is broken down into two components in one execution,
929-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
930-- scheduler passes to expand the whole expression. By the time @n-th@
931-- component is added to the scheduler queue, the first component would have
932-- received @n@ scheduler passes.
934-- Since all streams get interleaved, this operation is not suitable for
935-- folding an infinite lazy container of infinite size streams. However, if
936-- the streams are small, the streams on the left may get finished before more
937-- streams are added to the scheduler queue from the right side of the
938-- expression, so it may be possible to fold an infinite lazy container of
939-- streams. For example, if the streams are of size @n@ then at most @n@
940-- streams would be in the scheduler queue at a time.
942-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
943-- therefore the output of 'WAsyncT' even with a single thread of execution is
944-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
945-- its scheduling behavior.
947-- Any exceptions generated by a constituent stream are propagated to the
948-- output stream. The output and exceptions from a single stream are guaranteed
949-- to arrive in the same order in the resulting stream as they were generated
950-- in the input stream. However, the relative ordering of elements from
951-- different streams in the resulting stream can vary depending on scheduling
952-- and generation delays.
954-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
955-- concurrently using a round robin scheduling.
958-- main = S.'drain' . S.'fromWAsync' $ do
959-- n <- return 3 \<\> return 2 \<\> return 1
961-- threadDelay (n * 1000000)
962-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
965-- ThreadId 40: Delay 1
966-- ThreadId 39: Delay 2
967-- ThreadId 38: Delay 3
970newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
972instance MonadTrans WAsyncT where
974 lift = WAsyncT . K.fromEffect
976-- | A round robin parallely composing IO stream of elements of type @a@.
977-- See 'WAsyncT' documentation for more details.
979-- /Since: 0.2.0 ("Streamly")/
982type WAsync = WAsyncT IO
984------------------------------------------------------------------------------
986------------------------------------------------------------------------------
988{-# INLINE wAppend #-}
989{-# SPECIALIZE wAppend :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
990wAppend :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
991wAppend (WAsyncT m1) (WAsyncT m2) = WAsyncT $ wAsyncK m1 m2
993instance MonadAsync m => Semigroup (WAsyncT m a) where
996------------------------------------------------------------------------------
998------------------------------------------------------------------------------
1000instance MonadAsync m => Monoid (WAsyncT m a) where
1001 mempty = WAsyncT K.nil
1004------------------------------------------------------------------------------
1006------------------------------------------------------------------------------
1008{-# INLINE apWAsync #-}
1009{-# SPECIALIZE apWAsync ::
1010 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
1011apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
1012apWAsync (WAsyncT m1) (WAsyncT m2) =
1013 let f x1 = K.concatMapWith wAsyncK (K.fromPure . x1) m2
1014 in WAsyncT $ K.concatMapWith wAsyncK f m1
1016-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1017-- performance degradation (~2x).
1018instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1019 pure = WAsyncT . K.fromPure
1022------------------------------------------------------------------------------
1024------------------------------------------------------------------------------
1026-- GHC: if we change the implementation of bindWith with arguments in a
1027-- different order we see a significant performance degradation (~2x).
1028{-# INLINE bindWAsync #-}
1029{-# SPECIALIZE bindWAsync ::
1030 WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1031bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1032bindWAsync (WAsyncT m) f = WAsyncT $ K.bindWith wAsyncK m (getWAsyncT . f)
1034-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1035-- performance degradation (~2x).
1036instance MonadAsync m => Monad (WAsyncT m) where
1040------------------------------------------------------------------------------
1042------------------------------------------------------------------------------
1044MONAD_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