git.iko.soy/streamly
on eff3255 help sign in
streamly/src/Streamly/Internal/Data/Stream/Async.hs
994 lines 39 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3#include "inline.hs"
5-- |
6-- Module : Streamly.Internal.Data.Stream.Async
7-- Copyright : (c) 2017 Harendra Kumar
8--
9-- License : BSD3
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
12-- Portability : GHC
13--
14--
15module Streamly.Internal.Data.Stream.Async
16 (
17 AsyncT
18 , Async
19 , asyncly
20 , async
21 , (<|) --deprecated
22 , mkAsync
23 , mkAsyncK
25 , WAsyncT
26 , WAsync
27 , wAsyncly
28 , wAsync
29 )
30where
32import Control.Concurrent (myThreadId)
33import Control.Monad.Base (MonadBase(..), liftBaseDefault)
34import Control.Monad.Catch (MonadThrow, throwM)
35import Control.Monad.Trans.Control (MonadBaseControl (..))
36import Control.Concurrent.MVar (newEmptyMVar)
37-- import Control.Monad.Error.Class (MonadError(..))
38import Control.Monad.IO.Class (MonadIO(..))
39import Control.Monad.Reader.Class (MonadReader(..))
40import Control.Monad.State.Class (MonadState(..))
41import Control.Monad.Trans.Class (MonadTrans(lift))
42import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR)
43import Data.IORef (IORef, newIORef, readIORef)
44import Data.Maybe (fromJust)
45#if __GLASGOW_HASKELL__ < 808
46import Data.Semigroup (Semigroup(..))
47#endif
49import Prelude hiding (map)
50import qualified Data.Set as S
52import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)
53import Streamly.Internal.Data.Stream.SVar (fromSVar)
54import Streamly.Internal.Data.SVar
55import Streamly.Internal.Data.Stream.StreamK
56 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared)
57import qualified Streamly.Internal.Data.Stream.StreamK as K
58import qualified Streamly.Internal.Data.Stream.StreamD as D
60#include "Instances.hs"
62-------------------------------------------------------------------------------
63-- Async
64-------------------------------------------------------------------------------
66data WorkerStatus = Continue | Suspend
68{-# INLINE workLoopLIFO #-}
69workLoopLIFO
70 :: (MonadIO m, MonadBaseControl IO m)
71 => IORef [(RunInIO m, Stream m a)]
72 -> State Stream m a
73 -> SVar Stream m a
74 -> Maybe WorkerInfo
75 -> m ()
76workLoopLIFO q st sv winfo = run
78 where
80 stop = liftIO $ sendStop sv winfo
81 run = do
82 work <- dequeue
83 case work of
84 Nothing -> stop
85 Just (RunInIO runin, m) -> do
86 -- XXX when we finish we need to send the monadic state back to
87 -- the parent so that the state can be merged back. We capture
88 -- and return the state in the stop continuation.
89 --
90 -- Instead of using the run function we can just restore the
91 -- monad state here. That way it can work easily for
92 -- distributed case as well.
93 r <- liftIO $ runin $
94 foldStreamShared st yieldk single (return Continue) m
95 res <- restoreM r
96 case res of
97 Continue -> run
98 Suspend -> stop
100 single a = do
101 res <- liftIO $ sendYield sv winfo (ChildYield a)
102 return $ if res then Continue else Suspend
104 yieldk a r = do
105 res <- liftIO $ sendYield sv winfo (ChildYield a)
106 if res
107 then foldStreamShared st yieldk single (return Continue) r
108 else do
109 runInIO <- captureMonadState
110 liftIO $ enqueueLIFO sv q (runInIO, r)
111 return Suspend
113 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
114 [] -> ([], Nothing)
115 x : xs -> (xs, Just x)
117-- We duplicate workLoop for yield limit and no limit cases because it has
118-- around 40% performance overhead in the worst case.
119--
120-- XXX we can pass yinfo directly as an argument here so that we do not have to
121-- make a check every time.
122{-# INLINE workLoopLIFOLimited #-}
123workLoopLIFOLimited
124 :: forall m a. (MonadIO m, MonadBaseControl IO m)
125 => IORef [(RunInIO m, Stream m a)]
126 -> State Stream m a
127 -> SVar Stream m a
128 -> Maybe WorkerInfo
129 -> m ()
130workLoopLIFOLimited q st sv winfo = run
132 where
134 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
135 stop = liftIO $ sendStop sv winfo
136 run = do
137 work <- dequeue
138 case work of
139 Nothing -> stop
140 Just (RunInIO runin, m) -> do
141 -- XXX This is just a best effort minimization of concurrency
142 -- to the yield limit. If the stream is made of concurrent
143 -- streams we do not reserve the yield limit in the constituent
144 -- streams before executing the action. This can be done
145 -- though, by sharing the yield limit ref with downstream
146 -- actions via state passing. Just a todo.
147 yieldLimitOk <- liftIO $ decrementYieldLimit sv
148 if yieldLimitOk
149 then do
150 r <- liftIO $ runin $
151 foldStreamShared st yieldk single incrContinue m
152 res <- restoreM r
153 case res of
154 Continue -> run
155 Suspend -> stop
156 -- Avoid any side effects, undo the yield limit decrement if we
157 -- never yielded anything.
158 else liftIO $ do
159 enqueueLIFO sv q (RunInIO runin, m)
160 incrementYieldLimit sv
161 sendStop sv winfo
163 single a = do
164 res <- liftIO $ sendYield sv winfo (ChildYield a)
165 return $ if res then Continue else Suspend
167 -- XXX can we pass on the yield limit downstream to limit the concurrency
168 -- of constituent streams.
169 yieldk a r = do
170 res <- liftIO $ sendYield sv winfo (ChildYield a)
171 yieldLimitOk <- liftIO $ decrementYieldLimit sv
172 if res && yieldLimitOk
173 then foldStreamShared st yieldk single incrContinue r
174 else do
175 runInIO <- captureMonadState
176 liftIO $ incrementYieldLimit sv
177 liftIO $ enqueueLIFO sv q (runInIO, r)
178 return Suspend
180 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
181 [] -> ([], Nothing)
182 x : xs -> (xs, Just x)
184-------------------------------------------------------------------------------
185-- WAsync
186-------------------------------------------------------------------------------
188-- XXX we can remove sv as it is derivable from st
190{-# INLINE workLoopFIFO #-}
191workLoopFIFO
192 :: (MonadIO m, MonadBaseControl IO m)
193 => LinkedQueue (RunInIO m, Stream m a)
194 -> State Stream m a
195 -> SVar Stream m a
196 -> Maybe WorkerInfo
197 -> m ()
198workLoopFIFO q st sv winfo = run
200 where
202 stop = liftIO $ sendStop sv winfo
203 run = do
204 work <- liftIO $ tryPopR q
205 case work of
206 Nothing -> stop
207 Just (RunInIO runin, m) -> do
208 r <- liftIO $ runin $
209 foldStreamShared st yieldk single (return Continue) m
210 res <- restoreM r
211 case res of
212 Continue -> run
213 Suspend -> stop
215 single a = do
216 res <- liftIO $ sendYield sv winfo (ChildYield a)
217 return $ if res then Continue else Suspend
219 -- XXX in general we would like to yield "n" elements from a single stream
220 -- before moving on to the next. Single element granularity could be too
221 -- expensive in certain cases. Similarly, we can use time limit for
222 -- yielding.
223 yieldk a r = do
224 res <- liftIO $ sendYield sv winfo (ChildYield a)
225 runInIO <- captureMonadState
226 liftIO $ enqueueFIFO sv q (runInIO, r)
227 return $ if res then Continue else Suspend
229{-# INLINE workLoopFIFOLimited #-}
230workLoopFIFOLimited
231 :: forall m a. (MonadIO m, MonadBaseControl IO m)
232 => LinkedQueue (RunInIO m, Stream m a)
233 -> State Stream m a
234 -> SVar Stream m a
235 -> Maybe WorkerInfo
236 -> m ()
237workLoopFIFOLimited q st sv winfo = run
239 where
241 stop = liftIO $ sendStop sv winfo
242 incrContinue = liftIO (incrementYieldLimit sv) >> return Continue
243 run = do
244 work <- liftIO $ tryPopR q
245 case work of
246 Nothing -> stop
247 Just (RunInIO runin, m) -> do
248 yieldLimitOk <- liftIO $ decrementYieldLimit sv
249 if yieldLimitOk
250 then do
251 r <- liftIO $ runin $
252 foldStreamShared st yieldk single incrContinue m
253 res <- restoreM r
254 case res of
255 Continue -> run
256 Suspend -> stop
257 else liftIO $ do
258 enqueueFIFO sv q (RunInIO runin, m)
259 incrementYieldLimit sv
260 sendStop sv winfo
262 single a = do
263 res <- liftIO $ sendYield sv winfo (ChildYield a)
264 return $ if res then Continue else Suspend
266 yieldk a r = do
267 res <- liftIO $ sendYield sv winfo (ChildYield a)
268 runInIO <- captureMonadState
269 liftIO $ enqueueFIFO sv q (runInIO, r)
270 yieldLimitOk <- liftIO $ decrementYieldLimit sv
271 if res && yieldLimitOk
272 then return Continue
273 else liftIO $ do
274 incrementYieldLimit sv
275 return Suspend
277-------------------------------------------------------------------------------
278-- SVar creation
279-- This code belongs in SVar.hs but is kept here for perf reasons
280-------------------------------------------------------------------------------
282-- XXX we have this function in this file because passing runStreamLIFO as a
283-- function argument to this function results in a perf degradation of more
284-- than 10%. Need to investigate what the root cause is.
285-- Interestingly, the same thing does not make any difference for Ahead.
286getLifoSVar :: forall m a. MonadAsync m
287 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
288getLifoSVar st mrun = do
289 outQ <- newIORef ([], 0)
290 outQMv <- newEmptyMVar
291 active <- newIORef 0
292 wfw <- newIORef False
293 running <- newIORef S.empty
294 q <- newIORef ([] :: [(RunInIO m, Stream m a)])
295 yl <- case getYieldLimit st of
296 Nothing -> return Nothing
297 Just x -> Just <$> newIORef x
298 rateInfo <- getYieldRateInfo st
300 stats <- newSVarStats
301 tid <- myThreadId
303 let isWorkFinished _ = null <$> readIORef q
305 let isWorkFinishedLimited sv = do
306 yieldsDone <-
307 case remainingWork sv of
308 Just ref -> do
309 n <- readIORef ref
310 return (n <= 0)
311 Nothing -> return False
312 qEmpty <- null <$> readIORef q
313 return $ qEmpty || yieldsDone
315 let getSVar :: SVar Stream m a
316 -> (SVar Stream m a -> m [ChildEvent a])
317 -> (SVar Stream m a -> m Bool)
318 -> (SVar Stream m a -> IO Bool)
319 -> (IORef [(RunInIO m, Stream m a)]
320 -> State Stream m a
321 -> SVar Stream m a
322 -> Maybe WorkerInfo
323 -> m())
324 -> SVar Stream m a
325 getSVar sv readOutput postProc workDone wloop = SVar
326 { outputQueue = outQ
327 , outputQueueFromConsumer = undefined
328 , remainingWork = yl
329 , maxBufferLimit = getMaxBuffer st
330 , pushBufferSpace = undefined
331 , pushBufferPolicy = undefined
332 , pushBufferMVar = undefined
333 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
334 , yieldRateInfo = rateInfo
335 , outputDoorBell = outQMv
336 , outputDoorBellFromConsumer = undefined
337 , readOutputQ = readOutput sv
338 , postProcess = postProc sv
339 , workerThreads = running
340 , workLoop = wloop q st{streamVar = Just sv} sv
341 , enqueue = enqueueLIFO sv q
342 , isWorkDone = workDone sv
343 , isQueueDone = workDone sv
344 , needDoorBell = wfw
345 , svarStyle = AsyncVar
346 , svarStopStyle = StopNone
347 , svarStopBy = undefined
348 , svarMrun = mrun
349 , workerCount = active
350 , accountThread = delThread sv
351 , workerStopMVar = undefined
352 , svarRef = Nothing
353 , svarInspectMode = getInspectMode st
354 , svarCreator = tid
355 , aheadWorkQueue = undefined
356 , outputHeap = undefined
357 , svarStats = stats
358 }
360 let sv =
361 case getStreamRate st of
362 Nothing ->
363 case getYieldLimit st of
364 Nothing -> getSVar sv readOutputQBounded
365 postProcessBounded
366 isWorkFinished
367 workLoopLIFO
368 Just _ -> getSVar sv readOutputQBounded
369 postProcessBounded
370 isWorkFinishedLimited
371 workLoopLIFOLimited
372 Just _ ->
373 case getYieldLimit st of
374 Nothing -> getSVar sv readOutputQPaced
375 postProcessPaced
376 isWorkFinished
377 workLoopLIFO
378 Just _ -> getSVar sv readOutputQPaced
379 postProcessPaced
380 isWorkFinishedLimited
381 workLoopLIFOLimited
382 in return sv
384getFifoSVar :: forall m a. MonadAsync m
385 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
386getFifoSVar st mrun = do
387 outQ <- newIORef ([], 0)
388 outQMv <- newEmptyMVar
389 active <- newIORef 0
390 wfw <- newIORef False
391 running <- newIORef S.empty
392 q <- newQ
393 yl <- case getYieldLimit st of
394 Nothing -> return Nothing
395 Just x -> Just <$> newIORef x
396 rateInfo <- getYieldRateInfo st
398 stats <- newSVarStats
399 tid <- myThreadId
401 let isWorkFinished _ = nullQ q
402 let isWorkFinishedLimited sv = do
403 yieldsDone <-
404 case remainingWork sv of
405 Just ref -> do
406 n <- readIORef ref
407 return (n <= 0)
408 Nothing -> return False
409 qEmpty <- nullQ q
410 return $ qEmpty || yieldsDone
412 let getSVar :: SVar Stream m a
413 -> (SVar Stream m a -> m [ChildEvent a])
414 -> (SVar Stream m a -> m Bool)
415 -> (SVar Stream m a -> IO Bool)
416 -> (LinkedQueue (RunInIO m, Stream m a)
417 -> State Stream m a
418 -> SVar Stream m a
419 -> Maybe WorkerInfo
420 -> m())
421 -> SVar Stream m a
422 getSVar sv readOutput postProc workDone wloop = SVar
423 { outputQueue = outQ
424 , outputQueueFromConsumer = undefined
425 , remainingWork = yl
426 , maxBufferLimit = getMaxBuffer st
427 , pushBufferSpace = undefined
428 , pushBufferPolicy = undefined
429 , pushBufferMVar = undefined
430 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
431 , yieldRateInfo = rateInfo
432 , outputDoorBell = outQMv
433 , outputDoorBellFromConsumer = undefined
434 , readOutputQ = readOutput sv
435 , postProcess = postProc sv
436 , workerThreads = running
437 , workLoop = wloop q st{streamVar = Just sv} sv
438 , enqueue = enqueueFIFO sv q
439 , isWorkDone = workDone sv
440 , isQueueDone = workDone sv
441 , needDoorBell = wfw
442 , svarStyle = WAsyncVar
443 , svarStopStyle = StopNone
444 , svarStopBy = undefined
445 , svarMrun = mrun
446 , workerCount = active
447 , accountThread = delThread sv
448 , workerStopMVar = undefined
449 , svarRef = Nothing
450 , svarInspectMode = getInspectMode st
451 , svarCreator = tid
452 , aheadWorkQueue = undefined
453 , outputHeap = undefined
454 , svarStats = stats
455 }
457 let sv =
458 case getStreamRate st of
459 Nothing ->
460 case getYieldLimit st of
461 Nothing -> getSVar sv readOutputQBounded
462 postProcessBounded
463 isWorkFinished
464 workLoopFIFO
465 Just _ -> getSVar sv readOutputQBounded
466 postProcessBounded
467 isWorkFinishedLimited
468 workLoopFIFOLimited
469 Just _ ->
470 case getYieldLimit st of
471 Nothing -> getSVar sv readOutputQPaced
472 postProcessPaced
473 isWorkFinished
474 workLoopFIFO
475 Just _ -> getSVar sv readOutputQPaced
476 postProcessPaced
477 isWorkFinishedLimited
478 workLoopFIFOLimited
479 in return sv
481{-# INLINABLE newAsyncVar #-}
482newAsyncVar :: MonadAsync m
483 => State Stream m a -> Stream m a -> m (SVar Stream m a)
484newAsyncVar st m = do
485 mrun <- captureMonadState
486 sv <- liftIO $ getLifoSVar st mrun
487 sendFirstWorker sv m
489-- | Generate a stream asynchronously to keep it buffered, lazily consume
490-- from the buffer.
491--
492-- /Internal/
493--
494{-# INLINABLE mkAsyncK #-}
495mkAsyncK :: (IsStream t, MonadAsync m) => t m a -> t m a
496mkAsyncK m = mkStream $ \st yld sng stp -> do
497 sv <- newAsyncVar (adaptState st) (toStream m)
498 foldStream st yld sng stp $ fromSVar sv
500{-# INLINE_NORMAL mkAsyncD #-}
501mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a
502mkAsyncD m = D.Stream step Nothing
503 where
505 step gst Nothing = do
506 sv <- newAsyncVar gst (D.fromStreamD m)
507 return $ D.Skip $ Just $ D.fromSVar sv
509 step gst (Just (D.UnStream step1 st)) = do
510 r <- step1 gst st
511 return $ case r of
512 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
513 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
514 D.Stop -> D.Stop
516-- This is slightly faster than the CPS version above
517--
518-- | Make the stream producer and consumer run concurrently by introducing a
519-- buffer between them. The producer thread evaluates the input stream until
520-- the buffer fills, it terminates if the buffer is full and a worker thread is
521-- kicked off again to evaluate the remaining stream when there is space in the
522-- buffer. The consumer consumes the stream lazily from the buffer.
523--
524-- /Internal/
525--
526{-# INLINE_NORMAL mkAsync #-}
527mkAsync :: (K.IsStream t, MonadAsync m) => t m a -> t m a
528mkAsync = D.fromStreamD . mkAsyncD . D.toStreamD
530-- | Create a new SVar and enqueue one stream computation on it.
531{-# INLINABLE newWAsyncVar #-}
532newWAsyncVar :: MonadAsync m
533 => State Stream m a -> Stream m a -> m (SVar Stream m a)
534newWAsyncVar st m = do
535 mrun <- captureMonadState
536 sv <- liftIO $ getFifoSVar st mrun
537 sendFirstWorker sv m
539------------------------------------------------------------------------------
540-- Running streams concurrently
541------------------------------------------------------------------------------
543-- Concurrency rate control.
544--
545-- Our objective is to create more threads on demand if the consumer is running
546-- faster than us. As soon as we encounter a concurrent composition we create a
547-- push pull pair of threads. We use an SVar for communication between the
548-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
549-- The producer creates more threads if the SVar drains and becomes empty, that
550-- is the consumer is running faster.
551--
552-- XXX Note 1: This mechanism can be problematic if the initial production
553-- latency is high, we may end up creating too many threads. So we need some
554-- way to monitor and use the latency as well. Having a limit on the dispatches
555-- (programmer controlled) may also help.
556--
557-- TBD Note 2: We may want to run computations at the lower level of the
558-- composition tree serially even when they are composed using a parallel
559-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
560-- place of 'wAsync'. If we find that an SVar immediately above a computation
561-- gets drained empty we can switch to parallelizing the computation. For that
562-- we can use a state flag to fork the rest of the computation at any point of
563-- time inside the Monad bind operation if the consumer is running at a faster
564-- speed.
565--
566-- TBD Note 3: the binary operation ('parallel') composition allows us to
567-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
568-- chunksizes we will need to compose the parallel actions using a data
569-- constructor (A Free container) instead so that we can divide it in chunks of
570-- arbitrary size before dispatching. If the stream is composed of
571-- hierarchically composed grains of different sizes then we can always switch
572-- to a desired granularity depending on the consumer speed.
573--
574-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
575-- into just the number of CPUs.
577-- | Join two computations on the currently running 'SVar' queue for concurrent
578-- execution. When we are using parallel composition, an SVar is passed around
579-- as a state variable. We try to schedule a new parallel computation on the
580-- SVar passed to us. The first time, when no SVar exists, a new SVar is
581-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
582-- already scheduled on the SVar is further evaluated. For example, when (a
583-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
584-- the current scheduler queue.
585--
586-- The 'SVarStyle' required by the current composition context is passed as one
587-- of the parameters. If the scheduling and composition style of the new
588-- computation being scheduled is different than the style of the current SVar,
589-- then we create a new SVar and schedule it on that. The newly created SVar
590-- joins as one of the computations on the current SVar queue.
591--
592-- Cases when we need to switch to a new SVar:
593--
594-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
595-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
596-- because of the scheduling policy change.
597-- * if we 'adapt' a stream of type 'async' to a stream of type
598-- 'Parallel', we create a new SVar at the transitioning bind.
599-- * When the stream is switching from disjunctive composition to conjunctive
600-- composition and vice-versa we create a new SVar to isolate the scheduling
601-- of the two.
603forkSVarAsync :: (IsStream t, MonadAsync m)
604 => SVarStyle -> t m a -> t m a -> t m a
605forkSVarAsync style m1 m2 = mkStream $ \st yld sng stp -> do
606 sv <- case style of
607 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
608 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
609 _ -> error "illegal svar type"
610 foldStream st yld sng stp $ fromSVar sv
611 where
612 concurrently ma mb = mkStream $ \st yld sng stp -> do
613 runInIO <- captureMonadState
614 liftIO $ enqueue (fromJust $ streamVar st) $ (runInIO, mb)
615 foldStreamShared st yld sng stp ma
617{-# INLINE joinStreamVarAsync #-}
618joinStreamVarAsync :: (IsStream t, MonadAsync m)
619 => SVarStyle -> t m a -> t m a -> t m a
620joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
621 case streamVar st of
622 Just sv | svarStyle sv == style -> do
623 runInIO <- captureMonadState
624 liftIO $ enqueue sv $ (runInIO, toStream m2)
625 foldStreamShared st yld sng stp m1
626 _ -> foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
628------------------------------------------------------------------------------
629-- Semigroup and Monoid style compositions for parallel actions
630------------------------------------------------------------------------------
632infixr 6 `async`
634-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
635-- Merges two streams possibly concurrently, preferring the
636-- elements from the left one when available.
637--
638-- @since 0.2.0
639{-# INLINE async #-}
640async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
641async = joinStreamVarAsync AsyncVar
643-- | Same as 'async'.
644--
645-- @since 0.1.0
646{-# DEPRECATED (<|) "Please use 'async' instead." #-}
647{-# INLINE (<|) #-}
648(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
649(<|) = async
651-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
652-- huge difference in the performance of consM in IsStream instance even we
653-- have a SPECIALIZE in the instance.
654--
655-- | XXX we can implement it more efficienty by directly implementing instead
656-- of combining streams using async.
657{-# INLINE consMAsync #-}
658{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
659consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
660consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
662------------------------------------------------------------------------------
663-- AsyncT
664------------------------------------------------------------------------------
666-- | The 'Semigroup' operation (@<>@) for 'AsyncT' merges two streams
667-- concurrently with priority given to the first stream. In @s1 <> s2 <> s3
668-- ...@ the streams s1, s2 and s3 are scheduled for execution in that order.
669-- Multiple scheduled streams may be executed concurrently and the elements
670-- generated by them are served to the consumer as and when they become
671-- available. This behavior is similar to the scheduling and execution behavior
672-- of actions in a single async stream.
673--
674-- Since only a finite number of streams are executed concurrently, this
675-- operation can be used to fold an infinite lazy container of streams.
676--
677-- @
678-- import "Streamly"
679-- import qualified "Streamly.Prelude" as S
680-- import Control.Concurrent
681--
682-- main = (S.toList . 'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
683-- @
684-- @
685-- [1,2,3,4]
686-- @
687--
688-- Any exceptions generated by a constituent stream are propagated to the
689-- output stream. The output and exceptions from a single stream are guaranteed
690-- to arrive in the same order in the resulting stream as they were generated
691-- in the input stream. However, the relative ordering of elements from
692-- different streams in the resulting stream can vary depending on scheduling
693-- and generation delays.
694--
695-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
696-- concurrently based on demand. More concurrent iterations are started only
697-- if the previous iterations are not able to produce enough output for the
698-- consumer.
699--
700-- @
701-- main = 'drain' . 'asyncly' $ do
702-- n <- return 3 \<\> return 2 \<\> return 1
703-- S.yieldM $ do
704-- threadDelay (n * 1000000)
705-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
706-- @
707-- @
708-- ThreadId 40: Delay 1
709-- ThreadId 39: Delay 2
710-- ThreadId 38: Delay 3
711-- @
712--
713-- @since 0.1.0
714newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
715 deriving (MonadTrans)
717-- | A demand driven left biased parallely composing IO stream of elements of
718-- type @a@. See 'AsyncT' documentation for more details.
719--
720-- @since 0.2.0
721type Async = AsyncT IO
723-- | Fix the type of a polymorphic stream as 'AsyncT'.
724--
725-- @since 0.1.0
726asyncly :: IsStream t => AsyncT m a -> t m a
727asyncly = adapt
729instance IsStream AsyncT where
730 toStream = getAsyncT
731 fromStream = AsyncT
732 consM = consMAsync
733 (|:) = consMAsync
735------------------------------------------------------------------------------
736-- Semigroup
737------------------------------------------------------------------------------
739-- Monomorphically typed version of "async" for better performance of Semigroup
740-- instance.
741{-# INLINE mappendAsync #-}
742{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
743mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
744mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
746instance MonadAsync m => Semigroup (AsyncT m a) where
747 (<>) = mappendAsync
749------------------------------------------------------------------------------
750-- Monoid
751------------------------------------------------------------------------------
753instance MonadAsync m => Monoid (AsyncT m a) where
754 mempty = K.nil
755 mappend = (<>)
757------------------------------------------------------------------------------
758-- Applicative
759------------------------------------------------------------------------------
761{-# INLINE apAsync #-}
762{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
763apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
764apAsync (AsyncT m1) (AsyncT m2) =
765 let f x1 = K.concatMapBy async (pure . x1) m2
766 in AsyncT $ K.concatMapBy async f m1
768instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
769 {-# INLINE pure #-}
770 pure = AsyncT . K.yield
771 {-# INLINE (<*>) #-}
772 (<*>) = apAsync
774------------------------------------------------------------------------------
775-- Monad
776------------------------------------------------------------------------------
778-- GHC: if we change the implementation of bindWith with arguments in a
779-- different order we see a significant performance degradation (~2x).
780{-# INLINE bindAsync #-}
781{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
782bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
783bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
785-- GHC: if we specify arguments in the definition of (>>=) we see a significant
786-- performance degradation (~2x).
787instance MonadAsync m => Monad (AsyncT m) where
788 return = pure
789 (>>=) = bindAsync
791------------------------------------------------------------------------------
792-- Other instances
793------------------------------------------------------------------------------
795MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
797------------------------------------------------------------------------------
798-- WAsyncT
799------------------------------------------------------------------------------
801-- | XXX we can implement it more efficienty by directly implementing instead
802-- of combining streams using wAsync.
803{-# INLINE consMWAsync #-}
804{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
805consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
806consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
808infixr 6 `wAsync`
810-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
811-- Merges two streams concurrently choosing elements from both fairly.
812--
813-- @since 0.2.0
814{-# INLINE wAsync #-}
815wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
816wAsync = joinStreamVarAsync WAsyncVar
818-- | 'WAsyncT' is similar to 'WSerialT' but with concurrent execution.
819-- The 'Semigroup' operation (@<>@) for 'WAsyncT' merges two streams
820-- concurrently interleaving the actions from both the streams. In @s1
821-- <> s2 <> s3 ...@, the individual actions from streams @s1@, @s2@ and @s3@
822-- are scheduled for execution in a round-robin fashion. Multiple scheduled
823-- actions may be executed concurrently, the results from concurrent executions
824-- are consumed in the order in which they become available.
825--
826--
827-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
828-- contrast to the depth wise scheduling behavior of 'AsyncT'.
829--
830-- @
831-- import "Streamly"
832-- import qualified "Streamly.Prelude" as S
833-- import Control.Concurrent
834--
835-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
836-- @
837-- @
838-- [1,3,2,4]
839-- @
840--
841-- For this example, we are using @maxThreads 1@ so that concurrent thread
842-- scheduling does not affect the results and make them unpredictable. Let's
843-- now take a more general example:
844--
845-- @
846-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
847-- @
848-- @
849-- [1,4,2,7,5,3,8,6,9]
850-- @
851--
852-- This is how the execution of the above stream proceeds:
853--
854-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
855-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
856-- queue is represented by the rightmost item.
857-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
858-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
859-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
860-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
861-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
862-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
863-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
864-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
865-- 4. Note that the scheduler queue expands by one more stream component in
866-- every pass because one more @<>@ is broken down into two components. At this
867-- point there are no more @<>@ operations to be broken down further and the
868-- queue has reached its maximum size. Now these streams are scheduled in
869-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
870--
871-- As we see above, in a right associated expression composed with @<>@, only
872-- one @<>@ operation is broken down into two components in one execution,
873-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
874-- scheduler passes to expand the whole expression. By the time @n-th@
875-- component is added to the scheduler queue, the first component would have
876-- received @n@ scheduler passes.
877--
878-- Since all streams get interleaved, this operation is not suitable for
879-- folding an infinite lazy container of infinite size streams. However, if
880-- the streams are small, the streams on the left may get finished before more
881-- streams are added to the scheduler queue from the right side of the
882-- expression, so it may be possible to fold an infinite lazy container of
883-- streams. For example, if the streams are of size @n@ then at most @n@
884-- streams would be in the scheduler queue at a time.
885--
886-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
887-- therefore the output of 'WAsyncT' even with a single thread of execution is
888-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
889-- its scheduling behavior.
890--
891-- Any exceptions generated by a constituent stream are propagated to the
892-- output stream. The output and exceptions from a single stream are guaranteed
893-- to arrive in the same order in the resulting stream as they were generated
894-- in the input stream. However, the relative ordering of elements from
895-- different streams in the resulting stream can vary depending on scheduling
896-- and generation delays.
897--
898-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
899-- concurrently using a round robin scheduling.
900--
901-- @
902-- main = 'drain' . 'wAsyncly' $ do
903-- n <- return 3 \<\> return 2 \<\> return 1
904-- S.yieldM $ do
905-- threadDelay (n * 1000000)
906-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
907-- @
908-- @
909-- ThreadId 40: Delay 1
910-- ThreadId 39: Delay 2
911-- ThreadId 38: Delay 3
912-- @
913--
914-- @since 0.2.0
915newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
916 deriving (MonadTrans)
918-- | A round robin parallely composing IO stream of elements of type @a@.
919-- See 'WAsyncT' documentation for more details.
920--
921-- @since 0.2.0
922type WAsync = WAsyncT IO
924-- | Fix the type of a polymorphic stream as 'WAsyncT'.
925--
926-- @since 0.2.0
927wAsyncly :: IsStream t => WAsyncT m a -> t m a
928wAsyncly = adapt
930instance IsStream WAsyncT where
931 toStream = getWAsyncT
932 fromStream = WAsyncT
933 consM = consMWAsync
934 (|:) = consMWAsync
936------------------------------------------------------------------------------
937-- Semigroup
938------------------------------------------------------------------------------
940{-# INLINE mappendWAsync #-}
941{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
942mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
943mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
945instance MonadAsync m => Semigroup (WAsyncT m a) where
946 (<>) = mappendWAsync
948------------------------------------------------------------------------------
949-- Monoid
950------------------------------------------------------------------------------
952instance MonadAsync m => Monoid (WAsyncT m a) where
953 mempty = K.nil
954 mappend = (<>)
956------------------------------------------------------------------------------
957-- Applicative
958------------------------------------------------------------------------------
960{-# INLINE apWAsync #-}
961{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
962apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
963apWAsync (WAsyncT m1) (WAsyncT m2) =
964 let f x1 = K.concatMapBy wAsync (pure . x1) m2
965 in WAsyncT $ K.concatMapBy wAsync f m1
967-- GHC: if we specify arguments in the definition of (<*>) we see a significant
968-- performance degradation (~2x).
969instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
970 pure = WAsyncT . K.yield
971 (<*>) = apWAsync
973------------------------------------------------------------------------------
974-- Monad
975------------------------------------------------------------------------------
977-- GHC: if we change the implementation of bindWith with arguments in a
978-- different order we see a significant performance degradation (~2x).
979{-# INLINE bindWAsync #-}
980{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
981bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
982bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
984-- GHC: if we specify arguments in the definition of (>>=) we see a significant
985-- performance degradation (~2x).
986instance MonadAsync m => Monad (WAsyncT m) where
987 return = pure
988 (>>=) = bindWAsync
990------------------------------------------------------------------------------
991-- Other instances
992------------------------------------------------------------------------------
994MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)