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.
52 -- Add unfoldrM/iterateM?
55 -- | Uses a single channel to evaluate all actions.
62 -- | Uses a single channel to evaluate all actions.
69 -- | Use a channel for each pair.
70 -- combine/concur/conjoin
80 -- | Shares a single channel across many streams.
92 -- ** Stream of streams
98 -- | Shares a single channel across many streams.
102 , concatMapInterleave
117import Control.Concurrent (myThreadId, killThread)
118import Control.Monad (void, when)
119import Control.Monad.IO.Class (MonadIO(liftIO))
120import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
121import Streamly.Internal.Control.ForkLifted (forkManaged)
122import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)
123import Streamly.Internal.Data.Stream.Channel.Types
127import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
128import Streamly.Internal.Data.Stream.Type (Stream)
129import Streamly.Internal.Data.Stream.StreamD (Step(..))
131import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
132import qualified Streamly.Internal.Data.Stream as Stream
133import qualified Streamly.Internal.Data.Stream.StreamD as D
134import qualified Streamly.Internal.Data.Stream.StreamK as K
135import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
137import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
138import Streamly.Internal.Data.Stream.Concurrent.Channel
142-- Imports for example snippets in this module.
145-- >>> {-# LANGUAGE FlexibleContexts #-}
146-- >>> import Control.Concurrent (threadDelay)
147-- >>> import qualified Streamly.Data.Array as Array
148-- >>> import qualified Streamly.Data.Fold as Fold
149-- >>> import qualified Streamly.Data.Parser as Parser
150-- >>> import qualified Streamly.Internal.Data.Stream as Stream hiding (append2)
151-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream
152-- >>> import Prelude hiding (concatMap, concat, zipWith)
155-- threadDelay (n * 1000000) -- sleep for n seconds
156-- putStrLn (show n ++ " sec") -- print "n sec"
160-------------------------------------------------------------------------------
161-- Evaluating a stream
162-------------------------------------------------------------------------------
165{-# INLINE_NORMAL parEvalD #-}
166parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
167parEvalD modifier m = D.Stream step Nothing
171 chan <- newChannel modifier
172 sendFirstWorker chan (D.toStreamK m)
173 -- XXX should use an unfold to make this efficient
174 return $ D.Skip $ Just $ fromChannelD chan
176 step gst (Just (D.UnStream step1 st)) = do
179 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
180 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
184-- | Evaluate a stream asynchronously. In a serial stream, each element of the
185-- stream is generated as it is demanded by the consumer. `parEval` evaluates
186-- multiple elements of the stream ahead of time and serves the results from a
189-- Note that the evaluation requires only one thread as only one stream needs
190-- to be evaluated. Therefore, the concurrency options that are relevant to
191-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,
192-- ordered, stopWhen options won't have any effect.
194{-# INLINE parEval #-}
195parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
196parEval modifier input = withChannel modifier input (const id)
197 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
201-- >>> eval = Stream.parEval id
204eval :: MonadAsync m => Stream m a -> Stream m a
207-------------------------------------------------------------------------------
208-- combining two streams
209-------------------------------------------------------------------------------
211{-# INLINE _appendGeneric #-}
212_appendGeneric :: MonadAsync m =>
213 ((Config -> Config) -> m (Channel m a))
214 -> (Config -> Config)
218_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
223 chan <- newChan modifier
224 let cfg = modifier defaultConfig
225 done = K.nilM (stopChannel chan)
226 case getStopWhen cfg of
228 toChannelK chan stream2
229 toChannelK chan stream1
231 toChannelK chan stream2
232 toChannelK chan (K.serial stream1 done)
234 toChannelK chan (K.serial stream2 done)
235 toChannelK chan (K.serial stream1 done)
236 return $ Stream.toStreamK $ fromChannel chan
238-- | Create a new channel and add both the streams to it for async evaluation.
239-- The output stream is the result of the evaluation.
240{-# INLINE appendWithK #-}
241appendWithK :: MonadAsync m =>
242 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
243appendWithK modifier stream1 stream2 =
245 if getOrdered (modifier defaultConfig)
246 then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
247 else _appendGeneric Append.newChannel modifier stream1 stream2
249 parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
251-- | Evaluate the first stream in the current thread and add the second stream
252-- to the supplied channel. This is to be used by a worker thread.
254-- This can be used with parConcatMap:
257-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
260{-# INLINE _appendWithChanK #-}
261_appendWithChanK :: MonadAsync m =>
262 Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
263_appendWithChanK chan stream1 stream2 =
264 K.before (toChannelK chan stream2) stream1
267-- | Like 'append' but with a Config modifier.
268{-# INLINE combineWith #-}
269combineWith :: MonadAsync m =>
270 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
271combineWith modifier stream1 stream2 =
274 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
276-- | Binary operation to evaluate two streams concurrently prioritizing the
279-- If you want to combine more than two streams you almost always want the
280-- 'concat' operation instead. The performance of this operation degrades
281-- rapidly when more streams are combined as each operation adds one more
282-- concurrent channel. On the other hand, 'concat' uses a single channel for
283-- all streams. However, with this operation you can precisely control the
284-- scheduling by creating arbitrary shape expression trees.
286-- >>> append2 = Stream.combineWith id
288-- The following code finishes in 4 seconds:
290-- >>> stream1 = Stream.fromEffect (delay 4)
291-- >>> stream2 = Stream.fromEffect (delay 2)
292-- >>> Stream.fold Fold.toList $ stream1 `Stream.append2` stream2
297{-# INLINE append2 #-}
298append2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
299append2 = combineWith id
301-- | Like 'append' but interleaves the streams fairly instead of prioritizing
302-- the left stream. This schedules all streams in a round robin fashion over
303-- limited number of threads.
305-- >>> interleave2 = Stream.combineWith (Stream.interleaved True)
307{-# INLINE interleave2 #-}
308interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
309interleave2 = combineWith (interleaved True)
311-- | Like 'append' but with 'ordered' on.
313-- >>> ahead2 = Stream.combineWith (Stream.ordered True)
316ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
317ahead2 = combineWith (ordered True)
319-- | Like 'append2' but with 'eager' on.
321-- >>> parallel2 = Stream.combineWith (Stream.eager True)
323{-# INLINE parallel2 #-}
324parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
325parallel2 = combineWith (eager True)
327-- | Like 'parallel2' but stops the output as soon as the first stream stops.
329-- >>> parallelFst2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.FirstStops)
332{-# INLINE parallelFst2 #-}
333parallelFst2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
334parallelFst2 = combineWith (eager True . stopWhen FirstStops)
336-- | Like 'parallel2' but stops the output as soon as any of the two streams
339-- >>> parallelMin2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.AnyStops)
342{-# INLINE parallelMin2 #-}
343parallelMin2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
344parallelMin2 = combineWith (eager True . stopWhen AnyStops)
346-------------------------------------------------------------------------------
348-------------------------------------------------------------------------------
350-- | A runner function takes a queuing function @q@ and a stream, it splits the
351-- input stream, queuing the tail and using the head to generate a stream.
352-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
353-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
355{-# INLINE mkEnqueue #-}
356mkEnqueue :: MonadAsync m =>
358 -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
359 -> m (K.Stream m a -> m ())
360mkEnqueue chan runner = do
361 runInIO <- askRunInIO
364 -- Enqueue the outer loop
365 liftIO $ enqueue chan False (runInIO, runner q stream)
366 -- XXX In case of eager dispatch we can just directly dispatch
367 -- a worker with the tail stream here rather than first queuing
368 -- and then dispatching a worker which dequeues the work. The
369 -- older implementation did a direct dispatch here and its perf
370 -- characterstics looked much better.
374-- | Takes the head element of the input stream and queues the tail of the
375-- stream to the channel, then maps the supplied function on the head and
376-- evaluates the resulting stream.
378-- This function is designed to be used by worker threads on a channel to
379-- concurrently map and evaluate a stream.
380{-# INLINE parConcatMapChanK #-}
381parConcatMapChanK :: MonadAsync m =>
382 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
383parConcatMapChanK chan f stream =
384 let run q = concatMapDivK q f
385 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
386 -- K.parConcatMap (_appendWithChanK chan) f stream
388{-# INLINE parConcatMapChanKAny #-}
389parConcatMapChanKAny :: MonadAsync m =>
390 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
391parConcatMapChanKAny chan f stream =
392 let done = K.nilM (stopChannel chan)
393 run q = concatMapDivK q (\x -> K.serial (f x) done)
394 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
396{-# INLINE parConcatMapChanKFirst #-}
397parConcatMapChanKFirst :: MonadAsync m =>
398 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
399parConcatMapChanKFirst chan f stream =
400 let done = K.nilM (stopChannel chan)
401 run q = concatMapDivK q f
402 in K.concatEffect $ do
403 res <- K.uncons stream
405 Nothing -> return K.nil
407 q <- mkEnqueue chan run
409 return $ K.serial (f h) done
411{-# INLINE parConcatMapChanKGeneric #-}
412parConcatMapChanKGeneric :: MonadAsync m =>
415 -> (a -> K.Stream m b)
418parConcatMapChanKGeneric modifier chan f stream = do
419 let cfg = modifier defaultConfig
420 case getStopWhen cfg of
421 AllStop -> parConcatMapChanK chan f stream
422 FirstStops -> parConcatMapChanKFirst chan f stream
423 AnyStops -> parConcatMapChanKAny chan f stream
425-- XXX Add a deep evaluation variant that evaluates individual elements in the
426-- generated streams in parallel.
428-- | Allocate a channel and use it to concurrently evaluate the streams
429-- generated by the mapped function.
431{-# INLINE parConcatMapK #-}
432parConcatMapK :: MonadAsync m =>
433 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
434parConcatMapK modifier f input =
435 let g = parConcatMapChanKGeneric modifier
436 in withChannelK modifier input (`g` f)
438-- | Map each element of the input to a stream and then concurrently evaluate
439-- and concatenate the resulting streams. Multiple streams may be evaluated
440-- concurrently but earlier streams are perferred. Output from the streams are
441-- used as they arrive.
445-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
449-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
451-- The following streams finish in 4 seconds:
453-- >>> stream1 = Stream.fromEffect (delay 4)
454-- >>> stream2 = Stream.fromEffect (delay 2)
455-- >>> stream3 = Stream.fromEffect (delay 1)
456-- >>> f id [stream1, stream2, stream3]
462-- Limiting threads to 2 schedules the third stream only after one of the first
463-- two has finished, releasing a thread:
465-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
469-- When used with a Single thread it behaves like serial concatMap:
471-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
475-- >>> stream1 = Stream.fromList [1,2,3]
476-- >>> stream2 = Stream.fromList [4,5,6]
477-- >>> f (Stream.maxThreads 1) [stream1, stream2]
480{-# INLINE parConcatMap #-}
481parConcatMap :: MonadAsync m =>
482 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
483parConcatMap modifier f stream =
486 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
490-- >>> concatMap = Stream.parConcatMap id
492{-# INLINE concatMap #-}
493concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
494concatMap = parConcatMap id
496-- | Map each element of the input to a stream and then concurrently evaluate
497-- and interleave the resulting streams. Unlike 'concatMap' which prefers to
498-- evaluate the earlier stream first, this schedules all streams in a round
499-- robin fashion over the available threads.
501-- >>> concatMapInterleave = Stream.parConcatMap (Stream.interleaved True)
503-- When used with a single thread it behaves like serial interleaving:
505-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
507-- >>> stream1 = Stream.fromList [1,2,3]
508-- >>> stream2 = Stream.fromList [4,5,6]
509-- >>> f (Stream.maxThreads 1) [stream1, stream2]
512-- /Works only on finite number of streams/
514{-# INLINE concatMapInterleave #-}
515concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
516concatMapInterleave = parConcatMap (interleaved True)
518-- | Evaluate the streams in the input stream concurrently and combine them.
520-- >>> parConcat modifier = Stream.parConcatMap modifier id
522{-# INLINE parConcat #-}
523parConcat :: MonadAsync m =>
524 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
525parConcat modifier = parConcatMap modifier id
529-- >>> concat = Stream.parConcat id
532concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a
535-------------------------------------------------------------------------------
537-------------------------------------------------------------------------------
539-- | Like 'parConcat' but works on a list of streams.
541-- >>> parConcatList modifier = Stream.parConcat modifier . Stream.fromList
543{-# INLINE parConcatList #-}
544parConcatList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
545parConcatList modifier = parConcat modifier . Stream.fromList
547-- | Like 'concat' but works on a list of streams.
549-- >>> parLazy = Stream.parConcatList id
551{-# INLINE parLazy #-}
552parLazy :: MonadAsync m => [Stream m a] -> Stream m a
553parLazy = parConcatList id
555-- | Like 'parLazy' but interleaves the streams fairly instead of prioritizing
556-- the left stream. This schedules all streams in a round robin fashion over
557-- limited number of threads.
559-- >>> parLazyInterleaved = Stream.parConcatList (Stream.interleaved True)
561{-# INLINE parLazyInterleaved #-}
562parLazyInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
563parLazyInterleaved = parConcatList (interleaved True)
565-- | Like 'parLazy' but with 'ordered' on.
567-- >>> parLazyOrdered = Stream.parConcatList (Stream.ordered True)
569{-# INLINE parLazyOrdered #-}
570parLazyOrdered :: MonadAsync m => [Stream m a] -> Stream m a
571parLazyOrdered = parConcatList (ordered True)
573-- | Like 'parLazy' but with 'eager' on.
575-- >>> parEager = Stream.parConcatList (Stream.eager True)
577{-# INLINE parEager #-}
578parEager :: MonadAsync m => [Stream m a] -> Stream m a
579parEager = parConcatList (eager True)
581-- | Like 'parEager' but stops the output as soon as the first stream stops.
583-- >>> parEagerFst = Stream.parConcatList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
585{-# INLINE parEagerFst #-}
586parEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
587parEagerFst = parConcatList (eager True . stopWhen FirstStops)
589-- | Like 'parEager' but stops the output as soon as any of the two streams
594-- >>> parEagerMin = Stream.parConcatList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
596{-# INLINE parEagerMin #-}
597parEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
598parEagerMin = parConcatList (eager True . stopWhen AnyStops)
600-------------------------------------------------------------------------------
602-------------------------------------------------------------------------------
604-- | Apply an argument stream to a function stream concurrently. Uses a
605-- shared channel for all individual applications within a stream application.
606{-# INLINE parApply #-}
607{-# SPECIALIZE parApply ::
608 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
609parApply :: MonadAsync m =>
610 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
611parApply modifier stream1 stream2 =
614 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
618{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
619apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b
622-------------------------------------------------------------------------------
624-------------------------------------------------------------------------------
629-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
631-- Example, the following example finishes in 1 second as all actions run in
632-- parallel. Even though results are available out of order they are ordered
633-- due to the config option::
635-- >>> f x = delay x >> return x
636-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
642{-# INLINE parMapM #-}
643parMapM :: MonadAsync m =>
644 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
645parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
650-- >>> mapM = Stream.parMapM id
653mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
657-- >>> parSequence modifier = Stream.parMapM modifier id
659{-# INLINE parSequence #-}
660parSequence :: MonadAsync m =>
661 (Config -> Config) -> Stream m (m a) -> Stream m a
662parSequence modifier = parMapM modifier id
665-- >>> sequence = Stream.parSequence id
667{-# INLINE sequence #-}
668sequence :: MonadAsync m =>
669 Stream m (m a) -> Stream m a
670sequence = parSequence id
672-- | Evaluates the streams being zipped in separate threads than the consumer.
673-- The zip function is evaluated in the consumer thread.
675-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
677-- Multi-stream concurrency options won't apply here, see the notes in
680-- If you want to evaluate the zip function as well in a separate thread, you
681-- can use a 'parEval' on 'parZipWithM'.
683{-# INLINE parZipWithM #-}
684parZipWithM :: MonadAsync m
685 => (Config -> Config)
690parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
693-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
695-- >>> m1 = Stream.fromList [1,2,3]
696-- >>> m2 = Stream.fromList [4,5,6]
697-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
698-- [(1,4),(2,5),(3,6)]
700{-# INLINE parZipWith #-}
701parZipWith :: MonadAsync m
702 => (Config -> Config)
707parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
709-------------------------------------------------------------------------------
711-------------------------------------------------------------------------------
713-- | Same as 'concatIterate' but concurrent.
716{-# INLINE parConcatIterate #-}
717parConcatIterate :: MonadAsync m =>
722parConcatIterate modifier f input =
724 $ withChannelK modifier (Stream.toStreamK input) iterateStream
728 iterateStream channel stream =
729 parConcatMapChanKGeneric modifier channel (generate channel) stream
732 -- XXX The channel q should be FIFO for DFS, otherwise it is BFS
733 x `K.cons` iterateStream channel (Stream.toStreamK $ f x)
735-------------------------------------------------------------------------------
737-------------------------------------------------------------------------------
742-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
744-- Generate a stream by repeatedly executing a monadic action forever.
745{-# INLINE parRepeatM #-}
746parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
747parRepeatM cfg = parSequence cfg . Stream.repeat
752-- >>> repeatM = Stream.parRepeatM id
754{-# INLINE repeatM #-}
755repeatM :: MonadAsync m => m a -> Stream m a
756repeatM = parRepeatM id
758-- | Generate a stream by concurrently performing a monadic action @n@ times.
762-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
764-- Example, 'parReplicateM' in the following example executes all the
765-- replicated actions concurrently, thus taking only 1 second:
767-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
770{-# INLINE parReplicateM #-}
771parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
772parReplicateM cfg n = parSequence cfg . Stream.replicate n
777-- >>> replicateM n = Stream.parReplicateM id
779{-# INLINE replicateM #-}
780replicateM :: MonadAsync m => Int -> m a -> Stream m a
781replicateM = parReplicateM id
783-------------------------------------------------------------------------------
785-------------------------------------------------------------------------------
787-- Note: we can use another API with two callbacks stop and yield if we want
788-- the callback to be able to indicate end of stream.
790-- | Generates a callback and a stream pair. The callback returned is used to
791-- queue values to the stream. The stream is infinite, there is no way for the
792-- callback to indicate that it is done now.
796{-# INLINE_NORMAL newCallbackStream #-}
797newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
798newCallbackStream = do
799 chan <- newChannel (eager True)
801 -- XXX Add our own thread-id to the SVar as we can not know the callback's
802 -- thread-id and the callback is not run in a managed worker. We need to
803 -- handle this better.
805 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
811 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
812 -- XXX Use fromChannelD?
813 return (callback, fromChannel chan)
815-- | Supplies a stream generating callback to a callback setter function. Each
816-- invocation of the callback results in a value being generated in the
821{-# INLINE fromCallback #-}
822fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
823fromCallback setCallback = Stream.concatEffect $ do
824 (callback, stream) <- newCallbackStream
828{-# INLINE_NORMAL tapCountD #-}
832 -> (D.Stream m Int -> m b)
835tapCountD predicate fld (D.Stream step state) = D.Stream step' Nothing
838 {-# INLINE_LATE step' #-}
840 -- As long as we are using an "Int" for counts lockfree reads from
841 -- Var should work correctly on both 32-bit and 64-bit machines.
842 -- However, an Int on a 32-bit machine may overflow quickly.
843 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
846 $ Unboxed.toStreamD countVar
847 return $ Skip (Just (countVar, tid, state))
849 step' gst (Just (countVar, tid, st)) = do
854 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
855 return $ Yield x (Just (countVar, tid, s))
856 Skip s -> return $ Skip (Just (countVar, tid, s))
858 liftIO $ killThread tid
861-- | @tapCount predicate fold stream@ taps the count of those elements in the
862-- stream that pass the @predicate@. The resulting count stream is sent to
863-- another thread which folds it using @fold@.
865-- For example, to print the count of elements processed every second:
867-- >>> rate = Stream.rollingMap2 (flip (-)) . Stream.delayPost 1
868-- >>> report = Stream.fold (Fold.drainMapM print) . rate
869-- >>> tap = Stream.tapCount (const True) report
870-- >>> go = Stream.fold Fold.drain $ tap $ Stream.enumerateFrom 0
872-- Note: This may not work correctly on 32-bit machines because of Int
877{-# INLINE tapCount #-}
881 -> (Stream m Int -> m b)
884tapCount predicate f xs =
886 $ tapCountD predicate (f . Stream.fromStreamD)
887 $ Stream.toStreamD xs
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