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