1{-# LANGUAGE UndecidableInstances #-}
6-- Module : Streamly.Internal.Data.Stream.Parallel
7-- Copyright : (c) 2017 Composewell Technologies
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
15module Streamly.Internal.Data.Stream.Parallel
17 -- * Parallel Stream Type
22 -- * Merge Concurrently
27 -- * Evaluate Concurrently
42import Control.Concurrent (myThreadId, takeMVar)
43import Control.Monad (when)
44import Control.Monad.Base (MonadBase(..), liftBaseDefault)
45import Control.Monad.Catch (MonadThrow, throwM)
46-- import Control.Monad.Error.Class (MonadError(..))
47import Control.Monad.IO.Class (MonadIO(..))
48import Control.Monad.Reader.Class (MonadReader(..))
49import Control.Monad.State.Class (MonadState(..))
50import Control.Monad.Trans.Class (MonadTrans(lift))
51import Data.Functor (void)
52import Data.IORef (readIORef, writeIORef)
53import Data.Maybe (fromJust)
54#if __GLASGOW_HASKELL__ < 808
55import Data.Semigroup (Semigroup(..))
57import Prelude hiding (map)
59import qualified Data.Set as Set
61import Streamly.Internal.Data.Fold.Types (Fold)
62import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))
63import Streamly.Internal.Data.Stream.StreamK
64 (IsStream(..), Stream, mkStream, foldStream, foldStreamShared, adapt)
66import Streamly.Internal.Data.SVar
68import qualified Streamly.Internal.Data.Stream.StreamK as K (withLocal)
69import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
70import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
71import qualified Streamly.Internal.Data.Stream.SVar as SVar
73#include "Instances.hs"
78-- >>> import Prelude hiding (map)
79-- >>> import qualified Streamly.Prelude as Stream
80-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
82-------------------------------------------------------------------------------
84-------------------------------------------------------------------------------
86-------------------------------------------------------------------------------
87-- StreamK based worker routines
88-------------------------------------------------------------------------------
90{-# NOINLINE runOne #-}
93 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
95 case getYieldLimit st of
97 Just _ -> runOneLimited st m0 winfo
102 liftIO $ decrementBufferLimit sv
103 foldStreamShared st yieldk single stop m
105 sv = fromJust $ streamVar st
108 incrementBufferLimit sv
110 sendit a = liftIO $ void $ send sv (ChildYield a)
111 single a = sendit a >> (liftIO $ sendStop sv winfo)
112 yieldk a r = sendit a >> go r
116 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
117runOneLimited st m0 winfo = go m0
122 yieldLimitOk <- liftIO $ decrementYieldLimit sv
125 liftIO $ decrementBufferLimit sv
126 foldStreamShared st yieldk single stop m
128 liftIO $ cleanupSVarFromWorker sv
129 liftIO $ sendStop sv winfo
131 sv = fromJust $ streamVar st
134 incrementBufferLimit sv
135 incrementYieldLimit sv
137 sendit a = liftIO $ void $ send sv (ChildYield a)
138 single a = sendit a >> (liftIO $ sendStop sv winfo)
139 yieldk a r = sendit a >> go r
141-------------------------------------------------------------------------------
142-- Consing and appending a stream in parallel style
143-------------------------------------------------------------------------------
145-- Note that consing and appending requires StreamK as it would not scale well
146-- with StreamD unless we are only consing a very small number of streams or
147-- elements in a stream. StreamK allows us to manipulate control flow in a way
148-- which StreamD cannot allow. StreamK can make a jump without having to
149-- remember the past state.
151{-# NOINLINE forkSVarPar #-}
152forkSVarPar :: (IsStream t, MonadAsync m)
153 => SVarStopStyle -> t m a -> t m a -> t m a
154forkSVarPar ss m r = mkStream $ \st yld sng stp -> do
155 sv <- newParallelVar ss st
156 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
158 StopBy -> liftIO $ do
159 set <- readIORef (workerThreads sv)
160 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
162 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
163 foldStream st yld sng stp (SVar.fromSVar sv)
165{-# INLINE joinStreamVarPar #-}
166joinStreamVarPar :: (IsStream t, MonadAsync m)
167 => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a
168joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->
170 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
171 -- Here, WE ARE IN THE WORKER/PRODUCER THREAD, we know that because
172 -- the SVar exists. We are running under runOne and the output we
173 -- produce ultimately will be sent to the SVar by runOne.
175 -- If we came here the worker/runOne is evaluating a `parallel`
176 -- combinator. In this case, we always fork a new worker for the
177 -- first component (m1) in the parallel composition and continue to
178 -- evaluate the second component (m2) in the current worker thread.
180 -- When m1 is serially composed, the worker would evaluate it
181 -- without any further forks and the resulting output is sent to
182 -- the SVar and the evaluation terminates. If m1 is a `parallel`
183 -- composition of two streams the worker would again recurses here.
185 -- Similarly, when m2 is serially composed it gets evaluated here
186 -- and the resulting output is sent to the SVar by the runOne
187 -- wrapper. When m2 is composed with `parallel` it will again
188 -- recurse here and so on until it finally terminates.
190 -- When we create a right associated expression using `parallel`,
191 -- then m1 would always terminate without further forks or
192 -- recursion into this routine, therefore, the worker returns
193 -- immediately after evaluating it. And m2 would continue to
194 -- fork/recurse, therefore, the current thread always recurses and
195 -- forks new workers one after the other. This is a tail recursive
196 -- style execution, m2, the recursive expression always executed at
199 -- When the expression is left associated, the worker spawned would
200 -- get the forking/recursing responsibility and then again the
201 -- worker spawned by that worker would fork, thus creating layer
202 -- over layer of workers and a chain of threads leading to a very
203 -- inefficient execution.
204 pushWorkerPar sv (runOne st $ toStream m1)
205 foldStreamShared st yld sng stp m2
207 -- Here WE ARE IN THE CONSUMER THREAD, we create a new SVar, fork
208 -- worker threads to execute m1 and m2 and this thread starts
209 -- pulling the stream from the SVar.
210 foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
212-------------------------------------------------------------------------------
214-------------------------------------------------------------------------------
216-- | XXX we can implement it more efficienty by directly implementing instead
217-- of combining streams using parallel.
218{-# INLINE consMParallel #-}
219{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
220consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
221consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
225-- | Execute two streams concurrently and merge their outputs. For example, if
226-- stream @a@ is a serial stream consisting of @a1, a2, a3@ and stream @b@ is a
227-- serial stream consisting of @b1, b2, b3@ then stream @a `parallel` b@ may
228-- produce @a1, b1, a2, b2, a3, b3@ or @a1, a2, b1, a3, b2, b3@ or some other
229-- combination depending on the rate at which the two streams produce elements.
230-- However, the relative order of outputs from a single stream e.g. @a1, a2,
231-- a3@ would remain the same in the resulting stream. The effects in the two
232-- streams may occur concurrently.
234-- To run single actions (instead of streams) in parallel wrap them into
235-- singleton streams. The following trivial example is semantically equivalent
236-- to running the action @putStrLn "hello"@ in the current thread:
238-- >>> Stream.toList $ Stream.yieldM (putStrLn "hello") `Stream.parallel` Stream.nil
242-- Run two actions concurrently:
244-- >>> import Control.Concurrent (threadDelay)
245-- >>> Stream.toList $ Stream.yieldM (putStrLn "hello") `Stream.parallel` Stream.yieldM (threadDelay 100000 >> putStrLn "world")
250-- Run effects concurrently, disregarding their outputs:
252-- >>> Stream.toList $ nilM (putStrLn "hello") `parallel` Stream.nilM (threadDelay 100000 >> putStrLn "world")
257-- Run an effectful action, and a pure effect without any output, concurrently:
259-- >>> Stream.toList $ Stream.yieldM (return 1) `Stream.parallel` Stream.nilM (putStrLn "world")
263-- Note that 'parallel' is a polymorphic version of the @Semigroup@ operation
264-- @<>@ of 'ParallelT'.
266-- `nilM` is currently @Internal@.
268-- /Since: 0.2.0 ("Streamly")/
271{-# INLINE parallel #-}
272parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
273parallel = joinStreamVarPar ParallelVar StopNone
275-- This is a co-parallel like combinator for streams, where first stream is the
276-- main stream and the rest are just supporting it, when the first ends
279-- | Like `parallel` but stops the output as soon as the first stream stops.
282{-# INLINE parallelFst #-}
283parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
284parallelFst = joinStreamVarPar ParallelVar StopBy
286-- This is a race like combinator for streams.
288-- | Like `parallel` but stops the output as soon as any of the two streams
292{-# INLINE parallelMin #-}
293parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
294parallelMin = joinStreamVarPar ParallelVar StopAny
296------------------------------------------------------------------------------
297-- Convert a stream to parallel
298------------------------------------------------------------------------------
300-- | Like 'mkParallel' but uses StreamK internally.
304mkParallelK :: (IsStream t, MonadAsync m) => t m a -> t m a
305mkParallelK m = mkStream $ \st yld sng stp -> do
306 sv <- newParallelVar StopNone (adaptState st)
307 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
308 SVar.toSVarParallel st sv $ D.toStreamD m
309 foldStream st yld sng stp $ SVar.fromSVar sv
311-------------------------------------------------------------------------------
312-- Concurrent application and fold
313-------------------------------------------------------------------------------
315-- | Same as 'mkParallel' but for StreamD stream.
317{-# INLINE_NORMAL mkParallelD #-}
318mkParallelD :: MonadAsync m => D.Stream m a -> D.Stream m a
319mkParallelD m = D.Stream step Nothing
322 step gst Nothing = do
323 sv <- newParallelVar StopNone gst
324 SVar.toSVarParallel gst sv m
325 -- XXX use unfold instead?
326 return $ Skip $ Just $ SVar.fromSVarD sv
328 step gst (Just (D.UnStream step1 st)) = do
331 Yield a s -> Yield a (Just $ D.Stream step1 s)
332 Skip s -> Skip (Just $ D.Stream step1 s)
335-- Compare with mkAsync. mkAsync uses an Async style SVar whereas this uses a
336-- parallel style SVar for evaluation. Currently, parallel style cannot use
337-- rate control whereas Async style can use rate control. In async style SVar
338-- the worker thread terminates when the buffer is full whereas in Parallel
341-- | Make the stream producer and consumer run concurrently by introducing a
342-- buffer between them. The producer thread evaluates the input stream until
343-- the buffer fills, it blocks if the buffer is full until there is space in
344-- the buffer. The consumer consumes the stream lazily from the buffer.
346-- @mkParallel = D.fromStreamD . mkParallelD . D.toStreamD@
350{-# INLINE_NORMAL mkParallel #-}
351mkParallel :: (K.IsStream t, MonadAsync m) => t m a -> t m a
352mkParallel = D.fromStreamD . mkParallelD . D.toStreamD
354-------------------------------------------------------------------------------
356-------------------------------------------------------------------------------
358-- NOTE: In regular pull style streams, the consumer stream is pulling elements
359-- from the SVar and we have several workers producing elements and pushing to
360-- SVar. In case of folds, we, the parent stream driving the fold, are the
361-- stream producing worker, we start an SVar and start pushing to the SVar, the
362-- fold on the other side of the SVar is the consumer stream.
364-- In the pull stream case exceptions are propagated from the producing workers
365-- to the consumer stream, the exceptions are propagated on the same channel as
366-- the produced stream elements. However, in case of push style folds the
367-- current stream itself is the worker and the fold is the consumer, in this
368-- case we have to propagate the exceptions from the consumer to the producer.
369-- This is reverse of the pull case and we need a reverse direction channel
370-- to propagate the exception.
372-- | Redirect a copy of the stream to a supplied fold and run it concurrently
373-- in an independent thread. The fold may buffer some elements. The buffer size
374-- is determined by the prevailing 'maxBuffer' setting.
379-- -----stream m a ---------------stream m a-----
384-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
389-- Exceptions from the concurrently running fold are propagated to the current
390-- computation. Note that, because of buffering in the fold, exceptions may be
391-- delayed and may not correspond to the current element being processed in the
392-- parent stream, but we guarantee that before the parent stream stops the tap
393-- finishes and all exceptions from it are drained.
396-- Compare with 'tap'.
399{-# INLINE tapAsync #-}
400tapAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
401tapAsync f m = mkStream $ \st yld sng stp -> do
402 sv <- SVar.newFoldSVar st f
403 foldStreamShared st yld sng stp (SVar.teeToSVar sv m)
405data TapState fs st a = TapInit | Tapping !fs st | TapDone st
407-- | Like 'tapAsync' but uses a 'Fold' instead of a fold function.
409{-# INLINE_NORMAL tapAsyncF #-}
410tapAsyncF :: MonadAsync m => Fold m a b -> D.Stream m a -> D.Stream m a
411tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit
415 -- In general, a Stop event would come equipped with the result
416 -- of the fold. It is not used here but it would be useful in
417 -- applicative and distribute.
418 done <- SVar.fromConsumer svr
420 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
421 $ takeMVar (outputDoorBellFromConsumer svr)
425 liftIO $ sendStop svr Nothing
426 -- drain/wait until a stop event arrives from the fold.
429 {-# INLINE_LATE step #-}
430 step gst TapInit = do
431 sv <- SVar.newFoldSVarF gst f
432 return $ Skip (Tapping sv state1)
434 step gst (Tapping sv st) = do
438 done <- SVar.pushToFold sv a
441 -- XXX we do not need to wait synchronously here
443 return $ Yield a (TapDone s)
444 else return $ Yield a (Tapping sv s)
445 Skip s -> return $ Skip (Tapping sv s)
450 step gst (TapDone st) = do
453 Yield a s -> Yield a (TapDone s)
454 Skip s -> Skip (TapDone s)
457-- | Concurrently distribute a stream to a collection of fold functions,
458-- discarding the outputs of the folds.
461-- > Stream.drain $ Stream.distributeAsync_ [Stream.mapM_ print, Stream.mapM_ print] (Stream.enumerateFromTo 1 2)
470-- distributeAsync_ = flip (foldr tapAsync)
475{-# INLINE distributeAsync_ #-}
476distributeAsync_ :: (Foldable f, IsStream t, MonadAsync m)
477 => f (t m a -> m b) -> t m a -> t m a
478distributeAsync_ = flip (foldr tapAsync)
480------------------------------------------------------------------------------
482------------------------------------------------------------------------------
484-- | Async composition with strict concurrent execution of all streams.
486-- The 'Semigroup' instance of 'ParallelT' executes both the streams
487-- concurrently without any delay or without waiting for the consumer demand
488-- and /merges/ the results as they arrive. If the consumer does not consume
489-- the results, they are buffered upto a configured maximum, controlled by the
490-- 'maxBuffer' primitive. If the buffer becomes full the concurrent tasks will
491-- block until there is space in the buffer.
493-- Both 'WAsyncT' and 'ParallelT', evaluate the constituent streams fairly in a
494-- round robin fashion. The key difference is that 'WAsyncT' might wait for the
495-- consumer demand before it executes the tasks whereas 'ParallelT' starts
496-- executing all the tasks immediately without waiting for the consumer demand.
497-- For 'WAsyncT' the 'maxThreads' limit applies whereas for 'ParallelT' it does
498-- not apply. In other words, 'WAsyncT' can be lazy whereas 'ParallelT' is
501-- 'ParallelT' is useful for cases when the streams are required to be
502-- evaluated simultaneously irrespective of how the consumer consumes them e.g.
503-- when we want to race two tasks and want to start both strictly at the same
504-- time or if we have timers in the parallel tasks and our results depend on
505-- the timers being started at the same time. If we do not have such
506-- requirements then 'AsyncT' or 'AheadT' are recommended as they can be more
507-- efficient than 'ParallelT'.
510-- main = (S.'toList' . S.'parallely' $ (S.fromFoldable [1,2]) \<> (S.fromFoldable [3,4])) >>= print
516-- When streams with more than one element are merged, it yields whichever
517-- stream yields first without any bias, unlike the 'Async' style streams.
519-- Any exceptions generated by a constituent stream are propagated to the
520-- output stream. The output and exceptions from a single stream are guaranteed
521-- to arrive in the same order in the resulting stream as they were generated
522-- in the input stream. However, the relative ordering of elements from
523-- different streams in the resulting stream can vary depending on scheduling
524-- and generation delays.
526-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
527-- of the loop concurrently.
530-- import qualified "Streamly.Prelude" as S
531-- import Control.Concurrent
533-- main = S.'drain' . S.'parallely' $ do
534-- n <- return 3 \<\> return 2 \<\> return 1
536-- threadDelay (n * 1000000)
537-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
540-- ThreadId 40: Delay 1
541-- ThreadId 39: Delay 2
542-- ThreadId 38: Delay 3
545-- Note that parallel composition can only combine a finite number of
546-- streams as it needs to retain state for each unfinished stream.
548-- /Since: 0.1.0 ("Streamly")/
550-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
553newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
554 deriving (MonadTrans)
556-- | A parallely composing IO stream of elements of type @a@.
557-- See 'ParallelT' documentation for more details.
559-- /Since: 0.2.0 ("Streamly")/
562type Parallel = ParallelT IO
564-- | Fix the type of a polymorphic stream as 'ParallelT'.
566-- /Since: 0.1.0 ("Streamly")/
569parallely :: IsStream t => ParallelT m a -> t m a
572instance IsStream ParallelT where
573 toStream = getParallelT
574 fromStream = ParallelT
577 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
578 consM = consMParallel
581 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
584------------------------------------------------------------------------------
586------------------------------------------------------------------------------
588{-# INLINE mappendParallel #-}
589{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
590mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
591mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
593instance MonadAsync m => Semigroup (ParallelT m a) where
594 (<>) = mappendParallel
596------------------------------------------------------------------------------
598------------------------------------------------------------------------------
600instance MonadAsync m => Monoid (ParallelT m a) where
604------------------------------------------------------------------------------
606------------------------------------------------------------------------------
608{-# INLINE apParallel #-}
609{-# SPECIALIZE apParallel :: ParallelT IO (a -> b) -> ParallelT IO a -> ParallelT IO b #-}
610apParallel :: MonadAsync m => ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b
611apParallel (ParallelT m1) (ParallelT m2) =
612 let f x1 = K.concatMapBy parallel (pure . x1) m2
613 in ParallelT $ K.concatMapBy parallel f m1
615instance (Monad m, MonadAsync m) => Applicative (ParallelT m) where
617 pure = ParallelT . K.yield
621------------------------------------------------------------------------------
623------------------------------------------------------------------------------
625{-# INLINE bindParallel #-}
626{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
627bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
628bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
630instance MonadAsync m => Monad (ParallelT m) where
634------------------------------------------------------------------------------
636------------------------------------------------------------------------------
638MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
640-------------------------------------------------------------------------------
642-------------------------------------------------------------------------------
644-- Note: we can use another API with two callbacks stop and yield if we want
645-- the callback to be able to indicate end of stream.
647-- | Generates a callback and a stream pair. The callback returned is used to
648-- queue values to the stream. The stream is infinite, there is no way for the
649-- callback to indicate that it is done now.
653{-# INLINE_NORMAL newCallbackStream #-}
654newCallbackStream :: (K.IsStream t, MonadAsync m) => m (a -> m (), t m a)
655newCallbackStream = do
656 sv <- newParallelVar StopNone defState
658 -- XXX Add our own thread-id to the SVar as we can not know the callback's
659 -- thread-id and the callback is not run in a managed worker. We need to
660 -- handle this better.
661 liftIO myThreadId >>= modifyThread sv
663 let callback a = liftIO $ void $ send sv (ChildYield a)
664 -- XXX we can return an SVar and then the consumer can unfold from the
666 return (callback, D.fromStreamD (SVar.fromSVarD sv))
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list