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
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)
65#if __GLASGOW_HASKELL__ < 808
66import Data.Semigroup (Semigroup(..))
69import Streamly.Data.Fold (Fold)
70import Streamly.Internal.Control.Concurrent (MonadAsync)
71import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))
72import Streamly.Internal.Data.Stream.Type (Stream)
74import qualified Data.Set as Set
75import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
76 (Stream, foldStreamShared, mkStream, foldStream, fromEffect
77 , nil, concatMapWith, fromPure, bindWith, withLocal)
78import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
79 (Stream(..), mapM, toStreamK, fromStreamK)
80import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
81import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
82import qualified Streamly.Internal.Data.Stream as Stream
83 (catMaybes, dropWhile, fromStreamK, repeat, sequence, takeWhile, toStreamK)
85import Streamly.Internal.Data.SVar
86import Prelude hiding (map)
89#include "Instances.hs"
93-- >>> import qualified Streamly.Data.Stream as Stream
94-- >>> import Control.Concurrent (threadDelay)
97-- threadDelay (n * 1000000) -- sleep for n seconds
98-- putStrLn (show n ++ " sec") -- print "n sec"
102-------------------------------------------------------------------------------
104-------------------------------------------------------------------------------
106-------------------------------------------------------------------------------
107-- StreamK based worker routines
108-------------------------------------------------------------------------------
110{-# NOINLINE runOne #-}
113 => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()
115 case getYieldLimit st of
117 Just _ -> runOneLimited st m0 winfo
122 liftIO $ decrementBufferLimit sv
123 K.foldStreamShared st yieldk single stop m
125 sv = fromJust $ streamVar st
128 incrementBufferLimit sv
130 sendit a = liftIO $ void $ send sv (ChildYield a)
131 single a = sendit a >> liftIO (sendStop sv winfo)
132 yieldk a r = sendit a >> go r
136 => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()
137runOneLimited st m0 winfo = go m0
142 yieldLimitOk <- liftIO $ decrementYieldLimit sv
145 liftIO $ decrementBufferLimit sv
146 K.foldStreamShared st yieldk single stop m
148 liftIO $ cleanupSVarFromWorker sv
149 liftIO $ sendStop sv winfo
151 sv = fromJust $ streamVar st
154 incrementBufferLimit sv
155 incrementYieldLimit sv
157 sendit a = liftIO $ void $ send sv (ChildYield a)
158 single a = sendit a >> liftIO (sendStop sv winfo)
159 yieldk a r = sendit a >> go r
161-------------------------------------------------------------------------------
162-- Consing and appending a stream in parallel style
163-------------------------------------------------------------------------------
165-- Note that consing and appending requires StreamK as it would not scale well
166-- with StreamD unless we are only consing a very small number of streams or
167-- elements in a stream. StreamK allows us to manipulate control flow in a way
168-- which StreamD cannot allow. StreamK can make a jump without having to
169-- remember the past state.
171{-# NOINLINE forkSVarPar #-}
172forkSVarPar :: MonadAsync m
173 => SVarStopStyle -> K.Stream m a -> K.Stream m a -> K.Stream m a
174forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do
175 sv <- newParallelVar ss st
176 pushWorkerPar sv (runOne st{streamVar = Just sv} m)
178 StopBy -> liftIO $ do
179 set <- readIORef (workerThreads sv)
180 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
182 pushWorkerPar sv (runOne st{streamVar = Just sv} r)
183 K.foldStream st yld sng stp $ Stream.toStreamK (SVar.fromSVar sv)
185{-# INLINE joinStreamVarPar #-}
193joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->
195 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
196 -- Here, WE ARE IN THE WORKER/PRODUCER THREAD, we know that because
197 -- the SVar exists. We are running under runOne and the output we
198 -- produce ultimately will be sent to the SVar by runOne.
200 -- If we came here the worker/runOne is evaluating a `parallel`
201 -- combinator. In this case, we always fork a new worker for the
202 -- first component (m1) in the parallel composition and continue to
203 -- evaluate the second component (m2) in the current worker thread.
205 -- When m1 is serially composed, the worker would evaluate it
206 -- without any further forks and the resulting output is sent to
207 -- the SVar and the evaluation terminates. If m1 is a `parallel`
208 -- composition of two streams the worker would again recurses here.
210 -- Similarly, when m2 is serially composed it gets evaluated here
211 -- and the resulting output is sent to the SVar by the runOne
212 -- wrapper. When m2 is composed with `parallel` it will again
213 -- recurse here and so on until it finally terminates.
215 -- When we create a right associated expression using `parallel`,
216 -- then m1 would always terminate without further forks or
217 -- recursion into this routine, therefore, the worker returns
218 -- immediately after evaluating it. And m2 would continue to
219 -- fork/recurse, therefore, the current thread always recurses and
220 -- forks new workers one after the other. This is a tail recursive
221 -- style execution, m2, the recursive expression always executed at
224 -- When the expression is left associated, the worker spawned would
225 -- get the forking/recursing responsibility and then again the
226 -- worker spawned by that worker would fork, thus creating layer
227 -- over layer of workers and a chain of threads leading to a very
228 -- inefficient execution.
229 pushWorkerPar sv (runOne st m1)
230 K.foldStreamShared st yld sng stp m2
232 -- Here WE ARE IN THE CONSUMER THREAD, we create a new SVar, fork
233 -- worker threads to execute m1 and m2 and this thread starts
234 -- pulling the stream from the SVar.
235 K.foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
237-------------------------------------------------------------------------------
239-------------------------------------------------------------------------------
241{-# INLINE parallelK #-}
242parallelK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
243parallelK = joinStreamVarPar ParallelVar StopNone
245-- | XXX we can implement it more efficienty by directly implementing instead
246-- of combining streams using parallel.
248{-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
249consM :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
250consM m (ParallelT r) = ParallelT $ parallelK (K.fromEffect m) r
252-- This is a co-parallel like combinator for streams, where first stream is the
253-- main stream and the rest are just supporting it, when the first ends
256-- | Like `parallel` but stops the output as soon as the first stream stops.
259{-# INLINE parallelFstK #-}
260parallelFstK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
261parallelFstK = joinStreamVarPar ParallelVar StopBy
263-- This is a race like combinator for streams.
265-- | Like `parallel` but stops the output as soon as any of the two streams
269{-# INLINE parallelMinK #-}
270parallelMinK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
271parallelMinK = joinStreamVarPar ParallelVar StopAny
273------------------------------------------------------------------------------
274-- Convert a stream to parallel
275------------------------------------------------------------------------------
277-- | Like 'mkParallel' but uses StreamK internally.
281mkParallelK :: MonadAsync m => K.Stream m a -> K.Stream m a
282mkParallelK m = K.mkStream $ \st yld sng stp -> do
283 sv <- newParallelVar StopNone (adaptState st)
284 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
285 SVar.toSVarParallel st sv $ D.fromStreamK m
286 K.foldStream st yld sng stp $ Stream.toStreamK $ SVar.fromSVar sv
288-- | Same as 'mkParallel' but for StreamD stream.
290{-# INLINE_NORMAL mkParallelD #-}
291mkParallelD :: MonadAsync m => D.Stream m a -> D.Stream m a
292mkParallelD m = D.Stream step Nothing
295 step gst Nothing = do
296 sv <- newParallelVar StopNone gst
297 SVar.toSVarParallel gst sv m
298 -- XXX use unfold instead?
299 return $ Skip $ Just $ SVar.fromSVarD sv
301 step gst (Just (D.UnStream step1 st)) = do
304 Yield a s -> Yield a (Just $ D.Stream step1 s)
305 Skip s -> Skip (Just $ D.Stream step1 s)
308-------------------------------------------------------------------------------
310-------------------------------------------------------------------------------
312-- NOTE: In regular pull style streams, the consumer stream is pulling elements
313-- from the SVar and we have several workers producing elements and pushing to
314-- SVar. In case of folds, we, the parent stream driving the fold, are the
315-- stream producing worker, we start an SVar and start pushing to the SVar, the
316-- fold on the other side of the SVar is the consumer stream.
318-- In the pull stream case exceptions are propagated from the producing workers
319-- to the consumer stream, the exceptions are propagated on the same channel as
320-- the produced stream elements. However, in case of push style folds the
321-- current stream itself is the worker and the fold is the consumer, in this
322-- case we have to propagate the exceptions from the consumer to the producer.
323-- This is reverse of the pull case and we need a reverse direction channel
324-- to propagate the exception.
326-- | Redirect a copy of the stream to a supplied fold and run it concurrently
327-- in an independent thread. The fold may buffer some elements. The buffer size
328-- is determined by the prevailing 'Streamly.Prelude.maxBuffer' setting.
333-- -----stream m a ---------------stream m a-----
338-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
343-- Exceptions from the concurrently running fold are propagated to the current
344-- computation. Note that, because of buffering in the fold, exceptions may be
345-- delayed and may not correspond to the current element being processed in the
346-- parent stream, but we guarantee that before the parent stream stops the tap
347-- finishes and all exceptions from it are drained.
350-- Compare with 'tap'.
353{-# INLINE tapAsyncK #-}
355 MonadAsync m => (K.Stream m a -> m b) -> K.Stream m a -> K.Stream m a
356tapAsyncK f m = K.mkStream $ \st yld sng stp -> do
357 sv <- SVar.newFoldSVar st (f . Stream.toStreamK)
358 K.foldStreamShared st yld sng stp
359 $ Stream.toStreamK (SVar.teeToSVar sv $ Stream.fromStreamK m)
361data TapState fs st a = TapInit | Tapping !fs st | TapDone st
363-- | Like 'tapAsync' but uses a 'Fold' instead of a fold function.
365{-# INLINE_NORMAL tapAsyncF #-}
366tapAsyncF :: MonadAsync m => Fold m a b -> D.Stream m a -> D.Stream m a
367tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit
371 -- In general, a Stop event would come equipped with the result
372 -- of the fold. It is not used here but it would be useful in
373 -- applicative and distribute.
374 done <- SVar.fromConsumer svr
376 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
377 $ takeMVar (outputDoorBellFromConsumer svr)
381 liftIO $ sendStop svr Nothing
382 -- drain/wait until a stop event arrives from the fold.
385 {-# INLINE_LATE step #-}
386 step gst TapInit = do
387 sv <- SVar.newFoldSVarF gst f
388 return $ Skip (Tapping sv state1)
390 step gst (Tapping sv st) = do
394 done <- SVar.pushToFold sv a
397 -- XXX we do not need to wait synchronously here
399 return $ Yield a (TapDone s)
400 else return $ Yield a (Tapping sv s)
401 Skip s -> return $ Skip (Tapping sv s)
406 step gst (TapDone st) = do
409 Yield a s -> Yield a (TapDone s)
410 Skip s -> Skip (TapDone s)
413------------------------------------------------------------------------------
415------------------------------------------------------------------------------
417-- | For 'ParallelT' streams:
420-- (<>) = 'Streamly.Prelude.parallel'
421-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.parallel'
424-- See 'Streamly.Prelude.AsyncT', 'ParallelT' is similar except that all
425-- iterations are strictly concurrent while in 'AsyncT' it depends on the
426-- consumer demand and available threads. See 'parallel' for more details.
428-- /Since: 0.1.0 ("Streamly")/
430-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
433newtype ParallelT m a = ParallelT {getParallelT :: K.Stream m a}
434 deriving (MonadTrans)
436-- | A parallely composing IO stream of elements of type @a@.
437-- See 'ParallelT' documentation for more details.
439-- /Since: 0.2.0 ("Streamly")/
442type Parallel = ParallelT IO
444------------------------------------------------------------------------------
446------------------------------------------------------------------------------
449{-# SPECIALIZE append :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
450append :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
451append (ParallelT m1) (ParallelT m2) = ParallelT $ parallelK m1 m2
453instance MonadAsync m => Semigroup (ParallelT m a) where
456------------------------------------------------------------------------------
458------------------------------------------------------------------------------
460instance MonadAsync m => Monoid (ParallelT m a) where
461 mempty = ParallelT K.nil
464------------------------------------------------------------------------------
466------------------------------------------------------------------------------
468{-# INLINE apParallel #-}
469{-# SPECIALIZE apParallel ::
470 ParallelT IO (a -> b) -> ParallelT IO a -> ParallelT IO b #-}
471apParallel :: MonadAsync m =>
472 ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b
473apParallel (ParallelT m1) (ParallelT m2) =
474 let f x1 = K.concatMapWith parallelK (pure . x1) m2
475 in ParallelT $ K.concatMapWith parallelK f m1
477instance (Monad m, MonadAsync m) => Applicative (ParallelT m) where
479 pure = ParallelT . K.fromPure
484------------------------------------------------------------------------------
486------------------------------------------------------------------------------
489{-# SPECIALIZE bind ::
490 ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
491bind :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
492bind (ParallelT m) f = ParallelT $ K.bindWith parallelK m (getParallelT . f)
494instance MonadAsync m => Monad (ParallelT m) where
500------------------------------------------------------------------------------
502------------------------------------------------------------------------------
504MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
506-------------------------------------------------------------------------------
508-------------------------------------------------------------------------------
510-- Note: we can use another API with two callbacks stop and yield if we want
511-- the callback to be able to indicate end of stream.
513-- | Generates a callback and a stream pair. The callback returned is used to
514-- queue values to the stream. The stream is infinite, there is no way for the
515-- callback to indicate that it is done now.
519{-# INLINE_NORMAL newCallbackStream #-}
520newCallbackStream :: MonadAsync m => m (a -> m (), K.Stream m a)
521newCallbackStream = do
522 sv <- newParallelVar StopNone defState
524 -- XXX Add our own thread-id to the SVar as we can not know the callback's
525 -- thread-id and the callback is not run in a managed worker. We need to
526 -- handle this better.
527 liftIO myThreadId >>= modifyThread sv
529 let callback a = liftIO $ void $ send sv (ChildYield a)
530 -- XXX we can return an SVar and then the consumer can unfold from the
532 return (callback, D.toStreamK (SVar.fromSVarD sv))
534------------------------------------------------------------------------------
536------------------------------------------------------------------------------
538{-# INLINE parallelFst #-}
539parallelFst :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
542 $ parallelFstK (Stream.toStreamK m1) (Stream.toStreamK m2)
544-- | Intersperse a monadic action into the input stream after every @n@
547-- >>> import qualified Streamly.Data.Fold as Fold
548-- >>> import qualified Streamly.Internal.Data.Stream.Parallel as Parallel
549-- >>> Stream.fold Fold.drain $ Parallel.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
553{-# INLINE interjectSuffix #-}
554interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a
555interjectSuffix n f xs = xs `parallelFst` repeatM timed
556 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
557 repeatM = Stream.sequence . Stream.repeat
559-- XXX Notes from D.takeByTime (which was removed)
560-- XXX using getTime in the loop can be pretty expensive especially for
561-- computations where iterations are lightweight. We have the following
564-- 1) Run a timeout thread updating a flag asynchronously and check that
565-- flag here, that way we can have a cheap termination check.
567-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
569-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
570-- termination value of rdtsc in the beginning and then in each iteration just
571-- get rdtsc and check if we should terminate.
574-- | @takeInterval duration@ yields stream elements upto specified time
575-- @duration@ in seconds. The duration starts when the stream is evaluated for
576-- the first time, before the first element is yielded. The time duration is
577-- checked before generating each element, if the duration has expired the
580-- The total time taken in executing the stream is guaranteed to be /at least/
581-- @duration@, however, because the duration is checked before generating an
582-- element, the upper bound is indeterminate and depends on the time taken in
583-- generating and processing the last element.
585-- No element is yielded if the duration is zero. At least one element is
586-- yielded if the duration is non-zero.
590{-# INLINE takeInterval #-}
591takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
594 . Stream.takeWhile isNothing
595 . interjectSuffix d (return Nothing) . fmap Just
597-- | @dropInterval duration@ drops stream elements until specified @duration@ in
598-- seconds has passed. The duration begins when the stream is evaluated for the
599-- first time. The time duration is checked /after/ generating a stream element,
600-- the element is yielded if the duration has expired otherwise it is dropped.
602-- The time elapsed before starting to generate the first element is /at most/
603-- @duration@, however, because the duration expiry is checked after the
604-- element is generated, the lower bound is indeterminate and depends on the
605-- time taken in generating an element.
607-- All elements are yielded if the duration is zero.
611{-# INLINE dropInterval #-}
612dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
615 . Stream.dropWhile isNothing
616 . 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