2-- Module : Streamly.Data.Stream.Concurrent
3-- Copyright : (c) 2022 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
9-- This module provides concurrent streaming abstractions.
13-- The names in this module do not conflict with other stream modules,
14-- therefore, you can import it in the same namespace:
16-- >>> import qualified Streamly.Data.Stream.Concurrent as Stream
18module Streamly.Data.Stream.Concurrent
43 -- *** Scheduling behavior
52 -- | Stream combinators using a concurrent channel.
55 -- | Evaluates a serial stream asynchronously using a concurency channel.
60 -- | Uses a concurrency channel to evaluate multiple actions concurrently.
66 -- | Uses a concurrency channel to evaluate multiple mapped actions
73 -- | Use a channel for each pair.
79 -- *** List of streams
80 -- | Shares a single channel across many streams.
86 -- *** Stream of streams
92 -- | Shares a single channel across many streams.
105import Streamly.Internal.Data.Stream.Concurrent
106import Prelude hiding (mapM, sequence, concat, concatMap, zipWith)
110-- == Concurrency Channels
112-- At a lower level, concurrency is implemented using channels that support
113-- concurrent evaluation of streams. We create a channel, and add one or more
114-- streams to it. The channel evaluates multiple streams concurrently and then
115-- generates a single output stream from the results. How the streams are
116-- combined depends on the configuration of the channel.
118-- == Concurrency Primitives
120-- There are only a few fundamental abstractions for concurrency, 'parEval',
121-- 'parConcatMap', and 'parConcatIterate', all concurrency combinators can be
122-- expressed in terms of these.
124-- 'parEval', evaluates a single stream asynchronously, a worker thread runs
125-- the stream and buffers the results, and the consumer of the stream runs in
126-- another thread consuming it from the buffer, thus decoupling the production
127-- and consumption of the stream. This can be used to run different stages of a
128-- pipeline concurrently.
130-- 'parConcatMap' is used to evaluate multiple streams concurrently and combine
131-- the results. A stream generator function is mapped to the input stream and
132-- all the generated streams are then evaluated concurrently, and the results
135-- 'parConcatIterate' is like 'parConcatMap' but iterates a stream generator
136-- function recursively over the stream. This can be used to traverse trees or
139-- == Concurrency Configuration
141-- Concurrent combinators take a 'Config' argument which controls the
142-- concurrent behavior. For example, maximum number of threads to be used
143-- ('maxThreads') or the maxmimum size of the buffer ('maxBuffer'), or how the
144-- streams are scheduled with respect to each other ('interleaved'), or how the
145-- results are consumed ('ordered').
147-- Configuration is specified as @Config -> Config@ modifier functions that can
148-- be composed together using function composition. For example, to specify the
149-- maximum threads we can use @parConcatMap (maxThreads 10)@ if we also want to
150-- specify the maximum buffer we can compose the two options @parConcatMap
151-- (maxThreads 10 . maxBuffer 100)@. To use default configuration use 'id' as
152-- the config modifier e.g. @parConcatMap id@.
154-- See the @Configuration@ section and individual configuration options'
155-- documentation for the default behavior and default values of configuration
158-- == Scheduling behavior
160-- The most important configuration option is to control whether the output of
161-- the concurrent execution is consumed in the same order as the corresponding
162-- actions in the input stream or as soon as they arrive. The default is the
163-- latter, however, we can enforce the original order by using the 'ordered'
166-- Another important option controls whether the number of worker threads are
167-- automatically increased and decreased based on the consumption rate or
168-- threads are started as aggresively as possible until the 'maxThreads' or
169-- 'maxBuffer' limits are hit. The default is the former. However, the 'eager'
170-- option can be enabled to use the latter behavior. When 'eager' is on, even
171-- if the stream consumer thread blocks it does not make any impact on the
172-- scheduling of the available tasks.
174-- == Concurrent Combinators
176-- Using the few fundamental concurrency primitives we can implement all the
177-- usual streaming combinators with concurrent behavior. Combinators like
178-- 'unfoldrM', 'iterateM' that are inherently serial can be evaluated
179-- concurrently with respect to the consumer pipeline using 'parEval'.
180-- Combinators like 'zipWithM', 'mergeByM' can also use 'parEval' on the input
181-- streams to evaluate them concurrently before combining.
183-- Combinators like 'repeatM', 'replicateM', 'fromListM', 'sequence', 'mapM' in
184-- which all actions are independent of each other can be made concurrent using
185-- the 'parConcatMap' operation.
187-- A concurrent 'repeatM' repeats an action using multiple concurrent
188-- executions of the action. Similarly, a concurrent 'mapM' performs the mapped
189-- action in independent threads.
191-- Some common concurrent combinators are provided in this module.
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