@@ -18,23 +18,8 @@
1818--
1919-- We will add some more imports in the examples as needed.
2020--
21−-- = Overview
22−--
23−-- Streamly is a framework for modular data flow based programming and
24−-- declarative concurrency. Streamly API is similar to Haskell lists.
25−--
26−-- The type @SerialT IO a@ is an effectful equivalent of a list @[a]@ using the
27−-- IO monad. Streams can be constructed like lists, except that they use 'nil'
28−-- instead of '[]' and 'cons' instead of ':'. `cons` constructs a pure stream
29−-- which is more or less the same as a list:
30−--
31−-- >>> import Streamly.Prelude (SerialT, cons, consM, nil)
32−-- >>> stream = 1 `cons` 2 `cons` nil :: SerialT IO Int
33−-- >>> Stream.toList stream -- IO [Int]
34−-- [1,2]
35−--
36−-- To demonstrate the serial vs. concurrent execution behavior we will use the
37−-- following blocking IO action that blocks for @n@ seconds:
21+-- For effectful streams we will use the following IO action that blocks for
22+-- @n@ seconds:
3823--
3924-- >>> import Control.Concurrent (threadDelay)
4025-- >>> :{
@@ -48,15 +33,34 @@
4833-- 1 sec
4934-- 1
5035--
36+-- = Overview
37+--
38+-- Streamly is a framework for modular data flow based programming and
39+-- declarative concurrency. Powerful stream fusion framework in streamly
40+-- allows high performance combinatorial programming even when using byte level
41+-- streams. Streamly API is similar to Haskell lists.
42+--
43+-- The basic stream type is 'SerialT'. The type @SerialT IO a@ is an effectful
44+-- equivalent of a list @[a]@ using the IO monad. Streams can be constructed
45+-- like lists, except that they use 'nil' instead of '[]' and 'cons' instead of
46+-- ':'.
47+--
48+-- `cons` constructs a pure stream which is more or less the same as a list:
49+--
50+-- >>> import Streamly.Prelude (SerialT, cons, consM, nil)
51+-- >>> stream = 1 `cons` 2 `cons` nil :: SerialT IO Int
52+-- >>> Stream.toList stream -- IO [Int]
53+-- [1,2]
54+--
5155-- 'consM' constructs a stream from effectful actions:
5256--
53−-- @
54−-- > stream = delay 1 \`consM` delay 2 \`consM` 'nil'
55−-- > Stream.'toList' stream
57+-- >>> stream = delay 1 `consM` delay 2 `consM` nil
58+-- >>> Stream.toList stream
5659-- 1 sec
5760-- 2 sec
5861-- [1,2]
59−-- @
62+--
63+-- == Console Echo Program
6064--
6165-- In the following example, 'repeatM' generates an infinite stream of 'String'
6266-- by repeatedly performing the 'getLine' IO action. 'mapM' then applies
@@ -74,18 +78,105 @@
7478-- :}
7579-- @
7680--
77−-- This is a console echo program. It is an example of a declarative loop.
78−-- Compare it with an imperative @while@ loop.
81+-- This is a console echo program. It is an example of a declarative loop
82+-- written using streaming combinators. Compare it with an imperative @while@
83+-- loop.
7984--
8085-- Hopefully, this gives you an idea how we can program declaratively by
81−-- represneting loops using streams. For more details please see the
82−-- "Streamly.Tutorial" module.
86+-- representing loops using streams. In this module, you can find all
87+-- "Data.List" like functions and many more powerful combinators to perform
88+-- common programming tasks. Also see "Streamly.Internal.Data.Stream.IsStream"
89+-- module for many more @Pre-release@ combinators. See the
90+-- <https://github.com/composewell/streamly-examples> repository for many more
91+-- real world examples of stream programming.
92+--
93+-- == Polymorphic Combinators
94+--
95+-- Streamly has several stream types, 'SerialT' is one type of stream with
96+-- serial execution of actions, 'AsyncT' is another with concurrent execution.
97+-- The combinators in this module are polymorphic in stream type. For example,
98+--
99+-- @
100+-- repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
101+-- @
102+--
103+-- @t@ is the stream type, @m@ is the underlying 'Monad' of the stream (e.g.
104+-- IO) and @a@ is the type of elements in the stream (e.g. Int).
105+--
106+-- Stream elimination combinators accept a 'SerialT' type instead of a
107+-- polymorphic type to force a concrete monomorphic type by default, reducing
108+-- type errors. That's why in the console echo example above the stream type is
109+-- 'SerialT'.
110+--
111+-- @
112+-- drain :: Monad m => SerialT m a -> m ()
113+-- @
114+--
115+-- We can force a certain stream type in polymorphic code by using "Stream Type
116+-- Adaptors". For example, to force 'AsyncT':
117+--
118+-- >>> Stream.drain $ Stream.asyncly $ Stream.replicateM 10 $ delay 1
119+-- ...
120+--
121+-- == Combining two streams
122+--
123+-- Two streams can be combined to form a single stream in various interesting
124+-- ways. 'serial' (append), 'wSerial' (interleave), 'ahead' (concurrent,
125+-- ordered append), 'async' (lazy concurrent, unordered append) , 'wAsync'
126+-- (lazy concurrent, unordered interleave), 'parallel' (strict concurrent
127+-- merge), 'zipWith', 'zipAsyncWith' (concurrent zip), 'mergeBy',
128+-- 'mergeAsyncBy' (concurrent merge) are some ways of combining two streams.
129+--
130+-- For example, the `parallel` combinator schedules both the streams
131+-- concurrently.
132+--
133+-- >>> stream1 = Stream.fromListM [delay 3, delay 4]
134+-- >>> stream2 = Stream.fromListM [delay 1, delay 2]
135+-- >>> Stream.toList $ stream1 `parallel` stream2
136+-- ...
137+--
138+-- We can chain the operations to combine more than two streams:
139+--
140+-- >>> stream3 = Stream.fromListM [delay 1, delay 2]
141+-- >>> Stream.toList $ stream1 `parallel` stream2 `parallel` stream3
142+-- ...
143+--
144+-- Concurrent generation ('consM') and concurrent merging of streams is the
145+-- fundamental basis of all concurrency in streamly.
146+--
147+-- == Combining many streams
148+--
149+-- The 'concatMapWith' combinator can be used to generalize the two stream
150+-- combining combinators to @n@ streams. For example, we can use
151+-- @concatMapWith parallel@ to read concurrently from all incoming network
152+-- connections and combine the input streams into a single output stream:
153+--
154+-- @
155+-- import qualified Streamly.Network.Inet.TCP as TCP
156+-- import qualified Streamly.Network.Socket as Socket
157+--
158+-- Stream.unfold TCP.acceptOnPort 8090
159+-- & Stream.concatMapWith Stream.parallel (Stream.unfold Socket.read)
160+-- @
161+--
162+-- See the @streamly-examples@ repository for a full working example.
163+--
164+-- == Concurrent Nested Loops
165+--
166+-- The Monad instance of 'SerialT' is an example of nested looping. It is in
167+-- fact a list transformer. Different stream types provide different variants
168+-- of nested looping. For example, the 'Monad' instance of 'ParallelT' uses
169+-- @concatMapWith parallel@ as its bind operation. Therefore, each iteration of
170+-- the loop for 'ParallelT' stream can run concurrently. See the documentation
171+-- for individual stream types for the specific execution behavior of the
172+-- stream as well as the behavior of 'Semigroup' and 'Monad' instances.
83173--
84174-- == Stream Types
85175--
86−-- One way in which stream types are distinguished is how the sequence of
87−-- actions forming the stream are executed. Below we will see how 'SerialT',
88−-- 'AsyncT' and 'AheadT' stream types differ from each other.
176+-- Streamly has several stream types. These types differ in three fundamental
177+-- operations, 'consM' ('IsStream' instance), '<>' ('Semigroup' instance) and
178+-- '>>=' ('Monad' instance). Below we will see how 'consM' behaves for
179+-- 'SerialT', 'AsyncT' and 'AheadT' stream types.
89180--
90181-- 'SerialT' executes actions serially, so the total delay in the following
91182-- example is @2 + 1 = 3@ seconds:
@@ -116,55 +207,7 @@
116207-- 2 sec
117208-- [2,1]
118209--
119−-- == Concurrent Combinators
120−--
121−-- Like 'consM', there are several other stream generation operations whose
122−-- execution behavior depends on the stream type, they all follow behavior
123−-- similar to 'consM'.
124−--
125−-- By default, folds like 'drain' force the stream type to be 'SerialT', so
126−-- 'replicateM' in the following code runs serially, and takes 10 seconds:
127−--
128−-- >>> Stream.drain $ Stream.replicateM 10 $ delay 1
129−-- ...
130−--
131−-- We can use the 'asyncly' combinator to force the argument stream to be of
132−-- 'AsyncT' type, 'replicateM' in the following example executes the replicated
133−-- actions concurrently, thus taking only 1 second:
134−--
135−-- >>> Stream.drain $ Stream.asyncly $ Stream.replicateM 10 $ delay 1
136−-- ...
137−--
138−-- We can use 'mapM' to map an action concurrently:
139−--
140−-- >>> f x = delay 1 >> return (x + 1)
141−-- >>> Stream.toList $ Stream.aheadly $ Stream.mapM f $ Stream.fromList [1..3]
142−-- ...
143−-- [2,3,4]
144−--
145−-- 'aheadly' forces mapM to happen in 'AheadT' style, thus all three actions
146−-- take only one second even though each individual action blocks for a second.
147−--
148−-- See the documentation of individual combinators to check if it is concurrent
149−-- or not. The concurrent combinators necessarily have a @MonadAsync m@
150−-- constraint. However, a @MonadAsync m@ constraint does not necessarily mean
151−-- that the combinator is concurrent.
152−--
153−-- == Automatic Concurrency Control
154−--
155−-- 'SerialT' (and 'WSerialT') runs all tasks serially whereas 'ParallelT' runs
156−-- all tasks concurrently i.e. one thread per task. The stream types 'AsyncT',
157−-- 'WAsyncT', and 'AheadT' provide demand driven concurrency. It means that
158−-- based on the rate at which the consumer is consuming the stream, it
159−-- maintains the optimal number of threads to increase or decrease parallelism.
160−--
161−-- However, the programmer can control the maximum number of threads using
162−-- 'maxThreads'. It provides an upper bound on the concurrent IO requests or
163−-- CPU cores that can be used. 'maxBuffer' limits the number of evaluated
164−-- stream elements that we can buffer. See the "Concurrency Control" section
165−-- for details.
166−--
167−-- == Combining Two streams
210+-- == Semigroup Instance
168211--
169212-- Earlier we distinguished stream types based on the execution behavior of
170213-- actions within a stream. Stream types are also distinguished based on how
@@ -202,41 +245,53 @@
202245-- 'wSerial' respectively. The 'serial' combinator combines two streams of any
203246-- type in the same way as a serial stream combines.
204247--
205−-- == Combining N streams
248+-- == Concurrent Combinators
206249--
207−-- The 'concatMapWith' combinator can be used to generalize the two stream
208−-- combining combinators to @n@ streams.
250+-- Like 'consM', there are several other stream generation operations whose
251+-- execution behavior depends on the stream type, they all follow behavior
252+-- similar to 'consM'.
209253--
210−-- Let's take an example of the `parallel` combinator ('<>' operation of
211−-- 'ParallelT'), it schedules both the streams concurrently.
254+-- By default, folds like 'drain' force the stream type to be 'SerialT', so
255+-- 'replicateM' in the following code runs serially, and takes 10 seconds:
212256--
213−-- >>> stream1 = Stream.fromListM [delay 3, delay 4]
214−-- >>> stream2 = Stream.fromListM [delay 1, delay 2]
215−-- >>> Stream.toList $ stream1 `parallel` stream2
257+-- >>> Stream.drain $ Stream.replicateM 10 $ delay 1
216258-- ...
217259--
218−-- We can use @concatMapWith parallel@ to read concurrently from all incoming
219−-- network connections and combine the input streams into a single output
220−-- stream:
260+-- We can use the 'asyncly' combinator to force the argument stream to be of
261+-- 'AsyncT' type, 'replicateM' in the following example executes the replicated
262+-- actions concurrently, thus taking only 1 second:
221263--
222−-- @
223−-- import qualified Streamly.Network.Inet.TCP as TCP
224−-- import qualified Streamly.Network.Socket as Socket
264+-- >>> Stream.drain $ Stream.asyncly $ Stream.replicateM 10 $ delay 1
265+-- ...
225266--
226−-- Stream.unfold TCP.acceptOnPort 8090
227−-- & Stream.concatMapWith Stream.parallel (Stream.unfold Socket.read)
228−-- @
267+-- We can use 'mapM' to map an action concurrently:
229268--
230−-- See the @streamly-examples@ repository for a full working example.
269+-- >>> f x = delay 1 >> return (x + 1)
270+-- >>> Stream.toList $ Stream.aheadly $ Stream.mapM f $ Stream.fromList [1..3]
271+-- ...
272+-- [2,3,4]
231273--
232−-- The 'Monad' instance of 'ParallelT' uses @concatMapWith parallel@ as its
233−-- bind operation. Therefore, the monad composition behaves in the same way.
234−-- The monad instance of stream types is essentially nested looping, it means
235−-- each iteration of the loop for 'ParallelT' stream can run concurrently.
274+-- 'aheadly' forces mapM to happen in 'AheadT' style, thus all three actions
275+-- take only one second even though each individual action blocks for a second.
236276--
237−-- See the documentation for individual stream types for the specific execution
238−-- behavior of the stream as well as the behavior of 'Semigroup' and 'Monad'
239−-- instances.
277+-- See the documentation of individual combinators to check if it is concurrent
278+-- or not. The concurrent combinators necessarily have a @MonadAsync m@
279+-- constraint. However, a @MonadAsync m@ constraint does not necessarily mean
280+-- that the combinator is concurrent.
281+--
282+-- == Automatic Concurrency Control
283+--
284+-- 'SerialT' (and 'WSerialT') runs all tasks serially whereas 'ParallelT' runs
285+-- all tasks concurrently i.e. one thread per task. The stream types 'AsyncT',
286+-- 'WAsyncT', and 'AheadT' provide demand driven concurrency. It means that
287+-- based on the rate at which the consumer is consuming the stream, it
288+-- maintains the optimal number of threads to increase or decrease parallelism.
289+--
290+-- However, the programmer can control the maximum number of threads using
291+-- 'maxThreads'. It provides an upper bound on the concurrent IO requests or
292+-- CPU cores that can be used. 'maxBuffer' limits the number of evaluated
293+-- stream elements that we can buffer. See the "Concurrency Control" section
294+-- for details.
240295--
241296-- == Caveats
242297--
@@ -250,22 +305,6 @@
250305-- Stream.asyncly $ ... concurrent combinator here ... $ Stream.serially $ ...
251306-- @
252307--
253−-- == Polymorphic Combinators
254−--
255−-- The combinators in this module are polymorphic in stream type. The stream is
256−-- usually specified as @t m a@ where the stream type @t@ is a member of the
257−-- 'IsStream' type class (e.g. 'SerialT' or 'AsyncT'), @m@ is the underlying
258−-- 'Monad' of the stream (e.g. IO) and @a@ is the type of elements in the
259−-- stream (e.g. Int).
260−--
261−-- Deconstruction and folding combinators accept a 'SerialT' type instead of a
262−-- polymorphic type to ensure that streams always have a concrete monomorphic
263−-- type by default, reducing type errors.
264−--
265−-- In case you want to adapt a stream type to another stream type you can use
266−-- one of the type adapting combinators provided in the section "Stream Type
267−-- Adaptors".
268−--
269308-- == Conventions
270309--
271310-- Functions with the suffix @M@ are general functions that work on monadic