2-- Module : Streamly.Internal.Data.Stream.Concurrent
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Non-parallelizable stream combinators like unfoldrM, iterateM etc. can be
10-- evaluated concurrently with the stream consumer by using `eval`.
11-- Parallelizable combinators like repeatM, replicateM can generate the stream
12-- concurrently using 'concatMap'.
14-- Single effects related functionality can be moved to
15-- Data.Async/Control.Async.
16-- Common Channel functionality to Data.Channel.
17-- Stream channel to Data.Stream.Channel.
19module Streamly.Internal.Data.Stream.Concurrent
24 module Streamly.Internal.Data.Stream.Concurrent.Channel
30 -- | Stream combinators using a concurrent channel
33 -- | Evaluates a stream concurrently using a channel.
35 -- Add unfoldrM/iterateM?
38 -- | Uses a single channel to evaluate all actions.
43 -- | Uses a single channel to evaluate all actions.
48 -- | Use a channel for each pair.
56 -- | Shares a single channel across many streams.
65 -- ** Stream of streams
70 -- | Shares a single channel across many streams.
86import Control.Concurrent (myThreadId, killThread)
87import Control.Monad (void, when)
88import Control.Monad.IO.Class (MonadIO(liftIO))
89import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
90import Streamly.Internal.Control.ForkLifted (forkManaged)
91import Streamly.Internal.Data.Channel.Dispatcher (modifyThread)
92import Streamly.Internal.Data.Channel.Types (ChildEvent(..))
93import Streamly.Internal.Data.Channel.Worker (sendWithDoorBell)
94import Streamly.Internal.Data.Stream (Stream, Step(..))
95import Streamly.Internal.Data.SVar.Type (adaptState)
97import qualified Streamly.Internal.Data.MutArray as Unboxed
98import qualified Streamly.Internal.Data.Stream as Stream
99import qualified Streamly.Internal.Data.Stream as D
100import qualified Streamly.Internal.Data.StreamK as K
102import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
103import Streamly.Internal.Data.Stream.Concurrent.Channel
107-- Imports for example snippets in this module.
110-- >>> {-# LANGUAGE FlexibleContexts #-}
111-- >>> import Control.Concurrent (threadDelay)
112-- >>> import qualified Streamly.Data.Array as Array
113-- >>> import qualified Streamly.Data.Fold as Fold
114-- >>> import qualified Streamly.Data.Parser as Parser
115-- >>> import qualified Streamly.Data.StreamK as StreamK
116-- >>> import qualified Streamly.Internal.Data.Stream as Stream hiding (append2)
117-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream
118-- >>> import Prelude hiding (concatMap, concat, zipWith)
121-- threadDelay (n * 1000000) -- sleep for n seconds
122-- putStrLn (show n ++ " sec") -- print "n sec"
126-------------------------------------------------------------------------------
127-- Evaluating a stream
128-------------------------------------------------------------------------------
131{-# INLINE_NORMAL parEvalD #-}
132parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
133parEvalD modifier m = D.Stream step Nothing
137 chan <- newChannel modifier
138 sendFirstWorker chan (D.toStreamK m)
139 -- XXX should use an unfold to make this efficient
140 return $ D.Skip $ Just $ fromChannelD chan
142 step gst (Just (D.UnStream step1 st)) = do
145 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
146 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
150-- | 'parEval' evaluates a stream as a whole asynchronously with respect to
151-- the consumer of the stream. A worker thread evaluates multiple elements of
152-- the stream ahead of time and buffers the results; the consumer of the stream
153-- runs in another thread consuming the elements from the buffer, thus
154-- decoupling the production and consumption of the stream. 'parEval' can be
155-- used to run different stages of a pipeline concurrently.
157-- It is important to note that 'parEval' does not evaluate individual actions
158-- in the stream concurrently with respect to each other, it merely evaluates
159-- the stream serially but in a different thread than the consumer thread,
160-- thus the consumer and producer can run concurrently. See 'parMapM' and
161-- 'parSequence' to evaluate actions in the stream concurrently.
163-- The evaluation requires only one thread as only one stream needs to be
164-- evaluated. Therefore, the concurrency options that are relevant to multiple
165-- streams do not apply here e.g. maxThreads, eager, interleaved, ordered,
166-- stopWhen options do not have any effect on 'parEval'.
170-- >>> parUnfoldrM step = Stream.parEval id . Stream.unfoldrM step
171-- >>> parIterateM step = Stream.parEval id . Stream.iterateM step
172{-# INLINE parEval #-}
173parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
174parEval modifier input = withChannel modifier input (const id)
175 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
177-------------------------------------------------------------------------------
178-- combining two streams
179-------------------------------------------------------------------------------
181{-# INLINE _appendGeneric #-}
182_appendGeneric :: MonadAsync m =>
183 ((Config -> Config) -> m (Channel m a))
184 -> (Config -> Config)
188_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
193 chan <- newChan modifier
194 let cfg = modifier defaultConfig
195 done = K.nilM (stopChannel chan)
196 case getStopWhen cfg of
198 toChannelK chan stream2
199 toChannelK chan stream1
201 toChannelK chan stream2
202 toChannelK chan (K.append stream1 done)
204 toChannelK chan (K.append stream2 done)
205 toChannelK chan (K.append stream1 done)
206 return $ Stream.toStreamK $ fromChannel chan
208-- | Create a new channel and add both the streams to it for async evaluation.
209-- The output stream is the result of the evaluation.
210{-# INLINE appendWithK #-}
211appendWithK :: MonadAsync m =>
212 (Config -> Config) -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
213appendWithK modifier stream1 stream2 =
215 if getOrdered (modifier defaultConfig)
216 then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
217 else _appendGeneric Append.newChannel modifier stream1 stream2
219 parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
221-- | Evaluate the first stream in the current thread and add the second stream
222-- to the supplied channel. This is to be used by a worker thread.
224-- This can be used with parConcatMap:
227-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
230{-# INLINE _appendWithChanK #-}
231_appendWithChanK :: MonadAsync m =>
232 Channel m a -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
233_appendWithChanK chan stream1 stream2 =
234 K.before (toChannelK chan stream2) stream1
236-- | Binary operation to evaluate two streams concurrently using a channel.
238-- If you want to combine more than two streams you almost always want the
239-- 'parList' or `parConcat` operation instead. The performance of this
240-- operation degrades rapidly when more streams are combined as each operation
241-- adds one more concurrent channel. On the other hand, 'parConcat' uses a
242-- single channel for all streams. However, with this operation you can
243-- precisely control the scheduling by creating arbitrary shape expression
248-- >>> parTwo cfg x y = Stream.parList cfg [x, y]
250-- Example, the following code finishes in 4 seconds:
252-- >>> async = Stream.parTwo id
253-- >>> stream1 = Stream.fromEffect (delay 4)
254-- >>> stream2 = Stream.fromEffect (delay 2)
255-- >>> Stream.fold Fold.toList $ stream1 `async` stream2
261parTwo :: MonadAsync m =>
262 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
263parTwo modifier stream1 stream2 =
266 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
268-------------------------------------------------------------------------------
270-------------------------------------------------------------------------------
272-- | @concatMapDivK useTail useHead stream@, divides the stream in head and
273-- tail, maps a stream generator on the head and maps an action on the tail of
274-- a stream. Returns the stream generated by the head.
276-- Used for concurrent evaluation of streams using a Channel.
277{-# INLINE concatMapDivK #-}
278concatMapDivK :: Monad m =>
279 (K.StreamK m a -> m ())
280 -> (a -> K.StreamK m b)
283concatMapDivK useTail useHead stream =
284 K.mkStream $ \st yld sng stp -> do
285 let foldShared = K.foldStreamShared st yld sng stp
286 single a = foldShared $ useHead a
287 yieldk a r = useTail r >> single a
288 in K.foldStreamShared (adaptState st) yieldk single stp stream
290-------------------------------------------------------------------------------
292-------------------------------------------------------------------------------
294-- | A runner function takes a queuing function @q@ and a stream, it splits the
295-- input stream, queuing the tail and using the head to generate a stream.
296-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
297-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
299{-# INLINE mkEnqueue #-}
300mkEnqueue :: MonadAsync m =>
302 -> ((K.StreamK m a -> m ()) -> K.StreamK m a -> K.StreamK m b)
303 -> m (K.StreamK m a -> m ())
304mkEnqueue chan runner = do
305 runInIO <- askRunInIO
308 -- Enqueue the outer loop
309 liftIO $ enqueue chan False (runInIO, runner q stream)
310 -- XXX In case of eager dispatch we can just directly dispatch
311 -- a worker with the tail stream here rather than first queuing
312 -- and then dispatching a worker which dequeues the work. The
313 -- older implementation did a direct dispatch here and its perf
314 -- characterstics looked much better.
318-- | Takes the head element of the input stream and queues the tail of the
319-- stream to the channel, then maps the supplied function on the head and
320-- evaluates the resulting stream.
322-- This function is designed to be used by worker threads on a channel to
323-- concurrently map and evaluate a stream.
324{-# INLINE parConcatMapChanK #-}
325parConcatMapChanK :: MonadAsync m =>
326 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
327parConcatMapChanK chan f stream =
328 let run q = concatMapDivK q f
329 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
330 -- K.parConcatMap (_appendWithChanK chan) f stream
332{-# INLINE parConcatMapChanKAny #-}
333parConcatMapChanKAny :: MonadAsync m =>
334 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
335parConcatMapChanKAny chan f stream =
336 let done = K.nilM (stopChannel chan)
337 run q = concatMapDivK q (\x -> K.append (f x) done)
338 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
340{-# INLINE parConcatMapChanKFirst #-}
341parConcatMapChanKFirst :: MonadAsync m =>
342 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
343parConcatMapChanKFirst chan f stream =
344 let done = K.nilM (stopChannel chan)
345 run q = concatMapDivK q f
346 in K.concatEffect $ do
347 res <- K.uncons stream
349 Nothing -> return K.nil
351 q <- mkEnqueue chan run
353 return $ K.append (f h) done
355{-# INLINE parConcatMapChanKGeneric #-}
356parConcatMapChanKGeneric :: MonadAsync m =>
359 -> (a -> K.StreamK m b)
362parConcatMapChanKGeneric modifier chan f stream = do
363 let cfg = modifier defaultConfig
364 case getStopWhen cfg of
365 AllStop -> parConcatMapChanK chan f stream
366 FirstStops -> parConcatMapChanKFirst chan f stream
367 AnyStops -> parConcatMapChanKAny chan f stream
369-- XXX Add a deep evaluation variant that evaluates individual elements in the
370-- generated streams in parallel.
372-- | Allocate a channel and use it to concurrently evaluate the streams
373-- generated by the mapped function.
375{-# INLINE parConcatMapK #-}
376parConcatMapK :: MonadAsync m =>
377 (Config -> Config) -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
378parConcatMapK modifier f input =
379 let g = parConcatMapChanKGeneric modifier
380 in withChannelK modifier input (`g` f)
382-- | Map each element of the input to a stream and then concurrently evaluate
383-- and concatenate the resulting streams. Multiple streams may be evaluated
384-- concurrently but earlier streams are perferred. Output from the streams are
385-- used as they arrive.
389-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
393-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
395-- The following streams finish in 4 seconds:
397-- >>> stream1 = Stream.fromEffect (delay 4)
398-- >>> stream2 = Stream.fromEffect (delay 2)
399-- >>> stream3 = Stream.fromEffect (delay 1)
400-- >>> f id [stream1, stream2, stream3]
406-- Limiting threads to 2 schedules the third stream only after one of the first
407-- two has finished, releasing a thread:
409-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
413-- When used with a Single thread it behaves like serial concatMap:
415-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
419-- >>> stream1 = Stream.fromList [1,2,3]
420-- >>> stream2 = Stream.fromList [4,5,6]
421-- >>> f (Stream.maxThreads 1) [stream1, stream2]
424-- Schedule all streams in a round robin fashion over the available threads:
426-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
428-- >>> stream1 = Stream.fromList [1,2,3]
429-- >>> stream2 = Stream.fromList [4,5,6]
430-- >>> f (Stream.maxThreads 1) [stream1, stream2]
433{-# INLINE parConcatMap #-}
434parConcatMap :: MonadAsync m =>
435 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
436parConcatMap modifier f stream =
439 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
441-- | Evaluate the streams in the input stream concurrently and combine them.
443-- >>> parConcat modifier = Stream.parConcatMap modifier id
445{-# INLINE parConcat #-}
446parConcat :: MonadAsync m =>
447 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
448parConcat modifier = parConcatMap modifier id
450-------------------------------------------------------------------------------
452-------------------------------------------------------------------------------
454-- | Like 'parConcat' but works on a list of streams.
456-- >>> parList modifier = Stream.parConcat modifier . Stream.fromList
458{-# INLINE parList #-}
459parList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
460parList modifier = parConcat modifier . Stream.fromList
462-- | Like 'concat' but works on a list of streams.
464-- >>> parListLazy = Stream.parList id
466{-# INLINE parListLazy #-}
467parListLazy :: MonadAsync m => [Stream m a] -> Stream m a
468parListLazy = parList id
470-- | Like 'parListLazy' but interleaves the streams fairly instead of prioritizing
471-- the left stream. This schedules all streams in a round robin fashion over
472-- limited number of threads.
474-- >>> parListInterleaved = Stream.parList (Stream.interleaved True)
476{-# INLINE parListInterleaved #-}
477parListInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
478parListInterleaved = parList (interleaved True)
480-- | Like 'parListLazy' but with 'ordered' on.
482-- >>> parListOrdered = Stream.parList (Stream.ordered True)
484{-# INLINE parListOrdered #-}
485parListOrdered :: MonadAsync m => [Stream m a] -> Stream m a
486parListOrdered = parList (ordered True)
488-- | Like 'parListLazy' but with 'eager' on.
490-- >>> parListEager = Stream.parList (Stream.eager True)
492{-# INLINE parListEager #-}
493parListEager :: MonadAsync m => [Stream m a] -> Stream m a
494parListEager = parList (eager True)
496-- | Like 'parListEager' but stops the output as soon as the first stream stops.
498-- >>> parListEagerFst = Stream.parList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
500{-# INLINE parListEagerFst #-}
501parListEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
502parListEagerFst = parList (eager True . stopWhen FirstStops)
504-- | Like 'parListEager' but stops the output as soon as any of the two streams
509-- >>> parListEagerMin = Stream.parList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
511{-# INLINE parListEagerMin #-}
512parListEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
513parListEagerMin = parList (eager True . stopWhen AnyStops)
515-------------------------------------------------------------------------------
517-------------------------------------------------------------------------------
519-- | Apply an argument stream to a function stream concurrently. Uses a
520-- shared channel for all individual applications within a stream application.
521{-# INLINE parApply #-}
522{-# SPECIALIZE parApply ::
523 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
524parApply :: MonadAsync m =>
525 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
526parApply modifier stream1 stream2 =
529 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
532-------------------------------------------------------------------------------
534-------------------------------------------------------------------------------
539-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
541-- For example, the following finishes in 3 seconds (as opposed to 6 seconds)
542-- because all actions run in parallel. Even though results are available out
543-- of order they are ordered due to the config option:
545-- >>> f x = delay x >> return x
546-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
552{-# INLINE parMapM #-}
553parMapM :: MonadAsync m =>
554 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
555parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
559-- >>> parSequence modifier = Stream.parMapM modifier id
563-- >>> parFromListM = Stream.parSequence id . Stream.fromList
564-- >>> parFromFoldableM = Stream.parSequence id . StreamK.toStream . StreamK.fromFoldable
566{-# INLINE parSequence #-}
567parSequence :: MonadAsync m =>
568 (Config -> Config) -> Stream m (m a) -> Stream m a
569parSequence modifier = parMapM modifier id
571-- | Evaluates the streams being zipped in separate threads than the consumer.
572-- The zip function is evaluated in the consumer thread.
574-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
576-- Multi-stream concurrency options won't apply here, see the notes in
579-- If you want to evaluate the zip function as well in a separate thread, you
580-- can use a 'parEval' on 'parZipWithM'.
582{-# INLINE parZipWithM #-}
583parZipWithM :: MonadAsync m
584 => (Config -> Config)
589parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
592-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
594-- >>> m1 = Stream.fromList [1,2,3]
595-- >>> m2 = Stream.fromList [4,5,6]
596-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
597-- [(1,4),(2,5),(3,6)]
599{-# INLINE parZipWith #-}
600parZipWith :: MonadAsync m
601 => (Config -> Config)
606parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
608-- | Like 'mergeByM' but evaluates both the streams concurrently.
612-- >>> parMergeByM cfg f m1 m2 = Stream.mergeByM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
614{-# INLINE parMergeByM #-}
615parMergeByM :: MonadAsync m
616 => (Config -> Config)
617 -> (a -> a -> m Ordering)
621parMergeByM cfg f m1 m2 = Stream.mergeByM f (parEval cfg m1) (parEval cfg m2)
623-- | Like 'mergeBy' but evaluates both the streams concurrently.
627-- >>> parMergeBy cfg f = Stream.parMergeByM cfg (\a b -> return $ f a b)
629{-# INLINE parMergeBy #-}
630parMergeBy :: MonadAsync m
631 => (Config -> Config)
632 -> (a -> a -> Ordering)
636parMergeBy cfg f = parMergeByM cfg (\a b -> return $ f a b)
638-------------------------------------------------------------------------------
640-------------------------------------------------------------------------------
642-- | Same as 'concatIterate' but concurrent.
645{-# INLINE parConcatIterate #-}
646parConcatIterate :: MonadAsync m =>
651parConcatIterate modifier f input =
653 $ withChannelK modifier (Stream.toStreamK input) iterateStream
657 iterateStream channel =
658 parConcatMapChanKGeneric modifier channel (generate channel)
661 -- XXX The channel q should be FIFO for DFS, otherwise it is BFS
662 x `K.cons` iterateStream channel (Stream.toStreamK $ f x)
664-------------------------------------------------------------------------------
666-------------------------------------------------------------------------------
671-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
673-- Generate a stream by repeatedly executing a monadic action forever.
674{-# INLINE parRepeatM #-}
675parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
676parRepeatM cfg = parSequence cfg . Stream.repeat
678-- | Generate a stream by concurrently performing a monadic action @n@ times.
682-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
684-- Example, 'parReplicateM' in the following example executes all the
685-- replicated actions concurrently, thus taking only 1 second:
687-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
690{-# INLINE parReplicateM #-}
691parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
692parReplicateM cfg n = parSequence cfg . Stream.replicate n
694-------------------------------------------------------------------------------
696-------------------------------------------------------------------------------
698-- Note: we can use another API with two callbacks stop and yield if we want
699-- the callback to be able to indicate end of stream. Or we can use a Maybe
700-- stream where Nothing indicates end of stream.
702-- XXX Rename to parNewCallback
704-- | Returns an entangled pair of a callback and a stream i.e. whenever the
705-- callback is called a value appears in the stream. The stream is infinite,
706-- there is no way for the callback to indicate that it is done now.
708-- The callback queues a value to a concurrent channel associated with the
709-- stream. The stream can be evaluated safely in any thread.
713{-# INLINE_NORMAL newCallbackStream #-}
714newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
715newCallbackStream = do
716 chan <- newChannel (eager True)
718 -- XXX Add our own thread-id to the SVar as we can not know the callback's
719 -- thread-id and the callback is not run in a managed worker. We need to
720 -- handle this better. The caller thread might be killed by the Channel if
721 -- the stream evaluator dies.
724 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
726 -- XXX We can use a "Maybe a" here. Use Nothing to send a Stop event.
731 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
732 -- XXX Use fromChannelD?
733 return (callback, fromChannel chan)
735-- XXX Rename this to parSetCallback. Also take the Channel config as argument.
736-- What config can be set by user here?
738-- XXX What happens if an exception occurs when evaluating the stream? The
739-- result of callback can be used to communicate that. But we can only know
740-- about the exception on the next callback call. For better handling the user
741-- can supply an exception sender function as argument to fromCallback.
743-- | @fromCallback f@ creates an entangled pair of a callback and a stream i.e.
744-- whenever the callback is called a value appears in the stream. The function
745-- @f@ is invoked with the callback as argument, and the stream is returned.
746-- @f@ would store the callback for calling it later for generating values in
749-- The callback queues a value to a concurrent channel associated with the
750-- stream. The stream can be evaluated safely in any thread.
754{-# INLINE fromCallback #-}
755fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
756fromCallback setCallback = Stream.concatEffect $ do
757 (callback, stream) <- newCallbackStream
761-- | @parTapCount predicate fold stream@ taps the count of those elements in
762-- the stream that pass the @predicate@. The resulting count stream is sent to
763-- a @fold@ running concurrently in another thread.
765-- For example, to print the count of elements processed every second:
767-- >>> rate = Stream.rollingMap2 (flip (-)) . Stream.delayPost 1
768-- >>> report = Stream.fold (Fold.drainMapM print) . rate
769-- >>> tap = Stream.parTapCount (const True) report
770-- >>> go = Stream.fold Fold.drain $ tap $ Stream.enumerateFrom 0
772-- Note: This may not work correctly on 32-bit machines because of Int
777{-# INLINE_NORMAL parTapCount #-}
781 -> (D.Stream m Int -> m b)
784parTapCount predicate fld (D.Stream step state) = D.Stream step' Nothing
787 {-# INLINE_LATE step' #-}
789 -- As long as we are using an "Int" for counts lockfree reads from
790 -- Var should work correctly on both 32-bit and 64-bit machines.
791 -- However, an Int on a 32-bit machine may overflow quickly.
792 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
795 $ Unboxed.pollIntIORef countVar
796 return $ Skip (Just (countVar, tid, state))
798 step' gst (Just (countVar, tid, st)) = do
803 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
804 return $ Yield x (Just (countVar, tid, s))
805 Skip s -> return $ Skip (Just (countVar, tid, s))
807 liftIO $ killThread tid
810{-# DEPRECATED tapCount "Please use parTapCount instead." #-}
811-- | Same as 'parTapCount'. Deprecated.
812{-# INLINE tapCount #-}
816 -> (Stream m Int -> m b)
819tapCount = parTapCount
821-------------------------------------------------------------------------------
823-------------------------------------------------------------------------------
825-- Clone a stream into n streams, perform some processing on them and then zip
826-- or merge the results in different ways?
828-- For serial processing combining n scans into a single scan on the source
829-- stream would be the most efficient way of doing this. But this has a
830-- limitation that we process one element at a time through the combined scan.
831-- This is a way to combine the states of different scans in a modular way.
832-- This works well for cases where each scan consumes and produces one element
833-- at a time. If different scans produce elements by consuming different number
834-- of elements then this may become complicated, inconvenient to use.
836-- It does not make much sense to clone a stream to multiple free streams
837-- unless we enforce processing those streams in independent threads. If
838-- we are anyway running them in the same thread then there is not much point
839-- of cloning, we can just map a function on the stream to do multiple tasks in
842-- Cloning a stream to multiple free streams can provide independent buffering
843-- and speed of evaluation to each cloned stream pipeline. For example, we can
844-- parseBreak each stream independently using a different parser. The
845-- evaluation would be push driven. The source stream would be evaluated in a
846-- separate thread and we would push the generated elements to all the cloned
849-- 1. If the cloned streams have infinite buffers then this can lead to the
850-- source stream getting evaluated faster than consumers and buffering the
851-- entire source stream in cloned streams.
853-- 2. If the cloned streams have limited buffers, then they will all go at the
854-- speed of the slowest stream if they are run concurrently.
856-- 3. If the cloned streams have limited buffers and are evaluated serially
857-- then we may run into deadlock if we are deep evaluating one stream and the
858-- source gets blocked because other stream's buffer are full.
860-- This is somewhat like list sharing. And it will have the same space leak
861-- issues if used incorrectly. In fact, we can evaluate the source stream to
862-- generate a lazy list using unsafePerformIO and share that lazy list among
863-- multiple consumers. The evaluation of the list would drive the stream. And
864-- the list would be naturally shared across consumers which can use different
865-- buffering. This would be more like the lazy IO model. However, it may be
866-- better to use streams instead of lists because streams use a monad and lists
867-- are pure - pure lists can lead to the same issues as lazy IO when used in
870-- Therefore, for safety, it makes better sense to use consumers (Stream m a ->
871-- m b) rather than generating free streams as results. Each such consumer can
872-- be enforced to run in its own thread. We can also pass a result collector
873-- callback in a ReaderT env to collect the results from all these consumers
874-- into a single stream.
876-- parTap -- tap the stream to a single consumer
877-- parDistribute -- a finite list of consumers is specified. all consumers
878-- are guaranteed to get the entire stream from beginning. Run each consumer in
880-- parDistributeStream -- consumers can join the distribution channel
881-- dynamically, they will get the source stream from now onwards.
883-- XXX We could use Stream or StreamK, what are the pros and cons? The StreamK
884-- version can be used to implement parDistribute using foldr?
887parTap :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> Stream m a
888parTap f m = undefined
890-- Can we just use a parEval fold in tap?
891-- We can easily convert the Fold to "Stream m a -> m b" form. Check if this
892-- provides the same perf as above.
894parTap :: MonadAsync m => Fold m a b -> Stream m a -> Stream m a
895parTap f xs = undefined
897-- Can we just use a parallel distribute fold in tap?
898-- Maybe better to use a custom impl of distribute?
899{-# INLINE parDistribute #-}
900parDistribute :: (Foldable f, , MonadAsync m)
901 => f (Stream m a -> m b) -> Stream m a -> Stream m a
902parDistribute = flip (Prelude.foldr parTap)
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