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
36import Control.Concurrent (myThreadId, takeMVar)
37import Control.Monad (when)
38import Control.Monad.Base (MonadBase(..), liftBaseDefault)
39import Control.Monad.Catch (MonadThrow, throwM)
40-- import Control.Monad.Error.Class (MonadError(..))
41import Control.Monad.IO.Class (MonadIO(..))
42import Control.Monad.Reader.Class (MonadReader(..))
43import Control.Monad.State.Class (MonadState(..))
44import Control.Monad.Trans.Class (MonadTrans(lift))
45import Data.Functor (void)
46import Data.IORef (readIORef, writeIORef)
47import Data.Maybe (fromJust)
48#if __GLASGOW_HASKELL__ < 808
49import Data.Semigroup (Semigroup(..))
51import Prelude hiding (map)
53import qualified Data.Set as Set
55import Streamly.Internal.Data.Stream.SVar
56 (fromSVar, fromProducer, fromConsumer, pushToFold)
57import Streamly.Internal.Data.Stream.StreamK
58 (IsStream(..), Stream, mkStream, foldStream, foldStreamShared, adapt)
60import Streamly.Internal.Data.SVar
62import qualified Streamly.Internal.Data.Stream.StreamK as K
63import qualified Streamly.Internal.Data.Stream.StreamD as D
65#include "Instances.hs"
67-------------------------------------------------------------------------------
69-------------------------------------------------------------------------------
71-------------------------------------------------------------------------------
72-- StreamK based worker routines
73-------------------------------------------------------------------------------
75{-# NOINLINE runOne #-}
78 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
80 case getYieldLimit st of
82 Just _ -> runOneLimited st m0 winfo
87 liftIO $ decrementBufferLimit sv
88 foldStreamShared st yieldk single stop m
90 sv = fromJust $ streamVar st
93 incrementBufferLimit sv
95 sendit a = liftIO $ void $ send sv (ChildYield a)
96 single a = sendit a >> (liftIO $ sendStop sv winfo)
97 yieldk a r = sendit a >> go r
101 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
102runOneLimited st m0 winfo = go m0
107 yieldLimitOk <- liftIO $ decrementYieldLimit sv
110 liftIO $ decrementBufferLimit sv
111 foldStreamShared st yieldk single stop m
113 liftIO $ cleanupSVarFromWorker sv
114 liftIO $ sendStop sv winfo
116 sv = fromJust $ streamVar st
119 incrementBufferLimit sv
120 incrementYieldLimit sv
122 sendit a = liftIO $ void $ send sv (ChildYield a)
123 single a = sendit a >> (liftIO $ sendStop sv winfo)
124 yieldk a r = sendit a >> go r
126-------------------------------------------------------------------------------
127-- Consing and appending a stream in parallel style
128-------------------------------------------------------------------------------
130-- Note that consing and appending requires StreamK as it would not scale well
131-- with StreamD unless we are only consing a very small number of streams or
132-- elements in a stream. StreamK allows us to manipulate control flow in a way
133-- which StreamD cannot allow. StreamK can make a jump without having to
134-- remember the past state.
136{-# NOINLINE forkSVarPar #-}
137forkSVarPar :: (IsStream t, MonadAsync m)
138 => SVarStopStyle -> t m a -> t m a -> t m a
139forkSVarPar ss m r = mkStream $ \st yld sng stp -> do
140 sv <- newParallelVar ss st
141 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
143 StopBy -> liftIO $ do
144 set <- readIORef (workerThreads sv)
145 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
147 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
148 foldStream st yld sng stp (fromSVar sv)
150{-# INLINE joinStreamVarPar #-}
151joinStreamVarPar :: (IsStream t, MonadAsync m)
152 => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a
153joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->
155 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
156 -- Here, WE ARE IN THE WORKER/PRODUCER THREAD, we know that because
157 -- the SVar exists. We are running under runOne and the output we
158 -- produce ultimately will be sent to the SVar by runOne.
160 -- If we came here the worker/runOne is evaluating a `parallel`
161 -- combinator. In this case, we always fork a new worker for the
162 -- first component (m1) in the parallel composition and continue to
163 -- evaluate the second component (m2) in the current worker thread.
165 -- When m1 is serially composed, the worker would evaluate it
166 -- without any further forks and the resulting output is sent to
167 -- the SVar and the evaluation terminates. If m1 is a `parallel`
168 -- composition of two streams the worker would again recurses here.
170 -- Similarly, when m2 is serially composed it gets evaluated here
171 -- and the resulting output is sent to the SVar by the runOne
172 -- wrapper. When m2 is composed with `parallel` it will again
173 -- recurse here and so on until it finally terminates.
175 -- When we create a right associated expression using `parallel`,
176 -- then m1 would always terminate without further forks or
177 -- recursion into this routine, therefore, the worker returns
178 -- immediately after evaluating it. And m2 would continue to
179 -- fork/recurse, therefore, the current thread always recurses and
180 -- forks new workers one after the other. This is a tail recursive
181 -- style execution, m2, the recursive expression always executed at
184 -- When the expression is left associated, the worker spawned would
185 -- get the forking/recursing responsibility and then again the
186 -- worker spawned by that worker would fork, thus creating layer
187 -- over layer of workers and a chain of threads leading to a very
188 -- inefficient execution.
189 pushWorkerPar sv (runOne st $ toStream m1)
190 foldStreamShared st yld sng stp m2
192 -- Here WE ARE IN THE CONSUMER THREAD, we create a new SVar, fork
193 -- worker threads to execute m1 and m2 and this thread starts
194 -- pulling the stream from the SVar.
195 foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
197-------------------------------------------------------------------------------
199-------------------------------------------------------------------------------
201-- | XXX we can implement it more efficienty by directly implementing instead
202-- of combining streams using parallel.
203{-# INLINE consMParallel #-}
204{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
205consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
206consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
210-- | Execute two streams concurrently and merge their outputs. For example, if
211-- stream @a@ is a serial stream consisting of @a1, a2, a3@ and stream @b@ is a
212-- serial stream consisting of @b1, b2, b3@ then stream @a `parallel` b@ may
213-- produce @a1, b1, a2, b2, a3, b3@ or @a1, a2, b1, a3, b2, b3@ or some other
214-- combination depending on the rate at which the two streams produce elements.
215-- However, the relative order of outputs from a single stream e.g. @a1, a2,
216-- a3@ would remain the same in the resulting stream. The effects in the two
217-- streams may occur concurrently.
219-- To run single actions (instead of streams) in parallel wrap them into
220-- singleton streams. The following trivial example is semantically equivalent
221-- to running the action @putStrLn "hello"@ in the current thread:
223-- >>> S.toList $ S.yieldM (putStrLn "hello") `parallel` S.nil
227-- Run two actions concurrently:
229-- >>> S.toList $ S.yieldM (putStrLn "hello") `parallel` S.yieldM (putStrLn "world")
234-- Run effects concurrently, disregarding their outputs:
236-- >>> S.toList $ S.nilM (putStrLn "hello") `parallel` S.nilM (putStrLn "world")
241-- Run an effectful action, and a pure effect without any output, concurrently:
243-- >>> S.toList $ S.yieldM (return 1) `parallel` S.nilM (putStrLn "world")
247-- Note that 'parallel' is a polymorphic version of the @Semigroup@ operation
248-- @<>@ of 'ParallelT'.
250-- `nilM` is currently @Internal@.
252-- /Since: 0.2.0 ("Streamly")/
255{-# INLINE parallel #-}
256parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
257parallel = joinStreamVarPar ParallelVar StopNone
259-- This is a co-parallel like combinator for streams, where first stream is the
260-- main stream and the rest are just supporting it, when the first ends
263-- | Like `parallel` but stops the output as soon as the first stream stops.
266{-# INLINE parallelFst #-}
267parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
268parallelFst = joinStreamVarPar ParallelVar StopBy
270-- This is a race like combinator for streams.
272-- | Like `parallel` but stops the output as soon as any of the two streams
276{-# INLINE parallelMin #-}
277parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
278parallelMin = joinStreamVarPar ParallelVar StopAny
280------------------------------------------------------------------------------
281-- Convert a stream to parallel
282------------------------------------------------------------------------------
284-- | Generate a stream asynchronously to keep it buffered, lazily consume
289mkParallel :: (IsStream t, MonadAsync m) => t m a -> t m a
290mkParallel m = mkStream $ \st yld sng stp -> do
291 sv <- newParallelVar StopNone (adaptState st)
292 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
293 D.toSVarParallel st sv $ D.toStreamD m
294 foldStream st yld sng stp $ fromSVar sv
296------------------------------------------------------------------------------
297-- Clone and distribute a stream in parallel
298------------------------------------------------------------------------------
300-- Tap a stream and send the elements to the specified SVar in addition to
301-- yielding them again.
303-- XXX this could be written in StreamD style for better efficiency with fusion.
305{-# INLINE teeToSVar #-}
306teeToSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> t m a
307teeToSVar svr m = mkStream $ \st yld sng stp -> do
308 foldStreamShared st yld sng stp (go False m)
312 go False m0 = mkStream $ \st yld _ stp -> do
314 -- In general, a Stop event would come equipped with the result
315 -- of the fold. It is not used here but it would be useful in
316 -- applicative and distribute.
317 done <- fromConsumer svr
319 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
320 $ takeMVar (outputDoorBellFromConsumer svr)
324 liftIO $ sendStop svr Nothing
325 -- drain/wait until a stop event arrives from the fold.
328 stop = stopFold >> stp
330 done <- pushToFold svr a
331 yld a (go done (K.nilM stopFold))
332 yieldk a r = pushToFold svr a >>= \done -> yld a (go done r)
333 in foldStreamShared st yieldk single stop m0
337-- In case of folds the roles of worker and parent on an SVar are reversed. The
338-- parent stream pushes values to an SVar instead of pulling from it and a
339-- worker thread running the fold pulls from the SVar and folds the stream. We
340-- keep a separate channel for pushing exceptions in the reverse direction i.e.
341-- from the fold to the parent stream.
343-- Note: If we terminate due to an exception, we do not actively terminate the
344-- fold. It gets cleaned up by the GC.
346-- | Create an SVar with a fold consumer that will fold any elements sent to it
347-- using the supplied fold function.
348{-# INLINE newFoldSVar #-}
349newFoldSVar :: (IsStream t, MonadAsync m)
350 => State Stream m a -> (t m a -> m b) -> m (SVar Stream m a)
351newFoldSVar stt f = do
352 -- Buffer size for the SVar is derived from the current state
353 sv <- newParallelVar StopAny (adaptState stt)
355 -- Add the producer thread-id to the SVar.
356 liftIO myThreadId >>= modifyThread sv
358 void $ doFork (void $ f $ fromStream $ fromProducer sv)
360 (handleFoldException sv)
363-- NOTE: In regular pull style streams, the consumer stream is pulling elements
364-- from the SVar and we have several workers producing elements and pushing to
365-- SVar. In case of folds, we, the parent stream driving the fold, are the
366-- stream producing worker, we start an SVar and start pushing to the SVar, the
367-- fold on the other side of the SVar is the consumer stream.
369-- In the pull stream case exceptions are propagated from the producing workers
370-- to the consumer stream, the exceptions are propagated on the same channel as
371-- the produced stream elements. However, in case of push style folds the
372-- current stream itself is the worker and the fold is the consumer, in this
373-- case we have to propagate the exceptions from the consumer to the producer.
374-- This is reverse of the pull case and we need a reverse direction channel
375-- to propagate the exception.
377-- | Redirect a copy of the stream to a supplied fold and run it concurrently
378-- in an independent thread. The fold may buffer some elements. The buffer size
379-- is determined by the prevailing 'maxBuffer' setting.
384-- -----stream m a ---------------stream m a-----
389-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
394-- Exceptions from the concurrently running fold are propagated to the current
395-- computation. Note that, because of buffering in the fold, exceptions may be
396-- delayed and may not correspond to the current element being processed in the
397-- parent stream, but we guarantee that before the parent stream stops the tap
398-- finishes and all exceptions from it are drained.
401-- Compare with 'tap'.
404{-# INLINE tapAsync #-}
405tapAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
406tapAsync f m = mkStream $ \st yld sng stp -> do
407 sv <- newFoldSVar st f
408 foldStreamShared st yld sng stp (teeToSVar sv m)
410-- | Concurrently distribute a stream to a collection of fold functions,
411-- discarding the outputs of the folds.
413-- >>> S.drain $ distributeAsync_ [S.mapM_ print, S.mapM_ print] (S.enumerateFromTo 1 2)
416-- distributeAsync_ = flip (foldr tapAsync)
421{-# INLINE distributeAsync_ #-}
422distributeAsync_ :: (Foldable f, IsStream t, MonadAsync m)
423 => f (t m a -> m b) -> t m a -> t m a
424distributeAsync_ = flip (foldr tapAsync)
426------------------------------------------------------------------------------
428------------------------------------------------------------------------------
430-- | Async composition with strict concurrent execution of all streams.
432-- The 'Semigroup' instance of 'ParallelT' executes both the streams
433-- concurrently without any delay or without waiting for the consumer demand
434-- and /merges/ the results as they arrive. If the consumer does not consume
435-- the results, they are buffered upto a configured maximum, controlled by the
436-- 'maxBuffer' primitive. If the buffer becomes full the concurrent tasks will
437-- block until there is space in the buffer.
439-- Both 'WAsyncT' and 'ParallelT', evaluate the constituent streams fairly in a
440-- round robin fashion. The key difference is that 'WAsyncT' might wait for the
441-- consumer demand before it executes the tasks whereas 'ParallelT' starts
442-- executing all the tasks immediately without waiting for the consumer demand.
443-- For 'WAsyncT' the 'maxThreads' limit applies whereas for 'ParallelT' it does
444-- not apply. In other words, 'WAsyncT' can be lazy whereas 'ParallelT' is
447-- 'ParallelT' is useful for cases when the streams are required to be
448-- evaluated simultaneously irrespective of how the consumer consumes them e.g.
449-- when we want to race two tasks and want to start both strictly at the same
450-- time or if we have timers in the parallel tasks and our results depend on
451-- the timers being started at the same time. If we do not have such
452-- requirements then 'AsyncT' or 'AheadT' are recommended as they can be more
453-- efficient than 'ParallelT'.
456-- main = (S.'toList' . S.'parallely' $ (S.fromFoldable [1,2]) \<> (S.fromFoldable [3,4])) >>= print
462-- When streams with more than one element are merged, it yields whichever
463-- stream yields first without any bias, unlike the 'Async' style streams.
465-- Any exceptions generated by a constituent stream are propagated to the
466-- output stream. The output and exceptions from a single stream are guaranteed
467-- to arrive in the same order in the resulting stream as they were generated
468-- in the input stream. However, the relative ordering of elements from
469-- different streams in the resulting stream can vary depending on scheduling
470-- and generation delays.
472-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
473-- of the loop concurrently.
476-- import qualified "Streamly.Prelude" as S
477-- import Control.Concurrent
479-- main = S.'drain' . S.'parallely' $ do
480-- n <- return 3 \<\> return 2 \<\> return 1
482-- threadDelay (n * 1000000)
483-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
486-- ThreadId 40: Delay 1
487-- ThreadId 39: Delay 2
488-- ThreadId 38: Delay 3
491-- Note that parallel composition can only combine a finite number of
492-- streams as it needs to retain state for each unfinished stream.
494-- /Since: 0.1.0 ("Streamly")/
496-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
499newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
500 deriving (MonadTrans)
502-- | A parallely composing IO stream of elements of type @a@.
503-- See 'ParallelT' documentation for more details.
505-- /Since: 0.2.0 ("Streamly")/
508type Parallel = ParallelT IO
510-- | Fix the type of a polymorphic stream as 'ParallelT'.
512-- /Since: 0.1.0 ("Streamly")/
515parallely :: IsStream t => ParallelT m a -> t m a
518instance IsStream ParallelT where
519 toStream = getParallelT
520 fromStream = ParallelT
523 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
524 consM = consMParallel
527 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
530------------------------------------------------------------------------------
532------------------------------------------------------------------------------
534{-# INLINE mappendParallel #-}
535{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
536mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
537mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
539instance MonadAsync m => Semigroup (ParallelT m a) where
540 (<>) = mappendParallel
542------------------------------------------------------------------------------
544------------------------------------------------------------------------------
546instance MonadAsync m => Monoid (ParallelT m a) where
550------------------------------------------------------------------------------
552------------------------------------------------------------------------------
554{-# INLINE apParallel #-}
555{-# SPECIALIZE apParallel :: ParallelT IO (a -> b) -> ParallelT IO a -> ParallelT IO b #-}
556apParallel :: MonadAsync m => ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b
557apParallel (ParallelT m1) (ParallelT m2) =
558 let f x1 = K.concatMapBy parallel (pure . x1) m2
559 in ParallelT $ K.concatMapBy parallel f m1
561instance (Monad m, MonadAsync m) => Applicative (ParallelT m) where
563 pure = ParallelT . K.yield
567------------------------------------------------------------------------------
569------------------------------------------------------------------------------
571{-# INLINE bindParallel #-}
572{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
573bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
574bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
576instance MonadAsync m => Monad (ParallelT m) where
580------------------------------------------------------------------------------
582------------------------------------------------------------------------------
584MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
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