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://circleci.com/gh/composewell/streamly/tree/master)
8[](https://coveralls.io/github/composewell/streamly?branch=master)
11## Streaming Concurrently
13Haskell lists express pure computations using composable stream operations like
14`:`, `unfold`, `map`, `filter`, `zip` and `fold`. Streamly is exactly like
15lists except that it can express sequences of pure as well as monadic
16computations aka streams. More importantly, it can express monadic sequences
17with concurrent execution semantics without introducing any additional APIs.
19Streamly expresses concurrency using standard, well known abstractions.
20Concurrency semantics are defined for list operations, semigroup, applicative
21and monadic compositions. Programmer does not need to know any low level
22notions of concurrency like threads, locking or synchronization. Concurrent
23and non-concurrent programs are fundamentally the same. A chosen segment of
24the program can be made concurrent by annotating it with an appropriate
25combinator. We can choose a combinator for lookahead style or asynchronous
26concurrency. Concurrency is automatically scaled up or down based on the
27demand from the consumer application, we can finally say goodbye to managing
28thread pools and associated sizing issues. The result is truly fearless
29and declarative monadic concurrency.
31## Where to use streamly?
33Streamly is a general purpose programming framework. It can be used equally
34efficiently from a simple `Hello World!` program to a massively concurrent
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?".
38Streamly simplifies streaming and makes it as intuitive as plain lists. Unlike
39other streaming libraries, no fancy types are required. Streamly is simply a
40generalization of Haskell lists to monadic streaming optionally with concurrent
41composition. The basic stream type in streamly `SerialT m a` can be considered
42as a list type `[a]` parameterized by the monad `m`. For example, `SerialT IO
43a` is a moral equivalent of `[a]` in the IO monad. `SerialT Identity a`, is
44equivalent to pure lists. Streams are constructed very much like lists, except
45that they use `nil` and `cons` instead of `[]` and `:`. Unlike lists, streams
46can be constructed from monadic effects, not just pure elements. Streams are
47processed just like lists, with list like combinators, except that they are
48monadic and work in a streaming fashion. In other words streamly just completes
49what lists lack, you do not need to learn anything new. Please see [streamly vs
50lists](docs/streamly-vs-lists.md) for a detailed comparison.
52Not surprisingly, the monad instance of streamly is a list transformer, with
53concurrency capability.
55## Why data flow programming?
57If you need some convincing for using streaming or data flow programming
58paradigm itself then try to answer this question - why do we use lists in
59Haskell? It boils down to why we use functional programming in the first place.
60Haskell is successful in enforcing the functional data flow paradigm for pure
61computations using lists, but not for monadic computations. In the absence of a
62standard and easy to use data flow programming paradigm for monadic
63computations, and the IO monad providing an escape hatch to an imperative
64model, we just love to fall into the imperative trap, and start asking the same
65fundamental question again - why do we have to use the streaming data model?
67## Comparative Performance
69High performance and simplicity are the two primary goals of streamly.
70`Streamly` employs two different stream representations (CPS and direct style)
71and interconverts between the two to get the best of both worlds on different
72operations. It uses both foldr/build (for CPS style) and stream fusion (for
73direct style) techniques to fuse operations. In terms of performance,
74Streamly's goal is to compete with equivalent C programs. Streamly redefines
75"blazing fast" for streaming libraries, it competes with lists and `vector`.
76Other streaming libraries like "streaming", "pipes" and "conduit" are orders of
77magnitude slower on most microbenchmarks. See [streaming
78benchmarks](https://github.com/composewell/streaming-benchmarks) for detailed
81The following chart shows a comparison of those streamly and list operations
82where performance of the two differs by more than 10%. Positive y-axis displays
83how many times worse is a list operation compared to the same streamly
84operation, negative y-axis shows where streamly is worse compared to lists.
86
88Streamly uses lock-free synchronization for concurrent operations. It employs
89auto-scaling of the degree of concurrency based on demand. For CPU bound tasks
90it tries to keep the threads close to the number of CPUs available whereas for
91IO bound tasks more threads can be utilized. Parallelism can be utilized with
92little overhead even if the task size is very small. See [concurrency
93benchmarks](https://github.com/composewell/concurrency-benchmarks) for detailed
94performance results and a comparison with the `async` package.
96## Installing and using
98Please see [INSTALL.md](INSTALL.md) for instructions on how to use streamly
99with your Haskell build tool or package manager. You may want to go through it
100before jumping to run the examples below.
102The module `Streamly` provides just the core stream types, type casting and
103concurrency control combinators. Stream construction, transformation, folding,
104merging, zipping combinators are found in `Streamly.Prelude`.
106## Streaming Pipelines
108The following snippet provides a simple stream composition example that reads
109numbers from stdin, prints the squares of even numbers and exits if an even
110number more than 9 is entered.
114import qualified Streamly.Prelude as S
115import Data.Function ((&))
126Unlike `pipes` or `conduit` and like `vector` and `streaming`, `streamly`
127composes stream data instead of stream processors (functions). A stream is
128just like a list and is explicitly passed around to functions that process the
129stream. Therefore, no special operator is needed to join stages in a streaming
130pipeline, just the standard function application (`$`) or reverse function
131application (`&`) operator is enough.
133## Concurrent Stream Generation
135`consM` or its operator form `|:` can be used to construct a stream from
136monadic actions. A stream constructed with `consM` can run the monadic actions
137in the stream concurrently when used with appropriate stream type combinator
138(e.g. `asyncly`, `aheadly` or `parallely`).
140The following code finishes in 3 seconds (6 seconds when serial), note the
141order of elements in the resulting output, the outputs are consumed as soon as
142each action is finished (asyncly):
145> let p n = threadDelay (n * 1000000) >> return n
146> S.toList $ asyncly $ p 3 |: p 2 |: p 1 |: S.nil
150Use `aheadly` if you want speculative concurrency i.e. execute the actions in
151the stream concurrently but consume the results in the specified order:
154> S.toList $ aheadly $ p 3 |: p 2 |: p 1 |: S.nil
158Monadic stream generation functions e.g. `unfoldrM`, `replicateM`, `repeatM`,
159`iterateM` and `fromFoldableM` etc. can work concurrently.
161The following finishes in 10 seconds (100 seconds when serial):
164S.drain $ asyncly $ S.replicateM 10 $ p 10
167## Concurrency Auto Scaling
169Concurrency is auto-scaled i.e. more actions are executed concurrently if the
170consumer is consuming the stream at a higher speed. How many tasks are executed
171concurrently can be controlled by `maxThreads` and how many results are
172buffered ahead of consumption can be controlled by `maxBuffer`. See the
173documentation in the `Streamly` module.
175## Concurrent Streaming Pipelines
177Use `|&` or `|$` to apply stream processing functions concurrently. The
178following example prints a "hello" every second; if you use `&` instead of
179`|&` you will see that the delay doubles to 2 seconds instead because of serial
184 S.repeatM (threadDelay 1000000 >> return "hello")
185 |& S.mapM (\x -> threadDelay 1000000 >> putStrLn x)
188## Mapping Concurrently
190We can use `mapM` or `sequence` functions concurrently on a stream.
193> let p n = threadDelay (n * 1000000) >> return n
194> S.drain $ aheadly $ S.mapM (\x -> p 1 >> print x) (serially $ repeatM (p 1))
197## Serial and Concurrent Merging
199Semigroup and Monoid instances can be used to fold streams serially or
200concurrently. In the following example we compose ten actions in the
201stream, each with a delay of 1 to 10 seconds, respectively. Since all the
202actions are concurrent we see one output printed every second:
206import qualified Streamly.Prelude as S
207import Control.Concurrent (threadDelay)
209main = S.toList $ parallely $ foldMap delay [1..10]
210 where delay n = S.yieldM $ threadDelay (n * 1000000) >> print n
213Streams can be combined together in many ways. We provide some examples
214below, see the tutorial for more ways. We use the following `delay`
215function in the examples to demonstrate the concurrency aspects:
219import qualified Streamly.Prelude as S
220import Control.Concurrent
222delay n = S.yieldM $ do
223 threadDelay (n * 1000000)
225 putStrLn (show tid ++ ": Delay " ++ show n)
230main = S.drain $ delay 3 <> delay 2 <> delay 1
241main = S.drain . parallely $ delay 3 <> delay 2 <> delay 1
249## Nested Loops (aka List Transformer)
251The monad instance composes like a list monad.
255import qualified Streamly.Prelude as S
258 x <- S.fromFoldable [1,2]
259 y <- S.fromFoldable [3,4]
260 S.yieldM $ putStrLn $ show (x, y)
271## Concurrent Nested Loops
273To run the above code with speculative concurrency i.e. each iteration in the
274loop can run concurrently but the results are presented to the consumer of the
275output in the same order as serial execution:
278main = S.drain $ aheadly $ loops
281Different stream types execute the loop iterations in different ways. For
282example, `wSerially` interleaves the loop iterations. There are several
283concurrent stream styles to execute the loop iterations concurrently in
284different ways, see the `Streamly.Tutorial` module for a detailed treatment.
286## Magical Concurrency
288Streams can perform semigroup (<>) and monadic bind (>>=) operations
289concurrently using combinators like `asyncly`, `parallelly`. For example,
290to concurrently generate squares of a stream of numbers and then concurrently
291sum the square roots of all combinations of two streams:
295import qualified Streamly.Prelude as S
298 s <- S.sum $ asyncly $ do
299 -- Each square is performed concurrently, (<>) is concurrent
300 x2 <- foldMap (\x -> return $ x * x) [1..100]
301 y2 <- foldMap (\y -> return $ y * y) [1..100]
302 -- Each addition is performed concurrently, monadic bind is concurrent
303 return $ sqrt (x2 + y2)
307The concurrency facilities provided by streamly can be compared with
308[OpenMP](https://en.wikipedia.org/wiki/OpenMP) and
309[Cilk](https://en.wikipedia.org/wiki/Cilk) but with a more declarative
312## Example: Listing Directories Recursively/Concurrently
314The following code snippet lists a directory tree recursively, reading multiple
315directories concurrently:
318import Control.Monad.IO.Class (liftIO)
319import Path.IO (listDir, getCurrentDir) -- from path-io package
320import Streamly (AsyncT, adapt)
321import qualified Streamly.Prelude as S
323listDirRecursive :: AsyncT IO ()
324listDirRecursive = getCurrentDir >>= readdir >>= liftIO . mapM_ putStrLn
327 (dirs, files) <- listDir dir
328 S.yield (map show dirs ++ map show files) <> foldMap readdir dirs
331main = S.drain $ adapt $ listDirRecursive
334`AsyncT` is a stream monad transformer. If you are familiar with a list
335transformer, it is nothing but `ListT` with concurrency semantics. For example,
336the semigroup operation `<>` is concurrent. This makes `foldMap` concurrent
337too. You can replace `AsyncT` with `SerialT` and the above code will become
338serial, exactly equivalent to a `ListT`.
342For bounded concurrent streams, stream yield rate can be specified. For
343example, to print hello once every second you can simply write this:
347import Streamly.Prelude as S
349main = S.drain $ asyncly $ avgRate 1 $ S.repeatM $ putStrLn "hello"
352For some practical uses of rate control, see
353[AcidRain.hs](https://github.com/composewell/streamly/tree/master/examples/AcidRain.hs)
355[CirclingSquare.hs](https://github.com/composewell/streamly/tree/master/examples/CirclingSquare.hs)
357Concurrency of the stream is automatically controlled to match the specified
358rate. Rate control works precisely even at throughputs as high as millions of
359yields per second. For more sophisticated rate control see the haddock
364The `Streamly.Memory.Array` module provides immutable arrays. Arrays are the
365computing duals of streams. Streams are good at sequential access and immutable
366transformations of in-transit data whereas arrays are good at random access and
367in-place transformations of buffered data. Unlike streams which are potentially
368infinite, arrays are necessarily finite. Arrays can be used as an efficient
369interface between streams and external storage systems like memory, files and
370network. Streams and arrays complete each other to provide a general purpose
371computing system. The design of streamly as a general purpose computing
372framework is centered around these two fundamental aspects of computing and
375`Streamly.Memory.Array` uses pinned memory outside GC and therefore avoid any
376GC overhead for the storage in arrays. Streamly allows efficient
377transformations over arrays using streams. It uses arrays to transfer data to
378and from the operating system and to store data in memory.
382Folds are consumers of streams. `Streamly.Data.Fold` module provides a `Fold`
383type that represents a `foldl'`. Such folds can be efficiently composed
384allowing the compiler to perform stream fusion and therefore implement high
385performance combinators for consuming streams. A stream can be distributed to
386multiple folds, or it can be partitioned across multiple folds, or
387demultiplexed over multiple folds, or unzipped to two folds. We can also use
388folds to fold segments of stream generating a stream of the folded results.
390If you are familiar with the `foldl` library, these are the same composable
391left folds but simpler and better integrated with streamly, and with many more
392powerful ways of composing and applying them.
396Unfolds are duals of folds. Folds help us compose consumers of streams
397efficiently and unfolds help us compose producers of streams efficiently.
398`Streamly.Data.Unfold` provides an `Unfold` type that represents an `unfoldr`
399or a stream generator. Such generators can be combined together efficiently
400allowing the compiler to perform stream fusion and implement high performance
401stream merging combinators.
405The following code snippets implement some common Unix command line utilities
406using streamly. You can compile these with `ghc -O2 -fspec-constr-recursive=16
407-fmax-worker-args=16` and compare the performance with regular GNU coreutils
408available on your system. Though many of these are not most optimal solutions
409to keep them short and elegant. Source file
410[HandleIO.hs](https://github.com/composewell/streamly/tree/master/examples/HandleIO.hs)
411in the examples directory includes these examples.
416import qualified Streamly.Prelude as S
417import qualified Streamly.Data.Fold as FL
418import qualified Streamly.Memory.Array as A
419import qualified Streamly.FileSystem.Handle as FH
420import qualified System.IO as FH
422import Data.Char (ord)
423import System.Environment (getArgs)
424import System.IO (openFile, IOMode(..), stdout)
427 (name : _) <- getArgs
428 src <- openFile name ReadMode
432 (sname : dname : _) <- getArgs
433 src <- openFile sname ReadMode
434 dst <- openFile dname WriteMode
441cat = S.fold (FH.writeChunks stdout) . S.unfold FH.readChunks
448cp src dst = S.fold (FH.writeChunks dst) $ S.unfold FH.readChunks src
455wcl = S.length . S.splitOn (== 10) FL.drain . S.unfold FH.read
456main = withArg wcl >>= print
459### Average Line Length
464 . S.splitOn (== 10) FL.length
467 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length
468 toDouble = fmap (fromIntegral :: Int -> Double)
470main = withArg avgll >>= print
473### Line Length Histogram
475`classify` is not released yet, and is available in
476`Streamly.Internal.Data.Fold`
480 S.fold (FL.classify FL.length)
482 . S.splitOn (== 10) FL.length
486 bucket n = let i = n `mod` 10 in if i > 9 then (9,n) else (i,n)
488main = withArg llhisto >>= print
493Its easy to build concurrent client and server programs using streamly.
494`Streamly.Network.*` modules provide easy combinators to build network servers
495and client programs using streamly. See
496[FromFileClient.hs](https://github.com/composewell/streamly/tree/master/examples/FromFileClient.hs),
497[EchoServer.hs](https://github.com/composewell/streamly/tree/master/examples/EchoServer.hs),
498[FileSinkServer.hs](https://github.com/composewell/streamly/tree/master/examples/FileSinkServer.hs)
499in the examples directory.
503Exceptions can be thrown at any point using the `MonadThrow` instance. Standard
504exception handling combinators like `bracket`, `finally`, `handle`,
505`onException` are provided in `Streamly.Prelude` module.
507In presence of concurrency, synchronous exceptions work just the way they are
508supposed to work in non-concurrent code. When concurrent streams
509are combined together, exceptions from the constituent streams are propagated
510to the consumer stream. When an exception occurs in any of the constituent
511streams other concurrent streams are promptly terminated.
513There is no notion of explicit threads in streamly, therefore, no
514asynchronous exceptions to deal with. You can just ignore the zillions of
515blogs, talks, caveats about async exceptions. Async exceptions just don't
516exist. Please don't use things like `myThreadId` and `throwTo` just for fun!
518## Reactive Programming (FRP)
520Streamly is a foundation for first class reactive programming as well by virtue
521of integrating concurrency and streaming. See
522[AcidRain.hs](https://github.com/composewell/streamly/tree/master/examples/AcidRain.hs)
523for a console based FRP game example and
524[CirclingSquare.hs](https://github.com/composewell/streamly/tree/master/examples/CirclingSquare.hs)
525for an SDL based animation example.
529Streamly, short for streaming concurrently, provides monadic streams, with a
530simple API, almost identical to standard lists, and an in-built
531support for concurrency. By using stream-style combinators on stream
532composition, streams can be generated, merged, chained, mapped, zipped, and
533consumed concurrently – providing a generalized high level programming
534framework unifying streaming and concurrency. Controlled concurrency allows
535even infinite streams to be evaluated concurrently. Concurrency is auto scaled
536based on feedback from the stream consumer. The programmer does not have to be
537aware of threads, locking or synchronization to write scalable concurrent
540Streamly is a programmer first library, designed to be useful and friendly to
541programmers for solving practical problems in a simple and concise manner. Some
542key points in favor of streamly are:
544 * _Simplicity_: Simple list like streaming API, if you know how to use lists
545 then you know how to use streamly. This library is built with simplicity
546 and ease of use as a design goal.
547 * _Concurrency_: Simple, powerful, and scalable concurrency. Concurrency is
548 built-in, and not intrusive, concurrent programs are written exactly the
549 same way as non-concurrent ones.
550 * _Generality_: Unifies functionality provided by several disparate packages
551 (streaming, concurrency, list transformer, logic programming, reactive
552 programming) in a concise API.
553 * _Performance_: Streamly is designed for high performance. It employs stream
554 fusion optimizations for best possible performance. Serial peformance is
555 equivalent to the venerable `vector` library in most cases and even better
556 in some cases. Concurrent performance is unbeatable. See
557 [streaming-benchmarks](https://github.com/composewell/streaming-benchmarks)
558 for a comparison of popular streaming libraries on micro-benchmarks.
560The basic streaming functionality of streamly is equivalent to that provided by
561streaming libraries like
562[vector](https://hackage.haskell.org/package/vector),
563[streaming](https://hackage.haskell.org/package/streaming),
564[pipes](https://hackage.haskell.org/package/pipes), and
565[conduit](https://hackage.haskell.org/package/conduit).
566In addition to providing streaming functionality, streamly subsumes
567the functionality of list transformer libraries like `pipes` or
568[list-t](https://hackage.haskell.org/package/list-t), and also the logic
569programming library [logict](https://hackage.haskell.org/package/logict). On
570the concurrency side, it subsumes the functionality of the
571[async](https://hackage.haskell.org/package/async) package, and provides even
572higher level concurrent composition. Because it supports
573streaming with concurrency we can write FRP applications similar in concept to
574[Yampa](https://hackage.haskell.org/package/Yampa) or
575[reflex](https://hackage.haskell.org/package/reflex).
577See the `Comparison with existing packages` section at the end of the
578[tutorial](https://hackage.haskell.org/package/streamly/docs/Streamly-Tutorial.html).
582For more information, see:
584 * [Detailed tutorial](https://hackage.haskell.org/package/streamly/docs/Streamly-Tutorial.html)
585 * [Reference documentation](https://hackage.haskell.org/package/streamly)
586 * [Examples](examples)
588 * [Streaming benchmarks](https://github.com/composewell/streaming-benchmarks)
589 * [Concurrency benchmarks](https://github.com/composewell/concurrency-benchmarks)
591For additional unreleased/experimental APIs, build the haddock docs using:
594$ cabal haddock --haddock-option="--show-all"
595$ stack haddock --haddock-arguments "--show-all" --no-haddock-deps
600Please feel free to ask questions on the
601[streamly gitter channel](https://gitter.im/composewell/streamly).
602If you require professional support, consulting, training or timely
603enhancements to the library please contact
604[support@composewell.com](mailto:support@composewell.com).
608The following authors/libraries have influenced or inspired this library in a
611 * Roman Leshchinskiy (vector)
612 * Gabriel Gonzalez (foldl)
613 * Alberto G. Corona (transient)
615See the `credits` directory for full list of contributors, credits and licenses.
619The code is available under BSD-3 license
620[on github](https://github.com/composewell/streamly). Join the [gitter
621chat](https://gitter.im/composewell/streamly) channel for discussions. Please
622ask any questions on the gitter channel or [contact the maintainer
623directly](mailto:streamly@composewell.com). All contributions are welcome!
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