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
29 -- * Function application
38import Control.Monad (ap)
39import Control.Monad.Base (MonadBase(..), liftBaseDefault)
40import Control.Monad.Catch (MonadThrow, throwM)
41-- import Control.Monad.Error.Class (MonadError(..))
42import Control.Monad.IO.Class (MonadIO(..))
43import Control.Monad.Reader.Class (MonadReader(..))
44import Control.Monad.State.Class (MonadState(..))
45import Control.Monad.Trans.Class (MonadTrans(lift))
46import Data.Functor (void)
47import Data.IORef (readIORef, writeIORef)
48import Data.Maybe (fromJust)
49import Data.Semigroup (Semigroup(..))
50import Prelude hiding (map)
52import qualified Data.Set as Set
54import Streamly.Streams.SVar (fromSVar)
55import Streamly.Streams.Serial (map)
57import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream,
58 foldStreamShared, adapt)
59import qualified Streamly.Streams.StreamK as K
61#include "Instances.hs"
63-------------------------------------------------------------------------------
65-------------------------------------------------------------------------------
67{-# NOINLINE runOne #-}
70 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
72 case getYieldLimit st of
74 Just _ -> runOneLimited st m0 winfo
79 liftIO $ decrementBufferLimit sv
80 foldStreamShared st yieldk single stop m
82 sv = fromJust $ streamVar st
85 incrementBufferLimit sv
87 sendit a = liftIO $ void $ send sv (ChildYield a)
88 single a = sendit a >> (liftIO $ sendStop sv winfo)
89 yieldk a r = sendit a >> go r
93 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
94runOneLimited st m0 winfo = go m0
99 yieldLimitOk <- liftIO $ decrementYieldLimit sv
102 liftIO $ decrementBufferLimit sv
103 foldStreamShared st yieldk single stop m
104 else liftIO $ cleanupSVarFromWorker sv
106 sv = fromJust $ streamVar st
109 incrementBufferLimit sv
110 incrementYieldLimit sv
112 sendit a = liftIO $ void $ send sv (ChildYield a)
113 single a = sendit a >> (liftIO $ sendStop sv winfo)
114 yieldk a r = sendit a >> go r
116{-# NOINLINE forkSVarPar #-}
117forkSVarPar :: (IsStream t, MonadAsync m)
118 => SVarStopStyle -> t m a -> t m a -> t m a
119forkSVarPar ss m r = mkStream $ \st yld sng stp -> do
120 sv <- newParallelVar ss st
121 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
123 StopBy -> liftIO $ do
124 set <- readIORef (workerThreads sv)
125 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
127 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
128 foldStream st yld sng stp (fromSVar sv)
130{-# INLINE joinStreamVarPar #-}
131joinStreamVarPar :: (IsStream t, MonadAsync m)
132 => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a
133joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->
135 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
136 pushWorkerPar sv (runOne st $ toStream m1)
137 foldStreamShared st yld sng stp m2
138 _ -> foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
140-- | XXX we can implement it more efficienty by directly implementing instead
141-- of combining streams using parallel.
142{-# INLINE consMParallel #-}
143{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
144consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
145consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
147-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
148-- Merges two streams concurrently.
151{-# INLINE parallel #-}
152parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
153parallel = joinStreamVarPar ParallelVar StopNone
155-- This is a co-parallel like combinator for streams, where first stream is the
156-- main stream and the rest are just supporting it, when the first ends
159-- | Like `parallel` but stops the output as soon as the first stream stops.
162{-# INLINE parallelEndByFirst #-}
163parallelEndByFirst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
164parallelEndByFirst = joinStreamVarPar ParallelVar StopBy
166-- This is a race like combinator for streams.
168-- | Like `parallel` but stops the output as soon as any of the two streams
172{-# INLINE parallelEndByAny #-}
173parallelEndByAny :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
174parallelEndByAny = joinStreamVarPar ParallelVar StopAny
176------------------------------------------------------------------------------
177-- Convert a stream to parallel
178------------------------------------------------------------------------------
180mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
182 sv <- newParallelVar StopNone defState
183 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
186------------------------------------------------------------------------------
187-- Stream to stream concurrent function application
188------------------------------------------------------------------------------
190{-# INLINE applyParallel #-}
191applyParallel :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
192applyParallel f m = mkStream $ \st yld sng stp -> do
193 sv <- newParallelVar StopNone (adaptState st)
194 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
195 foldStream st yld sng stp $ f $ fromSVar sv
197------------------------------------------------------------------------------
198-- Stream runner concurrent function application
199------------------------------------------------------------------------------
201{-# INLINE foldParallel #-}
202foldParallel :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
204 sv <- newParallelVar StopNone defState
205 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
208------------------------------------------------------------------------------
209-- Concurrent Application
210------------------------------------------------------------------------------
218-- | Parallel function application operator for streams; just like the regular
219-- function application operator '$' except that it is concurrent. The
220-- following code prints a value every second even though each stage adds a 1
226-- S.mapM (\\x -> threadDelay 1000000 >> print x)
227-- |$ S.repeatM (threadDelay 1000000 >> return 1)
234(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
235f |$ x = applyParallel f x
237-- | Parallel reverse function application operator for streams; just like the
238-- regular reverse function application operator '&' except that it is
243-- S.repeatM (threadDelay 1000000 >> return 1)
244-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
251(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
254-- | Parallel function application operator; applies a @run@ or @fold@ function
255-- to a stream such that the fold consumer and the stream producer run in
256-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
257-- underlying monad. The @.@ at the end of the operator is a mnemonic for
258-- termination of the stream.
261-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
262-- |$. S.repeatM (threadDelay 1000000 >> return 1)
269(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
270f |$. x = foldParallel f x
272-- | Parallel reverse function application operator for applying a run or fold
273-- functions to a stream. Just like '|$.' except that the operands are reversed.
276-- S.repeatM (threadDelay 1000000 >> return 1)
277-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
284(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
287------------------------------------------------------------------------------
289------------------------------------------------------------------------------
291-- | Async composition with simultaneous traversal of all streams.
293-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
294-- running both strictly concurrently and yielding elements from both streams
295-- as they arrive. When multiple streams are combined using 'ParallelT' each
296-- one is evaluated in its own thread and the results produced are presented in
297-- the combined stream on a first come first serve basis.
299-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
300-- specific type of consumption pattern (depth first or breadth first). Since
301-- they are lookahead, they may introduce certain default latency in starting
302-- more concurrent tasks for efficiency reasons or may put a default limitation
303-- on the resource consumption (e.g. number of concurrent threads for
304-- lookahead). If we look at the implementation detail, they both can share a
305-- pool of worker threads to evaluate the streams in the desired pattern and at
306-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
307-- evaluate each stream.
309-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
310-- constituent streams fairly in a round robin fashion.
311-- However, the key difference is that 'WAsyncT' is lazy or pull driven
312-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
313-- starts concurrent evaluation of both the streams (in separate threads) and
314-- later picks the results whereas 'WAsyncT' may wait for a certain latency
315-- threshold before initiating concurrent evaluation of the next stream. The
316-- concurrent scheduling of the next stream or the degree of concurrency is
317-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
318-- is evaluated in a separate thread and results are /pushed/ to a shared
319-- output buffer, the evaluation rate is controlled by blocking when the buffer
322-- Concurrent lookahead streams are generally more efficient than
323-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
324-- they do not necessarily use a separate thread for each task. So they should
325-- be preferred over 'ParallelT' especially when efficiency is a concern and
326-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
327-- for cases when the streams are required to be evaluated simultaneously
328-- irrespective of how the consumer consumes them e.g. when we want to race
329-- two tasks and want to start both strictly at the same time or if we have
330-- timers in the parallel tasks and our results depend on the timers being
331-- started at the same time. We can say that 'ParallelT' is almost the same
332-- (modulo some implementation differences) as 'WAsyncT' when the latter is
333-- used with unlimited lookahead and zero latency in initiating lookahead.
336-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
342-- When streams with more than one element are merged, it yields whichever
343-- stream yields first without any bias, unlike the 'Async' style streams.
345-- Any exceptions generated by a constituent stream are propagated to the
346-- output stream. The output and exceptions from a single stream are guaranteed
347-- to arrive in the same order in the resulting stream as they were generated
348-- in the input stream. However, the relative ordering of elements from
349-- different streams in the resulting stream can vary depending on scheduling
350-- and generation delays.
352-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
353-- of the loop concurrently.
357-- import qualified "Streamly.Prelude" as S
358-- import Control.Concurrent
360-- main = 'drain' . 'parallely' $ do
361-- n <- return 3 \<\> return 2 \<\> return 1
363-- threadDelay (n * 1000000)
364-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
367-- ThreadId 40: Delay 1
368-- ThreadId 39: Delay 2
369-- ThreadId 38: Delay 3
372-- Note that parallel composition can only combine a finite number of
373-- streams as it needs to retain state for each unfinished stream.
376newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
377 deriving (MonadTrans)
379-- | A parallely composing IO stream of elements of type @a@.
380-- See 'ParallelT' documentation for more details.
383type Parallel = ParallelT IO
385-- | Fix the type of a polymorphic stream as 'ParallelT'.
388parallely :: IsStream t => ParallelT m a -> t m a
391instance IsStream ParallelT where
392 toStream = getParallelT
393 fromStream = ParallelT
396 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
397 consM = consMParallel
400 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
403------------------------------------------------------------------------------
405------------------------------------------------------------------------------
407{-# INLINE mappendParallel #-}
408{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
409mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
410mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
412instance MonadAsync m => Semigroup (ParallelT m a) where
413 (<>) = mappendParallel
415------------------------------------------------------------------------------
417------------------------------------------------------------------------------
419instance MonadAsync m => Monoid (ParallelT m a) where
423------------------------------------------------------------------------------
425------------------------------------------------------------------------------
427{-# INLINE bindParallel #-}
428{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
429bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
430bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
432instance MonadAsync m => Monad (ParallelT m) where
436------------------------------------------------------------------------------
438------------------------------------------------------------------------------
440MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
441MONAD_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