git.iko.soy/streamly
on f5fa162 help sign in
streamly/src/Streamly/Data/Stream/Concurrent.hs
174 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 -- * Concurrency
94 MonadAsync
96 -- ** Configuration
97 , Config
99 -- *** Limits
100 , maxThreads
101 , maxBuffer
103 -- *** Rate Control
104 , Rate(..)
105 , rate
106 , avgRate
107 , minRate
108 , maxRate
109 , constRate
111 -- *** Stop behavior
112 , StopWhen (..)
113 , stopWhen
115 -- *** Scheduling behavior
116 , eager
117 , ordered
118 , interleaved
120 -- *** Diagnostics
121 , inspect
123 -- ** Combinators
124 -- | Stream combinators using a concurrent channel.
126 -- *** Evaluate
127 -- | Evaluates a stream concurrently using a channel.
129 -- , eval
130 , parEval
132 -- *** Generate
133 -- | Uses a single channel to evaluate mapped actions concurrently.
134 , parRepeatM
135 , parReplicateM
137 -- *** Map
138 -- | Uses a single channel to evaluate mapped actions concurrently.
140 -- , mapM
141 , parMapM
142 -- , sequence
143 , parSequence
145 -- *** List of streams
146 -- | Shares a single channel across many streams.
148 -- , append
149 -- , ahead
150 -- , parallel
151 , parConcatList
152 -- , parallelFst
153 -- , parallelMin
154 -- , zipWithM
155 -- , zipWith
157 -- *** Stream of streams
158 -- **** Apply
160 -- , apply
161 , parApply
163 -- **** Concat
164 -- | Shares a single channel across many streams.
166 -- , concat
167 , parConcat
168 -- , concatMap
169 , parConcatMap
170 )
171where
173import Streamly.Internal.Data.Stream.Concurrent
174import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)