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)
54import Prelude hiding (map)
55import qualified Data.Set as S
57import Streamly.Internal.Control.Concurrent
58 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO, restoreM)
59import Streamly.Internal.Data.Atomics
60 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
61import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
62import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
64import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
65 (foldStreamShared, mkStream, foldStream, fromEffect
66 , nil, concatMapWith, fromPure, bindWith, withLocal)
67import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
68 (Stream(..), Step(..), mapM, toStreamK, fromStreamK)
69import qualified Streamly.Internal.Data.Stream as Stream (toStreamK)
71import Streamly.Internal.Data.SVar
73#include "Instances.hs"
76-- >>> import qualified Streamly.Prelude as Stream
77-- >>> import Control.Concurrent (threadDelay)
80-- threadDelay (n * 1000000) -- sleep for n seconds
81-- putStrLn (show n ++ " sec") -- print "n sec"
86-------------------------------------------------------------------------------
88-------------------------------------------------------------------------------
90-- Note: For purely right associated expressions this queue should have at most
91-- one element. It grows to more than one when we have left associcated
92-- expressions. Large left associated compositions can grow this to a
94{-# INLINE enqueueLIFO #-}
96 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
97enqueueLIFO sv q m = do
98 atomicModifyIORefCAS_ q $ \ms -> m : ms
101data WorkerStatus = Continue | Suspend
103{-# INLINE workLoopLIFO #-}
106 => IORef [(RunInIO m, Stream m a)]
111workLoopLIFO q st sv winfo = run
115 stop = liftIO $ sendStop sv winfo
120 Just (RunInIO runin, m) -> do
121 -- XXX when we finish we need to send the monadic state back to
122 -- the parent so that the state can be merged back. We capture
123 -- and return the state in the stop continuation.
125 -- Instead of using the run function we can just restore the
126 -- monad state here. That way it can work easily for
127 -- distributed case as well.
128 r <- liftIO $ runin $
129 K.foldStreamShared st yieldk single (return Continue) m
136 res <- liftIO $ sendYield sv winfo (ChildYield a)
137 return $ if res then Continue else Suspend
140 res <- liftIO $ sendYield sv winfo (ChildYield a)
142 then K.foldStreamShared st yieldk single (return Continue) r
144 runInIO <- askRunInIO
145 liftIO $ enqueueLIFO sv q (runInIO, r)
148 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
150 x : xs -> (xs, Just x)
152-- We duplicate workLoop for yield limit and no limit cases because it has
153-- around 40% performance overhead in the worst case.
155-- XXX we can pass yinfo directly as an argument here so that we do not have to
156-- make a check every time.
157{-# INLINE workLoopLIFOLimited #-}
159 :: forall m a. MonadRunInIO m
160 => IORef [(RunInIO m, Stream m a)]
165workLoopLIFOLimited q st sv winfo = run
169 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
170 stop = liftIO $ sendStop sv winfo
175 Just (RunInIO runin, m) -> do
176 -- XXX This is just a best effort minimization of concurrency
177 -- to the yield limit. If the stream is made of concurrent
178 -- streams we do not reserve the yield limit in the constituent
179 -- streams before executing the action. This can be done
180 -- though, by sharing the yield limit ref with downstream
181 -- actions via state passing. Just a todo.
182 yieldLimitOk <- liftIO $ decrementYieldLimit sv
185 r <- liftIO $ runin $
186 K.foldStreamShared st yieldk single incrContinue m
191 -- Avoid any side effects, undo the yield limit decrement if we
192 -- never yielded anything.
194 enqueueLIFO sv q (RunInIO runin, m)
195 incrementYieldLimit sv
199 res <- liftIO $ sendYield sv winfo (ChildYield a)
200 return $ if res then Continue else Suspend
202 -- XXX can we pass on the yield limit downstream to limit the concurrency
203 -- of constituent streams.
205 res <- liftIO $ sendYield sv winfo (ChildYield a)
206 yieldLimitOk <- liftIO $ decrementYieldLimit sv
207 if res && yieldLimitOk
208 then K.foldStreamShared st yieldk single incrContinue r
210 runInIO <- askRunInIO
211 liftIO $ incrementYieldLimit sv
212 liftIO $ enqueueLIFO sv q (runInIO, r)
215 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
217 x : xs -> (xs, Just x)
219-------------------------------------------------------------------------------
221-------------------------------------------------------------------------------
223-- XXX we can use the Ahead style sequence/heap mechanism to make the best
224-- effort to always try to finish the streams on the left side of an expression
225-- first as long as possible.
227{-# INLINE enqueueFIFO #-}
230 -> LinkedQueue (RunInIO m, t m a)
231 -> (RunInIO m, t m a)
233enqueueFIFO sv q m = do
237-- XXX we can remove sv as it is derivable from st
239{-# INLINE workLoopFIFO #-}
242 => LinkedQueue (RunInIO m, Stream m a)
247workLoopFIFO q st sv winfo = run
251 stop = liftIO $ sendStop sv winfo
253 work <- liftIO $ tryPopR q
256 Just (RunInIO runin, m) -> do
257 r <- liftIO $ runin $
258 K.foldStreamShared st yieldk single (return Continue) m
265 res <- liftIO $ sendYield sv winfo (ChildYield a)
266 return $ if res then Continue else Suspend
268 -- XXX in general we would like to yield "n" elements from a single stream
269 -- before moving on to the next. Single element granularity could be too
270 -- expensive in certain cases. Similarly, we can use time limit for
273 res <- liftIO $ sendYield sv winfo (ChildYield a)
274 runInIO <- askRunInIO
275 liftIO $ enqueueFIFO sv q (runInIO, r)
276 return $ if res then Continue else Suspend
278{-# INLINE workLoopFIFOLimited #-}
280 :: forall m a. MonadRunInIO m
281 => LinkedQueue (RunInIO m, Stream m a)
286workLoopFIFOLimited q st sv winfo = run
290 stop = liftIO $ sendStop sv winfo
291 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
293 work <- liftIO $ tryPopR q
296 Just (RunInIO runin, m) -> do
297 yieldLimitOk <- liftIO $ decrementYieldLimit sv
300 r <- liftIO $ runin $
301 K.foldStreamShared st yieldk single incrContinue m
307 enqueueFIFO sv q (RunInIO runin, m)
308 incrementYieldLimit sv
312 res <- liftIO $ sendYield sv winfo (ChildYield a)
313 return $ if res then Continue else Suspend
316 res <- liftIO $ sendYield sv winfo (ChildYield a)
317 runInIO <- askRunInIO
318 liftIO $ enqueueFIFO sv q (runInIO, r)
319 yieldLimitOk <- liftIO $ decrementYieldLimit sv
320 if res && yieldLimitOk
323 incrementYieldLimit sv
326-------------------------------------------------------------------------------
328-- This code belongs in SVar.hs but is kept here for perf reasons
329-------------------------------------------------------------------------------
331-- XXX we have this function in this file because passing runStreamLIFO as a
332-- function argument to this function results in a perf degradation of more
333-- than 10%. Need to investigate what the root cause is.
334-- Interestingly, the same thing does not make any difference for Ahead.
335getLifoSVar :: forall m a. MonadAsync m
336 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
337getLifoSVar st mrun = do
338 outQ <- newIORef ([], 0)
339 outQMv <- newEmptyMVar
341 wfw <- newIORef False
342 running <- newIORef S.empty
343 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
344 yl <- case getYieldLimit st of
345 Nothing -> return Nothing
346 Just x -> Just <$> newIORef x
347 rateInfo <- getYieldRateInfo st
349 stats <- newSVarStats
352 let isWorkFinished _ = null <$> readIORef q
354 let isWorkFinishedLimited sv = do
356 case remainingWork sv of
360 Nothing -> return False
361 qEmpty <- null <$> readIORef q
362 return $ qEmpty || yieldsDone
364 let getSVar :: SVar Stream m a
365 -> (SVar Stream m a -> m [ChildEvent a])
366 -> (SVar Stream m a -> m Bool)
367 -> (SVar Stream m a -> IO Bool)
368 -> (IORef [(RunInIO m, Stream m a)]
374 getSVar sv readOutput postProc workDone wloop = SVar
376 , outputQueueFromConsumer = undefined
378 , maxBufferLimit = getMaxBuffer st
379 , pushBufferSpace = undefined
380 , pushBufferPolicy = undefined
381 , pushBufferMVar = undefined
382 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
383 , yieldRateInfo = rateInfo
384 , outputDoorBell = outQMv
385 , outputDoorBellFromConsumer = undefined
386 , readOutputQ = readOutput sv
387 , postProcess = postProc sv
388 , workerThreads = running
389 , workLoop = wloop q st{streamVar = Just sv} sv
390 , enqueue = enqueueLIFO sv q
391 , isWorkDone = workDone sv
392 , isQueueDone = workDone sv
394 , svarStyle = AsyncVar
395 , svarStopStyle = StopNone
396 , svarStopBy = undefined
398 , workerCount = active
399 , accountThread = delThread sv
400 , workerStopMVar = undefined
402 , svarInspectMode = getInspectMode st
404 , aheadWorkQueue = undefined
405 , outputHeap = undefined
410 case getStreamRate st of
412 case getYieldLimit st of
413 Nothing -> getSVar sv readOutputQBounded
417 Just _ -> getSVar sv readOutputQBounded
419 isWorkFinishedLimited
422 case getYieldLimit st of
423 Nothing -> getSVar sv readOutputQPaced
427 Just _ -> getSVar sv readOutputQPaced
429 isWorkFinishedLimited
433getFifoSVar :: forall m a. MonadAsync m
434 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
435getFifoSVar st mrun = do
436 outQ <- newIORef ([], 0)
437 outQMv <- newEmptyMVar
439 wfw <- newIORef False
440 running <- newIORef S.empty
442 yl <- case getYieldLimit st of
443 Nothing -> return Nothing
444 Just x -> Just <$> newIORef x
445 rateInfo <- getYieldRateInfo st
447 stats <- newSVarStats
450 let isWorkFinished _ = nullQ q
451 let isWorkFinishedLimited sv = do
453 case remainingWork sv of
457 Nothing -> return False
459 return $ qEmpty || yieldsDone
461 let getSVar :: SVar Stream m a
462 -> (SVar Stream m a -> m [ChildEvent a])
463 -> (SVar Stream m a -> m Bool)
464 -> (SVar Stream m a -> IO Bool)
465 -> (LinkedQueue (RunInIO m, Stream m a)
471 getSVar sv readOutput postProc workDone wloop = SVar
473 , outputQueueFromConsumer = undefined
475 , maxBufferLimit = getMaxBuffer st
476 , pushBufferSpace = undefined
477 , pushBufferPolicy = undefined
478 , pushBufferMVar = undefined
479 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
480 , yieldRateInfo = rateInfo
481 , outputDoorBell = outQMv
482 , outputDoorBellFromConsumer = undefined
483 , readOutputQ = readOutput sv
484 , postProcess = postProc sv
485 , workerThreads = running
486 , workLoop = wloop q st{streamVar = Just sv} sv
487 , enqueue = enqueueFIFO sv q
488 , isWorkDone = workDone sv
489 , isQueueDone = workDone sv
491 , svarStyle = WAsyncVar
492 , svarStopStyle = StopNone
493 , svarStopBy = undefined
495 , workerCount = active
496 , accountThread = delThread sv
497 , workerStopMVar = undefined
499 , svarInspectMode = getInspectMode st
501 , aheadWorkQueue = undefined
502 , outputHeap = undefined
507 case getStreamRate st of
509 case getYieldLimit st of
510 Nothing -> getSVar sv readOutputQBounded
514 Just _ -> getSVar sv readOutputQBounded
516 isWorkFinishedLimited
519 case getYieldLimit st of
520 Nothing -> getSVar sv readOutputQPaced
524 Just _ -> getSVar sv readOutputQPaced
526 isWorkFinishedLimited
530{-# INLINABLE newAsyncVar #-}
531newAsyncVar :: MonadAsync m
532 => State Stream m a -> Stream m a -> m (SVar Stream m a)
535 sv <- liftIO $ getLifoSVar st mrun
538-- | Generate a stream asynchronously to keep it buffered, lazily consume
543{-# INLINABLE mkAsyncK #-}
544mkAsyncK :: MonadAsync m => Stream m a -> Stream m a
545mkAsyncK m = K.mkStream $ \st yld sng stp -> do
546 sv <- newAsyncVar (adaptState st) m
547 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
550-- This is slightly faster than the CPS version above
552{-# INLINE_NORMAL mkAsyncD #-}
553mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
554mkAsyncD m = D.Stream step Nothing
557 step gst Nothing = do
558 sv <- newAsyncVar gst (D.toStreamK m)
559 return $ D.Skip $ Just $ fromSVarD sv
561 step gst (Just (D.UnStream step1 st)) = do
564 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
565 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
568-- XXX We can pass captureMonadState instead of using MonadAsync
569-- | Create a new SVar and enqueue one stream computation on it.
570{-# INLINABLE newWAsyncVar #-}
571newWAsyncVar :: MonadAsync m
572 => State Stream m a -> Stream m a -> m (SVar Stream m a)
573newWAsyncVar st m = do
575 sv <- liftIO $ getFifoSVar st mrun
576 -- XXX Use just Stream and IO in all the functions below
577 -- XXX pass mrun instead of calling captureMonadState again inside it
580------------------------------------------------------------------------------
581-- Running streams concurrently
582------------------------------------------------------------------------------
584-- Concurrency rate control.
586-- Our objective is to create more threads on demand if the consumer is running
587-- faster than us. As soon as we encounter a concurrent composition we create a
588-- push pull pair of threads. We use an SVar for communication between the
589-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
590-- The producer creates more threads if the SVar drains and becomes empty, that
591-- is the consumer is running faster.
593-- XXX Note 1: This mechanism can be problematic if the initial production
594-- latency is high, we may end up creating too many threads. So we need some
595-- way to monitor and use the latency as well. Having a limit on the dispatches
596-- (programmer controlled) may also help.
598-- TBD Note 2: We may want to run computations at the lower level of the
599-- composition tree serially even when they are composed using a parallel
600-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
601-- place of 'wAsync'. If we find that an SVar immediately above a computation
602-- gets drained empty we can switch to parallelizing the computation. For that
603-- we can use a state flag to fork the rest of the computation at any point of
604-- time inside the Monad bind operation if the consumer is running at a faster
607-- TBD Note 3: the binary operation ('parallel') composition allows us to
608-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
609-- chunksizes we will need to compose the parallel actions using a data
610-- constructor (A Free container) instead so that we can divide it in chunks of
611-- arbitrary size before dispatching. If the stream is composed of
612-- hierarchically composed grains of different sizes then we can always switch
613-- to a desired granularity depending on the consumer speed.
615-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
616-- into just the number of CPUs.
618-- | Join two computations on the currently running 'SVar' queue for concurrent
619-- execution. When we are using parallel composition, an SVar is passed around
620-- as a state variable. We try to schedule a new parallel computation on the
621-- SVar passed to us. The first time, when no SVar exists, a new SVar is
622-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
623-- already scheduled on the SVar is further evaluated. For example, when (a
624-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
625-- the current scheduler queue.
627-- The 'SVarStyle' required by the current composition context is passed as one
628-- of the parameters. If the scheduling and composition style of the new
629-- computation being scheduled is different than the style of the current SVar,
630-- then we create a new SVar and schedule it on that. The newly created SVar
631-- joins as one of the computations on the current SVar queue.
633-- Cases when we need to switch to a new SVar:
635-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
636-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
637-- because of the scheduling policy change.
638-- * if we 'adapt' a stream of type 'async' to a stream of type
639-- 'Parallel', we create a new SVar at the transitioning bind.
640-- * When the stream is switching from disjunctive composition to conjunctive
641-- composition and vice-versa we create a new SVar to isolate the scheduling
644forkSVarAsync :: MonadAsync m
645 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
646forkSVarAsync style m1 m2 = K.mkStream $ \st yld sng stp -> do
648 AsyncVar -> newAsyncVar st (concurrently m1 m2)
649 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
650 _ -> error "illegal svar type"
651 K.foldStream st yld sng stp $ Stream.toStreamK $ fromSVar sv
653 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
654 runInIO <- askRunInIO
655 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
656 K.foldStreamShared st yld sng stp ma
658{-# INLINE joinStreamVarAsync #-}
659joinStreamVarAsync :: MonadAsync m
660 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
661joinStreamVarAsync style m1 m2 = K.mkStream $ \st yld sng stp ->
663 Just sv | svarStyle sv == style -> do
664 runInIO <- askRunInIO
665 liftIO $ enqueue sv (runInIO, m2)
666 K.foldStreamShared st yld sng stp m1
667 _ -> K.foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
669------------------------------------------------------------------------------
670-- Semigroup and Monoid style compositions for parallel actions
671------------------------------------------------------------------------------
674asyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
675asyncK = joinStreamVarAsync AsyncVar
677-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
678-- huge difference in the performance of consM in IsStream instance even we
679-- have a SPECIALIZE in the instance.
681-- | XXX we can implement it more efficienty by directly implementing instead
682-- of combining streams using async.
683{-# INLINE consMAsync #-}
684{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
685consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
686consMAsync m (AsyncT r) = AsyncT $ asyncK (K.fromEffect m) r
688------------------------------------------------------------------------------
690------------------------------------------------------------------------------
692-- | For 'AsyncT' streams:
695-- (<>) = 'Streamly.Prelude.async'
696-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
699-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
700-- executed concurrently a la the 'async' combinator, producing results and
701-- side effects of iterations out of order:
704-- Stream.toList $ Stream.fromAsync $ do
705-- x <- Stream.fromList [2,1] -- foreach x in stream
706-- Stream.fromEffect $ delay x
712-- Nested monad binds behave like nested @for@ loops with nested iterations
713-- executed concurrently, a la the 'async' combinator:
716-- Stream.toList $ Stream.fromAsync $ do
717-- x <- Stream.fromList [1,2] -- foreach x in stream
718-- y <- Stream.fromList [2,4] -- foreach y in stream
719-- Stream.fromEffect $ delay (x + y)
727-- The behavior can be explained as follows. All the iterations corresponding
728-- to the element @1@ in the first stream constitute one output stream and all
729-- the iterations corresponding to @2@ constitute another output stream and
730-- these two output streams are merged using 'async'.
732-- /Since: 0.1.0 ("Streamly")/
735newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
736 deriving (MonadTrans)
738-- | A demand driven left biased parallely composing IO stream of elements of
739-- type @a@. See 'AsyncT' documentation for more details.
741-- /Since: 0.2.0 ("Streamly")/
744type Async = AsyncT IO
746------------------------------------------------------------------------------
748------------------------------------------------------------------------------
750-- Monomorphically typed version of "async" for better performance of Semigroup
753{-# SPECIALIZE append :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
754append :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
755append (AsyncT m1) (AsyncT m2) = AsyncT $ asyncK m1 m2
757instance MonadAsync m => Semigroup (AsyncT m a) where
760------------------------------------------------------------------------------
762------------------------------------------------------------------------------
764instance MonadAsync m => Monoid (AsyncT m a) where
765 mempty = AsyncT K.nil
768------------------------------------------------------------------------------
770------------------------------------------------------------------------------
772{-# INLINE apAsync #-}
773{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
774apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
775apAsync (AsyncT m1) (AsyncT m2) =
776 let f x1 = K.concatMapWith asyncK (pure . x1) m2
777 in AsyncT $ K.concatMapWith asyncK f m1
779instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
781 pure = AsyncT . K.fromPure
786------------------------------------------------------------------------------
788------------------------------------------------------------------------------
790-- GHC: if we change the implementation of bindWith with arguments in a
791-- different order we see a significant performance degradation (~2x).
792{-# INLINE bindAsync #-}
793{-# SPECIALIZE bindAsync ::
794 AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
795bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
796bindAsync (AsyncT m) f = AsyncT $ K.bindWith asyncK m (getAsyncT . f)
798-- GHC: if we specify arguments in the definition of (>>=) we see a significant
799-- performance degradation (~2x).
800instance MonadAsync m => Monad (AsyncT m) where
804------------------------------------------------------------------------------
806------------------------------------------------------------------------------
808MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
810------------------------------------------------------------------------------
812------------------------------------------------------------------------------
814{-# INLINE wAsyncK #-}
815wAsyncK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
816wAsyncK = joinStreamVarAsync WAsyncVar
818-- | XXX we can implement it more efficienty by directly implementing instead
819-- of combining streams using wAsync.
820{-# INLINE consMWAsync #-}
821{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
822consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
823consMWAsync m (WAsyncT r) = WAsyncT $ wAsyncK (K.fromEffect m) r
825-- | For 'WAsyncT' streams:
828-- (<>) = 'Streamly.Prelude.wAsync'
829-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
832-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
833-- executed concurrently a la the 'wAsync' combinator, producing results and
834-- side effects of iterations out of order:
837-- Stream.toList $ Stream.fromWAsync $ do
838-- x <- Stream.fromList [2,1] -- foreach x in stream
839-- Stream.fromEffect $ delay x
845-- Nested monad binds behave like nested @for@ loops with nested iterations
846-- executed concurrently, a la the 'wAsync' combinator:
849-- Stream.toList $ Stream.fromWAsync $ do
850-- x <- Stream.fromList [1,2] -- foreach x in stream
851-- y <- Stream.fromList [2,4] -- foreach y in stream
852-- Stream.fromEffect $ delay (x + y)
860-- The behavior can be explained as follows. All the iterations corresponding
861-- to the element @1@ in the first stream constitute one 'WAsyncT' output
862-- stream and all the iterations corresponding to @2@ constitute another
863-- 'WAsyncT' output stream and these two output streams are merged using
866-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
867-- contrast to the depth wise scheduling behavior of 'AsyncT'.
869-- /Since: 0.2.0 ("Streamly")/
873-- XXX This documentation is redundant, need to be cleaned up/removed.
876-- import qualified "Streamly.Prelude" as S
877-- import Control.Concurrent
879-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
885-- For this example, we are using @maxThreads 1@ so that concurrent thread
886-- scheduling does not affect the results and make them unpredictable. Let's
887-- now take a more general example:
890-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
893-- [1,4,2,7,5,3,8,6,9]
896-- This is how the execution of the above stream proceeds:
898-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
899-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
900-- queue is represented by the rightmost item.
901-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
902-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
903-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
904-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
905-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
906-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
907-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
908-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
909-- 4. Note that the scheduler queue expands by one more stream component in
910-- every pass because one more @<>@ is broken down into two components. At this
911-- point there are no more @<>@ operations to be broken down further and the
912-- queue has reached its maximum size. Now these streams are scheduled in
913-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
915-- As we see above, in a right associated expression composed with @<>@, only
916-- one @<>@ operation is broken down into two components in one execution,
917-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
918-- scheduler passes to expand the whole expression. By the time @n-th@
919-- component is added to the scheduler queue, the first component would have
920-- received @n@ scheduler passes.
922-- Since all streams get interleaved, this operation is not suitable for
923-- folding an infinite lazy container of infinite size streams. However, if
924-- the streams are small, the streams on the left may get finished before more
925-- streams are added to the scheduler queue from the right side of the
926-- expression, so it may be possible to fold an infinite lazy container of
927-- streams. For example, if the streams are of size @n@ then at most @n@
928-- streams would be in the scheduler queue at a time.
930-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
931-- therefore the output of 'WAsyncT' even with a single thread of execution is
932-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
933-- its scheduling behavior.
935-- Any exceptions generated by a constituent stream are propagated to the
936-- output stream. The output and exceptions from a single stream are guaranteed
937-- to arrive in the same order in the resulting stream as they were generated
938-- in the input stream. However, the relative ordering of elements from
939-- different streams in the resulting stream can vary depending on scheduling
940-- and generation delays.
942-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
943-- concurrently using a round robin scheduling.
946-- main = S.'drain' . S.'fromWAsync' $ do
947-- n <- return 3 \<\> return 2 \<\> return 1
949-- threadDelay (n * 1000000)
950-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
953-- ThreadId 40: Delay 1
954-- ThreadId 39: Delay 2
955-- ThreadId 38: Delay 3
958newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
959 deriving (MonadTrans)
961-- | A round robin parallely composing IO stream of elements of type @a@.
962-- See 'WAsyncT' documentation for more details.
964-- /Since: 0.2.0 ("Streamly")/
967type WAsync = WAsyncT IO
969------------------------------------------------------------------------------
971------------------------------------------------------------------------------
973{-# INLINE wAppend #-}
974{-# SPECIALIZE wAppend :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
975wAppend :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
976wAppend (WAsyncT m1) (WAsyncT m2) = WAsyncT $ wAsyncK m1 m2
978instance MonadAsync m => Semigroup (WAsyncT m a) where
981------------------------------------------------------------------------------
983------------------------------------------------------------------------------
985instance MonadAsync m => Monoid (WAsyncT m a) where
986 mempty = WAsyncT K.nil
989------------------------------------------------------------------------------
991------------------------------------------------------------------------------
993{-# INLINE apWAsync #-}
994{-# SPECIALIZE apWAsync ::
995 WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
996apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
997apWAsync (WAsyncT m1) (WAsyncT m2) =
998 let f x1 = K.concatMapWith wAsyncK (pure . x1) m2
999 in WAsyncT $ K.concatMapWith wAsyncK f m1
1001-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1002-- performance degradation (~2x).
1003instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1004 pure = WAsyncT . K.fromPure
1007------------------------------------------------------------------------------
1009------------------------------------------------------------------------------
1011-- GHC: if we change the implementation of bindWith with arguments in a
1012-- different order we see a significant performance degradation (~2x).
1013{-# INLINE bindWAsync #-}
1014{-# SPECIALIZE bindWAsync ::
1015 WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1016bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1017bindWAsync (WAsyncT m) f = WAsyncT $ K.bindWith wAsyncK m (getWAsyncT . f)
1019-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1020-- performance degradation (~2x).
1021instance MonadAsync m => Monad (WAsyncT m) where
1025------------------------------------------------------------------------------
1027------------------------------------------------------------------------------
1029MONAD_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