1{-# LANGUAGE UndecidableInstances #-}
6-- Module : Streamly.Internal.Data.Stream.Async
7-- Copyright : (c) 2017 Composewell Technologies
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
14-- To run examples in this module:
16-- >>> import qualified Streamly.Prelude as Stream
17-- >>> import Control.Concurrent (threadDelay)
20-- threadDelay (n * 1000000) -- sleep for n seconds
21-- putStrLn (show n ++ " sec") -- print "n sec"
25module Streamly.Internal.Data.Stream.Async
42import Control.Concurrent (myThreadId)
43import Control.Monad.Base (MonadBase(..), liftBaseDefault)
44import Control.Monad.Catch (MonadThrow, throwM)
45import Control.Monad.Trans.Control (MonadBaseControl (..))
46import Control.Concurrent.MVar (newEmptyMVar)
47-- import Control.Monad.Error.Class (MonadError(..))
48import Control.Monad.IO.Class (MonadIO(..))
49import Control.Monad.Reader.Class (MonadReader(..))
50import Control.Monad.State.Class (MonadState(..))
51import Control.Monad.Trans.Class (MonadTrans(lift))
52import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR)
53import Data.IORef (IORef, newIORef, readIORef)
54import Data.Maybe (fromJust)
55#if __GLASGOW_HASKELL__ < 808
56import Data.Semigroup (Semigroup(..))
59import Prelude hiding (map)
60import qualified Data.Set as S
62import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)
63import Streamly.Internal.Data.Stream.SVar (fromSVar, fromSVarD)
64import Streamly.Internal.Data.SVar
65import Streamly.Internal.Data.Stream.StreamK.Type
66 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared)
68import qualified Streamly.Internal.Data.Stream.StreamK as K (withLocal)
69import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
70import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
72#include "Instances.hs"
75-- >>> import qualified Streamly.Prelude as Stream
76-- >>> import Control.Concurrent (threadDelay)
79-- threadDelay (n * 1000000) -- sleep for n seconds
80-- putStrLn (show n ++ " sec") -- print "n sec"
85-------------------------------------------------------------------------------
87-------------------------------------------------------------------------------
89data WorkerStatus = Continue | Suspend
91{-# INLINE workLoopLIFO #-}
93 :: (MonadIO m, MonadBaseControl IO m)
94 => IORef [(RunInIO m, Stream m a)]
99workLoopLIFO q st sv winfo = run
103 stop = liftIO $ sendStop sv winfo
108 Just (RunInIO runin, m) -> do
109 -- XXX when we finish we need to send the monadic state back to
110 -- the parent so that the state can be merged back. We capture
111 -- and return the state in the stop continuation.
113 -- Instead of using the run function we can just restore the
114 -- monad state here. That way it can work easily for
115 -- distributed case as well.
116 r <- liftIO $ runin $
117 foldStreamShared st yieldk single (return Continue) m
124 res <- liftIO $ sendYield sv winfo (ChildYield a)
125 return $ if res then Continue else Suspend
128 res <- liftIO $ sendYield sv winfo (ChildYield a)
130 then foldStreamShared st yieldk single (return Continue) r
132 runInIO <- captureMonadState
133 liftIO $ enqueueLIFO sv q (runInIO, r)
136 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
138 x : xs -> (xs, Just x)
140-- We duplicate workLoop for yield limit and no limit cases because it has
141-- around 40% performance overhead in the worst case.
143-- XXX we can pass yinfo directly as an argument here so that we do not have to
144-- make a check every time.
145{-# INLINE workLoopLIFOLimited #-}
147 :: forall m a. (MonadIO m, MonadBaseControl IO m)
148 => IORef [(RunInIO m, Stream m a)]
153workLoopLIFOLimited q st sv winfo = run
157 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
158 stop = liftIO $ sendStop sv winfo
163 Just (RunInIO runin, m) -> do
164 -- XXX This is just a best effort minimization of concurrency
165 -- to the yield limit. If the stream is made of concurrent
166 -- streams we do not reserve the yield limit in the constituent
167 -- streams before executing the action. This can be done
168 -- though, by sharing the yield limit ref with downstream
169 -- actions via state passing. Just a todo.
170 yieldLimitOk <- liftIO $ decrementYieldLimit sv
173 r <- liftIO $ runin $
174 foldStreamShared st yieldk single incrContinue m
179 -- Avoid any side effects, undo the yield limit decrement if we
180 -- never yielded anything.
182 enqueueLIFO sv q (RunInIO runin, m)
183 incrementYieldLimit sv
187 res <- liftIO $ sendYield sv winfo (ChildYield a)
188 return $ if res then Continue else Suspend
190 -- XXX can we pass on the yield limit downstream to limit the concurrency
191 -- of constituent streams.
193 res <- liftIO $ sendYield sv winfo (ChildYield a)
194 yieldLimitOk <- liftIO $ decrementYieldLimit sv
195 if res && yieldLimitOk
196 then foldStreamShared st yieldk single incrContinue r
198 runInIO <- captureMonadState
199 liftIO $ incrementYieldLimit sv
200 liftIO $ enqueueLIFO sv q (runInIO, r)
203 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
205 x : xs -> (xs, Just x)
207-------------------------------------------------------------------------------
209-------------------------------------------------------------------------------
211-- XXX we can remove sv as it is derivable from st
213{-# INLINE workLoopFIFO #-}
215 :: (MonadIO m, MonadBaseControl IO m)
216 => LinkedQueue (RunInIO m, Stream m a)
221workLoopFIFO q st sv winfo = run
225 stop = liftIO $ sendStop sv winfo
227 work <- liftIO $ tryPopR q
230 Just (RunInIO runin, m) -> do
231 r <- liftIO $ runin $
232 foldStreamShared st yieldk single (return Continue) m
239 res <- liftIO $ sendYield sv winfo (ChildYield a)
240 return $ if res then Continue else Suspend
242 -- XXX in general we would like to yield "n" elements from a single stream
243 -- before moving on to the next. Single element granularity could be too
244 -- expensive in certain cases. Similarly, we can use time limit for
247 res <- liftIO $ sendYield sv winfo (ChildYield a)
248 runInIO <- captureMonadState
249 liftIO $ enqueueFIFO sv q (runInIO, r)
250 return $ if res then Continue else Suspend
252{-# INLINE workLoopFIFOLimited #-}
254 :: forall m a. (MonadIO m, MonadBaseControl IO m)
255 => LinkedQueue (RunInIO m, Stream m a)
260workLoopFIFOLimited q st sv winfo = run
264 stop = liftIO $ sendStop sv winfo
265 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
267 work <- liftIO $ tryPopR q
270 Just (RunInIO runin, m) -> do
271 yieldLimitOk <- liftIO $ decrementYieldLimit sv
274 r <- liftIO $ runin $
275 foldStreamShared st yieldk single incrContinue m
281 enqueueFIFO sv q (RunInIO runin, m)
282 incrementYieldLimit sv
286 res <- liftIO $ sendYield sv winfo (ChildYield a)
287 return $ if res then Continue else Suspend
290 res <- liftIO $ sendYield sv winfo (ChildYield a)
291 runInIO <- captureMonadState
292 liftIO $ enqueueFIFO sv q (runInIO, r)
293 yieldLimitOk <- liftIO $ decrementYieldLimit sv
294 if res && yieldLimitOk
297 incrementYieldLimit sv
300-------------------------------------------------------------------------------
302-- This code belongs in SVar.hs but is kept here for perf reasons
303-------------------------------------------------------------------------------
305-- XXX we have this function in this file because passing runStreamLIFO as a
306-- function argument to this function results in a perf degradation of more
307-- than 10%. Need to investigate what the root cause is.
308-- Interestingly, the same thing does not make any difference for Ahead.
309getLifoSVar :: forall m a. MonadAsync m
310 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
311getLifoSVar st mrun = do
312 outQ <- newIORef ([], 0)
313 outQMv <- newEmptyMVar
315 wfw <- newIORef False
316 running <- newIORef S.empty
317 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
318 yl <- case getYieldLimit st of
319 Nothing -> return Nothing
320 Just x -> Just <$> newIORef x
321 rateInfo <- getYieldRateInfo st
323 stats <- newSVarStats
326 let isWorkFinished _ = null <$> readIORef q
328 let isWorkFinishedLimited sv = do
330 case remainingWork sv of
334 Nothing -> return False
335 qEmpty <- null <$> readIORef q
336 return $ qEmpty || yieldsDone
338 let getSVar :: SVar Stream m a
339 -> (SVar Stream m a -> m [ChildEvent a])
340 -> (SVar Stream m a -> m Bool)
341 -> (SVar Stream m a -> IO Bool)
342 -> (IORef [(RunInIO m, Stream m a)]
348 getSVar sv readOutput postProc workDone wloop = SVar
350 , outputQueueFromConsumer = undefined
352 , maxBufferLimit = getMaxBuffer st
353 , pushBufferSpace = undefined
354 , pushBufferPolicy = undefined
355 , pushBufferMVar = undefined
356 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
357 , yieldRateInfo = rateInfo
358 , outputDoorBell = outQMv
359 , outputDoorBellFromConsumer = undefined
360 , readOutputQ = readOutput sv
361 , postProcess = postProc sv
362 , workerThreads = running
363 , workLoop = wloop q st{streamVar = Just sv} sv
364 , enqueue = enqueueLIFO sv q
365 , isWorkDone = workDone sv
366 , isQueueDone = workDone sv
368 , svarStyle = AsyncVar
369 , svarStopStyle = StopNone
370 , svarStopBy = undefined
372 , workerCount = active
373 , accountThread = delThread sv
374 , workerStopMVar = undefined
376 , svarInspectMode = getInspectMode st
378 , aheadWorkQueue = undefined
379 , outputHeap = undefined
384 case getStreamRate st of
386 case getYieldLimit st of
387 Nothing -> getSVar sv readOutputQBounded
391 Just _ -> getSVar sv readOutputQBounded
393 isWorkFinishedLimited
396 case getYieldLimit st of
397 Nothing -> getSVar sv readOutputQPaced
401 Just _ -> getSVar sv readOutputQPaced
403 isWorkFinishedLimited
407getFifoSVar :: forall m a. MonadAsync m
408 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
409getFifoSVar st mrun = do
410 outQ <- newIORef ([], 0)
411 outQMv <- newEmptyMVar
413 wfw <- newIORef False
414 running <- newIORef S.empty
416 yl <- case getYieldLimit st of
417 Nothing -> return Nothing
418 Just x -> Just <$> newIORef x
419 rateInfo <- getYieldRateInfo st
421 stats <- newSVarStats
424 let isWorkFinished _ = nullQ q
425 let isWorkFinishedLimited sv = do
427 case remainingWork sv of
431 Nothing -> return False
433 return $ qEmpty || yieldsDone
435 let getSVar :: SVar Stream m a
436 -> (SVar Stream m a -> m [ChildEvent a])
437 -> (SVar Stream m a -> m Bool)
438 -> (SVar Stream m a -> IO Bool)
439 -> (LinkedQueue (RunInIO m, Stream m a)
445 getSVar sv readOutput postProc workDone wloop = SVar
447 , outputQueueFromConsumer = undefined
449 , maxBufferLimit = getMaxBuffer st
450 , pushBufferSpace = undefined
451 , pushBufferPolicy = undefined
452 , pushBufferMVar = undefined
453 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
454 , yieldRateInfo = rateInfo
455 , outputDoorBell = outQMv
456 , outputDoorBellFromConsumer = undefined
457 , readOutputQ = readOutput sv
458 , postProcess = postProc sv
459 , workerThreads = running
460 , workLoop = wloop q st{streamVar = Just sv} sv
461 , enqueue = enqueueFIFO sv q
462 , isWorkDone = workDone sv
463 , isQueueDone = workDone sv
465 , svarStyle = WAsyncVar
466 , svarStopStyle = StopNone
467 , svarStopBy = undefined
469 , workerCount = active
470 , accountThread = delThread sv
471 , workerStopMVar = undefined
473 , svarInspectMode = getInspectMode st
475 , aheadWorkQueue = undefined
476 , outputHeap = undefined
481 case getStreamRate st of
483 case getYieldLimit st of
484 Nothing -> getSVar sv readOutputQBounded
488 Just _ -> getSVar sv readOutputQBounded
490 isWorkFinishedLimited
493 case getYieldLimit st of
494 Nothing -> getSVar sv readOutputQPaced
498 Just _ -> getSVar sv readOutputQPaced
500 isWorkFinishedLimited
504{-# INLINABLE newAsyncVar #-}
505newAsyncVar :: MonadAsync m
506 => State Stream m a -> Stream m a -> m (SVar Stream m a)
508 mrun <- captureMonadState
509 sv <- liftIO $ getLifoSVar st mrun
512-- | Generate a stream asynchronously to keep it buffered, lazily consume
517{-# INLINABLE mkAsyncK #-}
518mkAsyncK :: (IsStream t, MonadAsync m) => t m a -> t m a
519mkAsyncK m = mkStream $ \st yld sng stp -> do
520 sv <- newAsyncVar (adaptState st) (toStream m)
521 foldStream st yld sng stp $ fromSVar sv
523{-# INLINE_NORMAL mkAsyncD #-}
524mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
525mkAsyncD m = D.Stream step Nothing
528 step gst Nothing = do
529 sv <- newAsyncVar gst (D.fromStreamD m)
530 return $ D.Skip $ Just $ fromSVarD sv
532 step gst (Just (D.UnStream step1 st)) = do
535 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
536 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
540-- This is slightly faster than the CPS version above
542-- | Make the stream producer and consumer run concurrently by introducing a
543-- buffer between them. The producer thread evaluates the input stream until
544-- the buffer fills, it terminates if the buffer is full and a worker thread is
545-- kicked off again to evaluate the remaining stream when there is space in the
546-- buffer. The consumer consumes the stream lazily from the buffer.
548-- /Since: 0.2.0 (Streamly)/
552{-# INLINE_NORMAL mkAsync #-}
553mkAsync :: (K.IsStream t, MonadAsync m) => t m a -> t m a
554mkAsync = D.fromStreamD . mkAsyncD . D.toStreamD
556-- | Create a new SVar and enqueue one stream computation on it.
557{-# INLINABLE newWAsyncVar #-}
558newWAsyncVar :: MonadAsync m
559 => State Stream m a -> Stream m a -> m (SVar Stream m a)
560newWAsyncVar st m = do
561 mrun <- captureMonadState
562 sv <- liftIO $ getFifoSVar st mrun
565------------------------------------------------------------------------------
566-- Running streams concurrently
567------------------------------------------------------------------------------
569-- Concurrency rate control.
571-- Our objective is to create more threads on demand if the consumer is running
572-- faster than us. As soon as we encounter a concurrent composition we create a
573-- push pull pair of threads. We use an SVar for communication between the
574-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
575-- The producer creates more threads if the SVar drains and becomes empty, that
576-- is the consumer is running faster.
578-- XXX Note 1: This mechanism can be problematic if the initial production
579-- latency is high, we may end up creating too many threads. So we need some
580-- way to monitor and use the latency as well. Having a limit on the dispatches
581-- (programmer controlled) may also help.
583-- TBD Note 2: We may want to run computations at the lower level of the
584-- composition tree serially even when they are composed using a parallel
585-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
586-- place of 'wAsync'. If we find that an SVar immediately above a computation
587-- gets drained empty we can switch to parallelizing the computation. For that
588-- we can use a state flag to fork the rest of the computation at any point of
589-- time inside the Monad bind operation if the consumer is running at a faster
592-- TBD Note 3: the binary operation ('parallel') composition allows us to
593-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
594-- chunksizes we will need to compose the parallel actions using a data
595-- constructor (A Free container) instead so that we can divide it in chunks of
596-- arbitrary size before dispatching. If the stream is composed of
597-- hierarchically composed grains of different sizes then we can always switch
598-- to a desired granularity depending on the consumer speed.
600-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
601-- into just the number of CPUs.
603-- | Join two computations on the currently running 'SVar' queue for concurrent
604-- execution. When we are using parallel composition, an SVar is passed around
605-- as a state variable. We try to schedule a new parallel computation on the
606-- SVar passed to us. The first time, when no SVar exists, a new SVar is
607-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
608-- already scheduled on the SVar is further evaluated. For example, when (a
609-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
610-- the current scheduler queue.
612-- The 'SVarStyle' required by the current composition context is passed as one
613-- of the parameters. If the scheduling and composition style of the new
614-- computation being scheduled is different than the style of the current SVar,
615-- then we create a new SVar and schedule it on that. The newly created SVar
616-- joins as one of the computations on the current SVar queue.
618-- Cases when we need to switch to a new SVar:
620-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
621-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
622-- because of the scheduling policy change.
623-- * if we 'adapt' a stream of type 'async' to a stream of type
624-- 'Parallel', we create a new SVar at the transitioning bind.
625-- * When the stream is switching from disjunctive composition to conjunctive
626-- composition and vice-versa we create a new SVar to isolate the scheduling
629forkSVarAsync :: (IsStream t, MonadAsync m)
630 => SVarStyle -> t m a -> t m a -> t m a
631forkSVarAsync style m1 m2 = mkStream $ \st yld sng stp -> do
633 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
634 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
635 _ -> error "illegal svar type"
636 foldStream st yld sng stp $ fromSVar sv
638 concurrently ma mb = mkStream $ \st yld sng stp -> do
639 runInIO <- captureMonadState
640 liftIO $ enqueue (fromJust $ streamVar st) $ (runInIO, mb)
641 foldStreamShared st yld sng stp ma
643{-# INLINE joinStreamVarAsync #-}
644joinStreamVarAsync :: (IsStream t, MonadAsync m)
645 => SVarStyle -> t m a -> t m a -> t m a
646joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
648 Just sv | svarStyle sv == style -> do
649 runInIO <- captureMonadState
650 liftIO $ enqueue sv $ (runInIO, toStream m2)
651 foldStreamShared st yld sng stp m1
652 _ -> foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
654------------------------------------------------------------------------------
655-- Semigroup and Monoid style compositions for parallel actions
656------------------------------------------------------------------------------
660-- | Merges two streams, both the streams may be evaluated concurrently,
661-- outputs from both are used as they arrive:
663-- >>> import Streamly.Prelude (async)
664-- >>> stream1 = Stream.fromEffect (delay 4)
665-- >>> stream2 = Stream.fromEffect (delay 2)
666-- >>> Stream.toList $ stream1 `async` stream2
671-- Multiple streams can be combined. With enough threads, all of them can be
672-- scheduled simultaneously:
674-- >>> stream3 = Stream.fromEffect (delay 1)
675-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
679-- With 2 threads, only two can be scheduled at a time, when one of those
680-- finishes, the third one gets scheduled:
682-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
686-- With a single thread, it becomes serial:
688-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
692-- Only streams are scheduled for async evaluation, how actions within a
693-- stream are evaluated depends on the stream type. If it is a concurrent
694-- stream they will be evaluated concurrently.
696-- In the following example, both the streams are scheduled for concurrent
697-- evaluation but each individual stream is evaluated serially:
699-- >>> stream1 = Stream.fromListM $ Prelude.map delay [3,3] -- SerialT IO Int
700-- >>> stream2 = Stream.fromListM $ Prelude.map delay [1,1] -- SerialT IO Int
701-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
705-- If total threads are 2, the third stream is scheduled only after one of the
706-- first two has finished:
708-- >>> stream3 = Stream.fromListM $ Prelude.map delay [2,2] -- SerialT IO Int
709-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
713-- Thus 'async' goes deep in first few streams rather than going wide in all
714-- streams. It prefers to evaluate the leftmost streams as much as possible.
715-- Because of this behavior, 'async' can be safely used to fold an infinite
716-- lazy container of streams.
718-- /Since: 0.2.0 ("Streamly")/
722async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
723async = joinStreamVarAsync AsyncVar
728{-# DEPRECATED (<|) "Please use 'async' instead." #-}
730(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
733-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
734-- huge difference in the performance of consM in IsStream instance even we
735-- have a SPECIALIZE in the instance.
737-- | XXX we can implement it more efficienty by directly implementing instead
738-- of combining streams using async.
739{-# INLINE consMAsync #-}
740{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
741consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
742consMAsync m r = fromStream $ K.fromEffect m `async` (toStream r)
744------------------------------------------------------------------------------
746------------------------------------------------------------------------------
748-- | For 'AsyncT' streams:
751-- (<>) = 'Streamly.Prelude.async'
752-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
755-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
756-- executed concurrently a la the 'async' combinator, producing results and
757-- side effects of iterations out of order:
760-- Stream.toList $ Stream.fromAsync $ do
761-- x <- Stream.fromList [2,1] -- foreach x in stream
762-- Stream.fromEffect $ delay x
768-- Nested monad binds behave like nested @for@ loops with nested iterations
769-- executed concurrently, a la the 'async' combinator:
772-- Stream.toList $ Stream.fromAsync $ do
773-- x <- Stream.fromList [1,2] -- foreach x in stream
774-- y <- Stream.fromList [2,4] -- foreach y in stream
775-- Stream.fromEffect $ delay (x + y)
783-- The behavior can be explained as follows. All the iterations corresponding
784-- to the element @1@ in the first stream constitute one output stream and all
785-- the iterations corresponding to @2@ constitute another output stream and
786-- these two output streams are merged using 'async'.
788-- /Since: 0.1.0 ("Streamly")/
791newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
792 deriving (MonadTrans)
794-- | A demand driven left biased parallely composing IO stream of elements of
795-- type @a@. See 'AsyncT' documentation for more details.
797-- /Since: 0.2.0 ("Streamly")/
800type Async = AsyncT IO
802-- | Fix the type of a polymorphic stream as 'AsyncT'.
804-- /Since: 0.1.0 ("Streamly")/
807fromAsync :: IsStream t => AsyncT m a -> t m a
810instance IsStream AsyncT where
816------------------------------------------------------------------------------
818------------------------------------------------------------------------------
820-- Monomorphically typed version of "async" for better performance of Semigroup
822{-# INLINE mappendAsync #-}
823{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
824mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
825mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
827instance MonadAsync m => Semigroup (AsyncT m a) where
830------------------------------------------------------------------------------
832------------------------------------------------------------------------------
834instance MonadAsync m => Monoid (AsyncT m a) where
838------------------------------------------------------------------------------
840------------------------------------------------------------------------------
842{-# INLINE apAsync #-}
843{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
844apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
845apAsync (AsyncT m1) (AsyncT m2) =
846 let f x1 = K.concatMapBy async (pure . x1) m2
847 in AsyncT $ K.concatMapBy async f m1
849instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
851 pure = AsyncT . K.fromPure
855------------------------------------------------------------------------------
857------------------------------------------------------------------------------
859-- GHC: if we change the implementation of bindWith with arguments in a
860-- different order we see a significant performance degradation (~2x).
861{-# INLINE bindAsync #-}
862{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
863bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
864bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
866-- GHC: if we specify arguments in the definition of (>>=) we see a significant
867-- performance degradation (~2x).
868instance MonadAsync m => Monad (AsyncT m) where
872------------------------------------------------------------------------------
874------------------------------------------------------------------------------
876MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
878------------------------------------------------------------------------------
880------------------------------------------------------------------------------
882-- | XXX we can implement it more efficienty by directly implementing instead
883-- of combining streams using wAsync.
884{-# INLINE consMWAsync #-}
885{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
886consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
887consMWAsync m r = fromStream $ K.fromEffect m `wAsync` (toStream r)
891-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
892-- singleton stream behavior. For multi-element streams, while 'async' is left
893-- biased i.e. it tries to evaluate the left side stream as much as possible,
894-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
895-- deep while 'wAsync' goes wide. However, outputs are always used as they
898-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
899-- starts behaving like 'wSerial'.
901-- >>> import Streamly.Prelude (wAsync)
902-- >>> stream1 = Stream.fromList [1,2,3]
903-- >>> stream2 = Stream.fromList [4,5,6]
904-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 1 $ stream1 `async` stream2
907-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
910-- With two threads available, and combining three streams:
912-- >>> stream3 = Stream.fromList [7,8,9]
913-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
914-- [1,2,3,4,5,6,7,8,9]
916-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
917-- [1,4,2,7,5,3,8,6,9]
919-- This operation cannot be used to fold an infinite lazy container of streams,
920-- because it schedules all the streams in a round robin manner.
922-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
923-- but the exact scheduling is slightly different in both cases.
927-- /Since: 0.2.0 ("Streamly")/
929-- Scheduling details:
931-- This is how the execution of the above example proceeds:
933-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
934-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
935-- queue is represented by the rightmost item.
936-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
937-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
938-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
939-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
940-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
941-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
942-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
943-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
944-- 4. Note that the scheduler queue expands by one more stream component in
945-- every pass because one more @<>@ is broken down into two components. At this
946-- point there are no more @<>@ operations to be broken down further and the
947-- queue has reached its maximum size. Now these streams are scheduled in
948-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
950-- As we see above, in a right associated expression composed with @<>@, only
951-- one @<>@ operation is broken down into two components in one execution,
952-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
953-- scheduler passes to expand the whole expression. By the time @n-th@
954-- component is added to the scheduler queue, the first component would have
955-- received @n@ scheduler passes.
958wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
959wAsync = joinStreamVarAsync WAsyncVar
961-- | For 'WAsyncT' streams:
964-- (<>) = 'Streamly.Prelude.wAsync'
965-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
968-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
969-- executed concurrently a la the 'wAsync' combinator, producing results and
970-- side effects of iterations out of order:
973-- Stream.toList $ Stream.fromWAsync $ do
974-- x <- Stream.fromList [2,1] -- foreach x in stream
975-- Stream.fromEffect $ delay x
981-- Nested monad binds behave like nested @for@ loops with nested iterations
982-- executed concurrently, a la the 'wAsync' combinator:
985-- Stream.toList $ Stream.fromWAsync $ do
986-- x <- Stream.fromList [1,2] -- foreach x in stream
987-- y <- Stream.fromList [2,4] -- foreach y in stream
988-- Stream.fromEffect $ delay (x + y)
996-- The behavior can be explained as follows. All the iterations corresponding
997-- to the element @1@ in the first stream constitute one 'WAsyncT' output
998-- stream and all the iterations corresponding to @2@ constitute another
999-- 'WAsyncT' output stream and these two output streams are merged using
1002-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
1003-- contrast to the depth wise scheduling behavior of 'AsyncT'.
1005-- /Since: 0.2.0 ("Streamly")/
1009-- XXX This documentation is redundant, need to be cleaned up/removed.
1012-- import qualified "Streamly.Prelude" as S
1013-- import Control.Concurrent
1015-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
1021-- For this example, we are using @maxThreads 1@ so that concurrent thread
1022-- scheduling does not affect the results and make them unpredictable. Let's
1023-- now take a more general example:
1026-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
1029-- [1,4,2,7,5,3,8,6,9]
1032-- This is how the execution of the above stream proceeds:
1034-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
1035-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
1036-- queue is represented by the rightmost item.
1037-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
1038-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
1039-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
1040-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
1041-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
1042-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
1043-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
1044-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
1045-- 4. Note that the scheduler queue expands by one more stream component in
1046-- every pass because one more @<>@ is broken down into two components. At this
1047-- point there are no more @<>@ operations to be broken down further and the
1048-- queue has reached its maximum size. Now these streams are scheduled in
1049-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
1051-- As we see above, in a right associated expression composed with @<>@, only
1052-- one @<>@ operation is broken down into two components in one execution,
1053-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
1054-- scheduler passes to expand the whole expression. By the time @n-th@
1055-- component is added to the scheduler queue, the first component would have
1056-- received @n@ scheduler passes.
1058-- Since all streams get interleaved, this operation is not suitable for
1059-- folding an infinite lazy container of infinite size streams. However, if
1060-- the streams are small, the streams on the left may get finished before more
1061-- streams are added to the scheduler queue from the right side of the
1062-- expression, so it may be possible to fold an infinite lazy container of
1063-- streams. For example, if the streams are of size @n@ then at most @n@
1064-- streams would be in the scheduler queue at a time.
1066-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
1067-- therefore the output of 'WAsyncT' even with a single thread of execution is
1068-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
1069-- its scheduling behavior.
1071-- Any exceptions generated by a constituent stream are propagated to the
1072-- output stream. The output and exceptions from a single stream are guaranteed
1073-- to arrive in the same order in the resulting stream as they were generated
1074-- in the input stream. However, the relative ordering of elements from
1075-- different streams in the resulting stream can vary depending on scheduling
1076-- and generation delays.
1078-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
1079-- concurrently using a round robin scheduling.
1082-- main = S.'drain' . S.'fromWAsync' $ do
1083-- n <- return 3 \<\> return 2 \<\> return 1
1085-- threadDelay (n * 1000000)
1086-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1089-- ThreadId 40: Delay 1
1090-- ThreadId 39: Delay 2
1091-- ThreadId 38: Delay 3
1094newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
1095 deriving (MonadTrans)
1097-- | A round robin parallely composing IO stream of elements of type @a@.
1098-- See 'WAsyncT' documentation for more details.
1100-- /Since: 0.2.0 ("Streamly")/
1103type WAsync = WAsyncT IO
1105-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1107-- /Since: 0.2.0 ("Streamly")/
1110fromWAsync :: IsStream t => WAsyncT m a -> t m a
1113instance IsStream WAsyncT where
1114 toStream = getWAsyncT
1115 fromStream = WAsyncT
1119------------------------------------------------------------------------------
1121------------------------------------------------------------------------------
1123{-# INLINE mappendWAsync #-}
1124{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1125mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
1126mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
1128instance MonadAsync m => Semigroup (WAsyncT m a) where
1129 (<>) = mappendWAsync
1131------------------------------------------------------------------------------
1133------------------------------------------------------------------------------
1135instance MonadAsync m => Monoid (WAsyncT m a) where
1139------------------------------------------------------------------------------
1141------------------------------------------------------------------------------
1143{-# INLINE apWAsync #-}
1144{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
1145apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
1146apWAsync (WAsyncT m1) (WAsyncT m2) =
1147 let f x1 = K.concatMapBy wAsync (pure . x1) m2
1148 in WAsyncT $ K.concatMapBy wAsync f m1
1150-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1151-- performance degradation (~2x).
1152instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1153 pure = WAsyncT . K.fromPure
1156------------------------------------------------------------------------------
1158------------------------------------------------------------------------------
1160-- GHC: if we change the implementation of bindWith with arguments in a
1161-- different order we see a significant performance degradation (~2x).
1162{-# INLINE bindWAsync #-}
1163{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1164bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1165bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
1167-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1168-- performance degradation (~2x).
1169instance MonadAsync m => Monad (WAsyncT m) where
1173------------------------------------------------------------------------------
1175------------------------------------------------------------------------------
1177MONAD_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