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)
52import Control.Monad.Base (MonadBase(..), liftBaseDefault)
53import Control.Monad.Catch (MonadThrow, throwM)
54-- import Control.Monad.Error.Class (MonadError(..))
55import Control.Monad.IO.Class (MonadIO(..))
56import Control.Monad.Reader.Class (MonadReader(..))
57import Control.Monad.State.Class (MonadState(..))
58import Control.Monad.Trans.Class (MonadTrans(lift))
59import Data.Functor (void)
60import Data.IORef (readIORef, writeIORef)
61import Data.Maybe (fromJust)
63import Streamly.Data.Fold (Fold)
64import Streamly.Internal.Control.Concurrent (MonadAsync)
65import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))
67import qualified Data.Set as Set
68import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
69 (StreamK, foldStreamShared, mkStream, foldStream, fromEffect
70 , nil, concatMapWith, fromPure, bindWith)
71import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
72 (Stream(..), mapM, toStreamK, fromStreamK)
73import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
74import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
75import qualified Streamly.Internal.Data.Stream.Serial as Stream
77import Streamly.Internal.Data.SVar
78import Prelude hiding (map)
81#include "Instances.hs"
85-- >>> import qualified Streamly.Prelude as Stream
86-- >>> import Control.Concurrent (threadDelay)
89-- threadDelay (n * 1000000) -- sleep for n seconds
90-- putStrLn (show n ++ " sec") -- print "n sec"
94{-# INLINABLE withLocal #-}
95withLocal :: MonadReader r m => (r -> r) -> K.StreamK m a -> K.StreamK m a
97 K.mkStream $ \st yld sng stp ->
98 let single = local f . sng
99 yieldk a r = local f $ yld a (withLocal f r)
100 in K.foldStream st yieldk single (local f stp) m
102-------------------------------------------------------------------------------
104-------------------------------------------------------------------------------
106-------------------------------------------------------------------------------
107-- StreamK based worker routines
108-------------------------------------------------------------------------------
110{-# NOINLINE runOne #-}
113 => State K.StreamK m a -> K.StreamK 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.StreamK m a -> K.StreamK 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.StreamK m a -> K.StreamK m a -> K.StreamK 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.StreamK m a -> K.StreamK m a -> K.StreamK 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.StreamK m a -> K.StreamK m a -> K.StreamK 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.StreamK m a -> K.StreamK m a -> K.StreamK 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.StreamK m a -> K.StreamK 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.StreamK m a -> m b) -> K.StreamK m a -> K.StreamK 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.StreamK m a}
435instance MonadTrans ParallelT where
437 lift = ParallelT . K.fromEffect
439-- | A parallely composing IO stream of elements of type @a@.
440-- See 'ParallelT' documentation for more details.
442-- /Since: 0.2.0 ("Streamly")/
445type Parallel = ParallelT IO
447------------------------------------------------------------------------------
449------------------------------------------------------------------------------
452{-# SPECIALIZE append :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
453append :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
454append (ParallelT m1) (ParallelT m2) = ParallelT $ parallelK m1 m2
456instance MonadAsync m => Semigroup (ParallelT m a) where
459------------------------------------------------------------------------------
461------------------------------------------------------------------------------
463instance MonadAsync m => Monoid (ParallelT m a) where
464 mempty = ParallelT K.nil
467------------------------------------------------------------------------------
469------------------------------------------------------------------------------
471{-# INLINE apParallel #-}
472{-# SPECIALIZE apParallel ::
473 ParallelT IO (a -> b) -> ParallelT IO a -> ParallelT IO b #-}
474apParallel :: MonadAsync m =>
475 ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b
476apParallel (ParallelT m1) (ParallelT m2) =
477 let f x1 = K.concatMapWith parallelK (K.fromPure . x1) m2
478 in ParallelT $ K.concatMapWith parallelK f m1
480instance (Monad m, MonadAsync m) => Applicative (ParallelT m) where
482 pure = ParallelT . K.fromPure
487------------------------------------------------------------------------------
489------------------------------------------------------------------------------
492{-# SPECIALIZE bind ::
493 ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
494bind :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
495bind (ParallelT m) f = ParallelT $ K.bindWith parallelK m (getParallelT . f)
497instance MonadAsync m => Monad (ParallelT m) where
503------------------------------------------------------------------------------
505------------------------------------------------------------------------------
507MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
509-------------------------------------------------------------------------------
511-------------------------------------------------------------------------------
513-- Note: we can use another API with two callbacks stop and yield if we want
514-- the callback to be able to indicate end of stream.
516-- | Generates a callback and a stream pair. The callback returned is used to
517-- queue values to the stream. The stream is infinite, there is no way for the
518-- callback to indicate that it is done now.
522{-# INLINE_NORMAL newCallbackStream #-}
523newCallbackStream :: MonadAsync m => m (a -> m (), K.StreamK m a)
524newCallbackStream = do
525 sv <- newParallelVar StopNone defState
527 -- XXX Add our own thread-id to the SVar as we can not know the callback's
528 -- thread-id and the callback is not run in a managed worker. We need to
529 -- handle this better.
530 liftIO myThreadId >>= modifyThread sv
532 let callback a = liftIO $ void $ send sv (ChildYield a)
533 -- XXX we can return an SVar and then the consumer can unfold from the
535 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