3[](https://hackage.haskell.org/package/streamly)
4[](https://gitter.im/composewell/streamly)
5[](https://travis-ci.org/composewell/streamly)
6[](https://ci.appveyor.com/project/harendra-kumar/streamly)
7[](https://coveralls.io/github/composewell/streamly?branch=master)
9## Stream`ing` `Concurrent`ly
11Streamly, short for streaming concurrently, provides monadic streams, with a
12simple API, almost identical to standard lists, and an in-built support for
13concurrency. By using stream-style combinators on stream composition,
14streams can be generated, merged, chained, mapped, zipped, and consumed
15concurrently – providing a generalized high level programming framework
16unifying streaming and concurrency. Controlled concurrency allows even infinite
17streams to be evaluated concurrently. Concurrency is auto scaled based on
18feedback from the stream consumer. The programmer does not have to be aware of
19threads, locking or synchronization to write scalable concurrent programs.
21The basic streaming functionality of streamly is equivalent to that provided by
22streaming libraries like
23[vector](https://hackage.haskell.org/package/vector),
24[streaming](https://hackage.haskell.org/package/streaming),
25[pipes](https://hackage.haskell.org/package/pipes), and
26[conduit](https://hackage.haskell.org/package/conduit).
27In addition to providing streaming functionality, streamly subsumes
28the functionality of list transformer libraries like `pipes` or
29[list-t](https://hackage.haskell.org/package/list-t), and also the logic
30programming library [logict](https://hackage.haskell.org/package/logict). On
31the concurrency side, it subsumes the functionality of the
32[async](https://hackage.haskell.org/package/async) package. Because it supports
33streaming with concurrency we can write FRP applications similar in concept to
34[Yampa](https://hackage.haskell.org/package/Yampa) or
35[reflex](https://hackage.haskell.org/package/reflex).
39 * _Simplicity_: Simple list like streaming API, if you know how to use lists
40 then you know how to use streamly. This library is built with simplicity
41 and ease of use as a design goal.
42 * _Concurrency_: Simple, powerful, and scalable concurrency. Concurrency is
43 built-in, and not intrusive, concurrent programs are written exactly the
44 same way as non-concurrent ones.
45 * _Generality_: Unifies functionality provided by several disparate packages
46 (streaming, concurrency, list transformer, logic programming, reactive
47 programming) in a concise API.
48 * _Performance_: Streamly is designed for high performance. It employs stream
49 fusion optimizations for best possible performance. Serial peformance is
50 equivalent to the venerable `vector` library in most cases and even better
51 in some cases. Concurrent performance is unbeatable. See
52 [streaming-benchmarks](https://github.com/composewell/streaming-benchmarks)
53 for a comparison of popular streaming libraries on micro-benchmarks.
55For more details on streaming library ecosystem and where streamly fits in,
57[streaming libraries](https://github.com/composewell/streaming-benchmarks#streaming-libraries).
58Also, see the [Comparison with Existing
59Packages](https://hackage.haskell.org/package/streamly/docs/Streamly-Tutorial.html)
60section in the streamly tutorial.
62For more information on streamly, see:
64 * [Streamly.Tutorial](https://hackage.haskell.org/package/streamly/docs/Streamly-Tutorial.html) module in the haddock documentation for a detailed introduction
65 * [examples](https://github.com/composewell/streamly/tree/master/examples) directory in the package for some simple practical examples
69Unlike `pipes` or `conduit` and like `vector` and `streaming`, `streamly`
70composes stream data instead of stream processors (functions). A stream is
71just like a list and is explicitly passed around to functions that process the
72stream. Therefore, no special operator is needed to join stages in a streaming
73pipeline, just the standard function application (`$`) or reverse function
74application (`&`) operator is enough. Combinators are provided in
75`Streamly.Prelude` to transform or fold streams.
77The following snippet provides a simple stream composition example that reads
78numbers from stdin, prints the squares of even numbers and exits if an even
79number more than 9 is entered.
83import qualified Streamly.Prelude as S
84import Data.Function ((&))
95## Concurrent Stream Generation
97Monadic construction and generation functions e.g. `consM`, `unfoldrM`,
98`replicateM`, `repeatM`, `iterateM` and `fromFoldableM` etc. work concurrently
99when used with appropriate stream type combinator (e.g. `asyncly`, `aheadly` or
102The following code finishes in 3 seconds (6 seconds when serial):
105> let p n = threadDelay (n * 1000000) >> return n
106> S.toList $ aheadly $ p 3 |: p 2 |: p 1 |: S.nil
109> S.toList $ parallely $ p 3 |: p 2 |: p 1 |: S.nil
113The following finishes in 10 seconds (100 seconds when serial):
116runStream $ asyncly $ S.replicateM 10 $ p 10
119## Concurrent Streaming Pipelines
121Use `|&` or `|$` to apply stream processing functions concurrently. The
122following example prints a "hello" every second; if you use `&` instead of
123`|&` you will see that the delay doubles to 2 seconds instead because of serial
128 S.repeatM (threadDelay 1000000 >> return "hello")
129 |& S.mapM (\x -> threadDelay 1000000 >> putStrLn x)
132## Mapping Concurrently
134We can use `mapM` or `sequence` functions concurrently on a stream.
137> let p n = threadDelay (n * 1000000) >> return n
138> runStream $ aheadly $ S.mapM (\x -> p 1 >> print x) (serially $ repeatM (p 1))
141## Serial and Concurrent Merging
143Semigroup and Monoid instances can be used to fold streams serially or
144concurrently. In the following example we compose ten actions in the
145stream, each with a delay of 1 to 10 seconds, respectively. Since all the
146actions are concurrent we see one output printed every second:
150import qualified Streamly.Prelude as S
151import Control.Concurrent (threadDelay)
153main = S.toList $ parallely $ foldMap delay [1..10]
154 where delay n = S.yieldM $ threadDelay (n * 1000000) >> print n
157Streams can be combined together in many ways. We provide some examples
158below, see the tutorial for more ways. We use the following `delay`
159function in the examples to demonstrate the concurrency aspects:
163import qualified Streamly.Prelude as S
164import Control.Concurrent
166delay n = S.yieldM $ do
167 threadDelay (n * 1000000)
169 putStrLn (show tid ++ ": Delay " ++ show n)
174main = runStream $ delay 3 <> delay 2 <> delay 1
185main = runStream . parallely $ delay 3 <> delay 2 <> delay 1
193## Nested Loops (aka List Transformer)
195The monad instance composes like a list monad.
199import qualified Streamly.Prelude as S
202 x <- S.fromFoldable [1,2]
203 y <- S.fromFoldable [3,4]
204 S.yieldM $ putStrLn $ show (x, y)
206main = runStream loops
215## Concurrent Nested Loops
217To run the above code with, lookahead style concurrency i.e. each iteration in
218the loop can run run concurrently by but the results are presented in the same
219order as serial execution:
222main = runStream $ aheadly $ loops
225To run it with depth first concurrency yielding results asynchronously in the
226same order as they become available (deep async composition):
229main = runStream $ asyncly $ loops
232To run it with breadth first concurrency and yeilding results asynchronously
233(wide async composition):
236main = runStream $ wAsyncly $ loops
239The above streams provide lazy/demand-driven concurrency which is automatically
240scaled as per demand and is controlled/bounded so that it can be used on
241infinite streams. The following combinator provides strict, unbounded
242concurrency irrespective of demand:
245main = runStream $ parallely $ loops
248To run it serially but interleaving the outer and inner loop iterations
249(breadth first serial):
252main = runStream $ wSerially $ loops
255## Magical Concurrency
257Streams can perform semigroup (<>) and monadic bind (>>=) operations
258concurrently using combinators like `asyncly`, `parallelly`. For example,
259to concurrently generate squares of a stream of numbers and then concurrently
260sum the square roots of all combinations of two streams:
264import qualified Streamly.Prelude as S
267 s <- S.sum $ asyncly $ do
268 -- Each square is performed concurrently, (<>) is concurrent
269 x2 <- foldMap (\x -> return $ x * x) [1..100]
270 y2 <- foldMap (\y -> return $ y * y) [1..100]
271 -- Each addition is performed concurrently, monadic bind is concurrent
272 return $ sqrt (x2 + y2)
276Of course, the actions running in parallel could be arbitrary IO actions. For
277example, to concurrently list the contents of a directory tree recursively:
280import Path.IO (listDir, getCurrentDir)
282import qualified Streamly.Prelude as S
284main = runStream $ aheadly $ getCurrentDir >>= readdir
286 (dirs, files) <- S.yieldM $ listDir d
287 S.yieldM $ mapM_ putStrLn $ map show files
288 -- read the subdirs concurrently, (<>) is concurrent
292In the above examples we do not think in terms of threads, locking or
293synchronization, rather we think in terms of what can run in parallel, the rest
294is taken care of automatically. When using `aheadly` the programmer does
295not have to worry about how many threads are to be created, they are
296automatically adjusted based on the demand of the consumer.
298The concurrency facilities provided by streamly can be compared with
299[OpenMP](https://en.wikipedia.org/wiki/OpenMP) and
300[Cilk](https://en.wikipedia.org/wiki/Cilk) but with a more declarative
303## Reactive Programming (FRP)
305Streamly is a foundation for first class reactive programming as well by virtue
306of integrating concurrency and streaming. See
307[AcidRain.hs](https://github.com/composewell/streamly/tree/master/examples/AcidRain.hs)
308for a console based FRP game example and
309[CirclingSquare.hs](https://github.com/composewell/streamly/tree/master/examples/CirclingSquare.hs)
310for an SDL based animation example.
314`Streamly` has best in class performance even though it generalizes streaming
315to concurrent composition that does not mean it sacrifices non-concurrent
317[streaming-benchmarks](https://github.com/composewell/streaming-benchmarks) for
318detailed performance comparison with regular streaming libraries and the
319explanation of the benchmarks. The following graphs show a summary, the first
320one measures how four pipeline stages in a series perform, the second one
321measures the performance of individual stream operations; in both cases the
322stream processes a million elements:
324
325
329The code is available under BSD-3 license
330[on github](https://github.com/composewell/streamly). Join the
331[gitter chat](https://gitter.im/composewell/streamly) channel for discussions.
332You can find some of the
333[todo items on the github wiki](https://github.com/composewell/streamly/wiki/Things-To-Do).
334Please ask on the gitter channel or [contact the maintainer directly](mailto:harendra.kumar@gmail.com)
335for more details on each item. All contributions are welcome!
337This library was originally inspired by the `transient` package authored by
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