git.iko.soy/streamly
on 03218ad help sign in
streamly/src/Streamly/Internal/Data/Stream/Async.hs
1175 lines 46 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)
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.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"
74-- $setup
75-- >>> import Control.Concurrent (threadDelay)
76-- >>> :{
77-- delay n = do
78-- threadDelay (n * 1000000) -- sleep for n seconds
79-- putStrLn (show n ++ " sec") -- print "n sec"
80-- return n -- IO Int
81-- :}
83-------------------------------------------------------------------------------
84-- Async
85-------------------------------------------------------------------------------
87data WorkerStatus = Continue | Suspend
89{-# INLINE workLoopLIFO #-}
90workLoopLIFO
91 :: (MonadIO m, MonadBaseControl IO m)
92 => IORef [(RunInIO m, Stream m a)]
93 -> State Stream m a
94 -> SVar Stream m a
95 -> Maybe WorkerInfo
96 -> m ()
97workLoopLIFO q st sv winfo = run
99 where
101 stop = liftIO $ sendStop sv winfo
102 run = do
103 work <- dequeue
104 case work of
105 Nothing -> stop
106 Just (RunInIO runin, m) -> do
107 -- XXX when we finish we need to send the monadic state back to
108 -- the parent so that the state can be merged back. We capture
109 -- and return the state in the stop continuation.
110 --
111 -- Instead of using the run function we can just restore the
112 -- monad state here. That way it can work easily for
113 -- distributed case as well.
114 r <- liftIO $ runin $
115 foldStreamShared st yieldk single (return Continue) m
116 res <- restoreM r
117 case res of
118 Continue -> run
119 Suspend -> stop
121 single a = do
122 res <- liftIO $ sendYield sv winfo (ChildYield a)
123 return $ if res then Continue else Suspend
125 yieldk a r = do
126 res <- liftIO $ sendYield sv winfo (ChildYield a)
127 if res
128 then foldStreamShared st yieldk single (return Continue) r
129 else do
130 runInIO <- captureMonadState
131 liftIO $ enqueueLIFO sv q (runInIO, r)
132 return Suspend
134 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
135 [] -> ([], Nothing)
136 x : xs -> (xs, Just x)
138-- We duplicate workLoop for yield limit and no limit cases because it has
139-- around 40% performance overhead in the worst case.
140--
141-- XXX we can pass yinfo directly as an argument here so that we do not have to
142-- make a check every time.
143{-# INLINE workLoopLIFOLimited #-}
144workLoopLIFOLimited
145 :: forall m a. (MonadIO m, MonadBaseControl IO m)
146 => IORef [(RunInIO m, Stream m a)]
147 -> State Stream m a
148 -> SVar Stream m a
149 -> Maybe WorkerInfo
150 -> m ()
151workLoopLIFOLimited q st sv winfo = run
153 where
155 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
156 stop = liftIO $ sendStop sv winfo
157 run = do
158 work <- dequeue
159 case work of
160 Nothing -> stop
161 Just (RunInIO runin, m) -> do
162 -- XXX This is just a best effort minimization of concurrency
163 -- to the yield limit. If the stream is made of concurrent
164 -- streams we do not reserve the yield limit in the constituent
165 -- streams before executing the action. This can be done
166 -- though, by sharing the yield limit ref with downstream
167 -- actions via state passing. Just a todo.
168 yieldLimitOk <- liftIO $ decrementYieldLimit sv
169 if yieldLimitOk
170 then do
171 r <- liftIO $ runin $
172 foldStreamShared st yieldk single incrContinue m
173 res <- restoreM r
174 case res of
175 Continue -> run
176 Suspend -> stop
177 -- Avoid any side effects, undo the yield limit decrement if we
178 -- never yielded anything.
179 else liftIO $ do
180 enqueueLIFO sv q (RunInIO runin, m)
181 incrementYieldLimit sv
182 sendStop sv winfo
184 single a = do
185 res <- liftIO $ sendYield sv winfo (ChildYield a)
186 return $ if res then Continue else Suspend
188 -- XXX can we pass on the yield limit downstream to limit the concurrency
189 -- of constituent streams.
190 yieldk a r = do
191 res <- liftIO $ sendYield sv winfo (ChildYield a)
192 yieldLimitOk <- liftIO $ decrementYieldLimit sv
193 if res && yieldLimitOk
194 then foldStreamShared st yieldk single incrContinue r
195 else do
196 runInIO <- captureMonadState
197 liftIO $ incrementYieldLimit sv
198 liftIO $ enqueueLIFO sv q (runInIO, r)
199 return Suspend
201 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
202 [] -> ([], Nothing)
203 x : xs -> (xs, Just x)
205-------------------------------------------------------------------------------
206-- WAsync
207-------------------------------------------------------------------------------
209-- XXX we can remove sv as it is derivable from st
211{-# INLINE workLoopFIFO #-}
212workLoopFIFO
213 :: (MonadIO m, MonadBaseControl IO m)
214 => LinkedQueue (RunInIO m, Stream m a)
215 -> State Stream m a
216 -> SVar Stream m a
217 -> Maybe WorkerInfo
218 -> m ()
219workLoopFIFO q st sv winfo = run
221 where
223 stop = liftIO $ sendStop sv winfo
224 run = do
225 work <- liftIO $ tryPopR q
226 case work of
227 Nothing -> stop
228 Just (RunInIO runin, m) -> do
229 r <- liftIO $ runin $
230 foldStreamShared st yieldk single (return Continue) m
231 res <- restoreM r
232 case res of
233 Continue -> run
234 Suspend -> stop
236 single a = do
237 res <- liftIO $ sendYield sv winfo (ChildYield a)
238 return $ if res then Continue else Suspend
240 -- XXX in general we would like to yield "n" elements from a single stream
241 -- before moving on to the next. Single element granularity could be too
242 -- expensive in certain cases. Similarly, we can use time limit for
243 -- yielding.
244 yieldk a r = do
245 res <- liftIO $ sendYield sv winfo (ChildYield a)
246 runInIO <- captureMonadState
247 liftIO $ enqueueFIFO sv q (runInIO, r)
248 return $ if res then Continue else Suspend
250{-# INLINE workLoopFIFOLimited #-}
251workLoopFIFOLimited
252 :: forall m a. (MonadIO m, MonadBaseControl IO m)
253 => LinkedQueue (RunInIO m, Stream m a)
254 -> State Stream m a
255 -> SVar Stream m a
256 -> Maybe WorkerInfo
257 -> m ()
258workLoopFIFOLimited q st sv winfo = run
260 where
262 stop = liftIO $ sendStop sv winfo
263 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
264 run = do
265 work <- liftIO $ tryPopR q
266 case work of
267 Nothing -> stop
268 Just (RunInIO runin, m) -> do
269 yieldLimitOk <- liftIO $ decrementYieldLimit sv
270 if yieldLimitOk
271 then do
272 r <- liftIO $ runin $
273 foldStreamShared st yieldk single incrContinue m
274 res <- restoreM r
275 case res of
276 Continue -> run
277 Suspend -> stop
278 else liftIO $ do
279 enqueueFIFO sv q (RunInIO runin, m)
280 incrementYieldLimit sv
281 sendStop sv winfo
283 single a = do
284 res <- liftIO $ sendYield sv winfo (ChildYield a)
285 return $ if res then Continue else Suspend
287 yieldk a r = do
288 res <- liftIO $ sendYield sv winfo (ChildYield a)
289 runInIO <- captureMonadState
290 liftIO $ enqueueFIFO sv q (runInIO, r)
291 yieldLimitOk <- liftIO $ decrementYieldLimit sv
292 if res && yieldLimitOk
293 then return Continue
294 else liftIO $ do
295 incrementYieldLimit sv
296 return Suspend
298-------------------------------------------------------------------------------
299-- SVar creation
300-- This code belongs in SVar.hs but is kept here for perf reasons
301-------------------------------------------------------------------------------
303-- XXX we have this function in this file because passing runStreamLIFO as a
304-- function argument to this function results in a perf degradation of more
305-- than 10%. Need to investigate what the root cause is.
306-- Interestingly, the same thing does not make any difference for Ahead.
307getLifoSVar :: forall m a. MonadAsync m
308 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
309getLifoSVar st mrun = do
310 outQ <- newIORef ([], 0)
311 outQMv <- newEmptyMVar
312 active <- newIORef 0
313 wfw <- newIORef False
314 running <- newIORef S.empty
315 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
316 yl <- case getYieldLimit st of
317 Nothing -> return Nothing
318 Just x -> Just <$> newIORef x
319 rateInfo <- getYieldRateInfo st
321 stats <- newSVarStats
322 tid <- myThreadId
324 let isWorkFinished _ = null <$> readIORef q
326 let isWorkFinishedLimited sv = do
327 yieldsDone <-
328 case remainingWork sv of
329 Just ref -> do
330 n <- readIORef ref
331 return (n <= 0)
332 Nothing -> return False
333 qEmpty <- null <$> readIORef q
334 return $ qEmpty || yieldsDone
336 let getSVar :: SVar Stream m a
337 -> (SVar Stream m a -> m [ChildEvent a])
338 -> (SVar Stream m a -> m Bool)
339 -> (SVar Stream m a -> IO Bool)
340 -> (IORef [(RunInIO m, Stream m a)]
341 -> State Stream m a
342 -> SVar Stream m a
343 -> Maybe WorkerInfo
344 -> m())
345 -> SVar Stream m a
346 getSVar sv readOutput postProc workDone wloop = SVar
347 { outputQueue = outQ
348 , outputQueueFromConsumer = undefined
349 , remainingWork = yl
350 , maxBufferLimit = getMaxBuffer st
351 , pushBufferSpace = undefined
352 , pushBufferPolicy = undefined
353 , pushBufferMVar = undefined
354 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
355 , yieldRateInfo = rateInfo
356 , outputDoorBell = outQMv
357 , outputDoorBellFromConsumer = undefined
358 , readOutputQ = readOutput sv
359 , postProcess = postProc sv
360 , workerThreads = running
361 , workLoop = wloop q st{streamVar = Just sv} sv
362 , enqueue = enqueueLIFO sv q
363 , isWorkDone = workDone sv
364 , isQueueDone = workDone sv
365 , needDoorBell = wfw
366 , svarStyle = AsyncVar
367 , svarStopStyle = StopNone
368 , svarStopBy = undefined
369 , svarMrun = mrun
370 , workerCount = active
371 , accountThread = delThread sv
372 , workerStopMVar = undefined
373 , svarRef = Nothing
374 , svarInspectMode = getInspectMode st
375 , svarCreator = tid
376 , aheadWorkQueue = undefined
377 , outputHeap = undefined
378 , svarStats = stats
379 }
381 let sv =
382 case getStreamRate st of
383 Nothing ->
384 case getYieldLimit st of
385 Nothing -> getSVar sv readOutputQBounded
386 postProcessBounded
387 isWorkFinished
388 workLoopLIFO
389 Just _ -> getSVar sv readOutputQBounded
390 postProcessBounded
391 isWorkFinishedLimited
392 workLoopLIFOLimited
393 Just _ ->
394 case getYieldLimit st of
395 Nothing -> getSVar sv readOutputQPaced
396 postProcessPaced
397 isWorkFinished
398 workLoopLIFO
399 Just _ -> getSVar sv readOutputQPaced
400 postProcessPaced
401 isWorkFinishedLimited
402 workLoopLIFOLimited
403 in return sv
405getFifoSVar :: forall m a. MonadAsync m
406 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
407getFifoSVar st mrun = do
408 outQ <- newIORef ([], 0)
409 outQMv <- newEmptyMVar
410 active <- newIORef 0
411 wfw <- newIORef False
412 running <- newIORef S.empty
413 q <- newQ
414 yl <- case getYieldLimit st of
415 Nothing -> return Nothing
416 Just x -> Just <$> newIORef x
417 rateInfo <- getYieldRateInfo st
419 stats <- newSVarStats
420 tid <- myThreadId
422 let isWorkFinished _ = nullQ q
423 let isWorkFinishedLimited sv = do
424 yieldsDone <-
425 case remainingWork sv of
426 Just ref -> do
427 n <- readIORef ref
428 return (n <= 0)
429 Nothing -> return False
430 qEmpty <- nullQ q
431 return $ qEmpty || yieldsDone
433 let getSVar :: SVar Stream m a
434 -> (SVar Stream m a -> m [ChildEvent a])
435 -> (SVar Stream m a -> m Bool)
436 -> (SVar Stream m a -> IO Bool)
437 -> (LinkedQueue (RunInIO m, Stream m a)
438 -> State Stream m a
439 -> SVar Stream m a
440 -> Maybe WorkerInfo
441 -> m())
442 -> SVar Stream m a
443 getSVar sv readOutput postProc workDone wloop = SVar
444 { outputQueue = outQ
445 , outputQueueFromConsumer = undefined
446 , remainingWork = yl
447 , maxBufferLimit = getMaxBuffer st
448 , pushBufferSpace = undefined
449 , pushBufferPolicy = undefined
450 , pushBufferMVar = undefined
451 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
452 , yieldRateInfo = rateInfo
453 , outputDoorBell = outQMv
454 , outputDoorBellFromConsumer = undefined
455 , readOutputQ = readOutput sv
456 , postProcess = postProc sv
457 , workerThreads = running
458 , workLoop = wloop q st{streamVar = Just sv} sv
459 , enqueue = enqueueFIFO sv q
460 , isWorkDone = workDone sv
461 , isQueueDone = workDone sv
462 , needDoorBell = wfw
463 , svarStyle = WAsyncVar
464 , svarStopStyle = StopNone
465 , svarStopBy = undefined
466 , svarMrun = mrun
467 , workerCount = active
468 , accountThread = delThread sv
469 , workerStopMVar = undefined
470 , svarRef = Nothing
471 , svarInspectMode = getInspectMode st
472 , svarCreator = tid
473 , aheadWorkQueue = undefined
474 , outputHeap = undefined
475 , svarStats = stats
476 }
478 let sv =
479 case getStreamRate st of
480 Nothing ->
481 case getYieldLimit st of
482 Nothing -> getSVar sv readOutputQBounded
483 postProcessBounded
484 isWorkFinished
485 workLoopFIFO
486 Just _ -> getSVar sv readOutputQBounded
487 postProcessBounded
488 isWorkFinishedLimited
489 workLoopFIFOLimited
490 Just _ ->
491 case getYieldLimit st of
492 Nothing -> getSVar sv readOutputQPaced
493 postProcessPaced
494 isWorkFinished
495 workLoopFIFO
496 Just _ -> getSVar sv readOutputQPaced
497 postProcessPaced
498 isWorkFinishedLimited
499 workLoopFIFOLimited
500 in return sv
502{-# INLINABLE newAsyncVar #-}
503newAsyncVar :: MonadAsync m
504 => State Stream m a -> Stream m a -> m (SVar Stream m a)
505newAsyncVar st m = do
506 mrun <- captureMonadState
507 sv <- liftIO $ getLifoSVar st mrun
508 sendFirstWorker sv m
510-- | Generate a stream asynchronously to keep it buffered, lazily consume
511-- from the buffer.
512--
513-- /Pre-release/
514--
515{-# INLINABLE mkAsyncK #-}
516mkAsyncK :: (IsStream t, MonadAsync m) => t m a -> t m a
517mkAsyncK m = mkStream $ \st yld sng stp -> do
518 sv <- newAsyncVar (adaptState st) (toStream m)
519 foldStream st yld sng stp $ fromSVar sv
521{-# INLINE_NORMAL mkAsyncD #-}
522mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
523mkAsyncD m = D.Stream step Nothing
524 where
526 step gst Nothing = do
527 sv <- newAsyncVar gst (D.fromStreamD m)
528 return $ D.Skip $ Just $ fromSVarD sv
530 step gst (Just (D.UnStream step1 st)) = do
531 r <- step1 gst st
532 return $ case r of
533 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
534 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
535 D.Stop -> D.Stop
537--
538-- This is slightly faster than the CPS version above
539--
540-- | Make the stream producer and consumer run concurrently by introducing a
541-- buffer between them. The producer thread evaluates the input stream until
542-- the buffer fills, it terminates if the buffer is full and a worker thread is
543-- kicked off again to evaluate the remaining stream when there is space in the
544-- buffer. The consumer consumes the stream lazily from the buffer.
545--
546-- /Since: 0.2.0 (Streamly)/
547--
548-- @since 0.8.0
549--
550{-# INLINE_NORMAL mkAsync #-}
551mkAsync :: (K.IsStream t, MonadAsync m) => t m a -> t m a
552mkAsync = D.fromStreamD . mkAsyncD . D.toStreamD
554-- | Create a new SVar and enqueue one stream computation on it.
555{-# INLINABLE newWAsyncVar #-}
556newWAsyncVar :: MonadAsync m
557 => State Stream m a -> Stream m a -> m (SVar Stream m a)
558newWAsyncVar st m = do
559 mrun <- captureMonadState
560 sv <- liftIO $ getFifoSVar st mrun
561 sendFirstWorker sv m
563------------------------------------------------------------------------------
564-- Running streams concurrently
565------------------------------------------------------------------------------
567-- Concurrency rate control.
568--
569-- Our objective is to create more threads on demand if the consumer is running
570-- faster than us. As soon as we encounter a concurrent composition we create a
571-- push pull pair of threads. We use an SVar for communication between the
572-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
573-- The producer creates more threads if the SVar drains and becomes empty, that
574-- is the consumer is running faster.
575--
576-- XXX Note 1: This mechanism can be problematic if the initial production
577-- latency is high, we may end up creating too many threads. So we need some
578-- way to monitor and use the latency as well. Having a limit on the dispatches
579-- (programmer controlled) may also help.
580--
581-- TBD Note 2: We may want to run computations at the lower level of the
582-- composition tree serially even when they are composed using a parallel
583-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
584-- place of 'wAsync'. If we find that an SVar immediately above a computation
585-- gets drained empty we can switch to parallelizing the computation. For that
586-- we can use a state flag to fork the rest of the computation at any point of
587-- time inside the Monad bind operation if the consumer is running at a faster
588-- speed.
589--
590-- TBD Note 3: the binary operation ('parallel') composition allows us to
591-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
592-- chunksizes we will need to compose the parallel actions using a data
593-- constructor (A Free container) instead so that we can divide it in chunks of
594-- arbitrary size before dispatching. If the stream is composed of
595-- hierarchically composed grains of different sizes then we can always switch
596-- to a desired granularity depending on the consumer speed.
597--
598-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
599-- into just the number of CPUs.
601-- | Join two computations on the currently running 'SVar' queue for concurrent
602-- execution. When we are using parallel composition, an SVar is passed around
603-- as a state variable. We try to schedule a new parallel computation on the
604-- SVar passed to us. The first time, when no SVar exists, a new SVar is
605-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
606-- already scheduled on the SVar is further evaluated. For example, when (a
607-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
608-- the current scheduler queue.
609--
610-- The 'SVarStyle' required by the current composition context is passed as one
611-- of the parameters. If the scheduling and composition style of the new
612-- computation being scheduled is different than the style of the current SVar,
613-- then we create a new SVar and schedule it on that. The newly created SVar
614-- joins as one of the computations on the current SVar queue.
615--
616-- Cases when we need to switch to a new SVar:
617--
618-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
619-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
620-- because of the scheduling policy change.
621-- * if we 'adapt' a stream of type 'async' to a stream of type
622-- 'Parallel', we create a new SVar at the transitioning bind.
623-- * When the stream is switching from disjunctive composition to conjunctive
624-- composition and vice-versa we create a new SVar to isolate the scheduling
625-- of the two.
627forkSVarAsync :: (IsStream t, MonadAsync m)
628 => SVarStyle -> t m a -> t m a -> t m a
629forkSVarAsync style m1 m2 = mkStream $ \st yld sng stp -> do
630 sv <- case style of
631 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
632 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
633 _ -> error "illegal svar type"
634 foldStream st yld sng stp $ fromSVar sv
635 where
636 concurrently ma mb = mkStream $ \st yld sng stp -> do
637 runInIO <- captureMonadState
638 liftIO $ enqueue (fromJust $ streamVar st) $ (runInIO, mb)
639 foldStreamShared st yld sng stp ma
641{-# INLINE joinStreamVarAsync #-}
642joinStreamVarAsync :: (IsStream t, MonadAsync m)
643 => SVarStyle -> t m a -> t m a -> t m a
644joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
645 case streamVar st of
646 Just sv | svarStyle sv == style -> do
647 runInIO <- captureMonadState
648 liftIO $ enqueue sv $ (runInIO, toStream m2)
649 foldStreamShared st yld sng stp m1
650 _ -> foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
652------------------------------------------------------------------------------
653-- Semigroup and Monoid style compositions for parallel actions
654------------------------------------------------------------------------------
656infixr 6 `async`
658-- | Merges two streams, both the streams may be evaluated concurrently,
659-- outputs from both are used as they arrive:
660--
661-- >>> import Streamly.Prelude (async)
662-- >>> stream1 = Stream.fromEffect (delay 4)
663-- >>> stream2 = Stream.fromEffect (delay 2)
664-- >>> Stream.toList $ stream1 `async` stream2
665-- 2 sec
666-- 4 sec
667-- [2,4]
668--
669-- Multiple streams can be combined. With enough threads, all of them can be
670-- scheduled simultaneously:
671--
672-- >>> stream3 = Stream.fromEffect (delay 1)
673-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
674-- ...
675-- [1,2,4]
676--
677-- With 2 threads, only two can be scheduled at a time, when one of those
678-- finishes, the third one gets scheduled:
679--
680-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
681-- ...
682-- [2,1,4]
683--
684-- With a single thread, it becomes serial:
685--
686-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
687-- ...
688-- [4,2,1]
689--
690-- Only streams are scheduled for async evaluation, how actions within a
691-- stream are evaluated depends on the stream type. If it is a concurrent
692-- stream they will be evaluated concurrently.
693--
694-- In the following example, both the streams are scheduled for concurrent
695-- evaluation but each individual stream is evaluated serially:
696--
697-- >>> stream1 = Stream.fromListM $ Prelude.map delay [3,3] -- SerialT IO Int
698-- >>> stream2 = Stream.fromListM $ Prelude.map delay [1,1] -- SerialT IO Int
699-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
700-- ...
701-- [1,1,3,3]
702--
703-- If total threads are 2, the third stream is scheduled only after one of the
704-- first two has finished:
705--
706-- >>> stream3 = Stream.fromListM $ Prelude.map delay [2,2] -- SerialT IO Int
707-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
708-- ...
709-- [1,1,3,2,3,2]
710--
711-- Thus 'async' goes deep in first few streams rather than going wide in all
712-- streams. It prefers to evaluate the leftmost streams as much as possible.
713-- Because of this behavior, 'async' can be safely used to fold an infinite
714-- lazy container of streams.
715--
716-- /Since: 0.2.0 ("Streamly")/
717--
718-- @since 0.8.0
719{-# INLINE async #-}
720async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
721async = joinStreamVarAsync AsyncVar
723-- | Same as 'async'.
724--
725-- @since 0.1.0
726{-# DEPRECATED (<|) "Please use 'async' instead." #-}
727{-# INLINE (<|) #-}
728(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
729(<|) = async
731-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
732-- huge difference in the performance of consM in IsStream instance even we
733-- have a SPECIALIZE in the instance.
734--
735-- | XXX we can implement it more efficienty by directly implementing instead
736-- of combining streams using async.
737{-# INLINE consMAsync #-}
738{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
739consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
740consMAsync m r = fromStream $ K.fromEffect m `async` (toStream r)
742------------------------------------------------------------------------------
743-- AsyncT
744------------------------------------------------------------------------------
746-- | For 'AsyncT' streams:
747--
748-- @
749-- (<>) = 'Streamly.Prelude.async'
750-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.async'
751-- @
752--
753-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
754-- executed concurrently a la the 'async' combinator, producing results and
755-- side effects of iterations out of order:
756--
757-- >>> :{
758-- Stream.toList $ Stream.fromAsync $ do
759-- x <- Stream.fromList [2,1] -- foreach x in stream
760-- Stream.fromEffect $ delay x
761-- :}
762-- 1 sec
763-- 2 sec
764-- [1,2]
765--
766-- Nested monad binds behave like nested @for@ loops with nested iterations
767-- executed concurrently, a la the 'async' combinator:
768--
769-- >>> :{
770-- Stream.toList $ Stream.fromAsync $ do
771-- x <- Stream.fromList [1,2] -- foreach x in stream
772-- y <- Stream.fromList [2,4] -- foreach y in stream
773-- Stream.fromEffect $ delay (x + y)
774-- :}
775-- 3 sec
776-- 4 sec
777-- 5 sec
778-- 6 sec
779-- [3,4,5,6]
780--
781-- The behavior can be explained as follows. All the iterations corresponding
782-- to the element @1@ in the first stream constitute one output stream and all
783-- the iterations corresponding to @2@ constitute another output stream and
784-- these two output streams are merged using 'async'.
785--
786-- /Since: 0.1.0 ("Streamly")/
787--
788-- @since 0.8.0
789newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
790 deriving (MonadTrans)
792-- | A demand driven left biased parallely composing IO stream of elements of
793-- type @a@. See 'AsyncT' documentation for more details.
794--
795-- /Since: 0.2.0 ("Streamly")/
796--
797-- @since 0.8.0
798type Async = AsyncT IO
800-- | Fix the type of a polymorphic stream as 'AsyncT'.
801--
802-- /Since: 0.1.0 ("Streamly")/
803--
804-- @since 0.8.0
805fromAsync :: IsStream t => AsyncT m a -> t m a
806fromAsync = adapt
808instance IsStream AsyncT where
809 toStream = getAsyncT
810 fromStream = AsyncT
811 consM = consMAsync
812 (|:) = consMAsync
814------------------------------------------------------------------------------
815-- Semigroup
816------------------------------------------------------------------------------
818-- Monomorphically typed version of "async" for better performance of Semigroup
819-- instance.
820{-# INLINE mappendAsync #-}
821{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
822mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
823mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
825instance MonadAsync m => Semigroup (AsyncT m a) where
826 (<>) = mappendAsync
828------------------------------------------------------------------------------
829-- Monoid
830------------------------------------------------------------------------------
832instance MonadAsync m => Monoid (AsyncT m a) where
833 mempty = K.nil
834 mappend = (<>)
836------------------------------------------------------------------------------
837-- Applicative
838------------------------------------------------------------------------------
840{-# INLINE apAsync #-}
841{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
842apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
843apAsync (AsyncT m1) (AsyncT m2) =
844 let f x1 = K.concatMapBy async (pure . x1) m2
845 in AsyncT $ K.concatMapBy async f m1
847instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
848 {-# INLINE pure #-}
849 pure = AsyncT . K.fromPure
850 {-# INLINE (<*>) #-}
851 (<*>) = apAsync
853------------------------------------------------------------------------------
854-- Monad
855------------------------------------------------------------------------------
857-- GHC: if we change the implementation of bindWith with arguments in a
858-- different order we see a significant performance degradation (~2x).
859{-# INLINE bindAsync #-}
860{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
861bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
862bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
864-- GHC: if we specify arguments in the definition of (>>=) we see a significant
865-- performance degradation (~2x).
866instance MonadAsync m => Monad (AsyncT m) where
867 return = pure
868 (>>=) = bindAsync
870------------------------------------------------------------------------------
871-- Other instances
872------------------------------------------------------------------------------
874MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
876------------------------------------------------------------------------------
877-- WAsyncT
878------------------------------------------------------------------------------
880-- | XXX we can implement it more efficienty by directly implementing instead
881-- of combining streams using wAsync.
882{-# INLINE consMWAsync #-}
883{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
884consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
885consMWAsync m r = fromStream $ K.fromEffect m `wAsync` (toStream r)
887infixr 6 `wAsync`
889-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
890-- singleton stream behavior. For multi-element streams, while 'async' is left
891-- biased i.e. it tries to evaluate the left side stream as much as possible,
892-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
893-- deep while 'wAsync' goes wide. However, outputs are always used as they
894-- arrive.
895--
896-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
897-- starts behaving like 'wSerial'.
898--
899-- >>> import Streamly.Prelude (wAsync)
900-- >>> stream1 = Stream.fromList [1,2,3]
901-- >>> stream2 = Stream.fromList [4,5,6]
902-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 1 $ stream1 `async` stream2
903-- [1,2,3,4,5,6]
904--
905-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
906-- [1,4,2,5,3,6]
907--
908-- With two threads available, and combining three streams:
909--
910-- >>> stream3 = Stream.fromList [7,8,9]
911-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
912-- [1,2,3,4,5,6,7,8,9]
913--
914-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
915-- [1,4,2,7,5,3,8,6,9]
916--
917-- This operation cannot be used to fold an infinite lazy container of streams,
918-- because it schedules all the streams in a round robin manner.
919--
920-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
921-- but the exact scheduling is slightly different in both cases.
922--
923-- @since 0.8.0
924--
925-- /Since: 0.2.0 ("Streamly")/
927-- Scheduling details:
928--
929-- This is how the execution of the above example proceeds:
930--
931-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
932-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
933-- queue is represented by the rightmost item.
934-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
935-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
936-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
937-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
938-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
939-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
940-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
941-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
942-- 4. Note that the scheduler queue expands by one more stream component in
943-- every pass because one more @<>@ is broken down into two components. At this
944-- point there are no more @<>@ operations to be broken down further and the
945-- queue has reached its maximum size. Now these streams are scheduled in
946-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
947--
948-- As we see above, in a right associated expression composed with @<>@, only
949-- one @<>@ operation is broken down into two components in one execution,
950-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
951-- scheduler passes to expand the whole expression. By the time @n-th@
952-- component is added to the scheduler queue, the first component would have
953-- received @n@ scheduler passes.
954--
955{-# INLINE wAsync #-}
956wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
957wAsync = joinStreamVarAsync WAsyncVar
959-- | For 'WAsyncT' streams:
960--
961-- @
962-- (<>) = 'Streamly.Prelude.wAsync'
963-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wAsync'
964-- @
965--
966-- A single 'Monad' bind behaves like a @for@ loop with iterations of the loop
967-- executed concurrently a la the 'wAsync' combinator, producing results and
968-- side effects of iterations out of order:
969--
970-- >>> :{
971-- Stream.toList $ Stream.fromWAsync $ do
972-- x <- Stream.fromList [2,1] -- foreach x in stream
973-- Stream.fromEffect $ delay x
974-- :}
975-- 1 sec
976-- 2 sec
977-- [1,2]
978--
979-- Nested monad binds behave like nested @for@ loops with nested iterations
980-- executed concurrently, a la the 'wAsync' combinator:
981--
982-- >>> :{
983-- Stream.toList $ Stream.fromWAsync $ do
984-- x <- Stream.fromList [1,2] -- foreach x in stream
985-- y <- Stream.fromList [2,4] -- foreach y in stream
986-- Stream.fromEffect $ delay (x + y)
987-- :}
988-- 3 sec
989-- 4 sec
990-- 5 sec
991-- 6 sec
992-- [3,4,5,6]
993--
994-- The behavior can be explained as follows. All the iterations corresponding
995-- to the element @1@ in the first stream constitute one 'WAsyncT' output
996-- stream and all the iterations corresponding to @2@ constitute another
997-- 'WAsyncT' output stream and these two output streams are merged using
998-- 'wAsync'.
999--
1000-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
1001-- contrast to the depth wise scheduling behavior of 'AsyncT'.
1003-- /Since: 0.2.0 ("Streamly")/
1005-- @since 0.8.0
1007-- XXX This documentation is redundant, need to be cleaned up/removed.
1009-- @
1010-- import qualified "Streamly.Prelude" as S
1011-- import Control.Concurrent
1013-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
1014-- @
1015-- @
1016-- [1,3,2,4]
1017-- @
1019-- For this example, we are using @maxThreads 1@ so that concurrent thread
1020-- scheduling does not affect the results and make them unpredictable. Let's
1021-- now take a more general example:
1023-- @
1024-- main = (S.toList . S.'fromWAsync' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
1025-- @
1026-- @
1027-- [1,4,2,7,5,3,8,6,9]
1028-- @
1030-- This is how the execution of the above stream proceeds:
1032-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
1033-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
1034-- queue is represented by the rightmost item.
1035-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
1036-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
1037-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
1038-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
1039-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
1040-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
1041-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
1042-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
1043-- 4. Note that the scheduler queue expands by one more stream component in
1044-- every pass because one more @<>@ is broken down into two components. At this
1045-- point there are no more @<>@ operations to be broken down further and the
1046-- queue has reached its maximum size. Now these streams are scheduled in
1047-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
1049-- As we see above, in a right associated expression composed with @<>@, only
1050-- one @<>@ operation is broken down into two components in one execution,
1051-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
1052-- scheduler passes to expand the whole expression. By the time @n-th@
1053-- component is added to the scheduler queue, the first component would have
1054-- received @n@ scheduler passes.
1056-- Since all streams get interleaved, this operation is not suitable for
1057-- folding an infinite lazy container of infinite size streams. However, if
1058-- the streams are small, the streams on the left may get finished before more
1059-- streams are added to the scheduler queue from the right side of the
1060-- expression, so it may be possible to fold an infinite lazy container of
1061-- streams. For example, if the streams are of size @n@ then at most @n@
1062-- streams would be in the scheduler queue at a time.
1064-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
1065-- therefore the output of 'WAsyncT' even with a single thread of execution is
1066-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
1067-- its scheduling behavior.
1069-- Any exceptions generated by a constituent stream are propagated to the
1070-- output stream. The output and exceptions from a single stream are guaranteed
1071-- to arrive in the same order in the resulting stream as they were generated
1072-- in the input stream. However, the relative ordering of elements from
1073-- different streams in the resulting stream can vary depending on scheduling
1074-- and generation delays.
1076-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
1077-- concurrently using a round robin scheduling.
1079-- @
1080-- main = S.'drain' . S.'fromWAsync' $ do
1081-- n <- return 3 \<\> return 2 \<\> return 1
1082-- S.fromEffect $ do
1083-- threadDelay (n * 1000000)
1084-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1085-- @
1086-- @
1087-- ThreadId 40: Delay 1
1088-- ThreadId 39: Delay 2
1089-- ThreadId 38: Delay 3
1090-- @
1092newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
1093 deriving (MonadTrans)
1095-- | A round robin parallely composing IO stream of elements of type @a@.
1096-- See 'WAsyncT' documentation for more details.
1098-- /Since: 0.2.0 ("Streamly")/
1100-- @since 0.8.0
1101type WAsync = WAsyncT IO
1103-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1105-- /Since: 0.2.0 ("Streamly")/
1107-- @since 0.8.0
1108fromWAsync :: IsStream t => WAsyncT m a -> t m a
1109fromWAsync = adapt
1111instance IsStream WAsyncT where
1112 toStream = getWAsyncT
1113 fromStream = WAsyncT
1114 consM = consMWAsync
1115 (|:) = consMWAsync
1117------------------------------------------------------------------------------
1118-- Semigroup
1119------------------------------------------------------------------------------
1121{-# INLINE mappendWAsync #-}
1122{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1123mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
1124mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
1126instance MonadAsync m => Semigroup (WAsyncT m a) where
1127 (<>) = mappendWAsync
1129------------------------------------------------------------------------------
1130-- Monoid
1131------------------------------------------------------------------------------
1133instance MonadAsync m => Monoid (WAsyncT m a) where
1134 mempty = K.nil
1135 mappend = (<>)
1137------------------------------------------------------------------------------
1138-- Applicative
1139------------------------------------------------------------------------------
1141{-# INLINE apWAsync #-}
1142{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
1143apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
1144apWAsync (WAsyncT m1) (WAsyncT m2) =
1145 let f x1 = K.concatMapBy wAsync (pure . x1) m2
1146 in WAsyncT $ K.concatMapBy wAsync f m1
1148-- GHC: if we specify arguments in the definition of (<*>) we see a significant
1149-- performance degradation (~2x).
1150instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
1151 pure = WAsyncT . K.fromPure
1152 (<*>) = apWAsync
1154------------------------------------------------------------------------------
1155-- Monad
1156------------------------------------------------------------------------------
1158-- GHC: if we change the implementation of bindWith with arguments in a
1159-- different order we see a significant performance degradation (~2x).
1160{-# INLINE bindWAsync #-}
1161{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
1162bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
1163bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
1165-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1166-- performance degradation (~2x).
1167instance MonadAsync m => Monad (WAsyncT m) where
1168 return = pure
1169 (>>=) = bindWAsync
1171------------------------------------------------------------------------------
1172-- Other instances
1173------------------------------------------------------------------------------
1175MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)