@@ -8,79 +8,6 @@
88--
99-- This module provides concurrent streaming abstractions.
1010--
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−--
8411-- == Programming Tips
8512--
8613-- The names in this module do not conflict with other stream modules,
@@ -91,6 +18,7 @@
9118module Streamly.Data.Stream.Concurrent
9219 (
9320 -- * Concurrency
21+ -- $concurrency
9422 MonadAsync
9523
9624 -- ** Configuration
@@ -180,3 +108,88 @@ where
180108
181109import Streamly.Internal.Data.Stream.Concurrent
182110import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
111+
112+-- $concurrency
113+--
114+-- == Concurrency Channels
115+--
116+-- At a lower level, concurrency is implemented using channels that support
117+-- concurrent evaluation of streams. We create a channel, and add one or more
118+-- streams to it. The channel evaluates multiple streams concurrently and then
119+-- generates a single output stream from the results. How the streams are
120+-- combined depends on the configuration of the channel.
121+--
122+-- == Concurrency Primitives
123+--
124+-- There are only a few fundamental abstractions for concurrency, 'parEval',
125+-- 'parConcatMap', and 'parConcatIterate', all concurrency combinators can be
126+-- expressed in terms of these.
127+--
128+-- 'parEval', evaluates a single stream asynchronously, a worker thread runs
129+-- the stream and buffers the results, and the consumer of the stream runs in
130+-- another thread consuming it from the buffer, thus decoupling the production
131+-- and consumption of the stream. This can be used to run different stages of a
132+-- pipeline concurrently.
133+--
134+-- 'parConcatMap' is used to evaluate multiple streams concurrently and combine
135+-- the results. A stream generator function is mapped to the input stream and
136+-- all the generated streams are then evaluated concurrently, and the results
137+-- are combined.
138+--
139+-- 'parConcatIterate' is like 'parConcatMap' but iterates a stream generator
140+-- function recursively over the stream. This can be used to traverse trees or
141+-- graphs.
142+--
143+-- == Concurrency Configuration
144+--
145+-- Concurrent combinators take a 'Config' argument which controls the
146+-- concurrent behavior. For example, maximum number of threads to be used
147+-- ('maxThreads') or the maxmimum size of the buffer ('maxBuffer'), or how the
148+-- streams are scheduled with respect to each other ('interleaved'), or how the
149+-- results are consumed ('ordered').
150+--
151+-- Configuration is specified as @Config -> Config@ modifier functions that can
152+-- be composed together using function composition. For example, to specify the
153+-- maximum threads we can use @parConcatMap (maxThreads 10)@ if we also want to
154+-- specify the maximum buffer we can compose the two options @parConcatMap
155+-- (maxThreads 10 . maxBuffer 100)@. To use default configuration use 'id' as
156+-- the config modifier e.g. @parConcatMap id@.
157+--
158+-- See the @Configuration@ section and individual configuration options'
159+-- documentation for the default behavior and default values of configuration
160+-- parameters.
161+--
162+-- == Scheduling behavior
163+--
164+-- The most important configuration option is to control whether the output of
165+-- the concurrent execution is consumed in the same order as the corresponding
166+-- actions in the input stream or as soon as they arrive. The default is the
167+-- latter, however, we can enforce the original order by using the 'ordered'
168+-- option.
169+--
170+-- Another important option controls whether the number of worker threads are
171+-- automatically increased and decreased based on the consumption rate or
172+-- threads are started as aggresively as possible until the 'maxThreads' or
173+-- 'maxBuffer' limits are hit. The default is the former. However, the 'eager'
174+-- option can be enabled to use the latter behavior. When 'eager' is on, even
175+-- if the stream consumer thread blocks it does not make any impact on the
176+-- scheduling of the available tasks.
177+--
178+-- == Concurrent Combinators
179+--
180+-- Using the few fundamental concurrency primitives we can implement all the
181+-- usual streaming combinators with concurrent behavior. Combinators like
182+-- 'unfoldrM', 'iterateM' that are inherently serial can be evaluated
183+-- concurrently with respect to the consumer pipeline using 'parEval'.
184+-- Combinators like 'zipWithM', 'mergeByM' can also use 'parEval' on the input
185+-- streams to evaluate them concurrently before combining.
186+--
187+-- Combinators like 'repeatM', 'replicateM', 'fromListM', 'sequence', 'mapM' in
188+-- which all actions are independent of each other can be made concurrent using
189+-- the 'parConcatMap' operation.
190+--
191+-- A concurrent 'repeatM' repeats an action using multiple concurrent
192+-- executions of the action. Similarly, a concurrent 'mapM' performs the mapped
193+-- action in independent threads.
194+--
195+-- Some common concurrent combinators are provided in this module.