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