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