2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE LambdaCase #-}
8{-# LANGUAGE MultiParamTypeClasses #-}
9{-# LANGUAGE ScopedTypeVariables #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
13-- Module : Streamly.Internal.Data.Stream.Async
14-- Copyright : (c) 2017 Harendra Kumar
17-- Maintainer : streamly@composewell.com
18-- Stability : experimental
22module Streamly.Internal.Data.Stream.Async
39import Control.Concurrent (myThreadId)
40import Control.Monad.Base (MonadBase(..), liftBaseDefault)
41import Control.Monad.Catch (MonadThrow, throwM)
42import Control.Monad.Trans.Control (MonadBaseControl (..))
43import Control.Concurrent.MVar (newEmptyMVar)
44-- import Control.Monad.Error.Class (MonadError(..))
45import Control.Monad.IO.Class (MonadIO(..))
46import Control.Monad.Reader.Class (MonadReader(..))
47import Control.Monad.State.Class (MonadState(..))
48import Control.Monad.Trans.Class (MonadTrans(lift))
49import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR)
50import Data.IORef (IORef, newIORef, readIORef)
51import Data.Maybe (fromJust)
52#if __GLASGOW_HASKELL__ < 808
53import Data.Semigroup (Semigroup(..))
56import Prelude hiding (map)
57import qualified Data.Set as S
59import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)
60import Streamly.Internal.Data.Stream.SVar (fromSVar)
61import Streamly.Internal.Data.SVar
62import Streamly.Internal.Data.Stream.StreamK
63 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared)
64import qualified Streamly.Internal.Data.Stream.StreamK as K
65import qualified Streamly.Internal.Data.Stream.StreamD as D
67#include "Instances.hs"
69-------------------------------------------------------------------------------
71-------------------------------------------------------------------------------
73data WorkerStatus = Continue | Suspend
75{-# INLINE workLoopLIFO #-}
77 :: (MonadIO m, MonadBaseControl IO m)
83workLoopLIFO q st sv winfo = run
87 mrun = runInIO $ svarMrun sv
90 let stop = liftIO $ sendStop sv winfo
94 -- XXX when we finish we need to send the monadic state back to
95 -- the parent so that the state can be merged back. We capture
96 -- and return the state in the stop continuation.
98 -- Instead of using the run function we can just restore the
99 -- monad state here. That way it can work easily for
100 -- distributed case as well.
102 foldStreamShared st yieldk single (return Continue) m
109 res <- liftIO $ sendYield sv winfo (ChildYield a)
110 return $ if res then Continue else Suspend
113 res <- liftIO $ sendYield sv winfo (ChildYield a)
115 then foldStreamShared st yieldk single (return Continue) r
117 -- XXX we also need to save the monadic state here
121 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
123 x : xs -> (xs, Just x)
125-- We duplicate workLoop for yield limit and no limit cases because it has
126-- around 40% performance overhead in the worst case.
128-- XXX we can pass yinfo directly as an argument here so that we do not have to
129-- make a check every time.
130{-# INLINE workLoopLIFOLimited #-}
132 :: (MonadIO m, MonadBaseControl IO m)
133 => IORef [Stream m a]
138workLoopLIFOLimited q st sv winfo = run
142 mrun = runInIO $ svarMrun sv
143 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
146 let stop = liftIO $ sendStop sv winfo
150 -- XXX This is just a best effort minimization of concurrency
151 -- to the yield limit. If the stream is made of concurrent
152 -- streams we do not reserve the yield limit in the constituent
153 -- streams before executing the action. This can be done
154 -- though, by sharing the yield limit ref with downstream
155 -- actions via state passing. Just a todo.
156 yieldLimitOk <- liftIO $ decrementYieldLimit sv
160 foldStreamShared st yieldk single incrContinue m
165 -- Avoid any side effects, undo the yield limit decrement if we
166 -- never yielded anything.
169 incrementYieldLimit sv
173 res <- liftIO $ sendYield sv winfo (ChildYield a)
174 return $ if res then Continue else Suspend
176 -- XXX can we pass on the yield limit downstream to limit the concurrency
177 -- of constituent streams.
179 res <- liftIO $ sendYield sv winfo (ChildYield a)
180 yieldLimitOk <- liftIO $ decrementYieldLimit sv
181 if res && yieldLimitOk
182 then foldStreamShared st yieldk single incrContinue r
184 incrementYieldLimit sv
188 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
190 x : xs -> (xs, Just x)
192-------------------------------------------------------------------------------
194-------------------------------------------------------------------------------
196-- XXX we can remove sv as it is derivable from st
198{-# INLINE workLoopFIFO #-}
200 :: (MonadIO m, MonadBaseControl IO m)
201 => LinkedQueue (Stream m a)
206workLoopFIFO q st sv winfo = run
210 mrun = runInIO $ svarMrun sv
212 work <- liftIO $ tryPopR q
213 let stop = liftIO $ sendStop sv winfo
218 foldStreamShared st yieldk single (return Continue) m
225 res <- liftIO $ sendYield sv winfo (ChildYield a)
226 return $ if res then Continue else Suspend
228 -- XXX in general we would like to yield "n" elements from a single stream
229 -- before moving on to the next. Single element granularity could be too
230 -- expensive in certain cases. Similarly, we can use time limit for
233 res <- liftIO $ sendYield sv winfo (ChildYield a)
234 liftIO $ enqueueFIFO sv q r
235 return $ if res then Continue else Suspend
237{-# INLINE workLoopFIFOLimited #-}
239 :: (MonadIO m, MonadBaseControl IO m)
240 => LinkedQueue (Stream m a)
245workLoopFIFOLimited q st sv winfo = run
249 mrun = runInIO $ svarMrun sv
250 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
252 work <- liftIO $ tryPopR q
253 let stop = liftIO $ sendStop sv winfo
257 yieldLimitOk <- liftIO $ decrementYieldLimit sv
261 foldStreamShared st yieldk single incrContinue m
268 incrementYieldLimit sv
272 res <- liftIO $ sendYield sv winfo (ChildYield a)
273 return $ if res then Continue else Suspend
276 res <- liftIO $ sendYield sv winfo (ChildYield a)
277 liftIO $ enqueueFIFO sv q r
278 yieldLimitOk <- liftIO $ decrementYieldLimit sv
279 if res && yieldLimitOk
282 incrementYieldLimit sv
285-------------------------------------------------------------------------------
287-- This code belongs in SVar.hs but is kept here for perf reasons
288-------------------------------------------------------------------------------
290-- XXX we have this function in this file because passing runStreamLIFO as a
291-- function argument to this function results in a perf degradation of more
292-- than 10%. Need to investigate what the root cause is.
293-- Interestingly, the same thing does not make any difference for Ahead.
294getLifoSVar :: forall m a. MonadAsync m
295 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
296getLifoSVar st mrun = do
297 outQ <- newIORef ([], 0)
298 outQMv <- newEmptyMVar
300 wfw <- newIORef False
301 running <- newIORef S.empty
303 yl <- case getYieldLimit st of
304 Nothing -> return Nothing
305 Just x -> Just <$> newIORef x
306 rateInfo <- getYieldRateInfo st
308 stats <- newSVarStats
311 let isWorkFinished _ = null <$> readIORef q
313 let isWorkFinishedLimited sv = do
315 case remainingWork sv of
319 Nothing -> return False
320 qEmpty <- null <$> readIORef q
321 return $ qEmpty || yieldsDone
323 let getSVar :: SVar Stream m a
324 -> (SVar Stream m a -> m [ChildEvent a])
325 -> (SVar Stream m a -> m Bool)
326 -> (SVar Stream m a -> IO Bool)
327 -> (IORef [Stream m a]
333 getSVar sv readOutput postProc workDone wloop = SVar
335 , outputQueueFromConsumer = undefined
337 , maxBufferLimit = getMaxBuffer st
338 , pushBufferSpace = undefined
339 , pushBufferPolicy = undefined
340 , pushBufferMVar = undefined
341 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
342 , yieldRateInfo = rateInfo
343 , outputDoorBell = outQMv
344 , outputDoorBellFromConsumer = undefined
345 , readOutputQ = readOutput sv
346 , postProcess = postProc sv
347 , workerThreads = running
348 , workLoop = wloop q st{streamVar = Just sv} sv
349 , enqueue = enqueueLIFO sv q
350 , isWorkDone = workDone sv
351 , isQueueDone = workDone sv
353 , svarStyle = AsyncVar
354 , svarStopStyle = StopNone
355 , svarStopBy = undefined
357 , workerCount = active
358 , accountThread = delThread sv
359 , workerStopMVar = undefined
361 , svarInspectMode = getInspectMode st
363 , aheadWorkQueue = undefined
364 , outputHeap = undefined
369 case getStreamRate st of
371 case getYieldLimit st of
372 Nothing -> getSVar sv readOutputQBounded
376 Just _ -> getSVar sv readOutputQBounded
378 isWorkFinishedLimited
381 case getYieldLimit st of
382 Nothing -> getSVar sv readOutputQPaced
386 Just _ -> getSVar sv readOutputQPaced
388 isWorkFinishedLimited
392getFifoSVar :: forall m a. MonadAsync m
393 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
394getFifoSVar st mrun = do
395 outQ <- newIORef ([], 0)
396 outQMv <- newEmptyMVar
398 wfw <- newIORef False
399 running <- newIORef S.empty
401 yl <- case getYieldLimit st of
402 Nothing -> return Nothing
403 Just x -> Just <$> newIORef x
404 rateInfo <- getYieldRateInfo st
406 stats <- newSVarStats
409 let isWorkFinished _ = nullQ q
410 let isWorkFinishedLimited sv = do
412 case remainingWork sv of
416 Nothing -> return False
418 return $ qEmpty || yieldsDone
420 let getSVar :: SVar Stream m a
421 -> (SVar Stream m a -> m [ChildEvent a])
422 -> (SVar Stream m a -> m Bool)
423 -> (SVar Stream m a -> IO Bool)
424 -> (LinkedQueue (Stream m a)
430 getSVar sv readOutput postProc workDone wloop = SVar
432 , outputQueueFromConsumer = undefined
434 , maxBufferLimit = getMaxBuffer st
435 , pushBufferSpace = undefined
436 , pushBufferPolicy = undefined
437 , pushBufferMVar = undefined
438 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
439 , yieldRateInfo = rateInfo
440 , outputDoorBell = outQMv
441 , outputDoorBellFromConsumer = undefined
442 , readOutputQ = readOutput sv
443 , postProcess = postProc sv
444 , workerThreads = running
445 , workLoop = wloop q st{streamVar = Just sv} sv
446 , enqueue = enqueueFIFO sv q
447 , isWorkDone = workDone sv
448 , isQueueDone = workDone sv
450 , svarStyle = WAsyncVar
451 , svarStopStyle = StopNone
452 , svarStopBy = undefined
454 , workerCount = active
455 , accountThread = delThread sv
456 , workerStopMVar = undefined
458 , svarInspectMode = getInspectMode st
460 , aheadWorkQueue = undefined
461 , outputHeap = undefined
466 case getStreamRate st of
468 case getYieldLimit st of
469 Nothing -> getSVar sv readOutputQBounded
473 Just _ -> getSVar sv readOutputQBounded
475 isWorkFinishedLimited
478 case getYieldLimit st of
479 Nothing -> getSVar sv readOutputQPaced
483 Just _ -> getSVar sv readOutputQPaced
485 isWorkFinishedLimited
489{-# INLINABLE newAsyncVar #-}
490newAsyncVar :: MonadAsync m
491 => State Stream m a -> Stream m a -> m (SVar Stream m a)
493 mrun <- captureMonadState
494 sv <- liftIO $ getLifoSVar st mrun
497-- XXX Get rid of this?
498-- | Make a stream asynchronous, triggers the computation and returns a stream
499-- in the underlying monad representing the output generated by the original
500-- computation. The returned action is exhaustible and must be drained once. If
501-- not drained fully we may have a thread blocked forever and once exhausted it
502-- will always return 'empty'.
505{-# INLINABLE mkAsync #-}
506mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
507mkAsync = mkAsync' defState
509{-# INLINABLE mkAsync' #-}
510mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)
511mkAsync' st m = fmap fromSVar (newAsyncVar st (toStream m))
513-- | Create a new SVar and enqueue one stream computation on it.
514{-# INLINABLE newWAsyncVar #-}
515newWAsyncVar :: MonadAsync m
516 => State Stream m a -> Stream m a -> m (SVar Stream m a)
517newWAsyncVar st m = do
518 mrun <- captureMonadState
519 sv <- liftIO $ getFifoSVar st mrun
522------------------------------------------------------------------------------
523-- Running streams concurrently
524------------------------------------------------------------------------------
526-- Concurrency rate control.
528-- Our objective is to create more threads on demand if the consumer is running
529-- faster than us. As soon as we encounter a concurrent composition we create a
530-- push pull pair of threads. We use an SVar for communication between the
531-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
532-- The producer creates more threads if the SVar drains and becomes empty, that
533-- is the consumer is running faster.
535-- XXX Note 1: This mechanism can be problematic if the initial production
536-- latency is high, we may end up creating too many threads. So we need some
537-- way to monitor and use the latency as well. Having a limit on the dispatches
538-- (programmer controlled) may also help.
540-- TBD Note 2: We may want to run computations at the lower level of the
541-- composition tree serially even when they are composed using a parallel
542-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
543-- place of 'wAsync'. If we find that an SVar immediately above a computation
544-- gets drained empty we can switch to parallelizing the computation. For that
545-- we can use a state flag to fork the rest of the computation at any point of
546-- time inside the Monad bind operation if the consumer is running at a faster
549-- TBD Note 3: the binary operation ('parallel') composition allows us to
550-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
551-- chunksizes we will need to compose the parallel actions using a data
552-- constructor (A Free container) instead so that we can divide it in chunks of
553-- arbitrary size before dispatching. If the stream is composed of
554-- hierarchically composed grains of different sizes then we can always switch
555-- to a desired granularity depending on the consumer speed.
557-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
558-- into just the number of CPUs.
560-- | Join two computations on the currently running 'SVar' queue for concurrent
561-- execution. When we are using parallel composition, an SVar is passed around
562-- as a state variable. We try to schedule a new parallel computation on the
563-- SVar passed to us. The first time, when no SVar exists, a new SVar is
564-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
565-- already scheduled on the SVar is further evaluated. For example, when (a
566-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
567-- the current scheduler queue.
569-- The 'SVarStyle' required by the current composition context is passed as one
570-- of the parameters. If the scheduling and composition style of the new
571-- computation being scheduled is different than the style of the current SVar,
572-- then we create a new SVar and schedule it on that. The newly created SVar
573-- joins as one of the computations on the current SVar queue.
575-- Cases when we need to switch to a new SVar:
577-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
578-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
579-- because of the scheduling policy change.
580-- * if we 'adapt' a stream of type 'async' to a stream of type
581-- 'Parallel', we create a new SVar at the transitioning bind.
582-- * When the stream is switching from disjunctive composition to conjunctive
583-- composition and vice-versa we create a new SVar to isolate the scheduling
586forkSVarAsync :: (IsStream t, MonadAsync m)
587 => SVarStyle -> t m a -> t m a -> t m a
588forkSVarAsync style m1 m2 = mkStream $ \st yld sng stp -> do
590 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
591 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
592 _ -> error "illegal svar type"
593 foldStream st yld sng stp $ fromSVar sv
595 concurrently ma mb = mkStream $ \st yld sng stp -> do
596 liftIO $ enqueue (fromJust $ streamVar st) mb
597 foldStreamShared st yld sng stp ma
599{-# INLINE joinStreamVarAsync #-}
600joinStreamVarAsync :: (IsStream t, MonadAsync m)
601 => SVarStyle -> t m a -> t m a -> t m a
602joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
604 Just sv | svarStyle sv == style -> do
605 liftIO $ enqueue sv (toStream m2)
606 foldStreamShared st yld sng stp m1
607 _ -> foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
609------------------------------------------------------------------------------
610-- Semigroup and Monoid style compositions for parallel actions
611------------------------------------------------------------------------------
613-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
614-- Merges two streams possibly concurrently, preferring the
615-- elements from the left one when available.
619async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
620async = joinStreamVarAsync AsyncVar
625{-# DEPRECATED (<|) "Please use 'async' instead." #-}
627(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
630-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
631-- huge difference in the performance of consM in IsStream instance even we
632-- have a SPECIALIZE in the instance.
634-- | XXX we can implement it more efficienty by directly implementing instead
635-- of combining streams using async.
636{-# INLINE consMAsync #-}
637{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
638consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
639consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
641------------------------------------------------------------------------------
643------------------------------------------------------------------------------
645-- | The 'Semigroup' operation (@<>@) for 'AsyncT' merges two streams
646-- concurrently with priority given to the first stream. In @s1 <> s2 <> s3
647-- ...@ the streams s1, s2 and s3 are scheduled for execution in that order.
648-- Multiple scheduled streams may be executed concurrently and the elements
649-- generated by them are served to the consumer as and when they become
650-- available. This behavior is similar to the scheduling and execution behavior
651-- of actions in a single async stream.
653-- Since only a finite number of streams are executed concurrently, this
654-- operation can be used to fold an infinite lazy container of streams.
658-- import qualified "Streamly.Prelude" as S
659-- import Control.Concurrent
661-- main = (S.toList . 'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
667-- Any exceptions generated by a constituent stream are propagated to the
668-- output stream. The output and exceptions from a single stream are guaranteed
669-- to arrive in the same order in the resulting stream as they were generated
670-- in the input stream. However, the relative ordering of elements from
671-- different streams in the resulting stream can vary depending on scheduling
672-- and generation delays.
674-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
675-- concurrently based on demand. More concurrent iterations are started only
676-- if the previous iterations are not able to produce enough output for the
680-- main = 'drain' . 'asyncly' $ do
681-- n <- return 3 \<\> return 2 \<\> return 1
683-- threadDelay (n * 1000000)
684-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
687-- ThreadId 40: Delay 1
688-- ThreadId 39: Delay 2
689-- ThreadId 38: Delay 3
693newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
694 deriving (MonadTrans)
696-- | A demand driven left biased parallely composing IO stream of elements of
697-- type @a@. See 'AsyncT' documentation for more details.
700type Async = AsyncT IO
702-- | Fix the type of a polymorphic stream as 'AsyncT'.
705asyncly :: IsStream t => AsyncT m a -> t m a
708instance IsStream AsyncT where
714------------------------------------------------------------------------------
716------------------------------------------------------------------------------
718-- Monomorphically typed version of "async" for better performance of Semigroup
720{-# INLINE mappendAsync #-}
721{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
722mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
723mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
725instance MonadAsync m => Semigroup (AsyncT m a) where
728------------------------------------------------------------------------------
730------------------------------------------------------------------------------
732instance MonadAsync m => Monoid (AsyncT m a) where
736------------------------------------------------------------------------------
738------------------------------------------------------------------------------
740{-# INLINE apAsync #-}
741{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
742apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
743apAsync (AsyncT m1) (AsyncT m2) =
744 let f x1 = K.concatMapBy async (pure . x1) m2
745 in AsyncT $ K.concatMapBy async f m1
747instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
749 pure = AsyncT . K.yield
753------------------------------------------------------------------------------
755------------------------------------------------------------------------------
757-- GHC: if we change the implementation of bindWith with arguments in a
758-- different order we see a significant performance degradation (~2x).
759{-# INLINE bindAsync #-}
760{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
761bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
762bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
764-- GHC: if we specify arguments in the definition of (>>=) we see a significant
765-- performance degradation (~2x).
766instance MonadAsync m => Monad (AsyncT m) where
770------------------------------------------------------------------------------
772------------------------------------------------------------------------------
774MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
776------------------------------------------------------------------------------
778------------------------------------------------------------------------------
780-- | XXX we can implement it more efficienty by directly implementing instead
781-- of combining streams using wAsync.
782{-# INLINE consMWAsync #-}
783{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
784consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
785consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
787-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
788-- Merges two streams concurrently choosing elements from both fairly.
792wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
793wAsync = joinStreamVarAsync WAsyncVar
795-- | 'WAsyncT' is similar to 'WSerialT' but with concurrent execution.
796-- The 'Semigroup' operation (@<>@) for 'WAsyncT' merges two streams
797-- concurrently interleaving the actions from both the streams. In @s1
798-- <> s2 <> s3 ...@, the individual actions from streams @s1@, @s2@ and @s3@
799-- are scheduled for execution in a round-robin fashion. Multiple scheduled
800-- actions may be executed concurrently, the results from concurrent executions
801-- are consumed in the order in which they become available.
804-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
805-- contrast to the depth wise scheduling behavior of 'AsyncT'.
809-- import qualified "Streamly.Prelude" as S
810-- import Control.Concurrent
812-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
818-- For this example, we are using @maxThreads 1@ so that concurrent thread
819-- scheduling does not affect the results and make them unpredictable. Let's
820-- now take a more general example:
823-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
826-- [1,4,2,7,5,3,8,6,9]
829-- This is how the execution of the above stream proceeds:
831-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
832-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
833-- queue is represented by the rightmost item.
834-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
835-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
836-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
837-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
838-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
839-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
840-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
841-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
842-- 4. Note that the scheduler queue expands by one more stream component in
843-- every pass because one more @<>@ is broken down into two components. At this
844-- point there are no more @<>@ operations to be broken down further and the
845-- queue has reached its maximum size. Now these streams are scheduled in
846-- round-robin fashion yielding @[2,7,5,3,8,8,9]@.
848-- As we see above, in a right associated expression composed with @<>@, only
849-- one @<>@ operation is broken down into two components in one execution,
850-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
851-- scheduler passes to expand the whole expression. By the time @n-th@
852-- component is added to the scheduler queue, the first component would have
853-- received @n@ scheduler passes.
855-- Since all streams get interleaved, this operation is not suitable for
856-- folding an infinite lazy container of infinite size streams. However, if
857-- the streams are small, the streams on the left may get finished before more
858-- streams are added to the scheduler queue from the right side of the
859-- expression, so it may be possible to fold an infinite lazy container of
860-- streams. For example, if the streams are of size @n@ then at most @n@
861-- streams would be in the scheduler queue at a time.
863-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
864-- therefore the output of 'WAsyncT' even with a single thread of execution is
865-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
866-- its scheduling behavior.
868-- Any exceptions generated by a constituent stream are propagated to the
869-- output stream. The output and exceptions from a single stream are guaranteed
870-- to arrive in the same order in the resulting stream as they were generated
871-- in the input stream. However, the relative ordering of elements from
872-- different streams in the resulting stream can vary depending on scheduling
873-- and generation delays.
875-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
876-- concurrently using a round robin scheduling.
879-- main = 'drain' . 'wAsyncly' $ do
880-- n <- return 3 \<\> return 2 \<\> return 1
882-- threadDelay (n * 1000000)
883-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
886-- ThreadId 40: Delay 1
887-- ThreadId 39: Delay 2
888-- ThreadId 38: Delay 3
892newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
893 deriving (MonadTrans)
895-- | A round robin parallely composing IO stream of elements of type @a@.
896-- See 'WAsyncT' documentation for more details.
899type WAsync = WAsyncT IO
901-- | Fix the type of a polymorphic stream as 'WAsyncT'.
904wAsyncly :: IsStream t => WAsyncT m a -> t m a
907instance IsStream WAsyncT where
908 toStream = getWAsyncT
913------------------------------------------------------------------------------
915------------------------------------------------------------------------------
917{-# INLINE mappendWAsync #-}
918{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
919mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
920mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
922instance MonadAsync m => Semigroup (WAsyncT m a) where
925------------------------------------------------------------------------------
927------------------------------------------------------------------------------
929instance MonadAsync m => Monoid (WAsyncT m a) where
933------------------------------------------------------------------------------
935------------------------------------------------------------------------------
937{-# INLINE apWAsync #-}
938{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
939apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
940apWAsync (WAsyncT m1) (WAsyncT m2) =
941 let f x1 = K.concatMapBy wAsync (pure . x1) m2
942 in WAsyncT $ K.concatMapBy wAsync f m1
944-- GHC: if we specify arguments in the definition of (<*>) we see a significant
945-- performance degradation (~2x).
946instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
947 pure = WAsyncT . K.yield
950------------------------------------------------------------------------------
952------------------------------------------------------------------------------
954-- GHC: if we change the implementation of bindWith with arguments in a
955-- different order we see a significant performance degradation (~2x).
956{-# INLINE bindWAsync #-}
957{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
958bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
959bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
961-- GHC: if we specify arguments in the definition of (>>=) we see a significant
962-- performance degradation (~2x).
963instance MonadAsync m => Monad (WAsyncT m) where
967------------------------------------------------------------------------------
969------------------------------------------------------------------------------
971MONAD_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