1{-# LANGUAGE UndecidableInstances #-}
2{-# OPTIONS_GHC -fno-warn-deprecations #-}
3{-# OPTIONS_GHC -Wno-redundant-constraints #-}
8-- Module : Streamly.Internal.Data.Stream.Async
9-- Copyright : (c) 2017 Composewell Technologies
12-- Maintainer : streamly@composewell.com
13-- Stability : experimental
16-- To run examples in this module:
18-- >>> import qualified Streamly.Prelude as Stream
19-- >>> import Control.Concurrent (threadDelay)
22-- threadDelay (n * 1000000) -- sleep for n seconds
23-- putStrLn (show n ++ " sec") -- print "n sec"
27module Streamly.Internal.Data.Stream.Async {-# DEPRECATED "Please use \"Streamly.Internal.Data.Stream.Concurrent\" from streamly package instead." #-}
43import Control.Concurrent (myThreadId)
44import Control.Monad.Base (MonadBase(..), liftBaseDefault)
45import Control.Monad.Catch (MonadThrow, throwM)
46import Control.Concurrent.MVar (newEmptyMVar)
47-- import Control.Monad.Error.Class (MonadError(..))
48import Control.Monad.IO.Class (MonadIO(..))
49import Control.Monad.Reader.Class (MonadReader(..))
50import Control.Monad.State.Class (MonadState(..))
51import Control.Monad.Trans.Class (MonadTrans(lift))
52import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)
53import Data.IORef (IORef, newIORef, readIORef)
54import Data.Maybe (fromJust)
56import Prelude hiding (map)
57import qualified Data.Set as S
59import Streamly.Internal.Control.Concurrent
60 (MonadRunInIO, MonadAsync, askRunInIO, restoreM)
61import Streamly.Internal.Data.Atomics
62 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
63import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
64import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
66import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
67 (foldStreamShared, mkStream, foldStream, fromEffect
68 , nil, concatMapWith, fromPure, bindWith)
69import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
70 (Stream(..), Step(..), mapM, toStreamK, fromStreamK)
71import qualified Streamly.Internal.Data.Stream.Serial as Stream (toStreamK)
73import Streamly.Internal.Data.SVar
75#include "Instances.hs"
78-- >>> :set -fno-warn-deprecations
79-- >>> import qualified Streamly.Prelude as Stream
80-- >>> import Control.Concurrent (threadDelay)
83-- threadDelay (n * 1000000) -- sleep for n seconds
84-- putStrLn (show n ++ " sec") -- print "n sec"
89{-# INLINABLE withLocal #-}
90withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
92 K.mkStream $ \st yld sng stp ->
93 let single = local f . sng
94 yieldk a r = local f $ yld a (withLocal f r)
95 in K.foldStream st yieldk single (local f stp) m
97-------------------------------------------------------------------------------
99-------------------------------------------------------------------------------
101-- Note: For purely right associated expressions this queue should have at most
102-- one element. It grows to more than one when we have left associcated
103-- expressions. Large left associated compositions can grow this to a
105{-# INLINE enqueueLIFO #-}
107 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
108enqueueLIFO sv q m = do
109 atomicModifyIORefCAS_ q $ \ms -> m : ms
112data WorkerStatus = Continue | Suspend
114{-# INLINE workLoopLIFO #-}
117 => IORef [(RunInIO m, Stream m a)]
122workLoopLIFO q st sv winfo = run
126 stop = liftIO $ sendStop sv winfo
131 Just (RunInIO runin, m) -> do
132 -- XXX when we finish we need to send the monadic state back to
133 -- the parent so that the state can be merged back. We capture
134 -- and return the state in the stop continuation.
136 -- Instead of using the run function we can just restore the
137 -- monad state here. That way it can work easily for
138 -- distributed case as well.
139 r <- liftIO $ runin $
140 K.foldStreamShared st yieldk single (return Continue) m
147 res <- liftIO $ sendYield sv winfo (ChildYield a)
148 return $ if res then Continue else Suspend
151 res <- liftIO $ sendYield sv winfo (ChildYield a)
153 then K.foldStreamShared st yieldk single (return Continue) r
155 runInIO <- askRunInIO
156 liftIO $ enqueueLIFO sv q (runInIO, r)
159 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
161 x : xs -> (xs, Just x)
163-- We duplicate workLoop for yield limit and no limit cases because it has
164-- around 40% performance overhead in the worst case.
166-- XXX we can pass yinfo directly as an argument here so that we do not have to
167-- make a check every time.
168{-# INLINE workLoopLIFOLimited #-}
170 :: forall m a. MonadRunInIO m
171 => IORef [(RunInIO m, Stream m a)]
176workLoopLIFOLimited q st sv winfo = run
180 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
181 stop = liftIO $ sendStop sv winfo
186 Just (RunInIO runin, m) -> do
187 -- XXX This is just a best effort minimization of concurrency
188 -- to the yield limit. If the stream is made of concurrent
189 -- streams we do not reserve the yield limit in the constituent
190 -- streams before executing the action. This can be done
191 -- though, by sharing the yield limit ref with downstream
192 -- actions via state passing. Just a todo.
193 yieldLimitOk <- liftIO $ decrementYieldLimit sv
196 r <- liftIO $ runin $
197 K.foldStreamShared st yieldk single incrContinue m
202 -- Avoid any side effects, undo the yield limit decrement if we
203 -- never yielded anything.
205 enqueueLIFO sv q (RunInIO runin, m)
206 incrementYieldLimit sv
210 res <- liftIO $ sendYield sv winfo (ChildYield a)
211 return $ if res then Continue else Suspend
213 -- XXX can we pass on the yield limit downstream to limit the concurrency
214 -- of constituent streams.
216 res <- liftIO $ sendYield sv winfo (ChildYield a)
217 yieldLimitOk <- liftIO $ decrementYieldLimit sv
218 if res && yieldLimitOk
219 then K.foldStreamShared st yieldk single incrContinue r
221 runInIO <- askRunInIO
222 liftIO $ incrementYieldLimit sv
223 liftIO $ enqueueLIFO sv q (runInIO, r)
226 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
228 x : xs -> (xs, Just x)
230-------------------------------------------------------------------------------
232-------------------------------------------------------------------------------
234-- XXX we can use the Ahead style sequence/heap mechanism to make the best
235-- effort to always try to finish the streams on the left side of an expression
236-- first as long as possible.
238{-# INLINE enqueueFIFO #-}
241 -> LinkedQueue (RunInIO m, t m a)
242 -> (RunInIO m, t m a)
244enqueueFIFO sv q m = do
248-- XXX we can remove sv as it is derivable from st
250{-# INLINE workLoopFIFO #-}
253 => LinkedQueue (RunInIO m, Stream m a)
258workLoopFIFO q st sv winfo = run
262 stop = liftIO $ sendStop sv winfo
264 work <- liftIO $ tryPopR q
267 Just (RunInIO runin, m) -> do
268 r <- liftIO $ runin $
269 K.foldStreamShared st yieldk single (return Continue) m
276 res <- liftIO $ sendYield sv winfo (ChildYield a)
277 return $ if res then Continue else Suspend
279 -- XXX in general we would like to yield "n" elements from a single stream
280 -- before moving on to the next. Single element granularity could be too
281 -- expensive in certain cases. Similarly, we can use time limit for
284 res <- liftIO $ sendYield sv winfo (ChildYield a)
285 runInIO <- askRunInIO
286 liftIO $ enqueueFIFO sv q (runInIO, r)
287 return $ if res then Continue else Suspend
289{-# INLINE workLoopFIFOLimited #-}
291 :: forall m a. MonadRunInIO m
292 => LinkedQueue (RunInIO m, Stream m a)
297workLoopFIFOLimited q st sv winfo = run
301 stop = liftIO $ sendStop sv winfo
302 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
304 work <- liftIO $ tryPopR q
307 Just (RunInIO runin, m) -> do
308 yieldLimitOk <- liftIO $ decrementYieldLimit sv
311 r <- liftIO $ runin $
312 K.foldStreamShared st yieldk single incrContinue m
318 enqueueFIFO sv q (RunInIO runin, m)
319 incrementYieldLimit sv
323 res <- liftIO $ sendYield sv winfo (ChildYield a)
324 return $ if res then Continue else Suspend
327 res <- liftIO $ sendYield sv winfo (ChildYield a)
328 runInIO <- askRunInIO
329 liftIO $ enqueueFIFO sv q (runInIO, r)
330 yieldLimitOk <- liftIO $ decrementYieldLimit sv
331 if res && yieldLimitOk
334 incrementYieldLimit sv
337-------------------------------------------------------------------------------
339-- This code belongs in SVar.hs but is kept here for perf reasons
340-------------------------------------------------------------------------------
342-- XXX we have this function in this file because passing runStreamLIFO as a
343-- function argument to this function results in a perf degradation of more
344-- than 10%. Need to investigate what the root cause is.
345-- Interestingly, the same thing does not make any difference for Ahead.
346getLifoSVar :: forall m a. MonadAsync m
347 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
348getLifoSVar st mrun = do
349 outQ <- newIORef ([], 0)
350 outQMv <- newEmptyMVar
352 wfw <- newIORef False
353 running <- newIORef S.empty
354 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
355 yl <- case getYieldLimit st of
356 Nothing -> return Nothing
357 Just x -> Just <$> newIORef x
358 rateInfo <- getYieldRateInfo st
360 stats <- newSVarStats
363 let isWorkFinished _ = null <$> readIORef q
365 let isWorkFinishedLimited sv = do
367 case remainingWork sv of
371 Nothing -> return False
372 qEmpty <- null <$> readIORef q
373 return $ qEmpty || yieldsDone
375 let getSVar :: SVar Stream m a
376 -> (SVar Stream m a -> m [ChildEvent a])
377 -> (SVar Stream m a -> m Bool)
378 -> (SVar Stream m a -> IO Bool)
379 -> (IORef [(RunInIO m, Stream m a)]
385 getSVar sv readOutput postProc workDone wloop = SVar
387 , outputQueueFromConsumer = undefined
389 , maxBufferLimit = getMaxBuffer st
390 , pushBufferSpace = undefined
391 , pushBufferPolicy = undefined
392 , pushBufferMVar = undefined
393 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
394 , yieldRateInfo = rateInfo
395 , outputDoorBell = outQMv
396 , outputDoorBellFromConsumer = undefined
397 , readOutputQ = readOutput sv
398 , postProcess = postProc sv
399 , workerThreads = running
400 , workLoop = wloop q st{streamVar = Just sv} sv
401 , enqueue = enqueueLIFO sv q
402 , isWorkDone = workDone sv
403 , isQueueDone = workDone sv
405 , svarStyle = AsyncVar
406 , svarStopStyle = StopNone
407 , svarStopBy = undefined
409 , workerCount = active
410 , accountThread = delThread sv
411 , workerStopMVar = undefined
413 , svarInspectMode = getInspectMode st
415 , aheadWorkQueue = undefined
416 , outputHeap = undefined
421 case getStreamRate st of
423 case getYieldLimit st of
424 Nothing -> getSVar sv readOutputQBounded
428 Just _ -> getSVar sv readOutputQBounded
430 isWorkFinishedLimited
433 case getYieldLimit st of
434 Nothing -> getSVar sv readOutputQPaced
438 Just _ -> getSVar sv readOutputQPaced
440 isWorkFinishedLimited
444getFifoSVar :: forall m a. MonadAsync m
445 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
446getFifoSVar st mrun = do
447 outQ <- newIORef ([], 0)
448 outQMv <- newEmptyMVar
450 wfw <- newIORef False
451 running <- newIORef S.empty
453 yl <- case getYieldLimit st of
454 Nothing -> return Nothing
455 Just x -> Just <$> newIORef x
456 rateInfo <- getYieldRateInfo st
458 stats <- newSVarStats
461 let isWorkFinished _ = nullQ q
462 let isWorkFinishedLimited sv = do
464 case remainingWork sv of
468 Nothing -> return False
470 return $ qEmpty || yieldsDone
472 let getSVar :: SVar Stream m a
473 -> (SVar Stream m a -> m [ChildEvent a])
474 -> (SVar Stream m a -> m Bool)
475 -> (SVar Stream m a -> IO Bool)
476 -> (LinkedQueue (RunInIO m, Stream m a)
482 getSVar sv readOutput postProc workDone wloop = SVar
484 , outputQueueFromConsumer = undefined
486 , maxBufferLimit = getMaxBuffer st
487 , pushBufferSpace = undefined
488 , pushBufferPolicy = undefined
489 , pushBufferMVar = undefined
490 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
491 , yieldRateInfo = rateInfo
492 , outputDoorBell = outQMv
493 , outputDoorBellFromConsumer = undefined
494 , readOutputQ = readOutput sv
495 , postProcess = postProc sv
496 , workerThreads = running
497 , workLoop = wloop q st{streamVar = Just sv} sv
498 , enqueue = enqueueFIFO sv q
499 , isWorkDone = workDone sv
500 , isQueueDone = workDone sv
502 , svarStyle = WAsyncVar
503 , svarStopStyle = StopNone
504 , svarStopBy = undefined
506 , workerCount = active
507 , accountThread = delThread sv
508 , workerStopMVar = undefined
510 , svarInspectMode = getInspectMode st
512 , aheadWorkQueue = undefined
513 , outputHeap = undefined
518 case getStreamRate st of
520 case getYieldLimit st of
521 Nothing -> getSVar sv readOutputQBounded
525 Just _ -> getSVar sv readOutputQBounded
527 isWorkFinishedLimited
530 case getYieldLimit st of
531 Nothing -> getSVar sv readOutputQPaced
535 Just _ -> getSVar sv readOutputQPaced
537 isWorkFinishedLimited
541{-# INLINABLE newAsyncVar #-}
542newAsyncVar :: MonadAsync m
543 => State Stream m a -> Stream m a -> m (SVar Stream m a)
546 sv <- liftIO $ getLifoSVar st mrun
549-- | Generate a stream asynchronously to keep it buffered, lazily consume
554{-# INLINABLE mkAsyncK #-}
555mkAsyncK :: MonadAsync m => Stream m a -> Stream m a
556mkAsyncK m = K.mkStream $ \st yld sng stp -> do
557 sv <- newAsyncVar (adaptState st) m
558 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
561-- This is slightly faster than the CPS version above
563{-# INLINE_NORMAL mkAsyncD #-}
564mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
565mkAsyncD m = D.Stream step Nothing
568 step gst Nothing = do
569 sv <- newAsyncVar gst (D.toStreamK m)
570 return $ D.Skip $ Just $ fromSVarD sv
572 step gst (Just (D.UnStream step1 st)) = do
575 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
576 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
579-- XXX We can pass captureMonadState instead of using MonadAsync
580-- | Create a new SVar and enqueue one stream computation on it.
581{-# INLINABLE newWAsyncVar #-}
582newWAsyncVar :: MonadAsync m
583 => State Stream m a -> Stream m a -> m (SVar Stream m a)
584newWAsyncVar st m = do
586 sv <- liftIO $ getFifoSVar st mrun
587 -- XXX Use just Stream and IO in all the functions below
588 -- XXX pass mrun instead of calling captureMonadState again inside it
591------------------------------------------------------------------------------
592-- Running streams concurrently
593------------------------------------------------------------------------------
595-- Concurrency rate control.
597-- Our objective is to create more threads on demand if the consumer is running
598-- faster than us. As soon as we encounter a concurrent composition we create a
599-- push pull pair of threads. We use an SVar for communication between the
600-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
601-- The producer creates more threads if the SVar drains and becomes empty, that
602-- is the consumer is running faster.
604-- XXX Note 1: This mechanism can be problematic if the initial production
605-- latency is high, we may end up creating too many threads. So we need some
606-- way to monitor and use the latency as well. Having a limit on the dispatches
607-- (programmer controlled) may also help.
609-- TBD Note 2: We may want to run computations at the lower level of the
610-- composition tree serially even when they are composed using a parallel
611-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
612-- place of 'wAsync'. If we find that an SVar immediately above a computation
613-- gets drained empty we can switch to parallelizing the computation. For that
614-- we can use a state flag to fork the rest of the computation at any point of
615-- time inside the Monad bind operation if the consumer is running at a faster
618-- TBD Note 3: the binary operation ('parallel') composition allows us to
619-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
620-- chunksizes we will need to compose the parallel actions using a data
621-- constructor (A Free container) instead so that we can divide it in chunks of
622-- arbitrary size before dispatching. If the stream is composed of
623-- hierarchically composed grains of different sizes then we can always switch
624-- to a desired granularity depending on the consumer speed.
626-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
627-- into just the number of CPUs.
629-- | Join two computations on the currently running 'SVar' queue for concurrent
630-- execution. When we are using parallel composition, an SVar is passed around
631-- as a state variable. We try to schedule a new parallel computation on the
632-- SVar passed to us. The first time, when no SVar exists, a new SVar is
633-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
634-- already scheduled on the SVar is further evaluated. For example, when (a
635-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
636-- the current scheduler queue.
638-- The 'SVarStyle' required by the current composition context is passed as one
639-- of the parameters. If the scheduling and composition style of the new
640-- computation being scheduled is different than the style of the current SVar,
641-- then we create a new SVar and schedule it on that. The newly created SVar
642-- joins as one of the computations on the current SVar queue.
644-- Cases when we need to switch to a new SVar:
646-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
647-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
648-- because of the scheduling policy change.
649-- * if we 'adapt' a stream of type 'async' to a stream of type
650-- 'Parallel', we create a new SVar at the transitioning bind.
651-- * When the stream is switching from disjunctive composition to conjunctive
652-- composition and vice-versa we create a new SVar to isolate the scheduling
655forkSVarAsync :: MonadAsync m
656 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
657forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do
659 AsyncVar -> newAsyncVar st (concurrently m1 m2)
660 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
661 _ -> error "illegal svar type"
662 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
664 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
665 runInIO <- askRunInIO
666 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
667 K.foldStreamShared st yld sng stp ma
669{-# INLINE joinStreamVarAsync #-}
670joinStreamVarAsync :: MonadAsync m
671 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
672joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->
674 Just sv | svarStyle sv == style -> do
675 runInIO <- askRunInIO
676 liftIO $ enqueue sv (runInIO, m2)
677 K.foldStreamShared st yld sng stp m1
678 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
680------------------------------------------------------------------------------
681-- Semigroup and Monoid style compositions for parallel actions
682------------------------------------------------------------------------------
685asyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
686asyncK = joinStreamVarAsync AsyncVar
688-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
689-- huge difference in the performance of consM in IsStream instance even we
690-- have a SPECIALIZE in the instance.
692-- | XXX we can implement it more efficienty by directly implementing instead
693-- of combining streams using async.
694{-# INLINE consMAsync #-}
695{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
696consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
697consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r
699------------------------------------------------------------------------------
701------------------------------------------------------------------------------
703-- | For 'AsyncT' streams:
706-- (<>) = 'Streamly.Prelude.async'
707-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
710-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
711-- executed concurrently a la the 'async' combinator, producing results and
712-- side effects of iterations out of order:
715-- Stream.toList $ Stream.fromAsync $ do
716-- x <- Stream.fromList [2,1] -- foreach x in stream
717-- Stream.fromEffect $ delay x
723-- Nested monad binds behave like nested @for@ loops with nested iterations
724-- executed concurrently, a la the 'async' combinator:
727-- Stream.toList $ Stream.fromAsync $ do
728-- x <- Stream.fromList [1,2] -- foreach x in stream
729-- y <- Stream.fromList [2,4] -- foreach y in stream
730-- Stream.fromEffect $ delay (x + y)
738-- The behavior can be explained as follows. All the iterations corresponding
739-- to the element @1@ in the first stream constitute one output stream and all
740-- the iterations corresponding to @2@ constitute another output stream and
741-- these two output streams are merged using 'async'.
743-- /Since: 0.1.0 ("Streamly")/
746newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
748instance MonadTrans AsyncT where
750 lift = AsyncT . K.fromEffect
752-- | A demand driven left biased parallely composing IO stream of elements of
753-- type @a@. See 'AsyncT' documentation for more details.
755-- /Since: 0.2.0 ("Streamly")/
758type Async = AsyncT IO
760------------------------------------------------------------------------------
762------------------------------------------------------------------------------
764-- Monomorphically typed version of "async" for better performance of Semigroup
767{-# SPECIALIZE append :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
768append :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
769append (AsyncT m1) (AsyncT m2) = AsyncT $ asyncK m1 m2
771instance MonadAsync m => Semigroup (AsyncT m a) where
774------------------------------------------------------------------------------
776------------------------------------------------------------------------------
778instance MonadAsync m => Monoid (AsyncT m a) where
779 mempty = AsyncT K.nil
782------------------------------------------------------------------------------
784------------------------------------------------------------------------------
786{-# INLINE apAsync #-}
787{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
788apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
789apAsync (AsyncT m1) (AsyncT m2) =
790 let f x1 = K.concatMapWith asyncK (K.fromPure . x1) m2
791 in AsyncT $ K.concatMapWith asyncK f m1
793instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
795 pure = AsyncT . K.fromPure
800------------------------------------------------------------------------------
802------------------------------------------------------------------------------
804-- GHC: if we change the implementation of bindWith with arguments in a
805-- different order we see a significant performance degradation (~2x).
806{-# INLINE bindAsync #-}
807{-# SPECIALIZE bindAsync ::
808 AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
809bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
810bindAsync (AsyncT m) f = AsyncT $ K.bindWith asyncK m (getAsyncT . f)
812-- GHC: if we specify arguments in the definition of (>>=) we see a significant
813-- performance degradation (~2x).
814instance MonadAsync m => Monad (AsyncT m) where
818------------------------------------------------------------------------------
820------------------------------------------------------------------------------
822MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
824------------------------------------------------------------------------------
826------------------------------------------------------------------------------
828{-# INLINE wAsyncK #-}
829wAsyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
830wAsyncK = joinStreamVarAsync WAsyncVar
832-- | XXX we can implement it more efficienty by directly implementing instead
833-- of combining streams using wAsync.
834{-# INLINE consMWAsync #-}
835{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
836consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
837consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r
839-- | For 'WAsyncT' streams:
842-- (<>) = 'Streamly.Prelude.wAsync'
843-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
846-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
847-- executed concurrently a la the 'wAsync' combinator, producing results and
848-- side effects of iterations out of order:
851-- Stream.toList $ Stream.fromWAsync $ do
852-- x <- Stream.fromList [2,1] -- foreach x in stream
853-- Stream.fromEffect $ delay x
859-- Nested monad binds behave like nested @for@ loops with nested iterations
860-- executed concurrently, a la the 'wAsync' combinator:
863-- Stream.toList $ Stream.fromWAsync $ do
864-- x <- Stream.fromList [1,2] -- foreach x in stream
865-- y <- Stream.fromList [2,4] -- foreach y in stream
866-- Stream.fromEffect $ delay (x + y)
874-- The behavior can be explained as follows. All the iterations corresponding
875-- to the element @1@ in the first stream constitute one 'WAsyncT' output
876-- stream and all the iterations corresponding to @2@ constitute another
877-- 'WAsyncT' output stream and these two output streams are merged using
880-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
881-- contrast to the depth wise scheduling behavior of 'AsyncT'.
883-- /Since: 0.2.0 ("Streamly")/
887-- XXX This documentation is redundant, need to be cleaned up/removed.
890-- import qualified "Streamly.Prelude" as S
891-- import Control.Concurrent
893-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
899-- For this example, we are using @maxThreads 1@ so that concurrent thread
900-- scheduling does not affect the results and make them unpredictable. Let's
901-- now take a more general example:
904-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
907-- [1,4,2,7,5,3,8,6,9]
910-- This is how the execution of the above stream proceeds:
912-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
913-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
914-- queue is represented by the rightmost item.
915-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
916-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
917-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
918-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
919-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
920-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
921-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
922-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
923-- 4. Note that the scheduler queue expands by one more stream component in
924-- every pass because one more @<>@ is broken down into two components. At this
925-- point there are no more @<>@ operations to be broken down further and the
926-- queue has reached its maximum size. Now these streams are scheduled in
927-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
929-- As we see above, in a right associated expression composed with @<>@, only
930-- one @<>@ operation is broken down into two components in one execution,
931-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
932-- scheduler passes to expand the whole expression. By the time @n-th@
933-- component is added to the scheduler queue, the first component would have
934-- received @n@ scheduler passes.
936-- Since all streams get interleaved, this operation is not suitable for
937-- folding an infinite lazy container of infinite size streams. However, if
938-- the streams are small, the streams on the left may get finished before more
939-- streams are added to the scheduler queue from the right side of the
940-- expression, so it may be possible to fold an infinite lazy container of
941-- streams. For example, if the streams are of size @n@ then at most @n@
942-- streams would be in the scheduler queue at a time.
944-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
945-- therefore the output of 'WAsyncT' even with a single thread of execution is
946-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
947-- its scheduling behavior.
949-- Any exceptions generated by a constituent stream are propagated to the
950-- output stream. The output and exceptions from a single stream are guaranteed
951-- to arrive in the same order in the resulting stream as they were generated
952-- in the input stream. However, the relative ordering of elements from
953-- different streams in the resulting stream can vary depending on scheduling
954-- and generation delays.
956-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
957-- concurrently using a round robin scheduling.
960-- main = S.'drain' . S.'fromWAsync' $ do
961-- n <- return 3 \<\> return 2 \<\> return 1
963-- threadDelay (n * 1000000)
964-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
967-- ThreadId 40: Delay 1
968-- ThreadId 39: Delay 2
969-- ThreadId 38: Delay 3
972newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
974instance MonadTrans WAsyncT where
976 lift = WAsyncT . K.fromEffect
978-- | A round robin parallely composing IO stream of elements of type @a@.
979-- See 'WAsyncT' documentation for more details.
981-- /Since: 0.2.0 ("Streamly")/
984type WAsync = WAsyncT IO
986------------------------------------------------------------------------------
988------------------------------------------------------------------------------
990{-# INLINE wAppend #-}
991{-# SPECIALIZE wAppend :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
992wAppend :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
993wAppend (WAsyncT m1) (WAsyncT m2) = WAsyncT $ wAsyncK m1 m2
995instance MonadAsync m => Semigroup (WAsyncT m a) where
998------------------------------------------------------------------------------
1000------------------------------------------------------------------------------
1002instance MonadAsync m => Monoid (WAsyncT m a) where
1003 mempty = WAsyncT K.nil
1006------------------------------------------------------------------------------
1008------------------------------------------------------------------------------
1010{-# INLINE apWAsync #-}
1011{-# SPECIALIZE apWAsync ::
1012 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
1013apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
1014apWAsync (WAsyncT m1) (WAsyncT m2) =
1015 let f x1 = K.concatMapWith wAsyncK (K.fromPure . x1) m2
1016 in WAsyncT $ K.concatMapWith wAsyncK f m1
1018-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1019-- performance degradation (~2x).
1020instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1021 pure = WAsyncT . K.fromPure
1024------------------------------------------------------------------------------
1026------------------------------------------------------------------------------
1028-- GHC: if we change the implementation of bindWith with arguments in a
1029-- different order we see a significant performance degradation (~2x).
1030{-# INLINE bindWAsync #-}
1031{-# SPECIALIZE bindWAsync ::
1032 WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1033bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1034bindWAsync (WAsyncT m) f = WAsyncT $ K.bindWith wAsyncK m (getWAsyncT . f)
1036-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1037-- performance degradation (~2x).
1038instance MonadAsync m => Monad (WAsyncT m) where
1042------------------------------------------------------------------------------
1044------------------------------------------------------------------------------
1046MONAD_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