1{-# LANGUAGE UndecidableInstances #-}
6-- Module : Streamly.Internal.Data.Stream.Async
7-- Copyright : (c) 2017 Composewell Technologies
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
14-- To run examples in this module:
16-- >>> import qualified Streamly.Prelude as Stream
17-- >>> import Control.Concurrent (threadDelay)
20-- threadDelay (n * 1000000) -- sleep for n seconds
21-- putStrLn (show n ++ " sec") -- print "n sec"
25module Streamly.Internal.Data.Stream.Async
41import Control.Concurrent (myThreadId)
42import Control.Monad.Base (MonadBase(..), liftBaseDefault)
43import Control.Monad.Catch (MonadThrow, throwM)
44import Control.Monad.Trans.Control (MonadBaseControl (..))
45import Control.Concurrent.MVar (newEmptyMVar)
46-- import Control.Monad.Error.Class (MonadError(..))
47import Control.Monad.IO.Class (MonadIO(..))
48import Control.Monad.Reader.Class (MonadReader(..))
49import Control.Monad.State.Class (MonadState(..))
50import Control.Monad.Trans.Class (MonadTrans(lift))
51import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)
52import Data.IORef (IORef, newIORef, readIORef)
53import Data.Maybe (fromJust)
54#if __GLASGOW_HASKELL__ < 808
55import Data.Semigroup (Semigroup(..))
58import Prelude hiding (map)
59import qualified Data.Set as S
61import Streamly.Internal.Control.Concurrent
62 (MonadAsync, RunInIO(..), captureMonadState)
63import Streamly.Internal.Data.Atomics
64 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
65import Streamly.Internal.Data.Stream.Serial (SerialT (..))
66import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
67import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
69import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
70import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
72import Streamly.Internal.Data.SVar.Type
73import Streamly.Internal.Data.SVar
75#include "Instances.hs"
78-- >>> import qualified Streamly.Prelude as Stream
79-- >>> import Control.Concurrent (threadDelay)
82-- threadDelay (n * 1000000) -- sleep for n seconds
83-- putStrLn (show n ++ " sec") -- print "n sec"
88-------------------------------------------------------------------------------
90-------------------------------------------------------------------------------
92-- Note: For purely right associated expressions this queue should have at most
93-- one element. It grows to more than one when we have left associcated
94-- expressions. Large left associated compositions can grow this to a
96{-# INLINE enqueueLIFO #-}
98 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
99enqueueLIFO sv q m = do
100 atomicModifyIORefCAS_ q $ \ms -> m : ms
103data WorkerStatus = Continue | Suspend
105{-# INLINE workLoopLIFO #-}
107 :: (MonadIO m, MonadBaseControl IO m)
108 => IORef [(RunInIO m, Stream m a)]
113workLoopLIFO q st sv winfo = run
117 stop = liftIO $ sendStop sv winfo
122 Just (RunInIO runin, m) -> do
123 -- XXX when we finish we need to send the monadic state back to
124 -- the parent so that the state can be merged back. We capture
125 -- and return the state in the stop continuation.
127 -- Instead of using the run function we can just restore the
128 -- monad state here. That way it can work easily for
129 -- distributed case as well.
130 r <- liftIO $ runin $
131 K.foldStreamShared st yieldk single (return Continue) m
138 res <- liftIO $ sendYield sv winfo (ChildYield a)
139 return $ if res then Continue else Suspend
142 res <- liftIO $ sendYield sv winfo (ChildYield a)
144 then K.foldStreamShared st yieldk single (return Continue) r
146 runInIO <- captureMonadState
147 liftIO $ enqueueLIFO sv q (runInIO, r)
150 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
152 x : xs -> (xs, Just x)
154-- We duplicate workLoop for yield limit and no limit cases because it has
155-- around 40% performance overhead in the worst case.
157-- XXX we can pass yinfo directly as an argument here so that we do not have to
158-- make a check every time.
159{-# INLINE workLoopLIFOLimited #-}
161 :: forall m a. (MonadIO m, MonadBaseControl IO m)
162 => IORef [(RunInIO m, Stream m a)]
167workLoopLIFOLimited q st sv winfo = run
171 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
172 stop = liftIO $ sendStop sv winfo
177 Just (RunInIO runin, m) -> do
178 -- XXX This is just a best effort minimization of concurrency
179 -- to the yield limit. If the stream is made of concurrent
180 -- streams we do not reserve the yield limit in the constituent
181 -- streams before executing the action. This can be done
182 -- though, by sharing the yield limit ref with downstream
183 -- actions via state passing. Just a todo.
184 yieldLimitOk <- liftIO $ decrementYieldLimit sv
187 r <- liftIO $ runin $
188 K.foldStreamShared st yieldk single incrContinue m
193 -- Avoid any side effects, undo the yield limit decrement if we
194 -- never yielded anything.
196 enqueueLIFO sv q (RunInIO runin, m)
197 incrementYieldLimit sv
201 res <- liftIO $ sendYield sv winfo (ChildYield a)
202 return $ if res then Continue else Suspend
204 -- XXX can we pass on the yield limit downstream to limit the concurrency
205 -- of constituent streams.
207 res <- liftIO $ sendYield sv winfo (ChildYield a)
208 yieldLimitOk <- liftIO $ decrementYieldLimit sv
209 if res && yieldLimitOk
210 then K.foldStreamShared st yieldk single incrContinue r
212 runInIO <- captureMonadState
213 liftIO $ incrementYieldLimit sv
214 liftIO $ enqueueLIFO sv q (runInIO, r)
217 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
219 x : xs -> (xs, Just x)
221-------------------------------------------------------------------------------
223-------------------------------------------------------------------------------
225-- XXX we can use the Ahead style sequence/heap mechanism to make the best
226-- effort to always try to finish the streams on the left side of an expression
227-- first as long as possible.
229{-# INLINE enqueueFIFO #-}
232 -> LinkedQueue (RunInIO m, t m a)
233 -> (RunInIO m, t m a)
235enqueueFIFO sv q m = do
239-- XXX we can remove sv as it is derivable from st
241{-# INLINE workLoopFIFO #-}
243 :: (MonadIO m, MonadBaseControl IO m)
244 => LinkedQueue (RunInIO m, Stream m a)
249workLoopFIFO q st sv winfo = run
253 stop = liftIO $ sendStop sv winfo
255 work <- liftIO $ tryPopR q
258 Just (RunInIO runin, m) -> do
259 r <- liftIO $ runin $
260 K.foldStreamShared st yieldk single (return Continue) m
267 res <- liftIO $ sendYield sv winfo (ChildYield a)
268 return $ if res then Continue else Suspend
270 -- XXX in general we would like to yield "n" elements from a single stream
271 -- before moving on to the next. Single element granularity could be too
272 -- expensive in certain cases. Similarly, we can use time limit for
275 res <- liftIO $ sendYield sv winfo (ChildYield a)
276 runInIO <- captureMonadState
277 liftIO $ enqueueFIFO sv q (runInIO, r)
278 return $ if res then Continue else Suspend
280{-# INLINE workLoopFIFOLimited #-}
282 :: forall m a. (MonadIO m, MonadBaseControl IO m)
283 => LinkedQueue (RunInIO m, Stream m a)
288workLoopFIFOLimited q st sv winfo = run
292 stop = liftIO $ sendStop sv winfo
293 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
295 work <- liftIO $ tryPopR q
298 Just (RunInIO runin, m) -> do
299 yieldLimitOk <- liftIO $ decrementYieldLimit sv
302 r <- liftIO $ runin $
303 K.foldStreamShared st yieldk single incrContinue m
309 enqueueFIFO sv q (RunInIO runin, m)
310 incrementYieldLimit sv
314 res <- liftIO $ sendYield sv winfo (ChildYield a)
315 return $ if res then Continue else Suspend
318 res <- liftIO $ sendYield sv winfo (ChildYield a)
319 runInIO <- captureMonadState
320 liftIO $ enqueueFIFO sv q (runInIO, r)
321 yieldLimitOk <- liftIO $ decrementYieldLimit sv
322 if res && yieldLimitOk
325 incrementYieldLimit sv
328-------------------------------------------------------------------------------
330-- This code belongs in SVar.hs but is kept here for perf reasons
331-------------------------------------------------------------------------------
333-- XXX we have this function in this file because passing runStreamLIFO as a
334-- function argument to this function results in a perf degradation of more
335-- than 10%. Need to investigate what the root cause is.
336-- Interestingly, the same thing does not make any difference for Ahead.
337getLifoSVar :: forall m a. MonadAsync m
338 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
339getLifoSVar st mrun = do
340 outQ <- newIORef ([], 0)
341 outQMv <- newEmptyMVar
343 wfw <- newIORef False
344 running <- newIORef S.empty
345 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
346 yl <- case getYieldLimit st of
347 Nothing -> return Nothing
348 Just x -> Just <$> newIORef x
349 rateInfo <- getYieldRateInfo st
351 stats <- newSVarStats
354 let isWorkFinished _ = null <$> readIORef q
356 let isWorkFinishedLimited sv = do
358 case remainingWork sv of
362 Nothing -> return False
363 qEmpty <- null <$> readIORef q
364 return $ qEmpty || yieldsDone
366 let getSVar :: SVar Stream m a
367 -> (SVar Stream m a -> m [ChildEvent a])
368 -> (SVar Stream m a -> m Bool)
369 -> (SVar Stream m a -> IO Bool)
370 -> (IORef [(RunInIO m, Stream m a)]
376 getSVar sv readOutput postProc workDone wloop = SVar
378 , outputQueueFromConsumer = undefined
380 , maxBufferLimit = getMaxBuffer st
381 , pushBufferSpace = undefined
382 , pushBufferPolicy = undefined
383 , pushBufferMVar = undefined
384 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
385 , yieldRateInfo = rateInfo
386 , outputDoorBell = outQMv
387 , outputDoorBellFromConsumer = undefined
388 , readOutputQ = readOutput sv
389 , postProcess = postProc sv
390 , workerThreads = running
391 , workLoop = wloop q st{streamVar = Just sv} sv
392 , enqueue = enqueueLIFO sv q
393 , isWorkDone = workDone sv
394 , isQueueDone = workDone sv
396 , svarStyle = AsyncVar
397 , svarStopStyle = StopNone
398 , svarStopBy = undefined
400 , workerCount = active
401 , accountThread = delThread sv
402 , workerStopMVar = undefined
404 , svarInspectMode = getInspectMode st
406 , aheadWorkQueue = undefined
407 , outputHeap = undefined
412 case getStreamRate st of
414 case getYieldLimit st of
415 Nothing -> getSVar sv readOutputQBounded
419 Just _ -> getSVar sv readOutputQBounded
421 isWorkFinishedLimited
424 case getYieldLimit st of
425 Nothing -> getSVar sv readOutputQPaced
429 Just _ -> getSVar sv readOutputQPaced
431 isWorkFinishedLimited
435getFifoSVar :: forall m a. MonadAsync m
436 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
437getFifoSVar st mrun = do
438 outQ <- newIORef ([], 0)
439 outQMv <- newEmptyMVar
441 wfw <- newIORef False
442 running <- newIORef S.empty
444 yl <- case getYieldLimit st of
445 Nothing -> return Nothing
446 Just x -> Just <$> newIORef x
447 rateInfo <- getYieldRateInfo st
449 stats <- newSVarStats
452 let isWorkFinished _ = nullQ q
453 let isWorkFinishedLimited sv = do
455 case remainingWork sv of
459 Nothing -> return False
461 return $ qEmpty || yieldsDone
463 let getSVar :: SVar Stream m a
464 -> (SVar Stream m a -> m [ChildEvent a])
465 -> (SVar Stream m a -> m Bool)
466 -> (SVar Stream m a -> IO Bool)
467 -> (LinkedQueue (RunInIO m, Stream m a)
473 getSVar sv readOutput postProc workDone wloop = SVar
475 , outputQueueFromConsumer = undefined
477 , maxBufferLimit = getMaxBuffer st
478 , pushBufferSpace = undefined
479 , pushBufferPolicy = undefined
480 , pushBufferMVar = undefined
481 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
482 , yieldRateInfo = rateInfo
483 , outputDoorBell = outQMv
484 , outputDoorBellFromConsumer = undefined
485 , readOutputQ = readOutput sv
486 , postProcess = postProc sv
487 , workerThreads = running
488 , workLoop = wloop q st{streamVar = Just sv} sv
489 , enqueue = enqueueFIFO sv q
490 , isWorkDone = workDone sv
491 , isQueueDone = workDone sv
493 , svarStyle = WAsyncVar
494 , svarStopStyle = StopNone
495 , svarStopBy = undefined
497 , workerCount = active
498 , accountThread = delThread sv
499 , workerStopMVar = undefined
501 , svarInspectMode = getInspectMode st
503 , aheadWorkQueue = undefined
504 , outputHeap = undefined
509 case getStreamRate st of
511 case getYieldLimit st of
512 Nothing -> getSVar sv readOutputQBounded
516 Just _ -> getSVar sv readOutputQBounded
518 isWorkFinishedLimited
521 case getYieldLimit st of
522 Nothing -> getSVar sv readOutputQPaced
526 Just _ -> getSVar sv readOutputQPaced
528 isWorkFinishedLimited
532{-# INLINABLE newAsyncVar #-}
533newAsyncVar :: MonadAsync m
534 => State Stream m a -> Stream m a -> m (SVar Stream m a)
536 mrun <- captureMonadState
537 sv <- liftIO $ getLifoSVar st mrun
540-- | Generate a stream asynchronously to keep it buffered, lazily consume
545{-# INLINABLE mkAsyncK #-}
546mkAsyncK :: MonadAsync m => Stream m a -> Stream m a
547mkAsyncK m = K.mkStream $ \st yld sng stp -> do
548 sv <- newAsyncVar (adaptState st) m
549 K.foldStream st yld sng stp $ getSerialT $ fromSVar sv
552-- This is slightly faster than the CPS version above
554{-# INLINE_NORMAL mkAsyncD #-}
555mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
556mkAsyncD m = D.Stream step Nothing
559 step gst Nothing = do
560 sv <- newAsyncVar gst (D.toStreamK m)
561 return $ D.Skip $ Just $ fromSVarD sv
563 step gst (Just (D.UnStream step1 st)) = do
566 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
567 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
570-- XXX We can pass captureMonadState instead of using MonadAsync
571-- | Create a new SVar and enqueue one stream computation on it.
572{-# INLINABLE newWAsyncVar #-}
573newWAsyncVar :: MonadAsync m
574 => State Stream m a -> Stream m a -> m (SVar Stream m a)
575newWAsyncVar st m = do
576 mrun <- captureMonadState
577 sv <- liftIO $ getFifoSVar st mrun
578 -- XXX Use just Stream and IO in all the functions below
579 -- XXX pass mrun instead of calling captureMonadState again inside it
582------------------------------------------------------------------------------
583-- Running streams concurrently
584------------------------------------------------------------------------------
586-- Concurrency rate control.
588-- Our objective is to create more threads on demand if the consumer is running
589-- faster than us. As soon as we encounter a concurrent composition we create a
590-- push pull pair of threads. We use an SVar for communication between the
591-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
592-- The producer creates more threads if the SVar drains and becomes empty, that
593-- is the consumer is running faster.
595-- XXX Note 1: This mechanism can be problematic if the initial production
596-- latency is high, we may end up creating too many threads. So we need some
597-- way to monitor and use the latency as well. Having a limit on the dispatches
598-- (programmer controlled) may also help.
600-- TBD Note 2: We may want to run computations at the lower level of the
601-- composition tree serially even when they are composed using a parallel
602-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
603-- place of 'wAsync'. If we find that an SVar immediately above a computation
604-- gets drained empty we can switch to parallelizing the computation. For that
605-- we can use a state flag to fork the rest of the computation at any point of
606-- time inside the Monad bind operation if the consumer is running at a faster
609-- TBD Note 3: the binary operation ('parallel') composition allows us to
610-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
611-- chunksizes we will need to compose the parallel actions using a data
612-- constructor (A Free container) instead so that we can divide it in chunks of
613-- arbitrary size before dispatching. If the stream is composed of
614-- hierarchically composed grains of different sizes then we can always switch
615-- to a desired granularity depending on the consumer speed.
617-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
618-- into just the number of CPUs.
620-- | Join two computations on the currently running 'SVar' queue for concurrent
621-- execution. When we are using parallel composition, an SVar is passed around
622-- as a state variable. We try to schedule a new parallel computation on the
623-- SVar passed to us. The first time, when no SVar exists, a new SVar is
624-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
625-- already scheduled on the SVar is further evaluated. For example, when (a
626-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
627-- the current scheduler queue.
629-- The 'SVarStyle' required by the current composition context is passed as one
630-- of the parameters. If the scheduling and composition style of the new
631-- computation being scheduled is different than the style of the current SVar,
632-- then we create a new SVar and schedule it on that. The newly created SVar
633-- joins as one of the computations on the current SVar queue.
635-- Cases when we need to switch to a new SVar:
637-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
638-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
639-- because of the scheduling policy change.
640-- * if we 'adapt' a stream of type 'async' to a stream of type
641-- 'Parallel', we create a new SVar at the transitioning bind.
642-- * When the stream is switching from disjunctive composition to conjunctive
643-- composition and vice-versa we create a new SVar to isolate the scheduling
646forkSVarAsync :: MonadAsync m
647 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
648forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do
650 AsyncVar -> newAsyncVar st (concurrently m1 m2)
651 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
652 _ -> error "illegal svar type"
653 K.foldStream st yld sng stp $ getSerialT $ fromSVar sv
655 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
656 runInIO <- captureMonadState
657 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
658 K.foldStreamShared st yld sng stp ma
660{-# INLINE joinStreamVarAsync #-}
661joinStreamVarAsync :: MonadAsync m
662 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
663joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->
665 Just sv | svarStyle sv == style -> do
666 runInIO <- captureMonadState
667 liftIO $ enqueue sv (runInIO, m2)
668 K.foldStreamShared st yld sng stp m1
669 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
671------------------------------------------------------------------------------
672-- Semigroup and Monoid style compositions for parallel actions
673------------------------------------------------------------------------------
676asyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
677asyncK = joinStreamVarAsync AsyncVar
679-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
680-- huge difference in the performance of consM in IsStream instance even we
681-- have a SPECIALIZE in the instance.
683-- | XXX we can implement it more efficienty by directly implementing instead
684-- of combining streams using async.
685{-# INLINE consMAsync #-}
686{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
687consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
688consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r
690------------------------------------------------------------------------------
692------------------------------------------------------------------------------
694-- | For 'AsyncT' streams:
697-- (<>) = 'Streamly.Prelude.async'
698-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
701-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
702-- executed concurrently a la the 'async' combinator, producing results and
703-- side effects of iterations out of order:
706-- Stream.toList $ Stream.fromAsync $ do
707-- x <- Stream.fromList [2,1] -- foreach x in stream
708-- Stream.fromEffect $ delay x
714-- Nested monad binds behave like nested @for@ loops with nested iterations
715-- executed concurrently, a la the 'async' combinator:
718-- Stream.toList $ Stream.fromAsync $ do
719-- x <- Stream.fromList [1,2] -- foreach x in stream
720-- y <- Stream.fromList [2,4] -- foreach y in stream
721-- Stream.fromEffect $ delay (x + y)
729-- The behavior can be explained as follows. All the iterations corresponding
730-- to the element @1@ in the first stream constitute one output stream and all
731-- the iterations corresponding to @2@ constitute another output stream and
732-- these two output streams are merged using 'async'.
734-- /Since: 0.1.0 ("Streamly")/
737newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
738 deriving (MonadTrans)
740-- | A demand driven left biased parallely composing IO stream of elements of
741-- type @a@. See 'AsyncT' documentation for more details.
743-- /Since: 0.2.0 ("Streamly")/
746type Async = AsyncT IO
748------------------------------------------------------------------------------
750------------------------------------------------------------------------------
752-- Monomorphically typed version of "async" for better performance of Semigroup
755{-# SPECIALIZE append :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
756append :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
757append (AsyncT m1) (AsyncT m2) = AsyncT $ asyncK m1 m2
759instance MonadAsync m => Semigroup (AsyncT m a) where
762------------------------------------------------------------------------------
764------------------------------------------------------------------------------
766instance MonadAsync m => Monoid (AsyncT m a) where
767 mempty = AsyncT K.nil
770------------------------------------------------------------------------------
772------------------------------------------------------------------------------
774{-# INLINE apAsync #-}
775{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
776apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
777apAsync (AsyncT m1) (AsyncT m2) =
778 let f x1 = K.concatMapWith asyncK (pure . x1) m2
779 in AsyncT $ K.concatMapWith asyncK f m1
781instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
783 pure = AsyncT . K.fromPure
788------------------------------------------------------------------------------
790------------------------------------------------------------------------------
792-- GHC: if we change the implementation of bindWith with arguments in a
793-- different order we see a significant performance degradation (~2x).
794{-# INLINE bindAsync #-}
795{-# SPECIALIZE bindAsync ::
796 AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
797bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
798bindAsync (AsyncT m) f = AsyncT $ K.bindWith asyncK m (getAsyncT . f)
800-- GHC: if we specify arguments in the definition of (>>=) we see a significant
801-- performance degradation (~2x).
802instance MonadAsync m => Monad (AsyncT m) where
806------------------------------------------------------------------------------
808------------------------------------------------------------------------------
810MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
812------------------------------------------------------------------------------
814------------------------------------------------------------------------------
816{-# INLINE wAsyncK #-}
817wAsyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
818wAsyncK = joinStreamVarAsync WAsyncVar
820-- | XXX we can implement it more efficienty by directly implementing instead
821-- of combining streams using wAsync.
822{-# INLINE consMWAsync #-}
823{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
824consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
825consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r
827-- | For 'WAsyncT' streams:
830-- (<>) = 'Streamly.Prelude.wAsync'
831-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
834-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
835-- executed concurrently a la the 'wAsync' combinator, producing results and
836-- side effects of iterations out of order:
839-- Stream.toList $ Stream.fromWAsync $ do
840-- x <- Stream.fromList [2,1] -- foreach x in stream
841-- Stream.fromEffect $ delay x
847-- Nested monad binds behave like nested @for@ loops with nested iterations
848-- executed concurrently, a la the 'wAsync' combinator:
851-- Stream.toList $ Stream.fromWAsync $ do
852-- x <- Stream.fromList [1,2] -- foreach x in stream
853-- y <- Stream.fromList [2,4] -- foreach y in stream
854-- Stream.fromEffect $ delay (x + y)
862-- The behavior can be explained as follows. All the iterations corresponding
863-- to the element @1@ in the first stream constitute one 'WAsyncT' output
864-- stream and all the iterations corresponding to @2@ constitute another
865-- 'WAsyncT' output stream and these two output streams are merged using
868-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
869-- contrast to the depth wise scheduling behavior of 'AsyncT'.
871-- /Since: 0.2.0 ("Streamly")/
875-- XXX This documentation is redundant, need to be cleaned up/removed.
878-- import qualified "Streamly.Prelude" as S
879-- import Control.Concurrent
881-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
887-- For this example, we are using @maxThreads 1@ so that concurrent thread
888-- scheduling does not affect the results and make them unpredictable. Let's
889-- now take a more general example:
892-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
895-- [1,4,2,7,5,3,8,6,9]
898-- This is how the execution of the above stream proceeds:
900-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
901-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
902-- queue is represented by the rightmost item.
903-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
904-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
905-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
906-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
907-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
908-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
909-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
910-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
911-- 4. Note that the scheduler queue expands by one more stream component in
912-- every pass because one more @<>@ is broken down into two components. At this
913-- point there are no more @<>@ operations to be broken down further and the
914-- queue has reached its maximum size. Now these streams are scheduled in
915-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
917-- As we see above, in a right associated expression composed with @<>@, only
918-- one @<>@ operation is broken down into two components in one execution,
919-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
920-- scheduler passes to expand the whole expression. By the time @n-th@
921-- component is added to the scheduler queue, the first component would have
922-- received @n@ scheduler passes.
924-- Since all streams get interleaved, this operation is not suitable for
925-- folding an infinite lazy container of infinite size streams. However, if
926-- the streams are small, the streams on the left may get finished before more
927-- streams are added to the scheduler queue from the right side of the
928-- expression, so it may be possible to fold an infinite lazy container of
929-- streams. For example, if the streams are of size @n@ then at most @n@
930-- streams would be in the scheduler queue at a time.
932-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
933-- therefore the output of 'WAsyncT' even with a single thread of execution is
934-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
935-- its scheduling behavior.
937-- Any exceptions generated by a constituent stream are propagated to the
938-- output stream. The output and exceptions from a single stream are guaranteed
939-- to arrive in the same order in the resulting stream as they were generated
940-- in the input stream. However, the relative ordering of elements from
941-- different streams in the resulting stream can vary depending on scheduling
942-- and generation delays.
944-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
945-- concurrently using a round robin scheduling.
948-- main = S.'drain' . S.'fromWAsync' $ do
949-- n <- return 3 \<\> return 2 \<\> return 1
951-- threadDelay (n * 1000000)
952-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
955-- ThreadId 40: Delay 1
956-- ThreadId 39: Delay 2
957-- ThreadId 38: Delay 3
960newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
961 deriving (MonadTrans)
963-- | A round robin parallely composing IO stream of elements of type @a@.
964-- See 'WAsyncT' documentation for more details.
966-- /Since: 0.2.0 ("Streamly")/
969type WAsync = WAsyncT IO
971------------------------------------------------------------------------------
973------------------------------------------------------------------------------
975{-# INLINE wAppend #-}
976{-# SPECIALIZE wAppend :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
977wAppend :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
978wAppend (WAsyncT m1) (WAsyncT m2) = WAsyncT $ wAsyncK m1 m2
980instance MonadAsync m => Semigroup (WAsyncT m a) where
983------------------------------------------------------------------------------
985------------------------------------------------------------------------------
987instance MonadAsync m => Monoid (WAsyncT m a) where
988 mempty = WAsyncT K.nil
991------------------------------------------------------------------------------
993------------------------------------------------------------------------------
995{-# INLINE apWAsync #-}
996{-# SPECIALIZE apWAsync ::
997 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
998apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
999apWAsync (WAsyncT m1) (WAsyncT m2) =
1000 let f x1 = K.concatMapWith wAsyncK (pure . x1) m2
1001 in WAsyncT $ K.concatMapWith wAsyncK f m1
1003-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1004-- performance degradation (~2x).
1005instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1006 pure = WAsyncT . K.fromPure
1009------------------------------------------------------------------------------
1011------------------------------------------------------------------------------
1013-- GHC: if we change the implementation of bindWith with arguments in a
1014-- different order we see a significant performance degradation (~2x).
1015{-# INLINE bindWAsync #-}
1016{-# SPECIALIZE bindWAsync ::
1017 WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1018bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1019bindWAsync (WAsyncT m) f = WAsyncT $ K.bindWith wAsyncK m (getWAsyncT . f)
1021-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1022-- performance degradation (~2x).
1023instance MonadAsync m => Monad (WAsyncT m) where
1027------------------------------------------------------------------------------
1029------------------------------------------------------------------------------
1031MONAD_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