1{-# LANGUAGE UndecidableInstances #-}
6-- Module : Streamly.Internal.Data.Stream.Async
7-- Copyright : (c) 2017 Harendra Kumar
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
15module Streamly.Internal.Data.Stream.Async
32import Control.Concurrent (myThreadId)
33import Control.Monad.Base (MonadBase(..), liftBaseDefault)
34import Control.Monad.Catch (MonadThrow, throwM)
35import Control.Monad.Trans.Control (MonadBaseControl (..))
36import Control.Concurrent.MVar (newEmptyMVar)
37-- import Control.Monad.Error.Class (MonadError(..))
38import Control.Monad.IO.Class (MonadIO(..))
39import Control.Monad.Reader.Class (MonadReader(..))
40import Control.Monad.State.Class (MonadState(..))
41import Control.Monad.Trans.Class (MonadTrans(lift))
42import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR)
43import Data.IORef (IORef, newIORef, readIORef)
44import Data.Maybe (fromJust)
45#if __GLASGOW_HASKELL__ < 808
46import Data.Semigroup (Semigroup(..))
49import Prelude hiding (map)
50import qualified Data.Set as S
52import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)
53import Streamly.Internal.Data.Stream.SVar (fromSVar)
54import Streamly.Internal.Data.SVar
55import Streamly.Internal.Data.Stream.StreamK
56 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared)
57import qualified Streamly.Internal.Data.Stream.StreamK as K
58import qualified Streamly.Internal.Data.Stream.StreamD as D
60#include "Instances.hs"
62-------------------------------------------------------------------------------
64-------------------------------------------------------------------------------
66data WorkerStatus = Continue | Suspend
68{-# INLINE workLoopLIFO #-}
70 :: (MonadIO m, MonadBaseControl IO m)
76workLoopLIFO q st sv winfo = run
80 mrun = runInIO $ svarMrun sv
83 let stop = liftIO $ sendStop sv winfo
87 -- XXX when we finish we need to send the monadic state back to
88 -- the parent so that the state can be merged back. We capture
89 -- and return the state in the stop continuation.
91 -- Instead of using the run function we can just restore the
92 -- monad state here. That way it can work easily for
93 -- distributed case as well.
95 foldStreamShared st yieldk single (return Continue) m
102 res <- liftIO $ sendYield sv winfo (ChildYield a)
103 return $ if res then Continue else Suspend
106 res <- liftIO $ sendYield sv winfo (ChildYield a)
108 then foldStreamShared st yieldk single (return Continue) r
110 -- XXX we also need to save the monadic state here
114 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
116 x : xs -> (xs, Just x)
118-- We duplicate workLoop for yield limit and no limit cases because it has
119-- around 40% performance overhead in the worst case.
121-- XXX we can pass yinfo directly as an argument here so that we do not have to
122-- make a check every time.
123{-# INLINE workLoopLIFOLimited #-}
125 :: (MonadIO m, MonadBaseControl IO m)
126 => IORef [Stream m a]
131workLoopLIFOLimited q st sv winfo = run
135 mrun = runInIO $ svarMrun sv
136 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
139 let stop = liftIO $ sendStop sv winfo
143 -- XXX This is just a best effort minimization of concurrency
144 -- to the yield limit. If the stream is made of concurrent
145 -- streams we do not reserve the yield limit in the constituent
146 -- streams before executing the action. This can be done
147 -- though, by sharing the yield limit ref with downstream
148 -- actions via state passing. Just a todo.
149 yieldLimitOk <- liftIO $ decrementYieldLimit sv
153 foldStreamShared st yieldk single incrContinue m
158 -- Avoid any side effects, undo the yield limit decrement if we
159 -- never yielded anything.
162 incrementYieldLimit sv
166 res <- liftIO $ sendYield sv winfo (ChildYield a)
167 return $ if res then Continue else Suspend
169 -- XXX can we pass on the yield limit downstream to limit the concurrency
170 -- of constituent streams.
172 res <- liftIO $ sendYield sv winfo (ChildYield a)
173 yieldLimitOk <- liftIO $ decrementYieldLimit sv
174 if res && yieldLimitOk
175 then foldStreamShared st yieldk single incrContinue r
177 incrementYieldLimit sv
181 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
183 x : xs -> (xs, Just x)
185-------------------------------------------------------------------------------
187-------------------------------------------------------------------------------
189-- XXX we can remove sv as it is derivable from st
191{-# INLINE workLoopFIFO #-}
193 :: (MonadIO m, MonadBaseControl IO m)
194 => LinkedQueue (Stream m a)
199workLoopFIFO q st sv winfo = run
203 mrun = runInIO $ svarMrun sv
205 work <- liftIO $ tryPopR q
206 let stop = liftIO $ sendStop sv winfo
211 foldStreamShared st yieldk single (return Continue) m
218 res <- liftIO $ sendYield sv winfo (ChildYield a)
219 return $ if res then Continue else Suspend
221 -- XXX in general we would like to yield "n" elements from a single stream
222 -- before moving on to the next. Single element granularity could be too
223 -- expensive in certain cases. Similarly, we can use time limit for
226 res <- liftIO $ sendYield sv winfo (ChildYield a)
227 liftIO $ enqueueFIFO sv q r
228 return $ if res then Continue else Suspend
230{-# INLINE workLoopFIFOLimited #-}
232 :: (MonadIO m, MonadBaseControl IO m)
233 => LinkedQueue (Stream m a)
238workLoopFIFOLimited q st sv winfo = run
242 mrun = runInIO $ svarMrun sv
243 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
245 work <- liftIO $ tryPopR q
246 let stop = liftIO $ sendStop sv winfo
250 yieldLimitOk <- liftIO $ decrementYieldLimit sv
254 foldStreamShared st yieldk single incrContinue m
261 incrementYieldLimit sv
265 res <- liftIO $ sendYield sv winfo (ChildYield a)
266 return $ if res then Continue else Suspend
269 res <- liftIO $ sendYield sv winfo (ChildYield a)
270 liftIO $ enqueueFIFO sv q r
271 yieldLimitOk <- liftIO $ decrementYieldLimit sv
272 if res && yieldLimitOk
275 incrementYieldLimit sv
278-------------------------------------------------------------------------------
280-- This code belongs in SVar.hs but is kept here for perf reasons
281-------------------------------------------------------------------------------
283-- XXX we have this function in this file because passing runStreamLIFO as a
284-- function argument to this function results in a perf degradation of more
285-- than 10%. Need to investigate what the root cause is.
286-- Interestingly, the same thing does not make any difference for Ahead.
287getLifoSVar :: forall m a. MonadAsync m
288 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
289getLifoSVar st mrun = do
290 outQ <- newIORef ([], 0)
291 outQMv <- newEmptyMVar
293 wfw <- newIORef False
294 running <- newIORef S.empty
296 yl <- case getYieldLimit st of
297 Nothing -> return Nothing
298 Just x -> Just <$> newIORef x
299 rateInfo <- getYieldRateInfo st
301 stats <- newSVarStats
304 let isWorkFinished _ = null <$> readIORef q
306 let isWorkFinishedLimited sv = do
308 case remainingWork sv of
312 Nothing -> return False
313 qEmpty <- null <$> readIORef q
314 return $ qEmpty || yieldsDone
316 let getSVar :: SVar Stream m a
317 -> (SVar Stream m a -> m [ChildEvent a])
318 -> (SVar Stream m a -> m Bool)
319 -> (SVar Stream m a -> IO Bool)
320 -> (IORef [Stream m a]
326 getSVar sv readOutput postProc workDone wloop = SVar
328 , outputQueueFromConsumer = undefined
330 , maxBufferLimit = getMaxBuffer st
331 , pushBufferSpace = undefined
332 , pushBufferPolicy = undefined
333 , pushBufferMVar = undefined
334 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
335 , yieldRateInfo = rateInfo
336 , outputDoorBell = outQMv
337 , outputDoorBellFromConsumer = undefined
338 , readOutputQ = readOutput sv
339 , postProcess = postProc sv
340 , workerThreads = running
341 , workLoop = wloop q st{streamVar = Just sv} sv
342 , enqueue = enqueueLIFO sv q
343 , isWorkDone = workDone sv
344 , isQueueDone = workDone sv
346 , svarStyle = AsyncVar
347 , svarStopStyle = StopNone
348 , svarStopBy = undefined
350 , workerCount = active
351 , accountThread = delThread sv
352 , workerStopMVar = undefined
354 , svarInspectMode = getInspectMode st
356 , aheadWorkQueue = undefined
357 , outputHeap = undefined
362 case getStreamRate st of
364 case getYieldLimit st of
365 Nothing -> getSVar sv readOutputQBounded
369 Just _ -> getSVar sv readOutputQBounded
371 isWorkFinishedLimited
374 case getYieldLimit st of
375 Nothing -> getSVar sv readOutputQPaced
379 Just _ -> getSVar sv readOutputQPaced
381 isWorkFinishedLimited
385getFifoSVar :: forall m a. MonadAsync m
386 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
387getFifoSVar st mrun = do
388 outQ <- newIORef ([], 0)
389 outQMv <- newEmptyMVar
391 wfw <- newIORef False
392 running <- newIORef S.empty
394 yl <- case getYieldLimit st of
395 Nothing -> return Nothing
396 Just x -> Just <$> newIORef x
397 rateInfo <- getYieldRateInfo st
399 stats <- newSVarStats
402 let isWorkFinished _ = nullQ q
403 let isWorkFinishedLimited sv = do
405 case remainingWork sv of
409 Nothing -> return False
411 return $ qEmpty || yieldsDone
413 let getSVar :: SVar Stream m a
414 -> (SVar Stream m a -> m [ChildEvent a])
415 -> (SVar Stream m a -> m Bool)
416 -> (SVar Stream m a -> IO Bool)
417 -> (LinkedQueue (Stream m a)
423 getSVar sv readOutput postProc workDone wloop = SVar
425 , outputQueueFromConsumer = undefined
427 , maxBufferLimit = getMaxBuffer st
428 , pushBufferSpace = undefined
429 , pushBufferPolicy = undefined
430 , pushBufferMVar = undefined
431 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
432 , yieldRateInfo = rateInfo
433 , outputDoorBell = outQMv
434 , outputDoorBellFromConsumer = undefined
435 , readOutputQ = readOutput sv
436 , postProcess = postProc sv
437 , workerThreads = running
438 , workLoop = wloop q st{streamVar = Just sv} sv
439 , enqueue = enqueueFIFO sv q
440 , isWorkDone = workDone sv
441 , isQueueDone = workDone sv
443 , svarStyle = WAsyncVar
444 , svarStopStyle = StopNone
445 , svarStopBy = undefined
447 , workerCount = active
448 , accountThread = delThread sv
449 , workerStopMVar = undefined
451 , svarInspectMode = getInspectMode st
453 , aheadWorkQueue = undefined
454 , outputHeap = undefined
459 case getStreamRate st of
461 case getYieldLimit st of
462 Nothing -> getSVar sv readOutputQBounded
466 Just _ -> getSVar sv readOutputQBounded
468 isWorkFinishedLimited
471 case getYieldLimit st of
472 Nothing -> getSVar sv readOutputQPaced
476 Just _ -> getSVar sv readOutputQPaced
478 isWorkFinishedLimited
482{-# INLINABLE newAsyncVar #-}
483newAsyncVar :: MonadAsync m
484 => State Stream m a -> Stream m a -> m (SVar Stream m a)
486 mrun <- captureMonadState
487 sv <- liftIO $ getLifoSVar st mrun
490-- | Generate a stream asynchronously to keep it buffered, lazily consume
495{-# INLINABLE mkAsyncK #-}
496mkAsyncK :: (IsStream t, MonadAsync m) => t m a -> t m a
497mkAsyncK m = mkStream $ \st yld sng stp -> do
498 sv <- newAsyncVar (adaptState st) (toStream m)
499 foldStream st yld sng stp $ fromSVar sv
501{-# INLINE_NORMAL mkAsyncD #-}
502mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
503mkAsyncD m = D.Stream step Nothing
506 step gst Nothing = do
507 sv <- newAsyncVar gst (D.fromStreamD m)
508 return $ D.Skip $ Just $ D.fromSVar sv
510 step gst (Just (D.UnStream step1 st)) = do
513 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
514 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
517-- This is slightly faster than the CPS version above
519-- | Make the stream producer and consumer run concurrently by introducing a
520-- buffer between them. The producer thread evaluates the input stream until
521-- the buffer fills, it terminates if the buffer is full and a worker thread is
522-- kicked off again to evaluate the remaining stream when there is space in the
523-- buffer. The consumer consumes the stream lazily from the buffer.
527{-# INLINE_NORMAL mkAsync #-}
528mkAsync :: (K.IsStream t, MonadAsync m) => t m a -> t m a
529mkAsync = D.fromStreamD . mkAsyncD . D.toStreamD
531-- | Create a new SVar and enqueue one stream computation on it.
532{-# INLINABLE newWAsyncVar #-}
533newWAsyncVar :: MonadAsync m
534 => State Stream m a -> Stream m a -> m (SVar Stream m a)
535newWAsyncVar st m = do
536 mrun <- captureMonadState
537 sv <- liftIO $ getFifoSVar st mrun
540------------------------------------------------------------------------------
541-- Running streams concurrently
542------------------------------------------------------------------------------
544-- Concurrency rate control.
546-- Our objective is to create more threads on demand if the consumer is running
547-- faster than us. As soon as we encounter a concurrent composition we create a
548-- push pull pair of threads. We use an SVar for communication between the
549-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
550-- The producer creates more threads if the SVar drains and becomes empty, that
551-- is the consumer is running faster.
553-- XXX Note 1: This mechanism can be problematic if the initial production
554-- latency is high, we may end up creating too many threads. So we need some
555-- way to monitor and use the latency as well. Having a limit on the dispatches
556-- (programmer controlled) may also help.
558-- TBD Note 2: We may want to run computations at the lower level of the
559-- composition tree serially even when they are composed using a parallel
560-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
561-- place of 'wAsync'. If we find that an SVar immediately above a computation
562-- gets drained empty we can switch to parallelizing the computation. For that
563-- we can use a state flag to fork the rest of the computation at any point of
564-- time inside the Monad bind operation if the consumer is running at a faster
567-- TBD Note 3: the binary operation ('parallel') composition allows us to
568-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
569-- chunksizes we will need to compose the parallel actions using a data
570-- constructor (A Free container) instead so that we can divide it in chunks of
571-- arbitrary size before dispatching. If the stream is composed of
572-- hierarchically composed grains of different sizes then we can always switch
573-- to a desired granularity depending on the consumer speed.
575-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
576-- into just the number of CPUs.
578-- | Join two computations on the currently running 'SVar' queue for concurrent
579-- execution. When we are using parallel composition, an SVar is passed around
580-- as a state variable. We try to schedule a new parallel computation on the
581-- SVar passed to us. The first time, when no SVar exists, a new SVar is
582-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
583-- already scheduled on the SVar is further evaluated. For example, when (a
584-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
585-- the current scheduler queue.
587-- The 'SVarStyle' required by the current composition context is passed as one
588-- of the parameters. If the scheduling and composition style of the new
589-- computation being scheduled is different than the style of the current SVar,
590-- then we create a new SVar and schedule it on that. The newly created SVar
591-- joins as one of the computations on the current SVar queue.
593-- Cases when we need to switch to a new SVar:
595-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
596-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
597-- because of the scheduling policy change.
598-- * if we 'adapt' a stream of type 'async' to a stream of type
599-- 'Parallel', we create a new SVar at the transitioning bind.
600-- * When the stream is switching from disjunctive composition to conjunctive
601-- composition and vice-versa we create a new SVar to isolate the scheduling
604forkSVarAsync :: (IsStream t, MonadAsync m)
605 => SVarStyle -> t m a -> t m a -> t m a
606forkSVarAsync style m1 m2 = mkStream $ \st yld sng stp -> do
608 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
609 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
610 _ -> error "illegal svar type"
611 foldStream st yld sng stp $ fromSVar sv
613 concurrently ma mb = mkStream $ \st yld sng stp -> do
614 liftIO $ enqueue (fromJust $ streamVar st) mb
615 foldStreamShared st yld sng stp ma
617{-# INLINE joinStreamVarAsync #-}
618joinStreamVarAsync :: (IsStream t, MonadAsync m)
619 => SVarStyle -> t m a -> t m a -> t m a
620joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
622 Just sv | svarStyle sv == style -> do
623 liftIO $ enqueue sv (toStream m2)
624 foldStreamShared st yld sng stp m1
625 _ -> foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
627------------------------------------------------------------------------------
628-- Semigroup and Monoid style compositions for parallel actions
629------------------------------------------------------------------------------
631-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
632-- Merges two streams possibly concurrently, preferring the
633-- elements from the left one when available.
637async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
638async = joinStreamVarAsync AsyncVar
643{-# DEPRECATED (<|) "Please use 'async' instead." #-}
645(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
648-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
649-- huge difference in the performance of consM in IsStream instance even we
650-- have a SPECIALIZE in the instance.
652-- | XXX we can implement it more efficienty by directly implementing instead
653-- of combining streams using async.
654{-# INLINE consMAsync #-}
655{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
656consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
657consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
659------------------------------------------------------------------------------
661------------------------------------------------------------------------------
663-- | The 'Semigroup' operation (@<>@) for 'AsyncT' merges two streams
664-- concurrently with priority given to the first stream. In @s1 <> s2 <> s3
665-- ...@ the streams s1, s2 and s3 are scheduled for execution in that order.
666-- Multiple scheduled streams may be executed concurrently and the elements
667-- generated by them are served to the consumer as and when they become
668-- available. This behavior is similar to the scheduling and execution behavior
669-- of actions in a single async stream.
671-- Since only a finite number of streams are executed concurrently, this
672-- operation can be used to fold an infinite lazy container of streams.
676-- import qualified "Streamly.Prelude" as S
677-- import Control.Concurrent
679-- main = (S.toList . 'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
685-- Any exceptions generated by a constituent stream are propagated to the
686-- output stream. The output and exceptions from a single stream are guaranteed
687-- to arrive in the same order in the resulting stream as they were generated
688-- in the input stream. However, the relative ordering of elements from
689-- different streams in the resulting stream can vary depending on scheduling
690-- and generation delays.
692-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
693-- concurrently based on demand. More concurrent iterations are started only
694-- if the previous iterations are not able to produce enough output for the
698-- main = 'drain' . 'asyncly' $ do
699-- n <- return 3 \<\> return 2 \<\> return 1
701-- threadDelay (n * 1000000)
702-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
705-- ThreadId 40: Delay 1
706-- ThreadId 39: Delay 2
707-- ThreadId 38: Delay 3
711newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
712 deriving (MonadTrans)
714-- | A demand driven left biased parallely composing IO stream of elements of
715-- type @a@. See 'AsyncT' documentation for more details.
718type Async = AsyncT IO
720-- | Fix the type of a polymorphic stream as 'AsyncT'.
723asyncly :: IsStream t => AsyncT m a -> t m a
726instance IsStream AsyncT where
732------------------------------------------------------------------------------
734------------------------------------------------------------------------------
736-- Monomorphically typed version of "async" for better performance of Semigroup
738{-# INLINE mappendAsync #-}
739{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
740mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
741mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
743instance MonadAsync m => Semigroup (AsyncT m a) where
746------------------------------------------------------------------------------
748------------------------------------------------------------------------------
750instance MonadAsync m => Monoid (AsyncT m a) where
754------------------------------------------------------------------------------
756------------------------------------------------------------------------------
758{-# INLINE apAsync #-}
759{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
760apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
761apAsync (AsyncT m1) (AsyncT m2) =
762 let f x1 = K.concatMapBy async (pure . x1) m2
763 in AsyncT $ K.concatMapBy async f m1
765instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
767 pure = AsyncT . K.yield
771------------------------------------------------------------------------------
773------------------------------------------------------------------------------
775-- GHC: if we change the implementation of bindWith with arguments in a
776-- different order we see a significant performance degradation (~2x).
777{-# INLINE bindAsync #-}
778{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
779bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
780bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
782-- GHC: if we specify arguments in the definition of (>>=) we see a significant
783-- performance degradation (~2x).
784instance MonadAsync m => Monad (AsyncT m) where
788------------------------------------------------------------------------------
790------------------------------------------------------------------------------
792MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
794------------------------------------------------------------------------------
796------------------------------------------------------------------------------
798-- | XXX we can implement it more efficienty by directly implementing instead
799-- of combining streams using wAsync.
800{-# INLINE consMWAsync #-}
801{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
802consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
803consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
805-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
806-- Merges two streams concurrently choosing elements from both fairly.
810wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
811wAsync = joinStreamVarAsync WAsyncVar
813-- | 'WAsyncT' is similar to 'WSerialT' but with concurrent execution.
814-- The 'Semigroup' operation (@<>@) for 'WAsyncT' merges two streams
815-- concurrently interleaving the actions from both the streams. In @s1
816-- <> s2 <> s3 ...@, the individual actions from streams @s1@, @s2@ and @s3@
817-- are scheduled for execution in a round-robin fashion. Multiple scheduled
818-- actions may be executed concurrently, the results from concurrent executions
819-- are consumed in the order in which they become available.
822-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
823-- contrast to the depth wise scheduling behavior of 'AsyncT'.
827-- import qualified "Streamly.Prelude" as S
828-- import Control.Concurrent
830-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
836-- For this example, we are using @maxThreads 1@ so that concurrent thread
837-- scheduling does not affect the results and make them unpredictable. Let's
838-- now take a more general example:
841-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
844-- [1,4,2,7,5,3,8,6,9]
847-- This is how the execution of the above stream proceeds:
849-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
850-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
851-- queue is represented by the rightmost item.
852-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
853-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
854-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
855-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
856-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
857-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
858-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
859-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
860-- 4. Note that the scheduler queue expands by one more stream component in
861-- every pass because one more @<>@ is broken down into two components. At this
862-- point there are no more @<>@ operations to be broken down further and the
863-- queue has reached its maximum size. Now these streams are scheduled in
864-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
866-- As we see above, in a right associated expression composed with @<>@, only
867-- one @<>@ operation is broken down into two components in one execution,
868-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
869-- scheduler passes to expand the whole expression. By the time @n-th@
870-- component is added to the scheduler queue, the first component would have
871-- received @n@ scheduler passes.
873-- Since all streams get interleaved, this operation is not suitable for
874-- folding an infinite lazy container of infinite size streams. However, if
875-- the streams are small, the streams on the left may get finished before more
876-- streams are added to the scheduler queue from the right side of the
877-- expression, so it may be possible to fold an infinite lazy container of
878-- streams. For example, if the streams are of size @n@ then at most @n@
879-- streams would be in the scheduler queue at a time.
881-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
882-- therefore the output of 'WAsyncT' even with a single thread of execution is
883-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
884-- its scheduling behavior.
886-- Any exceptions generated by a constituent stream are propagated to the
887-- output stream. The output and exceptions from a single stream are guaranteed
888-- to arrive in the same order in the resulting stream as they were generated
889-- in the input stream. However, the relative ordering of elements from
890-- different streams in the resulting stream can vary depending on scheduling
891-- and generation delays.
893-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
894-- concurrently using a round robin scheduling.
897-- main = 'drain' . 'wAsyncly' $ do
898-- n <- return 3 \<\> return 2 \<\> return 1
900-- threadDelay (n * 1000000)
901-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
904-- ThreadId 40: Delay 1
905-- ThreadId 39: Delay 2
906-- ThreadId 38: Delay 3
910newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
911 deriving (MonadTrans)
913-- | A round robin parallely composing IO stream of elements of type @a@.
914-- See 'WAsyncT' documentation for more details.
917type WAsync = WAsyncT IO
919-- | Fix the type of a polymorphic stream as 'WAsyncT'.
922wAsyncly :: IsStream t => WAsyncT m a -> t m a
925instance IsStream WAsyncT where
926 toStream = getWAsyncT
931------------------------------------------------------------------------------
933------------------------------------------------------------------------------
935{-# INLINE mappendWAsync #-}
936{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
937mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
938mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
940instance MonadAsync m => Semigroup (WAsyncT m a) where
943------------------------------------------------------------------------------
945------------------------------------------------------------------------------
947instance MonadAsync m => Monoid (WAsyncT m a) where
951------------------------------------------------------------------------------
953------------------------------------------------------------------------------
955{-# INLINE apWAsync #-}
956{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
957apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
958apWAsync (WAsyncT m1) (WAsyncT m2) =
959 let f x1 = K.concatMapBy wAsync (pure . x1) m2
960 in WAsyncT $ K.concatMapBy wAsync f m1
962-- GHC: if we specify arguments in the definition of (<*>) we see a significant
963-- performance degradation (~2x).
964instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
965 pure = WAsyncT . K.yield
968------------------------------------------------------------------------------
970------------------------------------------------------------------------------
972-- GHC: if we change the implementation of bindWith with arguments in a
973-- different order we see a significant performance degradation (~2x).
974{-# INLINE bindWAsync #-}
975{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
976bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
977bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
979-- GHC: if we specify arguments in the definition of (>>=) we see a significant
980-- performance degradation (~2x).
981instance MonadAsync m => Monad (WAsyncT m) where
985------------------------------------------------------------------------------
987------------------------------------------------------------------------------
989MONAD_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