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 (Stream, Step(..))
115import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
116import qualified Streamly.Internal.Data.Stream as Stream
117import qualified Streamly.Internal.Data.Stream as D
118import qualified Streamly.Internal.Data.StreamK as K
120import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
121import Streamly.Internal.Data.Stream.Concurrent.Channel
125-- Imports for example snippets in this module.
128-- >>> {-# LANGUAGE FlexibleContexts #-}
129-- >>> import Control.Concurrent (threadDelay)
130-- >>> import qualified Streamly.Data.Array as Array
131-- >>> import qualified Streamly.Data.Fold as Fold
132-- >>> import qualified Streamly.Data.Parser as Parser
133-- >>> import qualified Streamly.Data.StreamK as StreamK
134-- >>> import qualified Streamly.Internal.Data.Stream as Stream hiding (append2)
135-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream
136-- >>> import Prelude hiding (concatMap, concat, zipWith)
139-- threadDelay (n * 1000000) -- sleep for n seconds
140-- putStrLn (show n ++ " sec") -- print "n sec"
144-------------------------------------------------------------------------------
145-- Evaluating a stream
146-------------------------------------------------------------------------------
149{-# INLINE_NORMAL parEvalD #-}
150parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
151parEvalD modifier m = D.Stream step Nothing
155 chan <- newChannel modifier
156 sendFirstWorker chan (D.toStreamK m)
157 -- XXX should use an unfold to make this efficient
158 return $ D.Skip $ Just $ fromChannelD chan
160 step gst (Just (D.UnStream step1 st)) = do
163 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
164 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
168-- | Evaluate a stream asynchronously. In a serial stream, each element of the
169-- stream is generated as it is demanded by the consumer. `parEval` evaluates
170-- multiple elements of the stream ahead of time and serves the results from a
173-- Note that the evaluation requires only one thread as only one stream needs
174-- to be evaluated. Therefore, the concurrency options that are relevant to
175-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,
176-- ordered, stopWhen options won't have any effect.
180-- >>> parUnfoldrM step = Stream.parEval id . Stream.unfoldrM step
181-- >>> parIterateM step = Stream.parEval id . Stream.iterateM step
182{-# INLINE parEval #-}
183parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
184parEval modifier input = withChannel modifier input (const id)
185 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
187-------------------------------------------------------------------------------
188-- combining two streams
189-------------------------------------------------------------------------------
191{-# INLINE _appendGeneric #-}
192_appendGeneric :: MonadAsync m =>
193 ((Config -> Config) -> m (Channel m a))
194 -> (Config -> Config)
198_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
203 chan <- newChan modifier
204 let cfg = modifier defaultConfig
205 done = K.nilM (stopChannel chan)
206 case getStopWhen cfg of
208 toChannelK chan stream2
209 toChannelK chan stream1
211 toChannelK chan stream2
212 toChannelK chan (K.append stream1 done)
214 toChannelK chan (K.append stream2 done)
215 toChannelK chan (K.append stream1 done)
216 return $ Stream.toStreamK $ fromChannel chan
218-- | Create a new channel and add both the streams to it for async evaluation.
219-- The output stream is the result of the evaluation.
220{-# INLINE appendWithK #-}
221appendWithK :: MonadAsync m =>
222 (Config -> Config) -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
223appendWithK modifier stream1 stream2 =
225 if getOrdered (modifier defaultConfig)
226 then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
227 else _appendGeneric Append.newChannel modifier stream1 stream2
229 parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
231-- | Evaluate the first stream in the current thread and add the second stream
232-- to the supplied channel. This is to be used by a worker thread.
234-- This can be used with parConcatMap:
237-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
240{-# INLINE _appendWithChanK #-}
241_appendWithChanK :: MonadAsync m =>
242 Channel m a -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
243_appendWithChanK chan stream1 stream2 =
244 K.before (toChannelK chan stream2) stream1
246-- | Binary operation to evaluate two streams concurrently using a channel.
248-- If you want to combine more than two streams you almost always want the
249-- 'parList' or `parConcat` operation instead. The performance of this
250-- operation degrades rapidly when more streams are combined as each operation
251-- adds one more concurrent channel. On the other hand, 'parConcat' uses a
252-- single channel for all streams. However, with this operation you can
253-- precisely control the scheduling by creating arbitrary shape expression
258-- >>> parTwo cfg x y = Stream.parList cfg [x, y]
260-- Example, the following code finishes in 4 seconds:
262-- >>> async = Stream.parTwo id
263-- >>> stream1 = Stream.fromEffect (delay 4)
264-- >>> stream2 = Stream.fromEffect (delay 2)
265-- >>> Stream.fold Fold.toList $ stream1 `async` stream2
271parTwo :: MonadAsync m =>
272 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
273parTwo modifier stream1 stream2 =
276 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
278-------------------------------------------------------------------------------
280-------------------------------------------------------------------------------
282-- | A runner function takes a queuing function @q@ and a stream, it splits the
283-- input stream, queuing the tail and using the head to generate a stream.
284-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
285-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
287{-# INLINE mkEnqueue #-}
288mkEnqueue :: MonadAsync m =>
290 -> ((K.StreamK m a -> m ()) -> K.StreamK m a -> K.StreamK m b)
291 -> m (K.StreamK m a -> m ())
292mkEnqueue chan runner = do
293 runInIO <- askRunInIO
296 -- Enqueue the outer loop
297 liftIO $ enqueue chan False (runInIO, runner q stream)
298 -- XXX In case of eager dispatch we can just directly dispatch
299 -- a worker with the tail stream here rather than first queuing
300 -- and then dispatching a worker which dequeues the work. The
301 -- older implementation did a direct dispatch here and its perf
302 -- characterstics looked much better.
306-- | Takes the head element of the input stream and queues the tail of the
307-- stream to the channel, then maps the supplied function on the head and
308-- evaluates the resulting stream.
310-- This function is designed to be used by worker threads on a channel to
311-- concurrently map and evaluate a stream.
312{-# INLINE parConcatMapChanK #-}
313parConcatMapChanK :: MonadAsync m =>
314 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
315parConcatMapChanK chan f stream =
316 let run q = concatMapDivK q f
317 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
318 -- K.parConcatMap (_appendWithChanK chan) f stream
320{-# INLINE parConcatMapChanKAny #-}
321parConcatMapChanKAny :: MonadAsync m =>
322 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
323parConcatMapChanKAny chan f stream =
324 let done = K.nilM (stopChannel chan)
325 run q = concatMapDivK q (\x -> K.append (f x) done)
326 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
328{-# INLINE parConcatMapChanKFirst #-}
329parConcatMapChanKFirst :: MonadAsync m =>
330 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
331parConcatMapChanKFirst chan f stream =
332 let done = K.nilM (stopChannel chan)
333 run q = concatMapDivK q f
334 in K.concatEffect $ do
335 res <- K.uncons stream
337 Nothing -> return K.nil
339 q <- mkEnqueue chan run
341 return $ K.append (f h) done
343{-# INLINE parConcatMapChanKGeneric #-}
344parConcatMapChanKGeneric :: MonadAsync m =>
347 -> (a -> K.StreamK m b)
350parConcatMapChanKGeneric modifier chan f stream = do
351 let cfg = modifier defaultConfig
352 case getStopWhen cfg of
353 AllStop -> parConcatMapChanK chan f stream
354 FirstStops -> parConcatMapChanKFirst chan f stream
355 AnyStops -> parConcatMapChanKAny chan f stream
357-- XXX Add a deep evaluation variant that evaluates individual elements in the
358-- generated streams in parallel.
360-- | Allocate a channel and use it to concurrently evaluate the streams
361-- generated by the mapped function.
363{-# INLINE parConcatMapK #-}
364parConcatMapK :: MonadAsync m =>
365 (Config -> Config) -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
366parConcatMapK modifier f input =
367 let g = parConcatMapChanKGeneric modifier
368 in withChannelK modifier input (`g` f)
370-- | Map each element of the input to a stream and then concurrently evaluate
371-- and concatenate the resulting streams. Multiple streams may be evaluated
372-- concurrently but earlier streams are perferred. Output from the streams are
373-- used as they arrive.
377-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
381-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
383-- The following streams finish in 4 seconds:
385-- >>> stream1 = Stream.fromEffect (delay 4)
386-- >>> stream2 = Stream.fromEffect (delay 2)
387-- >>> stream3 = Stream.fromEffect (delay 1)
388-- >>> f id [stream1, stream2, stream3]
394-- Limiting threads to 2 schedules the third stream only after one of the first
395-- two has finished, releasing a thread:
397-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
401-- When used with a Single thread it behaves like serial concatMap:
403-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
407-- >>> stream1 = Stream.fromList [1,2,3]
408-- >>> stream2 = Stream.fromList [4,5,6]
409-- >>> f (Stream.maxThreads 1) [stream1, stream2]
412-- Schedule all streams in a round robin fashion over the available threads:
414-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
416-- >>> stream1 = Stream.fromList [1,2,3]
417-- >>> stream2 = Stream.fromList [4,5,6]
418-- >>> f (Stream.maxThreads 1) [stream1, stream2]
421{-# INLINE parConcatMap #-}
422parConcatMap :: MonadAsync m =>
423 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
424parConcatMap modifier f stream =
427 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
429-- | Evaluate the streams in the input stream concurrently and combine them.
431-- >>> parConcat modifier = Stream.parConcatMap modifier id
433{-# INLINE parConcat #-}
434parConcat :: MonadAsync m =>
435 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
436parConcat modifier = parConcatMap modifier id
438-------------------------------------------------------------------------------
440-------------------------------------------------------------------------------
442-- | Like 'parConcat' but works on a list of streams.
444-- >>> parList modifier = Stream.parConcat modifier . Stream.fromList
446{-# INLINE parList #-}
447parList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
448parList modifier = parConcat modifier . Stream.fromList
450-- | Like 'concat' but works on a list of streams.
452-- >>> parListLazy = Stream.parList id
454{-# INLINE parListLazy #-}
455parListLazy :: MonadAsync m => [Stream m a] -> Stream m a
456parListLazy = parList id
458-- | Like 'parListLazy' but interleaves the streams fairly instead of prioritizing
459-- the left stream. This schedules all streams in a round robin fashion over
460-- limited number of threads.
462-- >>> parListInterleaved = Stream.parList (Stream.interleaved True)
464{-# INLINE parListInterleaved #-}
465parListInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
466parListInterleaved = parList (interleaved True)
468-- | Like 'parListLazy' but with 'ordered' on.
470-- >>> parListOrdered = Stream.parList (Stream.ordered True)
472{-# INLINE parListOrdered #-}
473parListOrdered :: MonadAsync m => [Stream m a] -> Stream m a
474parListOrdered = parList (ordered True)
476-- | Like 'parListLazy' but with 'eager' on.
478-- >>> parListEager = Stream.parList (Stream.eager True)
480{-# INLINE parListEager #-}
481parListEager :: MonadAsync m => [Stream m a] -> Stream m a
482parListEager = parList (eager True)
484-- | Like 'parListEager' but stops the output as soon as the first stream stops.
486-- >>> parListEagerFst = Stream.parList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
488{-# INLINE parListEagerFst #-}
489parListEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
490parListEagerFst = parList (eager True . stopWhen FirstStops)
492-- | Like 'parListEager' but stops the output as soon as any of the two streams
497-- >>> parListEagerMin = Stream.parList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
499{-# INLINE parListEagerMin #-}
500parListEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
501parListEagerMin = parList (eager True . stopWhen AnyStops)
503-------------------------------------------------------------------------------
505-------------------------------------------------------------------------------
507-- | Apply an argument stream to a function stream concurrently. Uses a
508-- shared channel for all individual applications within a stream application.
509{-# INLINE parApply #-}
510{-# SPECIALIZE parApply ::
511 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
512parApply :: MonadAsync m =>
513 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
514parApply modifier stream1 stream2 =
517 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
520-------------------------------------------------------------------------------
522-------------------------------------------------------------------------------
527-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
529-- For example, the following finishes in 3 seconds (as opposed to 6 seconds)
530-- because all actions run in parallel. Even though results are available out
531-- of order they are ordered due to the config option:
533-- >>> f x = delay x >> return x
534-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
540{-# INLINE parMapM #-}
541parMapM :: MonadAsync m =>
542 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
543parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
547-- >>> parSequence modifier = Stream.parMapM modifier id
551-- >>> parFromListM = Stream.parSequence id . Stream.fromList
552-- >>> parFromFoldableM = Stream.parSequence id . StreamK.toStream . StreamK.fromFoldable
554{-# INLINE parSequence #-}
555parSequence :: MonadAsync m =>
556 (Config -> Config) -> Stream m (m a) -> Stream m a
557parSequence modifier = parMapM modifier id
559-- | Evaluates the streams being zipped in separate threads than the consumer.
560-- The zip function is evaluated in the consumer thread.
562-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
564-- Multi-stream concurrency options won't apply here, see the notes in
567-- If you want to evaluate the zip function as well in a separate thread, you
568-- can use a 'parEval' on 'parZipWithM'.
570{-# INLINE parZipWithM #-}
571parZipWithM :: MonadAsync m
572 => (Config -> Config)
577parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
580-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
582-- >>> m1 = Stream.fromList [1,2,3]
583-- >>> m2 = Stream.fromList [4,5,6]
584-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
585-- [(1,4),(2,5),(3,6)]
587{-# INLINE parZipWith #-}
588parZipWith :: MonadAsync m
589 => (Config -> Config)
594parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
596-- | Like 'mergeByM' but evaluates both the streams concurrently.
600-- >>> parMergeByM cfg f m1 m2 = Stream.mergeByM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
602{-# INLINE parMergeByM #-}
603parMergeByM :: MonadAsync m
604 => (Config -> Config)
605 -> (a -> a -> m Ordering)
609parMergeByM cfg f m1 m2 = Stream.mergeByM f (parEval cfg m1) (parEval cfg m2)
611-- | Like 'mergeBy' but evaluates both the streams concurrently.
615-- >>> parMergeBy cfg f = Stream.parMergeByM cfg (\a b -> return $ f a b)
617{-# INLINE parMergeBy #-}
618parMergeBy :: MonadAsync m
619 => (Config -> Config)
620 -> (a -> a -> Ordering)
624parMergeBy cfg f = parMergeByM cfg (\a b -> return $ f a b)
626-------------------------------------------------------------------------------
628-------------------------------------------------------------------------------
630-- | Same as 'concatIterate' but concurrent.
633{-# INLINE parConcatIterate #-}
634parConcatIterate :: MonadAsync m =>
639parConcatIterate modifier f input =
641 $ withChannelK modifier (Stream.toStreamK input) iterateStream
645 iterateStream channel stream =
646 parConcatMapChanKGeneric modifier channel (generate channel) stream
649 -- XXX The channel q should be FIFO for DFS, otherwise it is BFS
650 x `K.cons` iterateStream channel (Stream.toStreamK $ f x)
652-------------------------------------------------------------------------------
654-------------------------------------------------------------------------------
659-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
661-- Generate a stream by repeatedly executing a monadic action forever.
662{-# INLINE parRepeatM #-}
663parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
664parRepeatM cfg = parSequence cfg . Stream.repeat
666-- | Generate a stream by concurrently performing a monadic action @n@ times.
670-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
672-- Example, 'parReplicateM' in the following example executes all the
673-- replicated actions concurrently, thus taking only 1 second:
675-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
678{-# INLINE parReplicateM #-}
679parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
680parReplicateM cfg n = parSequence cfg . Stream.replicate n
682-------------------------------------------------------------------------------
684-------------------------------------------------------------------------------
686-- Note: we can use another API with two callbacks stop and yield if we want
687-- the callback to be able to indicate end of stream. Or we can use a Maybe
688-- stream where Nothing indicates end of stream.
690-- XXX Rename to parNewCallback
692-- | Returns an entangled pair of a callback and a stream i.e. whenever the
693-- callback is called a value appears in the stream. The stream is infinite,
694-- there is no way for the callback to indicate that it is done now.
696-- The callback queues a value to a concurrent channel associated with the
697-- stream. The stream can be evaluated safely in any thread.
701{-# INLINE_NORMAL newCallbackStream #-}
702newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
703newCallbackStream = do
704 chan <- newChannel (eager True)
706 -- XXX Add our own thread-id to the SVar as we can not know the callback's
707 -- thread-id and the callback is not run in a managed worker. We need to
708 -- handle this better. The caller thread might be killed by the Channel if
709 -- the stream evaluator dies.
712 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
714 -- XXX We can use a "Maybe a" here. Use Nothing to send a Stop event.
719 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
720 -- XXX Use fromChannelD?
721 return (callback, fromChannel chan)
723-- XXX Rename this to parSetCallback. Also take the Channel config as argument.
724-- What config can be set by user here?
726-- XXX What happens if an exception occurs when evaluating the stream? The
727-- result of callback can be used to communicate that. But we can only know
728-- about the exception on the next callback call. For better handling the user
729-- can supply an exception sender function as argument to fromCallback.
731-- | @fromCallback f@ creates an entangled pair of a callback and a stream i.e.
732-- whenever the callback is called a value appears in the stream. The function
733-- @f@ is invoked with the callback as argument, and the stream is returned.
734-- @f@ would store the callback for calling it later for generating values in
737-- The callback queues a value to a concurrent channel associated with the
738-- stream. The stream can be evaluated safely in any thread.
742{-# INLINE fromCallback #-}
743fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
744fromCallback setCallback = Stream.concatEffect $ do
745 (callback, stream) <- newCallbackStream
749-- | @parTapCount predicate fold stream@ taps the count of those elements in
750-- the stream that pass the @predicate@. The resulting count stream is sent to
751-- a @fold@ running concurrently in another thread.
753-- For example, to print the count of elements processed every second:
755-- >>> rate = Stream.rollingMap2 (flip (-)) . Stream.delayPost 1
756-- >>> report = Stream.fold (Fold.drainMapM print) . rate
757-- >>> tap = Stream.parTapCount (const True) report
758-- >>> go = Stream.fold Fold.drain $ tap $ Stream.enumerateFrom 0
760-- Note: This may not work correctly on 32-bit machines because of Int
765{-# INLINE_NORMAL parTapCount #-}
769 -> (D.Stream m Int -> m b)
772parTapCount predicate fld (D.Stream step state) = D.Stream step' Nothing
775 {-# INLINE_LATE step' #-}
777 -- As long as we are using an "Int" for counts lockfree reads from
778 -- Var should work correctly on both 32-bit and 64-bit machines.
779 -- However, an Int on a 32-bit machine may overflow quickly.
780 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
783 $ Unboxed.toStreamD countVar
784 return $ Skip (Just (countVar, tid, state))
786 step' gst (Just (countVar, tid, st)) = do
791 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
792 return $ Yield x (Just (countVar, tid, s))
793 Skip s -> return $ Skip (Just (countVar, tid, s))
795 liftIO $ killThread tid
798-- Deprecate in a major release.
799-- {-# DEPRECATED tapCount "Please use parTapCount instead." #-}
800-- | Same as 'parTapCount'. Deprecated.
801{-# INLINE tapCount #-}
805 -> (Stream m Int -> m b)
808tapCount = parTapCount
810-------------------------------------------------------------------------------
812-------------------------------------------------------------------------------
814-- Clone a stream into n streams, perform some processing on them and then zip
815-- or merge the results in different ways?
817-- For serial processing combining n scans into a single scan on the source
818-- stream would be the most efficient way of doing this. But this has a
819-- limitation that we process one element at a time through the combined scan.
820-- This is a way to combine the states of different scans in a modular way.
821-- This works well for cases where each scan consumes and produces one element
822-- at a time. If different scans produce elements by consuming different number
823-- of elements then this may become complicated, inconvenient to use.
825-- It does not make much sense to clone a stream to multiple free streams
826-- unless we enforce processing those streams in independent threads. If
827-- we are anyway running them in the same thread then there is not much point
828-- of cloning, we can just map a function on the stream to do multiple tasks in
831-- Cloning a stream to multiple free streams can provide independent buffering
832-- and speed of evaluation to each cloned stream pipeline. For example, we can
833-- parseBreak each stream independently using a different parser. The
834-- evaluation would be push driven. The source stream would be evaluated in a
835-- separate thread and we would push the generated elements to all the cloned
838-- 1. If the cloned streams have infinite buffers then this can lead to the
839-- source stream getting evaluated faster than consumers and buffering the
840-- entire source stream in cloned streams.
842-- 2. If the cloned streams have limited buffers, then they will all go at the
843-- speed of the slowest stream if they are run concurrently.
845-- 3. If the cloned streams have limited buffers and are evaluated serially
846-- then we may run into deadlock if we are deep evaluating one stream and the
847-- source gets blocked because other stream's buffer are full.
849-- This is somewhat like list sharing. And it will have the same space leak
850-- issues if used incorrectly. In fact, we can evaluate the source stream to
851-- generate a lazy list using unsafePerformIO and share that lazy list among
852-- multiple consumers. The evaluation of the list would drive the stream. And
853-- the list would be naturally shared across consumers which can use different
854-- buffering. This would be more like the lazy IO model. However, it may be
855-- better to use streams instead of lists because streams use a monad and lists
856-- are pure - pure lists can lead to the same issues as lazy IO when used in
859-- Therefore, for safety, it makes better sense to use consumers (Stream m a ->
860-- m b) rather than generating free streams as results. Each such consumer can
861-- be enforced to run in its own thread. We can also pass a result collector
862-- callback in a ReaderT env to collect the results from all these consumers
863-- into a single stream.
865-- parTap -- tap the stream to a single consumer
866-- parDistribute -- a finite list of consumers is specified. all consumers
867-- are guaranteed to get the entire stream from beginning. Run each consumer in
869-- parDistributeStream -- consumers can join the distribution channel
870-- dynamically, they will get the source stream from now onwards.
872-- XXX We could use Stream or StreamK, what are the pros and cons? The StreamK
873-- version can be used to implement parDistribute using foldr?
876parTap :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> Stream m a
877parTap f m = undefined
879-- Can we just use a parEval fold in tap?
880-- We can easily convert the Fold to "Stream m a -> m b" form. Check if this
881-- provides the same perf as above.
883parTap :: MonadAsync m => Fold m a b -> Stream m a -> Stream m a
884parTap f xs = undefined
886-- Can we just use a parallel distribute fold in tap?
887-- Maybe better to use a custom impl of distribute?
888{-# INLINE parDistribute #-}
889parDistribute :: (Foldable f, , MonadAsync m)
890 => f (Stream m a -> m b) -> Stream m a -> Stream m a
891parDistribute = 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