git.iko.soy/streamly
on bfeee8a help sign in
streamly/src/Streamly/Internal/Data/Stream/Concurrent.hs
774 lines 26 kB history raw
1-- |
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
7-- Portability : GHC
8--
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
20 (
21 -- * Imports
22 -- $setup
24 -- * Types
25 MonadAsync
27 -- * Configuration
28 , Config
29 , maxThreads
30 , maxBuffer
31 , eager
32 , StopWhen (..)
33 , stopWhen
34 , ordered
35 , interleaved
36 -- maxYields
37 , Rate(..)
38 , rate
39 , avgRate
40 , minRate
41 , maxRate
42 , constRate
43 , inspect
45 -- * Combinators
46 -- | Stream combinators using a concurrent channel
48 -- ** Evaluate
49 -- | Evaluates a stream concurrently using a channel.
50 , parEval
51 -- Add unfoldrM/iterateM?
53 -- ** Generate
54 -- | Uses a single channel to evaluate all actions.
55 , parRepeatM
56 , parReplicateM
58 -- ** Map
59 -- | Uses a single channel to evaluate all actions.
60 , parMapM
61 , parSequence
63 -- ** Combine two
64 -- | Use a channel for each pair.
65 , parTwo
66 , parZipWithM
67 , parZipWith
68 , parMergeByM
69 , parMergeBy
71 -- ** List of streams
72 -- | Shares a single channel across many streams.
73 , parListLazy
74 , parListOrdered
75 , parListInterleaved
76 , parListEager
77 , parListEagerFst
78 , parListEagerMin
79 , parList
81 -- ** Stream of streams
82 -- *** Apply
83 , parApply
85 -- *** Concat
86 -- | Shares a single channel across many streams.
87 , parConcat
88 , parConcatMap
90 -- *** ConcatIterate
91 , parConcatIterate
93 -- ** Reactive
94 , fromCallback
95 , tapCountD
96 , tapCount
97 )
98where
100#include "inline.hs"
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
109 ( ChildEvent(..)
110 , concatMapDivK
111 )
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
125-- $setup
126--
127-- Imports for example snippets in this module.
128--
129-- >>> :m
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.Internal.Data.Stream as Stream hiding (append2)
136-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream
137-- >>> import Prelude hiding (concatMap, concat, zipWith)
138-- >>> :{
139-- delay n = do
140-- threadDelay (n * 1000000) -- sleep for n seconds
141-- putStrLn (show n ++ " sec") -- print "n sec"
142-- return n -- IO Int
143-- :}
145-------------------------------------------------------------------------------
146-- Evaluating a stream
147-------------------------------------------------------------------------------
149{-
150{-# INLINE_NORMAL parEvalD #-}
151parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
152parEvalD modifier m = D.Stream step Nothing
153 where
155 step _ Nothing = do
156 chan <- newChannel modifier
157 sendFirstWorker chan (D.toStreamK m)
158 -- XXX should use an unfold to make this efficient
159 return $ D.Skip $ Just $ fromChannelD chan
161 step gst (Just (D.UnStream step1 st)) = do
162 r <- step1 gst st
163 return $ case r of
164 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
165 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
166 D.Stop -> D.Stop
167-}
169-- | Evaluate a stream asynchronously. In a serial stream, each element of the
170-- stream is generated as it is demanded by the consumer. `parEval` evaluates
171-- multiple elements of the stream ahead of time and serves the results from a
172-- buffer.
173--
174-- Note that the evaluation requires only one thread as only one stream needs
175-- to be evaluated. Therefore, the concurrency options that are relevant to
176-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,
177-- ordered, stopWhen options won't have any effect.
178--
179{-# INLINE parEval #-}
180parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
181parEval modifier input = withChannel modifier input (const id)
182 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
184-------------------------------------------------------------------------------
185-- combining two streams
186-------------------------------------------------------------------------------
188{-# INLINE _appendGeneric #-}
189_appendGeneric :: MonadAsync m =>
190 ((Config -> Config) -> m (Channel m a))
191 -> (Config -> Config)
192 -> K.StreamK m a
193 -> K.StreamK m a
194 -> K.StreamK m a
195_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
197 where
199 action = do
200 chan <- newChan modifier
201 let cfg = modifier defaultConfig
202 done = K.nilM (stopChannel chan)
203 case getStopWhen cfg of
204 AllStop -> do
205 toChannelK chan stream2
206 toChannelK chan stream1
207 FirstStops -> do
208 toChannelK chan stream2
209 toChannelK chan (K.append stream1 done)
210 AnyStops -> do
211 toChannelK chan (K.append stream2 done)
212 toChannelK chan (K.append stream1 done)
213 return $ Stream.toStreamK $ fromChannel chan
215-- | Create a new channel and add both the streams to it for async evaluation.
216-- The output stream is the result of the evaluation.
217{-# INLINE appendWithK #-}
218appendWithK :: MonadAsync m =>
219 (Config -> Config) -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
220appendWithK modifier stream1 stream2 =
221{-
222 if getOrdered (modifier defaultConfig)
223 then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
224 else _appendGeneric Append.newChannel modifier stream1 stream2
225-}
226 parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
228-- | Evaluate the first stream in the current thread and add the second stream
229-- to the supplied channel. This is to be used by a worker thread.
230--
231-- This can be used with parConcatMap:
232--
233-- @
234-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
235-- @
236--
237{-# INLINE _appendWithChanK #-}
238_appendWithChanK :: MonadAsync m =>
239 Channel m a -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a
240_appendWithChanK chan stream1 stream2 =
241 K.before (toChannelK chan stream2) stream1
243-- | Binary operation to evaluate two streams concurrently using a channel.
244--
245-- If you want to combine more than two streams you almost always want the
246-- 'parList' or `parConcat` operation instead. The performance of this
247-- operation degrades rapidly when more streams are combined as each operation
248-- adds one more concurrent channel. On the other hand, 'parConcat' uses a
249-- single channel for all streams. However, with this operation you can
250-- precisely control the scheduling by creating arbitrary shape expression
251-- trees.
252--
253-- Definition:
254--
255-- >>> parTwo cfg x y = Stream.parList cfg [x, y]
256--
257-- Example, the following code finishes in 4 seconds:
258--
259-- >>> async = Stream.parTwo id
260-- >>> stream1 = Stream.fromEffect (delay 4)
261-- >>> stream2 = Stream.fromEffect (delay 2)
262-- >>> Stream.fold Fold.toList $ stream1 `async` stream2
263-- 2 sec
264-- 4 sec
265-- [2,4]
266--
267{-# INLINE parTwo #-}
268parTwo :: MonadAsync m =>
269 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
270parTwo modifier stream1 stream2 =
271 Stream.fromStreamK
272 $ appendWithK
273 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
275-------------------------------------------------------------------------------
276-- concat streams
277-------------------------------------------------------------------------------
279-- | A runner function takes a queuing function @q@ and a stream, it splits the
280-- input stream, queuing the tail and using the head to generate a stream.
281-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
282-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
283-- between the two.
284{-# INLINE mkEnqueue #-}
285mkEnqueue :: MonadAsync m =>
286 Channel m b
287 -> ((K.StreamK m a -> m ()) -> K.StreamK m a -> K.StreamK m b)
288 -> m (K.StreamK m a -> m ())
289mkEnqueue chan runner = do
290 runInIO <- askRunInIO
291 return
292 $ let q stream = do
293 -- Enqueue the outer loop
294 liftIO $ enqueue chan False (runInIO, runner q stream)
295 -- XXX In case of eager dispatch we can just directly dispatch
296 -- a worker with the tail stream here rather than first queuing
297 -- and then dispatching a worker which dequeues the work. The
298 -- older implementation did a direct dispatch here and its perf
299 -- characterstics looked much better.
300 eagerDispatch chan
301 in q
303-- | Takes the head element of the input stream and queues the tail of the
304-- stream to the channel, then maps the supplied function on the head and
305-- evaluates the resulting stream.
306--
307-- This function is designed to be used by worker threads on a channel to
308-- concurrently map and evaluate a stream.
309{-# INLINE parConcatMapChanK #-}
310parConcatMapChanK :: MonadAsync m =>
311 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
312parConcatMapChanK chan f stream =
313 let run q = concatMapDivK q f
314 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
315 -- K.parConcatMap (_appendWithChanK chan) f stream
317{-# INLINE parConcatMapChanKAny #-}
318parConcatMapChanKAny :: MonadAsync m =>
319 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
320parConcatMapChanKAny chan f stream =
321 let done = K.nilM (stopChannel chan)
322 run q = concatMapDivK q (\x -> K.append (f x) done)
323 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
325{-# INLINE parConcatMapChanKFirst #-}
326parConcatMapChanKFirst :: MonadAsync m =>
327 Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
328parConcatMapChanKFirst chan f stream =
329 let done = K.nilM (stopChannel chan)
330 run q = concatMapDivK q f
331 in K.concatEffect $ do
332 res <- K.uncons stream
333 case res of
334 Nothing -> return K.nil
335 Just (h, t) -> do
336 q <- mkEnqueue chan run
337 q t
338 return $ K.append (f h) done
340{-# INLINE parConcatMapChanKGeneric #-}
341parConcatMapChanKGeneric :: MonadAsync m =>
342 (Config -> Config)
343 -> Channel m b
344 -> (a -> K.StreamK m b)
345 -> K.StreamK m a
346 -> K.StreamK m b
347parConcatMapChanKGeneric modifier chan f stream = do
348 let cfg = modifier defaultConfig
349 case getStopWhen cfg of
350 AllStop -> parConcatMapChanK chan f stream
351 FirstStops -> parConcatMapChanKFirst chan f stream
352 AnyStops -> parConcatMapChanKAny chan f stream
354-- XXX Add a deep evaluation variant that evaluates individual elements in the
355-- generated streams in parallel.
357-- | Allocate a channel and use it to concurrently evaluate the streams
358-- generated by the mapped function.
359--
360{-# INLINE parConcatMapK #-}
361parConcatMapK :: MonadAsync m =>
362 (Config -> Config) -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b
363parConcatMapK modifier f input =
364 let g = parConcatMapChanKGeneric modifier
365 in withChannelK modifier input (`g` f)
367-- | Map each element of the input to a stream and then concurrently evaluate
368-- and concatenate the resulting streams. Multiple streams may be evaluated
369-- concurrently but earlier streams are perferred. Output from the streams are
370-- used as they arrive.
371--
372-- Definition:
373--
374-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
375--
376-- Examples:
377--
378-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
379--
380-- The following streams finish in 4 seconds:
381--
382-- >>> stream1 = Stream.fromEffect (delay 4)
383-- >>> stream2 = Stream.fromEffect (delay 2)
384-- >>> stream3 = Stream.fromEffect (delay 1)
385-- >>> f id [stream1, stream2, stream3]
386-- 1 sec
387-- 2 sec
388-- 4 sec
389-- [1,2,4]
390--
391-- Limiting threads to 2 schedules the third stream only after one of the first
392-- two has finished, releasing a thread:
393--
394-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
395-- ...
396-- [2,1,4]
397--
398-- When used with a Single thread it behaves like serial concatMap:
399--
400-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
401-- ...
402-- [4,2,1]
403--
404-- >>> stream1 = Stream.fromList [1,2,3]
405-- >>> stream2 = Stream.fromList [4,5,6]
406-- >>> f (Stream.maxThreads 1) [stream1, stream2]
407-- [1,2,3,4,5,6]
408--
409-- Schedule all streams in a round robin fashion over the available threads:
410--
411-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
412--
413-- >>> stream1 = Stream.fromList [1,2,3]
414-- >>> stream2 = Stream.fromList [4,5,6]
415-- >>> f (Stream.maxThreads 1) [stream1, stream2]
416-- [1,4,2,5,3,6]
417--
418{-# INLINE parConcatMap #-}
419parConcatMap :: MonadAsync m =>
420 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
421parConcatMap modifier f stream =
422 Stream.fromStreamK
423 $ parConcatMapK
424 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
426-- | Evaluate the streams in the input stream concurrently and combine them.
427--
428-- >>> parConcat modifier = Stream.parConcatMap modifier id
429--
430{-# INLINE parConcat #-}
431parConcat :: MonadAsync m =>
432 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
433parConcat modifier = parConcatMap modifier id
435-------------------------------------------------------------------------------
436-- concat Lists
437-------------------------------------------------------------------------------
439-- | Like 'parConcat' but works on a list of streams.
440--
441-- >>> parList modifier = Stream.parConcat modifier . Stream.fromList
442--
443{-# INLINE parList #-}
444parList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
445parList modifier = parConcat modifier . Stream.fromList
447-- | Like 'concat' but works on a list of streams.
448--
449-- >>> parListLazy = Stream.parList id
450--
451{-# INLINE parListLazy #-}
452parListLazy :: MonadAsync m => [Stream m a] -> Stream m a
453parListLazy = parList id
455-- | Like 'parListLazy' but interleaves the streams fairly instead of prioritizing
456-- the left stream. This schedules all streams in a round robin fashion over
457-- limited number of threads.
458--
459-- >>> parListInterleaved = Stream.parList (Stream.interleaved True)
460--
461{-# INLINE parListInterleaved #-}
462parListInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
463parListInterleaved = parList (interleaved True)
465-- | Like 'parListLazy' but with 'ordered' on.
466--
467-- >>> parListOrdered = Stream.parList (Stream.ordered True)
468--
469{-# INLINE parListOrdered #-}
470parListOrdered :: MonadAsync m => [Stream m a] -> Stream m a
471parListOrdered = parList (ordered True)
473-- | Like 'parListLazy' but with 'eager' on.
474--
475-- >>> parListEager = Stream.parList (Stream.eager True)
476--
477{-# INLINE parListEager #-}
478parListEager :: MonadAsync m => [Stream m a] -> Stream m a
479parListEager = parList (eager True)
481-- | Like 'parListEager' but stops the output as soon as the first stream stops.
482--
483-- >>> parListEagerFst = Stream.parList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
484--
485{-# INLINE parListEagerFst #-}
486parListEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
487parListEagerFst = parList (eager True . stopWhen FirstStops)
489-- | Like 'parListEager' but stops the output as soon as any of the two streams
490-- stops.
491--
492-- Definition:
493--
494-- >>> parListEagerMin = Stream.parList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
495--
496{-# INLINE parListEagerMin #-}
497parListEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
498parListEagerMin = parList (eager True . stopWhen AnyStops)
500-------------------------------------------------------------------------------
501-- Applicative
502-------------------------------------------------------------------------------
504-- | Apply an argument stream to a function stream concurrently. Uses a
505-- shared channel for all individual applications within a stream application.
506{-# INLINE parApply #-}
507{-# SPECIALIZE parApply ::
508 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
509parApply :: MonadAsync m =>
510 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
511parApply modifier stream1 stream2 =
512 parConcatMap
513 modifier
514 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
515 stream1
517-------------------------------------------------------------------------------
518-- Map
519-------------------------------------------------------------------------------
521-- |
522-- Definition:
523--
524-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
525--
526-- Example, the following example finishes in 1 second as all actions run in
527-- parallel. Even though results are available out of order they are ordered
528-- due to the config option::
529--
530-- >>> f x = delay x >> return x
531-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
532-- 1 sec
533-- 2 sec
534-- 3 sec
535-- [3,2,1]
536--
537{-# INLINE parMapM #-}
538parMapM :: MonadAsync m =>
539 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
540parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
542-- |
543-- >>> parSequence modifier = Stream.parMapM modifier id
544--
545{-# INLINE parSequence #-}
546parSequence :: MonadAsync m =>
547 (Config -> Config) -> Stream m (m a) -> Stream m a
548parSequence modifier = parMapM modifier id
550-- | Evaluates the streams being zipped in separate threads than the consumer.
551-- The zip function is evaluated in the consumer thread.
552--
553-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
554--
555-- Multi-stream concurrency options won't apply here, see the notes in
556-- 'parEval'.
557--
558-- If you want to evaluate the zip function as well in a separate thread, you
559-- can use a 'parEval' on 'parZipWithM'.
560--
561{-# INLINE parZipWithM #-}
562parZipWithM :: MonadAsync m
563 => (Config -> Config)
564 -> (a -> b -> m c)
565 -> Stream m a
566 -> Stream m b
567 -> Stream m c
568parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
570-- |
571-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
572--
573-- >>> m1 = Stream.fromList [1,2,3]
574-- >>> m2 = Stream.fromList [4,5,6]
575-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
576-- [(1,4),(2,5),(3,6)]
577--
578{-# INLINE parZipWith #-}
579parZipWith :: MonadAsync m
580 => (Config -> Config)
581 -> (a -> b -> c)
582 -> Stream m a
583 -> Stream m b
584 -> Stream m c
585parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
587-- | Like 'mergeByM' but evaluates both the streams concurrently.
588--
589-- Definition:
590--
591-- >>> parMergeByM cfg f m1 m2 = Stream.mergeByM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
592--
593{-# INLINE parMergeByM #-}
594parMergeByM :: MonadAsync m
595 => (Config -> Config)
596 -> (a -> a -> m Ordering)
597 -> Stream m a
598 -> Stream m a
599 -> Stream m a
600parMergeByM cfg f m1 m2 = Stream.mergeByM f (parEval cfg m1) (parEval cfg m2)
602-- | Like 'mergeBy' but evaluates both the streams concurrently.
603--
604-- Definition:
605--
606-- >>> parMergeBy cfg f = Stream.parMergeByM cfg (\a b -> return $ f a b)
607--
608{-# INLINE parMergeBy #-}
609parMergeBy :: MonadAsync m
610 => (Config -> Config)
611 -> (a -> a -> Ordering)
612 -> Stream m a
613 -> Stream m a
614 -> Stream m a
615parMergeBy cfg f = parMergeByM cfg (\a b -> return $ f a b)
617-------------------------------------------------------------------------------
618-- concatIterate
619-------------------------------------------------------------------------------
621-- | Same as 'concatIterate' but concurrent.
622--
623-- /Pre-release/
624{-# INLINE parConcatIterate #-}
625parConcatIterate :: MonadAsync m =>
626 (Config -> Config)
627 -> (a -> Stream m a)
628 -> Stream m a
629 -> Stream m a
630parConcatIterate modifier f input =
631 Stream.fromStreamK
632 $ withChannelK modifier (Stream.toStreamK input) iterateStream
634 where
636 iterateStream channel stream =
637 parConcatMapChanKGeneric modifier channel (generate channel) stream
639 generate channel x =
640 -- XXX The channel q should be FIFO for DFS, otherwise it is BFS
641 x `K.cons` iterateStream channel (Stream.toStreamK $ f x)
643-------------------------------------------------------------------------------
644-- Generate
645-------------------------------------------------------------------------------
647-- |
648-- Definition:
649--
650-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
651--
652-- Generate a stream by repeatedly executing a monadic action forever.
653{-# INLINE parRepeatM #-}
654parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
655parRepeatM cfg = parSequence cfg . Stream.repeat
657-- | Generate a stream by concurrently performing a monadic action @n@ times.
658--
659-- Definition:
660--
661-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
662--
663-- Example, 'parReplicateM' in the following example executes all the
664-- replicated actions concurrently, thus taking only 1 second:
665--
666-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
667-- ...
668--
669{-# INLINE parReplicateM #-}
670parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
671parReplicateM cfg n = parSequence cfg . Stream.replicate n
673-------------------------------------------------------------------------------
674-- Reactive
675-------------------------------------------------------------------------------
677-- Note: we can use another API with two callbacks stop and yield if we want
678-- the callback to be able to indicate end of stream.
679--
680-- | Generates a callback and a stream pair. The callback returned is used to
681-- queue values to the stream. The stream is infinite, there is no way for the
682-- callback to indicate that it is done now.
683--
684-- /Pre-release/
685--
686{-# INLINE_NORMAL newCallbackStream #-}
687newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
688newCallbackStream = do
689 chan <- newChannel (eager True)
691 -- XXX Add our own thread-id to the SVar as we can not know the callback's
692 -- thread-id and the callback is not run in a managed worker. We need to
693 -- handle this better.
694 liftIO myThreadId
695 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
697 let callback a =
698 liftIO
699 $ void
700 $ sendWithDoorBell
701 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
702 -- XXX Use fromChannelD?
703 return (callback, fromChannel chan)
705-- | Supplies a stream generating callback to a callback setter function. Each
706-- invocation of the callback results in a value being generated in the
707-- resulting stream.
708--
709-- /Pre-release/
710--
711{-# INLINE fromCallback #-}
712fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
713fromCallback setCallback = Stream.concatEffect $ do
714 (callback, stream) <- newCallbackStream
715 setCallback callback
716 return stream
718{-# INLINE_NORMAL tapCountD #-}
719tapCountD
720 :: MonadAsync m
721 => (a -> Bool)
722 -> (D.Stream m Int -> m b)
723 -> D.Stream m a
724 -> D.Stream m a
725tapCountD predicate fld (D.Stream step state) = D.Stream step' Nothing
726 where
728 {-# INLINE_LATE step' #-}
729 step' _ Nothing = do
730 -- As long as we are using an "Int" for counts lockfree reads from
731 -- Var should work correctly on both 32-bit and 64-bit machines.
732 -- However, an Int on a 32-bit machine may overflow quickly.
733 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
734 tid <- forkManaged
735 $ void $ fld
736 $ Unboxed.toStreamD countVar
737 return $ Skip (Just (countVar, tid, state))
739 step' gst (Just (countVar, tid, st)) = do
740 r <- step gst st
741 case r of
742 Yield x s -> do
743 when (predicate x)
744 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
745 return $ Yield x (Just (countVar, tid, s))
746 Skip s -> return $ Skip (Just (countVar, tid, s))
747 Stop -> do
748 liftIO $ killThread tid
749 return Stop
751-- | @tapCount predicate fold stream@ taps the count of those elements in the
752-- stream that pass the @predicate@. The resulting count stream is sent to
753-- another thread which folds it using @fold@.
754--
755-- For example, to print the count of elements processed every second:
756--
757-- >>> rate = Stream.rollingMap2 (flip (-)) . Stream.delayPost 1
758-- >>> report = Stream.fold (Fold.drainMapM print) . rate
759-- >>> tap = Stream.tapCount (const True) report
760-- >>> go = Stream.fold Fold.drain $ tap $ Stream.enumerateFrom 0
761--
762-- Note: This may not work correctly on 32-bit machines because of Int
763-- overflow.
764--
765-- /Pre-release/
766--
767{-# INLINE tapCount #-}
768tapCount ::
769 (MonadAsync m)
770 => (a -> Bool)
771 -> (Stream m Int -> m b)
772 -> Stream m a
773 -> Stream m a
774tapCount = tapCountD