git.iko.soy/streamly
on fa75413 help sign in
streamly/src/Streamly/Internal/Data/Stream/Async.hs
864 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.Internal.Data.Stream.Async
14-- Copyright : (c) 2017 Harendra Kumar
15--
16-- License : BSD3
17-- Maintainer : streamly@composewell.com
18-- Stability : experimental
19-- Portability : GHC
20--
21--
22module Streamly.Internal.Data.Stream.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)
52#if __GLASGOW_HASKELL__ < 808
53import Data.Semigroup (Semigroup(..))
54#endif
56import Prelude hiding (map)
57import qualified Data.Set as S
59import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)
60import Streamly.Internal.Data.Stream.SVar (fromSVar)
61import Streamly.Internal.Data.Stream.Serial (map)
62import Streamly.Internal.Data.SVar
63import Streamly.Internal.Data.Stream.StreamK
64 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared,
65 foldStreamSVar)
66import qualified Streamly.Internal.Data.Stream.StreamK as K
68#include "Instances.hs"
70-------------------------------------------------------------------------------
71-- Async
72-------------------------------------------------------------------------------
74{-# INLINE workLoopLIFO #-}
75workLoopLIFO
76 :: MonadIO m
77 => IORef [Stream m a]
78 -> State Stream m a
79 -> SVar Stream m a
80 -> Maybe WorkerInfo
81 -> m ()
82workLoopLIFO q st sv winfo = run
84 where
86 run = do
87 work <- dequeue
88 case work of
89 Nothing -> liftIO $ sendStop sv winfo
90 Just m -> foldStreamSVar sv st yieldk single run m
92 single a = do
93 res <- liftIO $ sendYield sv winfo (ChildYield a)
94 if res then run else liftIO $ sendStop sv winfo
96 yieldk a r = do
97 res <- liftIO $ sendYield sv winfo (ChildYield a)
98 if res
99 then foldStreamSVar sv st yieldk single run r
100 else liftIO $ do
101 enqueueLIFO sv q r
102 sendStop sv winfo
104 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
105 [] -> ([], Nothing)
106 x : xs -> (xs, Just x)
108-- We duplicate workLoop for yield limit and no limit cases because it has
109-- around 40% performance overhead in the worst case.
110--
111-- XXX we can pass yinfo directly as an argument here so that we do not have to
112-- make a check every time.
113{-# INLINE workLoopLIFOLimited #-}
114workLoopLIFOLimited
115 :: MonadIO m
116 => IORef [Stream m a]
117 -> State Stream m a
118 -> SVar Stream m a
119 -> Maybe WorkerInfo
120 -> m ()
121workLoopLIFOLimited q st sv winfo = run
123 where
125 run = do
126 work <- dequeue
127 case work of
128 Nothing -> liftIO $ sendStop sv winfo
129 Just m -> do
130 -- XXX This is just a best effort minimization of concurrency
131 -- to the yield limit. If the stream is made of concurrent
132 -- streams we do not reserve the yield limit in the constituent
133 -- streams before executing the action. This can be done
134 -- though, by sharing the yield limit ref with downstream
135 -- actions via state passing. Just a todo.
136 yieldLimitOk <- liftIO $ decrementYieldLimit sv
137 if yieldLimitOk
138 then do
139 let stop = liftIO (incrementYieldLimit sv) >> run
140 foldStreamSVar sv st yieldk single stop m
141 -- Avoid any side effects, undo the yield limit decrement if we
142 -- never yielded anything.
143 else liftIO $ do
144 enqueueLIFO sv q m
145 incrementYieldLimit sv
146 sendStop sv winfo
148 single a = do
149 res <- liftIO $ sendYield sv winfo (ChildYield a)
150 if res then run else liftIO $ sendStop sv winfo
152 -- XXX can we pass on the yield limit downstream to limit the concurrency
153 -- of constituent streams.
154 yieldk a r = do
155 res <- liftIO $ sendYield sv winfo (ChildYield a)
156 yieldLimitOk <- liftIO $ decrementYieldLimit sv
157 let stop = liftIO (incrementYieldLimit sv) >> run
158 if res && yieldLimitOk
159 then foldStreamSVar sv st yieldk single stop r
160 else liftIO $ do
161 incrementYieldLimit sv
162 enqueueLIFO sv q r
163 sendStop sv winfo
165 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
166 [] -> ([], Nothing)
167 x : xs -> (xs, Just x)
169-------------------------------------------------------------------------------
170-- WAsync
171-------------------------------------------------------------------------------
173-- XXX we can remove sv as it is derivable from st
175{-# INLINE workLoopFIFO #-}
176workLoopFIFO
177 :: MonadIO m
178 => LinkedQueue (Stream m a)
179 -> State Stream m a
180 -> SVar Stream m a
181 -> Maybe WorkerInfo
182 -> m ()
183workLoopFIFO q st sv winfo = run
185 where
187 run = do
188 work <- liftIO $ tryPopR q
189 case work of
190 Nothing -> liftIO $ sendStop sv winfo
191 Just m -> foldStreamSVar sv st yieldk single run m
193 single a = do
194 res <- liftIO $ sendYield sv winfo (ChildYield a)
195 if res then run else liftIO $ sendStop sv winfo
197 yieldk a r = do
198 res <- liftIO $ sendYield sv winfo (ChildYield a)
199 if res
200 then foldStreamSVar sv st yieldk single run r
201 else liftIO $ do
202 enqueueFIFO sv q r
203 sendStop sv winfo
205{-# INLINE workLoopFIFOLimited #-}
206workLoopFIFOLimited
207 :: MonadIO m
208 => LinkedQueue (Stream m a)
209 -> State Stream m a
210 -> SVar Stream m a
211 -> Maybe WorkerInfo
212 -> m ()
213workLoopFIFOLimited q st sv winfo = run
215 where
217 run = do
218 work <- liftIO $ tryPopR q
219 case work of
220 Nothing -> liftIO $ sendStop sv winfo
221 Just m -> do
222 yieldLimitOk <- liftIO $ decrementYieldLimit sv
223 if yieldLimitOk
224 then do
225 let stop = liftIO (incrementYieldLimit sv) >> run
226 foldStreamSVar sv st yieldk single stop m
227 else liftIO $ do
228 enqueueFIFO sv q m
229 incrementYieldLimit sv
230 sendStop sv winfo
232 single a = do
233 res <- liftIO $ sendYield sv winfo (ChildYield a)
234 if res then run else liftIO $ sendStop sv winfo
236 yieldk a r = do
237 res <- liftIO $ sendYield sv winfo (ChildYield a)
238 yieldLimitOk <- liftIO $ decrementYieldLimit sv
239 let stop = liftIO (incrementYieldLimit sv) >> run
240 if res && yieldLimitOk
241 then foldStreamSVar sv st yieldk single stop r
242 else liftIO $ do
243 incrementYieldLimit sv
244 enqueueFIFO sv q r
245 sendStop sv winfo
247-------------------------------------------------------------------------------
248-- SVar creation
249-- This code belongs in SVar.hs but is kept here for perf reasons
250-------------------------------------------------------------------------------
252-- XXX we have this function in this file because passing runStreamLIFO as a
253-- function argument to this function results in a perf degradation of more
254-- than 10%. Need to investigate what the root cause is.
255-- Interestingly, the same thing does not make any difference for Ahead.
256getLifoSVar :: forall m a. MonadAsync m
257 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
258getLifoSVar st mrun = do
259 outQ <- newIORef ([], 0)
260 outQMv <- newEmptyMVar
261 active <- newIORef 0
262 wfw <- newIORef False
263 running <- newIORef S.empty
264 q <- newIORef []
265 yl <- case getYieldLimit st of
266 Nothing -> return Nothing
267 Just x -> Just <$> newIORef x
268 rateInfo <- getYieldRateInfo st
270 stats <- newSVarStats
271 tid <- myThreadId
273 let isWorkFinished _ = null <$> readIORef q
275 let isWorkFinishedLimited sv = do
276 yieldsDone <-
277 case remainingWork sv of
278 Just ref -> do
279 n <- readIORef ref
280 return (n <= 0)
281 Nothing -> return False
282 qEmpty <- null <$> readIORef q
283 return $ qEmpty || yieldsDone
285 let getSVar :: SVar Stream m a
286 -> (SVar Stream m a -> m [ChildEvent a])
287 -> (SVar Stream m a -> m Bool)
288 -> (SVar Stream m a -> IO Bool)
289 -> (IORef [Stream m a]
290 -> State Stream m a
291 -> SVar Stream m a
292 -> Maybe WorkerInfo
293 -> m())
294 -> SVar Stream m a
295 getSVar sv readOutput postProc workDone wloop = SVar
296 { outputQueue = outQ
297 , outputQueueFromConsumer = undefined
298 , remainingWork = yl
299 , maxBufferLimit = getMaxBuffer st
300 , pushBufferSpace = undefined
301 , pushBufferPolicy = undefined
302 , pushBufferMVar = undefined
303 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
304 , yieldRateInfo = rateInfo
305 , outputDoorBell = outQMv
306 , outputDoorBellFromConsumer = undefined
307 , readOutputQ = readOutput sv
308 , postProcess = postProc sv
309 , workerThreads = running
310 , workLoop = wloop q st{streamVar = Just sv} sv
311 , enqueue = enqueueLIFO sv q
312 , isWorkDone = workDone sv
313 , isQueueDone = workDone sv
314 , needDoorBell = wfw
315 , svarStyle = AsyncVar
316 , svarStopStyle = StopNone
317 , svarStopBy = undefined
318 , svarMrun = mrun
319 , workerCount = active
320 , accountThread = delThread sv
321 , workerStopMVar = undefined
322 , svarRef = Nothing
323 , svarInspectMode = getInspectMode st
324 , svarCreator = tid
325 , aheadWorkQueue = undefined
326 , outputHeap = undefined
327 , svarStats = stats
328 }
330 let sv =
331 case getStreamRate st of
332 Nothing ->
333 case getYieldLimit st of
334 Nothing -> getSVar sv readOutputQBounded
335 postProcessBounded
336 isWorkFinished
337 workLoopLIFO
338 Just _ -> getSVar sv readOutputQBounded
339 postProcessBounded
340 isWorkFinishedLimited
341 workLoopLIFOLimited
342 Just _ ->
343 case getYieldLimit st of
344 Nothing -> getSVar sv readOutputQPaced
345 postProcessPaced
346 isWorkFinished
347 workLoopLIFO
348 Just _ -> getSVar sv readOutputQPaced
349 postProcessPaced
350 isWorkFinishedLimited
351 workLoopLIFOLimited
352 in return sv
354getFifoSVar :: forall m a. MonadAsync m
355 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
356getFifoSVar st mrun = do
357 outQ <- newIORef ([], 0)
358 outQMv <- newEmptyMVar
359 active <- newIORef 0
360 wfw <- newIORef False
361 running <- newIORef S.empty
362 q <- newQ
363 yl <- case getYieldLimit st of
364 Nothing -> return Nothing
365 Just x -> Just <$> newIORef x
366 rateInfo <- getYieldRateInfo st
368 stats <- newSVarStats
369 tid <- myThreadId
371 let isWorkFinished _ = nullQ q
372 let isWorkFinishedLimited sv = do
373 yieldsDone <-
374 case remainingWork sv of
375 Just ref -> do
376 n <- readIORef ref
377 return (n <= 0)
378 Nothing -> return False
379 qEmpty <- nullQ q
380 return $ qEmpty || yieldsDone
382 let getSVar :: SVar Stream m a
383 -> (SVar Stream m a -> m [ChildEvent a])
384 -> (SVar Stream m a -> m Bool)
385 -> (SVar Stream m a -> IO Bool)
386 -> (LinkedQueue (Stream m a)
387 -> State Stream m a
388 -> SVar Stream m a
389 -> Maybe WorkerInfo
390 -> m())
391 -> SVar Stream m a
392 getSVar sv readOutput postProc workDone wloop = SVar
393 { outputQueue = outQ
394 , outputQueueFromConsumer = undefined
395 , remainingWork = yl
396 , maxBufferLimit = getMaxBuffer st
397 , pushBufferSpace = undefined
398 , pushBufferPolicy = undefined
399 , pushBufferMVar = undefined
400 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
401 , yieldRateInfo = rateInfo
402 , outputDoorBell = outQMv
403 , outputDoorBellFromConsumer = undefined
404 , readOutputQ = readOutput sv
405 , postProcess = postProc sv
406 , workerThreads = running
407 , workLoop = wloop q st{streamVar = Just sv} sv
408 , enqueue = enqueueFIFO sv q
409 , isWorkDone = workDone sv
410 , isQueueDone = workDone sv
411 , needDoorBell = wfw
412 , svarStyle = WAsyncVar
413 , svarStopStyle = StopNone
414 , svarStopBy = undefined
415 , svarMrun = mrun
416 , workerCount = active
417 , accountThread = delThread sv
418 , workerStopMVar = undefined
419 , svarRef = Nothing
420 , svarInspectMode = getInspectMode st
421 , svarCreator = tid
422 , aheadWorkQueue = undefined
423 , outputHeap = undefined
424 , svarStats = stats
425 }
427 let sv =
428 case getStreamRate st of
429 Nothing ->
430 case getYieldLimit st of
431 Nothing -> getSVar sv readOutputQBounded
432 postProcessBounded
433 isWorkFinished
434 workLoopFIFO
435 Just _ -> getSVar sv readOutputQBounded
436 postProcessBounded
437 isWorkFinishedLimited
438 workLoopFIFOLimited
439 Just _ ->
440 case getYieldLimit st of
441 Nothing -> getSVar sv readOutputQPaced
442 postProcessPaced
443 isWorkFinished
444 workLoopFIFO
445 Just _ -> getSVar sv readOutputQPaced
446 postProcessPaced
447 isWorkFinishedLimited
448 workLoopFIFOLimited
449 in return sv
451{-# INLINABLE newAsyncVar #-}
452newAsyncVar :: MonadAsync m
453 => State Stream m a -> Stream m a -> m (SVar Stream m a)
454newAsyncVar st m = do
455 mrun <- captureMonadState
456 sv <- liftIO $ getLifoSVar st mrun
457 sendFirstWorker sv m
459-- XXX Get rid of this?
460-- | Make a stream asynchronous, triggers the computation and returns a stream
461-- in the underlying monad representing the output generated by the original
462-- computation. The returned action is exhaustible and must be drained once. If
463-- not drained fully we may have a thread blocked forever and once exhausted it
464-- will always return 'empty'.
465--
466-- @since 0.2.0
467{-# INLINABLE mkAsync #-}
468mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
469mkAsync = mkAsync' defState
471{-# INLINABLE mkAsync' #-}
472mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)
473mkAsync' st m = fmap fromSVar (newAsyncVar st (toStream m))
475-- | Create a new SVar and enqueue one stream computation on it.
476{-# INLINABLE newWAsyncVar #-}
477newWAsyncVar :: MonadAsync m
478 => State Stream m a -> Stream m a -> m (SVar Stream m a)
479newWAsyncVar st m = do
480 mrun <- captureMonadState
481 sv <- liftIO $ getFifoSVar st mrun
482 sendFirstWorker sv m
484------------------------------------------------------------------------------
485-- Running streams concurrently
486------------------------------------------------------------------------------
488-- Concurrency rate control.
489--
490-- Our objective is to create more threads on demand if the consumer is running
491-- faster than us. As soon as we encounter a concurrent composition we create a
492-- push pull pair of threads. We use an SVar for communication between the
493-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
494-- The producer creates more threads if the SVar drains and becomes empty, that
495-- is the consumer is running faster.
496--
497-- XXX Note 1: This mechanism can be problematic if the initial production
498-- latency is high, we may end up creating too many threads. So we need some
499-- way to monitor and use the latency as well. Having a limit on the dispatches
500-- (programmer controlled) may also help.
501--
502-- TBD Note 2: We may want to run computations at the lower level of the
503-- composition tree serially even when they are composed using a parallel
504-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
505-- place of 'wAsync'. If we find that an SVar immediately above a computation
506-- gets drained empty we can switch to parallelizing the computation. For that
507-- we can use a state flag to fork the rest of the computation at any point of
508-- time inside the Monad bind operation if the consumer is running at a faster
509-- speed.
510--
511-- TBD Note 3: the binary operation ('parallel') composition allows us to
512-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
513-- chunksizes we will need to compose the parallel actions using a data
514-- constructor (A Free container) instead so that we can divide it in chunks of
515-- arbitrary size before dispatching. If the stream is composed of
516-- hierarchically composed grains of different sizes then we can always switch
517-- to a desired granularity depending on the consumer speed.
518--
519-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
520-- into just the number of CPUs.
522-- | Join two computations on the currently running 'SVar' queue for concurrent
523-- execution. When we are using parallel composition, an SVar is passed around
524-- as a state variable. We try to schedule a new parallel computation on the
525-- SVar passed to us. The first time, when no SVar exists, a new SVar is
526-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
527-- already scheduled on the SVar is further evaluated. For example, when (a
528-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
529-- the current scheduler queue.
530--
531-- The 'SVarStyle' required by the current composition context is passed as one
532-- of the parameters. If the scheduling and composition style of the new
533-- computation being scheduled is different than the style of the current SVar,
534-- then we create a new SVar and schedule it on that. The newly created SVar
535-- joins as one of the computations on the current SVar queue.
536--
537-- Cases when we need to switch to a new SVar:
538--
539-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
540-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
541-- because of the scheduling policy change.
542-- * if we 'adapt' a stream of type 'async' to a stream of type
543-- 'Parallel', we create a new SVar at the transitioning bind.
544-- * When the stream is switching from disjunctive composition to conjunctive
545-- composition and vice-versa we create a new SVar to isolate the scheduling
546-- of the two.
548forkSVarAsync :: (IsStream t, MonadAsync m)
549 => SVarStyle -> t m a -> t m a -> t m a
550forkSVarAsync style m1 m2 = mkStream $ \st stp sng yld -> do
551 sv <- case style of
552 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
553 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
554 _ -> error "illegal svar type"
555 foldStream st stp sng yld $ fromSVar sv
556 where
557 concurrently ma mb = mkStream $ \st stp sng yld -> do
558 liftIO $ enqueue (fromJust $ streamVar st) mb
559 foldStreamShared st stp sng yld ma
561{-# INLINE joinStreamVarAsync #-}
562joinStreamVarAsync :: (IsStream t, MonadAsync m)
563 => SVarStyle -> t m a -> t m a -> t m a
564joinStreamVarAsync style m1 m2 = mkStream $ \st stp sng yld ->
565 case streamVar st of
566 Just sv | svarStyle sv == style -> do
567 liftIO $ enqueue sv (toStream m2)
568 foldStreamShared st stp sng yld m1
569 _ -> foldStreamShared st stp sng yld (forkSVarAsync style m1 m2)
571------------------------------------------------------------------------------
572-- Semigroup and Monoid style compositions for parallel actions
573------------------------------------------------------------------------------
575-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
576-- Merges two streams possibly concurrently, preferring the
577-- elements from the left one when available.
578--
579-- @since 0.2.0
580{-# INLINE async #-}
581async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
582async = joinStreamVarAsync AsyncVar
584-- | Same as 'async'.
585--
586-- @since 0.1.0
587{-# DEPRECATED (<|) "Please use 'async' instead." #-}
588{-# INLINE (<|) #-}
589(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
590(<|) = async
592-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
593-- huge difference in the performance of consM in IsStream instance even we
594-- have a SPECIALIZE in the instance.
595--
596-- | XXX we can implement it more efficienty by directly implementing instead
597-- of combining streams using async.
598{-# INLINE consMAsync #-}
599{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
600consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
601consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
603------------------------------------------------------------------------------
604-- AsyncT
605------------------------------------------------------------------------------
607-- | The 'Semigroup' operation for 'AsyncT' appends two streams. The combined
608-- stream behaves like a single stream with the actions from the second stream
609-- appended to the first stream. The combined stream is evaluated in the
610-- asynchronous style. This operation can be used to fold an infinite lazy
611-- container of streams.
612--
613-- @
614-- import "Streamly"
615-- import qualified "Streamly.Prelude" as S
616-- import Control.Concurrent
617--
618-- main = (S.toList . 'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
619-- @
620-- @
621-- [1,2,3,4]
622-- @
623--
624-- Any exceptions generated by a constituent stream are propagated to the
625-- output stream. The output and exceptions from a single stream are guaranteed
626-- to arrive in the same order in the resulting stream as they were generated
627-- in the input stream. However, the relative ordering of elements from
628-- different streams in the resulting stream can vary depending on scheduling
629-- and generation delays.
630--
631-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
632-- concurrently based on demand. More concurrent iterations are started only
633-- if the previous iterations are not able to produce enough output for the
634-- consumer.
635--
636-- @
637-- main = 'drain' . 'asyncly' $ do
638-- n <- return 3 \<\> return 2 \<\> return 1
639-- S.yieldM $ do
640-- threadDelay (n * 1000000)
641-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
642-- @
643-- @
644-- ThreadId 40: Delay 1
645-- ThreadId 39: Delay 2
646-- ThreadId 38: Delay 3
647-- @
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-- | The 'Semigroup' operation for 'WAsyncT' interleaves the elements from the
745-- two streams. Therefore, when @a <> b@ is evaluated, one action is picked
746-- from stream @a@ for evaluation and then the next action is picked from
747-- stream @b@ and then the next action is again picked from stream @a@, going
748-- around in a round-robin fashion. Many such actions are executed concurrently
749-- depending on 'maxThreads' and 'maxBuffer' limits. Results are served to the
750-- consumer in the order completion of the actions.
751--
752-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we
753-- go in a round-robin fasion across all of them picking one action from each
754-- up to @z@ and then come back to @a@. Note that this operation cannot be
755-- used to fold a container of infinite streams as the state that it needs to
756-- maintain is proportional to the number of streams.
757--
758-- @
759-- import "Streamly"
760-- import qualified "Streamly.Prelude" as S
761-- import Control.Concurrent
762--
763-- main = (S.toList . 'wAsyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print
764-- @
765-- @
766-- [1,3,2,4]
767-- @
768--
769-- Any exceptions generated by a constituent stream are propagated to the
770-- output stream. The output and exceptions from a single stream are guaranteed
771-- to arrive in the same order in the resulting stream as they were generated
772-- in the input stream. However, the relative ordering of elements from
773-- different streams in the resulting stream can vary depending on scheduling
774-- and generation delays.
775--
776-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
777-- concurrently using a round robin scheduling.
778--
779-- @
780-- main = 'drain' . 'wAsyncly' $ do
781-- n <- return 3 \<\> return 2 \<\> return 1
782-- S.yieldM $ do
783-- threadDelay (n * 1000000)
784-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
785-- @
786-- @
787-- ThreadId 40: Delay 1
788-- ThreadId 39: Delay 2
789-- ThreadId 38: Delay 3
790-- @
791--
792-- @since 0.2.0
793newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
794 deriving (MonadTrans)
796-- | A round robin parallely composing IO stream of elements of type @a@.
797-- See 'WAsyncT' documentation for more details.
798--
799-- @since 0.2.0
800type WAsync = WAsyncT IO
802-- | Fix the type of a polymorphic stream as 'WAsyncT'.
803--
804-- @since 0.2.0
805wAsyncly :: IsStream t => WAsyncT m a -> t m a
806wAsyncly = adapt
808instance IsStream WAsyncT where
809 toStream = getWAsyncT
810 fromStream = WAsyncT
811 consM = consMWAsync
812 (|:) = consMWAsync
814------------------------------------------------------------------------------
815-- Semigroup
816------------------------------------------------------------------------------
818{-# INLINE mappendWAsync #-}
819{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
820mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
821mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
823instance MonadAsync m => Semigroup (WAsyncT m a) where
824 (<>) = mappendWAsync
826------------------------------------------------------------------------------
827-- Monoid
828------------------------------------------------------------------------------
830instance MonadAsync m => Monoid (WAsyncT m a) where
831 mempty = K.nil
832 mappend = (<>)
834------------------------------------------------------------------------------
835-- Monad
836------------------------------------------------------------------------------
838-- GHC: if we change the implementation of bindWith with arguments in a
839-- different order we see a significant performance degradation (~2x).
840{-# INLINE bindWAsync #-}
841{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
842bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
843bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
845-- GHC: if we specify arguments in the definition of (>>=) we see a significant
846-- performance degradation (~2x).
847instance MonadAsync m => Monad (WAsyncT m) where
848 return = pure
849 (>>=) = bindWAsync
851{-# INLINE apWAsync #-}
852{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
853apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
854apWAsync mf m = ap (adapt mf) (adapt m)
856instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
857 pure = WAsyncT . K.yield
858 (<*>) = apWAsync
860------------------------------------------------------------------------------
861-- Other instances
862------------------------------------------------------------------------------
864MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)