@@ -30,26 +30,27 @@ and declarative monadic concurrency.
31## Where to use streamly?
31## Where to use streamly?
33−Streamly is a general purpose programming framwework. It can be used equally
33+Streamly is a general purpose programming framework. It can be used equally
34efficiently from a simple `Hello World!` program to a massively concurrent
34efficiently from a simple `Hello World!` program to a massively concurrent
35application. The answer to the question, "where to use streamly?" - would be
35application. The answer to the question, "where to use streamly?" - would be
36similar to the answer to - "Where to use Haskell lists or the IO monad?".
36similar to the answer to - "Where to use Haskell lists or the IO monad?".
37−Streamly generalizes lists to monadic streams, and the `IO` monad to
37+
38−non-deterministic and concurrent stream composition. The `IO` monad is a
38+Streamly simplifies streaming and makes it as intuitive as plain lists. Unlike
39−special case of streamly; if we use single element streams the behavior of
39+other streaming libraries, no fancy types are required. Streamly is simply a
40−streamly becomes identical to the IO monad. The IO monad code can be replaced
40+generalization of Haskell lists to monadic streaming optionally with concurrent
41−with streamly by just prefixing the IO actions with `liftIO`, without any other
41+composition. Streamly stream type can be considered as a list type
42−changes, and without any loss of performance. Pure lists too are a special
42+parameterized by a monad. For example, `SerialT IO a` is an equivalent of `[a]`
43−case of streamly; if we use `Identity` as the underlying monad, streamly
43+in the IO monad. A stream in an `Identity` monad, `SerialT Identity a`, is
44−streams turn into pure lists. Non-concurrent programs are just a special case
44+equivalent to pure lists with equal or better performance. Streams are
45−of concurrent ones, simply adding a combinator turns a non-concurrent program
45+constructed just like lists are constructed, using `nil` and `cons` instead of
46−into a concurrent one.
46+`[]` and `:`. Unlike lists, streams can be constructed from monadic actions,
47−
47+not just pure elements. Streams are processed just like lists are processed.
48−In other words, streamly combines the functionality of lists and IO, with
48+Streamly provides all the list combinators and more, but they are monadic and
49−builtin concurrency. If you want to write a program that involves IO,
49+work in a streaming fashion. In other words streamly just completes what lists
50−concurrent or not, then you can just use streamly as the base monad, in fact,
50+lack, you do not need to learn anything new.
51−you could even use streamly for pure computations, as streamly performs at par
51+
52−with pure lists or `vector`.
52+Not surprisingly, the monad instance of streamly is a list transformer, with
53+concurrency capability.
54## Why data flow programming?
55## Why data flow programming?
@@ -71,49 +72,37 @@ for instructions on how to use streamly with your Haskell build tool or package
71manager. You may want to go through it before jumping to run the examples
72manager. You may want to go through it before jumping to run the examples
75+The module `Streamly` provides just the core stream types, type casting and
76+concurrency control combinators. Stream construction, transformation, folding,
77+merging, zipping combinators are found in `Streamly.Prelude`.
74## Show me an example
79## Show me an example
76−Here is an IO monad code to list a directory recursively:
81+The following code snippet lists a directory tree recursively, reading multiple
82+directories concurrently:
79import Control.Monad.IO.Class (liftIO)
85import Control.Monad.IO.Class (liftIO)
80import Path.IO (listDir, getCurrentDir) -- from path-io package
86import Path.IO (listDir, getCurrentDir) -- from path-io package
87+import Streamly (AsyncT, adapt)
88+import qualified Streamly.Prelude as S
82−listDirRecursive = getCurrentDir >>= readdir
90+listDirRecursive :: AsyncT IO ()
91+listDirRecursive = getCurrentDir >>= readdir >>= liftIO . mapM_ putStrLn
84 readdir dir = do
93 readdir dir = do
85 (dirs, files) <- listDir dir
94 (dirs, files) <- listDir dir
86− liftIO $ mapM_ putStrLn
87− $ map show dirs ++ map show files
95+ S.yield (map show dirs ++ map show files) <> foldMap readdir dirs
91−This is your usual IO monad code, with no streamly specific code whatsoever.
92−This is how you can run this:
96−main = listDirRecursive
99−And, this is how you can run exactly the same code using streamly with
100−lookahead style concurrency, the only difference is that this time multiple
101−directories are read concurrently:
104−import Streamly (aheadly)
105−import qualified Streamly.Prelude as S
107main :: IO ()
97main :: IO ()
108−main = S.drain $ aheadly $ listDirRecursive
98+main = S.drain $ adapt $ listDirRecursive
111−Isn't that magical? What's going on here? Streamly does not introduce any new
101+`AsyncT` is a stream monad transformer. If you are familiar with a list
112−abstractions, it just uses standard abstractions like `Semigroup` or
102+transformer, it is nothing but `ListT` with concurrency semantics. For example,
113−`Monoid` to combine monadic streams concurrently, the way lists combine a
103+the semigroup operation `<>` is concurrent. This makes `foldMap` concurrent
114−sequence of pure values non-concurrently. The `foldMap` in the code
104+too. You can replace `AsyncT` with `SerialT` and the above code will become
115−above turns into a concurrent monoidal composition of a stream of `readdir`
105+serial, exactly equivalent to a `ListT`.
118## How does it perform?
107## How does it perform?
@@ -219,33 +208,50 @@ composes stream data instead of stream processors (functions). A stream is
219just like a list and is explicitly passed around to functions that process the
208just like a list and is explicitly passed around to functions that process the
220stream. Therefore, no special operator is needed to join stages in a streaming
209stream. Therefore, no special operator is needed to join stages in a streaming
221pipeline, just the standard function application (`$`) or reverse function
210pipeline, just the standard function application (`$`) or reverse function
222−application (`&`) operator is enough. Combinators are provided in
211+application (`&`) operator is enough.
223−`Streamly.Prelude` to transform or fold streams.
225## Concurrent Stream Generation
213## Concurrent Stream Generation
227−Monadic construction and generation functions e.g. `consM`, `unfoldrM`,
215+`consM` or its operator form `|:` can be used to construct a stream from
228−`replicateM`, `repeatM`, `iterateM` and `fromFoldableM` etc. work concurrently
216+monadic actions. A stream constructed with `consM` can run the monadic actions
229−when used with appropriate stream type combinator (e.g. `asyncly`, `aheadly` or
217+in the stream concurrently when used with appropriate stream type combinator
230−`parallely`).
218+(e.g. `asyncly`, `aheadly` or `parallely`).
232−The following code finishes in 3 seconds (6 seconds when serial):
220+The following code finishes in 3 seconds (6 seconds when serial), note the
221+order of elements in the resulting output, the outputs are consumed as soon as
222+each action is finished (asyncly):
234``` haskell
224``` haskell
235> let p n = threadDelay (n * 1000000) >> return n
225> let p n = threadDelay (n * 1000000) >> return n
226+> S.toList $ asyncly $ p 3 |: p 2 |: p 1 |: S.nil
230+Use `aheadly` if you want speculative concurrency i.e. execute the actions in
231+the stream concurrently but consume the results in the specified order:
236> S.toList $ aheadly $ p 3 |: p 2 |: p 1 |: S.nil
234> S.toList $ aheadly $ p 3 |: p 2 |: p 1 |: S.nil
239−> S.toList $ parallely $ p 3 |: p 2 |: p 1 |: S.nil
238+Monadic stream generation functions e.g. `unfoldrM`, `replicateM`, `repeatM`,
239+`iterateM` and `fromFoldableM` etc. can work concurrently.
243The following finishes in 10 seconds (100 seconds when serial):
241The following finishes in 10 seconds (100 seconds when serial):
245``` haskell
243``` haskell
246S.drain $ asyncly $ S.replicateM 10 $ p 10
244S.drain $ asyncly $ S.replicateM 10 $ p 10
247+## Concurrency Auto Scaling
249+Concurrency is auto-scaled i.e. more actions are executed concurrently if the
250+consumer is consuming the stream at a higher speed. How many tasks are executed
251+concurrently can be controlled by `maxThreads` and how many results are
252+buffered ahead of consumption can be controlled by `maxBuffer`. See the
253+documentation in the `Streamly` module.
249## Concurrent Streaming Pipelines
255## Concurrent Streaming Pipelines
251Use `|&` or `|$` to apply stream processing functions concurrently. The
257Use `|&` or `|$` to apply stream processing functions concurrently. The
@@ -344,43 +350,18 @@ main = S.drain loops
345## Concurrent Nested Loops
351## Concurrent Nested Loops
347−To run the above code with, lookahead style concurrency i.e. each iteration in
353+To run the above code with speculative concurrency i.e. each iteration in the
348−the loop can run run concurrently by but the results are presented in the same
354+loop can run concurrently but the results are presented to the consumer of the
349−order as serial execution:
355+output in the same order as serial execution:
351``` haskell
357``` haskell
352main = S.drain $ aheadly $ loops
358main = S.drain $ aheadly $ loops
355−To run it with depth first concurrency yielding results asynchronously in the
361+Different stream types execute the loop iterations in different ways. For
356−same order as they become available (deep async composition):
362+example, `wSerially` interleaves the loop iterations. There are several
357−
363+concurrent stream styles to execute the loop iterations concurrently in
358−``` haskell
364+different ways, see the `Streamly.Tutorial` module for a detailed treatment.
359−main = S.drain $ asyncly $ loops
362−To run it with breadth first concurrency and yeilding results asynchronously
363−(wide async composition):
366−main = S.drain $ wAsyncly $ loops
369−The above streams provide lazy/demand-driven concurrency which is automatically
370−scaled as per demand and is controlled/bounded so that it can be used on
371−infinite streams. The following combinator provides strict, unbounded
372−concurrency irrespective of demand:
375−main = S.drain $ parallely $ loops
378−To run it serially but interleaving the outer and inner loop iterations
379−(breadth first serial):
382−main = S.drain $ wSerially $ loops
385## Magical Concurrency
366## Magical Concurrency
@@ -445,7 +426,6 @@ asynchronous exceptions to deal with. You can just ignore the zillions of
445blogs, talks, caveats about async exceptions. Async exceptions just don't
426blogs, talks, caveats about async exceptions. Async exceptions just don't
446exist. Please don't use things like `myThreadId` and `throwTo` just for fun!
427exist. Please don't use things like `myThreadId` and `throwTo` just for fun!
449## Reactive Programming (FRP)
429## Reactive Programming (FRP)
451Streamly is a foundation for first class reactive programming as well by virtue
431Streamly is a foundation for first class reactive programming as well by virtue