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-- | Evaluate a stream asynchronously. In a serial stream, each element of the
151-- stream is generated as it is demanded by the consumer. `parEval` evaluates
152-- multiple elements of the stream ahead of time and serves the results from a
155-- Note that the evaluation requires only one thread as only one stream needs
156-- to be evaluated. Therefore, the concurrency options that are relevant to
157-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,
158-- ordered, stopWhen options won't have any effect.
162-- >>> parUnfoldrM step = Stream.parEval id . Stream.unfoldrM step
163-- >>> parIterateM step = Stream.parEval id . Stream.iterateM step
164{-# INLINE parEval #-}
165parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
166parEval modifier input = withChannel modifier input (const id)
167 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
169-------------------------------------------------------------------------------
170-- combining two streams
171-------------------------------------------------------------------------------
173{-# INLINE _appendGeneric #-}
174_appendGeneric :: MonadAsync m =>
175 ((Config -> Config) -> m (Channel m a))
176 -> (Config -> Config)
180_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
185 chan <- newChan modifier
186 let cfg = modifier defaultConfig
187 done = K.nilM (stopChannel chan)
188 case getStopWhen cfg of
190 toChannelK chan stream2
191 toChannelK chan stream1
193 toChannelK chan stream2
194 toChannelK chan (K.append stream1 done)
196 toChannelK chan (K.append stream2 done)
197 toChannelK chan (K.append stream1 done)
198 return $ Stream.toStreamK $ fromChannel chan
200-- | Create a new channel and add both the streams to it for async evaluation.
201-- The output stream is the result of the evaluation.
202{-# INLINE appendWithK #-}
203appendWithK :: MonadAsync m =>
204 (Config -> Config) -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
205appendWithK modifier stream1 stream2 =
207 if getOrdered (modifier defaultConfig)
208 then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
209 else _appendGeneric Append.newChannel modifier stream1 stream2
211 parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
213-- | Evaluate the first stream in the current thread and add the second stream
214-- to the supplied channel. This is to be used by a worker thread.
216-- This can be used with parConcatMap:
219-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
222{-# INLINE _appendWithChanK #-}
223_appendWithChanK :: MonadAsync m =>
224 Channel m a -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
225_appendWithChanK chan stream1 stream2 =
226 K.before (toChannelK chan stream2) stream1
228-- | Binary operation to evaluate two streams concurrently using a channel.
230-- If you want to combine more than two streams you almost always want the
231-- 'parList' or `parConcat` operation instead. The performance of this
232-- operation degrades rapidly when more streams are combined as each operation
233-- adds one more concurrent channel. On the other hand, 'parConcat' uses a
234-- single channel for all streams. However, with this operation you can
235-- precisely control the scheduling by creating arbitrary shape expression
240-- >>> parTwo cfg x y = Stream.parList cfg [x, y]
242-- Example, the following code finishes in 4 seconds:
244-- >>> async = Stream.parTwo id
245-- >>> stream1 = Stream.fromEffect (delay 4)
246-- >>> stream2 = Stream.fromEffect (delay 2)
247-- >>> Stream.fold Fold.toList $ stream1 `async` stream2
253parTwo :: MonadAsync m =>
254 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
255parTwo modifier stream1 stream2 =
258 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
260-------------------------------------------------------------------------------
262-------------------------------------------------------------------------------
264-- | @concatMapDivK useTail useHead stream@, divides the stream in head and
265-- tail, maps a stream generator on the head and maps an action on the tail of
266-- a stream. Returns the stream generated by the head.
268-- Used for concurrent evaluation of streams using a Channel.
269{-# INLINE concatMapDivK #-}
270concatMapDivK :: Monad m =>
271 (K.StreamK m a -> m ())
272 -> (a -> K.StreamK m b)
275concatMapDivK useTail useHead stream =
276 K.mkStream $ \st yld sng stp -> do
277 let foldShared = K.foldStreamShared st yld sng stp
278 single a = foldShared $ useHead a
279 yieldk a r = useTail r >> single a
280 in K.foldStreamShared (adaptState st) yieldk single stp stream
282-------------------------------------------------------------------------------
284-------------------------------------------------------------------------------
286-- | A runner function takes a queuing function @q@ and a stream, it splits the
287-- input stream, queuing the tail and using the head to generate a stream.
288-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
289-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
291{-# INLINE mkEnqueue #-}
292mkEnqueue :: MonadAsync m =>
294 -> ((K.StreamK m a -> m ()) -> K.StreamK m a -> K.StreamK m b)
295 -> m (K.StreamK m a -> m ())
296mkEnqueue chan runner = do
297 runInIO <- askRunInIO
300 -- Enqueue the outer loop
301 liftIO $ enqueue chan False (runInIO, runner q stream)
302 -- XXX In case of eager dispatch we can just directly dispatch
303 -- a worker with the tail stream here rather than first queuing
304 -- and then dispatching a worker which dequeues the work. The
305 -- older implementation did a direct dispatch here and its perf
306 -- characterstics looked much better.
310-- | Takes the head element of the input stream and queues the tail of the
311-- stream to the channel, then maps the supplied function on the head and
312-- evaluates the resulting stream.
314-- This function is designed to be used by worker threads on a channel to
315-- concurrently map and evaluate a stream.
316{-# INLINE parConcatMapChanK #-}
317parConcatMapChanK :: MonadAsync m =>
318 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
319parConcatMapChanK chan f stream =
320 let run q = concatMapDivK q f
321 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
322 -- K.parConcatMap (_appendWithChanK chan) f stream
324{-# INLINE parConcatMapChanKAny #-}
325parConcatMapChanKAny :: MonadAsync m =>
326 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
327parConcatMapChanKAny chan f stream =
328 let done = K.nilM (stopChannel chan)
329 run q = concatMapDivK q (\x -> K.append (f x) done)
330 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
332{-# INLINE parConcatMapChanKFirst #-}
333parConcatMapChanKFirst :: MonadAsync m =>
334 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
335parConcatMapChanKFirst chan f stream =
336 let done = K.nilM (stopChannel chan)
337 run q = concatMapDivK q f
338 in K.concatEffect $ do
339 res <- K.uncons stream
341 Nothing -> return K.nil
343 q <- mkEnqueue chan run
345 return $ K.append (f h) done
347{-# INLINE parConcatMapChanKGeneric #-}
348parConcatMapChanKGeneric :: MonadAsync m =>
351 -> (a -> K.StreamK m b)
354parConcatMapChanKGeneric modifier chan f stream = do
355 let cfg = modifier defaultConfig
356 case getStopWhen cfg of
357 AllStop -> parConcatMapChanK chan f stream
358 FirstStops -> parConcatMapChanKFirst chan f stream
359 AnyStops -> parConcatMapChanKAny chan f stream
361-- XXX Add a deep evaluation variant that evaluates individual elements in the
362-- generated streams in parallel.
364-- | Allocate a channel and use it to concurrently evaluate the streams
365-- generated by the mapped function.
367{-# INLINE parConcatMapK #-}
368parConcatMapK :: MonadAsync m =>
369 (Config -> Config) -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
370parConcatMapK modifier f input =
371 let g = parConcatMapChanKGeneric modifier
372 in withChannelK modifier input (`g` f)
374-- | Map each element of the input to a stream and then concurrently evaluate
375-- and concatenate the resulting streams. Multiple streams may be evaluated
376-- concurrently but earlier streams are perferred. Output from the streams are
377-- used as they arrive.
381-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
385-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
387-- The following streams finish in 4 seconds:
389-- >>> stream1 = Stream.fromEffect (delay 4)
390-- >>> stream2 = Stream.fromEffect (delay 2)
391-- >>> stream3 = Stream.fromEffect (delay 1)
392-- >>> f id [stream1, stream2, stream3]
398-- Limiting threads to 2 schedules the third stream only after one of the first
399-- two has finished, releasing a thread:
401-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
405-- When used with a Single thread it behaves like serial concatMap:
407-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
411-- >>> stream1 = Stream.fromList [1,2,3]
412-- >>> stream2 = Stream.fromList [4,5,6]
413-- >>> f (Stream.maxThreads 1) [stream1, stream2]
416-- Schedule all streams in a round robin fashion over the available threads:
418-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
420-- >>> stream1 = Stream.fromList [1,2,3]
421-- >>> stream2 = Stream.fromList [4,5,6]
422-- >>> f (Stream.maxThreads 1) [stream1, stream2]
425{-# INLINE parConcatMap #-}
426parConcatMap :: MonadAsync m =>
427 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
428parConcatMap modifier f stream =
431 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
433-- | Evaluate the streams in the input stream concurrently and combine them.
435-- >>> parConcat modifier = Stream.parConcatMap modifier id
437{-# INLINE parConcat #-}
438parConcat :: MonadAsync m =>
439 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
440parConcat modifier = parConcatMap modifier id
442-------------------------------------------------------------------------------
444-------------------------------------------------------------------------------
446-- | Like 'parConcat' but works on a list of streams.
448-- >>> parList modifier = Stream.parConcat modifier . Stream.fromList
450{-# INLINE parList #-}
451parList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
452parList modifier = parConcat modifier . Stream.fromList
454-- | Like 'concat' but works on a list of streams.
456-- >>> parListLazy = Stream.parList id
458{-# INLINE parListLazy #-}
459parListLazy :: MonadAsync m => [Stream m a] -> Stream m a
460parListLazy = parList id
462-- | Like 'parListLazy' but interleaves the streams fairly instead of prioritizing
463-- the left stream. This schedules all streams in a round robin fashion over
464-- limited number of threads.
466-- >>> parListInterleaved = Stream.parList (Stream.interleaved True)
468{-# INLINE parListInterleaved #-}
469parListInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
470parListInterleaved = parList (interleaved True)
472-- | Like 'parListLazy' but with 'ordered' on.
474-- >>> parListOrdered = Stream.parList (Stream.ordered True)
476{-# INLINE parListOrdered #-}
477parListOrdered :: MonadAsync m => [Stream m a] -> Stream m a
478parListOrdered = parList (ordered True)
480-- | Like 'parListLazy' but with 'eager' on.
482-- >>> parListEager = Stream.parList (Stream.eager True)
484{-# INLINE parListEager #-}
485parListEager :: MonadAsync m => [Stream m a] -> Stream m a
486parListEager = parList (eager True)
488-- | Like 'parListEager' but stops the output as soon as the first stream stops.
490-- >>> parListEagerFst = Stream.parList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
492{-# INLINE parListEagerFst #-}
493parListEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
494parListEagerFst = parList (eager True . stopWhen FirstStops)
496-- | Like 'parListEager' but stops the output as soon as any of the two streams
501-- >>> parListEagerMin = Stream.parList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
503{-# INLINE parListEagerMin #-}
504parListEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
505parListEagerMin = parList (eager True . stopWhen AnyStops)
507-------------------------------------------------------------------------------
509-------------------------------------------------------------------------------
511-- | Apply an argument stream to a function stream concurrently. Uses a
512-- shared channel for all individual applications within a stream application.
513{-# INLINE parApply #-}
514{-# SPECIALIZE parApply ::
515 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
516parApply :: MonadAsync m =>
517 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
518parApply modifier stream1 stream2 =
521 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
524-------------------------------------------------------------------------------
526-------------------------------------------------------------------------------
531-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
533-- For example, the following finishes in 3 seconds (as opposed to 6 seconds)
534-- because all actions run in parallel. Even though results are available out
535-- of order they are ordered due to the config option:
537-- >>> f x = delay x >> return x
538-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
544{-# INLINE parMapM #-}
545parMapM :: MonadAsync m =>
546 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
547parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
551-- >>> parSequence modifier = Stream.parMapM modifier id
555-- >>> parFromListM = Stream.parSequence id . Stream.fromList
556-- >>> parFromFoldableM = Stream.parSequence id . StreamK.toStream . StreamK.fromFoldable
558{-# INLINE parSequence #-}
559parSequence :: MonadAsync m =>
560 (Config -> Config) -> Stream m (m a) -> Stream m a
561parSequence modifier = parMapM modifier id
563-- | Evaluates the streams being zipped in separate threads than the consumer.
564-- The zip function is evaluated in the consumer thread.
566-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
568-- Multi-stream concurrency options won't apply here, see the notes in
571-- If you want to evaluate the zip function as well in a separate thread, you
572-- can use a 'parEval' on 'parZipWithM'.
574{-# INLINE parZipWithM #-}
575parZipWithM :: MonadAsync m
576 => (Config -> Config)
581parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
584-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
586-- >>> m1 = Stream.fromList [1,2,3]
587-- >>> m2 = Stream.fromList [4,5,6]
588-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
589-- [(1,4),(2,5),(3,6)]
591{-# INLINE parZipWith #-}
592parZipWith :: MonadAsync m
593 => (Config -> Config)
598parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
600-- | Like 'mergeByM' but evaluates both the streams concurrently.
604-- >>> parMergeByM cfg f m1 m2 = Stream.mergeByM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
606{-# INLINE parMergeByM #-}
607parMergeByM :: MonadAsync m
608 => (Config -> Config)
609 -> (a -> a -> m Ordering)
613parMergeByM cfg f m1 m2 = Stream.mergeByM f (parEval cfg m1) (parEval cfg m2)
615-- | Like 'mergeBy' but evaluates both the streams concurrently.
619-- >>> parMergeBy cfg f = Stream.parMergeByM cfg (\a b -> return $ f a b)
621{-# INLINE parMergeBy #-}
622parMergeBy :: MonadAsync m
623 => (Config -> Config)
624 -> (a -> a -> Ordering)
628parMergeBy cfg f = parMergeByM cfg (\a b -> return $ f a b)
630-------------------------------------------------------------------------------
632-------------------------------------------------------------------------------
634-- | Same as 'concatIterate' but concurrent.
637{-# INLINE parConcatIterate #-}
638parConcatIterate :: MonadAsync m =>
643parConcatIterate modifier f input =
645 $ withChannelK modifier (Stream.toStreamK input) iterateStream
649 iterateStream channel stream =
650 parConcatMapChanKGeneric modifier channel (generate channel) stream
653 -- XXX The channel q should be FIFO for DFS, otherwise it is BFS
654 x `K.cons` iterateStream channel (Stream.toStreamK $ f x)
656-------------------------------------------------------------------------------
658-------------------------------------------------------------------------------
663-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
665-- Generate a stream by repeatedly executing a monadic action forever.
666{-# INLINE parRepeatM #-}
667parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
668parRepeatM cfg = parSequence cfg . Stream.repeat
670-- | Generate a stream by concurrently performing a monadic action @n@ times.
674-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
676-- Example, 'parReplicateM' in the following example executes all the
677-- replicated actions concurrently, thus taking only 1 second:
679-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
682{-# INLINE parReplicateM #-}
683parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
684parReplicateM cfg n = parSequence cfg . Stream.replicate n
686-------------------------------------------------------------------------------
688-------------------------------------------------------------------------------
690-- Note: we can use another API with two callbacks stop and yield if we want
691-- the callback to be able to indicate end of stream. Or we can use a Maybe
692-- stream where Nothing indicates end of stream.
694-- XXX Rename to parNewCallback
696-- | Returns an entangled pair of a callback and a stream i.e. whenever the
697-- callback is called a value appears in the stream. The stream is infinite,
698-- there is no way for the callback to indicate that it is done now.
700-- The callback queues a value to a concurrent channel associated with the
701-- stream. The stream can be evaluated safely in any thread.
705{-# INLINE_NORMAL newCallbackStream #-}
706newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
707newCallbackStream = do
708 chan <- newChannel (eager True)
710 -- XXX Add our own thread-id to the SVar as we can not know the callback's
711 -- thread-id and the callback is not run in a managed worker. We need to
712 -- handle this better. The caller thread might be killed by the Channel if
713 -- the stream evaluator dies.
716 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
718 -- XXX We can use a "Maybe a" here. Use Nothing to send a Stop event.
723 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
724 -- XXX Use fromChannelD?
725 return (callback, fromChannel chan)
727-- XXX Rename this to parSetCallback. Also take the Channel config as argument.
728-- What config can be set by user here?
730-- XXX What happens if an exception occurs when evaluating the stream? The
731-- result of callback can be used to communicate that. But we can only know
732-- about the exception on the next callback call. For better handling the user
733-- can supply an exception sender function as argument to fromCallback.
735-- | @fromCallback f@ creates an entangled pair of a callback and a stream i.e.
736-- whenever the callback is called a value appears in the stream. The function
737-- @f@ is invoked with the callback as argument, and the stream is returned.
738-- @f@ would store the callback for calling it later for generating values in
741-- The callback queues a value to a concurrent channel associated with the
742-- stream. The stream can be evaluated safely in any thread.
746{-# INLINE fromCallback #-}
747fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
748fromCallback setCallback = Stream.concatEffect $ do
749 (callback, stream) <- newCallbackStream
753-- | @parTapCount predicate fold stream@ taps the count of those elements in
754-- the stream that pass the @predicate@. The resulting count stream is sent to
755-- a @fold@ running concurrently in another thread.
757-- For example, to print the count of elements processed every second:
759-- >>> rate = Stream.rollingMap2 (flip (-)) . Stream.delayPost 1
760-- >>> report = Stream.fold (Fold.drainMapM print) . rate
761-- >>> tap = Stream.parTapCount (const True) report
762-- >>> go = Stream.fold Fold.drain $ tap $ Stream.enumerateFrom 0
764-- Note: This may not work correctly on 32-bit machines because of Int
769{-# INLINE_NORMAL parTapCount #-}
773 -> (D.Stream m Int -> m b)
776parTapCount predicate fld (D.Stream step state) = D.Stream step' Nothing
779 {-# INLINE_LATE step' #-}
781 -- As long as we are using an "Int" for counts lockfree reads from
782 -- Var should work correctly on both 32-bit and 64-bit machines.
783 -- However, an Int on a 32-bit machine may overflow quickly.
784 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
787 $ Unboxed.pollIntIORef countVar
788 return $ Skip (Just (countVar, tid, state))
790 step' gst (Just (countVar, tid, st)) = do
795 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
796 return $ Yield x (Just (countVar, tid, s))
797 Skip s -> return $ Skip (Just (countVar, tid, s))
799 liftIO $ killThread tid
802{-# DEPRECATED tapCount "Please use parTapCount instead." #-}
803-- | Same as 'parTapCount'. Deprecated.
804{-# INLINE tapCount #-}
808 -> (Stream m Int -> m b)
811tapCount = parTapCount
813-------------------------------------------------------------------------------
815-------------------------------------------------------------------------------
817-- Clone a stream into n streams, perform some processing on them and then zip
818-- or merge the results in different ways?
820-- For serial processing combining n scans into a single scan on the source
821-- stream would be the most efficient way of doing this. But this has a
822-- limitation that we process one element at a time through the combined scan.
823-- This is a way to combine the states of different scans in a modular way.
824-- This works well for cases where each scan consumes and produces one element
825-- at a time. If different scans produce elements by consuming different number
826-- of elements then this may become complicated, inconvenient to use.
828-- It does not make much sense to clone a stream to multiple free streams
829-- unless we enforce processing those streams in independent threads. If
830-- we are anyway running them in the same thread then there is not much point
831-- of cloning, we can just map a function on the stream to do multiple tasks in
834-- Cloning a stream to multiple free streams can provide independent buffering
835-- and speed of evaluation to each cloned stream pipeline. For example, we can
836-- parseBreak each stream independently using a different parser. The
837-- evaluation would be push driven. The source stream would be evaluated in a
838-- separate thread and we would push the generated elements to all the cloned
841-- 1. If the cloned streams have infinite buffers then this can lead to the
842-- source stream getting evaluated faster than consumers and buffering the
843-- entire source stream in cloned streams.
845-- 2. If the cloned streams have limited buffers, then they will all go at the
846-- speed of the slowest stream if they are run concurrently.
848-- 3. If the cloned streams have limited buffers and are evaluated serially
849-- then we may run into deadlock if we are deep evaluating one stream and the
850-- source gets blocked because other stream's buffer are full.
852-- This is somewhat like list sharing. And it will have the same space leak
853-- issues if used incorrectly. In fact, we can evaluate the source stream to
854-- generate a lazy list using unsafePerformIO and share that lazy list among
855-- multiple consumers. The evaluation of the list would drive the stream. And
856-- the list would be naturally shared across consumers which can use different
857-- buffering. This would be more like the lazy IO model. However, it may be
858-- better to use streams instead of lists because streams use a monad and lists
859-- are pure - pure lists can lead to the same issues as lazy IO when used in
862-- Therefore, for safety, it makes better sense to use consumers (Stream m a ->
863-- m b) rather than generating free streams as results. Each such consumer can
864-- be enforced to run in its own thread. We can also pass a result collector
865-- callback in a ReaderT env to collect the results from all these consumers
866-- into a single stream.
868-- parTap -- tap the stream to a single consumer
869-- parDistribute -- a finite list of consumers is specified. all consumers
870-- are guaranteed to get the entire stream from beginning. Run each consumer in
872-- parDistributeStream -- consumers can join the distribution channel
873-- dynamically, they will get the source stream from now onwards.
875-- XXX We could use Stream or StreamK, what are the pros and cons? The StreamK
876-- version can be used to implement parDistribute using foldr?
879parTap :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> Stream m a
880parTap f m = undefined
882-- Can we just use a parEval fold in tap?
883-- We can easily convert the Fold to "Stream m a -> m b" form. Check if this
884-- provides the same perf as above.
886parTap :: MonadAsync m => Fold m a b -> Stream m a -> Stream m a
887parTap f xs = undefined
889-- Can we just use a parallel distribute fold in tap?
890-- Maybe better to use a custom impl of distribute?
891{-# INLINE parDistribute #-}
892parDistribute :: (Foldable f, , MonadAsync m)
893 => f (Stream m a -> m b) -> Stream m a -> Stream m a
894parDistribute = 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