1{-# LANGUAGE UndecidableInstances #-}
4-- Module : Streamly.Internal.Data.Stream.Parallel
5-- Copyright : (c) 2017 Composewell Technologies
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12-- To run examples in this module:
14-- >>> import qualified Streamly.Data.Stream as Stream
15-- >>> import Control.Concurrent (threadDelay)
18-- threadDelay (n * 1000000) -- sleep for n seconds
19-- putStrLn (show n ++ " sec") -- print "n sec"
23module Streamly.Internal.Data.Stream.Parallel {-# DEPRECATED "Please use \"Streamly.Internal.Data.Stream.Concurrent\" instead." #-}
25 -- * Parallel Stream Type
30 -- * Merge Concurrently
35 -- * Evaluate Concurrently
53import Control.Concurrent (myThreadId, takeMVar, threadDelay)
54import Control.Monad (when)
55import Control.Monad.Base (MonadBase(..), liftBaseDefault)
56import Control.Monad.Catch (MonadThrow, throwM)
57-- import Control.Monad.Error.Class (MonadError(..))
58import Control.Monad.IO.Class (MonadIO(..))
59import Control.Monad.Reader.Class (MonadReader(..))
60import Control.Monad.State.Class (MonadState(..))
61import Control.Monad.Trans.Class (MonadTrans(lift))
62import Data.Functor (void)
63import Data.IORef (readIORef, writeIORef)
64import Data.Maybe (fromJust, isNothing)
66import Streamly.Data.Fold (Fold)
67import Streamly.Internal.Control.Concurrent (MonadAsync)
68import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))
69import Streamly.Internal.Data.Stream.Type (Stream)
71import qualified Data.Set as Set
72import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
73 (Stream, foldStreamShared, mkStream, foldStream, fromEffect
74 , nil, concatMapWith, fromPure, bindWith)
75import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
76 (Stream(..), mapM, toStreamK, fromStreamK)
77import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
78import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
79import qualified Streamly.Internal.Data.Stream as Stream
80 (catMaybes, dropWhile, fromStreamK, repeat, sequence, takeWhile, toStreamK)
82import Streamly.Internal.Data.SVar
83import Prelude hiding (map)
86#include "Instances.hs"
90-- >>> import qualified Streamly.Data.Stream as Stream
91-- >>> import Control.Concurrent (threadDelay)
94-- threadDelay (n * 1000000) -- sleep for n seconds
95-- putStrLn (show n ++ " sec") -- print "n sec"
99{-# INLINABLE withLocal #-}
100withLocal :: MonadReader r m => (r -> r) -> K.Stream m a -> K.Stream m a
102 K.mkStream $ \st yld sng stp ->
103 let single = local f . sng
104 yieldk a r = local f $ yld a (withLocal f r)
105 in K.foldStream st yieldk single (local f stp) m
107-------------------------------------------------------------------------------
109-------------------------------------------------------------------------------
111-------------------------------------------------------------------------------
112-- StreamK based worker routines
113-------------------------------------------------------------------------------
115{-# NOINLINE runOne #-}
118 => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()
120 case getYieldLimit st of
122 Just _ -> runOneLimited st m0 winfo
127 liftIO $ decrementBufferLimit sv
128 K.foldStreamShared st yieldk single stop m
130 sv = fromJust $ streamVar st
133 incrementBufferLimit sv
135 sendit a = liftIO $ void $ send sv (ChildYield a)
136 single a = sendit a >> liftIO (sendStop sv winfo)
137 yieldk a r = sendit a >> go r
141 => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()
142runOneLimited st m0 winfo = go m0
147 yieldLimitOk <- liftIO $ decrementYieldLimit sv
150 liftIO $ decrementBufferLimit sv
151 K.foldStreamShared st yieldk single stop m
153 liftIO $ cleanupSVarFromWorker sv
154 liftIO $ sendStop sv winfo
156 sv = fromJust $ streamVar st
159 incrementBufferLimit sv
160 incrementYieldLimit sv
162 sendit a = liftIO $ void $ send sv (ChildYield a)
163 single a = sendit a >> liftIO (sendStop sv winfo)
164 yieldk a r = sendit a >> go r
166-------------------------------------------------------------------------------
167-- Consing and appending a stream in parallel style
168-------------------------------------------------------------------------------
170-- Note that consing and appending requires StreamK as it would not scale well
171-- with StreamD unless we are only consing a very small number of streams or
172-- elements in a stream. StreamK allows us to manipulate control flow in a way
173-- which StreamD cannot allow. StreamK can make a jump without having to
174-- remember the past state.
176{-# NOINLINE forkSVarPar #-}
177forkSVarPar :: MonadAsync m
178 => SVarStopStyle -> K.Stream m a -> K.Stream m a -> K.Stream m a
179forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do
180 sv <- newParallelVar ss st
181 pushWorkerPar sv (runOne st{streamVar = Just sv} m)
183 StopBy -> liftIO $ do
184 set <- readIORef (workerThreads sv)
185 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
187 pushWorkerPar sv (runOne st{streamVar = Just sv} r)
188 K.foldStream st yld sng stp $ Stream.toStreamK (SVar.fromSVar sv)
190{-# INLINE joinStreamVarPar #-}
198joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->
200 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
201 -- Here, WE ARE IN THE WORKER/PRODUCER THREAD, we know that because
202 -- the SVar exists. We are running under runOne and the output we
203 -- produce ultimately will be sent to the SVar by runOne.
205 -- If we came here the worker/runOne is evaluating a `parallel`
206 -- combinator. In this case, we always fork a new worker for the
207 -- first component (m1) in the parallel composition and continue to
208 -- evaluate the second component (m2) in the current worker thread.
210 -- When m1 is serially composed, the worker would evaluate it
211 -- without any further forks and the resulting output is sent to
212 -- the SVar and the evaluation terminates. If m1 is a `parallel`
213 -- composition of two streams the worker would again recurses here.
215 -- Similarly, when m2 is serially composed it gets evaluated here
216 -- and the resulting output is sent to the SVar by the runOne
217 -- wrapper. When m2 is composed with `parallel` it will again
218 -- recurse here and so on until it finally terminates.
220 -- When we create a right associated expression using `parallel`,
221 -- then m1 would always terminate without further forks or
222 -- recursion into this routine, therefore, the worker returns
223 -- immediately after evaluating it. And m2 would continue to
224 -- fork/recurse, therefore, the current thread always recurses and
225 -- forks new workers one after the other. This is a tail recursive
226 -- style execution, m2, the recursive expression always executed at
229 -- When the expression is left associated, the worker spawned would
230 -- get the forking/recursing responsibility and then again the
231 -- worker spawned by that worker would fork, thus creating layer
232 -- over layer of workers and a chain of threads leading to a very
233 -- inefficient execution.
234 pushWorkerPar sv (runOne st m1)
235 K.foldStreamShared st yld sng stp m2
237 -- Here WE ARE IN THE CONSUMER THREAD, we create a new SVar, fork
238 -- worker threads to execute m1 and m2 and this thread starts
239 -- pulling the stream from the SVar.
240 K.foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
242-------------------------------------------------------------------------------
244-------------------------------------------------------------------------------
246{-# INLINE parallelK #-}
247parallelK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
248parallelK = joinStreamVarPar ParallelVar StopNone
250-- | XXX we can implement it more efficienty by directly implementing instead
251-- of combining streams using parallel.
253{-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
254consM :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
255consM m (ParallelT r) = ParallelT $ parallelK (K.fromEffect m) r
257-- This is a co-parallel like combinator for streams, where first stream is the
258-- main stream and the rest are just supporting it, when the first ends
261-- | Like `parallel` but stops the output as soon as the first stream stops.
264{-# INLINE parallelFstK #-}
265parallelFstK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
266parallelFstK = joinStreamVarPar ParallelVar StopBy
268-- This is a race like combinator for streams.
270-- | Like `parallel` but stops the output as soon as any of the two streams
274{-# INLINE parallelMinK #-}
275parallelMinK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
276parallelMinK = joinStreamVarPar ParallelVar StopAny
278------------------------------------------------------------------------------
279-- Convert a stream to parallel
280------------------------------------------------------------------------------
282-- | Like 'mkParallel' but uses StreamK internally.
286mkParallelK :: MonadAsync m => K.Stream m a -> K.Stream m a
287mkParallelK m = K.mkStream $ \st yld sng stp -> do
288 sv <- newParallelVar StopNone (adaptState st)
289 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
290 SVar.toSVarParallel st sv $ D.fromStreamK m
291 K.foldStream st yld sng stp $ Stream.toStreamK $ SVar.fromSVar sv
293-- | Same as 'mkParallel' but for StreamD stream.
295{-# INLINE_NORMAL mkParallelD #-}
296mkParallelD :: MonadAsync m => D.Stream m a -> D.Stream m a
297mkParallelD m = D.Stream step Nothing
300 step gst Nothing = do
301 sv <- newParallelVar StopNone gst
302 SVar.toSVarParallel gst sv m
303 -- XXX use unfold instead?
304 return $ Skip $ Just $ SVar.fromSVarD sv
306 step gst (Just (D.UnStream step1 st)) = do
309 Yield a s -> Yield a (Just $ D.Stream step1 s)
310 Skip s -> Skip (Just $ D.Stream step1 s)
313-------------------------------------------------------------------------------
315-------------------------------------------------------------------------------
317-- NOTE: In regular pull style streams, the consumer stream is pulling elements
318-- from the SVar and we have several workers producing elements and pushing to
319-- SVar. In case of folds, we, the parent stream driving the fold, are the
320-- stream producing worker, we start an SVar and start pushing to the SVar, the
321-- fold on the other side of the SVar is the consumer stream.
323-- In the pull stream case exceptions are propagated from the producing workers
324-- to the consumer stream, the exceptions are propagated on the same channel as
325-- the produced stream elements. However, in case of push style folds the
326-- current stream itself is the worker and the fold is the consumer, in this
327-- case we have to propagate the exceptions from the consumer to the producer.
328-- This is reverse of the pull case and we need a reverse direction channel
329-- to propagate the exception.
331-- | Redirect a copy of the stream to a supplied fold and run it concurrently
332-- in an independent thread. The fold may buffer some elements. The buffer size
333-- is determined by the prevailing 'Streamly.Prelude.maxBuffer' setting.
338-- -----stream m a ---------------stream m a-----
343-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
348-- Exceptions from the concurrently running fold are propagated to the current
349-- computation. Note that, because of buffering in the fold, exceptions may be
350-- delayed and may not correspond to the current element being processed in the
351-- parent stream, but we guarantee that before the parent stream stops the tap
352-- finishes and all exceptions from it are drained.
355-- Compare with 'tap'.
358{-# INLINE tapAsyncK #-}
360 MonadAsync m => (K.Stream m a -> m b) -> K.Stream m a -> K.Stream m a
361tapAsyncK f m = K.mkStream $ \st yld sng stp -> do
362 sv <- SVar.newFoldSVar st (f . Stream.toStreamK)
363 K.foldStreamShared st yld sng stp
364 $ Stream.toStreamK (SVar.teeToSVar sv $ Stream.fromStreamK m)
366data TapState fs st a = TapInit | Tapping !fs st | TapDone st
368-- | Like 'tapAsync' but uses a 'Fold' instead of a fold function.
370{-# INLINE_NORMAL tapAsyncF #-}
371tapAsyncF :: MonadAsync m => Fold m a b -> D.Stream m a -> D.Stream m a
372tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit
376 -- In general, a Stop event would come equipped with the result
377 -- of the fold. It is not used here but it would be useful in
378 -- applicative and distribute.
379 done <- SVar.fromConsumer svr
381 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
382 $ takeMVar (outputDoorBellFromConsumer svr)
386 liftIO $ sendStop svr Nothing
387 -- drain/wait until a stop event arrives from the fold.
390 {-# INLINE_LATE step #-}
391 step gst TapInit = do
392 sv <- SVar.newFoldSVarF gst f
393 return $ Skip (Tapping sv state1)
395 step gst (Tapping sv st) = do
399 done <- SVar.pushToFold sv a
402 -- XXX we do not need to wait synchronously here
404 return $ Yield a (TapDone s)
405 else return $ Yield a (Tapping sv s)
406 Skip s -> return $ Skip (Tapping sv s)
411 step gst (TapDone st) = do
414 Yield a s -> Yield a (TapDone s)
415 Skip s -> Skip (TapDone s)
418------------------------------------------------------------------------------
420------------------------------------------------------------------------------
422-- | For 'ParallelT' streams:
425-- (<>) = 'Streamly.Prelude.parallel'
426-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.parallel'
429-- See 'Streamly.Prelude.AsyncT', 'ParallelT' is similar except that all
430-- iterations are strictly concurrent while in 'AsyncT' it depends on the
431-- consumer demand and available threads. See 'parallel' for more details.
433-- /Since: 0.1.0 ("Streamly")/
435-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
438newtype ParallelT m a = ParallelT {getParallelT :: K.Stream m a}
440instance MonadTrans ParallelT where
442 lift = ParallelT . K.fromEffect
444-- | A parallely composing IO stream of elements of type @a@.
445-- See 'ParallelT' documentation for more details.
447-- /Since: 0.2.0 ("Streamly")/
450type Parallel = ParallelT IO
452------------------------------------------------------------------------------
454------------------------------------------------------------------------------
457{-# SPECIALIZE append :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
458append :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
459append (ParallelT m1) (ParallelT m2) = ParallelT $ parallelK m1 m2
461instance MonadAsync m => Semigroup (ParallelT m a) where
464------------------------------------------------------------------------------
466------------------------------------------------------------------------------
468instance MonadAsync m => Monoid (ParallelT m a) where
469 mempty = ParallelT K.nil
472------------------------------------------------------------------------------
474------------------------------------------------------------------------------
476{-# INLINE apParallel #-}
477{-# SPECIALIZE apParallel ::
478 ParallelT IO (a -> b) -> ParallelT IO a -> ParallelT IO b #-}
479apParallel :: MonadAsync m =>
480 ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b
481apParallel (ParallelT m1) (ParallelT m2) =
482 let f x1 = K.concatMapWith parallelK (K.fromPure . x1) m2
483 in ParallelT $ K.concatMapWith parallelK f m1
485instance (Monad m, MonadAsync m) => Applicative (ParallelT m) where
487 pure = ParallelT . K.fromPure
492------------------------------------------------------------------------------
494------------------------------------------------------------------------------
497{-# SPECIALIZE bind ::
498 ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
499bind :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
500bind (ParallelT m) f = ParallelT $ K.bindWith parallelK m (getParallelT . f)
502instance MonadAsync m => Monad (ParallelT m) where
508------------------------------------------------------------------------------
510------------------------------------------------------------------------------
512MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
514-------------------------------------------------------------------------------
516-------------------------------------------------------------------------------
518-- Note: we can use another API with two callbacks stop and yield if we want
519-- the callback to be able to indicate end of stream.
521-- | Generates a callback and a stream pair. The callback returned is used to
522-- queue values to the stream. The stream is infinite, there is no way for the
523-- callback to indicate that it is done now.
527{-# INLINE_NORMAL newCallbackStream #-}
528newCallbackStream :: MonadAsync m => m (a -> m (), K.Stream m a)
529newCallbackStream = do
530 sv <- newParallelVar StopNone defState
532 -- XXX Add our own thread-id to the SVar as we can not know the callback's
533 -- thread-id and the callback is not run in a managed worker. We need to
534 -- handle this better.
535 liftIO myThreadId >>= modifyThread sv
537 let callback a = liftIO $ void $ send sv (ChildYield a)
538 -- XXX we can return an SVar and then the consumer can unfold from the
540 return (callback, D.toStreamK (SVar.fromSVarD sv))
542------------------------------------------------------------------------------
544------------------------------------------------------------------------------
546{-# INLINE parallelFst #-}
547parallelFst :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
550 $ parallelFstK (Stream.toStreamK m1) (Stream.toStreamK m2)
552-- | Intersperse a monadic action into the input stream after every @n@
555-- >>> import qualified Streamly.Data.Fold as Fold
556-- >>> import qualified Streamly.Internal.Data.Stream.Parallel as Parallel
557-- >>> Stream.fold Fold.drain $ Parallel.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
561{-# INLINE interjectSuffix #-}
562interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a
563interjectSuffix n f xs = xs `parallelFst` repeatM timed
564 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
565 repeatM = Stream.sequence . Stream.repeat
567-- XXX Notes from D.takeByTime (which was removed)
568-- XXX using getTime in the loop can be pretty expensive especially for
569-- computations where iterations are lightweight. We have the following
572-- 1) Run a timeout thread updating a flag asynchronously and check that
573-- flag here, that way we can have a cheap termination check.
575-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
577-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
578-- termination value of rdtsc in the beginning and then in each iteration just
579-- get rdtsc and check if we should terminate.
582-- | @takeInterval duration@ yields stream elements upto specified time
583-- @duration@ in seconds. The duration starts when the stream is evaluated for
584-- the first time, before the first element is yielded. The time duration is
585-- checked before generating each element, if the duration has expired the
588-- The total time taken in executing the stream is guaranteed to be /at least/
589-- @duration@, however, because the duration is checked before generating an
590-- element, the upper bound is indeterminate and depends on the time taken in
591-- generating and processing the last element.
593-- No element is yielded if the duration is zero. At least one element is
594-- yielded if the duration is non-zero.
598{-# INLINE takeInterval #-}
599takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
602 . Stream.takeWhile isNothing
603 . interjectSuffix d (return Nothing) . fmap Just
605-- | @dropInterval duration@ drops stream elements until specified @duration@ in
606-- seconds has passed. The duration begins when the stream is evaluated for the
607-- first time. The time duration is checked /after/ generating a stream element,
608-- the element is yielded if the duration has expired otherwise it is dropped.
610-- The time elapsed before starting to generate the first element is /at most/
611-- @duration@, however, because the duration expiry is checked after the
612-- element is generated, the lower bound is indeterminate and depends on the
613-- time taken in generating an element.
615-- All elements are yielded if the duration is zero.
619{-# INLINE dropInterval #-}
620dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
623 . Stream.dropWhile isNothing
624 . interjectSuffix d (return Nothing) . fmap Just
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