2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE StandaloneDeriving #-}
9{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- Module : Streamly.Streams.Parallel
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
21module Streamly.Streams.Parallel
27 , runParallelT -- deprecated
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.Semigroup (Semigroup(..))
49import Streamly.Streams.SVar (fromSVar)
50import Streamly.Streams.StreamK
53#include "Instances.hs"
55-------------------------------------------------------------------------------
57-------------------------------------------------------------------------------
59{-# NOINLINE runOne #-}
60runOne :: MonadIO m => SVar Stream m a -> Stream m a -> m ()
61runOne sv m = (unStream m) (Just sv) stop single yieldk
65 stop = liftIO $ sendStop sv
66 sendit a = liftIO $ send sv (ChildYield a)
67 single a = sendit a >> stop
68 -- XXX there is no flow control in parallel case. We should perhaps use a
69 -- queue and queue it back on that and exit the thread when the outputQueue
70 -- overflows. Parallel is dangerous because it can accumulate unbounded
71 -- output in the buffer.
72 yieldk a r = void (sendit a) >> runOne sv r
74{-# NOINLINE forkSVarPar #-}
75forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
76forkSVarPar m r = Stream $ \_ stp sng yld -> do
78 pushWorkerPar sv (runOne sv m)
79 pushWorkerPar sv (runOne sv r)
80 (unStream (fromSVar sv)) Nothing stp sng yld
82{-# INLINE joinStreamVarPar #-}
83joinStreamVarPar :: MonadAsync m
84 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
85joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
87 Just sv | svarStyle sv == style -> do
88 pushWorkerPar sv (runOne sv m1) >> (unStream m2) svr stp sng yld
89 _ -> unStream (forkSVarPar m1 m2) Nothing stp sng yld
91{-# INLINE parallelStream #-}
92parallelStream :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
93parallelStream = joinStreamVarPar ParallelVar
95-- | XXX we can implement it more efficienty by directly implementing instead
96-- of combining streams using parallel.
97{-# INLINE consMParallel #-}
98consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
99consMParallel m r = yieldM m `parallelStream` r
101-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
102-- Merges two streams concurrently.
105{-# INLINE parallel #-}
106parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
107parallel m1 m2 = fromStream $ parallelStream (toStream m1) (toStream m2)
109------------------------------------------------------------------------------
110-- Convert a stream to parallel
111------------------------------------------------------------------------------
113mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
116 pushWorkerPar sv (runOne sv $ toStream m)
119------------------------------------------------------------------------------
120-- Stream to stream concurrent function application
121------------------------------------------------------------------------------
123{-# INLINE applyWith #-}
124applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
125applyWith f m = fromStream $ Stream $ \svr stp sng yld -> do
127 pushWorkerPar sv (runOne sv (toStream m))
128 unStream (toStream $ f $ fromSVar sv) svr stp sng yld
130------------------------------------------------------------------------------
131-- Stream runner concurrent function application
132------------------------------------------------------------------------------
134{-# INLINE runWith #-}
135runWith :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
138 pushWorkerPar sv (runOne sv $ toStream m)
141------------------------------------------------------------------------------
142-- Concurrent Application
143------------------------------------------------------------------------------
151-- | Parallel function application operator for streams; just like the regular
152-- function application operator '$' except that it is concurrent. The
153-- following code prints a value every second even though each stage adds a 1
159-- S.mapM (\\x -> threadDelay 1000000 >> print x)
160-- |$ S.repeatM (threadDelay 1000000 >> return 1)
167(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
168f |$ x = applyWith f x
170-- | Parallel reverse function application operator for streams; just like the
171-- regular reverse function application operator '&' except that it is
176-- S.repeatM (threadDelay 1000000 >> return 1)
177-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
184(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
187-- | Parallel function application operator; applies a @run@ or @fold@ function
188-- to a stream such that the fold consumer and the stream producer run in
189-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
190-- underlying monad. The @.@ at the end of the operator is a mnemonic for
191-- termination of the stream.
194-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
195-- |$. S.repeatM (threadDelay 1000000 >> return 1)
202(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
205-- | Parallel reverse function application operator for applying a run or fold
206-- functions to a stream. Just like '|$.' except that the operands are reversed.
209-- S.repeatM (threadDelay 1000000 >> return 1)
210-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
217(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
220------------------------------------------------------------------------------
222------------------------------------------------------------------------------
224-- | Async composition with simultaneous traversal of all streams.
226-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
227-- running both strictly concurrently and yielding elements from both streams
228-- as they arrive. When multiple streams are combined using 'ParallelT' each
229-- one is evaluated in its own thread and the results produced are presented in
230-- the combined stream on a first come first serve basis.
232-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
233-- specific type of consumption pattern (depth first or breadth first). Since
234-- they are lookahead, they may introduce certain default latency in starting
235-- more concurrent tasks for efficiency reasons or may put a default limitation
236-- on the resource consumption (e.g. number of concurrent threads for
237-- lookahead). If we look at the implementation detail, they both can share a
238-- pool of worker threads to evaluate the streams in the desired pattern and at
239-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
240-- evaluate each stream.
242-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
243-- constituent streams fairly in a round robin fashion.
244-- However, the key difference is that 'WAsyncT' is lazy or pull driven
245-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
246-- starts concurrent evaluation of both the streams (in separate threads) and
247-- later picks the results whereas 'WAsyncT' may wait for a certain latency
248-- threshold before initiating concurrent evaluation of the next stream. The
249-- concurrent scheduling of the next stream or the degree of concurrency is
250-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
251-- is evaluated in a separate thread and results are /pushed/ to a shared
252-- output buffer, the evaluation rate is controlled by blocking when the buffer
255-- Concurrent lookahead streams are generally more efficient than
256-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
257-- they do not necessarily use a separate thread for each task. So they should
258-- be preferred over 'ParallelT' especially when efficiency is a concern and
259-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
260-- for cases when the streams are required to be evaluated simultaneously
261-- irrespective of how the consumer consumes them e.g. when we want to race
262-- two tasks and want to start both strictly at the same time or if we have
263-- timers in the parallel tasks and our results depend on the timers being
264-- started at the same time. We can say that 'ParallelT' is almost the same
265-- (modulo some implementation differences) as 'WAsyncT' when the latter is
266-- used with unlimited lookahead and zero latency in initiating lookahead.
269-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
275-- When streams with more than one element are merged, it yields whichever
276-- stream yields first without any bias, unlike the 'Async' style streams.
278-- Any exceptions generated by a constituent stream are propagated to the
279-- output stream. The output and exceptions from a single stream are guaranteed
280-- to arrive in the same order in the resulting stream as they were generated
281-- in the input stream. However, the relative ordering of elements from
282-- different streams in the resulting stream can vary depending on scheduling
283-- and generation delays.
285-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
286-- of the loop concurrently.
290-- import qualified "Streamly.Prelude" as S
291-- import Control.Concurrent
293-- main = 'runStream' . 'parallely' $ do
294-- n <- return 3 \<\> return 2 \<\> return 1
296-- threadDelay (n * 1000000)
297-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
300-- ThreadId 40: Delay 1
301-- ThreadId 39: Delay 2
302-- ThreadId 38: Delay 3
305-- Note that parallel composition can only combine a finite number of
306-- streams as it needs to retain state for each unfinished stream.
309newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
310 deriving (Functor, MonadTrans)
312-- | A parallely composing IO stream of elements of type @a@.
313-- See 'ParallelT' documentation for more details.
316type Parallel a = ParallelT IO a
318-- | Fix the type of a polymorphic stream as 'ParallelT'.
321parallely :: IsStream t => ParallelT m a -> t m a
324instance IsStream ParallelT where
325 toStream = getParallelT
326 fromStream = ParallelT
329 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
330 consM m r = fromStream $ consMParallel m (toStream r)
333 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
336------------------------------------------------------------------------------
338------------------------------------------------------------------------------
340instance MonadAsync m => Semigroup (ParallelT m a) where
343------------------------------------------------------------------------------
345------------------------------------------------------------------------------
347instance MonadAsync m => Monoid (ParallelT m a) where
351------------------------------------------------------------------------------
353------------------------------------------------------------------------------
355instance MonadAsync m => Monad (ParallelT m) where
358 = ParallelT $ bindWith parallelStream m (getParallelT . f)
360------------------------------------------------------------------------------
362------------------------------------------------------------------------------
364MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
365MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
367-- | Same as @runStream . parallely@.
370{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
371runParallelT :: Monad m => ParallelT m a -> m ()
372runParallelT = runStream
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