git.iko.soy/streamly
on aaa775b help sign in
streamly/docs/User/HowTo/streamly-vs-async.md
235 lines 7.9 kB history raw
1# Streamly
3Streamly is a library to make concurrent programming a joy. The venerable
4`async` package is the go to package for concurrent programming for most
5Haskellers. Streamly is a higher level library than `async` and provides a lot
6more power and functionality, using a simpler and concise expression of
7concurrency. At a high level, you should be able to express everything with
8streamly that you can with `async`, if you can't please raise an issue. If you
9are familiar with `async`, in this document we highlight how streamly can be
10used where you would use `async`.
12## `async/wait` vs Concurrent Streams
14Unlike `async`, streamly does not use a spawn and `wait` model. Streamly uses
15a more high level approach to concurrency and has no explicit notion of
16threads. In streamly, we compose multiple actions as a stream and then express
17whether you want to run the actions in the stream `serially` or `parallely`.
18There are many different ways in which you can run streams concurrently, see
19the reference documentation for details.
21Since there is no explicit notion of threads in streamly, there are no
22equivalents of `async`, `wait`, `cancel`, `poll` or `link` combinators from the
23`async` package.
25Since streamly is a monad transformer it can work with all monads and not just
26IO, you won't need adaptations like `lifted-async` to use it for a generic
27monad.
29## Using Streamly for Concurrency
31You can write all of your program in a streamly monad and use the full power of
32the library. Streamly can be used as a direct replacement of the IO monad with
33no loss of performance, and no change in code except using `liftIO` or `fromEffect`
34to run any IO actions. Streamly IO monads (e.g. `SerialT IO`) are just a
35generalization of the IO monad with non-deterministic composition of streams
36added on top.
38However, if you would like to just run only some concurrent portions of your
39program using streamly, you can do that too. Just use `drain` if you want
40to run the stream without collecting the outputs of the concurrent actions or
41use `toList` if you want to convert the output stream into a list. Other
42stream folding operations can also be used, see the docs for more details.
44## Features as Compared with `async`
46Use the following imports to run the snippets shown below:
48```haskell
49import Streamly
50import Streamly.Prelude ((|:))
51import qualified Streamly.Prelude as S
52import qualified Data.Text as Text
53import Control.Concurrent (threadDelay)
54```
56Let us simulate a URL fetch with a delay of `n` seconds using the following
57functions:
59```haskell
60getURL :: Int -> IO String
61getURL n = threadDelay (n * 1000000) >> return (show n)
62getURLString = getURL
63getURLText n = getURL n >>= return . Text.pack
64```
66### concurrently
68You can run any number of actions concurrently. For example, to fetch two URLs
69concurrently:
71```haskell
72 urls <- S.toList $ fromParallel $ getURL 2 |: getURL 1 |: S.nil
73```
75This would return the results in their arrival order i.e. first 1 and then 2.
76If you want to preserve the order of the results, use the lookahead style
77using `fromAhead` instead. In the following example both URLs are fetched
78concurrently, and even though URL 1 arrives before URL 2 the results will
79return 2 first and then 1.
81```haskell
82 urls <- S.toList $ fromAhead $ getURL 2 |: getURL 1 |: S.nil
83```
85### concurrently_
87Use `drain` instead of `toList` to run the actions but ignore the results:
89```haskell
90 S.drain $ fromParallel $ getURL 1 |: getURL 2 |: S.nil
91```
93### Concurrent Applicative
95If the actions that you are executing result in different output types you can
96use applicative zip to collect the results or to directly apply them to a
97function:
99```haskell
100 tuples <- S.toList $ fromZipAsync $
101 (,) <$> S.fromEffect (getURLString 1) <*> S.fromEffect (getURLText 2)
102```
104### race
106There are two ways to achieve the race functionality, using `take` or using
107exceptions.
109#### `race` Using `take`
111We can run multiple actions concurrently and take the first result that
112arrives:
114```haskell
115 urls <- S.toList $ S.take 1 $ fromParallel $ getURL 1 |: getURL 2 |: S.nil
116```
118After the first result arrives, the rest of the actions are canceled
119automatically. In general, we can take first `n` results as they arrive:
121```haskell
122 urls <- S.toList $ S.take 2 $ fromParallel $ getURL 1 |: getURL 2 |: S.nil
123```
125#### `race` Using Exceptions
127When an exception occurs in a concurrent stream all the concurrently running
128actions are cacnceled on arrival of the exception. This can be used to
129implement the race functionality. Each action in the stream can use an
130exception to communicate the result. As soon as the first result arrives all
131other actions will be canceled, for example:
133```haskell
134 data Result = Result String deriving Show
135 instance Exception Result
137 main = do
138 url <- try $ S.drain $ fromParallel $
139 (getURL 2 >>= throwM . Result)
140 |: (getURL 1 >>= throwM . Result)
141 |: S.nil
142 case url of
143 Left (e :: SomeException) -> print e
144 Right _ -> undefined
145```
147### mapConcurrently
149There are many ways to map concurrently on a container and collect the results:
151You can create a concurrent stream from a `Foldable` container of monadic
152actions:
154```haskell
155 urls <- S.toList $ fromAhead $ S.fromFoldableM $ fmap getURL [1..3]
156```
158You can first convert a `Foldable` into a stream and then map an action on the
159stream concurrently:
161```haskell
162 urls <- S.toList $ fromAhead $ S.mapM getURL $ foldMap return [1..3]
163```
165You can map a monadic action to a `Foldable` container to convert it into a
166stream and at the same time fold it:
168```haskell
169 urls <- S.toList $ fromAhead $ foldMap (S.fromEffect . getURL) [1..3]
170```
172### replicateConcurrently
174Streamly has not just the equivalent of `replicateConcurrently` which is
175`replicateM` but many more ways to generate concurrent streams, for example,
176`|:`, `unfoldrM`, `repeatM`, `iterateM`, `fromFoldableM` etc. See the
177[Streamly.Prelude](https://hackage.haskell.org/package/streamly/docs/Streamly-Prelude.html)
178module documentation for more details.
180```haskell
181 xs <- S.toList $ fromParallel $ S.replicateM 2 $ getURL 1
182```
184### Functor
186The stream resulting from concurrent actions can be mapped serially or
187concurrently.
189To map serially just use `fmap`:
191```haskell
192 xs <- S.toList $ fromParallel $ fmap (+1) $ return 1 |: return 2 |: S.nil
193```
195To map a monadic action concurrently on all elements of the stream use `mapM`:
197```haskell
198 xs <- S.toList $ fromParallel $ S.mapM (\x -> return (x + 1))
199 $ return 1 |: return 2 |: S.nil
200```
202### Semigroup
204The `Semigroup` instances of streamly merge multiple streams serially or
205concurrently.
207### Monad
209The `Monad` instances of streamly nest loops concurrently (concurrent
210non-determinism).
212### Performance
214Streamly has very little concurrency overhead (ranging from a few 100
215nanoseconds to a few microseconds on a 2.2 GHz Intel Core i7), you can even run
216very lightweight actions in parallel without worrying about the overhead of
217concurrency. See the performance benchmarks [comparing streamly with the `async`
218package in this repo](https://github.com/composewell/concurrency-benchmarks).
220## Further Reading
222There is much more that you can do with streamly. For example, you can use the
223`maxThreads` combinator to restrict the total number of concurrent threads or
224use the `maxBuffer` combinator to restrict the total number of bufferred
225results or you can use the `avgRate` combinator to control the rate at which
226the concurrent actions are executed.
228See the [haddock documentation on
229hackage](https://hackage.haskell.org/package/streamly) and [a comprehensive tutorial
230here](https://streamly.composewell.com/streamly-0.8.2/Tutorial.html).
232## References
234* https://hackage.haskell.org/package/async
235* https://hackage.haskell.org/package/lifted-async