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
27 -- * Function application
36import Control.Monad (ap)
37import Control.Monad.Base (MonadBase(..), liftBaseDefault)
38import Control.Monad.Catch (MonadThrow, throwM)
39-- import Control.Monad.Error.Class (MonadError(..))
40import Control.Monad.IO.Class (MonadIO(..))
41import Control.Monad.Reader.Class (MonadReader(..))
42import Control.Monad.State.Class (MonadState(..))
43import Control.Monad.Trans.Class (MonadTrans(lift))
44import Data.Functor (void)
45import Data.Maybe (fromJust)
46import Data.Semigroup (Semigroup(..))
47import Prelude hiding (map)
49import Streamly.Streams.SVar (fromSVar)
50import Streamly.Streams.Serial (map)
52import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream,
53 foldStreamShared, adapt)
54import qualified Streamly.Streams.StreamK as K
56#include "Instances.hs"
58-------------------------------------------------------------------------------
60-------------------------------------------------------------------------------
62{-# NOINLINE runOne #-}
65 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
66runOne st m winfo = foldStreamShared st yieldk single stop m
70 sv = fromJust $ streamVar st
71 mrun = runInIO $ svarMrun sv
73 withLimitCheck action = do
74 yieldLimitOk <- liftIO $ decrementYieldLimitPost sv
77 else liftIO $ cleanupSVarFromWorker sv
79 stop = liftIO $ sendStop sv winfo
80 sendit a = liftIO $ sendYield sv winfo (ChildYield a)
81 single a = sendit a >> withLimitCheck stop
83 -- XXX there is no flow control in parallel case. We should perhaps use a
84 -- queue and queue it back on that and exit the thread when the outputQueue
85 -- overflows. Parallel is dangerous because it can accumulate unbounded
86 -- output in the buffer.
87 yieldk a r = void (sendit a)
88 >> withLimitCheck (void $ liftIO $ mrun $ runOne st r winfo)
90{-# NOINLINE forkSVarPar #-}
91forkSVarPar :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
92forkSVarPar m r = mkStream $ \st yld sng stp -> do
93 sv <- newParallelVar st
94 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
95 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
96 foldStream st yld sng stp (fromSVar sv)
98{-# INLINE joinStreamVarPar #-}
99joinStreamVarPar :: (IsStream t, MonadAsync m)
100 => SVarStyle -> t m a -> t m a -> t m a
101joinStreamVarPar style m1 m2 = mkStream $ \st yld sng stp ->
103 Just sv | svarStyle sv == style -> do
104 pushWorkerPar sv (runOne st $ toStream m1)
105 foldStreamShared st yld sng stp m2
106 _ -> foldStreamShared st yld sng stp (forkSVarPar m1 m2)
108-- | XXX we can implement it more efficienty by directly implementing instead
109-- of combining streams using parallel.
110{-# INLINE consMParallel #-}
111{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
112consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
113consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
115-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
116-- Merges two streams concurrently.
119{-# INLINE parallel #-}
120parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
121parallel = joinStreamVarPar ParallelVar
123------------------------------------------------------------------------------
124-- Convert a stream to parallel
125------------------------------------------------------------------------------
127mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
129 sv <- newParallelVar defState
130 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
133------------------------------------------------------------------------------
134-- Stream to stream concurrent function application
135------------------------------------------------------------------------------
137{-# INLINE applyWith #-}
138applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
139applyWith f m = mkStream $ \st yld sng stp -> do
140 sv <- newParallelVar (adaptState st)
141 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
142 foldStream st yld sng stp $ f $ fromSVar sv
144------------------------------------------------------------------------------
145-- Stream runner concurrent function application
146------------------------------------------------------------------------------
148{-# INLINE runWith #-}
149runWith :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
151 sv <- newParallelVar defState
152 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
155------------------------------------------------------------------------------
156-- Concurrent Application
157------------------------------------------------------------------------------
165-- | Parallel function application operator for streams; just like the regular
166-- function application operator '$' except that it is concurrent. The
167-- following code prints a value every second even though each stage adds a 1
173-- S.mapM (\\x -> threadDelay 1000000 >> print x)
174-- |$ S.repeatM (threadDelay 1000000 >> return 1)
181(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
182f |$ x = applyWith f x
184-- | Parallel reverse function application operator for streams; just like the
185-- regular reverse function application operator '&' except that it is
190-- S.repeatM (threadDelay 1000000 >> return 1)
191-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
198(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
201-- | Parallel function application operator; applies a @run@ or @fold@ function
202-- to a stream such that the fold consumer and the stream producer run in
203-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
204-- underlying monad. The @.@ at the end of the operator is a mnemonic for
205-- termination of the stream.
208-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
209-- |$. S.repeatM (threadDelay 1000000 >> return 1)
216(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
219-- | Parallel reverse function application operator for applying a run or fold
220-- functions to a stream. Just like '|$.' except that the operands are reversed.
223-- S.repeatM (threadDelay 1000000 >> return 1)
224-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
231(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
234------------------------------------------------------------------------------
236------------------------------------------------------------------------------
238-- | Async composition with simultaneous traversal of all streams.
240-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
241-- running both strictly concurrently and yielding elements from both streams
242-- as they arrive. When multiple streams are combined using 'ParallelT' each
243-- one is evaluated in its own thread and the results produced are presented in
244-- the combined stream on a first come first serve basis.
246-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
247-- specific type of consumption pattern (depth first or breadth first). Since
248-- they are lookahead, they may introduce certain default latency in starting
249-- more concurrent tasks for efficiency reasons or may put a default limitation
250-- on the resource consumption (e.g. number of concurrent threads for
251-- lookahead). If we look at the implementation detail, they both can share a
252-- pool of worker threads to evaluate the streams in the desired pattern and at
253-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
254-- evaluate each stream.
256-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
257-- constituent streams fairly in a round robin fashion.
258-- However, the key difference is that 'WAsyncT' is lazy or pull driven
259-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
260-- starts concurrent evaluation of both the streams (in separate threads) and
261-- later picks the results whereas 'WAsyncT' may wait for a certain latency
262-- threshold before initiating concurrent evaluation of the next stream. The
263-- concurrent scheduling of the next stream or the degree of concurrency is
264-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
265-- is evaluated in a separate thread and results are /pushed/ to a shared
266-- output buffer, the evaluation rate is controlled by blocking when the buffer
269-- Concurrent lookahead streams are generally more efficient than
270-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
271-- they do not necessarily use a separate thread for each task. So they should
272-- be preferred over 'ParallelT' especially when efficiency is a concern and
273-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
274-- for cases when the streams are required to be evaluated simultaneously
275-- irrespective of how the consumer consumes them e.g. when we want to race
276-- two tasks and want to start both strictly at the same time or if we have
277-- timers in the parallel tasks and our results depend on the timers being
278-- started at the same time. We can say that 'ParallelT' is almost the same
279-- (modulo some implementation differences) as 'WAsyncT' when the latter is
280-- used with unlimited lookahead and zero latency in initiating lookahead.
283-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
289-- When streams with more than one element are merged, it yields whichever
290-- stream yields first without any bias, unlike the 'Async' style streams.
292-- Any exceptions generated by a constituent stream are propagated to the
293-- output stream. The output and exceptions from a single stream are guaranteed
294-- to arrive in the same order in the resulting stream as they were generated
295-- in the input stream. However, the relative ordering of elements from
296-- different streams in the resulting stream can vary depending on scheduling
297-- and generation delays.
299-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
300-- of the loop concurrently.
304-- import qualified "Streamly.Prelude" as S
305-- import Control.Concurrent
307-- main = 'drain' . 'parallely' $ do
308-- n <- return 3 \<\> return 2 \<\> return 1
310-- threadDelay (n * 1000000)
311-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
314-- ThreadId 40: Delay 1
315-- ThreadId 39: Delay 2
316-- ThreadId 38: Delay 3
319-- Note that parallel composition can only combine a finite number of
320-- streams as it needs to retain state for each unfinished stream.
323newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
324 deriving (MonadTrans)
326-- | A parallely composing IO stream of elements of type @a@.
327-- See 'ParallelT' documentation for more details.
330type Parallel = ParallelT IO
332-- | Fix the type of a polymorphic stream as 'ParallelT'.
335parallely :: IsStream t => ParallelT m a -> t m a
338instance IsStream ParallelT where
339 toStream = getParallelT
340 fromStream = ParallelT
343 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
344 consM = consMParallel
347 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
350------------------------------------------------------------------------------
352------------------------------------------------------------------------------
354{-# INLINE mappendParallel #-}
355{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
356mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
357mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
359instance MonadAsync m => Semigroup (ParallelT m a) where
360 (<>) = mappendParallel
362------------------------------------------------------------------------------
364------------------------------------------------------------------------------
366instance MonadAsync m => Monoid (ParallelT m a) where
370------------------------------------------------------------------------------
372------------------------------------------------------------------------------
374{-# INLINE bindParallel #-}
375{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
376bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
377bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
379instance MonadAsync m => Monad (ParallelT m) where
383-- XXX Specialize the applicative instance
385------------------------------------------------------------------------------
387------------------------------------------------------------------------------
389MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
390MONAD_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