git.iko.soy/streamly
on 3ce061d help sign in
streamly/src/Streamly/Streams/Async.hs
841 lines 31 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 StandaloneDeriving #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
12#ifdef DIAGNOSTICS_VERBOSE
13#define DIAGNOSTICS
14#endif
16-- |
17-- Module : Streamly.Streams.Async
18-- Copyright : (c) 2017 Harendra Kumar
19--
20-- License : BSD3
21-- Maintainer : harendra.kumar@gmail.com
22-- Stability : experimental
23-- Portability : GHC
24--
25--
26module Streamly.Streams.Async
27 (
28 AsyncT
29 , Async
30 , asyncly
31 , async
32 , (<|) --deprecated
33 , mkAsync
34 , mkAsync'
36 , WAsyncT
37 , WAsync
38 , wAsyncly
39 , wAsync
40 )
41where
43import Control.Monad (ap)
44import Control.Monad.Base (MonadBase(..), liftBaseDefault)
45import Control.Monad.Catch (MonadThrow, throwM)
46import Control.Concurrent.MVar (newEmptyMVar)
47-- import Control.Monad.Error.Class (MonadError(..))
48import Control.Monad.IO.Class (MonadIO(..))
49import Control.Monad.Reader.Class (MonadReader(..))
50import Control.Monad.State.Class (MonadState(..))
51import Control.Monad.Trans.Class (MonadTrans(lift))
52import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR)
53import Data.IORef (IORef, newIORef, readIORef)
54import Data.Maybe (fromJust)
55import Data.Semigroup (Semigroup(..))
57import Prelude hiding (map)
58import qualified Data.Set as S
60import Streamly.Streams.SVar (fromSVar)
61import Streamly.Streams.Serial (map)
62import Streamly.SVar
63import Streamly.Streams.StreamK (IsStream(..), Stream(..), adapt)
64import qualified Streamly.Streams.StreamK as K
66#ifdef DIAGNOSTICS
67import Control.Concurrent (myThreadId)
68#endif
70#include "Instances.hs"
72-------------------------------------------------------------------------------
73-- Async
74-------------------------------------------------------------------------------
76{-# INLINE workLoopLIFO #-}
77workLoopLIFO
78 :: MonadIO m
79 => IORef [Stream m a]
80 -> State Stream m a
81 -> SVar Stream m a
82 -> Maybe WorkerInfo
83 -> m ()
84workLoopLIFO q st sv winfo = run
86 where
88 run = do
89 work <- dequeue
90 case work of
91 Nothing -> liftIO $ sendStop sv winfo
92 Just m -> unStream m st run single yieldk
94 single a = do
95 res <- liftIO $ sendYield sv winfo (ChildYield a)
96 if res then run else liftIO $ sendStop sv winfo
98 yieldk a r = do
99 res <- liftIO $ sendYield sv winfo (ChildYield a)
100 if res
101 then unStream r st run single yieldk
102 else liftIO $ do
103 enqueueLIFO sv q r
104 sendStop sv winfo
106 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
107 [] -> ([], Nothing)
108 x : xs -> (xs, Just x)
110-- We duplicate workLoop for yield limit and no limit cases because it has
111-- around 40% performance overhead in the worst case.
112--
113-- XXX we can pass yinfo directly as an argument here so that we do not have to
114-- make a check every time.
115{-# INLINE workLoopLIFOLimited #-}
116workLoopLIFOLimited
117 :: MonadIO m
118 => IORef [Stream m a]
119 -> State Stream m a
120 -> SVar Stream m a
121 -> Maybe WorkerInfo
122 -> m ()
123workLoopLIFOLimited q st sv winfo = run
125 where
127 run = do
128 work <- dequeue
129 case work of
130 Nothing -> liftIO $ sendStop sv winfo
131 Just m -> do
132 -- XXX This is just a best effort minimization of concurrency
133 -- to the yield limit. If the stream is made of concurrent
134 -- streams we do not reserve the yield limit in the constituent
135 -- streams before executing the action. This can be done
136 -- though, by sharing the yield limit ref with downstream
137 -- actions via state passing. Just a todo.
138 yieldLimitOk <- liftIO $ decrementYieldLimit sv
139 if yieldLimitOk
140 then do
141 let stop = liftIO (incrementYieldLimit sv) >> run
142 unStream m st stop single yieldk
143 -- Avoid any side effects, undo the yield limit decrement if we
144 -- never yielded anything.
145 else liftIO $ do
146 enqueueLIFO sv q m
147 incrementYieldLimit sv
148 sendStop sv winfo
150 single a = do
151 res <- liftIO $ sendYield sv winfo (ChildYield a)
152 if res then run else liftIO $ sendStop sv winfo
154 -- XXX can we pass on the yield limit downstream to limit the concurrency
155 -- of constituent streams.
156 yieldk a r = do
157 res <- liftIO $ sendYield sv winfo (ChildYield a)
158 yieldLimitOk <- liftIO $ decrementYieldLimit sv
159 let stop = liftIO (incrementYieldLimit sv) >> run
160 if res && yieldLimitOk
161 then unStream r st stop single yieldk
162 else liftIO $ do
163 incrementYieldLimit sv
164 enqueueLIFO sv q r
165 sendStop sv winfo
167 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
168 [] -> ([], Nothing)
169 x : xs -> (xs, Just x)
171-------------------------------------------------------------------------------
172-- WAsync
173-------------------------------------------------------------------------------
175{-# INLINE workLoopFIFO #-}
176workLoopFIFO
177 :: MonadIO m
178 => LinkedQueue (Stream m a)
179 -> State Stream m a
180 -> SVar Stream m a
181 -> Maybe WorkerInfo
182 -> m ()
183workLoopFIFO q st sv winfo = run
185 where
187 run = do
188 work <- liftIO $ tryPopR q
189 case work of
190 Nothing -> liftIO $ sendStop sv winfo
191 Just m -> unStream m st run single yieldk
193 single a = do
194 res <- liftIO $ sendYield sv winfo (ChildYield a)
195 if res then run else liftIO $ sendStop sv winfo
197 yieldk a r = do
198 res <- liftIO $ sendYield sv winfo (ChildYield a)
199 if res
200 then unStream r st run single yieldk
201 else liftIO $ do
202 enqueueFIFO sv q r
203 sendStop sv winfo
205{-# INLINE workLoopFIFOLimited #-}
206workLoopFIFOLimited
207 :: MonadIO m
208 => LinkedQueue (Stream m a)
209 -> State Stream m a
210 -> SVar Stream m a
211 -> Maybe WorkerInfo
212 -> m ()
213workLoopFIFOLimited q st sv winfo = run
215 where
217 run = do
218 work <- liftIO $ tryPopR q
219 case work of
220 Nothing -> liftIO $ sendStop sv winfo
221 Just m -> do
222 yieldLimitOk <- liftIO $ decrementYieldLimit sv
223 if yieldLimitOk
224 then do
225 let stop = liftIO (incrementYieldLimit sv) >> run
226 unStream m st stop single yieldk
227 else liftIO $ do
228 enqueueFIFO sv q m
229 incrementYieldLimit sv
230 sendStop sv winfo
232 single a = do
233 res <- liftIO $ sendYield sv winfo (ChildYield a)
234 if res then run else liftIO $ sendStop sv winfo
236 yieldk a r = do
237 res <- liftIO $ sendYield sv winfo (ChildYield a)
238 yieldLimitOk <- liftIO $ decrementYieldLimit sv
239 let stop = liftIO (incrementYieldLimit sv) >> run
240 if res && yieldLimitOk
241 then unStream r st stop single yieldk
242 else liftIO $ do
243 incrementYieldLimit sv
244 enqueueFIFO sv q r
245 sendStop sv winfo
247-------------------------------------------------------------------------------
248-- SVar creation
249-- This code belongs in SVar.hs but is kept here for perf reasons
250-------------------------------------------------------------------------------
252-- XXX we have this function in this file because passing runStreamLIFO as a
253-- function argument to this function results in a perf degradation of more
254-- than 10%. Need to investigate what the root cause is.
255-- Interestingly, the same thing does not make any difference for Ahead.
256getLifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)
257getLifoSVar st = do
258 outQ <- newIORef ([], 0)
259 outQMv <- newEmptyMVar
260 active <- newIORef 0
261 wfw <- newIORef False
262 running <- newIORef S.empty
263 q <- newIORef []
264 yl <- case getYieldLimit st of
265 Nothing -> return Nothing
266 Just x -> Just <$> newIORef x
267 rateInfo <- getYieldRateInfo st
269 disp <- newIORef 0
270 maxWrk <- newIORef 0
271 maxOq <- newIORef 0
272 maxHs <- newIORef 0
273 maxWq <- newIORef 0
274 avgLat <- newIORef (0, NanoSecs 0)
275 maxLat <- newIORef (NanoSecs 0)
276 minLat <- newIORef (NanoSecs 0)
277 stpTime <- newIORef Nothing
278#ifdef DIAGNOSTICS
279 tid <- myThreadId
280#endif
282 let isWorkFinished _ = null <$> readIORef q
284 let isWorkFinishedLimited sv = do
285 yieldsDone <-
286 case remainingWork sv of
287 Just ref -> do
288 n <- readIORef ref
289 return (n <= 0)
290 Nothing -> return False
291 qEmpty <- null <$> readIORef q
292 return $ qEmpty || yieldsDone
294 let getSVar sv readOutput postProc workDone wloop = SVar
295 { outputQueue = outQ
296 , remainingWork = yl
297 , maxBufferLimit = getMaxBuffer st
298 , maxWorkerLimit = getMaxThreads st
299 , yieldRateInfo = rateInfo
300 , outputDoorBell = outQMv
301 , readOutputQ = readOutput sv
302 , postProcess = postProc sv
303 , workerThreads = running
304 , workLoop = wloop q st{streamVar = Just sv} sv
305 , enqueue = enqueueLIFO sv q
306 , isWorkDone = workDone sv
307 , isQueueDone = workDone sv
308 , needDoorBell = wfw
309 , svarStyle = AsyncVar
310 , workerCount = active
311 , accountThread = delThread sv
312 , workerStopMVar = undefined
313 , svarRef = Nothing
314#ifdef DIAGNOSTICS
315 , svarCreator = tid
316 , aheadWorkQueue = undefined
317 , outputHeap = undefined
318#endif
319 , svarStats = SVarStats
320 { totalDispatches = disp
321 , maxWorkers = maxWrk
322 , maxOutQSize = maxOq
323 , maxHeapSize = maxHs
324 , maxWorkQSize = maxWq
325 , avgWorkerLatency = avgLat
326 , minWorkerLatency = minLat
327 , maxWorkerLatency = maxLat
328 , svarStopTime = stpTime
329 }
330 }
332 let sv =
333 case getStreamRate st of
334 Nothing ->
335 case getYieldLimit st of
336 Nothing -> getSVar sv readOutputQBounded
337 postProcessBounded
338 isWorkFinished
339 workLoopLIFO
340 Just _ -> getSVar sv readOutputQBounded
341 postProcessBounded
342 isWorkFinishedLimited
343 workLoopLIFOLimited
344 Just _ ->
345 case getYieldLimit st of
346 Nothing -> getSVar sv readOutputQPaced
347 postProcessPaced
348 isWorkFinished
349 workLoopLIFO
350 Just _ -> getSVar sv readOutputQPaced
351 postProcessPaced
352 isWorkFinishedLimited
353 workLoopLIFOLimited
354 in return sv
356getFifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)
357getFifoSVar st = do
358 outQ <- newIORef ([], 0)
359 outQMv <- newEmptyMVar
360 active <- newIORef 0
361 wfw <- newIORef False
362 running <- newIORef S.empty
363 q <- newQ
364 yl <- case getYieldLimit st of
365 Nothing -> return Nothing
366 Just x -> Just <$> newIORef x
367 rateInfo <- getYieldRateInfo st
369 disp <- newIORef 0
370 maxWrk <- newIORef 0
371 maxOq <- newIORef 0
372 maxHs <- newIORef 0
373 maxWq <- newIORef 0
374 avgLat <- newIORef (0, NanoSecs 0)
375 maxLat <- newIORef (NanoSecs 0)
376 minLat <- newIORef (NanoSecs 0)
377 stpTime <- newIORef Nothing
378#ifdef DIAGNOSTICS
379 tid <- myThreadId
380#endif
382 let isWorkFinished _ = nullQ q
383 let isWorkFinishedLimited sv = do
384 yieldsDone <-
385 case remainingWork sv of
386 Just ref -> do
387 n <- readIORef ref
388 return (n <= 0)
389 Nothing -> return False
390 qEmpty <- nullQ q
391 return $ qEmpty || yieldsDone
393 let getSVar sv readOutput postProc workDone wloop = SVar
394 { outputQueue = outQ
395 , remainingWork = yl
396 , maxBufferLimit = getMaxBuffer st
397 , maxWorkerLimit = getMaxThreads st
398 , yieldRateInfo = rateInfo
399 , outputDoorBell = outQMv
400 , readOutputQ = readOutput sv
401 , postProcess = postProc sv
402 , workerThreads = running
403 , workLoop = wloop q st{streamVar = Just sv} sv
404 , enqueue = enqueueFIFO sv q
405 , isWorkDone = workDone sv
406 , isQueueDone = workDone sv
407 , needDoorBell = wfw
408 , svarStyle = WAsyncVar
409 , workerCount = active
410 , accountThread = delThread sv
411 , workerStopMVar = undefined
412 , svarRef = Nothing
413#ifdef DIAGNOSTICS
414 , svarCreator = tid
415 , aheadWorkQueue = undefined
416 , outputHeap = undefined
417#endif
418 , svarStats = SVarStats
419 { totalDispatches = disp
420 , maxWorkers = maxWrk
421 , maxOutQSize = maxOq
422 , maxHeapSize = maxHs
423 , maxWorkQSize = maxWq
424 , avgWorkerLatency = avgLat
425 , minWorkerLatency = minLat
426 , maxWorkerLatency = maxLat
427 , svarStopTime = stpTime
428 }
429 }
431 let sv =
432 case getStreamRate st of
433 Nothing ->
434 case getYieldLimit st of
435 Nothing -> getSVar sv readOutputQBounded
436 postProcessBounded
437 isWorkFinished
438 workLoopFIFO
439 Just _ -> getSVar sv readOutputQBounded
440 postProcessBounded
441 isWorkFinishedLimited
442 workLoopFIFOLimited
443 Just _ ->
444 case getYieldLimit st of
445 Nothing -> getSVar sv readOutputQPaced
446 postProcessPaced
447 isWorkFinished
448 workLoopFIFO
449 Just _ -> getSVar sv readOutputQPaced
450 postProcessPaced
451 isWorkFinishedLimited
452 workLoopFIFOLimited
453 in return sv
455{-# INLINABLE newAsyncVar #-}
456newAsyncVar :: MonadAsync m
457 => State Stream m a -> Stream m a -> m (SVar Stream m a)
458newAsyncVar st m = do
459 sv <- liftIO $ getLifoSVar st
460 sendFirstWorker sv m
462-- XXX Get rid of this?
463-- | Make a stream asynchronous, triggers the computation and returns a stream
464-- in the underlying monad representing the output generated by the original
465-- computation. The returned action is exhaustible and must be drained once. If
466-- not drained fully we may have a thread blocked forever and once exhausted it
467-- will always return 'empty'.
468--
469-- @since 0.2.0
470{-# INLINABLE mkAsync #-}
471mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
472mkAsync m = newAsyncVar defState (toStream m) >>= return . fromSVar
474{-# INLINABLE mkAsync' #-}
475mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)
476mkAsync' st m = newAsyncVar st (toStream m) >>= return . fromSVar
478-- | Create a new SVar and enqueue one stream computation on it.
479{-# INLINABLE newWAsyncVar #-}
480newWAsyncVar :: MonadAsync m
481 => State Stream m a -> Stream m a -> m (SVar Stream m a)
482newWAsyncVar st m = do
483 sv <- liftIO $ getFifoSVar st
484 sendFirstWorker sv m
486------------------------------------------------------------------------------
487-- Running streams concurrently
488------------------------------------------------------------------------------
490-- Concurrency rate control.
491--
492-- Our objective is to create more threads on demand if the consumer is running
493-- faster than us. As soon as we encounter a concurrent composition we create a
494-- push pull pair of threads. We use an SVar for communication between the
495-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
496-- The producer creates more threads if the SVar drains and becomes empty, that
497-- is the consumer is running faster.
498--
499-- XXX Note 1: This mechanism can be problematic if the initial production
500-- latency is high, we may end up creating too many threads. So we need some
501-- way to monitor and use the latency as well. Having a limit on the dispatches
502-- (programmer controlled) may also help.
503--
504-- TBD Note 2: We may want to run computations at the lower level of the
505-- composition tree serially even when they are composed using a parallel
506-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
507-- place of 'wAsync'. If we find that an SVar immediately above a computation
508-- gets drained empty we can switch to parallelizing the computation. For that
509-- we can use a state flag to fork the rest of the computation at any point of
510-- time inside the Monad bind operation if the consumer is running at a faster
511-- speed.
512--
513-- TBD Note 3: the binary operation ('parallel') composition allows us to
514-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
515-- chunksizes we will need to compose the parallel actions using a data
516-- constructor (A Free container) instead so that we can divide it in chunks of
517-- arbitrary size before dispatching. If the stream is composed of
518-- hierarchically composed grains of different sizes then we can always switch
519-- to a desired granularity depending on the consumer speed.
520--
521-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
522-- into just the number of CPUs.
524-- | Join two computations on the currently running 'SVar' queue for concurrent
525-- execution. When we are using parallel composition, an SVar is passed around
526-- as a state variable. We try to schedule a new parallel computation on the
527-- SVar passed to us. The first time, when no SVar exists, a new SVar is
528-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
529-- already scheduled on the SVar is further evaluated. For example, when (a
530-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
531-- the current scheduler queue.
532--
533-- The 'SVarStyle' required by the current composition context is passed as one
534-- of the parameters. If the scheduling and composition style of the new
535-- computation being scheduled is different than the style of the current SVar,
536-- then we create a new SVar and schedule it on that. The newly created SVar
537-- joins as one of the computations on the current SVar queue.
538--
539-- Cases when we need to switch to a new SVar:
540--
541-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
542-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
543-- because of the scheduling policy change.
544-- * if we 'adapt' a stream of type 'async' to a stream of type
545-- 'Parallel', we create a new SVar at the transitioning bind.
546-- * When the stream is switching from disjunctive composition to conjunctive
547-- composition and vice-versa we create a new SVar to isolate the scheduling
548-- of the two.
550forkSVarAsync :: MonadAsync m
551 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
552forkSVarAsync style m1 m2 = Stream $ \st stp sng yld -> do
553 sv <- case style of
554 AsyncVar -> newAsyncVar st (concurrently m1 m2)
555 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
556 _ -> error "illegal svar type"
557 unStream (fromSVar sv) (rstState st) stp sng yld
558 where
559 concurrently ma mb = Stream $ \st stp sng yld -> do
560 liftIO $ enqueue (fromJust $ streamVar st) mb
561 unStream ma st stp sng yld
563{-# INLINE joinStreamVarAsync #-}
564joinStreamVarAsync :: MonadAsync m
565 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
566joinStreamVarAsync style m1 m2 = Stream $ \st stp sng yld -> do
567 case streamVar st of
568 Just sv | svarStyle sv == style ->
569 liftIO (enqueue sv m2) >> unStream m1 st stp sng yld
570 _ -> unStream (forkSVarAsync style m1 m2) st stp sng yld
572------------------------------------------------------------------------------
573-- Semigroup and Monoid style compositions for parallel actions
574------------------------------------------------------------------------------
576{-# INLINE asyncS #-}
577asyncS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
578asyncS = joinStreamVarAsync AsyncVar
580-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
581-- Merges two streams possibly concurrently, preferring the
582-- elements from the left one when available.
583--
584-- @since 0.2.0
585{-# INLINE async #-}
586async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
587async m1 m2 = fromStream $ Stream $ \st stp sng yld ->
588 unStream (joinStreamVarAsync AsyncVar (toStream m1) (toStream m2))
589 st stp sng yld
591-- | Same as 'async'.
592--
593-- @since 0.1.0
594{-# DEPRECATED (<|) "Please use 'async' instead." #-}
595{-# INLINE (<|) #-}
596(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
597(<|) = async
599-- | XXX we can implement it more efficienty by directly implementing instead
600-- of combining streams using async.
601{-# INLINE consMAsync #-}
602consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
603consMAsync m r = K.yieldM m `asyncS` r
605------------------------------------------------------------------------------
606-- AsyncT
607------------------------------------------------------------------------------
609-- | Deep async composition or async composition with depth first traversal. In
610-- a left to right 'Semigroup' composition it tries to yield elements from the
611-- left stream as long as it can, but it can run the right stream in parallel
612-- if it needs to, based on demand. The right stream can be run if the left
613-- stream blocks on IO or cannot produce elements fast enough for the consumer.
614--
615-- @
616-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
617-- @
618-- @
619-- [1,2,3,4]
620-- @
621--
622-- Any exceptions generated by a constituent stream are propagated to the
623-- output stream. The output and exceptions from a single stream are guaranteed
624-- to arrive in the same order in the resulting stream as they were generated
625-- in the input stream. However, the relative ordering of elements from
626-- different streams in the resulting stream can vary depending on scheduling
627-- and generation delays.
628--
629-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
630-- concurrently based on demand. More concurrent iterations are started only
631-- if the previous iterations are not able to produce enough output for the
632-- consumer.
633--
634-- @
635-- import "Streamly"
636-- import qualified "Streamly.Prelude" as S
637-- import Control.Concurrent
638--
639-- main = 'runStream' . 'asyncly' $ do
640-- n <- return 3 \<\> return 2 \<\> return 1
641-- S.yieldM $ do
642-- threadDelay (n * 1000000)
643-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
644-- @
645-- @
646-- ThreadId 40: Delay 1
647-- ThreadId 39: Delay 2
648-- ThreadId 38: Delay 3
649-- @
650--
651-- All iterations may run in the same thread if they do not block.
652--
653-- Note that async composition with depth first traversal can be used to
654-- combine infinite number of streams as it explores only a bounded number of
655-- streams at a time.
656--
657-- @since 0.1.0
658newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
659 deriving (MonadTrans)
661-- | A demand driven left biased parallely composing IO stream of elements of
662-- type @a@. See 'AsyncT' documentation for more details.
663--
664-- @since 0.2.0
665type Async a = AsyncT IO a
667-- | Fix the type of a polymorphic stream as 'AsyncT'.
668--
669-- @since 0.1.0
670asyncly :: IsStream t => AsyncT m a -> t m a
671asyncly = adapt
673instance IsStream AsyncT where
674 toStream = getAsyncT
675 fromStream = AsyncT
677 {-# INLINE consM #-}
678 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
679 consM m r = fromStream $ consMAsync m (toStream r)
681 {-# INLINE (|:) #-}
682 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
683 (|:) = consM
685------------------------------------------------------------------------------
686-- Semigroup
687------------------------------------------------------------------------------
689instance MonadAsync m => Semigroup (AsyncT m a) where
690 (<>) = async
692------------------------------------------------------------------------------
693-- Monoid
694------------------------------------------------------------------------------
696instance MonadAsync m => Monoid (AsyncT m a) where
697 mempty = K.nil
698 mappend = (<>)
700------------------------------------------------------------------------------
701-- Monad
702------------------------------------------------------------------------------
704instance MonadAsync m => Monad (AsyncT m) where
705 return = pure
706 (AsyncT m) >>= f = AsyncT $ K.bindWith asyncS m (getAsyncT . f)
708------------------------------------------------------------------------------
709-- Other instances
710------------------------------------------------------------------------------
712MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
713MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
715------------------------------------------------------------------------------
716-- WAsyncT
717------------------------------------------------------------------------------
719{-# INLINE wAsyncS #-}
720wAsyncS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
721wAsyncS = joinStreamVarAsync WAsyncVar
723-- | XXX we can implement it more efficienty by directly implementing instead
724-- of combining streams using wAsync.
725{-# INLINE consMWAsync #-}
726consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
727consMWAsync m r = K.yieldM m `wAsyncS` r
729-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
730-- Merges two streams concurrently choosing elements from both fairly.
731--
732-- @since 0.2.0
733{-# INLINE wAsync #-}
734wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
735wAsync m1 m2 = fromStream $ Stream $ \st stp sng yld ->
736 unStream (wAsyncS (toStream m1) (toStream m2)) st stp sng yld
738-- | Wide async composition or async composition with breadth first traversal.
739-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
740-- streams using a depth first travesal or in a round robin fashion, yielding
741-- elements from both streams alternately.
742--
743-- @
744-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
745-- @
746-- @
747-- [1,3,2,4]
748-- @
749--
750-- Any exceptions generated by a constituent stream are propagated to the
751-- output stream. The output and exceptions from a single stream are guaranteed
752-- to arrive in the same order in the resulting stream as they were generated
753-- in the input stream. However, the relative ordering of elements from
754-- different streams in the resulting stream can vary depending on scheduling
755-- and generation delays.
756--
757-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
758-- concurrently using a round robin scheduling.
759--
760-- @
761-- import "Streamly"
762-- import qualified "Streamly.Prelude" as S
763-- import Control.Concurrent
764--
765-- main = 'runStream' . 'wAsyncly' $ do
766-- n <- return 3 \<\> return 2 \<\> return 1
767-- S.yieldM $ do
768-- threadDelay (n * 1000000)
769-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
770-- @
771-- @
772-- ThreadId 40: Delay 1
773-- ThreadId 39: Delay 2
774-- ThreadId 38: Delay 3
775-- @
776--
777-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
778-- concurrently, unconditionally.
779--
780-- Note that async composition with breadth first traversal can only combine a
781-- finite number of streams as it needs to retain state for each unfinished
782-- stream.
783--
784-- @since 0.2.0
785newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
786 deriving (MonadTrans)
788-- | A round robin parallely composing IO stream of elements of type @a@.
789-- See 'WAsyncT' documentation for more details.
790--
791-- @since 0.2.0
792type WAsync a = WAsyncT IO a
794-- | Fix the type of a polymorphic stream as 'WAsyncT'.
795--
796-- @since 0.2.0
797wAsyncly :: IsStream t => WAsyncT m a -> t m a
798wAsyncly = adapt
800instance IsStream WAsyncT where
801 toStream = getWAsyncT
802 fromStream = WAsyncT
804 {-# INLINE consM #-}
805 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
806 consM m r = fromStream $ consMWAsync m (toStream r)
808 {-# INLINE (|:) #-}
809 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
810 (|:) = consM
812------------------------------------------------------------------------------
813-- Semigroup
814------------------------------------------------------------------------------
816instance MonadAsync m => Semigroup (WAsyncT m a) where
817 (<>) = wAsync
819------------------------------------------------------------------------------
820-- Monoid
821------------------------------------------------------------------------------
823instance MonadAsync m => Monoid (WAsyncT m a) where
824 mempty = K.nil
825 mappend = (<>)
827------------------------------------------------------------------------------
828-- Monad
829------------------------------------------------------------------------------
831instance MonadAsync m => Monad (WAsyncT m) where
832 return = pure
833 (WAsyncT m) >>= f =
834 WAsyncT $ K.bindWith wAsyncS m (getWAsyncT . f)
836------------------------------------------------------------------------------
837-- Other instances
838------------------------------------------------------------------------------
840MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
841MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)