git.iko.soy/streamly
on 03218ad help sign in
streamly/docs/Overview.md
469 lines 18 kB history raw
1## Overview
3These are some notes from an old version of README that may be
4useful. For a quick introduction please read the README.md at the repo
5root first.
7## Streaming Concurrently
9Haskell lists express pure computations using composable stream operations like
10`:`, `unfold`, `map`, `filter`, `zip` and `fold`. Streamly is exactly like
11lists except that it can express sequences of pure as well as monadic
12computations aka streams. More importantly, it can express monadic sequences
13with concurrent execution semantics without introducing any additional APIs.
15Streamly expresses concurrency using standard, well known abstractions.
16Concurrency semantics are defined for list operations, semigroup, applicative
17and monadic compositions. Programmer does not need to know any low level
18notions of concurrency like threads, locking or synchronization. Concurrent
19and non-concurrent programs are fundamentally the same. A chosen segment of
20the program can be made concurrent by annotating it with an appropriate
21combinator. We can choose a combinator for lookahead style or asynchronous
22concurrency. Concurrency is automatically scaled up or down based on the
23demand from the consumer application, we can finally say goodbye to managing
24thread pools and associated sizing issues. The result is truly fearless
25and declarative monadic concurrency.
27## Where to use streamly?
29Streamly is a general purpose programming framework. It can be used equally
30efficiently from a simple `Hello World!` program to a massively concurrent
31application. The answer to the question, "where to use streamly?" - would be
32similar to the answer to - "Where to use Haskell lists or the IO monad?".
34Streamly simplifies streaming and makes it as intuitive as plain lists. Unlike
35other streaming libraries, no fancy types are required. Streamly is simply a
36generalization of Haskell lists to monadic streaming optionally with concurrent
37composition. The basic stream type in streamly `SerialT m a` can be considered
38as a list type `[a]` parameterized by the monad `m`. For example, `SerialT IO
39a` is a moral equivalent of `[a]` in the IO monad. `SerialT Identity a`, is
40equivalent to pure lists. Streams are constructed very much like lists, except
41that they use `nil` and `cons` instead of `[]` and `:`. Unlike lists, streams
42can be constructed from monadic effects, not just pure elements. Streams are
43processed just like lists, with list like combinators, except that they are
44monadic and work in a streaming fashion. In other words streamly just completes
45what lists lack, you do not need to learn anything new. Please see [streamly vs
46lists](docs/streamly-vs-lists.md) for a detailed comparison.
48Not surprisingly, the monad instance of streamly is a list transformer, with
49concurrency capability.
51## Why data flow programming?
53If you need some convincing for using streaming or data flow programming
54paradigm itself then try to answer this question - why do we use lists in
55Haskell? It boils down to why we use functional programming in the first place.
56Haskell is successful in enforcing the functional data flow paradigm for pure
57computations using lists, but not for monadic computations. In the absence of a
58standard and easy to use data flow programming paradigm for monadic
59computations, and the IO monad providing an escape hatch to an imperative
60model, we just love to fall into the imperative trap, and start asking the same
61fundamental question again - why do we have to use the streaming data model?
63## Streaming Pipelines
65The following snippet provides a simple stream composition example that reads
66numbers from stdin, prints the squares of even numbers and exits if an even
67number more than 9 is entered.
69``` haskell
70import qualified Streamly.Prelude as S
71import Data.Function ((&))
73main = S.drain $
74 S.repeatM getLine
75 & fmap read
76 & S.filter even
77 & S.takeWhile (<= 9)
78 & fmap (\x -> x * x)
79 & S.mapM print
80```
82Unlike `pipes` or `conduit` and like `vector` and `streaming`, `streamly`
83composes stream data instead of stream processors (functions). A stream is
84just like a list and is explicitly passed around to functions that process the
85stream. Therefore, no special operator is needed to join stages in a streaming
86pipeline, just the standard function application (`$`) or reverse function
87application (`&`) operator is enough.
89## Concurrent Stream Generation
91`consM` or its operator form `|:` can be used to construct a stream from
92monadic actions. A stream constructed with `consM` can run the monadic actions
93in the stream concurrently when used with appropriate stream type combinator
94(e.g. `fromAsync`, `fromAhead` or `fromParallel`).
96The following code finishes in 3 seconds (6 seconds when serial), note the
97order of elements in the resulting output, the outputs are consumed as soon as
98each action is finished (asyncly):
100``` haskell
101> let p n = threadDelay (n * 1000000) >> return n
102> S.toList $ S.fromAsync $ p 3 |: p 2 |: p 1 |: S.nil
103[1,2,3]
104```
106Use `fromAhead` if you want speculative concurrency i.e. execute the actions in
107the stream concurrently but consume the results in the specified order:
109``` haskell
110> S.toList $ S.fromAhead $ p 3 |: p 2 |: p 1 |: S.nil
111[3,2,1]
112```
114Monadic stream generation functions e.g. `unfoldrM`, `replicateM`, `repeatM`,
115`iterateM` and `fromFoldableM` etc. can work concurrently.
117The following finishes in 10 seconds (100 seconds when serial):
119``` haskell
120S.drain $ S.fromAsync $ S.replicateM 10 $ p 10
121```
123## Concurrent Streaming Pipelines
125Use `|&` or `|$` to apply stream processing functions concurrently. The
126following example prints a "hello" every second; if you use `&` instead of
127`|&` you will see that the delay doubles to 2 seconds instead because of serial
128application.
130``` haskell
131main = S.drain $
132 S.repeatM (threadDelay 1000000 >> return "hello")
133 |& S.mapM (\x -> threadDelay 1000000 >> putStrLn x)
134```
136## Mapping Concurrently
138We can use `mapM` or `sequence` functions concurrently on a stream.
140``` haskell
141> let p n = threadDelay (n * 1000000) >> return n
142> S.drain $ S.fromAhead $ S.mapM (\x -> p 1 >> print x) (S.fromSerial $ S.repeatM (p 1))
143```
145## Serial and Concurrent Merging
147Semigroup and Monoid instances can be used to fold streams serially or
148concurrently. In the following example we compose ten actions in the
149stream, each with a delay of 1 to 10 seconds, respectively. Since all the
150actions are concurrent we see one output printed every second:
152``` haskell
153import qualified Streamly.Prelude as S
154import Control.Concurrent (threadDelay)
156main = S.toList $ S.fromParallel $ foldMap delay [1..10]
157 where delay n = S.fromEffect $ threadDelay (n * 1000000) >> print n
158```
160Streams can be combined together in many ways. We provide some examples
161below, see the tutorial for more ways. We use the following `delay`
162function in the examples to demonstrate the concurrency aspects:
164``` haskell
165import qualified Streamly.Prelude as S
166import Control.Concurrent
168delay n = S.fromEffect $ do
169 threadDelay (n * 1000000)
170 tid <- myThreadId
171 putStrLn (show tid ++ ": Delay " ++ show n)
172```
173### Serial
175``` haskell
176main = S.drain $ delay 3 <> delay 2 <> delay 1
177```
178```
179ThreadId 36: Delay 3
180ThreadId 36: Delay 2
181ThreadId 36: Delay 1
182```
184### Parallel
186``` haskell
187main = S.drain . S.fromParallel $ delay 3 <> delay 2 <> delay 1
188```
189```
190ThreadId 42: Delay 1
191ThreadId 41: Delay 2
192ThreadId 40: Delay 3
193```
195## Nested Loops (aka List Transformer)
197The monad instance composes like a list monad.
199``` haskell
200import qualified Streamly.Prelude as S
202loops = do
203 x <- S.fromFoldable [1,2]
204 y <- S.fromFoldable [3,4]
205 S.fromEffect $ putStrLn $ show (x, y)
207main = S.drain loops
208```
209```
210(1,3)
211(1,4)
212(2,3)
213(2,4)
214```
216## Concurrent Nested Loops
218To run the above code with speculative concurrency i.e. each iteration in the
219loop can run concurrently but the results are presented to the consumer of the
220output in the same order as serial execution:
222``` haskell
223main = S.drain $ S.fromAhead $ loops
224```
226Different stream types execute the loop iterations in different ways. For
227example, `fromWSerial` interleaves the loop iterations. There are several
228concurrent stream styles to execute the loop iterations concurrently in
229different ways, see the `Streamly.Tutorial` module for a detailed treatment.
231## Magical Concurrency
233Streams can perform semigroup (<>) and monadic bind (>>=) operations
234concurrently using combinators like `fromAsync`, `parallelly`. For example,
235to concurrently generate squares of a stream of numbers and then concurrently
236sum the square roots of all combinations of two streams:
238``` haskell
239import qualified Streamly.Prelude as S
241main = do
242 s <- S.sum $ S.fromAsync $ do
243 -- Each square is performed concurrently, (<>) is concurrent
244 x2 <- foldMap (\x -> return $ x * x) [1..100]
245 y2 <- foldMap (\y -> return $ y * y) [1..100]
246 -- Each addition is performed concurrently, monadic bind is concurrent
247 return $ sqrt (x2 + y2)
248 print s
249```
251## Rate Limiting
253For bounded concurrent streams, stream yield rate can be specified. For
254example, to print hello once every second you can simply write this:
256``` haskell
257import Streamly.Prelude as S
259main = S.drain $ S.fromAsync $ S.avgRate 1 $ S.repeatM $ putStrLn "hello"
260```
262For some practical uses of rate control, see
263[AcidRain.hs](https://github.com/composewell/streamly/tree/master/examples/AcidRain.hs)
264and
265[CirclingSquare.hs](https://github.com/composewell/streamly/tree/master/examples/CirclingSquare.hs)
267Concurrency of the stream is automatically controlled to match the specified
268rate. Rate control works precisely even at throughputs as high as millions of
269yields per second. For more sophisticated rate control see the haddock
270documentation.
272## Arrays
274The `Streamly.Data.Array.Foreign` module provides immutable arrays. Arrays are the
275computing duals of streams. Streams are good at sequential access and immutable
276transformations of in-transit data whereas arrays are good at random access and
277in-place transformations of buffered data. Unlike streams which are potentially
278infinite, arrays are necessarily finite. Arrays can be used as an efficient
279interface between streams and external storage systems like memory, files and
280network. Streams and arrays complete each other to provide a general purpose
281computing system. The design of streamly as a general purpose computing
282framework is centered around these two fundamental aspects of computing and
283storage.
285`Streamly.Data.Array.Foreign` uses pinned memory outside GC and therefore avoid any
286GC overhead for the storage in arrays. Streamly allows efficient
287transformations over arrays using streams. It uses arrays to transfer data to
288and from the operating system and to store data in memory.
290## Folds
292Folds are consumers of streams. `Streamly.Data.Fold` module provides a `Fold`
293type that represents a `foldl'`. Such folds can be efficiently composed
294allowing the compiler to perform stream fusion and therefore implement high
295performance combinators for consuming streams. A stream can be distributed to
296multiple folds, or it can be partitioned across multiple folds, or
297demultiplexed over multiple folds, or unzipped to two folds. We can also use
298folds to fold segments of stream generating a stream of the folded results.
300If you are familiar with the `foldl` library, these are the same composable
301left folds but simpler and better integrated with streamly, and with many more
302powerful ways of composing and applying them.
304## Unfolds
306Unfolds are duals of folds. Folds help us compose consumers of streams
307efficiently and unfolds help us compose producers of streams efficiently.
308`Streamly.Data.Unfold` provides an `Unfold` type that represents an `unfoldr`
309or a stream generator. Such generators can be combined together efficiently
310allowing the compiler to perform stream fusion and implement high performance
311stream merging combinators.
313## File IO
315The following code snippets implement some common Unix command line utilities
316using streamly. You can compile these with `ghc -O2 -fspec-constr-recursive=16
317-fmax-worker-args=16` and compare the performance with regular GNU coreutils
318available on your system. Though many of these are not most optimal solutions
319to keep them short and elegant. Source file
320[HandleIO.hs](https://github.com/composewell/streamly/tree/master/examples/HandleIO.hs)
321in the examples directory includes these examples.
323``` haskell
324module Main where
326import qualified Streamly.Prelude as S
327import qualified Streamly.Data.Fold as FL
328import qualified Streamly.Data.Array.Foreign as A
329import qualified Streamly.FileSystem.Handle as FH
330import qualified System.IO as FH
332import Data.Char (ord)
333import System.Environment (getArgs)
334import System.IO (openFile, IOMode(..), stdout)
336withArg f = do
337 (name : _) <- getArgs
338 src <- openFile name ReadMode
339 f src
341withArg2 f = do
342 (sname : dname : _) <- getArgs
343 src <- openFile sname ReadMode
344 dst <- openFile dname WriteMode
345 f src dst
346```
348### cat
350``` haskell
351cat = S.fold (FH.writeChunks stdout) . S.unfold FH.readChunks
352main = withArg cat
353```
355### cp
357``` haskell
358cp src dst = S.fold (FH.writeChunks dst) $ S.unfold FH.readChunks src
359main = withArg2 cp
360```
362### wc -l
364``` haskell
365wcl = S.length . S.splitOn (== 10) FL.drain . S.unfold FH.read
366main = withArg wcl >>= print
367```
369### Average Line Length
371``` haskell
372avgll =
373 S.fold avg
374 . S.splitOn (== 10) FL.length
375 . S.unfold FH.read
377 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length
378 toDouble = fmap (fromIntegral :: Int -> Double)
380main = withArg avgll >>= print
381```
383### Line Length Histogram
385`classify` is not released yet, and is available in
386`Streamly.Internal.Data.Fold`
388``` haskell
389llhisto =
390 S.fold (FL.classify FL.length)
391 . S.map bucket
392 . S.splitOn (== 10) FL.length
393 . S.unfold FH.read
395 where
396 bucket n = let i = n `mod` 10 in if i > 9 then (9,n) else (i,n)
398main = withArg llhisto >>= print
399```
401## Exceptions
403Exceptions can be thrown at any point using the `MonadThrow` instance. Standard
404exception handling combinators like `bracket`, `finally`, `handle`,
405`onException` are provided in `Streamly.Prelude` module.
407In presence of concurrency, synchronous exceptions work just the way they are
408supposed to work in non-concurrent code. When concurrent streams
409are combined together, exceptions from the constituent streams are propagated
410to the consumer stream. When an exception occurs in any of the constituent
411streams other concurrent streams are promptly terminated.
413There is no notion of explicit threads in streamly, therefore, no
414asynchronous exceptions to deal with. You can just ignore the zillions of
415blogs, talks, caveats about async exceptions. Async exceptions just don't
416exist. Please don't use things like `myThreadId` and `throwTo` just for fun!
418## Conclusion
420Streamly, short for streaming concurrently, provides monadic streams, with a
421simple API, almost identical to standard lists, and an in-built
422support for concurrency. By using stream-style combinators on stream
423composition, streams can be generated, merged, chained, mapped, zipped, and
424consumed concurrently – providing a generalized high level programming
425framework unifying streaming and concurrency. Controlled concurrency allows
426even infinite streams to be evaluated concurrently. Concurrency is auto scaled
427based on feedback from the stream consumer. The programmer does not have to be
428aware of threads, locking or synchronization to write scalable concurrent
429programs.
431Streamly is a programmer first library, designed to be useful and friendly to
432programmers for solving practical problems in a simple and concise manner. Some
433key points in favor of streamly are:
435 * _Simplicity_: Simple list like streaming API, if you know how to use lists
436 then you know how to use streamly. This library is built with simplicity
437 and ease of use as a design goal.
438 * _Concurrency_: Simple, powerful, and scalable concurrency. Concurrency is
439 built-in, and not intrusive, concurrent programs are written exactly the
440 same way as non-concurrent ones.
441 * _Generality_: Unifies functionality provided by several disparate packages
442 (streaming, concurrency, list transformer, logic programming, reactive
443 programming) in a concise API.
444 * _Performance_: Streamly is designed for high performance. It employs stream
445 fusion optimizations for best possible performance. Serial peformance is
446 equivalent to the venerable `vector` library in most cases and even better
447 in some cases. Concurrent performance is unbeatable. See
448 [streaming-benchmarks](https://github.com/composewell/streaming-benchmarks)
449 for a comparison of popular streaming libraries on micro-benchmarks.
451The basic streaming functionality of streamly is equivalent to that provided by
452streaming libraries like
453[vector](https://hackage.haskell.org/package/vector),
454[streaming](https://hackage.haskell.org/package/streaming),
455[pipes](https://hackage.haskell.org/package/pipes), and
456[conduit](https://hackage.haskell.org/package/conduit).
457In addition to providing streaming functionality, streamly subsumes
458the functionality of list transformer libraries like `pipes` or
459[list-t](https://hackage.haskell.org/package/list-t), and also the logic
460programming library [logict](https://hackage.haskell.org/package/logict). On
461the concurrency side, it subsumes the functionality of the
462[async](https://hackage.haskell.org/package/async) package, and provides even
463higher level concurrent composition. Because it supports
464streaming with concurrency we can write FRP applications similar in concept to
465[Yampa](https://hackage.haskell.org/package/Yampa) or
466[reflex](https://hackage.haskell.org/package/reflex).
468See the `Comparison with existing packages` section at the end of the
469[tutorial](https://hackage.haskell.org/package/streamly/docs/Streamly-Tutorial.html).