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.
108import Control.Concurrent (myThreadId, killThread)
109import Control.Monad (void, when)
110import Control.Monad.IO.Class (MonadIO(liftIO))
111import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
112import Streamly.Internal.Control.ForkLifted (forkManaged)
113import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)
114import Streamly.Internal.Data.Stream.Channel.Types
118import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
119import Streamly.Internal.Data.Stream.Type (Stream)
120import Streamly.Internal.Data.Stream.StreamD (Step(..))
122import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
123import qualified Streamly.Internal.Data.Stream as Stream
124import qualified Streamly.Internal.Data.Stream.StreamD as D
125import qualified Streamly.Internal.Data.Stream.StreamK as K
126import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
128import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
129import Streamly.Internal.Data.Stream.Concurrent.Channel
133-- Imports for example snippets in this module.
136-- >>> {-# LANGUAGE FlexibleContexts #-}
137-- >>> import Control.Concurrent (threadDelay)
138-- >>> import qualified Streamly.Data.Array as Array
139-- >>> import qualified Streamly.Data.Fold as Fold
140-- >>> import qualified Streamly.Data.Parser as Parser
141-- >>> import qualified Streamly.Data.Stream as Stream
142-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur
143-- >>> import Prelude hiding (concatMap, concat, zipWith)
146-- threadDelay (n * 1000000) -- sleep for n seconds
147-- putStrLn (show n ++ " sec") -- print "n sec"
151-------------------------------------------------------------------------------
152-- Evaluating a stream
153-------------------------------------------------------------------------------
156{-# INLINE_NORMAL evalWithD #-}
157evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
158evalWithD modifier m = D.Stream step Nothing
162 chan <- newChannel modifier
163 sendFirstWorker chan (D.toStreamK m)
164 -- XXX should use an unfold to make this efficient
165 return $ D.Skip $ Just $ fromChannelD chan
167 step gst (Just (D.UnStream step1 st)) = do
170 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
171 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
175-- | Like 'eval' but can specify a config modifier to change the concurrent
176-- channel parameters.
178-- > evalWith = withChannel (const id)
180{-# INLINE evalWith #-}
181evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
182evalWith modifier input = withChannel modifier input (const id)
183 -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream
185-- | Evaluate a stream asynchronously using a channel and serve the consumer
186-- from the evaluation buffer.
188-- >>> eval = Concur.evalWith id
191eval :: MonadAsync m => Stream m a -> Stream m a
194-------------------------------------------------------------------------------
195-- combining two streams
196-------------------------------------------------------------------------------
198{-# INLINE _appendGeneric #-}
199_appendGeneric :: MonadAsync m =>
200 ((Config -> Config) -> m (Channel m a))
201 -> (Config -> Config)
205_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
210 chan <- newChan modifier
211 let cfg = modifier defaultConfig
212 done = K.nilM (stopChannel chan)
213 case getStopWhen cfg of
215 toChannelK chan stream2
216 toChannelK chan stream1
218 toChannelK chan stream2
219 toChannelK chan (K.serial stream1 done)
221 toChannelK chan (K.serial stream2 done)
222 toChannelK chan (K.serial stream1 done)
223 return $ Stream.toStreamK $ fromChannel chan
225-- | Create a new channel and add both the streams to it for async evaluation.
226-- The output stream is the result of the evaluation.
227{-# INLINE appendWithK #-}
228appendWithK :: MonadAsync m =>
229 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
230appendWithK modifier stream1 stream2 =
232 if getOrdered (modifier defaultConfig)
233 then concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
234 else _appendGeneric Append.newChannel modifier stream1 stream2
236 concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
238-- | Evaluate the first stream in the current thread and add the second stream
239-- to the supplied channel. This is to be used by a worker thread.
241-- This can be used with concatMapWith:
244-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream
247{-# INLINE _appendWithChanK #-}
248_appendWithChanK :: MonadAsync m =>
249 Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
250_appendWithChanK chan stream1 stream2 =
251 K.before (toChannelK chan stream2) stream1
254-- | Like 'append' but with a Config modifier.
255{-# INLINE combineWith #-}
256combineWith :: MonadAsync m =>
257 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
258combineWith modifier stream1 stream2 =
261 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
263-- | Binary operation to evaluate two streams concurrently prioritizing the
266-- If you want to combine more than two streams you almost always want the
267-- 'concat' operation instead. The performance of this operation degrades
268-- rapidly when more streams are combined as each operation adds one more
269-- concurrent channel. On the other hand, 'concat' uses a single channel for
270-- all streams. However, with this operation you can precisely control the
271-- scheduling by creating arbitrary shape expression trees.
273-- >>> append2 = Concur.combineWith id
275-- The following code finishes in 4 seconds:
277-- >>> stream1 = Stream.fromEffect (delay 4)
278-- >>> stream2 = Stream.fromEffect (delay 2)
279-- >>> Stream.fold Fold.toList $ stream1 `Concur.append2` stream2
284{-# INLINE append2 #-}
285append2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
286append2 = combineWith id
288-- | Like 'append' but interleaves the streams fairly instead of prioritizing
289-- the left stream. This schedules all streams in a round robin fashion over
290-- limited number of threads.
292-- >>> interleave2 = Concur.combineWith (Concur.interleaved True)
294{-# INLINE interleave2 #-}
295interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
296interleave2 = combineWith (interleaved True)
298-- | Like 'append' but with 'ordered' on.
300-- >>> ahead2 = Concur.combineWith (Concur.ordered True)
303ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
304ahead2 = combineWith (ordered True)
306-- | Like 'append2' but with 'eager' on.
308-- >>> parallel2 = Concur.combineWith (Concur.eager True)
310{-# INLINE parallel2 #-}
311parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
312parallel2 = combineWith (eager True)
314-- | Like 'parallel2' but stops the output as soon as the first stream stops.
316-- >>> parallelFst2 = Concur.combineWith (Concur.eager True . Concur.stopWhen Concur.FirstStops)
319{-# INLINE parallelFst2 #-}
320parallelFst2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
321parallelFst2 = combineWith (eager True . stopWhen FirstStops)
323-- | Like 'parallel2' but stops the output as soon as any of the two streams
326-- >>> parallelMin2 = Concur.combineWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)
329{-# INLINE parallelMin2 #-}
330parallelMin2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
331parallelMin2 = combineWith (eager True . stopWhen AnyStops)
333-------------------------------------------------------------------------------
335-------------------------------------------------------------------------------
337-- | A runner function takes a queuing function @q@ and a stream, it splits the
338-- input stream, queuing the tail and using the head to generate a stream.
339-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
340-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
342{-# INLINE mkEnqueue #-}
343mkEnqueue :: MonadAsync m =>
345 -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
346 -> m (K.Stream m a -> m ())
347mkEnqueue chan runner = do
348 runInIO <- askRunInIO
351 -- Enqueue the outer loop
352 liftIO $ enqueue chan False (runInIO, runner q stream)
353 -- XXX In case of eager dispatch we can just directly dispatch
354 -- a worker with the tail stream here rather than first queuing
355 -- and then dispatching a worker which dequeues the work. The
356 -- older implementation did a direct dispatch here and its perf
357 -- characterstics looked much better.
361-- | Takes the head element of the input stream and queues the tail of the
362-- stream to the channel, then maps the supplied function on the head and
363-- evaluates the resulting stream.
365-- This function is designed to be used by worker threads on a channel to
366-- concurrently map and evaluate a stream.
367{-# INLINE concatMapWithChanK #-}
368concatMapWithChanK :: MonadAsync m =>
369 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
370concatMapWithChanK chan f stream =
371 let run q = concatMapDivK q f
372 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
373 -- K.concatMapWith (_appendWithChanK chan) f stream
375{-# INLINE concatMapWithChanKAny #-}
376concatMapWithChanKAny :: MonadAsync m =>
377 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
378concatMapWithChanKAny chan f stream =
379 let done = K.nilM (stopChannel chan)
380 run q = concatMapDivK q (\x -> K.serial (f x) done)
381 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
383{-# INLINE concatMapWithChanKFirst #-}
384concatMapWithChanKFirst :: MonadAsync m =>
385 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
386concatMapWithChanKFirst chan f stream =
387 let done = K.nilM (stopChannel chan)
388 run q = concatMapDivK q f
389 in K.concatEffect $ do
390 res <- K.uncons stream
392 Nothing -> return K.nil
394 q <- mkEnqueue chan run
396 return $ K.serial (f h) done
398{-# INLINE concatMapWithChanKGeneric #-}
399concatMapWithChanKGeneric :: MonadAsync m =>
402 -> (a -> K.Stream m b)
405concatMapWithChanKGeneric modifier chan f stream = do
406 let cfg = modifier defaultConfig
407 case getStopWhen cfg of
408 AllStop -> concatMapWithChanK chan f stream
409 FirstStops -> concatMapWithChanKFirst chan f stream
410 AnyStops -> concatMapWithChanKAny chan f stream
412-- XXX Add a deep evaluation variant that evaluates individual elements in the
413-- generated streams in parallel.
415-- | Allocate a channel and use it to concurrently evaluate the streams
416-- generated by the mapped function.
418{-# INLINE concatMapWithK #-}
419concatMapWithK :: MonadAsync m =>
420 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
421concatMapWithK modifier f input =
422 let g = concatMapWithChanKGeneric modifier
423 in withChannelK modifier input (`g` f)
425-- concatMapWith modifier f stream = concatWith modifier $ fmap f stream
427-- | Like 'concatMap' but we can also specify the concurrent channel's
428-- configuration parameters using Config modifiers.
430{-# INLINE concatMapWith #-}
431concatMapWith :: MonadAsync m =>
432 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
433concatMapWith modifier f stream =
436 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
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.
443-- >>> concatMap = Concur.concatMapWith id
445-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith cfg id $ Stream.fromList xs
447-- The following streams finish in 4 seconds:
449-- >>> stream1 = Stream.fromEffect (delay 4)
450-- >>> stream2 = Stream.fromEffect (delay 2)
451-- >>> stream3 = Stream.fromEffect (delay 1)
452-- >>> f id [stream1, stream2, stream3]
458-- Limiting threads to 2 schedules the third stream only after one of the first
459-- two has finished, releasing a thread:
461-- >>> f (Concur.maxThreads 2) [stream1, stream2, stream3]
465-- When used with a Single thread it behaves like serial concatMap:
467-- >>> f (Concur.maxThreads 1) [stream1, stream2, stream3]
471-- >>> stream1 = Stream.fromList [1,2,3]
472-- >>> stream2 = Stream.fromList [4,5,6]
473-- >>> f (Concur.maxThreads 1) [stream1, stream2]
476{-# INLINE concatMap #-}
477concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
478concatMap = concatMapWith id
480-- | Map each element of the input to a stream and then concurrently evaluate
481-- and interleave the resulting streams. Unlike 'concatMap' which prefers to
482-- evaluate the earlier stream first, this schedules all streams in a round
483-- robin fashion over the available threads.
485-- >>> concatMapInterleave = Concur.concatMapWith (Concur.interleaved True)
487-- When used with a single thread it behaves like serial interleaving:
489-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith (Concur.interleaved True . cfg) id $ Stream.fromList xs
491-- >>> stream1 = Stream.fromList [1,2,3]
492-- >>> stream2 = Stream.fromList [4,5,6]
493-- >>> f (Concur.maxThreads 1) [stream1, stream2]
496-- /Works only on finite number of streams/
498{-# INLINE concatMapInterleave #-}
499concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
500concatMapInterleave = concatMapWith (interleaved True)
502-- | Like 'concat' but we can also specify the concurrent channel's
503-- configuration parameters using Config modifiers.
505-- >>> concatWith modifier = Concur.concatMapWith modifier id
507{-# INLINE concatWith #-}
508concatWith :: MonadAsync m =>
509 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
510concatWith modifier = concatMapWith modifier id
512-- | Evaluate the streams in the input stream concurrently and combine them.
514-- >>> concat = Concur.concatWith id
517concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a
518concat = concatWith id
520-------------------------------------------------------------------------------
522-------------------------------------------------------------------------------
524-- | Like 'concatWith' but works on a list of streams.
526-- >>> concatListWith modifier = Concur.concatWith modifier . Stream.fromList
528{-# INLINE concatListWith #-}
529concatListWith :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
530concatListWith modifier = concatWith modifier . Stream.fromList
532-- | Like 'concat' but works on a list of streams.
534-- >>> append = Concur.concatListWith id
537append :: MonadAsync m => [Stream m a] -> Stream m a
538append = concatListWith id
540-- | Like 'append' but interleaves the streams fairly instead of prioritizing
541-- the left stream. This schedules all streams in a round robin fashion over
542-- limited number of threads.
544-- >>> interleave = Concur.concatListWith (Concur.interleaved True)
546{-# INLINE interleave #-}
547interleave :: MonadAsync m => [Stream m a] -> Stream m a
548interleave = concatListWith (interleaved True)
550-- | Like 'append' but with 'ordered' on.
552-- >>> ahead = Concur.concatListWith (Concur.ordered True)
555ahead :: MonadAsync m => [Stream m a] -> Stream m a
556ahead = concatListWith (ordered True)
558-- | Like 'append' but with 'eager' on.
560-- >>> parallel = Concur.concatListWith (Concur.eager True)
562{-# INLINE parallel #-}
563parallel :: MonadAsync m => [Stream m a] -> Stream m a
564parallel = concatListWith (eager True)
566-- | Like 'parallel' but stops the output as soon as the first stream stops.
568-- >>> parallelFst = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.FirstStops)
570{-# INLINE parallelFst #-}
571parallelFst :: MonadAsync m => [Stream m a] -> Stream m a
572parallelFst = concatListWith (eager True . stopWhen FirstStops)
574-- | Like 'parallel' but stops the output as soon as any of the two streams
577-- >>> parallelMin = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)
579{-# INLINE parallelMin #-}
580parallelMin :: MonadAsync m => [Stream m a] -> Stream m a
581parallelMin = concatListWith (eager True . stopWhen AnyStops)
583-------------------------------------------------------------------------------
585-------------------------------------------------------------------------------
587{-# INLINE applyWith #-}
588{-# SPECIALIZE applyWith ::
589 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
590applyWith :: MonadAsync m =>
591 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
592applyWith modifier stream1 stream2 =
595 (\g -> concatMapWith modifier (pure . g) stream2)
599{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
600apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b
603-------------------------------------------------------------------------------
605-------------------------------------------------------------------------------
608-- >>> mapMWith modifier f = Concur.concatMapWith modifier (Stream.fromEffect . f)
610{-# INLINE mapMWith #-}
611mapMWith :: MonadAsync m =>
612 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
613mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)
616-- >>> mapM = Concur.mapMWith id
618mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
622-- >>> sequenceWith modifier = Concur.mapMWith modifier id
624{-# INLINE sequenceWith #-}
625sequenceWith :: MonadAsync m =>
626 (Config -> Config) -> Stream m (m a) -> Stream m a
627sequenceWith modifier = mapMWith modifier id
630-- >>> sequence = Concur.sequenceWith id
632{-# INLINE sequence #-}
633sequence :: MonadAsync m =>
634 Stream m (m a) -> Stream m a
635sequence = sequenceWith id
638-- >>> zipWithM f m1 m2 = Stream.zipWithM f (Concur.eval m1) (Concur.eval m2)
640{-# INLINE zipWithM #-}
641zipWithM :: MonadAsync m
642 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
643zipWithM f m1 m2 = Stream.zipWithM f (eval m1) (eval m2)
646-- >>> zipWith f = Stream.zipWithM (\a b -> return $ f a b)
648-- >>> m1 = Stream.fromList [1,2,3]
649-- >>> m2 = Stream.fromList [4,5,6]
650-- >>> Stream.fold Fold.toList $ Concur.zipWith (,) m1 m2
651-- [(1,4),(2,5),(3,6)]
653{-# INLINE zipWith #-}
654zipWith :: MonadAsync m
655 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
656zipWith f = zipWithM (\a b -> return $ f a b)
658-------------------------------------------------------------------------------
660-------------------------------------------------------------------------------
662-- Note: we can use another API with two callbacks stop and yield if we want
663-- the callback to be able to indicate end of stream.
665-- | Generates a callback and a stream pair. The callback returned is used to
666-- queue values to the stream. The stream is infinite, there is no way for the
667-- callback to indicate that it is done now.
671{-# INLINE_NORMAL newCallbackStream #-}
672newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
673newCallbackStream = do
674 chan <- newChannel (eager True)
676 -- XXX Add our own thread-id to the SVar as we can not know the callback's
677 -- thread-id and the callback is not run in a managed worker. We need to
678 -- handle this better.
680 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
686 (outputQueue chan) (outputDoorBell chan) (ChildYield a)
687 -- XXX Use fromChannelD?
688 return (callback, fromChannel chan)
690-- | Takes a callback setter function and provides it with a callback. The
691-- callback when invoked adds a value at the tail of the stream. Returns a
692-- stream of values generated by the callback.
696{-# INLINE fromCallback #-}
697fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> Stream m a
698fromCallback setCallback = Stream.concatM $ do
699 (callback, stream) <- newCallbackStream
703{-# INLINE_NORMAL pollCountsD #-}
707 -> (D.Stream m Int -> m b)
710pollCountsD predicate fld (D.Stream step state) = D.Stream step' Nothing
713 {-# INLINE_LATE step' #-}
715 -- As long as we are using an "Int" for counts lockfree reads from
716 -- Var should work correctly on both 32-bit and 64-bit machines.
717 -- However, an Int on a 32-bit machine may overflow quickly.
718 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)
721 $ Unboxed.toStreamD countVar
722 return $ Skip (Just (countVar, tid, state))
724 step' gst (Just (countVar, tid, st)) = do
729 $ liftIO $ Unboxed.modifyIORef' countVar (+ 1)
730 return $ Yield x (Just (countVar, tid, s))
731 Skip s -> return $ Skip (Just (countVar, tid, s))
733 liftIO $ killThread tid
736-- | @pollCounts predicate transform fold stream@ counts those elements in the
737-- stream that pass the @predicate@. The resulting count stream is sent to
738-- another thread which transforms it using @transform@ and then folds it using
739-- @fold@. The thread is automatically cleaned up if the stream stops or
740-- aborts due to exception.
742-- For example, to print the count of elements processed every second:
745-- > Stream.drain $ Concur.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (Fold.drainBy print)
746-- $ Stream.enumerateFrom 0
749-- Note: This may not work correctly on 32-bit machines.
753{-# INLINE pollCounts #-}
757 -> (Stream m Int -> m b)
760pollCounts predicate f xs =
762 $ pollCountsD predicate (f . Stream.fromStreamD)
763 $ 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