git.iko.soy/streamly
on a02f2d0 help sign in
streamly/src/Streamly/Internal/Data/Stream/Async.hs
1207 lines 47 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3#include "inline.hs"
5-- |
6-- Module : Streamly.Internal.Data.Stream.Async
7-- Copyright : (c) 2017 Composewell Technologies
8--
9-- License : BSD3
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
12-- Portability : GHC
13--
14-- To run examples in this module:
15--
16-- >>> import qualified Streamly.Prelude as Stream
17-- >>> import Control.Concurrent (threadDelay)
18-- >>> :{
19-- delay n = do
20-- threadDelay (n * 1000000) -- sleep for n seconds
21-- putStrLn (show n ++ " sec") -- print "n sec"
22-- return n -- IO Int
23-- :}
24--
25module Streamly.Internal.Data.Stream.Async
26 (
27 AsyncT
28 , Async
29 , fromAsync
30 , async
31 , (<|) --deprecated
32 , mkAsync
33 , mkAsyncK
35 , WAsyncT
36 , WAsync
37 , fromWAsync
38 , wAsync
39 )
40where
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(..))
57#endif
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"
79-- $setup
80-- >>> import qualified Streamly.Prelude as Stream
81-- >>> import Control.Concurrent (threadDelay)
82-- >>> :{
83-- delay n = do
84-- threadDelay (n * 1000000) -- sleep for n seconds
85-- putStrLn (show n ++ " sec") -- print "n sec"
86-- return n -- IO Int
87-- :}
88--
90-------------------------------------------------------------------------------
91-- Async
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
97-- large size
98{-# INLINE enqueueLIFO #-}
99enqueueLIFO ::
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
103 ringDoorBell sv
105data WorkerStatus = Continue | Suspend
107{-# INLINE workLoopLIFO #-}
108workLoopLIFO
109 :: (MonadIO m, MonadBaseControl IO m)
110 => IORef [(RunInIO m, Stream m a)]
111 -> State Stream m a
112 -> SVar Stream m a
113 -> Maybe WorkerInfo
114 -> m ()
115workLoopLIFO q st sv winfo = run
117 where
119 stop = liftIO $ sendStop sv winfo
120 run = do
121 work <- dequeue
122 case work of
123 Nothing -> stop
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.
128 --
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
134 res <- restoreM r
135 case res of
136 Continue -> run
137 Suspend -> stop
139 single a = do
140 res <- liftIO $ sendYield sv winfo (ChildYield a)
141 return $ if res then Continue else Suspend
143 yieldk a r = do
144 res <- liftIO $ sendYield sv winfo (ChildYield a)
145 if res
146 then foldStreamShared st yieldk single (return Continue) r
147 else do
148 runInIO <- captureMonadState
149 liftIO $ enqueueLIFO sv q (runInIO, r)
150 return Suspend
152 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
153 [] -> ([], Nothing)
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.
158--
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 #-}
162workLoopLIFOLimited
163 :: forall m a. (MonadIO m, MonadBaseControl IO m)
164 => IORef [(RunInIO m, Stream m a)]
165 -> State Stream m a
166 -> SVar Stream m a
167 -> Maybe WorkerInfo
168 -> m ()
169workLoopLIFOLimited q st sv winfo = run
171 where
173 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
174 stop = liftIO $ sendStop sv winfo
175 run = do
176 work <- dequeue
177 case work of
178 Nothing -> stop
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
187 if yieldLimitOk
188 then do
189 r <- liftIO $ runin $
190 foldStreamShared st yieldk single incrContinue m
191 res <- restoreM r
192 case res of
193 Continue -> run
194 Suspend -> stop
195 -- Avoid any side effects, undo the yield limit decrement if we
196 -- never yielded anything.
197 else liftIO $ do
198 enqueueLIFO sv q (RunInIO runin, m)
199 incrementYieldLimit sv
200 sendStop sv winfo
202 single a = do
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.
208 yieldk a r = do
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
213 else do
214 runInIO <- captureMonadState
215 liftIO $ incrementYieldLimit sv
216 liftIO $ enqueueLIFO sv q (runInIO, r)
217 return Suspend
219 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
220 [] -> ([], Nothing)
221 x : xs -> (xs, Just x)
223-------------------------------------------------------------------------------
224-- WAsync
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 #-}
232enqueueFIFO ::
233 SVar t m a
234 -> LinkedQueue (RunInIO m, t m a)
235 -> (RunInIO m, t m a)
236 -> IO ()
237enqueueFIFO sv q m = do
238 pushL q m
239 ringDoorBell sv
241-- XXX we can remove sv as it is derivable from st
243{-# INLINE workLoopFIFO #-}
244workLoopFIFO
245 :: (MonadIO m, MonadBaseControl IO m)
246 => LinkedQueue (RunInIO m, Stream m a)
247 -> State Stream m a
248 -> SVar Stream m a
249 -> Maybe WorkerInfo
250 -> m ()
251workLoopFIFO q st sv winfo = run
253 where
255 stop = liftIO $ sendStop sv winfo
256 run = do
257 work <- liftIO $ tryPopR q
258 case work of
259 Nothing -> stop
260 Just (RunInIO runin, m) -> do
261 r <- liftIO $ runin $
262 foldStreamShared st yieldk single (return Continue) m
263 res <- restoreM r
264 case res of
265 Continue -> run
266 Suspend -> stop
268 single a = do
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
275 -- yielding.
276 yieldk a r = do
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 #-}
283workLoopFIFOLimited
284 :: forall m a. (MonadIO m, MonadBaseControl IO m)
285 => LinkedQueue (RunInIO m, Stream m a)
286 -> State Stream m a
287 -> SVar Stream m a
288 -> Maybe WorkerInfo
289 -> m ()
290workLoopFIFOLimited q st sv winfo = run
292 where
294 stop = liftIO $ sendStop sv winfo
295 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
296 run = do
297 work <- liftIO $ tryPopR q
298 case work of
299 Nothing -> stop
300 Just (RunInIO runin, m) -> do
301 yieldLimitOk <- liftIO $ decrementYieldLimit sv
302 if yieldLimitOk
303 then do
304 r <- liftIO $ runin $
305 foldStreamShared st yieldk single incrContinue m
306 res <- restoreM r
307 case res of
308 Continue -> run
309 Suspend -> stop
310 else liftIO $ do
311 enqueueFIFO sv q (RunInIO runin, m)
312 incrementYieldLimit sv
313 sendStop sv winfo
315 single a = do
316 res <- liftIO $ sendYield sv winfo (ChildYield a)
317 return $ if res then Continue else Suspend
319 yieldk a r = do
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
325 then return Continue
326 else liftIO $ do
327 incrementYieldLimit sv
328 return Suspend
330-------------------------------------------------------------------------------
331-- SVar creation
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
344 active <- newIORef 0
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
354 tid <- myThreadId
356 let isWorkFinished _ = null <$> readIORef q
358 let isWorkFinishedLimited sv = do
359 yieldsDone <-
360 case remainingWork sv of
361 Just ref -> do
362 n <- readIORef ref
363 return (n <= 0)
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)]
373 -> State Stream m a
374 -> SVar Stream m a
375 -> Maybe WorkerInfo
376 -> m())
377 -> SVar Stream m a
378 getSVar sv readOutput postProc workDone wloop = SVar
379 { outputQueue = outQ
380 , outputQueueFromConsumer = undefined
381 , remainingWork = yl
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
397 , needDoorBell = wfw
398 , svarStyle = AsyncVar
399 , svarStopStyle = StopNone
400 , svarStopBy = undefined
401 , svarMrun = mrun
402 , workerCount = active
403 , accountThread = delThread sv
404 , workerStopMVar = undefined
405 , svarRef = Nothing
406 , svarInspectMode = getInspectMode st
407 , svarCreator = tid
408 , aheadWorkQueue = undefined
409 , outputHeap = undefined
410 , svarStats = stats
411 }
413 let sv =
414 case getStreamRate st of
415 Nothing ->
416 case getYieldLimit st of
417 Nothing -> getSVar sv readOutputQBounded
418 postProcessBounded
419 isWorkFinished
420 workLoopLIFO
421 Just _ -> getSVar sv readOutputQBounded
422 postProcessBounded
423 isWorkFinishedLimited
424 workLoopLIFOLimited
425 Just _ ->
426 case getYieldLimit st of
427 Nothing -> getSVar sv readOutputQPaced
428 postProcessPaced
429 isWorkFinished
430 workLoopLIFO
431 Just _ -> getSVar sv readOutputQPaced
432 postProcessPaced
433 isWorkFinishedLimited
434 workLoopLIFOLimited
435 in return sv
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
442 active <- newIORef 0
443 wfw <- newIORef False
444 running <- newIORef S.empty
445 q <- newQ
446 yl <- case getYieldLimit st of
447 Nothing -> return Nothing
448 Just x -> Just <$> newIORef x
449 rateInfo <- getYieldRateInfo st
451 stats <- newSVarStats
452 tid <- myThreadId
454 let isWorkFinished _ = nullQ q
455 let isWorkFinishedLimited sv = do
456 yieldsDone <-
457 case remainingWork sv of
458 Just ref -> do
459 n <- readIORef ref
460 return (n <= 0)
461 Nothing -> return False
462 qEmpty <- nullQ q
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)
470 -> State Stream m a
471 -> SVar Stream m a
472 -> Maybe WorkerInfo
473 -> m())
474 -> SVar Stream m a
475 getSVar sv readOutput postProc workDone wloop = SVar
476 { outputQueue = outQ
477 , outputQueueFromConsumer = undefined
478 , remainingWork = yl
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
494 , needDoorBell = wfw
495 , svarStyle = WAsyncVar
496 , svarStopStyle = StopNone
497 , svarStopBy = undefined
498 , svarMrun = mrun
499 , workerCount = active
500 , accountThread = delThread sv
501 , workerStopMVar = undefined
502 , svarRef = Nothing
503 , svarInspectMode = getInspectMode st
504 , svarCreator = tid
505 , aheadWorkQueue = undefined
506 , outputHeap = undefined
507 , svarStats = stats
508 }
510 let sv =
511 case getStreamRate st of
512 Nothing ->
513 case getYieldLimit st of
514 Nothing -> getSVar sv readOutputQBounded
515 postProcessBounded
516 isWorkFinished
517 workLoopFIFO
518 Just _ -> getSVar sv readOutputQBounded
519 postProcessBounded
520 isWorkFinishedLimited
521 workLoopFIFOLimited
522 Just _ ->
523 case getYieldLimit st of
524 Nothing -> getSVar sv readOutputQPaced
525 postProcessPaced
526 isWorkFinished
527 workLoopFIFO
528 Just _ -> getSVar sv readOutputQPaced
529 postProcessPaced
530 isWorkFinishedLimited
531 workLoopFIFOLimited
532 in return sv
534{-# INLINABLE newAsyncVar #-}
535newAsyncVar :: MonadAsync m
536 => State Stream m a -> Stream m a -> m (SVar Stream m a)
537newAsyncVar st m = do
538 mrun <- captureMonadState
539 sv <- liftIO $ getLifoSVar st mrun
540 sendFirstWorker sv m
542-- | Generate a stream asynchronously to keep it buffered, lazily consume
543-- from the buffer.
544--
545-- /Pre-release/
546--
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
556 where
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
563 r <- step1 gst st
564 return $ case r of
565 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
566 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
567 D.Stop -> D.Stop
569--
570-- This is slightly faster than the CPS version above
571--
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.
577--
578-- /Since: 0.2.0 (Streamly)/
579--
580-- @since 0.8.0
581--
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
593 sendFirstWorker sv m
595------------------------------------------------------------------------------
596-- Running streams concurrently
597------------------------------------------------------------------------------
599-- Concurrency rate control.
600--
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.
607--
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.
612--
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
620-- speed.
621--
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.
629--
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.
641--
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.
647--
648-- Cases when we need to switch to a new SVar:
649--
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
657-- of the two.
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
662 sv <- case style of
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
667 where
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 ->
677 case streamVar st of
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------------------------------------------------------------------------------
688infixr 6 `async`
690-- | Merges two streams, both the streams may be evaluated concurrently,
691-- outputs from both are used as they arrive:
692--
693-- >>> import Streamly.Prelude (async)
694-- >>> stream1 = Stream.fromEffect (delay 4)
695-- >>> stream2 = Stream.fromEffect (delay 2)
696-- >>> Stream.toList $ stream1 `async` stream2
697-- 2 sec
698-- 4 sec
699-- [2,4]
700--
701-- Multiple streams can be combined. With enough threads, all of them can be
702-- scheduled simultaneously:
703--
704-- >>> stream3 = Stream.fromEffect (delay 1)
705-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
706-- ...
707-- [1,2,4]
708--
709-- With 2 threads, only two can be scheduled at a time, when one of those
710-- finishes, the third one gets scheduled:
711--
712-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
713-- ...
714-- [2,1,4]
715--
716-- With a single thread, it becomes serial:
717--
718-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
719-- ...
720-- [4,2,1]
721--
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.
725--
726-- In the following example, both the streams are scheduled for concurrent
727-- evaluation but each individual stream is evaluated serially:
728--
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]
732-- ...
733-- [1,1,3,3]
734--
735-- If total threads are 2, the third stream is scheduled only after one of the
736-- first two has finished:
737--
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]
740-- ...
741-- [1,1,3,2,3,2]
742--
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.
747--
748-- /Since: 0.2.0 ("Streamly")/
749--
750-- @since 0.8.0
751{-# INLINE async #-}
752async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
753async = joinStreamVarAsync AsyncVar
755-- | Same as 'async'.
756--
757-- @since 0.1.0
758{-# DEPRECATED (<|) "Please use 'async' instead." #-}
759{-# INLINE (<|) #-}
760(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
761(<|) = async
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.
766--
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------------------------------------------------------------------------------
775-- AsyncT
776------------------------------------------------------------------------------
778-- | For 'AsyncT' streams:
779--
780-- @
781-- (<>) = 'Streamly.Prelude.async'
782-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
783-- @
784--
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:
788--
789-- >>> :{
790-- Stream.toList $ Stream.fromAsync $ do
791-- x <- Stream.fromList [2,1] -- foreach x in stream
792-- Stream.fromEffect $ delay x
793-- :}
794-- 1 sec
795-- 2 sec
796-- [1,2]
797--
798-- Nested monad binds behave like nested @for@ loops with nested iterations
799-- executed concurrently, a la the 'async' combinator:
800--
801-- >>> :{
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)
806-- :}
807-- 3 sec
808-- 4 sec
809-- 5 sec
810-- 6 sec
811-- [3,4,5,6]
812--
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'.
817--
818-- /Since: 0.1.0 ("Streamly")/
819--
820-- @since 0.8.0
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.
826--
827-- /Since: 0.2.0 ("Streamly")/
828--
829-- @since 0.8.0
830type Async = AsyncT IO
832-- | Fix the type of a polymorphic stream as 'AsyncT'.
833--
834-- /Since: 0.1.0 ("Streamly")/
835--
836-- @since 0.8.0
837fromAsync :: IsStream t => AsyncT m a -> t m a
838fromAsync = adapt
840instance IsStream AsyncT where
841 toStream = getAsyncT
842 fromStream = AsyncT
843 consM = consMAsync
844 (|:) = consMAsync
846------------------------------------------------------------------------------
847-- Semigroup
848------------------------------------------------------------------------------
850-- Monomorphically typed version of "async" for better performance of Semigroup
851-- instance.
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
858 (<>) = mappendAsync
860------------------------------------------------------------------------------
861-- Monoid
862------------------------------------------------------------------------------
864instance MonadAsync m => Monoid (AsyncT m a) where
865 mempty = K.nil
866 mappend = (<>)
868------------------------------------------------------------------------------
869-- Applicative
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
880 {-# INLINE pure #-}
881 pure = AsyncT . K.fromPure
882 {-# INLINE (<*>) #-}
883 (<*>) = apAsync
885------------------------------------------------------------------------------
886-- Monad
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
899 return = pure
900 (>>=) = bindAsync
902------------------------------------------------------------------------------
903-- Other instances
904------------------------------------------------------------------------------
906MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
908------------------------------------------------------------------------------
909-- WAsyncT
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)
919infixr 6 `wAsync`
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
926-- arrive.
927--
928-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
929-- starts behaving like 'wSerial'.
930--
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
935-- [1,2,3,4,5,6]
936--
937-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
938-- [1,4,2,5,3,6]
939--
940-- With two threads available, and combining three streams:
941--
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]
945--
946-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
947-- [1,4,2,7,5,3,8,6,9]
948--
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.
951--
952-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
953-- but the exact scheduling is slightly different in both cases.
954--
955-- @since 0.8.0
956--
957-- /Since: 0.2.0 ("Streamly")/
959-- Scheduling details:
960--
961-- This is how the execution of the above example proceeds:
962--
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]@.
979--
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.
986--
987{-# INLINE wAsync #-}
988wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
989wAsync = joinStreamVarAsync WAsyncVar
991-- | For 'WAsyncT' streams:
992--
993-- @
994-- (<>) = 'Streamly.Prelude.wAsync'
995-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
996-- @
997--
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:
1002-- >>> :{
1003-- Stream.toList $ Stream.fromWAsync $ do
1004-- x <- Stream.fromList [2,1] -- foreach x in stream
1005-- Stream.fromEffect $ delay x
1006-- :}
1007-- 1 sec
1008-- 2 sec
1009-- [1,2]
1011-- Nested monad binds behave like nested @for@ loops with nested iterations
1012-- executed concurrently, a la the 'wAsync' combinator:
1014-- >>> :{
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)
1019-- :}
1020-- 3 sec
1021-- 4 sec
1022-- 5 sec
1023-- 6 sec
1024-- [3,4,5,6]
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
1030-- 'wAsync'.
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")/
1037-- @since 0.8.0
1039-- XXX This documentation is redundant, need to be cleaned up/removed.
1041-- @
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
1046-- @
1047-- @
1048-- [1,3,2,4]
1049-- @
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:
1055-- @
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
1057-- @
1058-- @
1059-- [1,4,2,7,5,3,8,6,9]
1060-- @
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.
1111-- @
1112-- main = S.'drain' . S.'fromWAsync' $ do
1113-- n <- return 3 \<\> return 2 \<\> return 1
1114-- S.fromEffect $ do
1115-- threadDelay (n * 1000000)
1116-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1117-- @
1118-- @
1119-- ThreadId 40: Delay 1
1120-- ThreadId 39: Delay 2
1121-- ThreadId 38: Delay 3
1122-- @
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")/
1132-- @since 0.8.0
1133type WAsync = WAsyncT IO
1135-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1137-- /Since: 0.2.0 ("Streamly")/
1139-- @since 0.8.0
1140fromWAsync :: IsStream t => WAsyncT m a -> t m a
1141fromWAsync = adapt
1143instance IsStream WAsyncT where
1144 toStream = getWAsyncT
1145 fromStream = WAsyncT
1146 consM = consMWAsync
1147 (|:) = consMWAsync
1149------------------------------------------------------------------------------
1150-- Semigroup
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------------------------------------------------------------------------------
1162-- Monoid
1163------------------------------------------------------------------------------
1165instance MonadAsync m => Monoid (WAsyncT m a) where
1166 mempty = K.nil
1167 mappend = (<>)
1169------------------------------------------------------------------------------
1170-- Applicative
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
1184 (<*>) = apWAsync
1186------------------------------------------------------------------------------
1187-- Monad
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
1200 return = pure
1201 (>>=) = bindWAsync
1203------------------------------------------------------------------------------
1204-- Other instances
1205------------------------------------------------------------------------------
1207MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)