git.iko.soy/streamly
on 2e489ea help sign in
streamly/src/Streamly/Internal/Data/Fold/Concurrent.hs
100 lines 3.7 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Fold.Concurrent
3-- Copyright : (c) 2022 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9-- Using 'parEval' a fold can be decoupled from the driver and evaluated
10-- concurrently with the driver. The driver just pushes an element to the
11-- fold's buffer and waits for async evaluation to finish.
12--
13-- Avoid scanning a stream using a concurrent fold. When scanning a stream
14-- using a concurrent fold we need to keep in mind that the result of the scan
15-- may be delayed because of the asynchronous execution. The results may not be
16-- same as in the case of a synchronous fold.
17--
18-- Stages in a fold pipeline can be made concurrent using 'parEval'.
19--
20-- The 'demux' combinator can be made concurrent by using 'parEval' on the fold
21-- returned by the fold-generating function. Thus, we can fold values for each
22-- key in the input stream concurrently.
23--
24-- Similarly, we can use 'parEval' with other cobminators like 'toMap',
25-- 'demuxToMap', 'classify', 'tee', 'distribute', 'partition' etc. Basically,
26-- any combinator that composes multiple folds or multiple instances of a fold
27-- is a good candidate for running folds concurrently.
29-- TODO:
30--
31-- Before a fold returns "done" it has to drain the child folds. For example,
32-- consider a "take" operation on a `parEval` fold, the take would return as
33-- soon as it has taken required number of elements irrespective of whether the
34-- child fold has yet finished or not.
35--
36-- parLmapM on a fold.
37--
38-- Use multiple worker folds to fold serial chunks of a stream and collect the
39-- results using another fold, combine using a monoid. The results can be
40-- collected out-of-order or in-order. This would be easier if each input
41-- element is a streamable chunk and each fold consumes one at a time. This is
42-- like parConcatMap in streams. We also need to have a lconcatMap to expand
43-- the chunks in the input to streams before folding. This will require an
44-- input Skip constructor. In fact, parLmapM would be implemented in terms of
45-- this like in streams.
46--
47-- Concurrent append: if one fold's buffer becomes full then use the next one
48-- Concurrent interleave/partition: Round robin to n folds.
49-- Concurrent distribute to multiple folds.
51module Streamly.Internal.Data.Fold.Concurrent
52 (
53 -- * Configuration
54 Config
55 , maxBuffer
56 , bound
57 , inspect
59 -- * Combinators
60 -- | Stream combinators using Async channel
62 , parEval
63 )
64where
66-- import Streamly.Internal.Control.Concurrent (MonadAsync)
67-- import Streamly.Internal.Data.Fold (Fold)
69import Streamly.Internal.Data.Fold.Concurrent.Channel
71-- $setup
72--
73-- Imports for example snippets in this module.
74--
75-- >>> :m
76-- >>> :set -XFlexibleContexts
77-- >>> import Control.Concurrent (threadDelay)
78-- >>> import qualified Streamly.Data.Array as Array
79-- >>> import qualified Streamly.Data.Fold as Fold
80-- >>> import qualified Streamly.Data.Parser as Parser
81-- >>> import qualified Streamly.Data.Stream as Stream
82-- >>> import qualified Streamly.Internal.Data.Fold.Concurrent as Fold
83-- >>> import Prelude hiding (concatMap, concat)
84-- >>> :{
85-- delay n = do
86-- threadDelay (n * 1000000) -- sleep for n seconds
87-- putStrLn (show n ++ " sec") -- print "n sec"
88-- return n -- IO Int
89-- :}
91{-
92-- | Evaluate a stream asynchronously using a channel and serve the consumer
93-- from the evaluation buffer.
94--
95-- >>> eval = Fold.parEval id
96--
97{-# INLINE eval #-}
98eval :: MonadAsync m => Fold m a b -> Fold m a b
99eval = parEval id
100-}