git.iko.soy/streamly
on 9268208 help sign in
streamly/src/Streamly/Streams/Async.hs
844 lines 32 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.Atomics (atomicModifyIORefCAS)
58import Streamly.Streams.SVar (fromSVar)
59import Streamly.Streams.Serial (map)
60import Streamly.SVar
61import Streamly.Streams.StreamK
62 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared,
63 foldStreamSVar)
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 -> Maybe 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 -> foldStreamSVar sv st yieldk single run m
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 foldStreamSVar sv st yieldk single run r
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 -> Maybe 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 let stop = liftIO (incrementYieldLimit sv) >> run
138 foldStreamSVar sv st yieldk single stop m
139 -- Avoid any side effects, undo the yield limit decrement if we
140 -- never yielded anything.
141 else liftIO $ do
142 enqueueLIFO sv q m
143 incrementYieldLimit sv
144 sendStop sv winfo
146 single a = do
147 res <- liftIO $ sendYield sv winfo (ChildYield a)
148 if res then run else liftIO $ sendStop sv winfo
150 -- XXX can we pass on the yield limit downstream to limit the concurrency
151 -- of constituent streams.
152 yieldk a r = do
153 res <- liftIO $ sendYield sv winfo (ChildYield a)
154 yieldLimitOk <- liftIO $ decrementYieldLimit sv
155 let stop = liftIO (incrementYieldLimit sv) >> run
156 if res && yieldLimitOk
157 then foldStreamSVar sv st yieldk single stop r
158 else liftIO $ do
159 incrementYieldLimit sv
160 enqueueLIFO sv q r
161 sendStop sv winfo
163 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
164 [] -> ([], Nothing)
165 x : xs -> (xs, Just x)
167-------------------------------------------------------------------------------
168-- WAsync
169-------------------------------------------------------------------------------
171-- XXX we can remove sv as it is derivable from st
173{-# INLINE workLoopFIFO #-}
174workLoopFIFO
175 :: MonadIO m
176 => LinkedQueue (Stream m a)
177 -> State Stream m a
178 -> SVar Stream m a
179 -> Maybe 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 -> foldStreamSVar sv st yieldk single run m
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 foldStreamSVar sv st yieldk single run r
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 -> Maybe 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 foldStreamSVar sv st yieldk single stop m
225 else liftIO $ do
226 enqueueFIFO sv q m
227 incrementYieldLimit sv
228 sendStop sv winfo
230 single a = do
231 res <- liftIO $ sendYield sv winfo (ChildYield a)
232 if res then run else liftIO $ sendStop sv winfo
234 yieldk a r = do
235 res <- liftIO $ sendYield sv winfo (ChildYield a)
236 yieldLimitOk <- liftIO $ decrementYieldLimit sv
237 let stop = liftIO (incrementYieldLimit sv) >> run
238 if res && yieldLimitOk
239 then foldStreamSVar sv st yieldk single stop r
240 else liftIO $ do
241 incrementYieldLimit sv
242 enqueueFIFO sv q r
243 sendStop sv winfo
245-------------------------------------------------------------------------------
246-- SVar creation
247-- This code belongs in SVar.hs but is kept here for perf reasons
248-------------------------------------------------------------------------------
250-- XXX we have this function in this file because passing runStreamLIFO as a
251-- function argument to this function results in a perf degradation of more
252-- than 10%. Need to investigate what the root cause is.
253-- Interestingly, the same thing does not make any difference for Ahead.
254getLifoSVar :: forall m a. MonadAsync m
255 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
256getLifoSVar st mrun = do
257 outQ <- newIORef ([], 0)
258 outQMv <- newEmptyMVar
259 active <- newIORef 0
260 wfw <- newIORef False
261 running <- newIORef S.empty
262 q <- newIORef []
263 yl <- case getYieldLimit st of
264 Nothing -> return Nothing
265 Just x -> Just <$> newIORef x
266 rateInfo <- getYieldRateInfo st
268 stats <- newSVarStats
269 tid <- myThreadId
271 let isWorkFinished _ = null <$> readIORef q
273 let isWorkFinishedLimited sv = do
274 yieldsDone <-
275 case remainingWork sv of
276 Just ref -> do
277 n <- readIORef ref
278 return (n <= 0)
279 Nothing -> return False
280 qEmpty <- null <$> readIORef q
281 return $ qEmpty || yieldsDone
283 let getSVar :: SVar Stream m a
284 -> (SVar Stream m a -> m [ChildEvent a])
285 -> (SVar Stream m a -> m Bool)
286 -> (SVar Stream m a -> IO Bool)
287 -> (IORef [Stream m a]
288 -> State Stream m a
289 -> SVar Stream m a
290 -> Maybe WorkerInfo
291 -> m())
292 -> SVar Stream m a
293 getSVar sv readOutput postProc workDone wloop = SVar
294 { outputQueue = outQ
295 , remainingWork = yl
296 , maxBufferLimit = getMaxBuffer st
297 , maxWorkerLimit = getMaxThreads st
298 , yieldRateInfo = rateInfo
299 , outputDoorBell = outQMv
300 , readOutputQ = readOutput sv
301 , postProcess = postProc sv
302 , workerThreads = running
303 , workLoop = wloop q st{streamVar = Just sv} sv
304 , enqueue = enqueueLIFO sv q
305 , isWorkDone = workDone sv
306 , isQueueDone = workDone sv
307 , needDoorBell = wfw
308 , svarStyle = AsyncVar
309 , svarMrun = mrun
310 , workerCount = active
311 , accountThread = delThread sv
312 , workerStopMVar = undefined
313 , svarRef = Nothing
314 , svarInspectMode = getInspectMode st
315 , svarCreator = tid
316 , aheadWorkQueue = undefined
317 , outputHeap = undefined
318 , svarStats = stats
319 }
321 let sv =
322 case getStreamRate st of
323 Nothing ->
324 case getYieldLimit st of
325 Nothing -> getSVar sv readOutputQBounded
326 postProcessBounded
327 isWorkFinished
328 workLoopLIFO
329 Just _ -> getSVar sv readOutputQBounded
330 postProcessBounded
331 isWorkFinishedLimited
332 workLoopLIFOLimited
333 Just _ ->
334 case getYieldLimit st of
335 Nothing -> getSVar sv readOutputQPaced
336 postProcessPaced
337 isWorkFinished
338 workLoopLIFO
339 Just _ -> getSVar sv readOutputQPaced
340 postProcessPaced
341 isWorkFinishedLimited
342 workLoopLIFOLimited
343 in return sv
345getFifoSVar :: forall m a. MonadAsync m
346 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
347getFifoSVar st mrun = do
348 outQ <- newIORef ([], 0)
349 outQMv <- newEmptyMVar
350 active <- newIORef 0
351 wfw <- newIORef False
352 running <- newIORef S.empty
353 q <- newQ
354 yl <- case getYieldLimit st of
355 Nothing -> return Nothing
356 Just x -> Just <$> newIORef x
357 rateInfo <- getYieldRateInfo st
359 stats <- newSVarStats
360 tid <- myThreadId
362 let isWorkFinished _ = nullQ q
363 let isWorkFinishedLimited sv = do
364 yieldsDone <-
365 case remainingWork sv of
366 Just ref -> do
367 n <- readIORef ref
368 return (n <= 0)
369 Nothing -> return False
370 qEmpty <- nullQ q
371 return $ qEmpty || yieldsDone
373 let getSVar :: SVar Stream m a
374 -> (SVar Stream m a -> m [ChildEvent a])
375 -> (SVar Stream m a -> m Bool)
376 -> (SVar Stream m a -> IO Bool)
377 -> (LinkedQueue (Stream m a)
378 -> State Stream m a
379 -> SVar Stream m a
380 -> Maybe WorkerInfo
381 -> m())
382 -> SVar Stream m a
383 getSVar sv readOutput postProc workDone wloop = SVar
384 { outputQueue = outQ
385 , remainingWork = yl
386 , maxBufferLimit = getMaxBuffer st
387 , maxWorkerLimit = getMaxThreads st
388 , yieldRateInfo = rateInfo
389 , outputDoorBell = outQMv
390 , readOutputQ = readOutput sv
391 , postProcess = postProc sv
392 , workerThreads = running
393 , workLoop = wloop q st{streamVar = Just sv} sv
394 , enqueue = enqueueFIFO sv q
395 , isWorkDone = workDone sv
396 , isQueueDone = workDone sv
397 , needDoorBell = wfw
398 , svarStyle = WAsyncVar
399 , svarMrun = mrun
400 , workerCount = active
401 , accountThread = delThread sv
402 , workerStopMVar = undefined
403 , svarRef = Nothing
404 , svarInspectMode = getInspectMode st
405 , svarCreator = tid
406 , aheadWorkQueue = undefined
407 , outputHeap = undefined
408 , svarStats = stats
409 }
411 let sv =
412 case getStreamRate st of
413 Nothing ->
414 case getYieldLimit st of
415 Nothing -> getSVar sv readOutputQBounded
416 postProcessBounded
417 isWorkFinished
418 workLoopFIFO
419 Just _ -> getSVar sv readOutputQBounded
420 postProcessBounded
421 isWorkFinishedLimited
422 workLoopFIFOLimited
423 Just _ ->
424 case getYieldLimit st of
425 Nothing -> getSVar sv readOutputQPaced
426 postProcessPaced
427 isWorkFinished
428 workLoopFIFO
429 Just _ -> getSVar sv readOutputQPaced
430 postProcessPaced
431 isWorkFinishedLimited
432 workLoopFIFOLimited
433 in return sv
435{-# INLINABLE newAsyncVar #-}
436newAsyncVar :: MonadAsync m
437 => State Stream m a -> Stream m a -> m (SVar Stream m a)
438newAsyncVar st m = do
439 mrun <- captureMonadState
440 sv <- liftIO $ getLifoSVar st mrun
441 sendFirstWorker sv m
443-- XXX Get rid of this?
444-- | Make a stream asynchronous, triggers the computation and returns a stream
445-- in the underlying monad representing the output generated by the original
446-- computation. The returned action is exhaustible and must be drained once. If
447-- not drained fully we may have a thread blocked forever and once exhausted it
448-- will always return 'empty'.
449--
450-- @since 0.2.0
451{-# INLINABLE mkAsync #-}
452mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
453mkAsync m = fmap fromSVar (newAsyncVar defState (toStream m))
455{-# INLINABLE mkAsync' #-}
456mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)
457mkAsync' st m = fmap fromSVar (newAsyncVar st (toStream m))
459-- | Create a new SVar and enqueue one stream computation on it.
460{-# INLINABLE newWAsyncVar #-}
461newWAsyncVar :: MonadAsync m
462 => State Stream m a -> Stream m a -> m (SVar Stream m a)
463newWAsyncVar st m = do
464 mrun <- captureMonadState
465 sv <- liftIO $ getFifoSVar st mrun
466 sendFirstWorker sv m
468------------------------------------------------------------------------------
469-- Running streams concurrently
470------------------------------------------------------------------------------
472-- Concurrency rate control.
473--
474-- Our objective is to create more threads on demand if the consumer is running
475-- faster than us. As soon as we encounter a concurrent composition we create a
476-- push pull pair of threads. We use an SVar for communication between the
477-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
478-- The producer creates more threads if the SVar drains and becomes empty, that
479-- is the consumer is running faster.
480--
481-- XXX Note 1: This mechanism can be problematic if the initial production
482-- latency is high, we may end up creating too many threads. So we need some
483-- way to monitor and use the latency as well. Having a limit on the dispatches
484-- (programmer controlled) may also help.
485--
486-- TBD Note 2: We may want to run computations at the lower level of the
487-- composition tree serially even when they are composed using a parallel
488-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
489-- place of 'wAsync'. If we find that an SVar immediately above a computation
490-- gets drained empty we can switch to parallelizing the computation. For that
491-- we can use a state flag to fork the rest of the computation at any point of
492-- time inside the Monad bind operation if the consumer is running at a faster
493-- speed.
494--
495-- TBD Note 3: the binary operation ('parallel') composition allows us to
496-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
497-- chunksizes we will need to compose the parallel actions using a data
498-- constructor (A Free container) instead so that we can divide it in chunks of
499-- arbitrary size before dispatching. If the stream is composed of
500-- hierarchically composed grains of different sizes then we can always switch
501-- to a desired granularity depending on the consumer speed.
502--
503-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
504-- into just the number of CPUs.
506-- | Join two computations on the currently running 'SVar' queue for concurrent
507-- execution. When we are using parallel composition, an SVar is passed around
508-- as a state variable. We try to schedule a new parallel computation on the
509-- SVar passed to us. The first time, when no SVar exists, a new SVar is
510-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
511-- already scheduled on the SVar is further evaluated. For example, when (a
512-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
513-- the current scheduler queue.
514--
515-- The 'SVarStyle' required by the current composition context is passed as one
516-- of the parameters. If the scheduling and composition style of the new
517-- computation being scheduled is different than the style of the current SVar,
518-- then we create a new SVar and schedule it on that. The newly created SVar
519-- joins as one of the computations on the current SVar queue.
520--
521-- Cases when we need to switch to a new SVar:
522--
523-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
524-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
525-- because of the scheduling policy change.
526-- * if we 'adapt' a stream of type 'async' to a stream of type
527-- 'Parallel', we create a new SVar at the transitioning bind.
528-- * When the stream is switching from disjunctive composition to conjunctive
529-- composition and vice-versa we create a new SVar to isolate the scheduling
530-- of the two.
532forkSVarAsync :: (IsStream t, MonadAsync m)
533 => SVarStyle -> t m a -> t m a -> t m a
534forkSVarAsync style m1 m2 = mkStream $ \st stp sng yld -> do
535 sv <- case style of
536 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
537 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
538 _ -> error "illegal svar type"
539 foldStream st stp sng yld $ fromSVar sv
540 where
541 concurrently ma mb = mkStream $ \st stp sng yld -> do
542 liftIO $ enqueue (fromJust $ streamVar st) mb
543 foldStreamShared st stp sng yld ma
545{-# INLINE joinStreamVarAsync #-}
546joinStreamVarAsync :: (IsStream t, MonadAsync m)
547 => SVarStyle -> t m a -> t m a -> t m a
548joinStreamVarAsync style m1 m2 = mkStream $ \st stp sng yld ->
549 case streamVar st of
550 Just sv | svarStyle sv == style -> do
551 liftIO $ enqueue sv (toStream m2)
552 foldStreamShared st stp sng yld m1
553 _ -> foldStreamShared st stp sng yld (forkSVarAsync style m1 m2)
555------------------------------------------------------------------------------
556-- Semigroup and Monoid style compositions for parallel actions
557------------------------------------------------------------------------------
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 = joinStreamVarAsync AsyncVar
568-- | Same as 'async'.
569--
570-- @since 0.1.0
571{-# DEPRECATED (<|) "Please use 'async' instead." #-}
572{-# INLINE (<|) #-}
573(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
574(<|) = async
576-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
577-- huge difference in the performance of consM in IsStream instance even we
578-- have a SPECIALIZE in the instance.
579--
580-- | XXX we can implement it more efficienty by directly implementing instead
581-- of combining streams using async.
582{-# INLINE consMAsync #-}
583{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
584consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
585consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
587------------------------------------------------------------------------------
588-- AsyncT
589------------------------------------------------------------------------------
591-- | Deep async composition or async composition with depth first traversal. In
592-- a left to right 'Semigroup' composition it tries to yield elements from the
593-- left stream as long as it can, but it can run the right stream in parallel
594-- if it needs to, based on demand. The right stream can be run if the left
595-- stream blocks on IO or cannot produce elements fast enough for the consumer.
596--
597-- @
598-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
599-- @
600-- @
601-- [1,2,3,4]
602-- @
603--
604-- Any exceptions generated by a constituent stream are propagated to the
605-- output stream. The output and exceptions from a single stream are guaranteed
606-- to arrive in the same order in the resulting stream as they were generated
607-- in the input stream. However, the relative ordering of elements from
608-- different streams in the resulting stream can vary depending on scheduling
609-- and generation delays.
610--
611-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
612-- concurrently based on demand. More concurrent iterations are started only
613-- if the previous iterations are not able to produce enough output for the
614-- consumer.
615--
616-- @
617-- import "Streamly"
618-- import qualified "Streamly.Prelude" as S
619-- import Control.Concurrent
620--
621-- main = 'drain' . 'asyncly' $ do
622-- n <- return 3 \<\> return 2 \<\> return 1
623-- S.yieldM $ do
624-- threadDelay (n * 1000000)
625-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
626-- @
627-- @
628-- ThreadId 40: Delay 1
629-- ThreadId 39: Delay 2
630-- ThreadId 38: Delay 3
631-- @
632--
633-- All iterations may run in the same thread if they do not block.
634--
635-- Note that async composition with depth first traversal can be used to
636-- combine infinite number of streams as it explores only a bounded number of
637-- streams at a time.
638--
639-- @since 0.1.0
640newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
641 deriving (MonadTrans)
643-- | A demand driven left biased parallely composing IO stream of elements of
644-- type @a@. See 'AsyncT' documentation for more details.
645--
646-- @since 0.2.0
647type Async = AsyncT IO
649-- | Fix the type of a polymorphic stream as 'AsyncT'.
650--
651-- @since 0.1.0
652asyncly :: IsStream t => AsyncT m a -> t m a
653asyncly = adapt
655instance IsStream AsyncT where
656 toStream = getAsyncT
657 fromStream = AsyncT
658 consM = consMAsync
659 (|:) = consMAsync
661------------------------------------------------------------------------------
662-- Semigroup
663------------------------------------------------------------------------------
665-- Monomorphically typed version of "async" for better performance of Semigroup
666-- instance.
667{-# INLINE mappendAsync #-}
668{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
669mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
670mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
672instance MonadAsync m => Semigroup (AsyncT m a) where
673 (<>) = mappendAsync
675------------------------------------------------------------------------------
676-- Monoid
677------------------------------------------------------------------------------
679instance MonadAsync m => Monoid (AsyncT m a) where
680 mempty = K.nil
681 mappend = (<>)
683------------------------------------------------------------------------------
684-- Monad
685------------------------------------------------------------------------------
687{-# INLINE bindAsync #-}
688{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
689bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
690bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
692instance MonadAsync m => Monad (AsyncT m) where
693 return = pure
694 (>>=) = bindAsync
696{-# INLINE apAsync #-}
697{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
698apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
699apAsync mf m = ap (adapt mf) (adapt m)
701instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
702 pure = AsyncT . K.yield
703 (<*>) = apAsync
705------------------------------------------------------------------------------
706-- Other instances
707------------------------------------------------------------------------------
709MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
711------------------------------------------------------------------------------
712-- WAsyncT
713------------------------------------------------------------------------------
715-- | XXX we can implement it more efficienty by directly implementing instead
716-- of combining streams using wAsync.
717{-# INLINE consMWAsync #-}
718{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
719consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
720consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
722-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
723-- Merges two streams concurrently choosing elements from both fairly.
724--
725-- @since 0.2.0
726{-# INLINE wAsync #-}
727wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
728wAsync = joinStreamVarAsync WAsyncVar
730-- | Wide async composition or async composition with breadth first traversal.
731-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
732-- streams using a depth first travesal or in a round robin fashion, yielding
733-- elements from both streams alternately.
734--
735-- @
736-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
737-- @
738-- @
739-- [1,3,2,4]
740-- @
741--
742-- Any exceptions generated by a constituent stream are propagated to the
743-- output stream. The output and exceptions from a single stream are guaranteed
744-- to arrive in the same order in the resulting stream as they were generated
745-- in the input stream. However, the relative ordering of elements from
746-- different streams in the resulting stream can vary depending on scheduling
747-- and generation delays.
748--
749-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
750-- concurrently using a round robin scheduling.
751--
752-- @
753-- import "Streamly"
754-- import qualified "Streamly.Prelude" as S
755-- import Control.Concurrent
756--
757-- main = 'drain' . 'wAsyncly' $ do
758-- n <- return 3 \<\> return 2 \<\> return 1
759-- S.yieldM $ do
760-- threadDelay (n * 1000000)
761-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
762-- @
763-- @
764-- ThreadId 40: Delay 1
765-- ThreadId 39: Delay 2
766-- ThreadId 38: Delay 3
767-- @
768--
769-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
770-- concurrently, unconditionally.
771--
772-- Note that async composition with breadth first traversal can only combine a
773-- finite number of streams as it needs to retain state for each unfinished
774-- stream.
775--
776-- @since 0.2.0
777newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
778 deriving (MonadTrans)
780-- | A round robin parallely composing IO stream of elements of type @a@.
781-- See 'WAsyncT' documentation for more details.
782--
783-- @since 0.2.0
784type WAsync = WAsyncT IO
786-- | Fix the type of a polymorphic stream as 'WAsyncT'.
787--
788-- @since 0.2.0
789wAsyncly :: IsStream t => WAsyncT m a -> t m a
790wAsyncly = adapt
792instance IsStream WAsyncT where
793 toStream = getWAsyncT
794 fromStream = WAsyncT
795 consM = consMWAsync
796 (|:) = consMWAsync
798------------------------------------------------------------------------------
799-- Semigroup
800------------------------------------------------------------------------------
802{-# INLINE mappendWAsync #-}
803{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
804mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
805mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
807instance MonadAsync m => Semigroup (WAsyncT m a) where
808 (<>) = mappendWAsync
810------------------------------------------------------------------------------
811-- Monoid
812------------------------------------------------------------------------------
814instance MonadAsync m => Monoid (WAsyncT m a) where
815 mempty = K.nil
816 mappend = (<>)
818------------------------------------------------------------------------------
819-- Monad
820------------------------------------------------------------------------------
822{-# INLINE bindWAsync #-}
823{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
824bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
825bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
827instance MonadAsync m => Monad (WAsyncT m) where
828 return = pure
829 (>>=) = bindWAsync
831{-# INLINE apWAsync #-}
832{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
833apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
834apWAsync mf m = ap (adapt mf) (adapt m)
836instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
837 pure = WAsyncT . K.yield
838 (<*>) = apWAsync
840------------------------------------------------------------------------------
841-- Other instances
842------------------------------------------------------------------------------
844MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)