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
114import Control.Concurrent (myThreadId, killThread)
115import Control.Monad (void, when)
116import Control.Monad.IO.Class (MonadIO(liftIO))
117import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
118import Streamly.Internal.Control.ForkLifted (forkManaged)
119import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)
120import Streamly.Internal.Data.Stream.Channel.Types
124import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
125import Streamly.Internal.Data.Stream.Type (Stream)
126import Streamly.Internal.Data.Stream.StreamD (Step(..))
128import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
129import qualified Streamly.Internal.Data.Stream as Stream
130import qualified Streamly.Internal.Data.Stream.StreamD as D
131import qualified Streamly.Internal.Data.Stream.StreamK as K
132import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
134import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
135import Streamly.Internal.Data.Stream.Concurrent.Channel
139-- Imports for example snippets in this module.
142-- >>> {-# LANGUAGE FlexibleContexts #-}
143-- >>> import Control.Concurrent (threadDelay)
144-- >>> import qualified Streamly.Data.Array as Array
145-- >>> import qualified Streamly.Data.Fold as Fold
146-- >>> import qualified Streamly.Data.Parser as Parser
147-- >>> import qualified Streamly.Data.Stream as Stream
148-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream
149-- >>> import Prelude hiding (concatMap, concat, zipWith)
152-- threadDelay (n * 1000000) -- sleep for n seconds
153-- putStrLn (show n ++ " sec") -- print "n sec"
157-------------------------------------------------------------------------------
158-- Evaluating a stream
159-------------------------------------------------------------------------------
162{-# INLINE_NORMAL parEvalD #-}
163parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
164parEvalD modifier m = D.Stream step Nothing
168 chan <- newChannel modifier
169 sendFirstWorker chan (D.toStreamK m)
170 -- XXX should use an unfold to make this efficient
171 return $ D.Skip $ Just $ fromChannelD chan
173 step gst (Just (D.UnStream step1 st)) = do
176 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
177 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
181-- | Evaluate a stream asynchronously using a channel and serve the consumer
182-- from the evaluation buffer.
184-- Note that the evaluation requires a single thread as only one stream needs
185-- to be evaluated. Therefore, the concurrency options that are relevant to
186-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,
187-- ordered, stopWhen options won't have any effect.
189{-# INLINE parEval #-}
190parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
191parEval modifier input = withChannel modifier input (const id)
192 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
196-- >>> eval = Stream.parEval id
199eval :: MonadAsync m => Stream m a -> Stream m a
202-------------------------------------------------------------------------------
203-- combining two streams
204-------------------------------------------------------------------------------
206{-# INLINE _appendGeneric #-}
207_appendGeneric :: MonadAsync m =>
208 ((Config -> Config) -> m (Channel m a))
209 -> (Config -> Config)
213_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
218 chan <- newChan modifier
219 let cfg = modifier defaultConfig
220 done = K.nilM (stopChannel chan)
221 case getStopWhen cfg of
223 toChannelK chan stream2
224 toChannelK chan stream1
226 toChannelK chan stream2
227 toChannelK chan (K.serial stream1 done)
229 toChannelK chan (K.serial stream2 done)
230 toChannelK chan (K.serial stream1 done)
231 return $ Stream.toStreamK $ fromChannel chan
233-- | Create a new channel and add both the streams to it for async evaluation.
234-- The output stream is the result of the evaluation.
235{-# INLINE appendWithK #-}
236appendWithK :: MonadAsync m =>
237 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
238appendWithK modifier stream1 stream2 =
240 if getOrdered (modifier defaultConfig)
241 then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
242 else _appendGeneric Append.newChannel modifier stream1 stream2
244 parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
246-- | Evaluate the first stream in the current thread and add the second stream
247-- to the supplied channel. This is to be used by a worker thread.
249-- This can be used with parConcatMap:
252-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
255{-# INLINE _appendWithChanK #-}
256_appendWithChanK :: MonadAsync m =>
257 Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
258_appendWithChanK chan stream1 stream2 =
259 K.before (toChannelK chan stream2) stream1
262-- | Like 'append' but with a Config modifier.
263{-# INLINE combineWith #-}
264combineWith :: MonadAsync m =>
265 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
266combineWith modifier stream1 stream2 =
269 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
271-- | Binary operation to evaluate two streams concurrently prioritizing the
274-- If you want to combine more than two streams you almost always want the
275-- 'concat' operation instead. The performance of this operation degrades
276-- rapidly when more streams are combined as each operation adds one more
277-- concurrent channel. On the other hand, 'concat' uses a single channel for
278-- all streams. However, with this operation you can precisely control the
279-- scheduling by creating arbitrary shape expression trees.
281-- >>> append2 = Stream.combineWith id
283-- The following code finishes in 4 seconds:
285-- >>> stream1 = Stream.fromEffect (delay 4)
286-- >>> stream2 = Stream.fromEffect (delay 2)
287-- >>> Stream.fold Fold.toList $ stream1 `Stream.append2` stream2
292{-# INLINE append2 #-}
293append2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
294append2 = combineWith id
296-- | Like 'append' but interleaves the streams fairly instead of prioritizing
297-- the left stream. This schedules all streams in a round robin fashion over
298-- limited number of threads.
300-- >>> interleave2 = Stream.combineWith (Stream.interleaved True)
302{-# INLINE interleave2 #-}
303interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
304interleave2 = combineWith (interleaved True)
306-- | Like 'append' but with 'ordered' on.
308-- >>> ahead2 = Stream.combineWith (Stream.ordered True)
311ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
312ahead2 = combineWith (ordered True)
314-- | Like 'append2' but with 'eager' on.
316-- >>> parallel2 = Stream.combineWith (Stream.eager True)
318{-# INLINE parallel2 #-}
319parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
320parallel2 = combineWith (eager True)
322-- | Like 'parallel2' but stops the output as soon as the first stream stops.
324-- >>> parallelFst2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.FirstStops)
327{-# INLINE parallelFst2 #-}
328parallelFst2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
329parallelFst2 = combineWith (eager True . stopWhen FirstStops)
331-- | Like 'parallel2' but stops the output as soon as any of the two streams
334-- >>> parallelMin2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.AnyStops)
337{-# INLINE parallelMin2 #-}
338parallelMin2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
339parallelMin2 = combineWith (eager True . stopWhen AnyStops)
341-------------------------------------------------------------------------------
343-------------------------------------------------------------------------------
345-- | A runner function takes a queuing function @q@ and a stream, it splits the
346-- input stream, queuing the tail and using the head to generate a stream.
347-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
348-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
350{-# INLINE mkEnqueue #-}
351mkEnqueue :: MonadAsync m =>
353 -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
354 -> m (K.Stream m a -> m ())
355mkEnqueue chan runner = do
356 runInIO <- askRunInIO
359 -- Enqueue the outer loop
360 liftIO $ enqueue chan False (runInIO, runner q stream)
361 -- XXX In case of eager dispatch we can just directly dispatch
362 -- a worker with the tail stream here rather than first queuing
363 -- and then dispatching a worker which dequeues the work. The
364 -- older implementation did a direct dispatch here and its perf
365 -- characterstics looked much better.
369-- | Takes the head element of the input stream and queues the tail of the
370-- stream to the channel, then maps the supplied function on the head and
371-- evaluates the resulting stream.
373-- This function is designed to be used by worker threads on a channel to
374-- concurrently map and evaluate a stream.
375{-# INLINE parConcatMapChanK #-}
376parConcatMapChanK :: MonadAsync m =>
377 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
378parConcatMapChanK chan f stream =
379 let run q = concatMapDivK q f
380 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
381 -- K.parConcatMap (_appendWithChanK chan) f stream
383{-# INLINE parConcatMapChanKAny #-}
384parConcatMapChanKAny :: MonadAsync m =>
385 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
386parConcatMapChanKAny chan f stream =
387 let done = K.nilM (stopChannel chan)
388 run q = concatMapDivK q (\x -> K.serial (f x) done)
389 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
391{-# INLINE parConcatMapChanKFirst #-}
392parConcatMapChanKFirst :: MonadAsync m =>
393 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
394parConcatMapChanKFirst chan f stream =
395 let done = K.nilM (stopChannel chan)
396 run q = concatMapDivK q f
397 in K.concatEffect $ do
398 res <- K.uncons stream
400 Nothing -> return K.nil
402 q <- mkEnqueue chan run
404 return $ K.serial (f h) done
406{-# INLINE parConcatMapChanKGeneric #-}
407parConcatMapChanKGeneric :: MonadAsync m =>
410 -> (a -> K.Stream m b)
413parConcatMapChanKGeneric modifier chan f stream = do
414 let cfg = modifier defaultConfig
415 case getStopWhen cfg of
416 AllStop -> parConcatMapChanK chan f stream
417 FirstStops -> parConcatMapChanKFirst chan f stream
418 AnyStops -> parConcatMapChanKAny chan f stream
420-- XXX Add a deep evaluation variant that evaluates individual elements in the
421-- generated streams in parallel.
423-- | Allocate a channel and use it to concurrently evaluate the streams
424-- generated by the mapped function.
426{-# INLINE parConcatMapK #-}
427parConcatMapK :: MonadAsync m =>
428 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
429parConcatMapK modifier f input =
430 let g = parConcatMapChanKGeneric modifier
431 in withChannelK modifier input (`g` f)
433-- | Map each element of the input to a stream and then concurrently evaluate
434-- and concatenate the resulting streams. Multiple streams may be evaluated
435-- concurrently but earlier streams are perferred. Output from the streams are
436-- used as they arrive.
440-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
444-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
446-- The following streams finish in 4 seconds:
448-- >>> stream1 = Stream.fromEffect (delay 4)
449-- >>> stream2 = Stream.fromEffect (delay 2)
450-- >>> stream3 = Stream.fromEffect (delay 1)
451-- >>> f id [stream1, stream2, stream3]
457-- Limiting threads to 2 schedules the third stream only after one of the first
458-- two has finished, releasing a thread:
460-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
464-- When used with a Single thread it behaves like serial concatMap:
466-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
470-- >>> stream1 = Stream.fromList [1,2,3]
471-- >>> stream2 = Stream.fromList [4,5,6]
472-- >>> f (Stream.maxThreads 1) [stream1, stream2]
475{-# INLINE parConcatMap #-}
476parConcatMap :: MonadAsync m =>
477 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
478parConcatMap modifier f stream =
481 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
485-- >>> concatMap = Stream.parConcatMap id
487{-# INLINE concatMap #-}
488concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
489concatMap = parConcatMap id
491-- | Map each element of the input to a stream and then concurrently evaluate
492-- and interleave the resulting streams. Unlike 'concatMap' which prefers to
493-- evaluate the earlier stream first, this schedules all streams in a round
494-- robin fashion over the available threads.
496-- >>> concatMapInterleave = Stream.parConcatMap (Stream.interleaved True)
498-- When used with a single thread it behaves like serial interleaving:
500-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
502-- >>> stream1 = Stream.fromList [1,2,3]
503-- >>> stream2 = Stream.fromList [4,5,6]
504-- >>> f (Stream.maxThreads 1) [stream1, stream2]
507-- /Works only on finite number of streams/
509{-# INLINE concatMapInterleave #-}
510concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
511concatMapInterleave = parConcatMap (interleaved True)
513-- | Evaluate the streams in the input stream concurrently and combine them.
515-- >>> parConcat modifier = Stream.parConcatMap modifier id
517{-# INLINE parConcat #-}
518parConcat :: MonadAsync m =>
519 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
520parConcat modifier = parConcatMap modifier id
524-- >>> concat = Stream.parConcat id
527concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a
530-------------------------------------------------------------------------------
532-------------------------------------------------------------------------------
534-- | Like 'parConcat' but works on a list of streams.
536-- >>> parConcatList modifier = Stream.parConcat modifier . Stream.fromList
538{-# INLINE parConcatList #-}
539parConcatList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
540parConcatList modifier = parConcat modifier . Stream.fromList
542-- | Like 'concat' but works on a list of streams.
544-- >>> parLazy = Stream.parConcatList id
546{-# INLINE parLazy #-}
547parLazy :: MonadAsync m => [Stream m a] -> Stream m a
548parLazy = parConcatList id
550-- | Like 'parLazy' but interleaves the streams fairly instead of prioritizing
551-- the left stream. This schedules all streams in a round robin fashion over
552-- limited number of threads.
554-- >>> parLazyInterleaved = Stream.parConcatList (Stream.interleaved True)
556{-# INLINE parLazyInterleaved #-}
557parLazyInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
558parLazyInterleaved = parConcatList (interleaved True)
560-- | Like 'parLazy' but with 'ordered' on.
562-- >>> parLazyOrdered = Stream.parConcatList (Stream.ordered True)
564{-# INLINE parLazyOrdered #-}
565parLazyOrdered :: MonadAsync m => [Stream m a] -> Stream m a
566parLazyOrdered = parConcatList (ordered True)
568-- | Like 'parLazy' but with 'eager' on.
570-- >>> parEager = Stream.parConcatList (Stream.eager True)
572{-# INLINE parEager #-}
573parEager :: MonadAsync m => [Stream m a] -> Stream m a
574parEager = parConcatList (eager True)
576-- | Like 'parEager' but stops the output as soon as the first stream stops.
578-- >>> parEagerFst = Stream.parConcatList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
580{-# INLINE parEagerFst #-}
581parEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
582parEagerFst = parConcatList (eager True . stopWhen FirstStops)
584-- | Like 'parEager' but stops the output as soon as any of the two streams
589-- >>> parEagerMin = Stream.parConcatList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
591{-# INLINE parEagerMin #-}
592parEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
593parEagerMin = parConcatList (eager True . stopWhen AnyStops)
595-------------------------------------------------------------------------------
597-------------------------------------------------------------------------------
599-- | Apply an argument stream to a function stream concurrently. Uses a
600-- shared channel for all individual applications within a stream application.
601{-# INLINE parApply #-}
602{-# SPECIALIZE parApply ::
603 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
604parApply :: MonadAsync m =>
605 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
606parApply modifier stream1 stream2 =
609 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
613{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
614apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b
617-------------------------------------------------------------------------------
619-------------------------------------------------------------------------------
624-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
626-- Example, the following example finishes in 1 second as all actions run in
627-- parallel. Even though results are available out of order they are ordered
628-- due to the config option::
630-- >>> f x = delay x >> return x
631-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
637{-# INLINE parMapM #-}
638parMapM :: MonadAsync m =>
639 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
640parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
645-- >>> mapM = Stream.parMapM id
648mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
652-- >>> parSequence modifier = Stream.parMapM modifier id
654{-# INLINE parSequence #-}
655parSequence :: MonadAsync m =>
656 (Config -> Config) -> Stream m (m a) -> Stream m a
657parSequence modifier = parMapM modifier id
660-- >>> sequence = Stream.parSequence id
662{-# INLINE sequence #-}
663sequence :: MonadAsync m =>
664 Stream m (m a) -> Stream m a
665sequence = parSequence id
667-- | Evaluates the streams being zipped in separate threads than the consumer.
668-- The zip function is evaluated in the consumer thread.
670-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
672-- Multi-stream concurrency options won't apply here, see the notes in
675-- If you want to evaluate the zip function as well in a separate thread, you
676-- can use a 'parEval' on 'parZipWithM'.
678{-# INLINE parZipWithM #-}
679parZipWithM :: MonadAsync m
680 => (Config -> Config)
685parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
688-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
690-- >>> m1 = Stream.fromList [1,2,3]
691-- >>> m2 = Stream.fromList [4,5,6]
692-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
693-- [(1,4),(2,5),(3,6)]
695{-# INLINE parZipWith #-}
696parZipWith :: MonadAsync m
697 => (Config -> Config)
702parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
704-------------------------------------------------------------------------------
706-------------------------------------------------------------------------------
711-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
713-- Generate a stream by repeatedly executing a monadic action forever.
714{-# INLINE parRepeatM #-}
715parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
716parRepeatM cfg = parSequence cfg . Stream.repeat
721-- >>> repeatM = Stream.parRepeatM id
723{-# INLINE repeatM #-}
724repeatM :: MonadAsync m => m a -> Stream m a
725repeatM = parRepeatM id
727-- | Generate a stream by concurrently performing a monadic action @n@ times.
731-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
733-- Example, 'parReplicateM' in the following example executes all the
734-- replicated actions concurrently, thus taking only 1 second:
736-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
739{-# INLINE parReplicateM #-}
740parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
741parReplicateM cfg n = parSequence cfg . Stream.replicate n
746-- >>> replicateM n = Stream.parReplicateM id
748{-# INLINE replicateM #-}
749replicateM :: MonadAsync m => Int -> m a -> Stream m a
750replicateM = parReplicateM id
752-------------------------------------------------------------------------------
754-------------------------------------------------------------------------------
756-- Note: we can use another API with two callbacks stop and yield if we want
757-- the callback to be able to indicate end of stream.
759-- | Generates a callback and a stream pair. The callback returned is used to
760-- queue values to the stream. The stream is infinite, there is no way for the
761-- callback to indicate that it is done now.
765{-# INLINE_NORMAL newCallbackStream #-}
766newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
767newCallbackStream = do
768 chan <- newChannel (eager True)
770 -- XXX Add our own thread-id to the SVar as we can not know the callback's
771 -- thread-id and the callback is not run in a managed worker. We need to
772 -- handle this better.
774 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
780 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
781 -- XXX Use fromChannelD?
782 return (callback, fromChannel chan)
784-- | Takes a callback setter function and provides it with a callback. The
785-- callback when invoked adds a value at the tail of the stream. Returns a
786-- stream of values generated by the callback.
790{-# INLINE fromCallback #-}
791fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
792fromCallback setCallback = Stream.concatEffect $ do
793 (callback, stream) <- newCallbackStream
797{-# INLINE_NORMAL pollCountsD #-}
801 -> (D.Stream m Int -> m b)
804pollCountsD predicate fld (D.Stream step state) = D.Stream step' Nothing
807 {-# INLINE_LATE step' #-}
809 -- As long as we are using an "Int" for counts lockfree reads from
810 -- Var should work correctly on both 32-bit and 64-bit machines.
811 -- However, an Int on a 32-bit machine may overflow quickly.
812 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
815 $ Unboxed.toStreamD countVar
816 return $ Skip (Just (countVar, tid, state))
818 step' gst (Just (countVar, tid, st)) = do
823 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
824 return $ Yield x (Just (countVar, tid, s))
825 Skip s -> return $ Skip (Just (countVar, tid, s))
827 liftIO $ killThread tid
830-- | @pollCounts predicate transform fold stream@ counts those elements in the
831-- stream that pass the @predicate@. The resulting count stream is sent to
832-- another thread which transforms it using @transform@ and then folds it using
833-- @fold@. The thread is automatically cleaned up if the stream stops or
834-- aborts due to exception.
836-- For example, to print the count of elements processed every second:
839-- > Stream.drain $ Stream.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (Fold.drainBy print)
840-- $ Stream.enumerateFrom 0
843-- Note: This may not work correctly on 32-bit machines.
847{-# INLINE pollCounts #-}
851 -> (Stream m Int -> m b)
854pollCounts predicate f xs =
856 $ pollCountsD predicate (f . Stream.fromStreamD)
857 $ 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