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