git.iko.soy/streamly
on 3ce574a help sign in
streamly/src/Streamly/Streams/Async.hs
856 lines 33 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 , svarStopStyle = StopNone
310 , svarStopBy = undefined
311 , svarMrun = mrun
312 , workerCount = active
313 , accountThread = delThread sv
314 , workerStopMVar = undefined
315 , svarRef = Nothing
316 , svarInspectMode = getInspectMode st
317 , svarCreator = tid
318 , aheadWorkQueue = undefined
319 , outputHeap = undefined
320 , svarStats = stats
321 }
323 let sv =
324 case getStreamRate st of
325 Nothing ->
326 case getYieldLimit st of
327 Nothing -> getSVar sv readOutputQBounded
328 postProcessBounded
329 isWorkFinished
330 workLoopLIFO
331 Just _ -> getSVar sv readOutputQBounded
332 postProcessBounded
333 isWorkFinishedLimited
334 workLoopLIFOLimited
335 Just _ ->
336 case getYieldLimit st of
337 Nothing -> getSVar sv readOutputQPaced
338 postProcessPaced
339 isWorkFinished
340 workLoopLIFO
341 Just _ -> getSVar sv readOutputQPaced
342 postProcessPaced
343 isWorkFinishedLimited
344 workLoopLIFOLimited
345 in return sv
347getFifoSVar :: forall m a. MonadAsync m
348 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
349getFifoSVar st mrun = 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 stats <- newSVarStats
362 tid <- myThreadId
364 let isWorkFinished _ = nullQ q
365 let isWorkFinishedLimited sv = do
366 yieldsDone <-
367 case remainingWork sv of
368 Just ref -> do
369 n <- readIORef ref
370 return (n <= 0)
371 Nothing -> return False
372 qEmpty <- nullQ q
373 return $ qEmpty || yieldsDone
375 let getSVar :: SVar Stream m a
376 -> (SVar Stream m a -> m [ChildEvent a])
377 -> (SVar Stream m a -> m Bool)
378 -> (SVar Stream m a -> IO Bool)
379 -> (LinkedQueue (Stream m a)
380 -> State Stream m a
381 -> SVar Stream m a
382 -> Maybe WorkerInfo
383 -> m())
384 -> SVar Stream m a
385 getSVar sv readOutput postProc workDone wloop = SVar
386 { outputQueue = outQ
387 , remainingWork = yl
388 , maxBufferLimit = getMaxBuffer st
389 , maxWorkerLimit = getMaxThreads st
390 , yieldRateInfo = rateInfo
391 , outputDoorBell = outQMv
392 , readOutputQ = readOutput sv
393 , postProcess = postProc sv
394 , workerThreads = running
395 , workLoop = wloop q st{streamVar = Just sv} sv
396 , enqueue = enqueueFIFO sv q
397 , isWorkDone = workDone sv
398 , isQueueDone = workDone sv
399 , needDoorBell = wfw
400 , svarStyle = WAsyncVar
401 , svarStopStyle = StopNone
402 , svarStopBy = undefined
403 , svarMrun = mrun
404 , workerCount = active
405 , accountThread = delThread sv
406 , workerStopMVar = undefined
407 , svarRef = Nothing
408 , svarInspectMode = getInspectMode st
409 , svarCreator = tid
410 , aheadWorkQueue = undefined
411 , outputHeap = undefined
412 , svarStats = stats
413 }
415 let sv =
416 case getStreamRate st of
417 Nothing ->
418 case getYieldLimit st of
419 Nothing -> getSVar sv readOutputQBounded
420 postProcessBounded
421 isWorkFinished
422 workLoopFIFO
423 Just _ -> getSVar sv readOutputQBounded
424 postProcessBounded
425 isWorkFinishedLimited
426 workLoopFIFOLimited
427 Just _ ->
428 case getYieldLimit st of
429 Nothing -> getSVar sv readOutputQPaced
430 postProcessPaced
431 isWorkFinished
432 workLoopFIFO
433 Just _ -> getSVar sv readOutputQPaced
434 postProcessPaced
435 isWorkFinishedLimited
436 workLoopFIFOLimited
437 in return sv
439{-# INLINABLE newAsyncVar #-}
440newAsyncVar :: MonadAsync m
441 => State Stream m a -> Stream m a -> m (SVar Stream m a)
442newAsyncVar st m = do
443 mrun <- captureMonadState
444 sv <- liftIO $ getLifoSVar st mrun
445 sendFirstWorker sv m
447-- XXX Get rid of this?
448-- | Make a stream asynchronous, triggers the computation and returns a stream
449-- in the underlying monad representing the output generated by the original
450-- computation. The returned action is exhaustible and must be drained once. If
451-- not drained fully we may have a thread blocked forever and once exhausted it
452-- will always return 'empty'.
453--
454-- @since 0.2.0
455{-# INLINABLE mkAsync #-}
456mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
457mkAsync m = fmap fromSVar (newAsyncVar defState (toStream m))
459{-# INLINABLE mkAsync' #-}
460mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)
461mkAsync' st m = fmap fromSVar (newAsyncVar st (toStream m))
463-- | Create a new SVar and enqueue one stream computation on it.
464{-# INLINABLE newWAsyncVar #-}
465newWAsyncVar :: MonadAsync m
466 => State Stream m a -> Stream m a -> m (SVar Stream m a)
467newWAsyncVar st m = do
468 mrun <- captureMonadState
469 sv <- liftIO $ getFifoSVar st mrun
470 sendFirstWorker sv m
472------------------------------------------------------------------------------
473-- Running streams concurrently
474------------------------------------------------------------------------------
476-- Concurrency rate control.
477--
478-- Our objective is to create more threads on demand if the consumer is running
479-- faster than us. As soon as we encounter a concurrent composition we create a
480-- push pull pair of threads. We use an SVar for communication between the
481-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
482-- The producer creates more threads if the SVar drains and becomes empty, that
483-- is the consumer is running faster.
484--
485-- XXX Note 1: This mechanism can be problematic if the initial production
486-- latency is high, we may end up creating too many threads. So we need some
487-- way to monitor and use the latency as well. Having a limit on the dispatches
488-- (programmer controlled) may also help.
489--
490-- TBD Note 2: We may want to run computations at the lower level of the
491-- composition tree serially even when they are composed using a parallel
492-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
493-- place of 'wAsync'. If we find that an SVar immediately above a computation
494-- gets drained empty we can switch to parallelizing the computation. For that
495-- we can use a state flag to fork the rest of the computation at any point of
496-- time inside the Monad bind operation if the consumer is running at a faster
497-- speed.
498--
499-- TBD Note 3: the binary operation ('parallel') composition allows us to
500-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
501-- chunksizes we will need to compose the parallel actions using a data
502-- constructor (A Free container) instead so that we can divide it in chunks of
503-- arbitrary size before dispatching. If the stream is composed of
504-- hierarchically composed grains of different sizes then we can always switch
505-- to a desired granularity depending on the consumer speed.
506--
507-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
508-- into just the number of CPUs.
510-- | Join two computations on the currently running 'SVar' queue for concurrent
511-- execution. When we are using parallel composition, an SVar is passed around
512-- as a state variable. We try to schedule a new parallel computation on the
513-- SVar passed to us. The first time, when no SVar exists, a new SVar is
514-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
515-- already scheduled on the SVar is further evaluated. For example, when (a
516-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
517-- the current scheduler queue.
518--
519-- The 'SVarStyle' required by the current composition context is passed as one
520-- of the parameters. If the scheduling and composition style of the new
521-- computation being scheduled is different than the style of the current SVar,
522-- then we create a new SVar and schedule it on that. The newly created SVar
523-- joins as one of the computations on the current SVar queue.
524--
525-- Cases when we need to switch to a new SVar:
526--
527-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
528-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
529-- because of the scheduling policy change.
530-- * if we 'adapt' a stream of type 'async' to a stream of type
531-- 'Parallel', we create a new SVar at the transitioning bind.
532-- * When the stream is switching from disjunctive composition to conjunctive
533-- composition and vice-versa we create a new SVar to isolate the scheduling
534-- of the two.
536forkSVarAsync :: (IsStream t, MonadAsync m)
537 => SVarStyle -> t m a -> t m a -> t m a
538forkSVarAsync style m1 m2 = mkStream $ \st stp sng yld -> do
539 sv <- case style of
540 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
541 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
542 _ -> error "illegal svar type"
543 foldStream st stp sng yld $ fromSVar sv
544 where
545 concurrently ma mb = mkStream $ \st stp sng yld -> do
546 liftIO $ enqueue (fromJust $ streamVar st) mb
547 foldStreamShared st stp sng yld ma
549{-# INLINE joinStreamVarAsync #-}
550joinStreamVarAsync :: (IsStream t, MonadAsync m)
551 => SVarStyle -> t m a -> t m a -> t m a
552joinStreamVarAsync style m1 m2 = mkStream $ \st stp sng yld ->
553 case streamVar st of
554 Just sv | svarStyle sv == style -> do
555 liftIO $ enqueue sv (toStream m2)
556 foldStreamShared st stp sng yld m1
557 _ -> foldStreamShared st stp sng yld (forkSVarAsync style m1 m2)
559------------------------------------------------------------------------------
560-- Semigroup and Monoid style compositions for parallel actions
561------------------------------------------------------------------------------
563-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
564-- Merges two streams possibly concurrently, preferring the
565-- elements from the left one when available.
566--
567-- @since 0.2.0
568{-# INLINE async #-}
569async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
570async = joinStreamVarAsync AsyncVar
572-- | Same as 'async'.
573--
574-- @since 0.1.0
575{-# DEPRECATED (<|) "Please use 'async' instead." #-}
576{-# INLINE (<|) #-}
577(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
578(<|) = async
580-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
581-- huge difference in the performance of consM in IsStream instance even we
582-- have a SPECIALIZE in the instance.
583--
584-- | XXX we can implement it more efficienty by directly implementing instead
585-- of combining streams using async.
586{-# INLINE consMAsync #-}
587{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
588consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
589consMAsync m r = fromStream $ K.yieldM m `async` (toStream 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 = 'drain' . 'asyncly' $ do
626-- n <- return 3 \<\> return 2 \<\> return 1
627-- S.yieldM $ 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 = AsyncT IO
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
662 consM = consMAsync
663 (|:) = consMAsync
665------------------------------------------------------------------------------
666-- Semigroup
667------------------------------------------------------------------------------
669-- Monomorphically typed version of "async" for better performance of Semigroup
670-- instance.
671{-# INLINE mappendAsync #-}
672{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
673mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
674mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
676instance MonadAsync m => Semigroup (AsyncT m a) where
677 (<>) = mappendAsync
679------------------------------------------------------------------------------
680-- Monoid
681------------------------------------------------------------------------------
683instance MonadAsync m => Monoid (AsyncT m a) where
684 mempty = K.nil
685 mappend = (<>)
687------------------------------------------------------------------------------
688-- Monad
689------------------------------------------------------------------------------
691-- GHC: if we change the implementation of bindWith with arguments in a
692-- different order we see a significant performance degradation (~2x).
693{-# INLINE bindAsync #-}
694{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
695bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
696bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
698-- GHC: if we specify arguments in the definition of (>>=) we see a significant
699-- performance degradation (~2x).
700instance MonadAsync m => Monad (AsyncT m) where
701 return = pure
702 (>>=) = bindAsync
704{-# INLINE apAsync #-}
705{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
706apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
707apAsync mf m = ap (adapt mf) (adapt m)
709instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
710 pure = AsyncT . K.yield
711 (<*>) = apAsync
713------------------------------------------------------------------------------
714-- Other instances
715------------------------------------------------------------------------------
717MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
719------------------------------------------------------------------------------
720-- WAsyncT
721------------------------------------------------------------------------------
723-- | XXX we can implement it more efficienty by directly implementing instead
724-- of combining streams using wAsync.
725{-# INLINE consMWAsync #-}
726{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
727consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
728consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
730-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
731-- Merges two streams concurrently choosing elements from both fairly.
732--
733-- @since 0.2.0
734{-# INLINE wAsync #-}
735wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
736wAsync = joinStreamVarAsync WAsyncVar
738-- | Wide async composition or async composition with breadth first traversal.
739-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
740-- streams using a depth first travesal or in a round robin fashion, yielding
741-- elements from both streams alternately.
742--
743-- @
744-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
745-- @
746-- @
747-- [1,3,2,4]
748-- @
749--
750-- Any exceptions generated by a constituent stream are propagated to the
751-- output stream. The output and exceptions from a single stream are guaranteed
752-- to arrive in the same order in the resulting stream as they were generated
753-- in the input stream. However, the relative ordering of elements from
754-- different streams in the resulting stream can vary depending on scheduling
755-- and generation delays.
756--
757-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
758-- concurrently using a round robin scheduling.
759--
760-- @
761-- import "Streamly"
762-- import qualified "Streamly.Prelude" as S
763-- import Control.Concurrent
764--
765-- main = 'drain' . 'wAsyncly' $ do
766-- n <- return 3 \<\> return 2 \<\> return 1
767-- S.yieldM $ do
768-- threadDelay (n * 1000000)
769-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
770-- @
771-- @
772-- ThreadId 40: Delay 1
773-- ThreadId 39: Delay 2
774-- ThreadId 38: Delay 3
775-- @
776--
777-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
778-- concurrently, unconditionally.
779--
780-- Note that async composition with breadth first traversal can only combine a
781-- finite number of streams as it needs to retain state for each unfinished
782-- stream.
783--
784-- @since 0.2.0
785newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
786 deriving (MonadTrans)
788-- | A round robin parallely composing IO stream of elements of type @a@.
789-- See 'WAsyncT' documentation for more details.
790--
791-- @since 0.2.0
792type WAsync = WAsyncT IO
794-- | Fix the type of a polymorphic stream as 'WAsyncT'.
795--
796-- @since 0.2.0
797wAsyncly :: IsStream t => WAsyncT m a -> t m a
798wAsyncly = adapt
800instance IsStream WAsyncT where
801 toStream = getWAsyncT
802 fromStream = WAsyncT
803 consM = consMWAsync
804 (|:) = consMWAsync
806------------------------------------------------------------------------------
807-- Semigroup
808------------------------------------------------------------------------------
810{-# INLINE mappendWAsync #-}
811{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
812mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
813mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
815instance MonadAsync m => Semigroup (WAsyncT m a) where
816 (<>) = mappendWAsync
818------------------------------------------------------------------------------
819-- Monoid
820------------------------------------------------------------------------------
822instance MonadAsync m => Monoid (WAsyncT m a) where
823 mempty = K.nil
824 mappend = (<>)
826------------------------------------------------------------------------------
827-- Monad
828------------------------------------------------------------------------------
830-- GHC: if we change the implementation of bindWith with arguments in a
831-- different order we see a significant performance degradation (~2x).
832{-# INLINE bindWAsync #-}
833{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
834bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
835bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
837-- GHC: if we specify arguments in the definition of (>>=) we see a significant
838-- performance degradation (~2x).
839instance MonadAsync m => Monad (WAsyncT m) where
840 return = pure
841 (>>=) = bindWAsync
843{-# INLINE apWAsync #-}
844{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
845apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
846apWAsync mf m = ap (adapt mf) (adapt m)
848instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
849 pure = WAsyncT . K.yield
850 (<*>) = apWAsync
852------------------------------------------------------------------------------
853-- Other instances
854------------------------------------------------------------------------------
856MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)