1{-# LANGUAGE UndecidableInstances #-}
6-- Module : Streamly.Internal.Data.Stream.IsStream.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.IsStream.Async
41import Control.Concurrent (myThreadId)
42import Control.Monad.Base (MonadBase(..), liftBaseDefault)
43import Control.Monad.Catch (MonadThrow, throwM)
44import Control.Concurrent.MVar (newEmptyMVar)
45-- import Control.Monad.Error.Class (MonadError(..))
46import Control.Monad.IO.Class (MonadIO(..))
47import Control.Monad.Reader.Class (MonadReader(..))
48import Control.Monad.State.Class (MonadState(..))
49import Control.Monad.Trans.Class (MonadTrans(lift))
50import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)
51import Data.IORef (IORef, newIORef, readIORef)
52import Data.Maybe (fromJust)
53#if __GLASGOW_HASKELL__ < 808
54import Data.Semigroup (Semigroup(..))
57import Prelude hiding (map)
58import qualified Data.Set as S
60import Streamly.Internal.Control.Concurrent
61 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO, restoreM)
62import Streamly.Internal.Data.Atomics
63 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
64import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
65import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
67import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
68 (foldStreamShared, mkStream, foldStream, fromEffect
69 , nil, concatMapWith, fromPure, bindWith, withLocal)
70import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
71 (Stream(..), Step(..), mapM, toStreamK, fromStreamK)
72import qualified Streamly.Internal.Data.Stream as Stream (toStreamK)
74import Streamly.Internal.Data.SVar
76#include "Instances.hs"
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-------------------------------------------------------------------------------
91-------------------------------------------------------------------------------
93-- Note: For purely right associated expressions this queue should have at most
94-- one element. It grows to more than one when we have left associcated
95-- expressions. Large left associated compositions can grow this to a
97{-# INLINE enqueueLIFO #-}
99 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
100enqueueLIFO sv q m = do
101 atomicModifyIORefCAS_ q $ \ms -> m : ms
104data WorkerStatus = Continue | Suspend
106{-# INLINE workLoopLIFO #-}
109 => IORef [(RunInIO m, Stream m a)]
114workLoopLIFO q st sv winfo = run
118 stop = liftIO $ sendStop sv winfo
123 Just (RunInIO runin, m) -> do
124 -- XXX when we finish we need to send the monadic state back to
125 -- the parent so that the state can be merged back. We capture
126 -- and return the state in the stop continuation.
128 -- Instead of using the run function we can just restore the
129 -- monad state here. That way it can work easily for
130 -- distributed case as well.
131 r <- liftIO $ runin $
132 K.foldStreamShared st yieldk single (return Continue) m
139 res <- liftIO $ sendYield sv winfo (ChildYield a)
140 return $ if res then Continue else Suspend
143 res <- liftIO $ sendYield sv winfo (ChildYield a)
145 then K.foldStreamShared st yieldk single (return Continue) r
147 runInIO <- askRunInIO
148 liftIO $ enqueueLIFO sv q (runInIO, r)
151 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
153 x : xs -> (xs, Just x)
155-- We duplicate workLoop for yield limit and no limit cases because it has
156-- around 40% performance overhead in the worst case.
158-- XXX we can pass yinfo directly as an argument here so that we do not have to
159-- make a check every time.
160{-# INLINE workLoopLIFOLimited #-}
162 :: forall m a. MonadRunInIO m
163 => IORef [(RunInIO m, Stream m a)]
168workLoopLIFOLimited q st sv winfo = run
172 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
173 stop = liftIO $ sendStop sv winfo
178 Just (RunInIO runin, m) -> do
179 -- XXX This is just a best effort minimization of concurrency
180 -- to the yield limit. If the stream is made of concurrent
181 -- streams we do not reserve the yield limit in the constituent
182 -- streams before executing the action. This can be done
183 -- though, by sharing the yield limit ref with downstream
184 -- actions via state passing. Just a todo.
185 yieldLimitOk <- liftIO $ decrementYieldLimit sv
188 r <- liftIO $ runin $
189 K.foldStreamShared st yieldk single incrContinue m
194 -- Avoid any side effects, undo the yield limit decrement if we
195 -- never yielded anything.
197 enqueueLIFO sv q (RunInIO runin, m)
198 incrementYieldLimit sv
202 res <- liftIO $ sendYield sv winfo (ChildYield a)
203 return $ if res then Continue else Suspend
205 -- XXX can we pass on the yield limit downstream to limit the concurrency
206 -- of constituent streams.
208 res <- liftIO $ sendYield sv winfo (ChildYield a)
209 yieldLimitOk <- liftIO $ decrementYieldLimit sv
210 if res && yieldLimitOk
211 then K.foldStreamShared st yieldk single incrContinue r
213 runInIO <- askRunInIO
214 liftIO $ incrementYieldLimit sv
215 liftIO $ enqueueLIFO sv q (runInIO, r)
218 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
220 x : xs -> (xs, Just x)
222-------------------------------------------------------------------------------
224-------------------------------------------------------------------------------
226-- XXX we can use the Ahead style sequence/heap mechanism to make the best
227-- effort to always try to finish the streams on the left side of an expression
228-- first as long as possible.
230{-# INLINE enqueueFIFO #-}
233 -> LinkedQueue (RunInIO m, t m a)
234 -> (RunInIO m, t m a)
236enqueueFIFO sv q m = do
240-- XXX we can remove sv as it is derivable from st
242{-# INLINE workLoopFIFO #-}
245 => LinkedQueue (RunInIO m, Stream m a)
250workLoopFIFO q st sv winfo = run
254 stop = liftIO $ sendStop sv winfo
256 work <- liftIO $ tryPopR q
259 Just (RunInIO runin, m) -> do
260 r <- liftIO $ runin $
261 K.foldStreamShared st yieldk single (return Continue) m
268 res <- liftIO $ sendYield sv winfo (ChildYield a)
269 return $ if res then Continue else Suspend
271 -- XXX in general we would like to yield "n" elements from a single stream
272 -- before moving on to the next. Single element granularity could be too
273 -- expensive in certain cases. Similarly, we can use time limit for
276 res <- liftIO $ sendYield sv winfo (ChildYield a)
277 runInIO <- askRunInIO
278 liftIO $ enqueueFIFO sv q (runInIO, r)
279 return $ if res then Continue else Suspend
281{-# INLINE workLoopFIFOLimited #-}
283 :: forall m a. MonadRunInIO m
284 => LinkedQueue (RunInIO m, Stream m a)
289workLoopFIFOLimited q st sv winfo = run
293 stop = liftIO $ sendStop sv winfo
294 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
296 work <- liftIO $ tryPopR q
299 Just (RunInIO runin, m) -> do
300 yieldLimitOk <- liftIO $ decrementYieldLimit sv
303 r <- liftIO $ runin $
304 K.foldStreamShared st yieldk single incrContinue m
310 enqueueFIFO sv q (RunInIO runin, m)
311 incrementYieldLimit sv
315 res <- liftIO $ sendYield sv winfo (ChildYield a)
316 return $ if res then Continue else Suspend
319 res <- liftIO $ sendYield sv winfo (ChildYield a)
320 runInIO <- askRunInIO
321 liftIO $ enqueueFIFO sv q (runInIO, r)
322 yieldLimitOk <- liftIO $ decrementYieldLimit sv
323 if res && yieldLimitOk
326 incrementYieldLimit sv
329-------------------------------------------------------------------------------
331-- This code belongs in SVar.hs but is kept here for perf reasons
332-------------------------------------------------------------------------------
334-- XXX we have this function in this file because passing runStreamLIFO as a
335-- function argument to this function results in a perf degradation of more
336-- than 10%. Need to investigate what the root cause is.
337-- Interestingly, the same thing does not make any difference for Ahead.
338getLifoSVar :: forall m a. MonadAsync m
339 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
340getLifoSVar st mrun = do
341 outQ <- newIORef ([], 0)
342 outQMv <- newEmptyMVar
344 wfw <- newIORef False
345 running <- newIORef S.empty
346 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
347 yl <- case getYieldLimit st of
348 Nothing -> return Nothing
349 Just x -> Just <$> newIORef x
350 rateInfo <- getYieldRateInfo st
352 stats <- newSVarStats
355 let isWorkFinished _ = null <$> readIORef q
357 let isWorkFinishedLimited sv = do
359 case remainingWork sv of
363 Nothing -> return False
364 qEmpty <- null <$> readIORef q
365 return $ qEmpty || yieldsDone
367 let getSVar :: SVar Stream m a
368 -> (SVar Stream m a -> m [ChildEvent a])
369 -> (SVar Stream m a -> m Bool)
370 -> (SVar Stream m a -> IO Bool)
371 -> (IORef [(RunInIO m, Stream m a)]
377 getSVar sv readOutput postProc workDone wloop = SVar
379 , outputQueueFromConsumer = undefined
381 , maxBufferLimit = getMaxBuffer st
382 , pushBufferSpace = undefined
383 , pushBufferPolicy = undefined
384 , pushBufferMVar = undefined
385 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
386 , yieldRateInfo = rateInfo
387 , outputDoorBell = outQMv
388 , outputDoorBellFromConsumer = undefined
389 , readOutputQ = readOutput sv
390 , postProcess = postProc sv
391 , workerThreads = running
392 , workLoop = wloop q st{streamVar = Just sv} sv
393 , enqueue = enqueueLIFO sv q
394 , isWorkDone = workDone sv
395 , isQueueDone = workDone sv
397 , svarStyle = AsyncVar
398 , svarStopStyle = StopNone
399 , svarStopBy = undefined
401 , workerCount = active
402 , accountThread = delThread sv
403 , workerStopMVar = undefined
405 , svarInspectMode = getInspectMode st
407 , aheadWorkQueue = undefined
408 , outputHeap = undefined
413 case getStreamRate st of
415 case getYieldLimit st of
416 Nothing -> getSVar sv readOutputQBounded
420 Just _ -> getSVar sv readOutputQBounded
422 isWorkFinishedLimited
425 case getYieldLimit st of
426 Nothing -> getSVar sv readOutputQPaced
430 Just _ -> getSVar sv readOutputQPaced
432 isWorkFinishedLimited
436getFifoSVar :: forall m a. MonadAsync m
437 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
438getFifoSVar st mrun = do
439 outQ <- newIORef ([], 0)
440 outQMv <- newEmptyMVar
442 wfw <- newIORef False
443 running <- newIORef S.empty
445 yl <- case getYieldLimit st of
446 Nothing -> return Nothing
447 Just x -> Just <$> newIORef x
448 rateInfo <- getYieldRateInfo st
450 stats <- newSVarStats
453 let isWorkFinished _ = nullQ q
454 let isWorkFinishedLimited sv = do
456 case remainingWork sv of
460 Nothing -> return False
462 return $ qEmpty || yieldsDone
464 let getSVar :: SVar Stream m a
465 -> (SVar Stream m a -> m [ChildEvent a])
466 -> (SVar Stream m a -> m Bool)
467 -> (SVar Stream m a -> IO Bool)
468 -> (LinkedQueue (RunInIO m, Stream m a)
474 getSVar sv readOutput postProc workDone wloop = SVar
476 , outputQueueFromConsumer = undefined
478 , maxBufferLimit = getMaxBuffer st
479 , pushBufferSpace = undefined
480 , pushBufferPolicy = undefined
481 , pushBufferMVar = undefined
482 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
483 , yieldRateInfo = rateInfo
484 , outputDoorBell = outQMv
485 , outputDoorBellFromConsumer = undefined
486 , readOutputQ = readOutput sv
487 , postProcess = postProc sv
488 , workerThreads = running
489 , workLoop = wloop q st{streamVar = Just sv} sv
490 , enqueue = enqueueFIFO sv q
491 , isWorkDone = workDone sv
492 , isQueueDone = workDone sv
494 , svarStyle = WAsyncVar
495 , svarStopStyle = StopNone
496 , svarStopBy = undefined
498 , workerCount = active
499 , accountThread = delThread sv
500 , workerStopMVar = undefined
502 , svarInspectMode = getInspectMode st
504 , aheadWorkQueue = undefined
505 , outputHeap = undefined
510 case getStreamRate st of
512 case getYieldLimit st of
513 Nothing -> getSVar sv readOutputQBounded
517 Just _ -> getSVar sv readOutputQBounded
519 isWorkFinishedLimited
522 case getYieldLimit st of
523 Nothing -> getSVar sv readOutputQPaced
527 Just _ -> getSVar sv readOutputQPaced
529 isWorkFinishedLimited
533{-# INLINABLE newAsyncVar #-}
534newAsyncVar :: MonadAsync m
535 => State Stream m a -> Stream m a -> m (SVar Stream m a)
538 sv <- liftIO $ getLifoSVar st mrun
541-- | Generate a stream asynchronously to keep it buffered, lazily consume
546{-# INLINABLE mkAsyncK #-}
547mkAsyncK :: MonadAsync m => Stream m a -> Stream m a
548mkAsyncK m = K.mkStream $ \st yld sng stp -> do
549 sv <- newAsyncVar (adaptState st) m
550 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
553-- This is slightly faster than the CPS version above
555{-# INLINE_NORMAL mkAsyncD #-}
556mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
557mkAsyncD m = D.Stream step Nothing
560 step gst Nothing = do
561 sv <- newAsyncVar gst (D.toStreamK m)
562 return $ D.Skip $ Just $ fromSVarD sv
564 step gst (Just (D.UnStream step1 st)) = do
567 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
568 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
571-- XXX We can pass captureMonadState instead of using MonadAsync
572-- | Create a new SVar and enqueue one stream computation on it.
573{-# INLINABLE newWAsyncVar #-}
574newWAsyncVar :: MonadAsync m
575 => State Stream m a -> Stream m a -> m (SVar Stream m a)
576newWAsyncVar st m = do
578 sv <- liftIO $ getFifoSVar st mrun
579 -- XXX Use just Stream and IO in all the functions below
580 -- XXX pass mrun instead of calling captureMonadState again inside it
583------------------------------------------------------------------------------
584-- Running streams concurrently
585------------------------------------------------------------------------------
587-- Concurrency rate control.
589-- Our objective is to create more threads on demand if the consumer is running
590-- faster than us. As soon as we encounter a concurrent composition we create a
591-- push pull pair of threads. We use an SVar for communication between the
592-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
593-- The producer creates more threads if the SVar drains and becomes empty, that
594-- is the consumer is running faster.
596-- XXX Note 1: This mechanism can be problematic if the initial production
597-- latency is high, we may end up creating too many threads. So we need some
598-- way to monitor and use the latency as well. Having a limit on the dispatches
599-- (programmer controlled) may also help.
601-- TBD Note 2: We may want to run computations at the lower level of the
602-- composition tree serially even when they are composed using a parallel
603-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
604-- place of 'wAsync'. If we find that an SVar immediately above a computation
605-- gets drained empty we can switch to parallelizing the computation. For that
606-- we can use a state flag to fork the rest of the computation at any point of
607-- time inside the Monad bind operation if the consumer is running at a faster
610-- TBD Note 3: the binary operation ('parallel') composition allows us to
611-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
612-- chunksizes we will need to compose the parallel actions using a data
613-- constructor (A Free container) instead so that we can divide it in chunks of
614-- arbitrary size before dispatching. If the stream is composed of
615-- hierarchically composed grains of different sizes then we can always switch
616-- to a desired granularity depending on the consumer speed.
618-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
619-- into just the number of CPUs.
621-- | Join two computations on the currently running 'SVar' queue for concurrent
622-- execution. When we are using parallel composition, an SVar is passed around
623-- as a state variable. We try to schedule a new parallel computation on the
624-- SVar passed to us. The first time, when no SVar exists, a new SVar is
625-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
626-- already scheduled on the SVar is further evaluated. For example, when (a
627-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
628-- the current scheduler queue.
630-- The 'SVarStyle' required by the current composition context is passed as one
631-- of the parameters. If the scheduling and composition style of the new
632-- computation being scheduled is different than the style of the current SVar,
633-- then we create a new SVar and schedule it on that. The newly created SVar
634-- joins as one of the computations on the current SVar queue.
636-- Cases when we need to switch to a new SVar:
638-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
639-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
640-- because of the scheduling policy change.
641-- * if we 'adapt' a stream of type 'async' to a stream of type
642-- 'Parallel', we create a new SVar at the transitioning bind.
643-- * When the stream is switching from disjunctive composition to conjunctive
644-- composition and vice-versa we create a new SVar to isolate the scheduling
647forkSVarAsync :: MonadAsync m
648 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
649forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do
651 AsyncVar -> newAsyncVar st (concurrently m1 m2)
652 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
653 _ -> error "illegal svar type"
654 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
656 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
657 runInIO <- askRunInIO
658 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
659 K.foldStreamShared st yld sng stp ma
661{-# INLINE joinStreamVarAsync #-}
662joinStreamVarAsync :: MonadAsync m
663 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
664joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->
666 Just sv | svarStyle sv == style -> do
667 runInIO <- askRunInIO
668 liftIO $ enqueue sv (runInIO, m2)
669 K.foldStreamShared st yld sng stp m1
670 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
672------------------------------------------------------------------------------
673-- Semigroup and Monoid style compositions for parallel actions
674------------------------------------------------------------------------------
677asyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
678asyncK = joinStreamVarAsync AsyncVar
680-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
681-- huge difference in the performance of consM in IsStream instance even we
682-- have a SPECIALIZE in the instance.
684-- | XXX we can implement it more efficienty by directly implementing instead
685-- of combining streams using async.
686{-# INLINE consMAsync #-}
687{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
688consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
689consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r
691------------------------------------------------------------------------------
693------------------------------------------------------------------------------
695-- | For 'AsyncT' streams:
698-- (<>) = 'Streamly.Prelude.async'
699-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
702-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
703-- executed concurrently a la the 'async' combinator, producing results and
704-- side effects of iterations out of order:
707-- Stream.toList $ Stream.fromAsync $ do
708-- x <- Stream.fromList [2,1] -- foreach x in stream
709-- Stream.fromEffect $ delay x
715-- Nested monad binds behave like nested @for@ loops with nested iterations
716-- executed concurrently, a la the 'async' combinator:
719-- Stream.toList $ Stream.fromAsync $ do
720-- x <- Stream.fromList [1,2] -- foreach x in stream
721-- y <- Stream.fromList [2,4] -- foreach y in stream
722-- Stream.fromEffect $ delay (x + y)
730-- The behavior can be explained as follows. All the iterations corresponding
731-- to the element @1@ in the first stream constitute one output stream and all
732-- the iterations corresponding to @2@ constitute another output stream and
733-- these two output streams are merged using 'async'.
735-- /Since: 0.1.0 ("Streamly")/
738newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
739 deriving (MonadTrans)
741-- | A demand driven left biased parallely composing IO stream of elements of
742-- type @a@. See 'AsyncT' documentation for more details.
744-- /Since: 0.2.0 ("Streamly")/
747type Async = AsyncT IO
749------------------------------------------------------------------------------
751------------------------------------------------------------------------------
753-- Monomorphically typed version of "async" for better performance of Semigroup
756{-# SPECIALIZE append :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
757append :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
758append (AsyncT m1) (AsyncT m2) = AsyncT $ asyncK m1 m2
760instance MonadAsync m => Semigroup (AsyncT m a) where
763------------------------------------------------------------------------------
765------------------------------------------------------------------------------
767instance MonadAsync m => Monoid (AsyncT m a) where
768 mempty = AsyncT K.nil
771------------------------------------------------------------------------------
773------------------------------------------------------------------------------
775{-# INLINE apAsync #-}
776{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
777apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
778apAsync (AsyncT m1) (AsyncT m2) =
779 let f x1 = K.concatMapWith asyncK (pure . x1) m2
780 in AsyncT $ K.concatMapWith asyncK f m1
782instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
784 pure = AsyncT . K.fromPure
789------------------------------------------------------------------------------
791------------------------------------------------------------------------------
793-- GHC: if we change the implementation of bindWith with arguments in a
794-- different order we see a significant performance degradation (~2x).
795{-# INLINE bindAsync #-}
796{-# SPECIALIZE bindAsync ::
797 AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
798bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
799bindAsync (AsyncT m) f = AsyncT $ K.bindWith asyncK m (getAsyncT . f)
801-- GHC: if we specify arguments in the definition of (>>=) we see a significant
802-- performance degradation (~2x).
803instance MonadAsync m => Monad (AsyncT m) where
807------------------------------------------------------------------------------
809------------------------------------------------------------------------------
811MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
813------------------------------------------------------------------------------
815------------------------------------------------------------------------------
817{-# INLINE wAsyncK #-}
818wAsyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
819wAsyncK = joinStreamVarAsync WAsyncVar
821-- | XXX we can implement it more efficienty by directly implementing instead
822-- of combining streams using wAsync.
823{-# INLINE consMWAsync #-}
824{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
825consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
826consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r
828-- | For 'WAsyncT' streams:
831-- (<>) = 'Streamly.Prelude.wAsync'
832-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
835-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
836-- executed concurrently a la the 'wAsync' combinator, producing results and
837-- side effects of iterations out of order:
840-- Stream.toList $ Stream.fromWAsync $ do
841-- x <- Stream.fromList [2,1] -- foreach x in stream
842-- Stream.fromEffect $ delay x
848-- Nested monad binds behave like nested @for@ loops with nested iterations
849-- executed concurrently, a la the 'wAsync' combinator:
852-- Stream.toList $ Stream.fromWAsync $ do
853-- x <- Stream.fromList [1,2] -- foreach x in stream
854-- y <- Stream.fromList [2,4] -- foreach y in stream
855-- Stream.fromEffect $ delay (x + y)
863-- The behavior can be explained as follows. All the iterations corresponding
864-- to the element @1@ in the first stream constitute one 'WAsyncT' output
865-- stream and all the iterations corresponding to @2@ constitute another
866-- 'WAsyncT' output stream and these two output streams are merged using
869-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
870-- contrast to the depth wise scheduling behavior of 'AsyncT'.
872-- /Since: 0.2.0 ("Streamly")/
876-- XXX This documentation is redundant, need to be cleaned up/removed.
879-- import qualified "Streamly.Prelude" as S
880-- import Control.Concurrent
882-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
888-- For this example, we are using @maxThreads 1@ so that concurrent thread
889-- scheduling does not affect the results and make them unpredictable. Let's
890-- now take a more general example:
893-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
896-- [1,4,2,7,5,3,8,6,9]
899-- This is how the execution of the above stream proceeds:
901-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
902-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
903-- queue is represented by the rightmost item.
904-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
905-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
906-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
907-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
908-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
909-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
910-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
911-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
912-- 4. Note that the scheduler queue expands by one more stream component in
913-- every pass because one more @<>@ is broken down into two components. At this
914-- point there are no more @<>@ operations to be broken down further and the
915-- queue has reached its maximum size. Now these streams are scheduled in
916-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
918-- As we see above, in a right associated expression composed with @<>@, only
919-- one @<>@ operation is broken down into two components in one execution,
920-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
921-- scheduler passes to expand the whole expression. By the time @n-th@
922-- component is added to the scheduler queue, the first component would have
923-- received @n@ scheduler passes.
925-- Since all streams get interleaved, this operation is not suitable for
926-- folding an infinite lazy container of infinite size streams. However, if
927-- the streams are small, the streams on the left may get finished before more
928-- streams are added to the scheduler queue from the right side of the
929-- expression, so it may be possible to fold an infinite lazy container of
930-- streams. For example, if the streams are of size @n@ then at most @n@
931-- streams would be in the scheduler queue at a time.
933-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
934-- therefore the output of 'WAsyncT' even with a single thread of execution is
935-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
936-- its scheduling behavior.
938-- Any exceptions generated by a constituent stream are propagated to the
939-- output stream. The output and exceptions from a single stream are guaranteed
940-- to arrive in the same order in the resulting stream as they were generated
941-- in the input stream. However, the relative ordering of elements from
942-- different streams in the resulting stream can vary depending on scheduling
943-- and generation delays.
945-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
946-- concurrently using a round robin scheduling.
949-- main = S.'drain' . S.'fromWAsync' $ do
950-- n <- return 3 \<\> return 2 \<\> return 1
952-- threadDelay (n * 1000000)
953-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
956-- ThreadId 40: Delay 1
957-- ThreadId 39: Delay 2
958-- ThreadId 38: Delay 3
961newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
962 deriving (MonadTrans)
964-- | A round robin parallely composing IO stream of elements of type @a@.
965-- See 'WAsyncT' documentation for more details.
967-- /Since: 0.2.0 ("Streamly")/
970type WAsync = WAsyncT IO
972------------------------------------------------------------------------------
974------------------------------------------------------------------------------
976{-# INLINE wAppend #-}
977{-# SPECIALIZE wAppend :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
978wAppend :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
979wAppend (WAsyncT m1) (WAsyncT m2) = WAsyncT $ wAsyncK m1 m2
981instance MonadAsync m => Semigroup (WAsyncT m a) where
984------------------------------------------------------------------------------
986------------------------------------------------------------------------------
988instance MonadAsync m => Monoid (WAsyncT m a) where
989 mempty = WAsyncT K.nil
992------------------------------------------------------------------------------
994------------------------------------------------------------------------------
996{-# INLINE apWAsync #-}
997{-# SPECIALIZE apWAsync ::
998 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
999apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
1000apWAsync (WAsyncT m1) (WAsyncT m2) =
1001 let f x1 = K.concatMapWith wAsyncK (pure . x1) m2
1002 in WAsyncT $ K.concatMapWith wAsyncK f m1
1004-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1005-- performance degradation (~2x).
1006instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1007 pure = WAsyncT . K.fromPure
1010------------------------------------------------------------------------------
1012------------------------------------------------------------------------------
1014-- GHC: if we change the implementation of bindWith with arguments in a
1015-- different order we see a significant performance degradation (~2x).
1016{-# INLINE bindWAsync #-}
1017{-# SPECIALIZE bindWAsync ::
1018 WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1019bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1020bindWAsync (WAsyncT m) f = WAsyncT $ K.bindWith wAsyncK m (getWAsyncT . f)
1022-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1023-- performance degradation (~2x).
1024instance MonadAsync m => Monad (WAsyncT m) where
1028------------------------------------------------------------------------------
1030------------------------------------------------------------------------------
1032MONAD_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