@@ -18,23 +18,8 @@
19-- We will add some more imports in the examples as needed.
19-- We will add some more imports in the examples as needed.
23−-- Streamly is a framework for modular data flow based programming and
24−-- declarative concurrency. Streamly API is similar to Haskell lists.
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:
31−-- >>> import Streamly.Prelude (SerialT, cons, consM, nil)
32−-- >>> stream = 1 `cons` 2 `cons` nil :: SerialT IO Int
33−-- >>> Stream.toList stream -- IO [Int]
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
39-- >>> import Control.Concurrent (threadDelay)
24-- >>> import Control.Concurrent (threadDelay)
@@ -48,15 +33,34 @@
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.
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
48+-- `cons` constructs a pure stream which is more or less the same as a list:
50+-- >>> import Streamly.Prelude (SerialT, cons, consM, nil)
51+-- >>> stream = 1 `cons` 2 `cons` nil :: SerialT IO Int
52+-- >>> Stream.toList stream -- IO [Int]
51-- 'consM' constructs a stream from effectful actions:
55-- 'consM' constructs a stream from effectful actions:
54−-- > stream = delay 1 \`consM` delay 2 \`consM` 'nil'
57+-- >>> stream = delay 1 `consM` delay 2 `consM` nil
55−-- > Stream.'toList' stream
58+-- >>> Stream.toList stream
63+-- == Console Echo Program
61-- In the following example, 'repeatM' generates an infinite stream of 'String'
65-- In the following example, 'repeatM' generates an infinite stream of 'String'
62-- by repeatedly performing the 'getLine' IO action. 'mapM' then applies
66-- by repeatedly performing the 'getLine' IO action. 'mapM' then applies
@@ -74,18 +78,105 @@
77−-- This is a console echo program. It is an example of a declarative loop.
81+-- This is a console echo program. It is an example of a declarative loop
78−-- Compare it with an imperative @while@ loop.
82+-- written using streaming combinators. Compare it with an imperative @while@
80-- Hopefully, this gives you an idea how we can program declaratively by
85-- Hopefully, this gives you an idea how we can program declaratively by
81−-- represneting loops using streams. For more details please see the
86+-- representing loops using streams. In this module, you can find all
82−-- "Streamly.Tutorial" module.
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.
93+-- == Polymorphic Combinators
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,
100+-- repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
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).
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
112+-- drain :: Monad m => SerialT m a -> m ()
115+-- We can force a certain stream type in polymorphic code by using "Stream Type
116+-- Adaptors". For example, to force 'AsyncT':
118+-- >>> Stream.drain $ Stream.asyncly $ Stream.replicateM 10 $ delay 1
121+-- == Combining two streams
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.
130+-- For example, the `parallel` combinator schedules both the streams
133+-- >>> stream1 = Stream.fromListM [delay 3, delay 4]
134+-- >>> stream2 = Stream.fromListM [delay 1, delay 2]
135+-- >>> Stream.toList $ stream1 `parallel` stream2
138+-- We can chain the operations to combine more than two streams:
140+-- >>> stream3 = Stream.fromListM [delay 1, delay 2]
141+-- >>> Stream.toList $ stream1 `parallel` stream2 `parallel` stream3
144+-- Concurrent generation ('consM') and concurrent merging of streams is the
145+-- fundamental basis of all concurrency in streamly.
147+-- == Combining many streams
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:
155+-- import qualified Streamly.Network.Inet.TCP as TCP
156+-- import qualified Streamly.Network.Socket as Socket
158+-- Stream.unfold TCP.acceptOnPort 8090
159+-- & Stream.concatMapWith Stream.parallel (Stream.unfold Socket.read)
162+-- See the @streamly-examples@ repository for a full working example.
164+-- == Concurrent Nested Loops
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.
84-- == Stream Types
174-- == Stream Types
86−-- One way in which stream types are distinguished is how the sequence of
176+-- Streamly has several stream types. These types differ in three fundamental
87−-- actions forming the stream are executed. Below we will see how 'SerialT',
177+-- operations, 'consM' ('IsStream' instance), '<>' ('Semigroup' instance) and
178+-- '>>=' ('Monad' instance). Below we will see how 'consM' behaves for
88−-- 'AsyncT' and 'AheadT' stream types differ from each other.
179+-- 'SerialT', 'AsyncT' and 'AheadT' stream types.
90-- 'SerialT' executes actions serially, so the total delay in the following
181-- 'SerialT' executes actions serially, so the total delay in the following
91-- example is @2 + 1 = 3@ seconds:
182-- example is @2 + 1 = 3@ seconds:
@@ -116,55 +207,7 @@
119−-- == Concurrent Combinators
210+-- == Semigroup Instance
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'.
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:
128−-- >>> Stream.drain $ Stream.replicateM 10 $ delay 1
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:
135−-- >>> Stream.drain $ Stream.asyncly $ Stream.replicateM 10 $ delay 1
138−-- We can use 'mapM' to map an action concurrently:
140−-- >>> f x = delay 1 >> return (x + 1)
141−-- >>> Stream.toList $ Stream.aheadly $ Stream.mapM f $ Stream.fromList [1..3]
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.
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.
153−-- == Automatic Concurrency Control
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.
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
167−-- == Combining Two streams
169-- Earlier we distinguished stream types based on the execution behavior of
212-- Earlier we distinguished stream types based on the execution behavior of
170-- actions within a stream. Stream types are also distinguished based on how
213-- actions within a stream. Stream types are also distinguished based on how
@@ -202,41 +245,53 @@
202-- 'wSerial' respectively. The 'serial' combinator combines two streams of any
245-- 'wSerial' respectively. The 'serial' combinator combines two streams of any
203-- type in the same way as a serial stream combines.
246-- type in the same way as a serial stream combines.
205−-- == Combining N streams
248+-- == Concurrent Combinators
207−-- The 'concatMapWith' combinator can be used to generalize the two stream
250+-- Like 'consM', there are several other stream generation operations whose
208−-- combining combinators to @n@ streams.
251+-- execution behavior depends on the stream type, they all follow behavior
252+-- similar to 'consM'.
210−-- Let's take an example of the `parallel` combinator ('<>' operation of
254+-- By default, folds like 'drain' force the stream type to be 'SerialT', so
211−-- 'ParallelT'), it schedules both the streams concurrently.
255+-- 'replicateM' in the following code runs serially, and takes 10 seconds:
213−-- >>> stream1 = Stream.fromListM [delay 3, delay 4]
257+-- >>> Stream.drain $ Stream.replicateM 10 $ delay 1
214−-- >>> stream2 = Stream.fromListM [delay 1, delay 2]
215−-- >>> Stream.toList $ stream1 `parallel` stream2
218−-- We can use @concatMapWith parallel@ to read concurrently from all incoming
260+-- We can use the 'asyncly' combinator to force the argument stream to be of
219−-- network connections and combine the input streams into a single output
261+-- 'AsyncT' type, 'replicateM' in the following example executes the replicated
220−-- stream:
262+-- actions concurrently, thus taking only 1 second:
222−-- @
264+-- >>> Stream.drain $ Stream.asyncly $ Stream.replicateM 10 $ delay 1
223−-- import qualified Streamly.Network.Inet.TCP as TCP
265+-- ...
224−-- import qualified Streamly.Network.Socket as Socket
226−-- Stream.unfold TCP.acceptOnPort 8090
267+-- We can use 'mapM' to map an action concurrently:
227−-- & Stream.concatMapWith Stream.parallel (Stream.unfold Socket.read)
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]
232−-- The 'Monad' instance of 'ParallelT' uses @concatMapWith parallel@ as its
274+-- 'aheadly' forces mapM to happen in 'AheadT' style, thus all three actions
233−-- bind operation. Therefore, the monad composition behaves in the same way.
275+-- take only one second even though each individual action blocks for a second.
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.
237−-- See the documentation for individual stream types for the specific execution
277+-- See the documentation of individual combinators to check if it is concurrent
238−-- behavior of the stream as well as the behavior of 'Semigroup' and 'Monad'
278+-- or not. The concurrent combinators necessarily have a @MonadAsync m@
239−-- instances.
279+-- constraint. However, a @MonadAsync m@ constraint does not necessarily mean
280+-- that the combinator is concurrent.
282+-- == Automatic Concurrency Control
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.
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
241-- == Caveats
296-- == Caveats
@@ -250,22 +305,6 @@
250-- Stream.asyncly $ ... concurrent combinator here ... $ Stream.serially $ ...
305-- Stream.asyncly $ ... concurrent combinator here ... $ Stream.serially $ ...
253−-- == Polymorphic Combinators
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).
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.
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
269-- == Conventions
308-- == Conventions
271-- Functions with the suffix @M@ are general functions that work on monadic
310-- Functions with the suffix @M@ are general functions that work on monadic