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)
44#if !(MIN_VERSION_transformers(0,6,0))
45import Control.Monad.Base (MonadBase(..), liftBaseDefault)
47import Control.Monad.Catch (MonadThrow, throwM)
48import Control.Concurrent.MVar (newEmptyMVar)
49-- import Control.Monad.Error.Class (MonadError(..))
50import Control.Monad.IO.Class (MonadIO(..))
51import Control.Monad.Reader.Class (MonadReader(..))
52import Control.Monad.State.Class (MonadState(..))
53#if !(MIN_VERSION_transformers(0,6,0))
54import Control.Monad.Trans.Class (MonadTrans(lift))
56import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)
57import Data.IORef (IORef, newIORef, readIORef)
58import Data.Maybe (fromJust)
60import Prelude hiding (map)
61import qualified Data.Set as S
63import Streamly.Internal.Control.Concurrent
64 (MonadRunInIO, MonadAsync, askRunInIO, restoreM)
65import Streamly.Internal.Data.Atomics
66 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
67import Streamly.Internal.Data.StreamK (Stream)
68import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
70import qualified Streamly.Internal.Data.StreamK as K
71 (foldStreamShared, mkStream, foldStream, fromEffect
72 , nil, concatMapWith, fromPure, bindWith)
73import qualified Streamly.Internal.Data.Stream as D
74 (Stream(..), Step(..), mapM, toStreamK, fromStreamK)
75import qualified Streamly.Internal.Data.Stream.Serial as Stream (toStreamK)
77import Streamly.Internal.Data.SVar
79#include "Instances.hs"
82-- >>> :set -fno-warn-deprecations
83-- >>> import qualified Streamly.Prelude as Stream
84-- >>> import Control.Concurrent (threadDelay)
87-- threadDelay (n * 1000000) -- sleep for n seconds
88-- putStrLn (show n ++ " sec") -- print "n sec"
93{-# INLINABLE withLocal #-}
94withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
96 K.mkStream $ \st yld sng stp ->
97 let single = local f . sng
98 yieldk a r = local f $ yld a (withLocal f r)
99 in K.foldStream st yieldk single (local f stp) m
101-------------------------------------------------------------------------------
103-------------------------------------------------------------------------------
105-- Note: For purely right associated expressions this queue should have at most
106-- one element. It grows to more than one when we have left associcated
107-- expressions. Large left associated compositions can grow this to a
109{-# INLINE enqueueLIFO #-}
111 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
112enqueueLIFO sv q m = do
113 atomicModifyIORefCAS_ q $ \ms -> m : ms
116data WorkerStatus = Continue | Suspend
118{-# INLINE workLoopLIFO #-}
121 => IORef [(RunInIO m, Stream m a)]
126workLoopLIFO q st sv winfo = run
130 stop = liftIO $ sendStop sv winfo
135 Just (RunInIO runin, m) -> do
136 -- XXX when we finish we need to send the monadic state back to
137 -- the parent so that the state can be merged back. We capture
138 -- and return the state in the stop continuation.
140 -- Instead of using the run function we can just restore the
141 -- monad state here. That way it can work easily for
142 -- distributed case as well.
143 r <- liftIO $ runin $
144 K.foldStreamShared st yieldk single (return Continue) m
151 res <- liftIO $ sendYield sv winfo (ChildYield a)
152 return $ if res then Continue else Suspend
155 res <- liftIO $ sendYield sv winfo (ChildYield a)
157 then K.foldStreamShared st yieldk single (return Continue) r
159 runInIO <- askRunInIO
160 liftIO $ enqueueLIFO sv q (runInIO, r)
163 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
165 x : xs -> (xs, Just x)
167-- We duplicate workLoop for yield limit and no limit cases because it has
168-- around 40% performance overhead in the worst case.
170-- XXX we can pass yinfo directly as an argument here so that we do not have to
171-- make a check every time.
172{-# INLINE workLoopLIFOLimited #-}
174 :: forall m a. MonadRunInIO m
175 => IORef [(RunInIO m, Stream m a)]
180workLoopLIFOLimited q st sv winfo = run
184 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
185 stop = liftIO $ sendStop sv winfo
190 Just (RunInIO runin, m) -> do
191 -- XXX This is just a best effort minimization of concurrency
192 -- to the yield limit. If the stream is made of concurrent
193 -- streams we do not reserve the yield limit in the constituent
194 -- streams before executing the action. This can be done
195 -- though, by sharing the yield limit ref with downstream
196 -- actions via state passing. Just a todo.
197 yieldLimitOk <- liftIO $ decrementYieldLimit sv
200 r <- liftIO $ runin $
201 K.foldStreamShared st yieldk single incrContinue m
206 -- Avoid any side effects, undo the yield limit decrement if we
207 -- never yielded anything.
209 enqueueLIFO sv q (RunInIO runin, m)
210 incrementYieldLimit sv
214 res <- liftIO $ sendYield sv winfo (ChildYield a)
215 return $ if res then Continue else Suspend
217 -- XXX can we pass on the yield limit downstream to limit the concurrency
218 -- of constituent streams.
220 res <- liftIO $ sendYield sv winfo (ChildYield a)
221 yieldLimitOk <- liftIO $ decrementYieldLimit sv
222 if res && yieldLimitOk
223 then K.foldStreamShared st yieldk single incrContinue r
225 runInIO <- askRunInIO
226 liftIO $ incrementYieldLimit sv
227 liftIO $ enqueueLIFO sv q (runInIO, r)
230 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
232 x : xs -> (xs, Just x)
234-------------------------------------------------------------------------------
236-------------------------------------------------------------------------------
238-- XXX we can use the Ahead style sequence/heap mechanism to make the best
239-- effort to always try to finish the streams on the left side of an expression
240-- first as long as possible.
242{-# INLINE enqueueFIFO #-}
245 -> LinkedQueue (RunInIO m, t m a)
246 -> (RunInIO m, t m a)
248enqueueFIFO sv q m = do
252-- XXX we can remove sv as it is derivable from st
254{-# INLINE workLoopFIFO #-}
257 => LinkedQueue (RunInIO m, Stream m a)
262workLoopFIFO q st sv winfo = run
266 stop = liftIO $ sendStop sv winfo
268 work <- liftIO $ tryPopR q
271 Just (RunInIO runin, m) -> do
272 r <- liftIO $ runin $
273 K.foldStreamShared st yieldk single (return Continue) m
280 res <- liftIO $ sendYield sv winfo (ChildYield a)
281 return $ if res then Continue else Suspend
283 -- XXX in general we would like to yield "n" elements from a single stream
284 -- before moving on to the next. Single element granularity could be too
285 -- expensive in certain cases. Similarly, we can use time limit for
288 res <- liftIO $ sendYield sv winfo (ChildYield a)
289 runInIO <- askRunInIO
290 liftIO $ enqueueFIFO sv q (runInIO, r)
291 return $ if res then Continue else Suspend
293{-# INLINE workLoopFIFOLimited #-}
295 :: forall m a. MonadRunInIO m
296 => LinkedQueue (RunInIO m, Stream m a)
301workLoopFIFOLimited q st sv winfo = run
305 stop = liftIO $ sendStop sv winfo
306 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
308 work <- liftIO $ tryPopR q
311 Just (RunInIO runin, m) -> do
312 yieldLimitOk <- liftIO $ decrementYieldLimit sv
315 r <- liftIO $ runin $
316 K.foldStreamShared st yieldk single incrContinue m
322 enqueueFIFO sv q (RunInIO runin, m)
323 incrementYieldLimit sv
327 res <- liftIO $ sendYield sv winfo (ChildYield a)
328 return $ if res then Continue else Suspend
331 res <- liftIO $ sendYield sv winfo (ChildYield a)
332 runInIO <- askRunInIO
333 liftIO $ enqueueFIFO sv q (runInIO, r)
334 yieldLimitOk <- liftIO $ decrementYieldLimit sv
335 if res && yieldLimitOk
338 incrementYieldLimit sv
341-------------------------------------------------------------------------------
343-- This code belongs in SVar.hs but is kept here for perf reasons
344-------------------------------------------------------------------------------
346-- XXX we have this function in this file because passing runStreamLIFO as a
347-- function argument to this function results in a perf degradation of more
348-- than 10%. Need to investigate what the root cause is.
349-- Interestingly, the same thing does not make any difference for Ahead.
350getLifoSVar :: forall m a. MonadAsync m
351 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
352getLifoSVar st mrun = do
353 outQ <- newIORef ([], 0)
354 outQMv <- newEmptyMVar
356 wfw <- newIORef False
357 running <- newIORef S.empty
358 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
359 yl <- case getYieldLimit st of
360 Nothing -> return Nothing
361 Just x -> Just <$> newIORef x
362 rateInfo <- getYieldRateInfo st
364 stats <- newSVarStats
367 let isWorkFinished _ = null <$> readIORef q
369 let isWorkFinishedLimited sv = do
371 case remainingWork sv of
375 Nothing -> return False
376 qEmpty <- null <$> readIORef q
377 return $ qEmpty || yieldsDone
379 let getSVar :: SVar Stream m a
380 -> (SVar Stream m a -> m [ChildEvent a])
381 -> (SVar Stream m a -> m Bool)
382 -> (SVar Stream m a -> IO Bool)
383 -> (IORef [(RunInIO m, Stream m a)]
389 getSVar sv readOutput postProc workDone wloop = SVar
391 , outputQueueFromConsumer = undefined
393 , maxBufferLimit = getMaxBuffer st
394 , pushBufferSpace = undefined
395 , pushBufferPolicy = undefined
396 , pushBufferMVar = undefined
397 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
398 , yieldRateInfo = rateInfo
399 , outputDoorBell = outQMv
400 , outputDoorBellFromConsumer = undefined
401 , readOutputQ = readOutput sv
402 , postProcess = postProc sv
403 , workerThreads = running
404 , workLoop = wloop q st{streamVar = Just sv} sv
405 , enqueue = enqueueLIFO sv q
406 , isWorkDone = workDone sv
407 , isQueueDone = workDone sv
409 , svarStyle = AsyncVar
410 , svarStopStyle = StopNone
411 , svarStopBy = undefined
413 , workerCount = active
414 , accountThread = delThread sv
415 , workerStopMVar = undefined
417 , svarInspectMode = getInspectMode st
419 , aheadWorkQueue = undefined
420 , outputHeap = undefined
425 case getStreamRate st of
427 case getYieldLimit st of
428 Nothing -> getSVar sv readOutputQBounded
432 Just _ -> getSVar sv readOutputQBounded
434 isWorkFinishedLimited
437 case getYieldLimit st of
438 Nothing -> getSVar sv readOutputQPaced
442 Just _ -> getSVar sv readOutputQPaced
444 isWorkFinishedLimited
448getFifoSVar :: forall m a. MonadAsync m
449 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
450getFifoSVar st mrun = do
451 outQ <- newIORef ([], 0)
452 outQMv <- newEmptyMVar
454 wfw <- newIORef False
455 running <- newIORef S.empty
457 yl <- case getYieldLimit st of
458 Nothing -> return Nothing
459 Just x -> Just <$> newIORef x
460 rateInfo <- getYieldRateInfo st
462 stats <- newSVarStats
465 let isWorkFinished _ = nullQ q
466 let isWorkFinishedLimited sv = do
468 case remainingWork sv of
472 Nothing -> return False
474 return $ qEmpty || yieldsDone
476 let getSVar :: SVar Stream m a
477 -> (SVar Stream m a -> m [ChildEvent a])
478 -> (SVar Stream m a -> m Bool)
479 -> (SVar Stream m a -> IO Bool)
480 -> (LinkedQueue (RunInIO m, Stream m a)
486 getSVar sv readOutput postProc workDone wloop = SVar
488 , outputQueueFromConsumer = undefined
490 , maxBufferLimit = getMaxBuffer st
491 , pushBufferSpace = undefined
492 , pushBufferPolicy = undefined
493 , pushBufferMVar = undefined
494 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
495 , yieldRateInfo = rateInfo
496 , outputDoorBell = outQMv
497 , outputDoorBellFromConsumer = undefined
498 , readOutputQ = readOutput sv
499 , postProcess = postProc sv
500 , workerThreads = running
501 , workLoop = wloop q st{streamVar = Just sv} sv
502 , enqueue = enqueueFIFO sv q
503 , isWorkDone = workDone sv
504 , isQueueDone = workDone sv
506 , svarStyle = WAsyncVar
507 , svarStopStyle = StopNone
508 , svarStopBy = undefined
510 , workerCount = active
511 , accountThread = delThread sv
512 , workerStopMVar = undefined
514 , svarInspectMode = getInspectMode st
516 , aheadWorkQueue = undefined
517 , outputHeap = undefined
522 case getStreamRate st of
524 case getYieldLimit st of
525 Nothing -> getSVar sv readOutputQBounded
529 Just _ -> getSVar sv readOutputQBounded
531 isWorkFinishedLimited
534 case getYieldLimit st of
535 Nothing -> getSVar sv readOutputQPaced
539 Just _ -> getSVar sv readOutputQPaced
541 isWorkFinishedLimited
545{-# INLINABLE newAsyncVar #-}
546newAsyncVar :: MonadAsync m
547 => State Stream m a -> Stream m a -> m (SVar Stream m a)
550 sv <- liftIO $ getLifoSVar st mrun
553-- | Generate a stream asynchronously to keep it buffered, lazily consume
558{-# INLINABLE mkAsyncK #-}
559mkAsyncK :: MonadAsync m => Stream m a -> Stream m a
560mkAsyncK m = K.mkStream $ \st yld sng stp -> do
561 sv <- newAsyncVar (adaptState st) m
562 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
565-- This is slightly faster than the CPS version above
567{-# INLINE_NORMAL mkAsyncD #-}
568mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
569mkAsyncD m = D.Stream step Nothing
572 step gst Nothing = do
573 sv <- newAsyncVar gst (D.toStreamK m)
574 return $ D.Skip $ Just $ fromSVarD sv
576 step gst (Just (D.UnStream step1 st)) = do
579 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
580 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
583-- XXX We can pass captureMonadState instead of using MonadAsync
584-- | Create a new SVar and enqueue one stream computation on it.
585{-# INLINABLE newWAsyncVar #-}
586newWAsyncVar :: MonadAsync m
587 => State Stream m a -> Stream m a -> m (SVar Stream m a)
588newWAsyncVar st m = do
590 sv <- liftIO $ getFifoSVar st mrun
591 -- XXX Use just Stream and IO in all the functions below
592 -- XXX pass mrun instead of calling captureMonadState again inside it
595------------------------------------------------------------------------------
596-- Running streams concurrently
597------------------------------------------------------------------------------
599-- Concurrency rate control.
601-- Our objective is to create more threads on demand if the consumer is running
602-- faster than us. As soon as we encounter a concurrent composition we create a
603-- push pull pair of threads. We use an SVar for communication between the
604-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
605-- The producer creates more threads if the SVar drains and becomes empty, that
606-- is the consumer is running faster.
608-- XXX Note 1: This mechanism can be problematic if the initial production
609-- latency is high, we may end up creating too many threads. So we need some
610-- way to monitor and use the latency as well. Having a limit on the dispatches
611-- (programmer controlled) may also help.
613-- TBD Note 2: We may want to run computations at the lower level of the
614-- composition tree serially even when they are composed using a parallel
615-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
616-- place of 'wAsync'. If we find that an SVar immediately above a computation
617-- gets drained empty we can switch to parallelizing the computation. For that
618-- we can use a state flag to fork the rest of the computation at any point of
619-- time inside the Monad bind operation if the consumer is running at a faster
622-- TBD Note 3: the binary operation ('parallel') composition allows us to
623-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
624-- chunksizes we will need to compose the parallel actions using a data
625-- constructor (A Free container) instead so that we can divide it in chunks of
626-- arbitrary size before dispatching. If the stream is composed of
627-- hierarchically composed grains of different sizes then we can always switch
628-- to a desired granularity depending on the consumer speed.
630-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
631-- into just the number of CPUs.
633-- | Join two computations on the currently running 'SVar' queue for concurrent
634-- execution. When we are using parallel composition, an SVar is passed around
635-- as a state variable. We try to schedule a new parallel computation on the
636-- SVar passed to us. The first time, when no SVar exists, a new SVar is
637-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
638-- already scheduled on the SVar is further evaluated. For example, when (a
639-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
640-- the current scheduler queue.
642-- The 'SVarStyle' required by the current composition context is passed as one
643-- of the parameters. If the scheduling and composition style of the new
644-- computation being scheduled is different than the style of the current SVar,
645-- then we create a new SVar and schedule it on that. The newly created SVar
646-- joins as one of the computations on the current SVar queue.
648-- Cases when we need to switch to a new SVar:
650-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
651-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
652-- because of the scheduling policy change.
653-- * if we 'adapt' a stream of type 'async' to a stream of type
654-- 'Parallel', we create a new SVar at the transitioning bind.
655-- * When the stream is switching from disjunctive composition to conjunctive
656-- composition and vice-versa we create a new SVar to isolate the scheduling
659forkSVarAsync :: MonadAsync m
660 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
661forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do
663 AsyncVar -> newAsyncVar st (concurrently m1 m2)
664 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
665 _ -> error "illegal svar type"
666 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
668 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
669 runInIO <- askRunInIO
670 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
671 K.foldStreamShared st yld sng stp ma
673{-# INLINE joinStreamVarAsync #-}
674joinStreamVarAsync :: MonadAsync m
675 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
676joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->
678 Just sv | svarStyle sv == style -> do
679 runInIO <- askRunInIO
680 liftIO $ enqueue sv (runInIO, m2)
681 K.foldStreamShared st yld sng stp m1
682 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
684------------------------------------------------------------------------------
685-- Semigroup and Monoid style compositions for parallel actions
686------------------------------------------------------------------------------
689asyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
690asyncK = joinStreamVarAsync AsyncVar
692-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
693-- huge difference in the performance of consM in IsStream instance even we
694-- have a SPECIALIZE in the instance.
696-- | XXX we can implement it more efficienty by directly implementing instead
697-- of combining streams using async.
698{-# INLINE consMAsync #-}
699{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
700consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
701consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r
703------------------------------------------------------------------------------
705------------------------------------------------------------------------------
707-- | For 'AsyncT' streams:
710-- (<>) = 'Streamly.Prelude.async'
711-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
714-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
715-- executed concurrently a la the 'async' combinator, producing results and
716-- side effects of iterations out of order:
719-- Stream.toList $ Stream.fromAsync $ do
720-- x <- Stream.fromList [2,1] -- foreach x in stream
721-- Stream.fromEffect $ delay x
727-- Nested monad binds behave like nested @for@ loops with nested iterations
728-- executed concurrently, a la the 'async' combinator:
731-- Stream.toList $ Stream.fromAsync $ do
732-- x <- Stream.fromList [1,2] -- foreach x in stream
733-- y <- Stream.fromList [2,4] -- foreach y in stream
734-- Stream.fromEffect $ delay (x + y)
742-- The behavior can be explained as follows. All the iterations corresponding
743-- to the element @1@ in the first stream constitute one output stream and all
744-- the iterations corresponding to @2@ constitute another output stream and
745-- these two output streams are merged using 'async'.
747-- /Since: 0.1.0 ("Streamly")/
750newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
752#if !(MIN_VERSION_transformers(0,6,0))
753instance MonadTrans AsyncT where
755 lift = AsyncT . K.fromEffect
758-- | A demand driven left biased parallely composing IO stream of elements of
759-- type @a@. See 'AsyncT' documentation for more details.
761-- /Since: 0.2.0 ("Streamly")/
764type Async = AsyncT IO
766------------------------------------------------------------------------------
768------------------------------------------------------------------------------
770-- Monomorphically typed version of "async" for better performance of Semigroup
773{-# SPECIALIZE append :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
774append :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
775append (AsyncT m1) (AsyncT m2) = AsyncT $ asyncK m1 m2
777instance MonadAsync m => Semigroup (AsyncT m a) where
780------------------------------------------------------------------------------
782------------------------------------------------------------------------------
784instance MonadAsync m => Monoid (AsyncT m a) where
785 mempty = AsyncT K.nil
788------------------------------------------------------------------------------
790------------------------------------------------------------------------------
792{-# INLINE apAsync #-}
793{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
794apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
795apAsync (AsyncT m1) (AsyncT m2) =
796 let f x1 = K.concatMapWith asyncK (K.fromPure . x1) m2
797 in AsyncT $ K.concatMapWith asyncK f m1
799instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
801 pure = AsyncT . K.fromPure
806------------------------------------------------------------------------------
808------------------------------------------------------------------------------
810-- GHC: if we change the implementation of bindWith with arguments in a
811-- different order we see a significant performance degradation (~2x).
812{-# INLINE bindAsync #-}
813{-# SPECIALIZE bindAsync ::
814 AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
815bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
816bindAsync (AsyncT m) f = AsyncT $ K.bindWith asyncK m (getAsyncT . f)
818-- GHC: if we specify arguments in the definition of (>>=) we see a significant
819-- performance degradation (~2x).
820instance MonadAsync m => Monad (AsyncT m) where
824------------------------------------------------------------------------------
826------------------------------------------------------------------------------
828#if !(MIN_VERSION_transformers(0,6,0))
829instance (MonadBase b m, Monad m, MonadAsync m) => MonadBase b (AsyncT m) where
830 liftBase = liftBaseDefault
833MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
835------------------------------------------------------------------------------
837------------------------------------------------------------------------------
839{-# INLINE wAsyncK #-}
840wAsyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
841wAsyncK = joinStreamVarAsync WAsyncVar
843-- | XXX we can implement it more efficienty by directly implementing instead
844-- of combining streams using wAsync.
845{-# INLINE consMWAsync #-}
846{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
847consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
848consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r
850-- | For 'WAsyncT' streams:
853-- (<>) = 'Streamly.Prelude.wAsync'
854-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
857-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
858-- executed concurrently a la the 'wAsync' combinator, producing results and
859-- side effects of iterations out of order:
862-- Stream.toList $ Stream.fromWAsync $ do
863-- x <- Stream.fromList [2,1] -- foreach x in stream
864-- Stream.fromEffect $ delay x
870-- Nested monad binds behave like nested @for@ loops with nested iterations
871-- executed concurrently, a la the 'wAsync' combinator:
874-- Stream.toList $ Stream.fromWAsync $ do
875-- x <- Stream.fromList [1,2] -- foreach x in stream
876-- y <- Stream.fromList [2,4] -- foreach y in stream
877-- Stream.fromEffect $ delay (x + y)
885-- The behavior can be explained as follows. All the iterations corresponding
886-- to the element @1@ in the first stream constitute one 'WAsyncT' output
887-- stream and all the iterations corresponding to @2@ constitute another
888-- 'WAsyncT' output stream and these two output streams are merged using
891-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
892-- contrast to the depth wise scheduling behavior of 'AsyncT'.
894-- /Since: 0.2.0 ("Streamly")/
898-- XXX This documentation is redundant, need to be cleaned up/removed.
901-- import qualified "Streamly.Prelude" as S
902-- import Control.Concurrent
904-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
910-- For this example, we are using @maxThreads 1@ so that concurrent thread
911-- scheduling does not affect the results and make them unpredictable. Let's
912-- now take a more general example:
915-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
918-- [1,4,2,7,5,3,8,6,9]
921-- This is how the execution of the above stream proceeds:
923-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
924-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
925-- queue is represented by the rightmost item.
926-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
927-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
928-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
929-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
930-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
931-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
932-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
933-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
934-- 4. Note that the scheduler queue expands by one more stream component in
935-- every pass because one more @<>@ is broken down into two components. At this
936-- point there are no more @<>@ operations to be broken down further and the
937-- queue has reached its maximum size. Now these streams are scheduled in
938-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
940-- As we see above, in a right associated expression composed with @<>@, only
941-- one @<>@ operation is broken down into two components in one execution,
942-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
943-- scheduler passes to expand the whole expression. By the time @n-th@
944-- component is added to the scheduler queue, the first component would have
945-- received @n@ scheduler passes.
947-- Since all streams get interleaved, this operation is not suitable for
948-- folding an infinite lazy container of infinite size streams. However, if
949-- the streams are small, the streams on the left may get finished before more
950-- streams are added to the scheduler queue from the right side of the
951-- expression, so it may be possible to fold an infinite lazy container of
952-- streams. For example, if the streams are of size @n@ then at most @n@
953-- streams would be in the scheduler queue at a time.
955-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
956-- therefore the output of 'WAsyncT' even with a single thread of execution is
957-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
958-- its scheduling behavior.
960-- Any exceptions generated by a constituent stream are propagated to the
961-- output stream. The output and exceptions from a single stream are guaranteed
962-- to arrive in the same order in the resulting stream as they were generated
963-- in the input stream. However, the relative ordering of elements from
964-- different streams in the resulting stream can vary depending on scheduling
965-- and generation delays.
967-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
968-- concurrently using a round robin scheduling.
971-- main = S.'drain' . S.'fromWAsync' $ do
972-- n <- return 3 \<\> return 2 \<\> return 1
974-- threadDelay (n * 1000000)
975-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
978-- ThreadId 40: Delay 1
979-- ThreadId 39: Delay 2
980-- ThreadId 38: Delay 3
983newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
985#if !(MIN_VERSION_transformers(0,6,0))
986instance MonadTrans WAsyncT where
988 lift = WAsyncT . K.fromEffect
991-- | A round robin parallely composing IO stream of elements of type @a@.
992-- See 'WAsyncT' documentation for more details.
994-- /Since: 0.2.0 ("Streamly")/
997type WAsync = WAsyncT IO
999------------------------------------------------------------------------------
1001------------------------------------------------------------------------------
1003{-# INLINE wAppend #-}
1004{-# SPECIALIZE wAppend :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1005wAppend :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
1006wAppend (WAsyncT m1) (WAsyncT m2) = WAsyncT $ wAsyncK m1 m2
1008instance MonadAsync m => Semigroup (WAsyncT m a) where
1011------------------------------------------------------------------------------
1013------------------------------------------------------------------------------
1015instance MonadAsync m => Monoid (WAsyncT m a) where
1016 mempty = WAsyncT K.nil
1019------------------------------------------------------------------------------
1021------------------------------------------------------------------------------
1023{-# INLINE apWAsync #-}
1024{-# SPECIALIZE apWAsync ::
1025 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
1026apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
1027apWAsync (WAsyncT m1) (WAsyncT m2) =
1028 let f x1 = K.concatMapWith wAsyncK (K.fromPure . x1) m2
1029 in WAsyncT $ K.concatMapWith wAsyncK f m1
1031-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1032-- performance degradation (~2x).
1033instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1034 pure = WAsyncT . K.fromPure
1037------------------------------------------------------------------------------
1039------------------------------------------------------------------------------
1041-- GHC: if we change the implementation of bindWith with arguments in a
1042-- different order we see a significant performance degradation (~2x).
1043{-# INLINE bindWAsync #-}
1044{-# SPECIALIZE bindWAsync ::
1045 WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1046bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1047bindWAsync (WAsyncT m) f = WAsyncT $ K.bindWith wAsyncK m (getWAsyncT . f)
1049-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1050-- performance degradation (~2x).
1051instance MonadAsync m => Monad (WAsyncT m) where
1055------------------------------------------------------------------------------
1057------------------------------------------------------------------------------
1059#if !(MIN_VERSION_transformers(0,6,0))
1060instance (MonadBase b m, Monad m, MonadAsync m) => MonadBase b (WAsyncT m) where
1061 liftBase = liftBaseDefault
1064MONAD_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