git.iko.soy/streamly
on de680fb help sign in
streamly/src/Streamly/Internal/Data/Stream/Async.hs
1010 lines 40 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-- /Since: 0.8.0 ("Streamly.Prelude")/
517--
518-- This is slightly faster than the CPS version above
519--
520-- | Make the stream producer and consumer run concurrently by introducing a
521-- buffer between them. The producer thread evaluates the input stream until
522-- the buffer fills, it terminates if the buffer is full and a worker thread is
523-- kicked off again to evaluate the remaining stream when there is space in the
524-- buffer. The consumer consumes the stream lazily from the buffer.
525--
526-- /Internal/
527--
528{-# INLINE_NORMAL mkAsync #-}
529mkAsync :: (K.IsStream t, MonadAsync m) => t m a -> t m a
530mkAsync = D.fromStreamD . mkAsyncD . D.toStreamD
532-- | Create a new SVar and enqueue one stream computation on it.
533{-# INLINABLE newWAsyncVar #-}
534newWAsyncVar :: MonadAsync m
535 => State Stream m a -> Stream m a -> m (SVar Stream m a)
536newWAsyncVar st m = do
537 mrun <- captureMonadState
538 sv <- liftIO $ getFifoSVar st mrun
539 sendFirstWorker sv m
541------------------------------------------------------------------------------
542-- Running streams concurrently
543------------------------------------------------------------------------------
545-- Concurrency rate control.
546--
547-- Our objective is to create more threads on demand if the consumer is running
548-- faster than us. As soon as we encounter a concurrent composition we create a
549-- push pull pair of threads. We use an SVar for communication between the
550-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
551-- The producer creates more threads if the SVar drains and becomes empty, that
552-- is the consumer is running faster.
553--
554-- XXX Note 1: This mechanism can be problematic if the initial production
555-- latency is high, we may end up creating too many threads. So we need some
556-- way to monitor and use the latency as well. Having a limit on the dispatches
557-- (programmer controlled) may also help.
558--
559-- TBD Note 2: We may want to run computations at the lower level of the
560-- composition tree serially even when they are composed using a parallel
561-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
562-- place of 'wAsync'. If we find that an SVar immediately above a computation
563-- gets drained empty we can switch to parallelizing the computation. For that
564-- we can use a state flag to fork the rest of the computation at any point of
565-- time inside the Monad bind operation if the consumer is running at a faster
566-- speed.
567--
568-- TBD Note 3: the binary operation ('parallel') composition allows us to
569-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
570-- chunksizes we will need to compose the parallel actions using a data
571-- constructor (A Free container) instead so that we can divide it in chunks of
572-- arbitrary size before dispatching. If the stream is composed of
573-- hierarchically composed grains of different sizes then we can always switch
574-- to a desired granularity depending on the consumer speed.
575--
576-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
577-- into just the number of CPUs.
579-- | Join two computations on the currently running 'SVar' queue for concurrent
580-- execution. When we are using parallel composition, an SVar is passed around
581-- as a state variable. We try to schedule a new parallel computation on the
582-- SVar passed to us. The first time, when no SVar exists, a new SVar is
583-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
584-- already scheduled on the SVar is further evaluated. For example, when (a
585-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
586-- the current scheduler queue.
587--
588-- The 'SVarStyle' required by the current composition context is passed as one
589-- of the parameters. If the scheduling and composition style of the new
590-- computation being scheduled is different than the style of the current SVar,
591-- then we create a new SVar and schedule it on that. The newly created SVar
592-- joins as one of the computations on the current SVar queue.
593--
594-- Cases when we need to switch to a new SVar:
595--
596-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
597-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
598-- because of the scheduling policy change.
599-- * if we 'adapt' a stream of type 'async' to a stream of type
600-- 'Parallel', we create a new SVar at the transitioning bind.
601-- * When the stream is switching from disjunctive composition to conjunctive
602-- composition and vice-versa we create a new SVar to isolate the scheduling
603-- of the two.
605forkSVarAsync :: (IsStream t, MonadAsync m)
606 => SVarStyle -> t m a -> t m a -> t m a
607forkSVarAsync style m1 m2 = mkStream $ \st yld sng stp -> do
608 sv <- case style of
609 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
610 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
611 _ -> error "illegal svar type"
612 foldStream st yld sng stp $ fromSVar sv
613 where
614 concurrently ma mb = mkStream $ \st yld sng stp -> do
615 runInIO <- captureMonadState
616 liftIO $ enqueue (fromJust $ streamVar st) $ (runInIO, mb)
617 foldStreamShared st yld sng stp ma
619{-# INLINE joinStreamVarAsync #-}
620joinStreamVarAsync :: (IsStream t, MonadAsync m)
621 => SVarStyle -> t m a -> t m a -> t m a
622joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
623 case streamVar st of
624 Just sv | svarStyle sv == style -> do
625 runInIO <- captureMonadState
626 liftIO $ enqueue sv $ (runInIO, toStream m2)
627 foldStreamShared st yld sng stp m1
628 _ -> foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
630------------------------------------------------------------------------------
631-- Semigroup and Monoid style compositions for parallel actions
632------------------------------------------------------------------------------
634infixr 6 `async`
636-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
637-- Merges two streams possibly concurrently, preferring the
638-- elements from the left one when available.
639--
640-- /Since: 0.2.0 ("Streamly")/
641--
642-- @since 0.8.0
643{-# INLINE async #-}
644async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
645async = joinStreamVarAsync AsyncVar
647-- | Same as 'async'.
648--
649-- @since 0.1.0
650{-# DEPRECATED (<|) "Please use 'async' instead." #-}
651{-# INLINE (<|) #-}
652(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
653(<|) = async
655-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
656-- huge difference in the performance of consM in IsStream instance even we
657-- have a SPECIALIZE in the instance.
658--
659-- | XXX we can implement it more efficienty by directly implementing instead
660-- of combining streams using async.
661{-# INLINE consMAsync #-}
662{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
663consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
664consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
666------------------------------------------------------------------------------
667-- AsyncT
668------------------------------------------------------------------------------
670-- | The 'Semigroup' operation (@<>@) for 'AsyncT' merges two streams
671-- concurrently with priority given to the first stream. In @s1 <> s2 <> s3
672-- ...@ the streams s1, s2 and s3 are scheduled for execution in that order.
673-- Multiple scheduled streams may be executed concurrently and the elements
674-- generated by them are served to the consumer as and when they become
675-- available. This behavior is similar to the scheduling and execution behavior
676-- of actions in a single async stream.
677--
678-- Since only a finite number of streams are executed concurrently, this
679-- operation can be used to fold an infinite lazy container of streams.
680--
681-- @
682-- import qualified "Streamly.Prelude" as S
683-- import Control.Concurrent
684--
685-- main = (S.toList . S.'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
686-- @
687-- @
688-- [1,2,3,4]
689-- @
690--
691-- Any exceptions generated by a constituent stream are propagated to the
692-- output stream. The output and exceptions from a single stream are guaranteed
693-- to arrive in the same order in the resulting stream as they were generated
694-- in the input stream. However, the relative ordering of elements from
695-- different streams in the resulting stream can vary depending on scheduling
696-- and generation delays.
697--
698-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
699-- concurrently based on demand. More concurrent iterations are started only
700-- if the previous iterations are not able to produce enough output for the
701-- consumer.
702--
703-- @
704-- main = S.'drain' . S.'asyncly' $ do
705-- n <- return 3 \<\> return 2 \<\> return 1
706-- S.yieldM $ do
707-- threadDelay (n * 1000000)
708-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
709-- @
710-- @
711-- ThreadId 40: Delay 1
712-- ThreadId 39: Delay 2
713-- ThreadId 38: Delay 3
714-- @
715--
716-- /Since: 0.1.0 ("Streamly")/
717--
718-- @since 0.8.0
719newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
720 deriving (MonadTrans)
722-- | A demand driven left biased parallely composing IO stream of elements of
723-- type @a@. See 'AsyncT' documentation for more details.
724--
725-- /Since: 0.2.0 ("Streamly")/
726--
727-- @since 0.8.0
728type Async = AsyncT IO
730-- | Fix the type of a polymorphic stream as 'AsyncT'.
731--
732-- /Since: 0.1.0 ("Streamly")/
733--
734-- @since 0.8.0
735asyncly :: IsStream t => AsyncT m a -> t m a
736asyncly = adapt
738instance IsStream AsyncT where
739 toStream = getAsyncT
740 fromStream = AsyncT
741 consM = consMAsync
742 (|:) = consMAsync
744------------------------------------------------------------------------------
745-- Semigroup
746------------------------------------------------------------------------------
748-- Monomorphically typed version of "async" for better performance of Semigroup
749-- instance.
750{-# INLINE mappendAsync #-}
751{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
752mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
753mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
755instance MonadAsync m => Semigroup (AsyncT m a) where
756 (<>) = mappendAsync
758------------------------------------------------------------------------------
759-- Monoid
760------------------------------------------------------------------------------
762instance MonadAsync m => Monoid (AsyncT m a) where
763 mempty = K.nil
764 mappend = (<>)
766------------------------------------------------------------------------------
767-- Applicative
768------------------------------------------------------------------------------
770{-# INLINE apAsync #-}
771{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
772apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
773apAsync (AsyncT m1) (AsyncT m2) =
774 let f x1 = K.concatMapBy async (pure . x1) m2
775 in AsyncT $ K.concatMapBy async f m1
777instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
778 {-# INLINE pure #-}
779 pure = AsyncT . K.yield
780 {-# INLINE (<*>) #-}
781 (<*>) = apAsync
783------------------------------------------------------------------------------
784-- Monad
785------------------------------------------------------------------------------
787-- GHC: if we change the implementation of bindWith with arguments in a
788-- different order we see a significant performance degradation (~2x).
789{-# INLINE bindAsync #-}
790{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
791bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
792bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
794-- GHC: if we specify arguments in the definition of (>>=) we see a significant
795-- performance degradation (~2x).
796instance MonadAsync m => Monad (AsyncT m) where
797 return = pure
798 (>>=) = bindAsync
800------------------------------------------------------------------------------
801-- Other instances
802------------------------------------------------------------------------------
804MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
806------------------------------------------------------------------------------
807-- WAsyncT
808------------------------------------------------------------------------------
810-- | XXX we can implement it more efficienty by directly implementing instead
811-- of combining streams using wAsync.
812{-# INLINE consMWAsync #-}
813{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
814consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
815consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
817infixr 6 `wAsync`
819-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
820-- Merges two streams concurrently choosing elements from both fairly.
821--
822-- @since 0.8.0
823--
824-- /Since: 0.2.0 ("Streamly")/
825{-# INLINE wAsync #-}
826wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
827wAsync = joinStreamVarAsync WAsyncVar
829-- | 'WAsyncT' is similar to 'WSerialT' but with concurrent execution.
830-- The 'Semigroup' operation (@<>@) for 'WAsyncT' merges two streams
831-- concurrently interleaving the actions from both the streams. In @s1
832-- <> s2 <> s3 ...@, the individual actions from streams @s1@, @s2@ and @s3@
833-- are scheduled for execution in a round-robin fashion. Multiple scheduled
834-- actions may be executed concurrently, the results from concurrent executions
835-- are consumed in the order in which they become available.
836--
837--
838-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
839-- contrast to the depth wise scheduling behavior of 'AsyncT'.
840--
841-- @
842-- import qualified "Streamly.Prelude" as S
843-- import Control.Concurrent
844--
845-- main = (S.toList . S.'wAsyncly' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
846-- @
847-- @
848-- [1,3,2,4]
849-- @
850--
851-- For this example, we are using @maxThreads 1@ so that concurrent thread
852-- scheduling does not affect the results and make them unpredictable. Let's
853-- now take a more general example:
854--
855-- @
856-- main = (S.toList . S.'wAsyncly' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print
857-- @
858-- @
859-- [1,4,2,7,5,3,8,6,9]
860-- @
861--
862-- This is how the execution of the above stream proceeds:
863--
864-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
865-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
866-- queue is represented by the rightmost item.
867-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
868-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
869-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
870-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
871-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
872-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
873-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
874-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
875-- 4. Note that the scheduler queue expands by one more stream component in
876-- every pass because one more @<>@ is broken down into two components. At this
877-- point there are no more @<>@ operations to be broken down further and the
878-- queue has reached its maximum size. Now these streams are scheduled in
879-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
880--
881-- As we see above, in a right associated expression composed with @<>@, only
882-- one @<>@ operation is broken down into two components in one execution,
883-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
884-- scheduler passes to expand the whole expression. By the time @n-th@
885-- component is added to the scheduler queue, the first component would have
886-- received @n@ scheduler passes.
887--
888-- Since all streams get interleaved, this operation is not suitable for
889-- folding an infinite lazy container of infinite size streams. However, if
890-- the streams are small, the streams on the left may get finished before more
891-- streams are added to the scheduler queue from the right side of the
892-- expression, so it may be possible to fold an infinite lazy container of
893-- streams. For example, if the streams are of size @n@ then at most @n@
894-- streams would be in the scheduler queue at a time.
895--
896-- Note that 'WSerialT' and 'WAsyncT' differ in their scheduling behavior,
897-- therefore the output of 'WAsyncT' even with a single thread of execution is
898-- not the same as that of 'WSerialT' See notes in 'WSerialT' for details about
899-- its scheduling behavior.
900--
901-- Any exceptions generated by a constituent stream are propagated to the
902-- output stream. The output and exceptions from a single stream are guaranteed
903-- to arrive in the same order in the resulting stream as they were generated
904-- in the input stream. However, the relative ordering of elements from
905-- different streams in the resulting stream can vary depending on scheduling
906-- and generation delays.
907--
908-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
909-- concurrently using a round robin scheduling.
910--
911-- @
912-- main = S.'drain' . S.'wAsyncly' $ do
913-- n <- return 3 \<\> return 2 \<\> return 1
914-- S.yieldM $ do
915-- threadDelay (n * 1000000)
916-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
917-- @
918-- @
919-- ThreadId 40: Delay 1
920-- ThreadId 39: Delay 2
921-- ThreadId 38: Delay 3
922-- @
923--
924-- /Since: 0.2.0 ("Streamly")/
925--
926-- @since 0.8.0
927newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
928 deriving (MonadTrans)
930-- | A round robin parallely composing IO stream of elements of type @a@.
931-- See 'WAsyncT' documentation for more details.
932--
933-- /Since: 0.2.0 ("Streamly")/
934--
935-- @since 0.8.0
936type WAsync = WAsyncT IO
938-- | Fix the type of a polymorphic stream as 'WAsyncT'.
939--
940-- /Since: 0.2.0 ("Streamly")/
941--
942-- @since 0.8.0
943wAsyncly :: IsStream t => WAsyncT m a -> t m a
944wAsyncly = adapt
946instance IsStream WAsyncT where
947 toStream = getWAsyncT
948 fromStream = WAsyncT
949 consM = consMWAsync
950 (|:) = consMWAsync
952------------------------------------------------------------------------------
953-- Semigroup
954------------------------------------------------------------------------------
956{-# INLINE mappendWAsync #-}
957{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
958mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
959mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
961instance MonadAsync m => Semigroup (WAsyncT m a) where
962 (<>) = mappendWAsync
964------------------------------------------------------------------------------
965-- Monoid
966------------------------------------------------------------------------------
968instance MonadAsync m => Monoid (WAsyncT m a) where
969 mempty = K.nil
970 mappend = (<>)
972------------------------------------------------------------------------------
973-- Applicative
974------------------------------------------------------------------------------
976{-# INLINE apWAsync #-}
977{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
978apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
979apWAsync (WAsyncT m1) (WAsyncT m2) =
980 let f x1 = K.concatMapBy wAsync (pure . x1) m2
981 in WAsyncT $ K.concatMapBy wAsync f m1
983-- GHC: if we specify arguments in the definition of (<*>) we see a significant
984-- performance degradation (~2x).
985instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
986 pure = WAsyncT . K.yield
987 (<*>) = apWAsync
989------------------------------------------------------------------------------
990-- Monad
991------------------------------------------------------------------------------
993-- GHC: if we change the implementation of bindWith with arguments in a
994-- different order we see a significant performance degradation (~2x).
995{-# INLINE bindWAsync #-}
996{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
997bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
998bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
1000-- GHC: if we specify arguments in the definition of (>>=) we see a significant
1001-- performance degradation (~2x).
1002instance MonadAsync m => Monad (WAsyncT m) where
1003 return = pure
1004 (>>=) = bindWAsync
1006------------------------------------------------------------------------------
1007-- Other instances
1008------------------------------------------------------------------------------
1010MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)