git.iko.soy/streamly
on 45e2189 help sign in
streamly/src/Streamly/Data/Stream/Concurrent.hs
173 lines 5.8 kB history raw
1-- |
2-- Module : Streamly.Data.Stream.Concurrent
3-- Copyright : (c) 2022 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : released
7-- Portability : GHC
8--
9-- This module provides concurrent streaming abstractions.
10--
11-- == Concurrency Channels
12--
13-- At a lower level, concurrency is implemented using channels that support
14-- concurrent evaluation of streams. We create a channel, and add one or more
15-- streams to it. The channel evaluates multiple streams concurrently and then
16-- generates a single output stream from the results.
17--
18-- == Concurrency Primitives
19--
20-- There are only two fundamental abstractions for concurrency, all other
21-- combinators in this module can be expressed in terms of those two.
22--
23-- The first one is 'evalWith', it evaluates a single stream asynchronously, a
24-- worker thread runs the stream and buffers the results, and the consumer of
25-- the stream runs in another thread consuming it from the buffer, thus
26-- decoupling the production and consumption of the stream. This can be used to
27-- run different stages of a pipeline concurrently.
28--
29-- The second one is 'concatMapWith', unlike 'evalWith' this is used to
30-- evaluate multiple streams concurrently and combine the results. A stream
31-- generator function is mapped to the input stream and all the generated
32-- streams are then evaluated concurrently, and the results are combined.
33--
34-- == Concurrency Configuration
35--
36-- Combinators ending with the "With" suffix ('concatMapWith') take a 'Config'
37-- argument which controls the concurrent behavior. For example, maximum number
38-- of threads to be used ('maxThreads') or the maxmimum size of the buffer
39-- ('maxBuffer'), or how the streams are scheduled with respect to each other
40-- ('interleaved'), or how the results are consumed ('ordered').
41--
42-- Configuration is specified as @Config -> Config@ modifier functions that can
43-- be composed together using function composition. For example, to specify the
44-- maximum threads we can use @concatMapWith (maxThreads 10)@ if we also want
45-- to specify the maximum buffer we can compose the two options
46-- @concatMapwith (maxThreads 10 . maxBuffer 100)@. To use default configuration
47-- use 'id' as the config modifier e.g. @concatMapWith id@.
48--
49-- See the "Configuration" section and individual configuration options'
50-- documentation for the default behavior and values.
51--
52-- == Scheduling behavior
53--
54-- The most important option is to control whether the output of the concurrent
55-- execution is consumed in the same order as the corresponding actions in the
56-- input stream or as soon as they arrive. The default is the latter, however,
57-- we can enforce the original order by using the 'ordered' option.
58--
59-- Another important option controls whether the number of worker threads are
60-- automatically increased or decreased based on the consumption rate or
61-- threads are started as aggresively as possible until the maxThreads or
62-- maxBuffer limits are hit. The default is the former. However, the 'eager'
63-- option can be enabled to use the latter behavior. When 'eager' is on,
64-- consumer thread does not make any impact on scheduling of the available
65-- tasks.
66--
67-- == Concurrent Combinators
68--
69-- Using the two fundamental concurrency primitives we can implement all the
70-- usual streaming combinators with concurrent behavior. Combinators like
71-- 'unfoldrM', 'iterateM' that are inherently serial can be evaluated
72-- concurrently with respect to the consumer pipeline using 'eval'. Combinators
73-- like 'zipWithM', 'mergeByM' can also use 'evalWith' on the input streams to
74-- make them concurrent.
75--
76-- Combinators like 'repeatM', 'replicateM', 'fromListM', 'sequence', 'mapM' in
77-- which all actions are independent of each other can be made concurrent using
78-- the concurrent 'concatMapWith' operation.
79--
80-- A concurrent 'repeatM' would repeat an action using multiple concurrent
81-- executions of the action. Similarly, a concurrent 'mapM' would perform the
82-- mapped action in independent threads.
83--
84-- == Programming Tips
85--
86-- The names in this module do not conflict with other stream modules,
87-- therefore, you can import it in the same namespace:
88--
89-- >>> import qualified Streamly.Data.Stream.Concurrent as Stream
91module Streamly.Data.Stream.Concurrent
92 (
93 MonadAsync
95 -- * Configuration
96 , Config
98 -- ** Limits
99 , maxThreads
100 , maxBuffer
102 -- ** Rate Control
103 , Rate(..)
104 , rate
105 , avgRate
106 , minRate
107 , maxRate
108 , constRate
110 -- ** Stop behavior
111 , StopWhen (..)
112 , stopWhen
114 -- ** Scheduling behavior
115 , eager
116 , ordered
117 , interleaved
119 -- ** Diagnostics
120 , inspect
122 -- * Combinators
123 -- | Stream combinators using a concurrent channel.
125 -- ** Evaluate
126 -- | Evaluates a stream concurrently using a channel.
128 -- , eval
129 , parEval
131 -- ** Generate
132 -- | Uses a single channel to evaluate mapped actions concurrently.
133 , parRepeatM
134 , parReplicateM
136 -- ** Map
137 -- | Uses a single channel to evaluate mapped actions concurrently.
139 -- , mapM
140 , parMapM
141 -- , sequence
142 , parSequence
144 -- ** List of streams
145 -- | Shares a single channel across many streams.
147 -- , append
148 -- , ahead
149 -- , parallel
150 , parConcatList
151 -- , parallelFst
152 -- , parallelMin
153 -- , zipWithM
154 -- , zipWith
156 -- ** Stream of streams
157 -- *** Apply
159 -- , apply
160 , parApply
162 -- *** Concat
163 -- | Shares a single channel across many streams.
165 -- , concat
166 , parConcat
167 -- , concatMap
168 , parConcatMap
169 )
170where
172import Streamly.Internal.Data.Stream.Concurrent
173import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)