git.iko.soy/streamly
on bc5617f help sign in
streamly/docs/Overview.md
468 lines 17 kB history raw
1# Streaming
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## Streamly
9Streamly is a general computing framework based on data flow programming also
10known as streaming. Moreover streamly supports concurrent dataflow programming.
12Streaming in general enables writing modular, composable and scalable
13applications with ease, and concurrency allows you to make them scale and
14perform well. Streamly enables writing scalable concurrent applications
15without being aware of threads or synchronization. No explicit thread
16control is needed. Where applicable, concurrency rate is automatically
17controlled based on the demand by the consumer. However, combinators can be
18used to fine tune the concurrency control.
20Streaming and concurrency together enable expressing reactive applications
21conveniently. See the @CirclingSquare@ example in
22<https://github.com/composewell/streamly-examples Streamly Examples> for
23a simple SDL based FRP example. To summarize, streamly provides a unified
24computing framework for streaming, non-determinism and functional reactive
25programming in an elegant and simple API that is a natural extension of pure
26lists to monadic streams.
28## What are streams?
30In simple terms, streams are the functional equivalent of loops in
31imperative programming.
33A stream is a representation of potentially infinite data sequence. You
34can compose a pipeline of functions or stream processors to process
35an input stream of data to produce an output stream. We call it a
36form of dataflow programming as data flows through the processing
37logic. In imperative programming there is no clear separation of data
38and logic. The logic can arbitrarily examine and mutate data which
39creates a problem due to complex interleaving of state and logic in the
40program.
42In streamly there are two fundamental data structures, streams and arrays.
43Streams are for dataflow style processing while arrays are for storing data.
44Both taken together are powerful tools for general purpose programming in a
45functional or dataflow style.
47## Loops vs Streams
49In imperative programming when we have to process a sequence of items
50or an array of data we run a loop over it, each iteration of the loop
51examines the data to do something with it and produce an output.
53Loops are a low level, monolithic and general concept. Whereas streams
54are high level, structured and modular way of expressing what you usualy
55do with loops. Streams allow you to write different parts of the loop
56as separate modular combinators and then compose them to create bigger
57loops.
59## Streaming Concurrently
61Haskell lists express pure computations using composable stream operations like
62`:`, `unfold`, `map`, `filter`, `zip` and `fold`. Streamly is exactly like
63lists except that it can express sequences of pure as well as monadic
64computations aka streams. More importantly, it can express monadic sequences
65with concurrent execution semantics without introducing any additional APIs.
67Streamly expresses concurrency using standard, well known abstractions.
68Concurrency semantics are defined for list operations, semigroup, applicative
69and monadic compositions. Programmer does not need to know any low level
70notions of concurrency like threads, locking or synchronization. Concurrent
71and non-concurrent programs are fundamentally the same. A chosen segment of
72the program can be made concurrent by annotating it with an appropriate
73combinator. We can choose a combinator for lookahead style or asynchronous
74concurrency. Concurrency is automatically scaled up or down based on the
75demand from the consumer application, we can finally say goodbye to managing
76thread pools and associated sizing issues. The result is truly fearless
77and declarative monadic concurrency.
79## Where to use streamly?
81Streamly is a general purpose programming framework. It can be used equally
82efficiently from a simple `Hello World!` program to a massively concurrent
83application. The answer to the question, "where to use streamly?" - would be
84similar to the answer to - "Where to use Haskell lists or the IO monad?".
86Streamly simplifies streaming and makes it as intuitive as plain lists. Unlike
87other streaming libraries, no fancy types are required. Streamly is simply a
88generalization of Haskell lists to monadic streaming optionally with concurrent
89composition. The basic stream type in streamly `SerialT m a` can be considered
90as a list type `[a]` parameterized by the monad `m`. For example, `SerialT IO
91a` is a moral equivalent of `[a]` in the IO monad. `SerialT Identity a`, is
92equivalent to pure lists. Streams are constructed very much like lists, except
93that they use `nil` and `cons` instead of `[]` and `:`. Unlike lists, streams
94can be constructed from monadic effects, not just pure elements. Streams are
95processed just like lists, with list like combinators, except that they are
96monadic and work in a streaming fashion. In other words streamly just completes
97what lists lack, you do not need to learn anything new. Please see [streamly vs
98lists](docs/streamly-vs-lists.md) for a detailed comparison.
100Not surprisingly, the monad instance of streamly is a list transformer, with
101concurrency capability.
103## Why data flow programming?
105If you need some convincing for using streaming or data flow programming
106paradigm itself then try to answer this question - why do we use lists in
107Haskell? It boils down to why we use functional programming in the first place.
108Haskell is successful in enforcing the functional data flow paradigm for pure
109computations using lists, but not for monadic computations. In the absence of a
110standard and easy to use data flow programming paradigm for monadic
111computations, and the IO monad providing an escape hatch to an imperative
112model, we just love to fall into the imperative trap, and start asking the same
113fundamental question again - why do we have to use the streaming data model?
115## Streaming Pipelines
117The following snippet provides a simple stream composition example that reads
118numbers from stdin, prints the squares of even numbers and exits if an even
119number more than 9 is entered.
121``` haskell
122import qualified Streamly.Prelude as S
123import Data.Function ((&))
125main = S.drain $
126 S.repeatM getLine
127 & fmap read
128 & S.filter even
129 & S.takeWhile (<= 9)
130 & fmap (\x -> x * x)
131 & S.mapM print
132```
134Unlike `pipes` or `conduit` and like `vector` and `streaming`, `streamly`
135composes stream data instead of stream processors (functions). A stream is
136just like a list and is explicitly passed around to functions that process the
137stream. Therefore, no special operator is needed to join stages in a streaming
138pipeline, just the standard function application (`$`) or reverse function
139application (`&`) operator is enough.
141## Concurrent Stream Generation
143`consM` or its operator form `|:` can be used to construct a stream from
144monadic actions. A stream constructed with `consM` can run the monadic actions
145in the stream concurrently when used with appropriate stream type combinator
146(e.g. `fromAsync`, `fromAhead` or `fromParallel`).
148The following code finishes in 3 seconds (6 seconds when serial), note the
149order of elements in the resulting output, the outputs are consumed as soon as
150each action is finished (asyncly):
152``` haskell
153> let p n = threadDelay (n * 1000000) >> return n
154> S.toList $ S.fromAsync $ p 3 |: p 2 |: p 1 |: S.nil
155[1,2,3]
156```
158Use `fromAhead` if you want speculative concurrency i.e. execute the actions in
159the stream concurrently but consume the results in the specified order:
161``` haskell
162> S.toList $ S.fromAhead $ p 3 |: p 2 |: p 1 |: S.nil
163[3,2,1]
164```
166Monadic stream generation functions e.g. `unfoldrM`, `replicateM`, `repeatM`,
167`iterateM` and `fromFoldableM` etc. can work concurrently.
169The following finishes in 10 seconds (100 seconds when serial):
171``` haskell
172S.drain $ S.fromAsync $ S.replicateM 10 $ p 10
173```
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
180application.
182``` haskell
183main = S.drain $
184 S.repeatM (threadDelay 1000000 >> return "hello")
185 |& S.mapM (\x -> threadDelay 1000000 >> putStrLn x)
186```
188## Mapping Concurrently
190We can use `mapM` or `sequence` functions concurrently on a stream.
192``` haskell
193> let p n = threadDelay (n * 1000000) >> return n
194> S.drain $ S.fromAhead $ S.mapM (\x -> p 1 >> print x) (S.fromSerial $ S.repeatM (p 1))
195```
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:
204``` haskell
205import qualified Streamly.Prelude as S
206import Control.Concurrent (threadDelay)
208main = S.toList $ S.fromParallel $ foldMap delay [1..10]
209 where delay n = S.fromEffect $ threadDelay (n * 1000000) >> print n
210```
212Streams can be combined together in many ways. We provide some examples
213below, see the tutorial for more ways. We use the following `delay`
214function in the examples to demonstrate the concurrency aspects:
216``` haskell
217import qualified Streamly.Prelude as S
218import Control.Concurrent
220delay n = S.fromEffect $ do
221 threadDelay (n * 1000000)
222 tid <- myThreadId
223 putStrLn (show tid ++ ": Delay " ++ show n)
224```
225### Serial
227``` haskell
228main = S.drain $ delay 3 <> delay 2 <> delay 1
229```
230```
231ThreadId 36: Delay 3
232ThreadId 36: Delay 2
233ThreadId 36: Delay 1
234```
236### Parallel
238``` haskell
239main = S.drain . S.fromParallel $ delay 3 <> delay 2 <> delay 1
240```
241```
242ThreadId 42: Delay 1
243ThreadId 41: Delay 2
244ThreadId 40: Delay 3
245```
247## Nested Loops (aka List Transformer)
249The monad instance composes like a list monad.
251``` haskell
252import qualified Streamly.Prelude as S
254loops = do
255 x <- S.fromFoldable [1,2]
256 y <- S.fromFoldable [3,4]
257 S.fromEffect $ putStrLn $ show (x, y)
259main = S.drain loops
260```
261```
262(1,3)
263(1,4)
264(2,3)
265(2,4)
266```
268## Concurrent Nested Loops
270To run the above code with speculative concurrency i.e. each iteration in the
271loop can run concurrently but the results are presented to the consumer of the
272output in the same order as serial execution:
274``` haskell
275main = S.drain $ S.fromAhead $ loops
276```
278Different stream types execute the loop iterations in different ways. For
279example, `fromWSerial` interleaves the loop iterations. There are several
280concurrent stream styles to execute the loop iterations concurrently in
281different ways, see the `Streamly.Tutorial` module for a detailed treatment.
283## Magical Concurrency
285Streams can perform semigroup (<>) and monadic bind (>>=) operations
286concurrently using combinators like `fromAsync`, `parallelly`. For example,
287to concurrently generate squares of a stream of numbers and then concurrently
288sum the square roots of all combinations of two streams:
290``` haskell
291import qualified Streamly.Prelude as S
293main = do
294 s <- S.sum $ S.fromAsync $ do
295 -- Each square is performed concurrently, (<>) is concurrent
296 x2 <- foldMap (\x -> return $ x * x) [1..100]
297 y2 <- foldMap (\y -> return $ y * y) [1..100]
298 -- Each addition is performed concurrently, monadic bind is concurrent
299 return $ sqrt (x2 + y2)
300 print s
301```
303## Rate Limiting
305For bounded concurrent streams, stream yield rate can be specified. For
306example, to print hello once every second you can simply write this:
308``` haskell
309import Streamly.Prelude as S
311main = S.drain $ S.fromAsync $ S.avgRate 1 $ S.repeatM $ putStrLn "hello"
312```
314For some practical uses of rate control, see
315[AcidRain.hs](https://github.com/composewell/streamly-examples/tree/master/examples/AcidRain.hs)
316and
317[CirclingSquare.hs](https://github.com/composewell/streamly-examples/tree/master/examples/CirclingSquare.hs)
319Concurrency of the stream is automatically controlled to match the specified
320rate. Rate control works precisely even at throughputs as high as millions of
321yields per second. For more sophisticated rate control see the haddock
322documentation.
324## Arrays
326The `Streamly.Data.Array.Foreign` module provides immutable arrays. Arrays are the
327computing duals of streams. Streams are good at sequential access and immutable
328transformations of in-transit data whereas arrays are good at random access and
329in-place transformations of buffered data. Unlike streams which are potentially
330infinite, arrays are necessarily finite. Arrays can be used as an efficient
331interface between streams and external storage systems like memory, files and
332network. Streams and arrays complete each other to provide a general purpose
333computing system. The design of streamly as a general purpose computing
334framework is centered around these two fundamental aspects of computing and
335storage.
337`Streamly.Data.Array.Foreign` uses pinned memory outside GC and therefore avoid any
338GC overhead for the storage in arrays. Streamly allows efficient
339transformations over arrays using streams. It uses arrays to transfer data to
340and from the operating system and to store data in memory.
342## Folds
344Folds are consumers of streams. `Streamly.Data.Fold` module provides a `Fold`
345type that represents a `foldl'`. Such folds can be efficiently composed
346allowing the compiler to perform stream fusion and therefore implement high
347performance combinators for consuming streams. A stream can be distributed to
348multiple folds, or it can be partitioned across multiple folds, or
349demultiplexed over multiple folds, or unzipped to two folds. We can also use
350folds to fold segments of stream generating a stream of the folded results.
352If you are familiar with the `foldl` library, these are the same composable
353left folds but simpler and better integrated with streamly, and with many more
354powerful ways of composing and applying them.
356## Unfolds
358Unfolds are duals of folds. Folds help us compose consumers of streams
359efficiently and unfolds help us compose producers of streams efficiently.
360`Streamly.Data.Unfold` provides an `Unfold` type that represents an `unfoldr`
361or a stream generator. Such generators can be combined together efficiently
362allowing the compiler to perform stream fusion and implement high performance
363stream merging combinators.
365## File IO
367The following code snippets implement some common Unix command line utilities
368using streamly. You can compile these with `ghc -O2 -fspec-constr-recursive=16
369-fmax-worker-args=16` and compare the performance with regular GNU coreutils
370available on your system. Though many of these are not most optimal solutions
371to keep them short and elegant. Source file
372[HandleIO.hs](https://github.com/composewell/streamly-examples/tree/master/examples/HandleIO.hs)
373in the examples directory includes these examples.
375``` haskell
376module Main where
378import qualified Streamly.Prelude as S
379import qualified Streamly.Data.Fold as FL
380import qualified Streamly.Data.Array.Foreign as A
381import qualified Streamly.FileSystem.Handle as FH
382import qualified System.IO as FH
384import Data.Char (ord)
385import System.Environment (getArgs)
386import System.IO (openFile, IOMode(..), stdout)
388withArg f = do
389 (name : _) <- getArgs
390 src <- openFile name ReadMode
391 f src
393withArg2 f = do
394 (sname : dname : _) <- getArgs
395 src <- openFile sname ReadMode
396 dst <- openFile dname WriteMode
397 f src dst
398```
400### cat
402``` haskell
403cat = S.fold (FH.writeChunks stdout) . S.unfold FH.readChunks
404main = withArg cat
405```
407### cp
409``` haskell
410cp src dst = S.fold (FH.writeChunks dst) $ S.unfold FH.readChunks src
411main = withArg2 cp
412```
414### wc -l
416``` haskell
417wcl = S.length . S.splitOn (== 10) FL.drain . S.unfold FH.read
418main = withArg wcl >>= print
419```
421### Average Line Length
423``` haskell
424avgll =
425 S.fold avg
426 . S.splitOn (== 10) FL.length
427 . S.unfold FH.read
429 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length
430 toDouble = fmap (fromIntegral :: Int -> Double)
432main = withArg avgll >>= print
433```
435### Line Length Histogram
437`classify` is not released yet, and is available in
438`Streamly.Internal.Data.Fold`
440``` haskell
441llhisto =
442 S.fold (FL.classify FL.length)
443 . S.map bucket
444 . S.splitOn (== 10) FL.length
445 . S.unfold FH.read
447 where
448 bucket n = let i = n `mod` 10 in if i > 9 then (9,n) else (i,n)
450main = withArg llhisto >>= print
451```
453## Exceptions
455Exceptions can be thrown at any point using the `MonadThrow` instance. Standard
456exception handling combinators like `bracket`, `finally`, `handle`,
457`onException` are provided in `Streamly.Prelude` module.
459In presence of concurrency, synchronous exceptions work just the way they are
460supposed to work in non-concurrent code. When concurrent streams
461are combined together, exceptions from the constituent streams are propagated
462to the consumer stream. When an exception occurs in any of the constituent
463streams other concurrent streams are promptly terminated.
465There is no notion of explicit threads in streamly, therefore, no
466asynchronous exceptions to deal with. You can just ignore the zillions of
467blogs, talks, caveats about async exceptions. Async exceptions just don't
468exist. Please don't use things like `myThreadId` and `throwTo` just for fun!