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
46 -- | Stream combinators using a concurrent channel
49 -- | Evaluates a stream concurrently using a channel.
51 -- Add unfoldrM/iterateM?
54 -- | Uses a single channel to evaluate all actions.
59 -- | Uses a single channel to evaluate all actions.
64 -- | Use a channel for each pair.
72 -- | Shares a single channel across many streams.
81 -- ** Stream of streams
86 -- | Shares a single channel across many streams.
102import Control.Concurrent (myThreadId, killThread)
103import Control.Monad (void, when)
104import Control.Monad.IO.Class (MonadIO(liftIO))
105import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
106import Streamly.Internal.Control.ForkLifted (forkManaged)
107import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)
108import Streamly.Internal.Data.Stream.Channel.Types
112import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
113import Streamly.Internal.Data.Stream.StreamD.Type (Stream)
114import Streamly.Internal.Data.Stream.StreamD (Step(..))
116import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
117import qualified Streamly.Internal.Data.Stream.StreamD as Stream
118import qualified Streamly.Internal.Data.Stream.StreamD as D
119import qualified Streamly.Internal.Data.Stream.StreamK as K
120import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
122import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
123import Streamly.Internal.Data.Stream.Concurrent.Channel
127-- Imports for example snippets in this module.
130-- >>> {-# LANGUAGE FlexibleContexts #-}
131-- >>> import Control.Concurrent (threadDelay)
132-- >>> import qualified Streamly.Data.Array as Array
133-- >>> import qualified Streamly.Data.Fold as Fold
134-- >>> import qualified Streamly.Data.Parser as Parser
135-- >>> import qualified Streamly.Data.StreamK as StreamK
136-- >>> import qualified Streamly.Internal.Data.Stream as Stream hiding (append2)
137-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream
138-- >>> import Prelude hiding (concatMap, concat, zipWith)
141-- threadDelay (n * 1000000) -- sleep for n seconds
142-- putStrLn (show n ++ " sec") -- print "n sec"
146-------------------------------------------------------------------------------
147-- Evaluating a stream
148-------------------------------------------------------------------------------
151{-# INLINE_NORMAL parEvalD #-}
152parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
153parEvalD modifier m = D.Stream step Nothing
157 chan <- newChannel modifier
158 sendFirstWorker chan (D.toStreamK m)
159 -- XXX should use an unfold to make this efficient
160 return $ D.Skip $ Just $ fromChannelD chan
162 step gst (Just (D.UnStream step1 st)) = do
165 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
166 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
170-- | Evaluate a stream asynchronously. In a serial stream, each element of the
171-- stream is generated as it is demanded by the consumer. `parEval` evaluates
172-- multiple elements of the stream ahead of time and serves the results from a
175-- Note that the evaluation requires only one thread as only one stream needs
176-- to be evaluated. Therefore, the concurrency options that are relevant to
177-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,
178-- ordered, stopWhen options won't have any effect.
182-- >>> parUnfoldrM step = Stream.parEval id . Stream.unfoldrM step
183-- >>> parIterateM step = Stream.parEval id . Stream.iterateM step
184{-# INLINE parEval #-}
185parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
186parEval modifier input = withChannel modifier input (const id)
187 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
189-------------------------------------------------------------------------------
190-- combining two streams
191-------------------------------------------------------------------------------
193{-# INLINE _appendGeneric #-}
194_appendGeneric :: MonadAsync m =>
195 ((Config -> Config) -> m (Channel m a))
196 -> (Config -> Config)
200_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
205 chan <- newChan modifier
206 let cfg = modifier defaultConfig
207 done = K.nilM (stopChannel chan)
208 case getStopWhen cfg of
210 toChannelK chan stream2
211 toChannelK chan stream1
213 toChannelK chan stream2
214 toChannelK chan (K.append stream1 done)
216 toChannelK chan (K.append stream2 done)
217 toChannelK chan (K.append stream1 done)
218 return $ Stream.toStreamK $ fromChannel chan
220-- | Create a new channel and add both the streams to it for async evaluation.
221-- The output stream is the result of the evaluation.
222{-# INLINE appendWithK #-}
223appendWithK :: MonadAsync m =>
224 (Config -> Config) -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
225appendWithK modifier stream1 stream2 =
227 if getOrdered (modifier defaultConfig)
228 then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
229 else _appendGeneric Append.newChannel modifier stream1 stream2
231 parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
233-- | Evaluate the first stream in the current thread and add the second stream
234-- to the supplied channel. This is to be used by a worker thread.
236-- This can be used with parConcatMap:
239-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
242{-# INLINE _appendWithChanK #-}
243_appendWithChanK :: MonadAsync m =>
244 Channel m a -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
245_appendWithChanK chan stream1 stream2 =
246 K.before (toChannelK chan stream2) stream1
248-- | Binary operation to evaluate two streams concurrently using a channel.
250-- If you want to combine more than two streams you almost always want the
251-- 'parList' or `parConcat` operation instead. The performance of this
252-- operation degrades rapidly when more streams are combined as each operation
253-- adds one more concurrent channel. On the other hand, 'parConcat' uses a
254-- single channel for all streams. However, with this operation you can
255-- precisely control the scheduling by creating arbitrary shape expression
260-- >>> parTwo cfg x y = Stream.parList cfg [x, y]
262-- Example, the following code finishes in 4 seconds:
264-- >>> async = Stream.parTwo id
265-- >>> stream1 = Stream.fromEffect (delay 4)
266-- >>> stream2 = Stream.fromEffect (delay 2)
267-- >>> Stream.fold Fold.toList $ stream1 `async` stream2
273parTwo :: MonadAsync m =>
274 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
275parTwo modifier stream1 stream2 =
278 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
280-------------------------------------------------------------------------------
282-------------------------------------------------------------------------------
284-- | A runner function takes a queuing function @q@ and a stream, it splits the
285-- input stream, queuing the tail and using the head to generate a stream.
286-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
287-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
289{-# INLINE mkEnqueue #-}
290mkEnqueue :: MonadAsync m =>
292 -> ((K.StreamK m a -> m ()) -> K.StreamK m a -> K.StreamK m b)
293 -> m (K.StreamK m a -> m ())
294mkEnqueue chan runner = do
295 runInIO <- askRunInIO
298 -- Enqueue the outer loop
299 liftIO $ enqueue chan False (runInIO, runner q stream)
300 -- XXX In case of eager dispatch we can just directly dispatch
301 -- a worker with the tail stream here rather than first queuing
302 -- and then dispatching a worker which dequeues the work. The
303 -- older implementation did a direct dispatch here and its perf
304 -- characterstics looked much better.
308-- | Takes the head element of the input stream and queues the tail of the
309-- stream to the channel, then maps the supplied function on the head and
310-- evaluates the resulting stream.
312-- This function is designed to be used by worker threads on a channel to
313-- concurrently map and evaluate a stream.
314{-# INLINE parConcatMapChanK #-}
315parConcatMapChanK :: MonadAsync m =>
316 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
317parConcatMapChanK chan f stream =
318 let run q = concatMapDivK q f
319 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
320 -- K.parConcatMap (_appendWithChanK chan) f stream
322{-# INLINE parConcatMapChanKAny #-}
323parConcatMapChanKAny :: MonadAsync m =>
324 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
325parConcatMapChanKAny chan f stream =
326 let done = K.nilM (stopChannel chan)
327 run q = concatMapDivK q (\x -> K.append (f x) done)
328 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
330{-# INLINE parConcatMapChanKFirst #-}
331parConcatMapChanKFirst :: MonadAsync m =>
332 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
333parConcatMapChanKFirst chan f stream =
334 let done = K.nilM (stopChannel chan)
335 run q = concatMapDivK q f
336 in K.concatEffect $ do
337 res <- K.uncons stream
339 Nothing -> return K.nil
341 q <- mkEnqueue chan run
343 return $ K.append (f h) done
345{-# INLINE parConcatMapChanKGeneric #-}
346parConcatMapChanKGeneric :: MonadAsync m =>
349 -> (a -> K.StreamK m b)
352parConcatMapChanKGeneric modifier chan f stream = do
353 let cfg = modifier defaultConfig
354 case getStopWhen cfg of
355 AllStop -> parConcatMapChanK chan f stream
356 FirstStops -> parConcatMapChanKFirst chan f stream
357 AnyStops -> parConcatMapChanKAny chan f stream
359-- XXX Add a deep evaluation variant that evaluates individual elements in the
360-- generated streams in parallel.
362-- | Allocate a channel and use it to concurrently evaluate the streams
363-- generated by the mapped function.
365{-# INLINE parConcatMapK #-}
366parConcatMapK :: MonadAsync m =>
367 (Config -> Config) -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
368parConcatMapK modifier f input =
369 let g = parConcatMapChanKGeneric modifier
370 in withChannelK modifier input (`g` f)
372-- | Map each element of the input to a stream and then concurrently evaluate
373-- and concatenate the resulting streams. Multiple streams may be evaluated
374-- concurrently but earlier streams are perferred. Output from the streams are
375-- used as they arrive.
379-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
383-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
385-- The following streams finish in 4 seconds:
387-- >>> stream1 = Stream.fromEffect (delay 4)
388-- >>> stream2 = Stream.fromEffect (delay 2)
389-- >>> stream3 = Stream.fromEffect (delay 1)
390-- >>> f id [stream1, stream2, stream3]
396-- Limiting threads to 2 schedules the third stream only after one of the first
397-- two has finished, releasing a thread:
399-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
403-- When used with a Single thread it behaves like serial concatMap:
405-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
409-- >>> stream1 = Stream.fromList [1,2,3]
410-- >>> stream2 = Stream.fromList [4,5,6]
411-- >>> f (Stream.maxThreads 1) [stream1, stream2]
414-- Schedule all streams in a round robin fashion over the available threads:
416-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
418-- >>> stream1 = Stream.fromList [1,2,3]
419-- >>> stream2 = Stream.fromList [4,5,6]
420-- >>> f (Stream.maxThreads 1) [stream1, stream2]
423{-# INLINE parConcatMap #-}
424parConcatMap :: MonadAsync m =>
425 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
426parConcatMap modifier f stream =
429 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
431-- | Evaluate the streams in the input stream concurrently and combine them.
433-- >>> parConcat modifier = Stream.parConcatMap modifier id
435{-# INLINE parConcat #-}
436parConcat :: MonadAsync m =>
437 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
438parConcat modifier = parConcatMap modifier id
440-------------------------------------------------------------------------------
442-------------------------------------------------------------------------------
444-- | Like 'parConcat' but works on a list of streams.
446-- >>> parList modifier = Stream.parConcat modifier . Stream.fromList
448{-# INLINE parList #-}
449parList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
450parList modifier = parConcat modifier . Stream.fromList
452-- | Like 'concat' but works on a list of streams.
454-- >>> parListLazy = Stream.parList id
456{-# INLINE parListLazy #-}
457parListLazy :: MonadAsync m => [Stream m a] -> Stream m a
458parListLazy = parList id
460-- | Like 'parListLazy' but interleaves the streams fairly instead of prioritizing
461-- the left stream. This schedules all streams in a round robin fashion over
462-- limited number of threads.
464-- >>> parListInterleaved = Stream.parList (Stream.interleaved True)
466{-# INLINE parListInterleaved #-}
467parListInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
468parListInterleaved = parList (interleaved True)
470-- | Like 'parListLazy' but with 'ordered' on.
472-- >>> parListOrdered = Stream.parList (Stream.ordered True)
474{-# INLINE parListOrdered #-}
475parListOrdered :: MonadAsync m => [Stream m a] -> Stream m a
476parListOrdered = parList (ordered True)
478-- | Like 'parListLazy' but with 'eager' on.
480-- >>> parListEager = Stream.parList (Stream.eager True)
482{-# INLINE parListEager #-}
483parListEager :: MonadAsync m => [Stream m a] -> Stream m a
484parListEager = parList (eager True)
486-- | Like 'parListEager' but stops the output as soon as the first stream stops.
488-- >>> parListEagerFst = Stream.parList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
490{-# INLINE parListEagerFst #-}
491parListEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
492parListEagerFst = parList (eager True . stopWhen FirstStops)
494-- | Like 'parListEager' but stops the output as soon as any of the two streams
499-- >>> parListEagerMin = Stream.parList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
501{-# INLINE parListEagerMin #-}
502parListEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
503parListEagerMin = parList (eager True . stopWhen AnyStops)
505-------------------------------------------------------------------------------
507-------------------------------------------------------------------------------
509-- | Apply an argument stream to a function stream concurrently. Uses a
510-- shared channel for all individual applications within a stream application.
511{-# INLINE parApply #-}
512{-# SPECIALIZE parApply ::
513 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
514parApply :: MonadAsync m =>
515 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
516parApply modifier stream1 stream2 =
519 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
522-------------------------------------------------------------------------------
524-------------------------------------------------------------------------------
529-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
531-- For example, the following finishes in 3 seconds (as opposed to 6 seconds)
532-- because all actions run in parallel. Even though results are available out
533-- of order they are ordered due to the config option:
535-- >>> f x = delay x >> return x
536-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
542{-# INLINE parMapM #-}
543parMapM :: MonadAsync m =>
544 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
545parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
549-- >>> parSequence modifier = Stream.parMapM modifier id
553-- >>> parFromListM = Stream.parSequence id . Stream.fromList
554-- >>> parFromFoldableM = Stream.parSequence id . StreamK.toStream . StreamK.fromFoldable
556{-# INLINE parSequence #-}
557parSequence :: MonadAsync m =>
558 (Config -> Config) -> Stream m (m a) -> Stream m a
559parSequence modifier = parMapM modifier id
561-- | Evaluates the streams being zipped in separate threads than the consumer.
562-- The zip function is evaluated in the consumer thread.
564-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
566-- Multi-stream concurrency options won't apply here, see the notes in
569-- If you want to evaluate the zip function as well in a separate thread, you
570-- can use a 'parEval' on 'parZipWithM'.
572{-# INLINE parZipWithM #-}
573parZipWithM :: MonadAsync m
574 => (Config -> Config)
579parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
582-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
584-- >>> m1 = Stream.fromList [1,2,3]
585-- >>> m2 = Stream.fromList [4,5,6]
586-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
587-- [(1,4),(2,5),(3,6)]
589{-# INLINE parZipWith #-}
590parZipWith :: MonadAsync m
591 => (Config -> Config)
596parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
598-- | Like 'mergeByM' but evaluates both the streams concurrently.
602-- >>> parMergeByM cfg f m1 m2 = Stream.mergeByM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
604{-# INLINE parMergeByM #-}
605parMergeByM :: MonadAsync m
606 => (Config -> Config)
607 -> (a -> a -> m Ordering)
611parMergeByM cfg f m1 m2 = Stream.mergeByM f (parEval cfg m1) (parEval cfg m2)
613-- | Like 'mergeBy' but evaluates both the streams concurrently.
617-- >>> parMergeBy cfg f = Stream.parMergeByM cfg (\a b -> return $ f a b)
619{-# INLINE parMergeBy #-}
620parMergeBy :: MonadAsync m
621 => (Config -> Config)
622 -> (a -> a -> Ordering)
626parMergeBy cfg f = parMergeByM cfg (\a b -> return $ f a b)
628-------------------------------------------------------------------------------
630-------------------------------------------------------------------------------
632-- | Same as 'concatIterate' but concurrent.
635{-# INLINE parConcatIterate #-}
636parConcatIterate :: MonadAsync m =>
641parConcatIterate modifier f input =
643 $ withChannelK modifier (Stream.toStreamK input) iterateStream
647 iterateStream channel stream =
648 parConcatMapChanKGeneric modifier channel (generate channel) stream
651 -- XXX The channel q should be FIFO for DFS, otherwise it is BFS
652 x `K.cons` iterateStream channel (Stream.toStreamK $ f x)
654-------------------------------------------------------------------------------
656-------------------------------------------------------------------------------
661-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
663-- Generate a stream by repeatedly executing a monadic action forever.
664{-# INLINE parRepeatM #-}
665parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
666parRepeatM cfg = parSequence cfg . Stream.repeat
668-- | Generate a stream by concurrently performing a monadic action @n@ times.
672-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
674-- Example, 'parReplicateM' in the following example executes all the
675-- replicated actions concurrently, thus taking only 1 second:
677-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
680{-# INLINE parReplicateM #-}
681parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
682parReplicateM cfg n = parSequence cfg . Stream.replicate n
684-------------------------------------------------------------------------------
686-------------------------------------------------------------------------------
688-- Note: we can use another API with two callbacks stop and yield if we want
689-- the callback to be able to indicate end of stream. Or we can use a Maybe
690-- stream where Nothing indicates end of stream.
692-- XXX Rename to parNewCallback
694-- | Returns an entangled pair of a callback and a stream i.e. whenever the
695-- callback is called a value appears in the stream. The stream is infinite,
696-- there is no way for the callback to indicate that it is done now.
698-- The callback queues a value to a concurrent channel associated with the
699-- stream. The stream can be evaluated safely in any thread.
703{-# INLINE_NORMAL newCallbackStream #-}
704newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
705newCallbackStream = do
706 chan <- newChannel (eager True)
708 -- XXX Add our own thread-id to the SVar as we can not know the callback's
709 -- thread-id and the callback is not run in a managed worker. We need to
710 -- handle this better. The caller thread might be killed by the Channel if
711 -- the stream evaluator dies.
714 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
716 -- XXX We can use a "Maybe a" here. Use Nothing to send a Stop event.
721 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
722 -- XXX Use fromChannelD?
723 return (callback, fromChannel chan)
725-- XXX Rename this to parSetCallback. Also take the Channel config as argument.
726-- What config can be set by user here?
728-- XXX What happens if an exception occurs when evaluating the stream? The
729-- result of callback can be used to communicate that. But we can only know
730-- about the exception on the next callback call. For better handling the user
731-- can supply an exception sender function as argument to fromCallback.
733-- | @fromCallback f@ creates an entangled pair of a callback and a stream i.e.
734-- whenever the callback is called a value appears in the stream. The function
735-- @f@ is invoked with the callback as argument, and the stream is returned.
736-- @f@ would store the callback for calling it later for generating values in
739-- The callback queues a value to a concurrent channel associated with the
740-- stream. The stream can be evaluated safely in any thread.
744{-# INLINE fromCallback #-}
745fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
746fromCallback setCallback = Stream.concatEffect $ do
747 (callback, stream) <- newCallbackStream
751-- | @parTapCount predicate fold stream@ taps the count of those elements in
752-- the stream that pass the @predicate@. The resulting count stream is sent to
753-- a @fold@ running concurrently in another thread.
755-- For example, to print the count of elements processed every second:
757-- >>> rate = Stream.rollingMap2 (flip (-)) . Stream.delayPost 1
758-- >>> report = Stream.fold (Fold.drainMapM print) . rate
759-- >>> tap = Stream.parTapCount (const True) report
760-- >>> go = Stream.fold Fold.drain $ tap $ Stream.enumerateFrom 0
762-- Note: This may not work correctly on 32-bit machines because of Int
767{-# INLINE_NORMAL parTapCount #-}
771 -> (D.Stream m Int -> m b)
774parTapCount predicate fld (D.Stream step state) = D.Stream step' Nothing
777 {-# INLINE_LATE step' #-}
779 -- As long as we are using an "Int" for counts lockfree reads from
780 -- Var should work correctly on both 32-bit and 64-bit machines.
781 -- However, an Int on a 32-bit machine may overflow quickly.
782 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
785 $ Unboxed.toStreamD countVar
786 return $ Skip (Just (countVar, tid, state))
788 step' gst (Just (countVar, tid, st)) = do
793 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
794 return $ Yield x (Just (countVar, tid, s))
795 Skip s -> return $ Skip (Just (countVar, tid, s))
797 liftIO $ killThread tid
800-- Deprecate in a major release.
801-- {-# DEPRECATED tapCount "Please use parTapCount instead." #-}
802-- | Same as 'parTapCount'. Deprecated.
803{-# INLINE tapCount #-}
807 -> (Stream m Int -> m b)
810tapCount = parTapCount
812-------------------------------------------------------------------------------
814-------------------------------------------------------------------------------
816-- Clone a stream into n streams, perform some processing on them and then zip
817-- or merge the results in different ways?
819-- For serial processing combining n scans into a single scan on the source
820-- stream would be the most efficient way of doing this. But this has a
821-- limitation that we process one element at a time through the combined scan.
822-- This is a way to combine the states of different scans in a modular way.
823-- This works well for cases where each scan consumes and produces one element
824-- at a time. If different scans produce elements by consuming different number
825-- of elements then this may become complicated, inconvenient to use.
827-- It does not make much sense to clone a stream to multiple free streams
828-- unless we enforce processing those streams in independent threads. If
829-- we are anyway running them in the same thread then there is not much point
830-- of cloning, we can just map a function on the stream to do multiple tasks in
833-- Cloning a stream to multiple free streams can provide independent buffering
834-- and speed of evaluation to each cloned stream pipeline. For example, we can
835-- parseBreak each stream independently using a different parser. The
836-- evaluation would be push driven. The source stream would be evaluated in a
837-- separate thread and we would push the generated elements to all the cloned
840-- 1. If the cloned streams have infinite buffers then this can lead to the
841-- source stream getting evaluated faster than consumers and buffering the
842-- entire source stream in cloned streams.
844-- 2. If the cloned streams have limited buffers, then they will all go at the
845-- speed of the slowest stream if they are run concurrently.
847-- 3. If the cloned streams have limited buffers and are evaluated serially
848-- then we may run into deadlock if we are deep evaluating one stream and the
849-- source gets blocked because other stream's buffer are full.
851-- This is somewhat like list sharing. And it will have the same space leak
852-- issues if used incorrectly. In fact, we can evaluate the source stream to
853-- generate a lazy list using unsafePerformIO and share that lazy list among
854-- multiple consumers. The evaluation of the list would drive the stream. And
855-- the list would be naturally shared across consumers which can use different
856-- buffering. This would be more like the lazy IO model. However, it may be
857-- better to use streams instead of lists because streams use a monad and lists
858-- are pure - pure lists can lead to the same issues as lazy IO when used in
861-- Therefore, for safety, it makes better sense to use consumers (Stream m a ->
862-- m b) rather than generating free streams as results. Each such consumer can
863-- be enforced to run in its own thread. We can also pass a result collector
864-- callback in a ReaderT env to collect the results from all these consumers
865-- into a single stream.
867-- parTap -- tap the stream to a single consumer
868-- parDistribute -- a finite list of consumers is specified. all consumers
869-- are guaranteed to get the entire stream from beginning. Run each consumer in
871-- parDistributeStream -- consumers can join the distribution channel
872-- dynamically, they will get the source stream from now onwards.
874-- XXX We could use Stream or StreamK, what are the pros and cons? The StreamK
875-- version can be used to implement parDistribute using foldr?
878parTap :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> Stream m a
879parTap f m = undefined
881-- Can we just use a parEval fold in tap?
882-- We can easily convert the Fold to "Stream m a -> m b" form. Check if this
883-- provides the same perf as above.
885parTap :: MonadAsync m => Fold m a b -> Stream m a -> Stream m a
886parTap f xs = undefined
888-- Can we just use a parallel distribute fold in tap?
889-- Maybe better to use a custom impl of distribute?
890{-# INLINE parDistribute #-}
891parDistribute :: (Foldable f, , MonadAsync m)
892 => f (Stream m a -> m b) -> Stream m a -> Stream m a
893parDistribute = 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