1{-# LANGUAGE UndecidableInstances #-}
2{-# OPTIONS_GHC -fno-warn-deprecations #-}
5-- Module : Streamly.Internal.Data.Stream.Parallel
6-- Copyright : (c) 2017 Composewell Technologies
9-- Maintainer : streamly@composewell.com
10-- Stability : experimental
13-- To run examples in this module:
15-- >>> import qualified Streamly.Prelude as Stream
16-- >>> import Control.Concurrent (threadDelay)
19-- threadDelay (n * 1000000) -- sleep for n seconds
20-- putStrLn (show n ++ " sec") -- print "n sec"
24module Streamly.Internal.Data.Stream.Parallel {-# DEPRECATED "Please use \"Streamly.Internal.Data.Stream.Concurrent\" instead." #-}
26 -- * Parallel Stream Type
31 -- * Merge Concurrently
36 -- * Evaluate Concurrently
49import Control.Concurrent (myThreadId, takeMVar)
50import Control.Monad (when)
51import Control.Monad.Base (MonadBase(..), liftBaseDefault)
52import Control.Monad.Catch (MonadThrow, throwM)
53-- import Control.Monad.Error.Class (MonadError(..))
54import Control.Monad.IO.Class (MonadIO(..))
55import Control.Monad.Reader.Class (MonadReader(..))
56import Control.Monad.State.Class (MonadState(..))
57import Control.Monad.Trans.Class (MonadTrans(lift))
58import Data.Functor (void)
59import Data.IORef (readIORef, writeIORef)
60import Data.Maybe (fromJust)
62import Streamly.Data.Fold (Fold)
63import Streamly.Internal.Control.Concurrent (MonadAsync)
64import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))
66import qualified Data.Set as Set
67import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
68 (StreamK, foldStreamShared, mkStream, foldStream, fromEffect
69 , nil, concatMapWith, fromPure, bindWith)
70import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
71 (Stream(..), mapM, toStreamK, fromStreamK)
72import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
73import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
74import qualified Streamly.Internal.Data.Stream.Serial as Stream
76import Streamly.Internal.Data.SVar
77import Prelude hiding (map)
80#include "Instances.hs"
84-- >>> import qualified Streamly.Prelude as Stream
85-- >>> import Control.Concurrent (threadDelay)
88-- threadDelay (n * 1000000) -- sleep for n seconds
89-- putStrLn (show n ++ " sec") -- print "n sec"
93{-# INLINABLE withLocal #-}
94withLocal :: MonadReader r m => (r -> r) -> K.StreamK m a -> K.StreamK m a
96 K.mkStream $ \st yld sng stp ->
97 let single = local f . sng
98 yieldk a r = local f $ yld a (withLocal f r)
99 in K.foldStream st yieldk single (local f stp) m
101-------------------------------------------------------------------------------
103-------------------------------------------------------------------------------
105-------------------------------------------------------------------------------
106-- StreamK based worker routines
107-------------------------------------------------------------------------------
109{-# NOINLINE runOne #-}
112 => State K.StreamK m a -> K.StreamK m a -> Maybe WorkerInfo -> m ()
114 case getYieldLimit st of
116 Just _ -> runOneLimited st m0 winfo
121 liftIO $ decrementBufferLimit sv
122 K.foldStreamShared st yieldk single stop m
124 sv = fromJust $ streamVar st
127 incrementBufferLimit sv
129 sendit a = liftIO $ void $ send sv (ChildYield a)
130 single a = sendit a >> liftIO (sendStop sv winfo)
131 yieldk a r = sendit a >> go r
135 => State K.StreamK m a -> K.StreamK m a -> Maybe WorkerInfo -> m ()
136runOneLimited st m0 winfo = go m0
141 yieldLimitOk <- liftIO $ decrementYieldLimit sv
144 liftIO $ decrementBufferLimit sv
145 K.foldStreamShared st yieldk single stop m
147 liftIO $ cleanupSVarFromWorker sv
148 liftIO $ sendStop sv winfo
150 sv = fromJust $ streamVar st
153 incrementBufferLimit sv
154 incrementYieldLimit sv
156 sendit a = liftIO $ void $ send sv (ChildYield a)
157 single a = sendit a >> liftIO (sendStop sv winfo)
158 yieldk a r = sendit a >> go r
160-------------------------------------------------------------------------------
161-- Consing and appending a stream in parallel style
162-------------------------------------------------------------------------------
164-- Note that consing and appending requires StreamK as it would not scale well
165-- with StreamD unless we are only consing a very small number of streams or
166-- elements in a stream. StreamK allows us to manipulate control flow in a way
167-- which StreamD cannot allow. StreamK can make a jump without having to
168-- remember the past state.
170{-# NOINLINE forkSVarPar #-}
171forkSVarPar :: MonadAsync m
172 => SVarStopStyle -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
173forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do
174 sv <- newParallelVar ss st
175 pushWorkerPar sv (runOne st{streamVar = Just sv} m)
177 StopBy -> liftIO $ do
178 set <- readIORef (workerThreads sv)
179 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
181 pushWorkerPar sv (runOne st{streamVar = Just sv} r)
182 K.foldStream st yld sng stp $ Stream.toStreamK (SVar.fromSVar sv)
184{-# INLINE joinStreamVarPar #-}
192joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->
194 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
195 -- Here, WE ARE IN THE WORKER/PRODUCER THREAD, we know that because
196 -- the SVar exists. We are running under runOne and the output we
197 -- produce ultimately will be sent to the SVar by runOne.
199 -- If we came here the worker/runOne is evaluating a `parallel`
200 -- combinator. In this case, we always fork a new worker for the
201 -- first component (m1) in the parallel composition and continue to
202 -- evaluate the second component (m2) in the current worker thread.
204 -- When m1 is serially composed, the worker would evaluate it
205 -- without any further forks and the resulting output is sent to
206 -- the SVar and the evaluation terminates. If m1 is a `parallel`
207 -- composition of two streams the worker would again recurses here.
209 -- Similarly, when m2 is serially composed it gets evaluated here
210 -- and the resulting output is sent to the SVar by the runOne
211 -- wrapper. When m2 is composed with `parallel` it will again
212 -- recurse here and so on until it finally terminates.
214 -- When we create a right associated expression using `parallel`,
215 -- then m1 would always terminate without further forks or
216 -- recursion into this routine, therefore, the worker returns
217 -- immediately after evaluating it. And m2 would continue to
218 -- fork/recurse, therefore, the current thread always recurses and
219 -- forks new workers one after the other. This is a tail recursive
220 -- style execution, m2, the recursive expression always executed at
223 -- When the expression is left associated, the worker spawned would
224 -- get the forking/recursing responsibility and then again the
225 -- worker spawned by that worker would fork, thus creating layer
226 -- over layer of workers and a chain of threads leading to a very
227 -- inefficient execution.
228 pushWorkerPar sv (runOne st m1)
229 K.foldStreamShared st yld sng stp m2
231 -- Here WE ARE IN THE CONSUMER THREAD, we create a new SVar, fork
232 -- worker threads to execute m1 and m2 and this thread starts
233 -- pulling the stream from the SVar.
234 K.foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
236-------------------------------------------------------------------------------
238-------------------------------------------------------------------------------
240{-# INLINE parallelK #-}
241parallelK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a
242parallelK = joinStreamVarPar ParallelVar StopNone
244-- | XXX we can implement it more efficienty by directly implementing instead
245-- of combining streams using parallel.
247{-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
248consM :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
249consM m (ParallelT r) = ParallelT $ parallelK (K.fromEffect m) r
251-- This is a co-parallel like combinator for streams, where first stream is the
252-- main stream and the rest are just supporting it, when the first ends
255-- | Like `parallel` but stops the output as soon as the first stream stops.
258{-# INLINE parallelFstK #-}
259parallelFstK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a
260parallelFstK = joinStreamVarPar ParallelVar StopBy
262-- This is a race like combinator for streams.
264-- | Like `parallel` but stops the output as soon as any of the two streams
268{-# INLINE parallelMinK #-}
269parallelMinK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a
270parallelMinK = joinStreamVarPar ParallelVar StopAny
272------------------------------------------------------------------------------
273-- Convert a stream to parallel
274------------------------------------------------------------------------------
276-- | Like 'mkParallel' but uses StreamK internally.
280mkParallelK :: MonadAsync m => K.StreamK m a -> K.StreamK m a
281mkParallelK m = K.mkStream $ \st yld sng stp -> do
282 sv <- newParallelVar StopNone (adaptState st)
283 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
284 SVar.toSVarParallel st sv $ D.fromStreamK m
285 K.foldStream st yld sng stp $ Stream.toStreamK $ SVar.fromSVar sv
287-- | Same as 'mkParallel' but for StreamD stream.
289{-# INLINE_NORMAL mkParallelD #-}
290mkParallelD :: MonadAsync m => D.Stream m a -> D.Stream m a
291mkParallelD m = D.Stream step Nothing
294 step gst Nothing = do
295 sv <- newParallelVar StopNone gst
296 SVar.toSVarParallel gst sv m
297 -- XXX use unfold instead?
298 return $ Skip $ Just $ SVar.fromSVarD sv
300 step gst (Just (D.UnStream step1 st)) = do
303 Yield a s -> Yield a (Just $ D.Stream step1 s)
304 Skip s -> Skip (Just $ D.Stream step1 s)
307-------------------------------------------------------------------------------
309-------------------------------------------------------------------------------
311-- NOTE: In regular pull style streams, the consumer stream is pulling elements
312-- from the SVar and we have several workers producing elements and pushing to
313-- SVar. In case of folds, we, the parent stream driving the fold, are the
314-- stream producing worker, we start an SVar and start pushing to the SVar, the
315-- fold on the other side of the SVar is the consumer stream.
317-- In the pull stream case exceptions are propagated from the producing workers
318-- to the consumer stream, the exceptions are propagated on the same channel as
319-- the produced stream elements. However, in case of push style folds the
320-- current stream itself is the worker and the fold is the consumer, in this
321-- case we have to propagate the exceptions from the consumer to the producer.
322-- This is reverse of the pull case and we need a reverse direction channel
323-- to propagate the exception.
325-- | Redirect a copy of the stream to a supplied fold and run it concurrently
326-- in an independent thread. The fold may buffer some elements. The buffer size
327-- is determined by the prevailing 'Streamly.Prelude.maxBuffer' setting.
332-- -----stream m a ---------------stream m a-----
337-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
342-- Exceptions from the concurrently running fold are propagated to the current
343-- computation. Note that, because of buffering in the fold, exceptions may be
344-- delayed and may not correspond to the current element being processed in the
345-- parent stream, but we guarantee that before the parent stream stops the tap
346-- finishes and all exceptions from it are drained.
349-- Compare with 'tap'.
352{-# INLINE tapAsyncK #-}
354 MonadAsync m => (K.StreamK m a -> m b) -> K.StreamK m a -> K.StreamK m a
355tapAsyncK f m = K.mkStream $ \st yld sng stp -> do
356 sv <- SVar.newFoldSVar st (f . Stream.toStreamK)
357 K.foldStreamShared st yld sng stp
358 $ Stream.toStreamK (SVar.teeToSVar sv $ Stream.fromStreamK m)
360data TapState fs st a = TapInit | Tapping !fs st | TapDone st
362-- | Like 'tapAsync' but uses a 'Fold' instead of a fold function.
364{-# INLINE_NORMAL tapAsyncF #-}
365tapAsyncF :: MonadAsync m => Fold m a b -> D.Stream m a -> D.Stream m a
366tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit
370 -- In general, a Stop event would come equipped with the result
371 -- of the fold. It is not used here but it would be useful in
372 -- applicative and distribute.
373 done <- SVar.fromConsumer svr
375 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
376 $ takeMVar (outputDoorBellFromConsumer svr)
380 liftIO $ sendStop svr Nothing
381 -- drain/wait until a stop event arrives from the fold.
384 {-# INLINE_LATE step #-}
385 step gst TapInit = do
386 sv <- SVar.newFoldSVarF gst f
387 return $ Skip (Tapping sv state1)
389 step gst (Tapping sv st) = do
393 done <- SVar.pushToFold sv a
396 -- XXX we do not need to wait synchronously here
398 return $ Yield a (TapDone s)
399 else return $ Yield a (Tapping sv s)
400 Skip s -> return $ Skip (Tapping sv s)
405 step gst (TapDone st) = do
408 Yield a s -> Yield a (TapDone s)
409 Skip s -> Skip (TapDone s)
412------------------------------------------------------------------------------
414------------------------------------------------------------------------------
416-- | For 'ParallelT' streams:
419-- (<>) = 'Streamly.Prelude.parallel'
420-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.parallel'
423-- See 'Streamly.Prelude.AsyncT', 'ParallelT' is similar except that all
424-- iterations are strictly concurrent while in 'AsyncT' it depends on the
425-- consumer demand and available threads. See 'parallel' for more details.
427-- /Since: 0.1.0 ("Streamly")/
429-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
432newtype ParallelT m a = ParallelT {getParallelT :: K.StreamK m a}
434instance MonadTrans ParallelT where
436 lift = ParallelT . K.fromEffect
438-- | A parallely composing IO stream of elements of type @a@.
439-- See 'ParallelT' documentation for more details.
441-- /Since: 0.2.0 ("Streamly")/
444type Parallel = ParallelT IO
446------------------------------------------------------------------------------
448------------------------------------------------------------------------------
451{-# SPECIALIZE append :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
452append :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
453append (ParallelT m1) (ParallelT m2) = ParallelT $ parallelK m1 m2
455instance MonadAsync m => Semigroup (ParallelT m a) where
458------------------------------------------------------------------------------
460------------------------------------------------------------------------------
462instance MonadAsync m => Monoid (ParallelT m a) where
463 mempty = ParallelT K.nil
466------------------------------------------------------------------------------
468------------------------------------------------------------------------------
470{-# INLINE apParallel #-}
471{-# SPECIALIZE apParallel ::
472 ParallelT IO (a -> b) -> ParallelT IO a -> ParallelT IO b #-}
473apParallel :: MonadAsync m =>
474 ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b
475apParallel (ParallelT m1) (ParallelT m2) =
476 let f x1 = K.concatMapWith parallelK (K.fromPure . x1) m2
477 in ParallelT $ K.concatMapWith parallelK f m1
479instance (Monad m, MonadAsync m) => Applicative (ParallelT m) where
481 pure = ParallelT . K.fromPure
486------------------------------------------------------------------------------
488------------------------------------------------------------------------------
491{-# SPECIALIZE bind ::
492 ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
493bind :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
494bind (ParallelT m) f = ParallelT $ K.bindWith parallelK m (getParallelT . f)
496instance MonadAsync m => Monad (ParallelT m) where
502------------------------------------------------------------------------------
504------------------------------------------------------------------------------
506MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
508-------------------------------------------------------------------------------
510-------------------------------------------------------------------------------
512-- Note: we can use another API with two callbacks stop and yield if we want
513-- the callback to be able to indicate end of stream.
515-- | Generates a callback and a stream pair. The callback returned is used to
516-- queue values to the stream. The stream is infinite, there is no way for the
517-- callback to indicate that it is done now.
521{-# INLINE_NORMAL newCallbackStream #-}
522newCallbackStream :: MonadAsync m => m (a -> m (), K.StreamK m a)
523newCallbackStream = do
524 sv <- newParallelVar StopNone defState
526 -- XXX Add our own thread-id to the SVar as we can not know the callback's
527 -- thread-id and the callback is not run in a managed worker. We need to
528 -- handle this better.
529 liftIO myThreadId >>= modifyThread sv
531 let callback a = liftIO $ void $ send sv (ChildYield a)
532 -- XXX we can return an SVar and then the consumer can unfold from the
534 return (callback, D.toStreamK (SVar.fromSVarD sv))
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