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, pushL)
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.Control.Concurrent
63 (MonadAsync, RunInIO(..), captureMonadState)
64import Streamly.Internal.Data.Atomics
65 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
66import Streamly.Internal.Data.Stream.SVar.Generate (fromSVar, fromSVarD)
67import Streamly.Internal.Data.Stream.StreamK.Type
68 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared)
70import qualified Streamly.Internal.Data.Stream.StreamK as K (withLocal)
71import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
72import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
74import Streamly.Internal.Data.SVar.Type
75import Streamly.Internal.Data.SVar
77#include "Instances.hs"
80-- >>> import qualified Streamly.Prelude as Stream
81-- >>> import Control.Concurrent (threadDelay)
84-- threadDelay (n * 1000000) -- sleep for n seconds
85-- putStrLn (show n ++ " sec") -- print "n sec"
90-------------------------------------------------------------------------------
92-------------------------------------------------------------------------------
94-- Note: For purely right associated expressions this queue should have at most
95-- one element. It grows to more than one when we have left associcated
96-- expressions. Large left associated compositions can grow this to a
98{-# INLINE enqueueLIFO #-}
100 SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
101enqueueLIFO sv q m = do
102 atomicModifyIORefCAS_ q $ \ms -> m : ms
105data WorkerStatus = Continue | Suspend
107{-# INLINE workLoopLIFO #-}
109 :: (MonadIO m, MonadBaseControl IO m)
110 => IORef [(RunInIO m, Stream m a)]
115workLoopLIFO q st sv winfo = run
119 stop = liftIO $ sendStop sv winfo
124 Just (RunInIO runin, m) -> do
125 -- XXX when we finish we need to send the monadic state back to
126 -- the parent so that the state can be merged back. We capture
127 -- and return the state in the stop continuation.
129 -- Instead of using the run function we can just restore the
130 -- monad state here. That way it can work easily for
131 -- distributed case as well.
132 r <- liftIO $ runin $
133 foldStreamShared st yieldk single (return Continue) m
140 res <- liftIO $ sendYield sv winfo (ChildYield a)
141 return $ if res then Continue else Suspend
144 res <- liftIO $ sendYield sv winfo (ChildYield a)
146 then foldStreamShared st yieldk single (return Continue) r
148 runInIO <- captureMonadState
149 liftIO $ enqueueLIFO sv q (runInIO, r)
152 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
154 x : xs -> (xs, Just x)
156-- We duplicate workLoop for yield limit and no limit cases because it has
157-- around 40% performance overhead in the worst case.
159-- XXX we can pass yinfo directly as an argument here so that we do not have to
160-- make a check every time.
161{-# INLINE workLoopLIFOLimited #-}
163 :: forall m a. (MonadIO m, MonadBaseControl IO m)
164 => IORef [(RunInIO m, Stream m a)]
169workLoopLIFOLimited q st sv winfo = run
173 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
174 stop = liftIO $ sendStop sv winfo
179 Just (RunInIO runin, m) -> do
180 -- XXX This is just a best effort minimization of concurrency
181 -- to the yield limit. If the stream is made of concurrent
182 -- streams we do not reserve the yield limit in the constituent
183 -- streams before executing the action. This can be done
184 -- though, by sharing the yield limit ref with downstream
185 -- actions via state passing. Just a todo.
186 yieldLimitOk <- liftIO $ decrementYieldLimit sv
189 r <- liftIO $ runin $
190 foldStreamShared st yieldk single incrContinue m
195 -- Avoid any side effects, undo the yield limit decrement if we
196 -- never yielded anything.
198 enqueueLIFO sv q (RunInIO runin, m)
199 incrementYieldLimit sv
203 res <- liftIO $ sendYield sv winfo (ChildYield a)
204 return $ if res then Continue else Suspend
206 -- XXX can we pass on the yield limit downstream to limit the concurrency
207 -- of constituent streams.
209 res <- liftIO $ sendYield sv winfo (ChildYield a)
210 yieldLimitOk <- liftIO $ decrementYieldLimit sv
211 if res && yieldLimitOk
212 then foldStreamShared st yieldk single incrContinue r
214 runInIO <- captureMonadState
215 liftIO $ incrementYieldLimit sv
216 liftIO $ enqueueLIFO sv q (runInIO, r)
219 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
221 x : xs -> (xs, Just x)
223-------------------------------------------------------------------------------
225-------------------------------------------------------------------------------
227-- XXX we can use the Ahead style sequence/heap mechanism to make the best
228-- effort to always try to finish the streams on the left side of an expression
229-- first as long as possible.
231{-# INLINE enqueueFIFO #-}
234 -> LinkedQueue (RunInIO m, t m a)
235 -> (RunInIO m, t m a)
237enqueueFIFO sv q m = do
241-- XXX we can remove sv as it is derivable from st
243{-# INLINE workLoopFIFO #-}
245 :: (MonadIO m, MonadBaseControl IO m)
246 => LinkedQueue (RunInIO m, Stream m a)
251workLoopFIFO q st sv winfo = run
255 stop = liftIO $ sendStop sv winfo
257 work <- liftIO $ tryPopR q
260 Just (RunInIO runin, m) -> do
261 r <- liftIO $ runin $
262 foldStreamShared st yieldk single (return Continue) m
269 res <- liftIO $ sendYield sv winfo (ChildYield a)
270 return $ if res then Continue else Suspend
272 -- XXX in general we would like to yield "n" elements from a single stream
273 -- before moving on to the next. Single element granularity could be too
274 -- expensive in certain cases. Similarly, we can use time limit for
277 res <- liftIO $ sendYield sv winfo (ChildYield a)
278 runInIO <- captureMonadState
279 liftIO $ enqueueFIFO sv q (runInIO, r)
280 return $ if res then Continue else Suspend
282{-# INLINE workLoopFIFOLimited #-}
284 :: forall m a. (MonadIO m, MonadBaseControl IO m)
285 => LinkedQueue (RunInIO m, Stream m a)
290workLoopFIFOLimited q st sv winfo = run
294 stop = liftIO $ sendStop sv winfo
295 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
297 work <- liftIO $ tryPopR q
300 Just (RunInIO runin, m) -> do
301 yieldLimitOk <- liftIO $ decrementYieldLimit sv
304 r <- liftIO $ runin $
305 foldStreamShared st yieldk single incrContinue m
311 enqueueFIFO sv q (RunInIO runin, m)
312 incrementYieldLimit sv
316 res <- liftIO $ sendYield sv winfo (ChildYield a)
317 return $ if res then Continue else Suspend
320 res <- liftIO $ sendYield sv winfo (ChildYield a)
321 runInIO <- captureMonadState
322 liftIO $ enqueueFIFO sv q (runInIO, r)
323 yieldLimitOk <- liftIO $ decrementYieldLimit sv
324 if res && yieldLimitOk
327 incrementYieldLimit sv
330-------------------------------------------------------------------------------
332-- This code belongs in SVar.hs but is kept here for perf reasons
333-------------------------------------------------------------------------------
335-- XXX we have this function in this file because passing runStreamLIFO as a
336-- function argument to this function results in a perf degradation of more
337-- than 10%. Need to investigate what the root cause is.
338-- Interestingly, the same thing does not make any difference for Ahead.
339getLifoSVar :: forall m a. MonadAsync m
340 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
341getLifoSVar st mrun = do
342 outQ <- newIORef ([], 0)
343 outQMv <- newEmptyMVar
345 wfw <- newIORef False
346 running <- newIORef S.empty
347 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
348 yl <- case getYieldLimit st of
349 Nothing -> return Nothing
350 Just x -> Just <$> newIORef x
351 rateInfo <- getYieldRateInfo st
353 stats <- newSVarStats
356 let isWorkFinished _ = null <$> readIORef q
358 let isWorkFinishedLimited sv = do
360 case remainingWork sv of
364 Nothing -> return False
365 qEmpty <- null <$> readIORef q
366 return $ qEmpty || yieldsDone
368 let getSVar :: SVar Stream m a
369 -> (SVar Stream m a -> m [ChildEvent a])
370 -> (SVar Stream m a -> m Bool)
371 -> (SVar Stream m a -> IO Bool)
372 -> (IORef [(RunInIO m, Stream m a)]
378 getSVar sv readOutput postProc workDone wloop = SVar
380 , outputQueueFromConsumer = undefined
382 , maxBufferLimit = getMaxBuffer st
383 , pushBufferSpace = undefined
384 , pushBufferPolicy = undefined
385 , pushBufferMVar = undefined
386 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
387 , yieldRateInfo = rateInfo
388 , outputDoorBell = outQMv
389 , outputDoorBellFromConsumer = undefined
390 , readOutputQ = readOutput sv
391 , postProcess = postProc sv
392 , workerThreads = running
393 , workLoop = wloop q st{streamVar = Just sv} sv
394 , enqueue = enqueueLIFO sv q
395 , isWorkDone = workDone sv
396 , isQueueDone = workDone sv
398 , svarStyle = AsyncVar
399 , svarStopStyle = StopNone
400 , svarStopBy = undefined
402 , workerCount = active
403 , accountThread = delThread sv
404 , workerStopMVar = undefined
406 , svarInspectMode = getInspectMode st
408 , aheadWorkQueue = undefined
409 , outputHeap = undefined
414 case getStreamRate st of
416 case getYieldLimit st of
417 Nothing -> getSVar sv readOutputQBounded
421 Just _ -> getSVar sv readOutputQBounded
423 isWorkFinishedLimited
426 case getYieldLimit st of
427 Nothing -> getSVar sv readOutputQPaced
431 Just _ -> getSVar sv readOutputQPaced
433 isWorkFinishedLimited
437getFifoSVar :: forall m a. MonadAsync m
438 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
439getFifoSVar st mrun = do
440 outQ <- newIORef ([], 0)
441 outQMv <- newEmptyMVar
443 wfw <- newIORef False
444 running <- newIORef S.empty
446 yl <- case getYieldLimit st of
447 Nothing -> return Nothing
448 Just x -> Just <$> newIORef x
449 rateInfo <- getYieldRateInfo st
451 stats <- newSVarStats
454 let isWorkFinished _ = nullQ q
455 let isWorkFinishedLimited sv = do
457 case remainingWork sv of
461 Nothing -> return False
463 return $ qEmpty || yieldsDone
465 let getSVar :: SVar Stream m a
466 -> (SVar Stream m a -> m [ChildEvent a])
467 -> (SVar Stream m a -> m Bool)
468 -> (SVar Stream m a -> IO Bool)
469 -> (LinkedQueue (RunInIO m, Stream m a)
475 getSVar sv readOutput postProc workDone wloop = SVar
477 , outputQueueFromConsumer = undefined
479 , maxBufferLimit = getMaxBuffer st
480 , pushBufferSpace = undefined
481 , pushBufferPolicy = undefined
482 , pushBufferMVar = undefined
483 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
484 , yieldRateInfo = rateInfo
485 , outputDoorBell = outQMv
486 , outputDoorBellFromConsumer = undefined
487 , readOutputQ = readOutput sv
488 , postProcess = postProc sv
489 , workerThreads = running
490 , workLoop = wloop q st{streamVar = Just sv} sv
491 , enqueue = enqueueFIFO sv q
492 , isWorkDone = workDone sv
493 , isQueueDone = workDone sv
495 , svarStyle = WAsyncVar
496 , svarStopStyle = StopNone
497 , svarStopBy = undefined
499 , workerCount = active
500 , accountThread = delThread sv
501 , workerStopMVar = undefined
503 , svarInspectMode = getInspectMode st
505 , aheadWorkQueue = undefined
506 , outputHeap = undefined
511 case getStreamRate st of
513 case getYieldLimit st of
514 Nothing -> getSVar sv readOutputQBounded
518 Just _ -> getSVar sv readOutputQBounded
520 isWorkFinishedLimited
523 case getYieldLimit st of
524 Nothing -> getSVar sv readOutputQPaced
528 Just _ -> getSVar sv readOutputQPaced
530 isWorkFinishedLimited
534{-# INLINABLE newAsyncVar #-}
535newAsyncVar :: MonadAsync m
536 => State Stream m a -> Stream m a -> m (SVar Stream m a)
538 mrun <- captureMonadState
539 sv <- liftIO $ getLifoSVar st mrun
542-- | Generate a stream asynchronously to keep it buffered, lazily consume
547{-# INLINABLE mkAsyncK #-}
548mkAsyncK :: (IsStream t, MonadAsync m) => t m a -> t m a
549mkAsyncK m = mkStream $ \st yld sng stp -> do
550 sv <- newAsyncVar (adaptState st) (toStream m)
551 foldStream st yld sng stp $ fromSVar sv
553{-# INLINE_NORMAL mkAsyncD #-}
554mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
555mkAsyncD m = D.Stream step Nothing
558 step gst Nothing = do
559 sv <- newAsyncVar gst (D.fromStreamD m)
560 return $ D.Skip $ Just $ fromSVarD sv
562 step gst (Just (D.UnStream step1 st)) = do
565 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
566 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
570-- This is slightly faster than the CPS version above
572-- | Make the stream producer and consumer run concurrently by introducing a
573-- buffer between them. The producer thread evaluates the input stream until
574-- the buffer fills, it terminates if the buffer is full and a worker thread is
575-- kicked off again to evaluate the remaining stream when there is space in the
576-- buffer. The consumer consumes the stream lazily from the buffer.
578-- /Since: 0.2.0 (Streamly)/
582{-# INLINE_NORMAL mkAsync #-}
583mkAsync :: (K.IsStream t, MonadAsync m) => t m a -> t m a
584mkAsync = D.fromStreamD . mkAsyncD . D.toStreamD
586-- | Create a new SVar and enqueue one stream computation on it.
587{-# INLINABLE newWAsyncVar #-}
588newWAsyncVar :: MonadAsync m
589 => State Stream m a -> Stream m a -> m (SVar Stream m a)
590newWAsyncVar st m = do
591 mrun <- captureMonadState
592 sv <- liftIO $ getFifoSVar st mrun
595------------------------------------------------------------------------------
596-- Running streams concurrently
597------------------------------------------------------------------------------
599-- Concurrency rate control.
601-- Our objective is to create more threads on demand if the consumer is running
602-- faster than us. As soon as we encounter a concurrent composition we create a
603-- push pull pair of threads. We use an SVar for communication between the
604-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
605-- The producer creates more threads if the SVar drains and becomes empty, that
606-- is the consumer is running faster.
608-- XXX Note 1: This mechanism can be problematic if the initial production
609-- latency is high, we may end up creating too many threads. So we need some
610-- way to monitor and use the latency as well. Having a limit on the dispatches
611-- (programmer controlled) may also help.
613-- TBD Note 2: We may want to run computations at the lower level of the
614-- composition tree serially even when they are composed using a parallel
615-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
616-- place of 'wAsync'. If we find that an SVar immediately above a computation
617-- gets drained empty we can switch to parallelizing the computation. For that
618-- we can use a state flag to fork the rest of the computation at any point of
619-- time inside the Monad bind operation if the consumer is running at a faster
622-- TBD Note 3: the binary operation ('parallel') composition allows us to
623-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
624-- chunksizes we will need to compose the parallel actions using a data
625-- constructor (A Free container) instead so that we can divide it in chunks of
626-- arbitrary size before dispatching. If the stream is composed of
627-- hierarchically composed grains of different sizes then we can always switch
628-- to a desired granularity depending on the consumer speed.
630-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
631-- into just the number of CPUs.
633-- | Join two computations on the currently running 'SVar' queue for concurrent
634-- execution. When we are using parallel composition, an SVar is passed around
635-- as a state variable. We try to schedule a new parallel computation on the
636-- SVar passed to us. The first time, when no SVar exists, a new SVar is
637-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
638-- already scheduled on the SVar is further evaluated. For example, when (a
639-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
640-- the current scheduler queue.
642-- The 'SVarStyle' required by the current composition context is passed as one
643-- of the parameters. If the scheduling and composition style of the new
644-- computation being scheduled is different than the style of the current SVar,
645-- then we create a new SVar and schedule it on that. The newly created SVar
646-- joins as one of the computations on the current SVar queue.
648-- Cases when we need to switch to a new SVar:
650-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
651-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
652-- because of the scheduling policy change.
653-- * if we 'adapt' a stream of type 'async' to a stream of type
654-- 'Parallel', we create a new SVar at the transitioning bind.
655-- * When the stream is switching from disjunctive composition to conjunctive
656-- composition and vice-versa we create a new SVar to isolate the scheduling
659forkSVarAsync :: (IsStream t, MonadAsync m)
660 => SVarStyle -> t m a -> t m a -> t m a
661forkSVarAsync style m1 m2 = mkStream $ \st yld sng stp -> do
663 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
664 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
665 _ -> error "illegal svar type"
666 foldStream st yld sng stp $ fromSVar sv
668 concurrently ma mb = mkStream $ \st yld sng stp -> do
669 runInIO <- captureMonadState
670 liftIO $ enqueue (fromJust $ streamVar st) $ (runInIO, mb)
671 foldStreamShared st yld sng stp ma
673{-# INLINE joinStreamVarAsync #-}
674joinStreamVarAsync :: (IsStream t, MonadAsync m)
675 => SVarStyle -> t m a -> t m a -> t m a
676joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
678 Just sv | svarStyle sv == style -> do
679 runInIO <- captureMonadState
680 liftIO $ enqueue sv $ (runInIO, toStream m2)
681 foldStreamShared st yld sng stp m1
682 _ -> foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
684------------------------------------------------------------------------------
685-- Semigroup and Monoid style compositions for parallel actions
686------------------------------------------------------------------------------
690-- | Merges two streams, both the streams may be evaluated concurrently,
691-- outputs from both are used as they arrive:
693-- >>> import Streamly.Prelude (async)
694-- >>> stream1 = Stream.fromEffect (delay 4)
695-- >>> stream2 = Stream.fromEffect (delay 2)
696-- >>> Stream.toList $ stream1 `async` stream2
701-- Multiple streams can be combined. With enough threads, all of them can be
702-- scheduled simultaneously:
704-- >>> stream3 = Stream.fromEffect (delay 1)
705-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
709-- With 2 threads, only two can be scheduled at a time, when one of those
710-- finishes, the third one gets scheduled:
712-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
716-- With a single thread, it becomes serial:
718-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
722-- Only streams are scheduled for async evaluation, how actions within a
723-- stream are evaluated depends on the stream type. If it is a concurrent
724-- stream they will be evaluated concurrently.
726-- In the following example, both the streams are scheduled for concurrent
727-- evaluation but each individual stream is evaluated serially:
729-- >>> stream1 = Stream.fromListM $ Prelude.map delay [3,3] -- SerialT IO Int
730-- >>> stream2 = Stream.fromListM $ Prelude.map delay [1,1] -- SerialT IO Int
731-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
735-- If total threads are 2, the third stream is scheduled only after one of the
736-- first two has finished:
738-- >>> stream3 = Stream.fromListM $ Prelude.map delay [2,2] -- SerialT IO Int
739-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
743-- Thus 'async' goes deep in first few streams rather than going wide in all
744-- streams. It prefers to evaluate the leftmost streams as much as possible.
745-- Because of this behavior, 'async' can be safely used to fold an infinite
746-- lazy container of streams.
748-- /Since: 0.2.0 ("Streamly")/
752async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
753async = joinStreamVarAsync AsyncVar
758{-# DEPRECATED (<|) "Please use 'async' instead." #-}
760(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
763-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
764-- huge difference in the performance of consM in IsStream instance even we
765-- have a SPECIALIZE in the instance.
767-- | XXX we can implement it more efficienty by directly implementing instead
768-- of combining streams using async.
769{-# INLINE consMAsync #-}
770{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
771consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
772consMAsync m r = fromStream $ K.fromEffect m `async` (toStream r)
774------------------------------------------------------------------------------
776------------------------------------------------------------------------------
778-- | For 'AsyncT' streams:
781-- (<>) = 'Streamly.Prelude.async'
782-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
785-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
786-- executed concurrently a la the 'async' combinator, producing results and
787-- side effects of iterations out of order:
790-- Stream.toList $ Stream.fromAsync $ do
791-- x <- Stream.fromList [2,1] -- foreach x in stream
792-- Stream.fromEffect $ delay x
798-- Nested monad binds behave like nested @for@ loops with nested iterations
799-- executed concurrently, a la the 'async' combinator:
802-- Stream.toList $ Stream.fromAsync $ do
803-- x <- Stream.fromList [1,2] -- foreach x in stream
804-- y <- Stream.fromList [2,4] -- foreach y in stream
805-- Stream.fromEffect $ delay (x + y)
813-- The behavior can be explained as follows. All the iterations corresponding
814-- to the element @1@ in the first stream constitute one output stream and all
815-- the iterations corresponding to @2@ constitute another output stream and
816-- these two output streams are merged using 'async'.
818-- /Since: 0.1.0 ("Streamly")/
821newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
822 deriving (MonadTrans)
824-- | A demand driven left biased parallely composing IO stream of elements of
825-- type @a@. See 'AsyncT' documentation for more details.
827-- /Since: 0.2.0 ("Streamly")/
830type Async = AsyncT IO
832-- | Fix the type of a polymorphic stream as 'AsyncT'.
834-- /Since: 0.1.0 ("Streamly")/
837fromAsync :: IsStream t => AsyncT m a -> t m a
840instance IsStream AsyncT where
846------------------------------------------------------------------------------
848------------------------------------------------------------------------------
850-- Monomorphically typed version of "async" for better performance of Semigroup
852{-# INLINE mappendAsync #-}
853{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
854mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
855mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
857instance MonadAsync m => Semigroup (AsyncT m a) where
860------------------------------------------------------------------------------
862------------------------------------------------------------------------------
864instance MonadAsync m => Monoid (AsyncT m a) where
868------------------------------------------------------------------------------
870------------------------------------------------------------------------------
872{-# INLINE apAsync #-}
873{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
874apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
875apAsync (AsyncT m1) (AsyncT m2) =
876 let f x1 = K.concatMapBy async (pure . x1) m2
877 in AsyncT $ K.concatMapBy async f m1
879instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
881 pure = AsyncT . K.fromPure
885------------------------------------------------------------------------------
887------------------------------------------------------------------------------
889-- GHC: if we change the implementation of bindWith with arguments in a
890-- different order we see a significant performance degradation (~2x).
891{-# INLINE bindAsync #-}
892{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
893bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
894bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
896-- GHC: if we specify arguments in the definition of (>>=) we see a significant
897-- performance degradation (~2x).
898instance MonadAsync m => Monad (AsyncT m) where
902------------------------------------------------------------------------------
904------------------------------------------------------------------------------
906MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
908------------------------------------------------------------------------------
910------------------------------------------------------------------------------
912-- | XXX we can implement it more efficienty by directly implementing instead
913-- of combining streams using wAsync.
914{-# INLINE consMWAsync #-}
915{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
916consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
917consMWAsync m r = fromStream $ K.fromEffect m `wAsync` (toStream r)
921-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
922-- singleton stream behavior. For multi-element streams, while 'async' is left
923-- biased i.e. it tries to evaluate the left side stream as much as possible,
924-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
925-- deep while 'wAsync' goes wide. However, outputs are always used as they
928-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
929-- starts behaving like 'wSerial'.
931-- >>> import Streamly.Prelude (wAsync)
932-- >>> stream1 = Stream.fromList [1,2,3]
933-- >>> stream2 = Stream.fromList [4,5,6]
934-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 1 $ stream1 `async` stream2
937-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
940-- With two threads available, and combining three streams:
942-- >>> stream3 = Stream.fromList [7,8,9]
943-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
944-- [1,2,3,4,5,6,7,8,9]
946-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
947-- [1,4,2,7,5,3,8,6,9]
949-- This operation cannot be used to fold an infinite lazy container of streams,
950-- because it schedules all the streams in a round robin manner.
952-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
953-- but the exact scheduling is slightly different in both cases.
957-- /Since: 0.2.0 ("Streamly")/
959-- Scheduling details:
961-- This is how the execution of the above example proceeds:
963-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
964-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
965-- queue is represented by the rightmost item.
966-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
967-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
968-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
969-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
970-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
971-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
972-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
973-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
974-- 4. Note that the scheduler queue expands by one more stream component in
975-- every pass because one more @<>@ is broken down into two components. At this
976-- point there are no more @<>@ operations to be broken down further and the
977-- queue has reached its maximum size. Now these streams are scheduled in
978-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
980-- As we see above, in a right associated expression composed with @<>@, only
981-- one @<>@ operation is broken down into two components in one execution,
982-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
983-- scheduler passes to expand the whole expression. By the time @n-th@
984-- component is added to the scheduler queue, the first component would have
985-- received @n@ scheduler passes.
988wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
989wAsync = joinStreamVarAsync WAsyncVar
991-- | For 'WAsyncT' streams:
994-- (<>) = 'Streamly.Prelude.wAsync'
995-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
998-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
999-- executed concurrently a la the 'wAsync' combinator, producing results and
1000-- side effects of iterations out of order:
1003-- Stream.toList $ Stream.fromWAsync $ do
1004-- x <- Stream.fromList [2,1] -- foreach x in stream
1005-- Stream.fromEffect $ delay x
1011-- Nested monad binds behave like nested @for@ loops with nested iterations
1012-- executed concurrently, a la the 'wAsync' combinator:
1015-- Stream.toList $ Stream.fromWAsync $ do
1016-- x <- Stream.fromList [1,2] -- foreach x in stream
1017-- y <- Stream.fromList [2,4] -- foreach y in stream
1018-- Stream.fromEffect $ delay (x + y)
1026-- The behavior can be explained as follows. All the iterations corresponding
1027-- to the element @1@ in the first stream constitute one 'WAsyncT' output
1028-- stream and all the iterations corresponding to @2@ constitute another
1029-- 'WAsyncT' output stream and these two output streams are merged using
1032-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
1033-- contrast to the depth wise scheduling behavior of 'AsyncT'.
1035-- /Since: 0.2.0 ("Streamly")/
1039-- XXX This documentation is redundant, need to be cleaned up/removed.
1042-- import qualified "Streamly.Prelude" as S
1043-- import Control.Concurrent
1045-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
1051-- For this example, we are using @maxThreads 1@ so that concurrent thread
1052-- scheduling does not affect the results and make them unpredictable. Let's
1053-- now take a more general example:
1056-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
1059-- [1,4,2,7,5,3,8,6,9]
1062-- This is how the execution of the above stream proceeds:
1064-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
1065-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
1066-- queue is represented by the rightmost item.
1067-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
1068-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
1069-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
1070-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
1071-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
1072-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
1073-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
1074-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
1075-- 4. Note that the scheduler queue expands by one more stream component in
1076-- every pass because one more @<>@ is broken down into two components. At this
1077-- point there are no more @<>@ operations to be broken down further and the
1078-- queue has reached its maximum size. Now these streams are scheduled in
1079-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
1081-- As we see above, in a right associated expression composed with @<>@, only
1082-- one @<>@ operation is broken down into two components in one execution,
1083-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
1084-- scheduler passes to expand the whole expression. By the time @n-th@
1085-- component is added to the scheduler queue, the first component would have
1086-- received @n@ scheduler passes.
1088-- Since all streams get interleaved, this operation is not suitable for
1089-- folding an infinite lazy container of infinite size streams. However, if
1090-- the streams are small, the streams on the left may get finished before more
1091-- streams are added to the scheduler queue from the right side of the
1092-- expression, so it may be possible to fold an infinite lazy container of
1093-- streams. For example, if the streams are of size @n@ then at most @n@
1094-- streams would be in the scheduler queue at a time.
1096-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
1097-- therefore the output of 'WAsyncT' even with a single thread of execution is
1098-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
1099-- its scheduling behavior.
1101-- Any exceptions generated by a constituent stream are propagated to the
1102-- output stream. The output and exceptions from a single stream are guaranteed
1103-- to arrive in the same order in the resulting stream as they were generated
1104-- in the input stream. However, the relative ordering of elements from
1105-- different streams in the resulting stream can vary depending on scheduling
1106-- and generation delays.
1108-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
1109-- concurrently using a round robin scheduling.
1112-- main = S.'drain' . S.'fromWAsync' $ do
1113-- n <- return 3 \<\> return 2 \<\> return 1
1115-- threadDelay (n * 1000000)
1116-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1119-- ThreadId 40: Delay 1
1120-- ThreadId 39: Delay 2
1121-- ThreadId 38: Delay 3
1124newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
1125 deriving (MonadTrans)
1127-- | A round robin parallely composing IO stream of elements of type @a@.
1128-- See 'WAsyncT' documentation for more details.
1130-- /Since: 0.2.0 ("Streamly")/
1133type WAsync = WAsyncT IO
1135-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1137-- /Since: 0.2.0 ("Streamly")/
1140fromWAsync :: IsStream t => WAsyncT m a -> t m a
1143instance IsStream WAsyncT where
1144 toStream = getWAsyncT
1145 fromStream = WAsyncT
1149------------------------------------------------------------------------------
1151------------------------------------------------------------------------------
1153{-# INLINE mappendWAsync #-}
1154{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1155mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
1156mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
1158instance MonadAsync m => Semigroup (WAsyncT m a) where
1159 (<>) = mappendWAsync
1161------------------------------------------------------------------------------
1163------------------------------------------------------------------------------
1165instance MonadAsync m => Monoid (WAsyncT m a) where
1169------------------------------------------------------------------------------
1171------------------------------------------------------------------------------
1173{-# INLINE apWAsync #-}
1174{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
1175apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
1176apWAsync (WAsyncT m1) (WAsyncT m2) =
1177 let f x1 = K.concatMapBy wAsync (pure . x1) m2
1178 in WAsyncT $ K.concatMapBy wAsync f m1
1180-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1181-- performance degradation (~2x).
1182instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1183 pure = WAsyncT . K.fromPure
1186------------------------------------------------------------------------------
1188------------------------------------------------------------------------------
1190-- GHC: if we change the implementation of bindWith with arguments in a
1191-- different order we see a significant performance degradation (~2x).
1192{-# INLINE bindWAsync #-}
1193{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1194bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1195bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
1197-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1198-- performance degradation (~2x).
1199instance MonadAsync m => Monad (WAsyncT m) where
1203------------------------------------------------------------------------------
1205------------------------------------------------------------------------------
1207MONAD_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