git.iko.soy/streamly
on bcd787e help sign in
streamly/src/Streamly/Streams/Async.hs
843 lines 32 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE LambdaCase #-}
8{-# LANGUAGE MultiParamTypeClasses #-}
9{-# LANGUAGE ScopedTypeVariables #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- |
13-- Module : Streamly.Streams.Async
14-- Copyright : (c) 2017 Harendra Kumar
15--
16-- License : BSD3
17-- Maintainer : harendra.kumar@gmail.com
18-- Stability : experimental
19-- Portability : GHC
20--
21--
22module Streamly.Streams.Async
23 (
24 AsyncT
25 , Async
26 , asyncly
27 , async
28 , (<|) --deprecated
29 , mkAsync
30 , mkAsync'
32 , WAsyncT
33 , WAsync
34 , wAsyncly
35 , wAsync
36 )
37where
39import Control.Concurrent (myThreadId)
40import Control.Monad (ap)
41import Control.Monad.Base (MonadBase(..), liftBaseDefault)
42import Control.Monad.Catch (MonadThrow, throwM)
43import Control.Concurrent.MVar (newEmptyMVar)
44-- import Control.Monad.Error.Class (MonadError(..))
45import Control.Monad.IO.Class (MonadIO(..))
46import Control.Monad.Reader.Class (MonadReader(..))
47import Control.Monad.State.Class (MonadState(..))
48import Control.Monad.Trans.Class (MonadTrans(lift))
49import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR)
50import Data.IORef (IORef, newIORef, readIORef)
51import Data.Maybe (fromJust)
52import Data.Semigroup (Semigroup(..))
54import Prelude hiding (map)
55import qualified Data.Set as S
57import Streamly.Streams.SVar (fromSVar)
58import Streamly.Streams.Serial (map)
59import Streamly.SVar
60import Streamly.Streams.StreamK
61 (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared,
62 foldStreamSVar)
63import qualified Streamly.Streams.StreamK as K
65#include "Instances.hs"
67-------------------------------------------------------------------------------
68-- Async
69-------------------------------------------------------------------------------
71{-# INLINE workLoopLIFO #-}
72workLoopLIFO
73 :: MonadIO m
74 => IORef [Stream m a]
75 -> State Stream m a
76 -> SVar Stream m a
77 -> Maybe WorkerInfo
78 -> m ()
79workLoopLIFO q st sv winfo = run
81 where
83 run = do
84 work <- dequeue
85 case work of
86 Nothing -> liftIO $ sendStop sv winfo
87 Just m -> foldStreamSVar sv st yieldk single run m
89 single a = do
90 res <- liftIO $ sendYield sv winfo (ChildYield a)
91 if res then run else liftIO $ sendStop sv winfo
93 yieldk a r = do
94 res <- liftIO $ sendYield sv winfo (ChildYield a)
95 if res
96 then foldStreamSVar sv st yieldk single run r
97 else liftIO $ do
98 enqueueLIFO sv q r
99 sendStop sv winfo
101 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
102 [] -> ([], Nothing)
103 x : xs -> (xs, Just x)
105-- We duplicate workLoop for yield limit and no limit cases because it has
106-- around 40% performance overhead in the worst case.
107--
108-- XXX we can pass yinfo directly as an argument here so that we do not have to
109-- make a check every time.
110{-# INLINE workLoopLIFOLimited #-}
111workLoopLIFOLimited
112 :: MonadIO m
113 => IORef [Stream m a]
114 -> State Stream m a
115 -> SVar Stream m a
116 -> Maybe WorkerInfo
117 -> m ()
118workLoopLIFOLimited q st sv winfo = run
120 where
122 run = do
123 work <- dequeue
124 case work of
125 Nothing -> liftIO $ sendStop sv winfo
126 Just m -> do
127 -- XXX This is just a best effort minimization of concurrency
128 -- to the yield limit. If the stream is made of concurrent
129 -- streams we do not reserve the yield limit in the constituent
130 -- streams before executing the action. This can be done
131 -- though, by sharing the yield limit ref with downstream
132 -- actions via state passing. Just a todo.
133 yieldLimitOk <- liftIO $ decrementYieldLimit sv
134 if yieldLimitOk
135 then do
136 let stop = liftIO (incrementYieldLimit sv) >> run
137 foldStreamSVar sv st yieldk single stop m
138 -- Avoid any side effects, undo the yield limit decrement if we
139 -- never yielded anything.
140 else liftIO $ do
141 enqueueLIFO sv q m
142 incrementYieldLimit sv
143 sendStop sv winfo
145 single a = do
146 res <- liftIO $ sendYield sv winfo (ChildYield a)
147 if res then run else liftIO $ sendStop sv winfo
149 -- XXX can we pass on the yield limit downstream to limit the concurrency
150 -- of constituent streams.
151 yieldk a r = do
152 res <- liftIO $ sendYield sv winfo (ChildYield a)
153 yieldLimitOk <- liftIO $ decrementYieldLimit sv
154 let stop = liftIO (incrementYieldLimit sv) >> run
155 if res && yieldLimitOk
156 then foldStreamSVar sv st yieldk single stop r
157 else liftIO $ do
158 incrementYieldLimit sv
159 enqueueLIFO sv q r
160 sendStop sv winfo
162 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
163 [] -> ([], Nothing)
164 x : xs -> (xs, Just x)
166-------------------------------------------------------------------------------
167-- WAsync
168-------------------------------------------------------------------------------
170-- XXX we can remove sv as it is derivable from st
172{-# INLINE workLoopFIFO #-}
173workLoopFIFO
174 :: MonadIO m
175 => LinkedQueue (Stream m a)
176 -> State Stream m a
177 -> SVar Stream m a
178 -> Maybe WorkerInfo
179 -> m ()
180workLoopFIFO q st sv winfo = run
182 where
184 run = do
185 work <- liftIO $ tryPopR q
186 case work of
187 Nothing -> liftIO $ sendStop sv winfo
188 Just m -> foldStreamSVar sv st yieldk single run m
190 single a = do
191 res <- liftIO $ sendYield sv winfo (ChildYield a)
192 if res then run else liftIO $ sendStop sv winfo
194 yieldk a r = do
195 res <- liftIO $ sendYield sv winfo (ChildYield a)
196 if res
197 then foldStreamSVar sv st yieldk single run r
198 else liftIO $ do
199 enqueueFIFO sv q r
200 sendStop sv winfo
202{-# INLINE workLoopFIFOLimited #-}
203workLoopFIFOLimited
204 :: MonadIO m
205 => LinkedQueue (Stream m a)
206 -> State Stream m a
207 -> SVar Stream m a
208 -> Maybe WorkerInfo
209 -> m ()
210workLoopFIFOLimited q st sv winfo = run
212 where
214 run = do
215 work <- liftIO $ tryPopR q
216 case work of
217 Nothing -> liftIO $ sendStop sv winfo
218 Just m -> do
219 yieldLimitOk <- liftIO $ decrementYieldLimit sv
220 if yieldLimitOk
221 then do
222 let stop = liftIO (incrementYieldLimit sv) >> run
223 foldStreamSVar sv st yieldk single stop m
224 else liftIO $ do
225 enqueueFIFO sv q m
226 incrementYieldLimit sv
227 sendStop sv winfo
229 single a = do
230 res <- liftIO $ sendYield sv winfo (ChildYield a)
231 if res then run else liftIO $ sendStop sv winfo
233 yieldk a r = do
234 res <- liftIO $ sendYield sv winfo (ChildYield a)
235 yieldLimitOk <- liftIO $ decrementYieldLimit sv
236 let stop = liftIO (incrementYieldLimit sv) >> run
237 if res && yieldLimitOk
238 then foldStreamSVar sv st yieldk single stop r
239 else liftIO $ do
240 incrementYieldLimit sv
241 enqueueFIFO sv q r
242 sendStop sv winfo
244-------------------------------------------------------------------------------
245-- SVar creation
246-- This code belongs in SVar.hs but is kept here for perf reasons
247-------------------------------------------------------------------------------
249-- XXX we have this function in this file because passing runStreamLIFO as a
250-- function argument to this function results in a perf degradation of more
251-- than 10%. Need to investigate what the root cause is.
252-- Interestingly, the same thing does not make any difference for Ahead.
253getLifoSVar :: forall m a. MonadAsync m
254 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
255getLifoSVar st mrun = do
256 outQ <- newIORef ([], 0)
257 outQMv <- newEmptyMVar
258 active <- newIORef 0
259 wfw <- newIORef False
260 running <- newIORef S.empty
261 q <- newIORef []
262 yl <- case getYieldLimit st of
263 Nothing -> return Nothing
264 Just x -> Just <$> newIORef x
265 rateInfo <- getYieldRateInfo st
267 stats <- newSVarStats
268 tid <- myThreadId
270 let isWorkFinished _ = null <$> readIORef q
272 let isWorkFinishedLimited sv = do
273 yieldsDone <-
274 case remainingWork sv of
275 Just ref -> do
276 n <- readIORef ref
277 return (n <= 0)
278 Nothing -> return False
279 qEmpty <- null <$> readIORef q
280 return $ qEmpty || yieldsDone
282 let getSVar :: SVar Stream m a
283 -> (SVar Stream m a -> m [ChildEvent a])
284 -> (SVar Stream m a -> m Bool)
285 -> (SVar Stream m a -> IO Bool)
286 -> (IORef [Stream m a]
287 -> State Stream m a
288 -> SVar Stream m a
289 -> Maybe WorkerInfo
290 -> m())
291 -> SVar Stream m a
292 getSVar sv readOutput postProc workDone wloop = SVar
293 { outputQueue = outQ
294 , remainingWork = yl
295 , maxBufferLimit = getMaxBuffer st
296 , maxWorkerLimit = getMaxThreads st
297 , yieldRateInfo = rateInfo
298 , outputDoorBell = outQMv
299 , readOutputQ = readOutput sv
300 , postProcess = postProc sv
301 , workerThreads = running
302 , workLoop = wloop q st{streamVar = Just sv} sv
303 , enqueue = enqueueLIFO sv q
304 , isWorkDone = workDone sv
305 , isQueueDone = workDone sv
306 , needDoorBell = wfw
307 , svarStyle = AsyncVar
308 , svarMrun = mrun
309 , workerCount = active
310 , accountThread = delThread sv
311 , workerStopMVar = undefined
312 , svarRef = Nothing
313 , svarInspectMode = getInspectMode st
314 , svarCreator = tid
315 , aheadWorkQueue = undefined
316 , outputHeap = undefined
317 , svarStats = stats
318 }
320 let sv =
321 case getStreamRate st of
322 Nothing ->
323 case getYieldLimit st of
324 Nothing -> getSVar sv readOutputQBounded
325 postProcessBounded
326 isWorkFinished
327 workLoopLIFO
328 Just _ -> getSVar sv readOutputQBounded
329 postProcessBounded
330 isWorkFinishedLimited
331 workLoopLIFOLimited
332 Just _ ->
333 case getYieldLimit st of
334 Nothing -> getSVar sv readOutputQPaced
335 postProcessPaced
336 isWorkFinished
337 workLoopLIFO
338 Just _ -> getSVar sv readOutputQPaced
339 postProcessPaced
340 isWorkFinishedLimited
341 workLoopLIFOLimited
342 in return sv
344getFifoSVar :: forall m a. MonadAsync m
345 => State Stream m a -> RunInIO m -> IO (SVar Stream m a)
346getFifoSVar st mrun = do
347 outQ <- newIORef ([], 0)
348 outQMv <- newEmptyMVar
349 active <- newIORef 0
350 wfw <- newIORef False
351 running <- newIORef S.empty
352 q <- newQ
353 yl <- case getYieldLimit st of
354 Nothing -> return Nothing
355 Just x -> Just <$> newIORef x
356 rateInfo <- getYieldRateInfo st
358 stats <- newSVarStats
359 tid <- myThreadId
361 let isWorkFinished _ = nullQ q
362 let isWorkFinishedLimited sv = do
363 yieldsDone <-
364 case remainingWork sv of
365 Just ref -> do
366 n <- readIORef ref
367 return (n <= 0)
368 Nothing -> return False
369 qEmpty <- nullQ q
370 return $ qEmpty || yieldsDone
372 let getSVar :: SVar Stream m a
373 -> (SVar Stream m a -> m [ChildEvent a])
374 -> (SVar Stream m a -> m Bool)
375 -> (SVar Stream m a -> IO Bool)
376 -> (LinkedQueue (Stream m a)
377 -> State Stream m a
378 -> SVar Stream m a
379 -> Maybe WorkerInfo
380 -> m())
381 -> SVar Stream m a
382 getSVar sv readOutput postProc workDone wloop = SVar
383 { outputQueue = outQ
384 , remainingWork = yl
385 , maxBufferLimit = getMaxBuffer st
386 , maxWorkerLimit = getMaxThreads st
387 , yieldRateInfo = rateInfo
388 , outputDoorBell = outQMv
389 , readOutputQ = readOutput sv
390 , postProcess = postProc sv
391 , workerThreads = running
392 , workLoop = wloop q st{streamVar = Just sv} sv
393 , enqueue = enqueueFIFO sv q
394 , isWorkDone = workDone sv
395 , isQueueDone = workDone sv
396 , needDoorBell = wfw
397 , svarStyle = WAsyncVar
398 , svarMrun = mrun
399 , workerCount = active
400 , accountThread = delThread sv
401 , workerStopMVar = undefined
402 , svarRef = Nothing
403 , svarInspectMode = getInspectMode st
404 , svarCreator = tid
405 , aheadWorkQueue = undefined
406 , outputHeap = undefined
407 , svarStats = stats
408 }
410 let sv =
411 case getStreamRate st of
412 Nothing ->
413 case getYieldLimit st of
414 Nothing -> getSVar sv readOutputQBounded
415 postProcessBounded
416 isWorkFinished
417 workLoopFIFO
418 Just _ -> getSVar sv readOutputQBounded
419 postProcessBounded
420 isWorkFinishedLimited
421 workLoopFIFOLimited
422 Just _ ->
423 case getYieldLimit st of
424 Nothing -> getSVar sv readOutputQPaced
425 postProcessPaced
426 isWorkFinished
427 workLoopFIFO
428 Just _ -> getSVar sv readOutputQPaced
429 postProcessPaced
430 isWorkFinishedLimited
431 workLoopFIFOLimited
432 in return sv
434{-# INLINABLE newAsyncVar #-}
435newAsyncVar :: MonadAsync m
436 => State Stream m a -> Stream m a -> m (SVar Stream m a)
437newAsyncVar st m = do
438 mrun <- captureMonadState
439 sv <- liftIO $ getLifoSVar st mrun
440 sendFirstWorker sv m
442-- XXX Get rid of this?
443-- | Make a stream asynchronous, triggers the computation and returns a stream
444-- in the underlying monad representing the output generated by the original
445-- computation. The returned action is exhaustible and must be drained once. If
446-- not drained fully we may have a thread blocked forever and once exhausted it
447-- will always return 'empty'.
448--
449-- @since 0.2.0
450{-# INLINABLE mkAsync #-}
451mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
452mkAsync m = fmap fromSVar (newAsyncVar defState (toStream m))
454{-# INLINABLE mkAsync' #-}
455mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)
456mkAsync' st m = fmap fromSVar (newAsyncVar st (toStream m))
458-- | Create a new SVar and enqueue one stream computation on it.
459{-# INLINABLE newWAsyncVar #-}
460newWAsyncVar :: MonadAsync m
461 => State Stream m a -> Stream m a -> m (SVar Stream m a)
462newWAsyncVar st m = do
463 mrun <- captureMonadState
464 sv <- liftIO $ getFifoSVar st mrun
465 sendFirstWorker sv m
467------------------------------------------------------------------------------
468-- Running streams concurrently
469------------------------------------------------------------------------------
471-- Concurrency rate control.
472--
473-- Our objective is to create more threads on demand if the consumer is running
474-- faster than us. As soon as we encounter a concurrent composition we create a
475-- push pull pair of threads. We use an SVar for communication between the
476-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
477-- The producer creates more threads if the SVar drains and becomes empty, that
478-- is the consumer is running faster.
479--
480-- XXX Note 1: This mechanism can be problematic if the initial production
481-- latency is high, we may end up creating too many threads. So we need some
482-- way to monitor and use the latency as well. Having a limit on the dispatches
483-- (programmer controlled) may also help.
484--
485-- TBD Note 2: We may want to run computations at the lower level of the
486-- composition tree serially even when they are composed using a parallel
487-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
488-- place of 'wAsync'. If we find that an SVar immediately above a computation
489-- gets drained empty we can switch to parallelizing the computation. For that
490-- we can use a state flag to fork the rest of the computation at any point of
491-- time inside the Monad bind operation if the consumer is running at a faster
492-- speed.
493--
494-- TBD Note 3: the binary operation ('parallel') composition allows us to
495-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
496-- chunksizes we will need to compose the parallel actions using a data
497-- constructor (A Free container) instead so that we can divide it in chunks of
498-- arbitrary size before dispatching. If the stream is composed of
499-- hierarchically composed grains of different sizes then we can always switch
500-- to a desired granularity depending on the consumer speed.
501--
502-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
503-- into just the number of CPUs.
505-- | Join two computations on the currently running 'SVar' queue for concurrent
506-- execution. When we are using parallel composition, an SVar is passed around
507-- as a state variable. We try to schedule a new parallel computation on the
508-- SVar passed to us. The first time, when no SVar exists, a new SVar is
509-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
510-- already scheduled on the SVar is further evaluated. For example, when (a
511-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
512-- the current scheduler queue.
513--
514-- The 'SVarStyle' required by the current composition context is passed as one
515-- of the parameters. If the scheduling and composition style of the new
516-- computation being scheduled is different than the style of the current SVar,
517-- then we create a new SVar and schedule it on that. The newly created SVar
518-- joins as one of the computations on the current SVar queue.
519--
520-- Cases when we need to switch to a new SVar:
521--
522-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
523-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
524-- because of the scheduling policy change.
525-- * if we 'adapt' a stream of type 'async' to a stream of type
526-- 'Parallel', we create a new SVar at the transitioning bind.
527-- * When the stream is switching from disjunctive composition to conjunctive
528-- composition and vice-versa we create a new SVar to isolate the scheduling
529-- of the two.
531forkSVarAsync :: (IsStream t, MonadAsync m)
532 => SVarStyle -> t m a -> t m a -> t m a
533forkSVarAsync style m1 m2 = mkStream $ \st stp sng yld -> do
534 sv <- case style of
535 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
536 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
537 _ -> error "illegal svar type"
538 foldStream st stp sng yld $ fromSVar sv
539 where
540 concurrently ma mb = mkStream $ \st stp sng yld -> do
541 liftIO $ enqueue (fromJust $ streamVar st) mb
542 foldStreamShared st stp sng yld ma
544{-# INLINE joinStreamVarAsync #-}
545joinStreamVarAsync :: (IsStream t, MonadAsync m)
546 => SVarStyle -> t m a -> t m a -> t m a
547joinStreamVarAsync style m1 m2 = mkStream $ \st stp sng yld ->
548 case streamVar st of
549 Just sv | svarStyle sv == style -> do
550 liftIO $ enqueue sv (toStream m2)
551 foldStreamShared st stp sng yld m1
552 _ -> foldStreamShared st stp sng yld (forkSVarAsync style m1 m2)
554------------------------------------------------------------------------------
555-- Semigroup and Monoid style compositions for parallel actions
556------------------------------------------------------------------------------
558-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
559-- Merges two streams possibly concurrently, preferring the
560-- elements from the left one when available.
561--
562-- @since 0.2.0
563{-# INLINE async #-}
564async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
565async = joinStreamVarAsync AsyncVar
567-- | Same as 'async'.
568--
569-- @since 0.1.0
570{-# DEPRECATED (<|) "Please use 'async' instead." #-}
571{-# INLINE (<|) #-}
572(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
573(<|) = async
575-- IMPORTANT: using a monomorphically typed and SPECIALIZED consMAsync makes a
576-- huge difference in the performance of consM in IsStream instance even we
577-- have a SPECIALIZE in the instance.
578--
579-- | XXX we can implement it more efficienty by directly implementing instead
580-- of combining streams using async.
581{-# INLINE consMAsync #-}
582{-# SPECIALIZE consMAsync :: IO a -> AsyncT IO a -> AsyncT IO a #-}
583consMAsync :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a
584consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)
586------------------------------------------------------------------------------
587-- AsyncT
588------------------------------------------------------------------------------
590-- | Deep async composition or async composition with depth first traversal. In
591-- a left to right 'Semigroup' composition it tries to yield elements from the
592-- left stream as long as it can, but it can run the right stream in parallel
593-- if it needs to, based on demand. The right stream can be run if the left
594-- stream blocks on IO or cannot produce elements fast enough for the consumer.
595--
596-- @
597-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
598-- @
599-- @
600-- [1,2,3,4]
601-- @
602--
603-- Any exceptions generated by a constituent stream are propagated to the
604-- output stream. The output and exceptions from a single stream are guaranteed
605-- to arrive in the same order in the resulting stream as they were generated
606-- in the input stream. However, the relative ordering of elements from
607-- different streams in the resulting stream can vary depending on scheduling
608-- and generation delays.
609--
610-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
611-- concurrently based on demand. More concurrent iterations are started only
612-- if the previous iterations are not able to produce enough output for the
613-- consumer.
614--
615-- @
616-- import "Streamly"
617-- import qualified "Streamly.Prelude" as S
618-- import Control.Concurrent
619--
620-- main = 'runStream' . 'asyncly' $ do
621-- n <- return 3 \<\> return 2 \<\> return 1
622-- S.yieldM $ do
623-- threadDelay (n * 1000000)
624-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
625-- @
626-- @
627-- ThreadId 40: Delay 1
628-- ThreadId 39: Delay 2
629-- ThreadId 38: Delay 3
630-- @
631--
632-- All iterations may run in the same thread if they do not block.
633--
634-- Note that async composition with depth first traversal can be used to
635-- combine infinite number of streams as it explores only a bounded number of
636-- streams at a time.
637--
638-- @since 0.1.0
639newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
640 deriving (MonadTrans)
642-- | A demand driven left biased parallely composing IO stream of elements of
643-- type @a@. See 'AsyncT' documentation for more details.
644--
645-- @since 0.2.0
646type Async = AsyncT IO
648-- | Fix the type of a polymorphic stream as 'AsyncT'.
649--
650-- @since 0.1.0
651asyncly :: IsStream t => AsyncT m a -> t m a
652asyncly = adapt
654instance IsStream AsyncT where
655 toStream = getAsyncT
656 fromStream = AsyncT
657 consM = consMAsync
658 (|:) = consMAsync
660------------------------------------------------------------------------------
661-- Semigroup
662------------------------------------------------------------------------------
664-- Monomorphically typed version of "async" for better performance of Semigroup
665-- instance.
666{-# INLINE mappendAsync #-}
667{-# SPECIALIZE mappendAsync :: AsyncT IO a -> AsyncT IO a -> AsyncT IO a #-}
668mappendAsync :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
669mappendAsync m1 m2 = fromStream $ async (toStream m1) (toStream m2)
671instance MonadAsync m => Semigroup (AsyncT m a) where
672 (<>) = mappendAsync
674------------------------------------------------------------------------------
675-- Monoid
676------------------------------------------------------------------------------
678instance MonadAsync m => Monoid (AsyncT m a) where
679 mempty = K.nil
680 mappend = (<>)
682------------------------------------------------------------------------------
683-- Monad
684------------------------------------------------------------------------------
686{-# INLINE bindAsync #-}
687{-# SPECIALIZE bindAsync :: AsyncT IO a -> (a -> AsyncT IO b) -> AsyncT IO b #-}
688bindAsync :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
689bindAsync m f = fromStream $ K.bindWith async (adapt m) (\a -> adapt $ f a)
691instance MonadAsync m => Monad (AsyncT m) where
692 return = pure
693 (>>=) = bindAsync
695{-# INLINE apAsync #-}
696{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}
697apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b
698apAsync mf m = ap (adapt mf) (adapt m)
700instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where
701 pure = AsyncT . K.yield
702 (<*>) = apAsync
704------------------------------------------------------------------------------
705-- Other instances
706------------------------------------------------------------------------------
708MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
710------------------------------------------------------------------------------
711-- WAsyncT
712------------------------------------------------------------------------------
714-- | XXX we can implement it more efficienty by directly implementing instead
715-- of combining streams using wAsync.
716{-# INLINE consMWAsync #-}
717{-# SPECIALIZE consMWAsync :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
718consMWAsync :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a
719consMWAsync m r = fromStream $ K.yieldM m `wAsync` (toStream r)
721-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
722-- Merges two streams concurrently choosing elements from both fairly.
723--
724-- @since 0.2.0
725{-# INLINE wAsync #-}
726wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
727wAsync = joinStreamVarAsync WAsyncVar
729-- | Wide async composition or async composition with breadth first traversal.
730-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
731-- streams using a depth first travesal or in a round robin fashion, yielding
732-- elements from both streams alternately.
733--
734-- @
735-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
736-- @
737-- @
738-- [1,3,2,4]
739-- @
740--
741-- Any exceptions generated by a constituent stream are propagated to the
742-- output stream. The output and exceptions from a single stream are guaranteed
743-- to arrive in the same order in the resulting stream as they were generated
744-- in the input stream. However, the relative ordering of elements from
745-- different streams in the resulting stream can vary depending on scheduling
746-- and generation delays.
747--
748-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
749-- concurrently using a round robin scheduling.
750--
751-- @
752-- import "Streamly"
753-- import qualified "Streamly.Prelude" as S
754-- import Control.Concurrent
755--
756-- main = 'runStream' . 'wAsyncly' $ do
757-- n <- return 3 \<\> return 2 \<\> return 1
758-- S.yieldM $ do
759-- threadDelay (n * 1000000)
760-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
761-- @
762-- @
763-- ThreadId 40: Delay 1
764-- ThreadId 39: Delay 2
765-- ThreadId 38: Delay 3
766-- @
767--
768-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
769-- concurrently, unconditionally.
770--
771-- Note that async composition with breadth first traversal can only combine a
772-- finite number of streams as it needs to retain state for each unfinished
773-- stream.
774--
775-- @since 0.2.0
776newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}
777 deriving (MonadTrans)
779-- | A round robin parallely composing IO stream of elements of type @a@.
780-- See 'WAsyncT' documentation for more details.
781--
782-- @since 0.2.0
783type WAsync = WAsyncT IO
785-- | Fix the type of a polymorphic stream as 'WAsyncT'.
786--
787-- @since 0.2.0
788wAsyncly :: IsStream t => WAsyncT m a -> t m a
789wAsyncly = adapt
791instance IsStream WAsyncT where
792 toStream = getWAsyncT
793 fromStream = WAsyncT
794 consM = consMWAsync
795 (|:) = consMWAsync
797------------------------------------------------------------------------------
798-- Semigroup
799------------------------------------------------------------------------------
801{-# INLINE mappendWAsync #-}
802{-# SPECIALIZE mappendWAsync :: WAsyncT IO a -> WAsyncT IO a -> WAsyncT IO a #-}
803mappendWAsync :: MonadAsync m => WAsyncT m a -> WAsyncT m a -> WAsyncT m a
804mappendWAsync m1 m2 = fromStream $ wAsync (toStream m1) (toStream m2)
806instance MonadAsync m => Semigroup (WAsyncT m a) where
807 (<>) = mappendWAsync
809------------------------------------------------------------------------------
810-- Monoid
811------------------------------------------------------------------------------
813instance MonadAsync m => Monoid (WAsyncT m a) where
814 mempty = K.nil
815 mappend = (<>)
817------------------------------------------------------------------------------
818-- Monad
819------------------------------------------------------------------------------
821{-# INLINE bindWAsync #-}
822{-# SPECIALIZE bindWAsync :: WAsyncT IO a -> (a -> WAsyncT IO b) -> WAsyncT IO b #-}
823bindWAsync :: MonadAsync m => WAsyncT m a -> (a -> WAsyncT m b) -> WAsyncT m b
824bindWAsync m f = fromStream $ K.bindWith wAsync (adapt m) (\a -> adapt $ f a)
826instance MonadAsync m => Monad (WAsyncT m) where
827 return = pure
828 (>>=) = bindWAsync
830{-# INLINE apWAsync #-}
831{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}
832apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b
833apWAsync mf m = ap (adapt mf) (adapt m)
835instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where
836 pure = WAsyncT . K.yield
837 (<*>) = apWAsync
839------------------------------------------------------------------------------
840-- Other instances
841------------------------------------------------------------------------------
843MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)