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.
60 -- Add repeatM/replicateM?
63 -- | Use a channel for each pair.
64 -- combine/concur/conjoin
74 -- | Shares a single channel across many streams.
85 -- ** Stream of streams
87 -- | Uses a separate channel for each application.
92 -- | Shares a single channel across many streams.
106import Control.Concurrent (myThreadId)
107import Control.Monad (void)
108import Control.Monad.IO.Class (MonadIO(liftIO))
109import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
110import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)
111import Streamly.Internal.Data.Stream.Channel.Types
115import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
116import Streamly.Internal.Data.Stream.Type (Stream)
118import qualified Streamly.Internal.Data.Stream as Stream
119import qualified Streamly.Internal.Data.Stream.StreamK as K
120import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
122import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
123import Streamly.Internal.Data.Stream.Concurrent.Channel
127-- Imports for example snippets in this module.
130-- >>> {-# LANGUAGE FlexibleContexts #-}
131-- >>> import Control.Concurrent (threadDelay)
132-- >>> import qualified Streamly.Data.Array.Unboxed as Array
133-- >>> import qualified Streamly.Data.Fold as Fold
134-- >>> import qualified Streamly.Data.Parser as Parser
135-- >>> import qualified Streamly.Data.Stream as Stream
136-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Async
137-- >>> import Prelude hiding (concatMap, concat, zipWith)
140-- threadDelay (n * 1000000) -- sleep for n seconds
141-- putStrLn (show n ++ " sec") -- print "n sec"
145-------------------------------------------------------------------------------
146-- Evaluating a stream
147-------------------------------------------------------------------------------
150{-# INLINE_NORMAL evalWithD #-}
151evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
152evalWithD modifier m = D.Stream step Nothing
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
164 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
165 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
169-- | Like 'eval' but can specify a config modifier to change the concurrent
170-- channel parameters.
172-- > evalWith = withChannel (const id)
174{-# INLINE evalWith #-}
175evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
176evalWith modifier input = withChannel modifier input (const id)
177 -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream
179-- | Evaluate a stream asynchronously using a channel and serve the consumer
180-- from the evaluation buffer.
182-- >>> eval = Async.evalWith id
185eval :: MonadAsync m => Stream m a -> Stream m a
188-------------------------------------------------------------------------------
189-- combining two streams
190-------------------------------------------------------------------------------
192{-# INLINE _appendGeneric #-}
193_appendGeneric :: MonadAsync m =>
194 ((Config -> Config) -> m (Channel m a))
195 -> (Config -> Config)
199_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
204 chan <- newChan modifier
205 let cfg = modifier defaultConfig
206 done = K.nilM (stopChannel chan)
207 case getStopWhen cfg of
209 toChannelK chan stream2
210 toChannelK chan stream1
212 toChannelK chan stream2
213 toChannelK chan (K.serial stream1 done)
215 toChannelK chan (K.serial stream2 done)
216 toChannelK chan (K.serial stream1 done)
217 return $ Stream.toStreamK $ fromChannel chan
219-- | Create a new channel and add both the streams to it for async evaluation.
220-- The output stream is the result of the evaluation.
221{-# INLINE appendWithK #-}
222appendWithK :: MonadAsync m =>
223 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
224appendWithK modifier stream1 stream2 =
226 if getOrdered (modifier defaultConfig)
227 then concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
228 else _appendGeneric Append.newChannel modifier stream1 stream2
230 concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
232-- | Evaluate the first stream in the current thread and add the second stream
233-- to the supplied channel. This is to be used by a worker thread.
235-- This can be used with concatMapWith:
238-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream
241{-# INLINE _appendWithChanK #-}
242_appendWithChanK :: MonadAsync m =>
243 Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
244_appendWithChanK chan stream1 stream2 =
245 K.before (toChannelK chan stream2) stream1
248-- | Like 'append' but with a Config modifier.
249{-# INLINE combineWith #-}
250combineWith :: MonadAsync m =>
251 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
252combineWith modifier stream1 stream2 =
255 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
257-- | Binary operation to evaluate two streams concurrently prioritizing the
260-- If you want to combine more than two streams you almost always want the
261-- 'concat' operation instead. The performance of this operation degrades
262-- rapidly when more streams are combined as each operation adds one more
263-- concurrent channel. On the other hand, 'concat' uses a single channel for
264-- all streams. However, with this operation you can precisely control the
265-- scheduling by creating arbitrary shape expression trees.
267-- >>> append2 = Async.combineWith id
269-- The following code finishes in 4 seconds:
271-- >>> stream1 = Stream.fromEffect (delay 4)
272-- >>> stream2 = Stream.fromEffect (delay 2)
273-- >>> Stream.fold Fold.toList $ stream1 `Async.append2` stream2
278{-# INLINE append2 #-}
279append2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
280append2 = combineWith id
282-- | Like 'append' but interleaves the streams fairly instead of prioritizing
283-- the left stream. This schedules all streams in a round robin fashion over
284-- limited number of threads.
286-- >>> interleave2 = Async.combineWith Async.interleaved
288{-# INLINE interleave2 #-}
289interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
290interleave2 = combineWith interleaved
292-- | Like 'append' but with 'ordered' on.
294-- >>> ahead2 = Async.combineWith Async.ordered
297ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
298ahead2 = combineWith ordered
300-- | Like 'append2' but with 'eager' on.
302-- >>> parallel2 = Async.combineWith Async.eager
304{-# INLINE parallel2 #-}
305parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
306parallel2 = combineWith eager
308-- | Like 'parallel2' but stops the output as soon as the first stream stops.
310-- >>> parallelFst2 = Async.combineWith (Async.eager . Async.stopWhen Async.FirstStops)
313{-# INLINE parallelFst2 #-}
314parallelFst2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
315parallelFst2 = combineWith (eager . stopWhen FirstStops)
317-- | Like 'parallel2' but stops the output as soon as any of the two streams
320-- >>> parallelMin2 = Async.combineWith (Async.eager . Async.stopWhen Async.AnyStops)
323{-# INLINE parallelMin2 #-}
324parallelMin2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
325parallelMin2 = combineWith (eager . stopWhen AnyStops)
327-------------------------------------------------------------------------------
329-------------------------------------------------------------------------------
331-- | A runner function takes a queuing function @q@ and a stream, it splits the
332-- input stream, queuing the tail and using the head to generate a stream.
333-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
334-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
336{-# INLINE mkEnqueue #-}
337mkEnqueue :: MonadAsync m =>
339 -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
340 -> m (K.Stream m a -> m ())
341mkEnqueue chan runner = do
342 runInIO <- askRunInIO
345 -- Enqueue the outer loop
346 liftIO $ enqueue chan False (runInIO, runner q stream)
347 -- XXX In case of eager dispatch we can just directly dispatch
348 -- a worker with the tail stream here rather than first queuing
349 -- and then dispatching a worker which dequeues the work. The
350 -- older implementation did a direct dispatch here and its perf
351 -- characterstics looked much better.
355-- | Takes the head element of the input stream and queues the tail of the
356-- stream to the channel, then maps the supplied function on the head and
357-- evaluates the resulting stream.
359-- This function is designed to be used by worker threads on a channel to
360-- concurrently map and evaluate a stream.
361{-# INLINE concatMapWithChanK #-}
362concatMapWithChanK :: MonadAsync m =>
363 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
364concatMapWithChanK chan f stream =
365 let run q = concatMapDivK q f
366 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
367 -- K.concatMapWith (_appendWithChanK chan) f stream
369{-# INLINE concatMapWithChanKAny #-}
370concatMapWithChanKAny :: MonadAsync m =>
371 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
372concatMapWithChanKAny chan f stream =
373 let done = K.nilM (stopChannel chan)
374 run q = concatMapDivK q (\x -> K.serial (f x) done)
375 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
377{-# INLINE concatMapWithChanKFirst #-}
378concatMapWithChanKFirst :: MonadAsync m =>
379 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
380concatMapWithChanKFirst chan f stream =
381 let done = K.nilM (stopChannel chan)
382 run q = concatMapDivK q f
383 in K.concatEffect $ do
384 res <- K.uncons stream
386 Nothing -> return K.nil
388 q <- mkEnqueue chan run
390 return $ K.serial (f h) done
392{-# INLINE concatMapWithChanKGeneric #-}
393concatMapWithChanKGeneric :: MonadAsync m =>
396 -> (a -> K.Stream m b)
399concatMapWithChanKGeneric modifier chan f stream = do
400 let cfg = modifier defaultConfig
401 case getStopWhen cfg of
402 AllStop -> concatMapWithChanK chan f stream
403 FirstStops -> concatMapWithChanKFirst chan f stream
404 AnyStops -> concatMapWithChanKAny chan f stream
406-- XXX Add a deep evaluation variant that evaluates individual elements in the
407-- generated streams in parallel.
409-- | Allocate a channel and use it to concurrently evaluate the streams
410-- generated by the mapped function.
412{-# INLINE concatMapWithK #-}
413concatMapWithK :: MonadAsync m =>
414 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
415concatMapWithK modifier f input =
416 let g = concatMapWithChanKGeneric modifier
417 in withChannelK modifier input (`g` f)
419-- concatMapWith modifier f stream = concatWith modifier $ fmap f stream
421-- | Like 'concatMap' but we can also specify the concurrent channel's
422-- configuration parameters using Config modifiers.
424{-# INLINE concatMapWith #-}
425concatMapWith :: MonadAsync m =>
426 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
427concatMapWith modifier f stream =
430 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
432-- | Map each element of the input to a stream and then concurrently evaluate
433-- and concatenate the resulting streams. Multiple streams may be evaluated
434-- concurrently but earlier streams are perferred. Output from the streams are
435-- used as they arrive.
437-- >>> concatMap = Async.concatMapWith id
439-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith cfg id $ Stream.fromList xs
441-- The following streams finish in 4 seconds:
443-- >>> stream1 = Stream.fromEffect (delay 4)
444-- >>> stream2 = Stream.fromEffect (delay 2)
445-- >>> stream3 = Stream.fromEffect (delay 1)
446-- >>> f id [stream1, stream2, stream3]
452-- Limiting threads to 2 schedules the third stream only after one of the first
453-- two has finished, releasing a thread:
455-- >>> f (Async.maxThreads 2) [stream1, stream2, stream3]
459-- When used with a Single thread it behaves like serial concatMap:
461-- >>> f (Async.maxThreads 1) [stream1, stream2, stream3]
465-- >>> stream1 = Stream.fromList [1,2,3]
466-- >>> stream2 = Stream.fromList [4,5,6]
467-- >>> f (Async.maxThreads 1) [stream1, stream2]
470{-# INLINE concatMap #-}
471concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
472concatMap = concatMapWith id
474-- | Map each element of the input to a stream and then concurrently evaluate
475-- and interleave the resulting streams. Unlike 'concatMap' which prefers to
476-- evaluate the earlier stream first, this schedules all streams in a round
477-- robin fashion over the available threads.
479-- >>> concatMapInterleave = Async.concatMapWith Async.interleaved
481-- When used with a single thread it behaves like serial interleaving:
483-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith (Async.interleaved . cfg) id $ Stream.fromList xs
485-- >>> stream1 = Stream.fromList [1,2,3]
486-- >>> stream2 = Stream.fromList [4,5,6]
487-- >>> f (Async.maxThreads 1) [stream1, stream2]
490-- /Works only on finite number of streams/
492{-# INLINE concatMapInterleave #-}
493concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
494concatMapInterleave = concatMapWith interleaved
496-- | Like 'concat' but we can also specify the concurrent channel's
497-- configuration parameters using Config modifiers.
499-- >>> concatWith modifier = Async.concatMapWith modifier id
501{-# INLINE concatWith #-}
502concatWith :: MonadAsync m =>
503 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
504concatWith modifier = concatMapWith modifier id
506-- | Evaluate the streams in the input stream concurrently and combine them.
508-- >>> concat = Async.concatWith id
511concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a
512concat = concatWith id
514-------------------------------------------------------------------------------
516-------------------------------------------------------------------------------
518-- | Like 'concatWith' but works on a list of streams.
520-- >>> concatListWith modifier = Async.concatWith modifier . Stream.fromList
522{-# INLINE concatListWith #-}
523concatListWith :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
524concatListWith modifier = concatWith modifier . Stream.fromList
526-- | Like 'concat' but works on a list of streams.
528-- >>> append = Async.concatListWith id
531append :: MonadAsync m => [Stream m a] -> Stream m a
532append = concatListWith id
534-- | Like 'append' but interleaves the streams fairly instead of prioritizing
535-- the left stream. This schedules all streams in a round robin fashion over
536-- limited number of threads.
538-- >>> interleave = Async.concatListWith Async.interleaved
540{-# INLINE interleave #-}
541interleave :: MonadAsync m => [Stream m a] -> Stream m a
542interleave = concatListWith interleaved
544-- | Like 'append' but with 'ordered' on.
546-- >>> ahead = Async.concatListWith Async.ordered
549ahead :: MonadAsync m => [Stream m a] -> Stream m a
550ahead = concatListWith ordered
552-- | Like 'append' but with 'eager' on.
554-- >>> parallel = Async.concatListWith Async.eager
556{-# INLINE parallel #-}
557parallel :: MonadAsync m => [Stream m a] -> Stream m a
558parallel = concatListWith eager
560-- | Like 'parallel' but stops the output as soon as the first stream stops.
562-- >>> parallelFst = Async.concatListWith (Async.eager . Async.stopWhen Async.FirstStops)
565{-# INLINE parallelFst #-}
566parallelFst :: MonadAsync m => [Stream m a] -> Stream m a
567parallelFst = concatListWith (eager . stopWhen FirstStops)
569-- | Like 'parallel' but stops the output as soon as any of the two streams
572-- >>> parallelMin = Async.concatListWith (Async.eager . Async.stopWhen Async.AnyStops)
575{-# INLINE parallelMin #-}
576parallelMin :: MonadAsync m => [Stream m a] -> Stream m a
577parallelMin = concatListWith (eager . stopWhen AnyStops)
579-------------------------------------------------------------------------------
581-------------------------------------------------------------------------------
583{-# INLINE applyWith #-}
584{-# SPECIALIZE applyWith ::
585 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
586applyWith :: MonadAsync m =>
587 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
588applyWith modifier stream1 stream2 =
591 (\g -> concatMapWith modifier (pure . g) stream2)
595{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
596apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b
599-------------------------------------------------------------------------------
601-------------------------------------------------------------------------------
604-- >>> mapMWith modifier f = Async.concatMapWith modifier (Stream.fromEffect . f)
606{-# INLINE mapMWith #-}
607mapMWith :: MonadAsync m =>
608 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
609mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)
612-- >>> mapM = Async.mapMWith id
614mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
618-- >>> sequenceWith modifier = Async.mapMWith modifier id
620{-# INLINE sequenceWith #-}
621sequenceWith :: MonadAsync m =>
622 (Config -> Config) -> Stream m (m a) -> Stream m a
623sequenceWith modifier = mapMWith modifier id
626-- >>> sequence = Async.sequenceWith id
628{-# INLINE sequence #-}
629sequence :: MonadAsync m =>
630 Stream m (m a) -> Stream m a
631sequence = sequenceWith id
634-- >>> zipWithM f m1 m2 = Stream.zipWithM f (Async.eval m1) (Async.eval m2)
636{-# INLINE zipWithM #-}
637zipWithM :: MonadAsync m
638 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
639zipWithM f m1 m2 = Stream.zipWithM f (eval m1) (eval m2)
642-- >>> zipWith f = Stream.zipWithM (\a b -> return $ f a b)
644-- >>> m1 = Stream.fromList [1,2,3]
645-- >>> m2 = Stream.fromList [4,5,6]
646-- >>> Stream.fold Fold.toList $ Async.zipWith (,) m1 m2
647-- [(1,4),(2,5),(3,6)]
649{-# INLINE zipWith #-}
650zipWith :: MonadAsync m
651 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
652zipWith f = zipWithM (\a b -> return $ f a b)
654-------------------------------------------------------------------------------
656-------------------------------------------------------------------------------
658-- Note: we can use another API with two callbacks stop and yield if we want
659-- the callback to be able to indicate end of stream.
661-- | Generates a callback and a stream pair. The callback returned is used to
662-- queue values to the stream. The stream is infinite, there is no way for the
663-- callback to indicate that it is done now.
667{-# INLINE_NORMAL newCallbackStream #-}
668newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
669newCallbackStream = do
670 chan <- newChannel eager
672 -- XXX Add our own thread-id to the SVar as we can not know the callback's
673 -- thread-id and the callback is not run in a managed worker. We need to
674 -- handle this better.
676 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
682 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
683 -- XXX Use fromChannelD?
684 return (callback, fromChannel chan)
686-- | Takes a callback setter function and provides it with a callback. The
687-- callback when invoked adds a value at the tail of the stream. Returns a
688-- stream of values generated by the callback.
692{-# INLINE fromCallback #-}
693fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
694fromCallback setCallback = Stream.concatM $ do
695 (callback, stream) <- newCallbackStream
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