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
28 -- * Function application
37import Control.Monad (ap)
38import Control.Monad.Base (MonadBase(..), liftBaseDefault)
39import Control.Monad.Catch (MonadThrow, throwM)
40-- import Control.Monad.Error.Class (MonadError(..))
41import Control.Monad.IO.Class (MonadIO(..))
42import Control.Monad.Reader.Class (MonadReader(..))
43import Control.Monad.State.Class (MonadState(..))
44import Control.Monad.Trans.Class (MonadTrans(lift))
45import Data.Functor (void)
46import Data.IORef (newIORef)
47import Data.Maybe (fromJust)
48import Data.Semigroup (Semigroup(..))
49import Prelude hiding (map)
52import Streamly.Streams.SVar (fromSVar)
53import Streamly.Streams.Serial (map)
55import Streamly.Streams.StreamK (IsStream(..), Stream(..), adapt)
56import qualified Streamly.Streams.StreamK as K
58#include "Instances.hs"
60-------------------------------------------------------------------------------
62-------------------------------------------------------------------------------
64{-# NOINLINE runOne #-}
65runOne :: MonadIO m => State Stream m a -> Stream m a -> m ()
66runOne st m = unStream m st stop single yieldk
72 wls <- newIORef (0, fromNanoSecs 0)
73 return $ WorkerInfo 0 yc wls
74 sv = fromJust $ streamVar st
75 stop = liftIO $ winfo >>= sendStop sv
76 sendit a = liftIO $ winfo >>= \x -> sendYield sv x (ChildYield a)
77 single a = sendit a >> stop
78 -- XXX there is no flow control in parallel case. We should perhaps use a
79 -- queue and queue it back on that and exit the thread when the outputQueue
80 -- overflows. Parallel is dangerous because it can accumulate unbounded
81 -- output in the buffer.
82 yieldk a r = void (sendit a) >> runOne st r
84{-# NOINLINE forkSVarPar #-}
85forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
86forkSVarPar m r = Stream $ \st stp sng yld -> do
88 pushWorkerPar sv (runOne st{streamVar = Just sv} m)
89 pushWorkerPar sv (runOne st{streamVar = Just sv} r)
90 (unStream (fromSVar sv)) (rstState st) stp sng yld
92{-# INLINE joinStreamVarPar #-}
93joinStreamVarPar :: MonadAsync m
94 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
95joinStreamVarPar style m1 m2 = Stream $ \st stp sng yld ->
97 Just sv | svarStyle sv == style -> do
98 pushWorkerPar sv (runOne st m1)
99 unStream m2 st stp sng yld
100 _ -> unStream (forkSVarPar m1 m2) (rstState st) stp sng yld
102{-# INLINE parallelStream #-}
103parallelStream :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
104parallelStream = joinStreamVarPar ParallelVar
106-- | XXX we can implement it more efficienty by directly implementing instead
107-- of combining streams using parallel.
108{-# INLINE consMParallel #-}
109consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
110consMParallel m r = K.yieldM m `parallelStream` r
112-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
113-- Merges two streams concurrently.
116{-# INLINE parallel #-}
117parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
118parallel m1 m2 = fromStream $ parallelStream (toStream m1) (toStream m2)
120------------------------------------------------------------------------------
121-- Convert a stream to parallel
122------------------------------------------------------------------------------
124mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
127 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
130------------------------------------------------------------------------------
131-- Stream to stream concurrent function application
132------------------------------------------------------------------------------
134{-# INLINE applyWith #-}
135applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
136applyWith f m = fromStream $ Stream $ \st stp sng yld -> do
138 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
139 unStream (toStream $ f $ fromSVar sv) (rstState st) stp sng yld
141------------------------------------------------------------------------------
142-- Stream runner concurrent function application
143------------------------------------------------------------------------------
145{-# INLINE runWith #-}
146runWith :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
149 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
152------------------------------------------------------------------------------
153-- Concurrent Application
154------------------------------------------------------------------------------
162-- | Parallel function application operator for streams; just like the regular
163-- function application operator '$' except that it is concurrent. The
164-- following code prints a value every second even though each stage adds a 1
170-- S.mapM (\\x -> threadDelay 1000000 >> print x)
171-- |$ S.repeatM (threadDelay 1000000 >> return 1)
178(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
179f |$ x = applyWith f x
181-- | Parallel reverse function application operator for streams; just like the
182-- regular reverse function application operator '&' except that it is
187-- S.repeatM (threadDelay 1000000 >> return 1)
188-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
195(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
198-- | Parallel function application operator; applies a @run@ or @fold@ function
199-- to a stream such that the fold consumer and the stream producer run in
200-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
201-- underlying monad. The @.@ at the end of the operator is a mnemonic for
202-- termination of the stream.
205-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
206-- |$. S.repeatM (threadDelay 1000000 >> return 1)
213(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
216-- | Parallel reverse function application operator for applying a run or fold
217-- functions to a stream. Just like '|$.' except that the operands are reversed.
220-- S.repeatM (threadDelay 1000000 >> return 1)
221-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
228(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
231------------------------------------------------------------------------------
233------------------------------------------------------------------------------
235-- | Async composition with simultaneous traversal of all streams.
237-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
238-- running both strictly concurrently and yielding elements from both streams
239-- as they arrive. When multiple streams are combined using 'ParallelT' each
240-- one is evaluated in its own thread and the results produced are presented in
241-- the combined stream on a first come first serve basis.
243-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
244-- specific type of consumption pattern (depth first or breadth first). Since
245-- they are lookahead, they may introduce certain default latency in starting
246-- more concurrent tasks for efficiency reasons or may put a default limitation
247-- on the resource consumption (e.g. number of concurrent threads for
248-- lookahead). If we look at the implementation detail, they both can share a
249-- pool of worker threads to evaluate the streams in the desired pattern and at
250-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
251-- evaluate each stream.
253-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
254-- constituent streams fairly in a round robin fashion.
255-- However, the key difference is that 'WAsyncT' is lazy or pull driven
256-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
257-- starts concurrent evaluation of both the streams (in separate threads) and
258-- later picks the results whereas 'WAsyncT' may wait for a certain latency
259-- threshold before initiating concurrent evaluation of the next stream. The
260-- concurrent scheduling of the next stream or the degree of concurrency is
261-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
262-- is evaluated in a separate thread and results are /pushed/ to a shared
263-- output buffer, the evaluation rate is controlled by blocking when the buffer
266-- Concurrent lookahead streams are generally more efficient than
267-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
268-- they do not necessarily use a separate thread for each task. So they should
269-- be preferred over 'ParallelT' especially when efficiency is a concern and
270-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
271-- for cases when the streams are required to be evaluated simultaneously
272-- irrespective of how the consumer consumes them e.g. when we want to race
273-- two tasks and want to start both strictly at the same time or if we have
274-- timers in the parallel tasks and our results depend on the timers being
275-- started at the same time. We can say that 'ParallelT' is almost the same
276-- (modulo some implementation differences) as 'WAsyncT' when the latter is
277-- used with unlimited lookahead and zero latency in initiating lookahead.
280-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
286-- When streams with more than one element are merged, it yields whichever
287-- stream yields first without any bias, unlike the 'Async' style streams.
289-- Any exceptions generated by a constituent stream are propagated to the
290-- output stream. The output and exceptions from a single stream are guaranteed
291-- to arrive in the same order in the resulting stream as they were generated
292-- in the input stream. However, the relative ordering of elements from
293-- different streams in the resulting stream can vary depending on scheduling
294-- and generation delays.
296-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
297-- of the loop concurrently.
301-- import qualified "Streamly.Prelude" as S
302-- import Control.Concurrent
304-- main = 'runStream' . 'parallely' $ do
305-- n <- return 3 \<\> return 2 \<\> return 1
307-- threadDelay (n * 1000000)
308-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
311-- ThreadId 40: Delay 1
312-- ThreadId 39: Delay 2
313-- ThreadId 38: Delay 3
316-- Note that parallel composition can only combine a finite number of
317-- streams as it needs to retain state for each unfinished stream.
320newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
321 deriving (MonadTrans)
323-- | A parallely composing IO stream of elements of type @a@.
324-- See 'ParallelT' documentation for more details.
327type Parallel a = ParallelT IO a
329-- | Fix the type of a polymorphic stream as 'ParallelT'.
332parallely :: IsStream t => ParallelT m a -> t m a
335instance IsStream ParallelT where
336 toStream = getParallelT
337 fromStream = ParallelT
340 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
341 consM m r = fromStream $ consMParallel m (toStream r)
344 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
347------------------------------------------------------------------------------
349------------------------------------------------------------------------------
351instance MonadAsync m => Semigroup (ParallelT m a) where
354------------------------------------------------------------------------------
356------------------------------------------------------------------------------
358instance MonadAsync m => Monoid (ParallelT m a) where
362------------------------------------------------------------------------------
364------------------------------------------------------------------------------
366instance MonadAsync m => Monad (ParallelT m) where
369 = ParallelT $ K.bindWith parallelStream m (getParallelT . f)
371------------------------------------------------------------------------------
373------------------------------------------------------------------------------
375MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
376MONAD_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