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 : streamly@composewell.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)
52#if __GLASGOW_HASKELL__ < 808
53import Data.Semigroup (Semigroup(..))
55import Prelude hiding (map)
57import qualified Data.Set as Set
59import Streamly.Streams.SVar (fromSVar, fromStreamVar)
60import Streamly.Streams.Serial (map)
61import Streamly.Internal.Data.SVar
62import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream,
63 foldStreamShared, adapt)
64import qualified Streamly.Streams.StreamK as K
66#include "Instances.hs"
68-------------------------------------------------------------------------------
70-------------------------------------------------------------------------------
72{-# NOINLINE runOne #-}
75 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
77 case getYieldLimit st of
79 Just _ -> runOneLimited st m0 winfo
84 liftIO $ decrementBufferLimit sv
85 foldStreamShared st yieldk single stop m
87 sv = fromJust $ streamVar st
90 incrementBufferLimit sv
92 sendit a = liftIO $ void $ send sv (ChildYield a)
93 single a = sendit a >> (liftIO $ sendStop sv winfo)
94 yieldk a r = sendit a >> go r
98 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
99runOneLimited st m0 winfo = go m0
104 yieldLimitOk <- liftIO $ decrementYieldLimit sv
107 liftIO $ decrementBufferLimit sv
108 foldStreamShared st yieldk single stop m
110 liftIO $ cleanupSVarFromWorker sv
111 liftIO $ sendStop sv winfo
113 sv = fromJust $ streamVar st
116 incrementBufferLimit sv
117 incrementYieldLimit sv
119 sendit a = liftIO $ void $ send sv (ChildYield a)
120 single a = sendit a >> (liftIO $ sendStop sv winfo)
121 yieldk a r = sendit a >> go r
123{-# NOINLINE forkSVarPar #-}
124forkSVarPar :: (IsStream t, MonadAsync m)
125 => SVarStopStyle -> t m a -> t m a -> t m a
126forkSVarPar ss m r = mkStream $ \st yld sng stp -> do
127 sv <- newParallelVar ss st
128 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
130 StopBy -> liftIO $ do
131 set <- readIORef (workerThreads sv)
132 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
134 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
135 foldStream st yld sng stp (fromSVar sv)
137{-# INLINE joinStreamVarPar #-}
138joinStreamVarPar :: (IsStream t, MonadAsync m)
139 => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a
140joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->
142 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
143 pushWorkerPar sv (runOne st $ toStream m1)
144 foldStreamShared st yld sng stp m2
145 _ -> foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
147-- | XXX we can implement it more efficienty by directly implementing instead
148-- of combining streams using parallel.
149{-# INLINE consMParallel #-}
150{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
151consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
152consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
154-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
155-- Merges two streams concurrently.
158{-# INLINE parallel #-}
159parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
160parallel = joinStreamVarPar ParallelVar StopNone
162-- This is a co-parallel like combinator for streams, where first stream is the
163-- main stream and the rest are just supporting it, when the first ends
166-- | Like `parallel` but stops the output as soon as the first stream stops.
169{-# INLINE parallelFst #-}
170parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
171parallelFst = joinStreamVarPar ParallelVar StopBy
173-- This is a race like combinator for streams.
175-- | Like `parallel` but stops the output as soon as any of the two streams
179{-# INLINE parallelMin #-}
180parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
181parallelMin = joinStreamVarPar ParallelVar StopAny
183------------------------------------------------------------------------------
184-- Convert a stream to parallel
185------------------------------------------------------------------------------
187mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
189 sv <- newParallelVar StopNone defState
190 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
193------------------------------------------------------------------------------
194-- Stream to stream concurrent function application
195------------------------------------------------------------------------------
197{-# INLINE applyParallel #-}
198applyParallel :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
199applyParallel f m = mkStream $ \st yld sng stp -> do
200 sv <- newParallelVar StopNone (adaptState st)
201 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
202 foldStream st yld sng stp $ f $ fromSVar sv
204------------------------------------------------------------------------------
205-- Stream runner concurrent function application
206------------------------------------------------------------------------------
208{-# INLINE foldParallel #-}
209foldParallel :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
211 sv <- newParallelVar StopNone defState
212 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
215{-# INLINE teeToSVar #-}
216teeToSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> t m a
217teeToSVar svr m = mkStream $ \st yld sng stp -> do
218 foldStreamShared st yld sng stp (go svr m)
222 go sv m0 = mkStream $ \st yld sng stp -> do
225 incrementBufferLimit sv
227 -- XXX wait for Stop response event on exception channel,
228 -- drain the exception channel.
230 sendit a = liftIO $ do
231 -- XXX check for exceptions before decrement so that we do not
232 -- block forever if the child already exited with an exception.
233 decrementBufferLimit sv
234 void $ send sv (ChildYield a)
235 single a = sendit a >> (liftIO $ sendStop sv Nothing) >> sng a
236 yieldk a r = sendit a >> yld a (go sv r)
237 in foldStreamShared st yieldk single stop m0
239-- In case of folds the roles of worker and parent on an SVar are reversed. The
240-- parent stream pushes values to an SVar instead of pulling from it and a
241-- worker thread running the fold pulls from the SVar and folds the stream. We
242-- keep a separate channel for pushing exceptions in the reverse direction i.e.
243-- from the fold to the parent stream.
245-- NOTE: If we use fromSVar here it will kill the main computation (the parent)
246-- when the SVar goes away so we use fromStreamVar instead.
248{-# NOINLINE handleChildException #-}
249handleChildException :: SVar t m a -> SomeException -> IO ()
250handleChildException _sv e = do
252 -- void $ sendReverse sv (ChildStop tid (Just e))
255-- | Redirect a copy of the stream to a supplied fold and run it concurrently
256-- in an independent thread. The fold may buffer some elements. The buffer size
257-- is determined by the prevailing 'maxBuffer' setting.
262-- -----stream m a ---------------stream m a-----
267-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
272-- Exceptions from the concurrently running fold are propagated to the current
273-- computation. Note that, because of buffering in the fold, exceptions may be
274-- delayed and may not correspond to the current element being processed in the
275-- parent stream, but we guarantee that the tap finishes and all exceptions
276-- from it are drained before the parent stream stops.
279-- Compare with 'tap'.
282{-# INLINE tapAsync #-}
283tapAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
284tapAsync f m = mkStream $ \st yld sng stp -> do
285 -- Buffer size for the SVar is derived from the current state
286 sv <- newParallelVar StopNone (adaptState st)
287 -- XXX exception handling
288 -- XXX if we terminate due to an exception, do we need to actively
289 -- terminate the fold?
290 liftIO myThreadId >>= modifyThread sv
291 void $ doFork (void $ f $ fromStream $ fromStreamVar sv)
292 (svarMrun sv) (handleChildException sv)
293 foldStreamShared st yld sng stp (teeToSVar sv m)
295------------------------------------------------------------------------------
296-- Concurrent Application
297------------------------------------------------------------------------------
305-- | Parallel function application operator for streams; just like the regular
306-- function application operator '$' except that it is concurrent. The
307-- following code prints a value every second even though each stage adds a 1
313-- S.mapM (\\x -> threadDelay 1000000 >> print x)
314-- |$ S.repeatM (threadDelay 1000000 >> return 1)
321(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
322f |$ x = applyParallel f x
324-- | Parallel reverse function application operator for streams; just like the
325-- regular reverse function application operator '&' except that it is
330-- S.repeatM (threadDelay 1000000 >> return 1)
331-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
338(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
341-- | Parallel function application operator; applies a @run@ or @fold@ function
342-- to a stream such that the fold consumer and the stream producer run in
343-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
344-- underlying monad. The @.@ at the end of the operator is a mnemonic for
345-- termination of the stream.
348-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
349-- |$. S.repeatM (threadDelay 1000000 >> return 1)
356(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
357f |$. x = foldParallel f x
359-- | Parallel reverse function application operator for applying a run or fold
360-- functions to a stream. Just like '|$.' except that the operands are reversed.
363-- S.repeatM (threadDelay 1000000 >> return 1)
364-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
371(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
374------------------------------------------------------------------------------
376------------------------------------------------------------------------------
378-- | Async composition with strict concurrent execution of all streams.
380-- The 'Semigroup' instance of 'ParallelT' executes both the streams
381-- concurrently without any delay or without waiting for the consumer demand
382-- and /merges/ the results as they arrive. If the consumer does not consume
383-- the results, they are buffered upto a configured maximum, controlled by the
384-- 'maxBuffer' primitive. If the buffer becomes full the concurrent tasks will
385-- block until there is space in the buffer.
387-- Both 'WAsyncT' and 'ParallelT', evaluate the constituent streams fairly in a
388-- round robin fashion. The key difference is that 'WAsyncT' might wait for the
389-- consumer demand before it executes the tasks whereas 'ParallelT' starts
390-- executing all the tasks immediately without waiting for the consumer demand.
391-- For 'WAsyncT' the 'maxThreads' limit applies whereas for 'ParallelT' it does
392-- not apply. In other words, 'WAsyncT' can be lazy whereas 'ParallelT' is
395-- 'ParallelT' is useful for cases when the streams are required to be
396-- evaluated simultaneously irrespective of how the consumer consumes them e.g.
397-- when we want to race two tasks and want to start both strictly at the same
398-- time or if we have timers in the parallel tasks and our results depend on
399-- the timers being started at the same time. If we do not have such
400-- requirements then 'AsyncT' or 'AheadT' are recommended as they can be more
401-- efficient than 'ParallelT'.
404-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
410-- When streams with more than one element are merged, it yields whichever
411-- stream yields first without any bias, unlike the 'Async' style streams.
413-- Any exceptions generated by a constituent stream are propagated to the
414-- output stream. The output and exceptions from a single stream are guaranteed
415-- to arrive in the same order in the resulting stream as they were generated
416-- in the input stream. However, the relative ordering of elements from
417-- different streams in the resulting stream can vary depending on scheduling
418-- and generation delays.
420-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
421-- of the loop concurrently.
425-- import qualified "Streamly.Prelude" as S
426-- import Control.Concurrent
428-- main = 'drain' . 'parallely' $ do
429-- n <- return 3 \<\> return 2 \<\> return 1
431-- threadDelay (n * 1000000)
432-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
435-- ThreadId 40: Delay 1
436-- ThreadId 39: Delay 2
437-- ThreadId 38: Delay 3
440-- Note that parallel composition can only combine a finite number of
441-- streams as it needs to retain state for each unfinished stream.
443-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
446newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
447 deriving (MonadTrans)
449-- | A parallely composing IO stream of elements of type @a@.
450-- See 'ParallelT' documentation for more details.
453type Parallel = ParallelT IO
455-- | Fix the type of a polymorphic stream as 'ParallelT'.
458parallely :: IsStream t => ParallelT m a -> t m a
461instance IsStream ParallelT where
462 toStream = getParallelT
463 fromStream = ParallelT
466 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
467 consM = consMParallel
470 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
473------------------------------------------------------------------------------
475------------------------------------------------------------------------------
477{-# INLINE mappendParallel #-}
478{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
479mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
480mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
482instance MonadAsync m => Semigroup (ParallelT m a) where
483 (<>) = mappendParallel
485------------------------------------------------------------------------------
487------------------------------------------------------------------------------
489instance MonadAsync m => Monoid (ParallelT m a) where
493------------------------------------------------------------------------------
495------------------------------------------------------------------------------
497{-# INLINE bindParallel #-}
498{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
499bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
500bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
502instance MonadAsync m => Monad (ParallelT m) where
506------------------------------------------------------------------------------
508------------------------------------------------------------------------------
510MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
511MONAD_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