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