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