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