1{-# LANGUAGE UndecidableInstances #-}
2{-# OPTIONS_GHC -fno-warn-deprecations #-}
3{-# OPTIONS_GHC -Wno-redundant-constraints #-}
6-- Module : Streamly.Internal.Data.Stream.Parallel
7-- Copyright : (c) 2017 Composewell Technologies
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
14-- To run examples in this module:
16-- >>> import qualified Streamly.Prelude as Stream
17-- >>> import Control.Concurrent (threadDelay)
20-- threadDelay (n * 1000000) -- sleep for n seconds
21-- putStrLn (show n ++ " sec") -- print "n sec"
25module Streamly.Internal.Data.Stream.Parallel {-# DEPRECATED "Please use \"Streamly.Internal.Data.Stream.Concurrent\" instead." #-}
27 -- * Parallel Stream Type
32 -- * Merge Concurrently
37 -- * Evaluate Concurrently
50import Control.Concurrent (myThreadId, takeMVar)
51import Control.Monad (when)
52#if !(MIN_VERSION_transformers(0,6,0))
53import Control.Monad.Base (MonadBase(..), liftBaseDefault)
55import Control.Monad.Catch (MonadThrow, throwM)
56-- import Control.Monad.Error.Class (MonadError(..))
57import Control.Monad.IO.Class (MonadIO(..))
58import Control.Monad.Reader.Class (MonadReader(..))
59import Control.Monad.State.Class (MonadState(..))
60#if !(MIN_VERSION_transformers(0,6,0))
61import Control.Monad.Trans.Class (MonadTrans(lift))
63import Data.Functor (void)
64import Data.IORef (readIORef, writeIORef)
65import Data.Maybe (fromJust)
67import Streamly.Data.Fold (Fold)
68import Streamly.Internal.Control.Concurrent (MonadAsync)
69import Streamly.Internal.Data.Stream (Step(..))
71import qualified Data.Set as Set
72import qualified Streamly.Internal.Data.StreamK as K
73 (StreamK, foldStreamShared, mkStream, foldStream, fromEffect
74 , nil, concatMapWith, fromPure, bindWith)
75import qualified Streamly.Internal.Data.Stream 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.Serial as Stream
81import Streamly.Internal.Data.SVar
82import Prelude hiding (map)
85#include "Instances.hs"
89-- >>> import qualified Streamly.Prelude as Stream
90-- >>> import Control.Concurrent (threadDelay)
93-- threadDelay (n * 1000000) -- sleep for n seconds
94-- putStrLn (show n ++ " sec") -- print "n sec"
98{-# INLINABLE withLocal #-}
99withLocal :: MonadReader r m => (r -> r) -> K.StreamK m a -> K.StreamK m a
101 K.mkStream $ \st yld sng stp ->
102 let single = local f . sng
103 yieldk a r = local f $ yld a (withLocal f r)
104 in K.foldStream st yieldk single (local f stp) m
106-------------------------------------------------------------------------------
108-------------------------------------------------------------------------------
110-------------------------------------------------------------------------------
111-- StreamK based worker routines
112-------------------------------------------------------------------------------
114{-# NOINLINE runOne #-}
117 => State K.StreamK m a -> K.StreamK m a -> Maybe WorkerInfo -> m ()
119 case getYieldLimit st of
121 Just _ -> runOneLimited st m0 winfo
126 liftIO $ decrementBufferLimit sv
127 K.foldStreamShared st yieldk single stop m
129 sv = fromJust $ streamVar st
132 incrementBufferLimit sv
134 sendit a = liftIO $ void $ send sv (ChildYield a)
135 single a = sendit a >> liftIO (sendStop sv winfo)
136 yieldk a r = sendit a >> go r
140 => State K.StreamK m a -> K.StreamK m a -> Maybe WorkerInfo -> m ()
141runOneLimited st m0 winfo = go m0
146 yieldLimitOk <- liftIO $ decrementYieldLimit sv
149 liftIO $ decrementBufferLimit sv
150 K.foldStreamShared st yieldk single stop m
152 liftIO $ cleanupSVarFromWorker sv
153 liftIO $ sendStop sv winfo
155 sv = fromJust $ streamVar st
158 incrementBufferLimit sv
159 incrementYieldLimit sv
161 sendit a = liftIO $ void $ send sv (ChildYield a)
162 single a = sendit a >> liftIO (sendStop sv winfo)
163 yieldk a r = sendit a >> go r
165-------------------------------------------------------------------------------
166-- Consing and appending a stream in parallel style
167-------------------------------------------------------------------------------
169-- Note that consing and appending requires StreamK as it would not scale well
170-- with StreamD unless we are only consing a very small number of streams or
171-- elements in a stream. StreamK allows us to manipulate control flow in a way
172-- which StreamD cannot allow. StreamK can make a jump without having to
173-- remember the past state.
175{-# NOINLINE forkSVarPar #-}
176forkSVarPar :: MonadAsync m
177 => SVarStopStyle -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
178forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do
179 sv <- newParallelVar ss st
180 pushWorkerPar sv (runOne st{streamVar = Just sv} m)
182 StopBy -> liftIO $ do
183 set <- readIORef (workerThreads sv)
184 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
186 pushWorkerPar sv (runOne st{streamVar = Just sv} r)
187 K.foldStream st yld sng stp $ Stream.toStreamK (SVar.fromSVar sv)
189{-# INLINE joinStreamVarPar #-}
197joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->
199 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
200 -- Here, WE ARE IN THE WORKER/PRODUCER THREAD, we know that because
201 -- the SVar exists. We are running under runOne and the output we
202 -- produce ultimately will be sent to the SVar by runOne.
204 -- If we came here the worker/runOne is evaluating a `parallel`
205 -- combinator. In this case, we always fork a new worker for the
206 -- first component (m1) in the parallel composition and continue to
207 -- evaluate the second component (m2) in the current worker thread.
209 -- When m1 is serially composed, the worker would evaluate it
210 -- without any further forks and the resulting output is sent to
211 -- the SVar and the evaluation terminates. If m1 is a `parallel`
212 -- composition of two streams the worker would again recurses here.
214 -- Similarly, when m2 is serially composed it gets evaluated here
215 -- and the resulting output is sent to the SVar by the runOne
216 -- wrapper. When m2 is composed with `parallel` it will again
217 -- recurse here and so on until it finally terminates.
219 -- When we create a right associated expression using `parallel`,
220 -- then m1 would always terminate without further forks or
221 -- recursion into this routine, therefore, the worker returns
222 -- immediately after evaluating it. And m2 would continue to
223 -- fork/recurse, therefore, the current thread always recurses and
224 -- forks new workers one after the other. This is a tail recursive
225 -- style execution, m2, the recursive expression always executed at
228 -- When the expression is left associated, the worker spawned would
229 -- get the forking/recursing responsibility and then again the
230 -- worker spawned by that worker would fork, thus creating layer
231 -- over layer of workers and a chain of threads leading to a very
232 -- inefficient execution.
233 pushWorkerPar sv (runOne st m1)
234 K.foldStreamShared st yld sng stp m2
236 -- Here WE ARE IN THE CONSUMER THREAD, we create a new SVar, fork
237 -- worker threads to execute m1 and m2 and this thread starts
238 -- pulling the stream from the SVar.
239 K.foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
241-------------------------------------------------------------------------------
243-------------------------------------------------------------------------------
245{-# INLINE parallelK #-}
246parallelK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a
247parallelK = joinStreamVarPar ParallelVar StopNone
249-- | XXX we can implement it more efficienty by directly implementing instead
250-- of combining streams using parallel.
252{-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
253consM :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
254consM m (ParallelT r) = ParallelT $ parallelK (K.fromEffect m) r
256-- This is a co-parallel like combinator for streams, where first stream is the
257-- main stream and the rest are just supporting it, when the first ends
260-- | Like `parallel` but stops the output as soon as the first stream stops.
263{-# INLINE parallelFstK #-}
264parallelFstK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a
265parallelFstK = joinStreamVarPar ParallelVar StopBy
267-- This is a race like combinator for streams.
269-- | Like `parallel` but stops the output as soon as any of the two streams
273{-# INLINE parallelMinK #-}
274parallelMinK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a
275parallelMinK = joinStreamVarPar ParallelVar StopAny
277------------------------------------------------------------------------------
278-- Convert a stream to parallel
279------------------------------------------------------------------------------
281-- | Like 'mkParallel' but uses StreamK internally.
285mkParallelK :: MonadAsync m => K.StreamK m a -> K.StreamK m a
286mkParallelK m = K.mkStream $ \st yld sng stp -> do
287 sv <- newParallelVar StopNone (adaptState st)
288 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
289 SVar.toSVarParallel st sv $ D.fromStreamK m
290 K.foldStream st yld sng stp $ Stream.toStreamK $ SVar.fromSVar sv
292-- | Same as 'mkParallel' but for StreamD stream.
294{-# INLINE_NORMAL mkParallelD #-}
295mkParallelD :: MonadAsync m => D.Stream m a -> D.Stream m a
296mkParallelD m = D.Stream step Nothing
299 step gst Nothing = do
300 sv <- newParallelVar StopNone gst
301 SVar.toSVarParallel gst sv m
302 -- XXX use unfold instead?
303 return $ Skip $ Just $ SVar.fromSVarD sv
305 step gst (Just (D.UnStream step1 st)) = do
308 Yield a s -> Yield a (Just $ D.Stream step1 s)
309 Skip s -> Skip (Just $ D.Stream step1 s)
312-------------------------------------------------------------------------------
314-------------------------------------------------------------------------------
316-- NOTE: In regular pull style streams, the consumer stream is pulling elements
317-- from the SVar and we have several workers producing elements and pushing to
318-- SVar. In case of folds, we, the parent stream driving the fold, are the
319-- stream producing worker, we start an SVar and start pushing to the SVar, the
320-- fold on the other side of the SVar is the consumer stream.
322-- In the pull stream case exceptions are propagated from the producing workers
323-- to the consumer stream, the exceptions are propagated on the same channel as
324-- the produced stream elements. However, in case of push style folds the
325-- current stream itself is the worker and the fold is the consumer, in this
326-- case we have to propagate the exceptions from the consumer to the producer.
327-- This is reverse of the pull case and we need a reverse direction channel
328-- to propagate the exception.
330-- | Redirect a copy of the stream to a supplied fold and run it concurrently
331-- in an independent thread. The fold may buffer some elements. The buffer size
332-- is determined by the prevailing 'Streamly.Prelude.maxBuffer' setting.
337-- -----stream m a ---------------stream m a-----
342-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
347-- Exceptions from the concurrently running fold are propagated to the current
348-- computation. Note that, because of buffering in the fold, exceptions may be
349-- delayed and may not correspond to the current element being processed in the
350-- parent stream, but we guarantee that before the parent stream stops the tap
351-- finishes and all exceptions from it are drained.
354-- Compare with 'tap'.
357{-# INLINE tapAsyncK #-}
359 MonadAsync m => (K.StreamK m a -> m b) -> K.StreamK m a -> K.StreamK m a
360tapAsyncK f m = K.mkStream $ \st yld sng stp -> do
361 sv <- SVar.newFoldSVar st (f . Stream.toStreamK)
362 K.foldStreamShared st yld sng stp
363 $ Stream.toStreamK (SVar.teeToSVar sv $ Stream.fromStreamK m)
365data TapState fs st a = TapInit | Tapping !fs st | TapDone st
367-- | Like 'tapAsync' but uses a 'Fold' instead of a fold function.
369{-# INLINE_NORMAL tapAsyncF #-}
370tapAsyncF :: MonadAsync m => Fold m a b -> D.Stream m a -> D.Stream m a
371tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit
375 -- In general, a Stop event would come equipped with the result
376 -- of the fold. It is not used here but it would be useful in
377 -- applicative and distribute.
378 done <- SVar.fromConsumer svr
380 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
381 $ takeMVar (outputDoorBellFromConsumer svr)
385 liftIO $ sendStop svr Nothing
386 -- drain/wait until a stop event arrives from the fold.
389 {-# INLINE_LATE step #-}
390 step gst TapInit = do
391 sv <- SVar.newFoldSVarF gst f
392 return $ Skip (Tapping sv state1)
394 step gst (Tapping sv st) = do
398 done <- SVar.pushToFold sv a
401 -- XXX we do not need to wait synchronously here
403 return $ Yield a (TapDone s)
404 else return $ Yield a (Tapping sv s)
405 Skip s -> return $ Skip (Tapping sv s)
410 step gst (TapDone st) = do
413 Yield a s -> Yield a (TapDone s)
414 Skip s -> Skip (TapDone s)
417------------------------------------------------------------------------------
419------------------------------------------------------------------------------
421-- | For 'ParallelT' streams:
424-- (<>) = 'Streamly.Prelude.parallel'
425-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.parallel'
428-- See 'Streamly.Prelude.AsyncT', 'ParallelT' is similar except that all
429-- iterations are strictly concurrent while in 'AsyncT' it depends on the
430-- consumer demand and available threads. See 'parallel' for more details.
432-- /Since: 0.1.0 ("Streamly")/
434-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
437newtype ParallelT m a = ParallelT {getParallelT :: K.StreamK m a}
439#if !(MIN_VERSION_transformers(0,6,0))
440instance MonadTrans ParallelT where
442 lift = ParallelT . K.fromEffect
445-- | A parallely composing IO stream of elements of type @a@.
446-- See 'ParallelT' documentation for more details.
448-- /Since: 0.2.0 ("Streamly")/
451type Parallel = ParallelT IO
453------------------------------------------------------------------------------
455------------------------------------------------------------------------------
458{-# SPECIALIZE append :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
459append :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
460append (ParallelT m1) (ParallelT m2) = ParallelT $ parallelK m1 m2
462instance MonadAsync m => Semigroup (ParallelT m a) where
465------------------------------------------------------------------------------
467------------------------------------------------------------------------------
469instance MonadAsync m => Monoid (ParallelT m a) where
470 mempty = ParallelT K.nil
473------------------------------------------------------------------------------
475------------------------------------------------------------------------------
477{-# INLINE apParallel #-}
478{-# SPECIALIZE apParallel ::
479 ParallelT IO (a -> b) -> ParallelT IO a -> ParallelT IO b #-}
480apParallel :: MonadAsync m =>
481 ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b
482apParallel (ParallelT m1) (ParallelT m2) =
483 let f x1 = K.concatMapWith parallelK (K.fromPure . x1) m2
484 in ParallelT $ K.concatMapWith parallelK f m1
486instance (Monad m, MonadAsync m) => Applicative (ParallelT m) where
488 pure = ParallelT . K.fromPure
493------------------------------------------------------------------------------
495------------------------------------------------------------------------------
498{-# SPECIALIZE bind ::
499 ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
500bind :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
501bind (ParallelT m) f = ParallelT $ K.bindWith parallelK m (getParallelT . f)
503instance MonadAsync m => Monad (ParallelT m) where
509------------------------------------------------------------------------------
511------------------------------------------------------------------------------
513#if !(MIN_VERSION_transformers(0,6,0))
514instance (MonadBase b m, Monad m, MonadAsync m) => MonadBase b (ParallelT m) where
515 liftBase = liftBaseDefault
518MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
520-------------------------------------------------------------------------------
522-------------------------------------------------------------------------------
524-- Note: we can use another API with two callbacks stop and yield if we want
525-- the callback to be able to indicate end of stream.
527-- | Generates a callback and a stream pair. The callback returned is used to
528-- queue values to the stream. The stream is infinite, there is no way for the
529-- callback to indicate that it is done now.
533{-# INLINE_NORMAL newCallbackStream #-}
534newCallbackStream :: MonadAsync m => m (a -> m (), K.StreamK m a)
535newCallbackStream = do
536 sv <- newParallelVar StopNone defState
538 -- XXX Add our own thread-id to the SVar as we can not know the callback's
539 -- thread-id and the callback is not run in a managed worker. We need to
540 -- handle this better.
541 liftIO myThreadId >>= modifyThread sv
543 let callback a = liftIO $ void $ send sv (ChildYield a)
544 -- XXX we can return an SVar and then the consumer can unfold from the
546 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