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.Internal.Data.Stream.Parallel
14-- Copyright : (c) 2017 Harendra Kumar
17-- Maintainer : streamly@composewell.com
18-- Stability : experimental
22module Streamly.Internal.Data.Stream.Parallel
37import Control.Concurrent (myThreadId, takeMVar)
38import Control.Exception (SomeException(..))
39import Control.Monad (ap, when)
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
58 (fromSVar, fromProducer, fromConsumer)
59import Streamly.Internal.Data.Stream.Serial (map)
60import Streamly.Internal.Data.Stream.StreamK
61 (IsStream(..), Stream, mkStream, foldStream, foldStreamShared, adapt)
63import Streamly.Internal.Data.SVar
65import qualified Streamly.Internal.Data.Stream.StreamK as K
66import qualified Streamly.Internal.Data.Stream.StreamD as D
68#include "Instances.hs"
70-------------------------------------------------------------------------------
72-------------------------------------------------------------------------------
74-------------------------------------------------------------------------------
75-- StreamK based worker routines
76-------------------------------------------------------------------------------
78{-# NOINLINE runOne #-}
81 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
83 case getYieldLimit st of
85 Just _ -> runOneLimited st m0 winfo
90 liftIO $ decrementBufferLimit sv
91 foldStreamShared st yieldk single stop m
93 sv = fromJust $ streamVar st
96 incrementBufferLimit sv
98 sendit a = liftIO $ void $ send sv (ChildYield a)
99 single a = sendit a >> (liftIO $ sendStop sv winfo)
100 yieldk a r = sendit a >> go r
104 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
105runOneLimited st m0 winfo = go m0
110 yieldLimitOk <- liftIO $ decrementYieldLimit sv
113 liftIO $ decrementBufferLimit sv
114 foldStreamShared st yieldk single stop m
116 liftIO $ cleanupSVarFromWorker sv
117 liftIO $ sendStop sv winfo
119 sv = fromJust $ streamVar st
122 incrementBufferLimit sv
123 incrementYieldLimit sv
125 sendit a = liftIO $ void $ send sv (ChildYield a)
126 single a = sendit a >> (liftIO $ sendStop sv winfo)
127 yieldk a r = sendit a >> go r
129-------------------------------------------------------------------------------
130-- Consing and appending a stream in parallel style
131-------------------------------------------------------------------------------
133-- Note that consing and appending requires StreamK as it would not scale well
134-- with StreamD unless we are only consing a very small number of streams or
135-- elements in a stream. StreamK allows us to manipulate control flow in a way
136-- which StreamD cannot allow. StreamK can make a jump without having to
137-- remember the past state.
139{-# NOINLINE forkSVarPar #-}
140forkSVarPar :: (IsStream t, MonadAsync m)
141 => SVarStopStyle -> t m a -> t m a -> t m a
142forkSVarPar ss m r = mkStream $ \st yld sng stp -> do
143 sv <- newParallelVar ss st
144 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
146 StopBy -> liftIO $ do
147 set <- readIORef (workerThreads sv)
148 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
150 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
151 foldStream st yld sng stp (fromSVar sv)
153{-# INLINE joinStreamVarPar #-}
154joinStreamVarPar :: (IsStream t, MonadAsync m)
155 => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a
156joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->
158 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
159 -- Here, WE ARE IN THE WORKER/PRODUCER THREAD, we know that because
160 -- the SVar exists. We are running under runOne and the output we
161 -- produce ultimately will be sent to the SVar by runOne.
163 -- If we came here the worker/runOne is evaluating a `parallel`
164 -- combinator. In this case, we always fork a new worker for the
165 -- first component (m1) in the parallel composition and continue to
166 -- evaluate the second component (m2) in the current worker thread.
168 -- When m1 is serially composed, the worker would evaluate it
169 -- without any further forks and the resulting output is sent to
170 -- the SVar and the evaluation terminates. If m1 is a `parallel`
171 -- composition of two streams the worker would again recurses here.
173 -- Similarly, when m2 is serially composed it gets evaluated here
174 -- and the resulting output is sent to the SVar by the runOne
175 -- wrapper. When m2 is composed with `parallel` it will again
176 -- recurse here and so on until it finally terminates.
178 -- When we create a right associated expression using `parallel`,
179 -- then m1 would always terminate without further forks or
180 -- recursion into this routine, therefore, the worker returns
181 -- immediately after evaluating it. And m2 would continue to
182 -- fork/recurse, therefore, the current thread always recurses and
183 -- forks new workers one after the other. This is a tail recursive
184 -- style execution, m2, the recursive expression always executed at
187 -- When the expression is left associated, the worker spawned would
188 -- get the forking/recursing responsibility and then again the
189 -- worker spawned by that worker would fork, thus creating layer
190 -- over layer of workers and a chain of threads leading to a very
191 -- inefficient execution.
192 pushWorkerPar sv (runOne st $ toStream m1)
193 foldStreamShared st yld sng stp m2
195 -- Here WE ARE IN THE CONSUMER THREAD, we create a new SVar, fork
196 -- worker threads to execute m1 and m2 and this thread starts
197 -- pulling the stream from the SVar.
198 foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
200-------------------------------------------------------------------------------
202-------------------------------------------------------------------------------
204-- | XXX we can implement it more efficienty by directly implementing instead
205-- of combining streams using parallel.
206{-# INLINE consMParallel #-}
207{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
208consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
209consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
211-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
212-- Merges two streams concurrently.
215{-# INLINE parallel #-}
216parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
217parallel = joinStreamVarPar ParallelVar StopNone
219-- This is a co-parallel like combinator for streams, where first stream is the
220-- main stream and the rest are just supporting it, when the first ends
223-- | Like `parallel` but stops the output as soon as the first stream stops.
226{-# INLINE parallelFst #-}
227parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
228parallelFst = joinStreamVarPar ParallelVar StopBy
230-- This is a race like combinator for streams.
232-- | Like `parallel` but stops the output as soon as any of the two streams
236{-# INLINE parallelMin #-}
237parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
238parallelMin = joinStreamVarPar ParallelVar StopAny
240------------------------------------------------------------------------------
241-- Convert a stream to parallel
242------------------------------------------------------------------------------
244mkParallel :: (IsStream t, MonadAsync m)
245 => State Stream m a -> t m a -> m (t m a)
247 sv <- newParallelVar StopNone defState
248 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
249 D.toSVarParallel st sv $ D.toStreamD m
252------------------------------------------------------------------------------
253-- Stream to stream concurrent function application
254------------------------------------------------------------------------------
256{-# INLINE applyParallel #-}
257applyParallel :: (IsStream t, MonadAsync m)
258 => (t m a -> t m b) -> t m a -> t m b
259applyParallel f m = mkStream $ \st yld sng stp -> do
260 sv <- newParallelVar StopNone (adaptState st)
261 -- pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
262 D.toSVarParallel (adaptState st) sv $ D.toStreamD m
263 foldStream st yld sng stp $ f $ fromSVar sv
265------------------------------------------------------------------------------
266-- Stream runner concurrent function application
267------------------------------------------------------------------------------
269{-# INLINE foldParallel #-}
270foldParallel :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
272 sv <- newParallelVar StopNone defState
273 -- pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
274 D.toSVarParallel defState sv $ D.toStreamD m
277------------------------------------------------------------------------------
278-- Clone and distribute a stream in parallel
279------------------------------------------------------------------------------
281-- push values to a fold worker via an SVar. Returns whether the fold is done.
282{-# INLINE pushToFold #-}
283pushToFold :: MonadAsync m => SVar Stream m a -> a -> m Bool
285 -- Check for exceptions before decrement so that we do not
286 -- block forever if the child already exited with an exception.
288 -- We avoid a race between the consumer fold sending an event and we
289 -- blocking on decrementBufferLimit by waking up the producer thread in
290 -- sendToProducer before any event is sent by the fold to the producer
292 let qref = outputQueueFromConsumer sv
294 (_, n) <- liftIO $ readIORef qref
301 decrementBufferLimit sv
302 void $ send sv (ChildYield a)
305-- Tap a stream and send the elements to the specified SVar in addition to
306-- yielding them again.
308-- XXX this could be written in StreamD style for better efficiency with fusion.
310{-# INLINE teeToSVar #-}
311teeToSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> t m a
312teeToSVar svr m = mkStream $ \st yld sng stp -> do
313 foldStreamShared st yld sng stp (go False m)
317 go False m0 = mkStream $ \st yld _ stp -> do
319 -- In general, a Stop event would come equipped with the result
320 -- of the fold. It is not used here but it would be useful in
321 -- applicative and distribute.
322 done <- fromConsumer svr
324 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
325 $ takeMVar (outputDoorBellFromConsumer svr)
329 liftIO $ sendStop svr Nothing
330 -- drain/wait until a stop event arrives from the fold.
333 stop = stopFold >> stp
335 done <- pushToFold svr a
336 yld a (go done (K.nilM stopFold))
337 yieldk a r = pushToFold svr a >>= \done -> yld a (go done r)
338 in foldStreamShared st yieldk single stop m0
342-- In case of folds the roles of worker and parent on an SVar are reversed. The
343-- parent stream pushes values to an SVar instead of pulling from it and a
344-- worker thread running the fold pulls from the SVar and folds the stream. We
345-- keep a separate channel for pushing exceptions in the reverse direction i.e.
346-- from the fold to the parent stream.
348-- Note: If we terminate due to an exception, we do not actively terminate the
349-- fold. It gets cleaned up by the GC.
351{-# NOINLINE handleFoldException #-}
352handleFoldException :: SVar t m a -> SomeException -> IO ()
353handleFoldException sv e = do
355 void $ sendToProducer sv (ChildStop tid (Just e))
357{-# NOINLINE sendStopToConsumer #-}
358sendStopToConsumer :: MonadIO m => SVar t m a -> m ()
359sendStopToConsumer sv = liftIO $ do
361 void $ sendToProducer sv (ChildStop tid Nothing)
363-- | Create an SVar with a fold consumer that will fold any elements sent to it
364-- using the supplied fold function.
365{-# INLINE newFoldSVar #-}
366newFoldSVar :: (IsStream t, MonadAsync m)
367 => State Stream m a -> (t m a -> m b) -> m (SVar Stream m a)
368newFoldSVar stt f = do
369 -- Buffer size for the SVar is derived from the current state
370 sv <- newParallelVar StopAny (adaptState stt)
372 -- Add the producer thread-id to the SVar.
373 liftIO myThreadId >>= modifyThread sv
375 -- A wrapper to send a Stop event back to the producer when the fold
376 -- receives a stop from the producer.
377 let pull m = mkStream $ \st yld _ stp -> do
378 let stop = sendStopToConsumer sv >> stp
379 single a = yld a (K.nilM (sendStopToConsumer sv))
380 yieldk a r = yld a (pull r)
381 in foldStreamShared st yieldk single stop m
383 void $ doFork (void $ f $ fromStream $ pull $ fromProducer sv)
385 (handleFoldException sv)
388-- NOTE: In regular pull style streams, the consumer stream is pulling elements
389-- from the SVar and we have several workers producing elements and pushing to
390-- SVar. In case of folds, we, the parent stream driving the fold, are the
391-- stream producing worker, we start an SVar and start pushing to the SVar, the
392-- fold on the other side of the SVar is the consumer stream.
394-- In the pull stream case exceptions are propagated from the producing workers
395-- to the consumer stream, the exceptions are propagated on the same channel as
396-- the produced stream elements. However, in case of push style folds the
397-- current stream itself is the worker and the fold is the consumer, in this
398-- case we have to propagate the exceptions from the consumer to the producer.
399-- This is reverse of the pull case and we need a reverse direction channel
400-- to propagate the exception.
402-- | Redirect a copy of the stream to a supplied fold and run it concurrently
403-- in an independent thread. The fold may buffer some elements. The buffer size
404-- is determined by the prevailing 'maxBuffer' setting.
409-- -----stream m a ---------------stream m a-----
414-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
419-- Exceptions from the concurrently running fold are propagated to the current
420-- computation. Note that, because of buffering in the fold, exceptions may be
421-- delayed and may not correspond to the current element being processed in the
422-- parent stream, but we guarantee that before the parent stream stops the tap
423-- finishes and all exceptions from it are drained.
426-- Compare with 'tap'.
429{-# INLINE tapAsync #-}
430tapAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
431tapAsync f m = mkStream $ \st yld sng stp -> do
432 sv <- newFoldSVar st f
433 foldStreamShared st yld sng stp (teeToSVar sv m)
435------------------------------------------------------------------------------
437------------------------------------------------------------------------------
439-- | Async composition with strict concurrent execution of all streams.
441-- The 'Semigroup' instance of 'ParallelT' executes both the streams
442-- concurrently without any delay or without waiting for the consumer demand
443-- and /merges/ the results as they arrive. If the consumer does not consume
444-- the results, they are buffered upto a configured maximum, controlled by the
445-- 'maxBuffer' primitive. If the buffer becomes full the concurrent tasks will
446-- block until there is space in the buffer.
448-- Both 'WAsyncT' and 'ParallelT', evaluate the constituent streams fairly in a
449-- round robin fashion. The key difference is that 'WAsyncT' might wait for the
450-- consumer demand before it executes the tasks whereas 'ParallelT' starts
451-- executing all the tasks immediately without waiting for the consumer demand.
452-- For 'WAsyncT' the 'maxThreads' limit applies whereas for 'ParallelT' it does
453-- not apply. In other words, 'WAsyncT' can be lazy whereas 'ParallelT' is
456-- 'ParallelT' is useful for cases when the streams are required to be
457-- evaluated simultaneously irrespective of how the consumer consumes them e.g.
458-- when we want to race two tasks and want to start both strictly at the same
459-- time or if we have timers in the parallel tasks and our results depend on
460-- the timers being started at the same time. If we do not have such
461-- requirements then 'AsyncT' or 'AheadT' are recommended as they can be more
462-- efficient than 'ParallelT'.
465-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
471-- When streams with more than one element are merged, it yields whichever
472-- stream yields first without any bias, unlike the 'Async' style streams.
474-- Any exceptions generated by a constituent stream are propagated to the
475-- output stream. The output and exceptions from a single stream are guaranteed
476-- to arrive in the same order in the resulting stream as they were generated
477-- in the input stream. However, the relative ordering of elements from
478-- different streams in the resulting stream can vary depending on scheduling
479-- and generation delays.
481-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
482-- of the loop concurrently.
486-- import qualified "Streamly.Prelude" as S
487-- import Control.Concurrent
489-- main = 'drain' . 'parallely' $ do
490-- n <- return 3 \<\> return 2 \<\> return 1
492-- threadDelay (n * 1000000)
493-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
496-- ThreadId 40: Delay 1
497-- ThreadId 39: Delay 2
498-- ThreadId 38: Delay 3
501-- Note that parallel composition can only combine a finite number of
502-- streams as it needs to retain state for each unfinished stream.
504-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
507newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
508 deriving (MonadTrans)
510-- | A parallely composing IO stream of elements of type @a@.
511-- See 'ParallelT' documentation for more details.
514type Parallel = ParallelT IO
516-- | Fix the type of a polymorphic stream as 'ParallelT'.
519parallely :: IsStream t => ParallelT m a -> t m a
522instance IsStream ParallelT where
523 toStream = getParallelT
524 fromStream = ParallelT
527 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
528 consM = consMParallel
531 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
534------------------------------------------------------------------------------
536------------------------------------------------------------------------------
538{-# INLINE mappendParallel #-}
539{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
540mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
541mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
543instance MonadAsync m => Semigroup (ParallelT m a) where
544 (<>) = mappendParallel
546------------------------------------------------------------------------------
548------------------------------------------------------------------------------
550instance MonadAsync m => Monoid (ParallelT m a) where
554------------------------------------------------------------------------------
556------------------------------------------------------------------------------
558{-# INLINE bindParallel #-}
559{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
560bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
561bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
563instance MonadAsync m => Monad (ParallelT m) where
567------------------------------------------------------------------------------
569------------------------------------------------------------------------------
571MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
572MONAD_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