2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE UndecidableInstances #-} -- XXX
13-- Module : Streamly.Streams.Parallel
14-- Copyright : (c) 2017 Harendra Kumar
17-- Maintainer : streamly@composewell.com
18-- Stability : experimental
22module Streamly.Streams.Parallel
37import Control.Concurrent (myThreadId)
38import Control.Exception (SomeException(..), throwIO)
39import Control.Monad (ap)
40import Control.Monad.Base (MonadBase(..), liftBaseDefault)
41import Control.Monad.Catch (MonadThrow, throwM)
42-- import Control.Monad.Error.Class (MonadError(..))
43import Control.Monad.IO.Class (MonadIO(..))
44import Control.Monad.Reader.Class (MonadReader(..))
45import Control.Monad.State.Class (MonadState(..))
46import Control.Monad.Trans.Class (MonadTrans(lift))
47import Data.Functor (void)
48import Data.IORef (readIORef, writeIORef)
49import Data.Maybe (fromJust)
50#if __GLASGOW_HASKELL__ < 808
51import Data.Semigroup (Semigroup(..))
53import Prelude hiding (map)
55import qualified Data.Set as Set
57import Streamly.Internal.Data.Stream.SVar (fromSVar, fromStreamVar)
58import Streamly.Internal.Data.Stream.Serial (map)
59import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream, mkStream, foldStream,
60 foldStreamShared, adapt)
62import Streamly.Internal.Data.SVar hiding (handleChildException)
64import qualified Streamly.Internal.Data.Stream.StreamK as K
65import qualified Streamly.Internal.Data.Stream.StreamD as D
67#include "Instances.hs"
69-------------------------------------------------------------------------------
71-------------------------------------------------------------------------------
73-------------------------------------------------------------------------------
74-- StreamK based worker routines
75-------------------------------------------------------------------------------
77{-# NOINLINE runOne #-}
80 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
82 case getYieldLimit st of
84 Just _ -> runOneLimited st m0 winfo
89 liftIO $ decrementBufferLimit sv
90 foldStreamShared st yieldk single stop m
92 sv = fromJust $ streamVar st
95 incrementBufferLimit sv
97 sendit a = liftIO $ void $ send sv (ChildYield a)
98 single a = sendit a >> (liftIO $ sendStop sv winfo)
99 yieldk a r = sendit a >> go r
103 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
104runOneLimited st m0 winfo = go m0
109 yieldLimitOk <- liftIO $ decrementYieldLimit sv
112 liftIO $ decrementBufferLimit sv
113 foldStreamShared st yieldk single stop m
115 liftIO $ cleanupSVarFromWorker sv
116 liftIO $ sendStop sv winfo
118 sv = fromJust $ streamVar st
121 incrementBufferLimit sv
122 incrementYieldLimit sv
124 sendit a = liftIO $ void $ send sv (ChildYield a)
125 single a = sendit a >> (liftIO $ sendStop sv winfo)
126 yieldk a r = sendit a >> go r
128-------------------------------------------------------------------------------
129-- Consing and appending a stream in parallel style
130-------------------------------------------------------------------------------
132-- Note that consing and appending requires StreamK as it would not scale well
133-- with StreamD unless we are only consing a very small number of streams or
134-- elements in a stream. StreamK allows us to manipulate control flow in a way
135-- which StreamD cannot allow. StreamK can make a jump without having to
136-- remember the past state.
138{-# NOINLINE forkSVarPar #-}
139forkSVarPar :: (IsStream t, MonadAsync m)
140 => SVarStopStyle -> t m a -> t m a -> t m a
141forkSVarPar ss m r = mkStream $ \st yld sng stp -> do
142 sv <- newParallelVar ss st
143 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
145 StopBy -> liftIO $ do
146 set <- readIORef (workerThreads sv)
147 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
149 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
150 foldStream st yld sng stp (fromSVar sv)
152{-# INLINE joinStreamVarPar #-}
153joinStreamVarPar :: (IsStream t, MonadAsync m)
154 => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a
155joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->
157 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
158 -- Here, WE ARE IN THE WORKER/PRODUCER THREAD, we know that because
159 -- the SVar exists. We are running under runOne and the output we
160 -- produce ultimately will be sent to the SVar by runOne.
162 -- If we came here the worker/runOne is evaluating a `parallel`
163 -- combinator. In this case, we always fork a new worker for the
164 -- first component (m1) in the parallel composition and continue to
165 -- evaluate the second component (m2) in the current worker thread.
167 -- When m1 is serially composed, the worker would evaluate it
168 -- without any further forks and the resulting output is sent to
169 -- the SVar and the evaluation terminates. If m1 is a `parallel`
170 -- composition of two streams the worker would again recurses here.
172 -- Similarly, when m2 is serially composed it gets evaluated here
173 -- and the resulting output is sent to the SVar by the runOne
174 -- wrapper. When m2 is composed with `parallel` it will again
175 -- recurse here and so on until it finally terminates.
177 -- When we create a right associated expression using `parallel`,
178 -- then m1 would always terminate without further forks or
179 -- recursion into this routine, therefore, the worker returns
180 -- immediately after evaluating it. And m2 would continue to
181 -- fork/recurse, therefore, the current thread always recurses and
182 -- forks new workers one after the other. This is a tail recursive
183 -- style execution, m2, the recursive expression always executed at
186 -- When the expression is left associated, the worker spawned would
187 -- get the forking/recursing responsibility and then again the
188 -- worker spawned by that worker would fork, thus creating layer
189 -- over layer of workers and a chain of threads leading to a very
190 -- inefficient execution.
191 pushWorkerPar sv (runOne st $ toStream m1)
192 foldStreamShared st yld sng stp m2
194 -- Here WE ARE IN THE CONSUMER THREAD, we create a new SVar, fork
195 -- worker threads to execute m1 and m2 and this thread starts
196 -- pulling the stream from the SVar.
197 foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
199-------------------------------------------------------------------------------
201-------------------------------------------------------------------------------
203-- | XXX we can implement it more efficienty by directly implementing instead
204-- of combining streams using parallel.
205{-# INLINE consMParallel #-}
206{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
207consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
208consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
210-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
211-- Merges two streams concurrently.
214{-# INLINE parallel #-}
215parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
216parallel = joinStreamVarPar ParallelVar StopNone
218-- This is a co-parallel like combinator for streams, where first stream is the
219-- main stream and the rest are just supporting it, when the first ends
222-- | Like `parallel` but stops the output as soon as the first stream stops.
225{-# INLINE parallelFst #-}
226parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
227parallelFst = joinStreamVarPar ParallelVar StopBy
229-- This is a race like combinator for streams.
231-- | Like `parallel` but stops the output as soon as any of the two streams
235{-# INLINE parallelMin #-}
236parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
237parallelMin = joinStreamVarPar ParallelVar StopAny
239------------------------------------------------------------------------------
240-- Convert a stream to parallel
241------------------------------------------------------------------------------
243mkParallel :: (IsStream t, MonadAsync m)
244 => State Stream m a -> t m a -> m (t m a)
246 sv <- newParallelVar StopNone defState
247 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
248 D.toSVarParallel st sv $ D.toStreamD m
251------------------------------------------------------------------------------
252-- Stream to stream concurrent function application
253------------------------------------------------------------------------------
255{-# INLINE applyParallel #-}
256applyParallel :: (IsStream t, MonadAsync m)
257 => (t m a -> t m b) -> t m a -> t m b
258applyParallel f m = mkStream $ \st yld sng stp -> do
259 sv <- newParallelVar StopNone (adaptState st)
260 -- pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
261 D.toSVarParallel (adaptState st) sv $ D.toStreamD m
262 foldStream st yld sng stp $ f $ fromSVar sv
264------------------------------------------------------------------------------
265-- Stream runner concurrent function application
266------------------------------------------------------------------------------
268{-# INLINE foldParallel #-}
269foldParallel :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
271 sv <- newParallelVar StopNone defState
272 -- pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
273 D.toSVarParallel defState sv $ D.toStreamD m
276------------------------------------------------------------------------------
278------------------------------------------------------------------------------
280{-# INLINE teeToSVar #-}
281teeToSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> t m a
282teeToSVar svr m = mkStream $ \st yld sng stp -> do
283 foldStreamShared st yld sng stp (go svr m)
287 go sv m0 = mkStream $ \st yld sng stp -> do
290 incrementBufferLimit sv
292 -- XXX wait for Stop response event on exception channel,
293 -- drain the exception channel.
295 sendit a = liftIO $ do
296 -- XXX check for exceptions before decrement so that we do not
297 -- block forever if the child already exited with an exception.
298 decrementBufferLimit sv
299 void $ send sv (ChildYield a)
300 single a = sendit a >> (liftIO $ sendStop sv Nothing) >> sng a
301 yieldk a r = sendit a >> yld a (go sv r)
302 in foldStreamShared st yieldk single stop m0
304-- In case of folds the roles of worker and parent on an SVar are reversed. The
305-- parent stream pushes values to an SVar instead of pulling from it and a
306-- worker thread running the fold pulls from the SVar and folds the stream. We
307-- keep a separate channel for pushing exceptions in the reverse direction i.e.
308-- from the fold to the parent stream.
310-- NOTE: If we use fromSVar here it will kill the main computation (the parent)
311-- when the SVar goes away so we use fromStreamVar instead.
313{-# NOINLINE handleChildException #-}
314handleChildException :: SVar t m a -> SomeException -> IO ()
315handleChildException _sv e = do
317 -- void $ sendReverse sv (ChildStop tid (Just e))
320-- | Redirect a copy of the stream to a supplied fold and run it concurrently
321-- in an independent thread. The fold may buffer some elements. The buffer size
322-- is determined by the prevailing 'maxBuffer' setting.
327-- -----stream m a ---------------stream m a-----
332-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
337-- Exceptions from the concurrently running fold are propagated to the current
338-- computation. Note that, because of buffering in the fold, exceptions may be
339-- delayed and may not correspond to the current element being processed in the
340-- parent stream, but we guarantee that the tap finishes and all exceptions
341-- from it are drained before the parent stream stops.
344-- Compare with 'tap'.
347{-# INLINE tapAsync #-}
348tapAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
349tapAsync f m = mkStream $ \st yld sng stp -> do
350 -- Buffer size for the SVar is derived from the current state
351 sv <- newParallelVar StopNone (adaptState st)
352 -- XXX exception handling
353 -- XXX if we terminate due to an exception, do we need to actively
354 -- terminate the fold?
355 liftIO myThreadId >>= modifyThread sv
356 void $ doFork (void $ f $ fromStream $ fromStreamVar sv)
357 (svarMrun sv) (handleChildException sv)
358 foldStreamShared st yld sng stp (teeToSVar sv m)
360------------------------------------------------------------------------------
362------------------------------------------------------------------------------
364-- | Async composition with strict concurrent execution of all streams.
366-- The 'Semigroup' instance of 'ParallelT' executes both the streams
367-- concurrently without any delay or without waiting for the consumer demand
368-- and /merges/ the results as they arrive. If the consumer does not consume
369-- the results, they are buffered upto a configured maximum, controlled by the
370-- 'maxBuffer' primitive. If the buffer becomes full the concurrent tasks will
371-- block until there is space in the buffer.
373-- Both 'WAsyncT' and 'ParallelT', evaluate the constituent streams fairly in a
374-- round robin fashion. The key difference is that 'WAsyncT' might wait for the
375-- consumer demand before it executes the tasks whereas 'ParallelT' starts
376-- executing all the tasks immediately without waiting for the consumer demand.
377-- For 'WAsyncT' the 'maxThreads' limit applies whereas for 'ParallelT' it does
378-- not apply. In other words, 'WAsyncT' can be lazy whereas 'ParallelT' is
381-- 'ParallelT' is useful for cases when the streams are required to be
382-- evaluated simultaneously irrespective of how the consumer consumes them e.g.
383-- when we want to race two tasks and want to start both strictly at the same
384-- time or if we have timers in the parallel tasks and our results depend on
385-- the timers being started at the same time. If we do not have such
386-- requirements then 'AsyncT' or 'AheadT' are recommended as they can be more
387-- efficient than 'ParallelT'.
390-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
396-- When streams with more than one element are merged, it yields whichever
397-- stream yields first without any bias, unlike the 'Async' style streams.
399-- Any exceptions generated by a constituent stream are propagated to the
400-- output stream. The output and exceptions from a single stream are guaranteed
401-- to arrive in the same order in the resulting stream as they were generated
402-- in the input stream. However, the relative ordering of elements from
403-- different streams in the resulting stream can vary depending on scheduling
404-- and generation delays.
406-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
407-- of the loop concurrently.
411-- import qualified "Streamly.Prelude" as S
412-- import Control.Concurrent
414-- main = 'drain' . 'parallely' $ do
415-- n <- return 3 \<\> return 2 \<\> return 1
417-- threadDelay (n * 1000000)
418-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
421-- ThreadId 40: Delay 1
422-- ThreadId 39: Delay 2
423-- ThreadId 38: Delay 3
426-- Note that parallel composition can only combine a finite number of
427-- streams as it needs to retain state for each unfinished stream.
429-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
432newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
433 deriving (MonadTrans)
435-- | A parallely composing IO stream of elements of type @a@.
436-- See 'ParallelT' documentation for more details.
439type Parallel = ParallelT IO
441-- | Fix the type of a polymorphic stream as 'ParallelT'.
444parallely :: IsStream t => ParallelT m a -> t m a
447instance IsStream ParallelT where
448 toStream = getParallelT
449 fromStream = ParallelT
452 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
453 consM = consMParallel
456 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
459------------------------------------------------------------------------------
461------------------------------------------------------------------------------
463{-# INLINE mappendParallel #-}
464{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
465mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
466mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
468instance MonadAsync m => Semigroup (ParallelT m a) where
469 (<>) = mappendParallel
471------------------------------------------------------------------------------
473------------------------------------------------------------------------------
475instance MonadAsync m => Monoid (ParallelT m a) where
479------------------------------------------------------------------------------
481------------------------------------------------------------------------------
483{-# INLINE bindParallel #-}
484{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
485bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
486bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
488instance MonadAsync m => Monad (ParallelT m) where
492------------------------------------------------------------------------------
494------------------------------------------------------------------------------
496MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
497MONAD_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