2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE UndecidableInstances #-} -- XXX
11-- Module : Streamly.Streams.Parallel
12-- Copyright : (c) 2017 Harendra Kumar
15-- Maintainer : harendra.kumar@gmail.com
16-- Stability : experimental
20module Streamly.Streams.Parallel
30 -- * Function application
39import Control.Concurrent (myThreadId)
40import Control.Exception (SomeException(..), throwIO)
41import Control.Monad (ap)
42import Control.Monad.Base (MonadBase(..), liftBaseDefault)
43import Control.Monad.Catch (MonadThrow, throwM)
44-- import Control.Monad.Error.Class (MonadError(..))
45import Control.Monad.IO.Class (MonadIO(..))
46import Control.Monad.Reader.Class (MonadReader(..))
47import Control.Monad.State.Class (MonadState(..))
48import Control.Monad.Trans.Class (MonadTrans(lift))
49import Data.Functor (void)
50import Data.IORef (readIORef, writeIORef)
51import Data.Maybe (fromJust)
52import Data.Semigroup (Semigroup(..))
53import Prelude hiding (map)
55import qualified Data.Set as Set
57import Streamly.Streams.SVar (fromSVar, fromStreamVar)
58import Streamly.Streams.Serial (map)
60import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream,
61 foldStreamShared, adapt)
62import qualified Streamly.Streams.StreamK as K
64#include "Instances.hs"
66-------------------------------------------------------------------------------
68-------------------------------------------------------------------------------
70{-# NOINLINE runOne #-}
73 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
75 case getYieldLimit st of
77 Just _ -> runOneLimited st m0 winfo
82 liftIO $ decrementBufferLimit sv
83 foldStreamShared st yieldk single stop m
85 sv = fromJust $ streamVar st
88 incrementBufferLimit sv
90 sendit a = liftIO $ void $ send sv (ChildYield a)
91 single a = sendit a >> (liftIO $ sendStop sv winfo)
92 yieldk a r = sendit a >> go r
96 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
97runOneLimited st m0 winfo = go m0
102 yieldLimitOk <- liftIO $ decrementYieldLimit sv
105 liftIO $ decrementBufferLimit sv
106 foldStreamShared st yieldk single stop m
107 else liftIO $ cleanupSVarFromWorker sv
109 sv = fromJust $ streamVar st
112 incrementBufferLimit sv
113 incrementYieldLimit sv
115 sendit a = liftIO $ void $ send sv (ChildYield a)
116 single a = sendit a >> (liftIO $ sendStop sv winfo)
117 yieldk a r = sendit a >> go r
119{-# NOINLINE forkSVarPar #-}
120forkSVarPar :: (IsStream t, MonadAsync m)
121 => SVarStopStyle -> t m a -> t m a -> t m a
122forkSVarPar ss m r = mkStream $ \st yld sng stp -> do
123 sv <- newParallelVar ss st
124 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
126 StopBy -> liftIO $ do
127 set <- readIORef (workerThreads sv)
128 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
130 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
131 foldStream st yld sng stp (fromSVar sv)
133{-# INLINE joinStreamVarPar #-}
134joinStreamVarPar :: (IsStream t, MonadAsync m)
135 => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a
136joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->
138 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
139 pushWorkerPar sv (runOne st $ toStream m1)
140 foldStreamShared st yld sng stp m2
141 _ -> foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
143-- | XXX we can implement it more efficienty by directly implementing instead
144-- of combining streams using parallel.
145{-# INLINE consMParallel #-}
146{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
147consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
148consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
151-- Different variants of parallel differ in how the combined stream ends:
153-- - - parallel - ends when all stream end
154-- parallelEndByFirst - parallel1 - ends when the first stream ends
155-- parallelEndByAny - parallel_ - ends when any stream ends
157-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
158-- Merges two streams concurrently.
161{-# INLINE parallel #-}
162parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
163parallel = joinStreamVarPar ParallelVar StopNone
165-- This is a co-parallel like combinator for streams, where first stream is the
166-- main stream and the rest are just supporting it, when the first ends
169-- | Like `parallel` but stops the output as soon as the first stream stops.
172{-# INLINE parallelEndByFirst #-}
173parallelEndByFirst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
174parallelEndByFirst = joinStreamVarPar ParallelVar StopBy
176-- This is a race like combinator for streams.
178-- | Like `parallel` but stops the output as soon as any of the two streams
182{-# INLINE parallelEndByAny #-}
183parallelEndByAny :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
184parallelEndByAny = joinStreamVarPar ParallelVar StopAny
186------------------------------------------------------------------------------
187-- Convert a stream to parallel
188------------------------------------------------------------------------------
190mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
192 sv <- newParallelVar StopNone defState
193 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
196------------------------------------------------------------------------------
197-- Stream to stream concurrent function application
198------------------------------------------------------------------------------
200{-# INLINE applyParallel #-}
201applyParallel :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
202applyParallel f m = mkStream $ \st yld sng stp -> do
203 sv <- newParallelVar StopNone (adaptState st)
204 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
205 foldStream st yld sng stp $ f $ fromSVar sv
207------------------------------------------------------------------------------
208-- Stream runner concurrent function application
209------------------------------------------------------------------------------
211{-# INLINE foldParallel #-}
212foldParallel :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
214 sv <- newParallelVar StopNone defState
215 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
218{-# INLINE teeToSVar #-}
219teeToSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> t m a
220teeToSVar svr m = mkStream $ \st yld sng stp -> do
221 foldStreamShared st yld sng stp (go svr m)
225 go sv m0 = mkStream $ \st yld sng stp -> do
228 incrementBufferLimit sv
230 -- XXX wait for Stop response event on exception channel,
231 -- drain the exception channel.
233 sendit a = liftIO $ do
234 -- XXX check for exceptions before decrement so that we do not
235 -- block forever if the child already exited with an exception.
236 decrementBufferLimit sv
237 void $ send sv (ChildYield a)
238 single a = sendit a >> (liftIO $ sendStop sv Nothing) >> sng a
239 yieldk a r = sendit a >> yld a (go sv r)
240 in foldStreamShared st yieldk single stop m0
242-- In case of folds the roles of worker and parent on an SVar are reversed. The
243-- parent stream pushes values to an SVar instead of pulling from it and a
244-- worker thread running the fold pulls from the SVar and folds the stream. We
245-- keep a separate channel for pushing exceptions in the reverse direction i.e.
246-- from the fold to the parent stream.
248-- NOTE: If we use fromSVar here it will kill the main computation (the parent)
249-- when the SVar goes away so we use fromStreamVar instead.
251{-# NOINLINE handleChildException #-}
252handleChildException :: SVar t m a -> SomeException -> IO ()
253handleChildException _sv e = do
255 -- void $ sendReverse sv (ChildStop tid (Just e))
258-- | Redirect a copy of the stream to a supplied fold and run it concurrently
259-- in an independent thread. The fold may buffer some elements. The buffer size
260-- is determined by the prevailing 'maxBuffer' setting.
265-- -----stream m a ---------------stream m a-----
270-- > S.drain $ S.tee (S.mapM_ print) (S.enumerateFromTo 1 2)
275-- Exceptions from the concurrently running fold are propagated to the current
276-- computation. Note that, because of buffering in the fold, exceptions may be
277-- delayed and may not correspond to the current element being processed in the
278-- parent stream, but we guarantee that the tap finishes and all exceptions
279-- from it are drained before the parent stream stops.
282-- Compare with 'tap'.
286tee :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
287tee f m = mkStream $ \st yld sng stp -> do
288 -- Buffer size for the SVar is derived from the current state
289 sv <- newParallelVar StopNone (adaptState st)
290 -- XXX exception handling
291 -- XXX if we terminate due to an exception, do we need to actively
292 -- terminate the fold?
293 liftIO myThreadId >>= modifyThread sv
294 void $ doFork (void $ f $ fromStream $ fromStreamVar sv)
295 (svarMrun sv) (handleChildException sv)
296 foldStreamShared st yld sng stp (teeToSVar sv m)
298------------------------------------------------------------------------------
299-- Concurrent Application
300------------------------------------------------------------------------------
308-- | Parallel function application operator for streams; just like the regular
309-- function application operator '$' except that it is concurrent. The
310-- following code prints a value every second even though each stage adds a 1
316-- S.mapM (\\x -> threadDelay 1000000 >> print x)
317-- |$ S.repeatM (threadDelay 1000000 >> return 1)
324(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
325f |$ x = applyParallel f x
327-- | Parallel reverse function application operator for streams; just like the
328-- regular reverse function application operator '&' except that it is
333-- S.repeatM (threadDelay 1000000 >> return 1)
334-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
341(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
344-- | Parallel function application operator; applies a @run@ or @fold@ function
345-- to a stream such that the fold consumer and the stream producer run in
346-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
347-- underlying monad. The @.@ at the end of the operator is a mnemonic for
348-- termination of the stream.
351-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
352-- |$. S.repeatM (threadDelay 1000000 >> return 1)
359(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
360f |$. x = foldParallel f x
362-- | Parallel reverse function application operator for applying a run or fold
363-- functions to a stream. Just like '|$.' except that the operands are reversed.
366-- S.repeatM (threadDelay 1000000 >> return 1)
367-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
374(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
377------------------------------------------------------------------------------
379------------------------------------------------------------------------------
381-- | Async composition with simultaneous traversal of all streams.
383-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
384-- running both strictly concurrently and yielding elements from both streams
385-- as they arrive. When multiple streams are combined using 'ParallelT' each
386-- one is evaluated in its own thread and the results produced are presented in
387-- the combined stream on a first come first serve basis.
389-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
390-- specific type of consumption pattern (depth first or breadth first). Since
391-- they are lookahead, they may introduce certain default latency in starting
392-- more concurrent tasks for efficiency reasons or may put a default limitation
393-- on the resource consumption (e.g. number of concurrent threads for
394-- lookahead). If we look at the implementation detail, they both can share a
395-- pool of worker threads to evaluate the streams in the desired pattern and at
396-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
397-- evaluate each stream.
399-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
400-- constituent streams fairly in a round robin fashion.
401-- However, the key difference is that 'WAsyncT' is lazy or pull driven
402-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
403-- starts concurrent evaluation of both the streams (in separate threads) and
404-- later picks the results whereas 'WAsyncT' may wait for a certain latency
405-- threshold before initiating concurrent evaluation of the next stream. The
406-- concurrent scheduling of the next stream or the degree of concurrency is
407-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
408-- is evaluated in a separate thread and results are /pushed/ to a shared
409-- output buffer, the evaluation rate is controlled by blocking when the buffer
412-- Concurrent lookahead streams are generally more efficient than
413-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
414-- they do not necessarily use a separate thread for each task. So they should
415-- be preferred over 'ParallelT' especially when efficiency is a concern and
416-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
417-- for cases when the streams are required to be evaluated simultaneously
418-- irrespective of how the consumer consumes them e.g. when we want to race
419-- two tasks and want to start both strictly at the same time or if we have
420-- timers in the parallel tasks and our results depend on the timers being
421-- started at the same time. We can say that 'ParallelT' is almost the same
422-- (modulo some implementation differences) as 'WAsyncT' when the latter is
423-- used with unlimited lookahead and zero latency in initiating lookahead.
426-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
432-- When streams with more than one element are merged, it yields whichever
433-- stream yields first without any bias, unlike the 'Async' style streams.
435-- Any exceptions generated by a constituent stream are propagated to the
436-- output stream. The output and exceptions from a single stream are guaranteed
437-- to arrive in the same order in the resulting stream as they were generated
438-- in the input stream. However, the relative ordering of elements from
439-- different streams in the resulting stream can vary depending on scheduling
440-- and generation delays.
442-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
443-- of the loop concurrently.
447-- import qualified "Streamly.Prelude" as S
448-- import Control.Concurrent
450-- main = 'drain' . 'parallely' $ do
451-- n <- return 3 \<\> return 2 \<\> return 1
453-- threadDelay (n * 1000000)
454-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
457-- ThreadId 40: Delay 1
458-- ThreadId 39: Delay 2
459-- ThreadId 38: Delay 3
462-- Note that parallel composition can only combine a finite number of
463-- streams as it needs to retain state for each unfinished stream.
466newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
467 deriving (MonadTrans)
469-- | A parallely composing IO stream of elements of type @a@.
470-- See 'ParallelT' documentation for more details.
473type Parallel = ParallelT IO
475-- | Fix the type of a polymorphic stream as 'ParallelT'.
478parallely :: IsStream t => ParallelT m a -> t m a
481instance IsStream ParallelT where
482 toStream = getParallelT
483 fromStream = ParallelT
486 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
487 consM = consMParallel
490 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
493------------------------------------------------------------------------------
495------------------------------------------------------------------------------
497{-# INLINE mappendParallel #-}
498{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
499mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
500mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
502instance MonadAsync m => Semigroup (ParallelT m a) where
503 (<>) = mappendParallel
505------------------------------------------------------------------------------
507------------------------------------------------------------------------------
509instance MonadAsync m => Monoid (ParallelT m a) where
513------------------------------------------------------------------------------
515------------------------------------------------------------------------------
517{-# INLINE bindParallel #-}
518{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
519bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
520bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
522instance MonadAsync m => Monad (ParallelT m) where
526------------------------------------------------------------------------------
528------------------------------------------------------------------------------
530MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
531MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
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