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