1# [Streamly][]: Idiomatic Haskell with the Performance of C
3[](https://gitter.im/composewell/streamly)
4[](https://hackage.haskell.org/package/streamly)
10[Streamly]: https://streamly.composewell.com/
11[streamly-examples]: https://github.com/composewell/streamly-examples
12[streaming-benchmarks]: https://github.com/composewell/streaming-benchmarks
13[concurrency-benchmarks]: https://github.com/composewell/concurrency-benchmarks
16Keep all the unstable links here so that they can be updated to stable
17links (for online docs) before we release.
21[WordCountModular.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCountModular.hs
22[WordCount.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCount.hs
23[WordCount.c]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCount.c
24[WordCountParallel.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCountParallel.hs
25[WordCountUTF8.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCountUTF8.hs
26[WordServer.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordServer.hs
27[MergeServer.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/MergeServer.hs
28[ListDir.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/ListDir.hs
29[Rate.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/Rate.hs
30[AcidRain.hs]: https://github.com/composewell/streamly-examples/tree/master/examples/AcidRain.hs
31[CirclingSquare.hs]: https://github.com/composewell/streamly-examples/tree/master/examples/CirclingSquare.hs
34[LICENSE]: https://github.com/composewell/streamly/blob/master/LICENSE
35[CONTRIBUTING.md]: https://github.com/composewell/streamly/blob/master/CONTRIBUTING.md
36[docs]: https://github.com/composewell/streamly/blob/master/docs/
38Streamly is a Haskell library that provides the building blocks to build
39safe, scalable, modular and high performance software. Streamly offers:
41* The type safety of Haskell.
42* The performance of C programs.
43* Powerful abstractions for structuring your code.
44* Idiomatic functional programming.
45* Declarative concurrency for the seamless use of multiprocessing hardware.
49This guide introduces programming with [Streamly][] using a few practical
52* We will start with a simple program that [counts the number of words
53 in a text](#modular-word-counting). We will then transform this program
54 into a [concurrent](#concurrent-word-counting) program that can efficiently
55 use multiprocessing hardware.
56* Next, we will create a [concurrent network
57 server](#a-concurrent-network-server). We then show
58 how to write a network server that [merges multiple
59 streams](#merging-incoming-streams) concurrently.
60* Our third example shows how to list a directory tree concurrently,
61 by reading [multiple directories in
62 parallel](#listing-directories-recursivelyconcurrently).
63* Finally, we will look at how to [rate limit](#rate-limiting) stream
66The guide then looks at how Streamly achieves its
67[performance](#performance). It [concludes](#notes) with a brief
68discussion about Streamly's design philosophy, and with suggestions for
73### Installing Streamly
75If you wish to follow along with this guide, you will need to have
76[Streamly][] installed.
78Please see the [Getting Started With The Streamly Package](./INSTALL.md)
79guide for instructions on how to install [Streamly][].
81If you wish to run benchmarks, please be sure to build your
82application using the instructions in the [Build And Compile Options
85### An overview of the types used in these examples
87As an expository device, we have indicated the types at the intermediate
88stages of stream computations as comments in the examples below.
89The meaning of these types are:
91* A `SerialT IO a` is a serial stream of values of type `a` in the IO Monad.
92* An `AsyncT IO a` is a concurrent (asynchronous) stream of values of type
94* An `Unfold IO a b` is a representation of a function that converts a seed
95 value of type `a` into a stream of values of type `b` in the IO Monad.
96* A `Fold IO a b` is a representation of a function that converts a stream of
97 type `a` to a final accumulator of type `b` in the IO Monad.
99### A Note on Module Naming
101Some of the examples below use modules from the `Internal` Streamly package
102hierarchy. These are not really internal to the library. We classify
103`Streamly` modules into two categories:
105* _Released modules and APIs_: These modules and APIs are
106 stable. Significant changes to these modules and APIs will cause
107 Streamly's version number to change according to the package versioning
109* _Pre-release modules and APIs_: These modules and APIs have not been
110 formally released yet. They may change in the near future, and such
111 changes will not necessarily be reflected in Streamly's package
112 version number. As yet unreleased modules and APIs reside in the
113 `Internal` namespace.
117### Modular Word Counting
119A `Fold` in Streamly is a composable stream consumer. For our first
120example, we will use `Fold`s to count the number of bytes, words and lines
121present in a file. We will then compose individual `Fold`s together to
122count words, bytes and lines at the same time.
124Please see the file [WordCountModular.hs][] for the complete example
125program, including the imports that we have omitted here.
127#### Count Bytes (wc -c)
129We start with a code fragment that counts the number of bytes in a file:
132import qualified Streamly.Data.Fold as Fold
133import qualified Streamly.Internal.FileSystem.File as File
134import qualified Streamly.Prelude as Stream
136wcb :: String -> IO Int
138 File.toBytes file -- SerialT IO Word8
139 & Stream.fold Fold.length -- IO Int
142### Count Lines (wc -l)
144The next code fragment shows how to count the number of lines in a file:
147-- ASCII character 10 is a newline.
148countl :: Int -> Word8 -> Int
149countl n ch = if ch == 10 then n + 1 else n
151-- The fold accepts a stream of `Word8` and returns a line count (`Int`).
152nlines :: Monad m => Fold m Word8 Int
153nlines = Fold.foldl' countl 0
155wcl :: String -> IO Int
157 File.toBytes file -- SerialT IO Word8
158 & Stream.fold nlines -- IO Int
161### Count Words (wc -w)
163Our final code fragment counts the number of whitespace-separated words
167countw :: (Int, Bool) -> Word8 -> (Int, Bool)
168countw (n, wasSpace) ch =
169 if isSpace $ chr $ fromIntegral ch
171 else (if wasSpace then n + 1 else n, False)
173-- The fold accepts a stream of `Word8` and returns a word count (`Int`).
174nwords :: Monad m => Fold m Word8 Int
175nwords = fst <$> Fold.foldl' countw (0, True)
177wcw :: String -> IO Int
179 File.toBytes file -- SerialT IO Word8
180 & Stream.fold nwords -- IO Int
183### Counting Bytes, Words and Lines Together
185By using the `Tee` combinator we can compose the three folds that count
186bytes, lines and words individually into a single fold that counts all
187three at once. The applicative instance of `Tee` distributes its input
188to all the supplied folds (`Fold.length`, `nlines`, and `nwords`) and
189then combines the outputs from the folds using the supplied combiner
193import qualified Streamly.Internal.Data.Fold.Tee as Tee
195-- The fold accepts a stream of `Word8` and returns the three counts.
196countAll :: Fold IO Word8 (Int, Int, Int)
197countAll = Tee.toFold $ (,,) <$> Tee Fold.length <*> Tee nlines <*> Tee nwords
199wc :: String -> IO (Int, Int, Int)
201 File.toBytes file -- SerialT IO Word8
202 & Stream.fold countAll -- IO (Int, Int, Int)
205This example demonstrates the excellent modularity offered by
206[Streamly][]'s simple and concise API. Experienced Haskellers will
207notice that we have not used bytestrings—we instead used a stream of
208`Word8` values, simplifying our program.
210### The Performance of Word Counting
212We compare two equivalent implementations: one using [Streamly][],
213and the other using C.
215The performance of the [Streamly word counting
216implementation][WordCount.hs] is:
219$ time WordCount-hs gutenberg-500MB.txt
22011242220 97050938 574714449 gutenberg-500MB.txt
227The performance of an equivalent [wc implementation in C][WordCount.c] is:
230$ time WordCount-c gutenberg-500MB.txt
23111242220 97050938 574714449 gutenberg-500MB.txt
238### Concurrent Word Counting
240In our next example we show how the task of counting words, lines,
241and bytes could be done in parallel on multiprocessor hardware.
243To count words in parallel we first divide the stream into chunks
244(arrays), do the counting within each chunk, and then add all the
245counts across chunks. We use the same code as above except that we use
246arrays for our input data.
248Please see the file [WordCountParallel.hs][] for the complete working
249code for this example, including the imports that we have omitted below.
251The `countArray` function counts the line, word, char counts in one chunk:
254import qualified Streamly.Data.Array.Foreign as Array
256countArray :: Array Word8 -> IO Counts
258 Stream.unfold Array.read arr -- SerialT IO Word8
259 & Stream.decodeLatin1 -- SerialT IO Char
260 & Stream.foldl' count (Counts 0 0 0 True) -- IO Counts
263Here the function `count` and the `Counts` data type are defined in the
264`WordCount` helper module defined in [WordCount.hs][].
266When combining the counts in two contiguous chunks, we need to check
267whether the first element of the next chunk is a whitespace character
268in order to determine if the same word continues in the next chunk or
269whether the chunk starts with a new word. The `partialCounts` function
270adds a `Bool` flag to `Counts` returned by `countArray` to indicate
271whether the first character in the chunk is a space.
274partialCounts :: Array Word8 -> IO (Bool, Counts)
275partialCounts arr = do
276 let r = Array.getIndex arr 0
279 counts <- countArray arr
280 return (isSpace (chr (fromIntegral x)), counts)
281 Nothing -> return (False, Counts 0 0 0 True)
284`addCounts` then adds the counts from two consecutive chunks:
287addCounts :: (Bool, Counts) -> (Bool, Counts) -> (Bool, Counts)
288addCounts (sp1, Counts l1 w1 c1 ws1) (sp2, Counts l2 w2 c2 ws2) =
290 if not ws1 && not sp2 -- No space between two chunks.
293 in (sp1, Counts (l1 + l2) wcount (c1 + c2) ws2)
296To count in parallel we now only need to divide the stream into arrays,
297apply our counting function to each array, and then combine the counts
301wc :: String -> IO (Bool, Counts)
303 Stream.unfold File.readChunks file -- AheadT IO (Array Word8)
304 & Stream.mapM partialCounts -- AheadT IO (Bool, Counts)
305 & Stream.maxThreads numCapabilities -- AheadT IO (Bool, Counts)
306 & Stream.fromAhead -- SerialT IO (Bool, Counts)
307 & Stream.foldl' addCounts (False, Counts 0 0 0 True) -- IO (Bool, Counts)
310Please note that the only difference between a concurrent and a
311non-concurrent program lies in the use of the `Stream.fromAhead`
312combinator. If we remove the call to `Stream.fromAhead`, we would
313still have a perfectly valid and performant serial program. Notice
314how succinctly and idiomatically we have expressed the concurrent word
317A benchmark with 2 CPUs:
320$ time WordCount-hs-parallel gutenberg-500MB.txt
32111242220 97050938 574714449 gutenberg-500MB.txt
328These example programs have assumed ASCII encoded input data. For UTF-8
329streams, we have a [concurrent wc implementation][WordCountUTF8.hs]
330with UTF-8 decoding. This concurrent implementation performs as well
331as the standard `wc` program in serial benchmarks. In concurrent mode
332[Streamly][]'s implementation can utilise multiple processing cores if
333these are present, and can thereby run much faster than the standard
336Streamly provides concurrency facilities similar
337to [OpenMP](https://en.wikipedia.org/wiki/OpenMP) and
338[Cilk](https://en.wikipedia.org/wiki/Cilk) but with a more declarative
339style of expression. With Streamly you can write concurrent programs
340with ease, with support for different types of concurrent scheduling.
342### A Concurrent Network Server
344We now move to a slightly more complicated example: we simulate a
345dictionary lookup server which can serve word meanings to multiple
346clients concurrently. This example demonstrates the use of the concurrent
349Please see the file [WordServer.hs][] for the complete code for this
350example, including the imports that we have omitted below.
353import qualified Streamly.Data.Fold as Fold
354import qualified Streamly.Network.Inet.TCP as TCP
355import qualified Streamly.Network.Socket as Socket
356import qualified Streamly.Unicode.Stream as Unicode
358-- Simulate network/db query by adding a delay.
359fetch :: String -> IO (String, String)
360fetch w = threadDelay 1000000 >> return (w,w)
362-- Read lines of whitespace separated list of words from a socket, fetch the
363-- meanings of each word concurrently and return the meanings separated by
364-- newlines, in same order as the words were received. Repeat until the
365-- connection is closed.
366lookupWords :: Socket -> IO ()
368 Stream.unfold Socket.read sk -- SerialT IO Word8
369 & Unicode.decodeLatin1 -- SerialT IO Char
370 & Stream.wordsBy isSpace Fold.toList -- SerialT IO String
371 & Stream.fromSerial -- AheadT IO String
372 & Stream.mapM fetch -- AheadT IO (String, String)
373 & Stream.fromAhead -- SerialT IO (String, String)
374 & Stream.map show -- SerialT IO String
375 & Stream.intersperse "\n" -- SerialT IO String
376 & Unicode.encodeStrings Unicode.encodeLatin1 -- SerialT IO (Array Word8)
377 & Stream.fold (Socket.writeChunks sk) -- IO ()
379serve :: Socket -> IO ()
380serve sk = finally (lookupWords sk) (close sk)
382-- | Run a server on port 8091. Accept and handle connections concurrently. The
383-- connection handler is "serve" (i.e. lookupWords). You can use "telnet" or
384-- "nc" as a client to try it out.
387 Stream.unfold TCP.acceptOnPort 8091 -- SerialT IO Socket
388 & Stream.fromSerial -- AsyncT IO ()
389 & Stream.mapM serve -- AsyncT IO ()
390 & Stream.fromAsync -- SerialT IO ()
391 & Stream.drain -- IO ()
394### Merging Incoming Streams
396In the next example, we show how to merge logs coming from multiple
397nodes in your network. These logs are merged at line boundaries and
398the merged logs are written to a file or to a network destination.
399This example uses the `concatMapWith` combinator to merge multiple
402Please see the file [MergeServer.hs][] for the complete working code,
403including the imports that we have omitted below.
406import qualified Streamly.Data.Unfold as Unfold
407import qualified Streamly.Network.Socket as Socket
409-- | Read a line stream from a socket.
410-- Note: lines are buffered, and we could add a limit to the
411-- buffering for safety.
412readLines :: Socket -> SerialT IO (Array Char)
414 Stream.unfold Socket.read sk -- SerialT IO Word8
415 & Unicode.decodeLatin1 -- SerialT IO Char
416 & Stream.splitWithSuffix (== '\n') Array.write -- SerialT IO String
418recv :: Socket -> SerialT IO (Array Char)
419recv sk = Stream.finally (liftIO $ close sk) (readLines sk)
421-- | Starts a server at port 8091 listening for lines with space separated
422-- words. Multiple clients can connect to the server and send streams of lines.
423-- The server handles all the connections concurrently, merges the incoming
424-- streams at line boundaries and writes the merged stream to a file.
425server :: Handle -> IO ()
427 Stream.unfold TCP.acceptOnPort 8090 -- SerialT IO Socket
428 & Stream.concatMapWith Stream.parallel recv -- SerialT IO (Array Char)
429 & Stream.unfoldMany Array.read -- SerialT IO Char
430 & Unicode.encodeLatin1 -- SerialT IO Word8
431 & Stream.fold (Handle.write file) -- IO ()
434main = withFile "output.txt" AppendMode server
437### Listing Directories Recursively/Concurrently
439Our next example lists a directory tree recursively, reading
440multiple directories concurrently.
442This example uses the tree traversing combinator `iterateMapLeftsWith`.
443This combinator maps a stream generator on the `Left` values in its
444input stream (directory names in this case), feeding the resulting `Left`
445values back to the input, while it lets the `Right` values (file names
446in this case) pass through to the output. The `Stream.ahead` stream
447joining combinator then makes it iterate on the directories concurrently.
449Please see the file [ListDir.hs][] for the complete working code,
450including the imports that we have omitted below.
453import Streamly.Internal.Data.Stream.IsStream (iterateMapLeftsWith)
455import qualified Streamly.Prelude as Stream
456import qualified Streamly.Internal.FileSystem.Dir as Dir (toEither)
458-- Lists a directory as a stream of (Either Dir File).
459listDir :: String -> SerialT IO (Either String String)
461 Dir.toEither dir -- SerialT IO (Either String String)
462 & Stream.map (bimap mkAbs mkAbs) -- SerialT IO (Either String String)
464 where mkAbs x = dir ++ "/" ++ x
466-- | List the current directory recursively using concurrent processing.
469 hSetBuffering stdout LineBuffering
470 let start = Stream.fromPure (Left ".")
471 Stream.iterateMapLeftsWith Stream.ahead listDir start
477For bounded concurrent streams, a stream yield rate can be specified
478easily. For example, to print "tick" once every second you can simply
484 Stream.repeatM (pure "tick") -- AsyncT IO String
485 & Stream.timestamped -- AsyncT IO (AbsTime, String)
486 & Stream.avgRate 1 -- AsyncT IO (AbsTime, String)
487 & Stream.fromAsync -- SerialT IO (AbsTime, String)
488 & Stream.mapM_ print -- IO ()
491Please see the file [Rate.hs][] for the complete working code.
493The concurrency of the stream is automatically controlled to match the
494specified rate. [Streamly][]'s rate control works precisely even at
495throughputs as high as millions of yields per second.
497For more sophisticated rate control needs please see the Streamly [reference
498documentation][Streamly].
500### Reactive Programming
502Streamly supports reactive (time domain) programming because of its
503support for declarative concurrency. Please see the `Streamly.Prelude`
504module for time-specific combinators like `intervalsOf`, and
505folds like `takeInterval` in `Streamly.Internal.Data.Fold`.
506Please also see the pre-release sampling combinators in the
507`Streamly.Internal.Data.Stream.IsStream.Top` module for `throttle` and
508`debounce` like operations.
510The examples [AcidRain.hs][] and [CirclingSquare.hs][] demonstrate
511reactive programming using [Streamly][].
515If you would like to view more examples, please visit the [Streamly
516Examples][streamly-examples] web page.
520* [Streaming Benchmarks][streaming-benchmarks]
521* [Concurrency Benchmarks][concurrency-benchmarks]
522* Functional Conf 2019 [Video](https://www.youtube.com/watch?v=uzsqgdMMgtk) | [Slides](https://www.slideshare.net/HarendraKumar10/streamly-concurrent-data-flow-programming)
523* [Other Guides](docs/)
524* [Streamly Homepage][Streamly]
528As you have seen in the word count example above, [Streamly][] offers
529highly modular abstractions for building programs while also offering
530the performance close to an equivalent (imperative) C program.
532Streamly offers excellent performance even for byte-at-a-time stream
533operations using efficient abstractions like `Unfold`s and terminating
534`Fold`s. Byte-at-a-time stream operations can simplify programming
535because the developer does not have to deal explicitly with chunking
536and re-combining data.
538Streamly exploits GHC's stream fusion optimizations (`case-of-case` and
539`spec-constr`) aggressively to achieve C-like speed, while also offering
540highly modular abstractions to developers.
542[Streamly][] will usually perform very well without any
543compiler plugins. However, we have fixed some deficiencies
544that we had noticed in GHC's optimizer using a [compiler
545plugin](https://github.com/composewell/fusion-plugin). We hope to fold
546these optimizations into GHC in the future; until then we recommend that
547you use this plugin for applications that are performance sensitive.
551We measured several Haskell streaming implementations
552using various micro-benchmarks. Please see the [streaming
553benchmarks][streaming-benchmarks] page for a detailed comparison of
554Streamly against other streaming libraries.
556Our results show that [Streamly][] is the fastest effectful streaming
557implementation on almost all the measured microbenchmarks. In many cases
558it runs up to 100x faster, and in some cases even 1000x faster than
559some of the tested alternatives. In some composite operation benchmarks
560[Streamly][] turns out to be significantly faster than Haskell's list
563*Note*: If you can write a program in some other way or with some other
564language that runs significantly faster than what [Streamly][] offers,
565please let us know and we will improve.
569Streamly comes equipped with a very powerful set of abstractions to
570accomplish many kinds of programming tasks: it provides support for
571programming with streams and arrays, for reading and writing from the
572file system and from the network, for time domain programming (reactive
573programming), and for reacting to file system events using `fsnotify`.
575Please view [Streamly's documentation][Streamly] for more information
576about Streamly's features.
580Streamly uses lock-free synchronization for achieving concurrent
581operation with low overheads. The number of tasks performed concurrently
582are determined automatically based on the rate at which a consumer
583is consuming the results. In other words, you do not need to manage
584thread pools or decide how many threads to use for a particular task.
585For CPU-bound tasks Streamly will try to keep the number of threads
586close to the number of CPUs available; for IO-bound tasks it will utilize
589The parallelism available during program execution can be utilized with
590very little overhead even where the task size is very
591small, because Streamly will automatically switch between
592serial or batched execution of tasks on the same CPU depending
593on whichever is more efficient. Please see our [concurrency
594benchmarks][concurrency-benchmarks] for more detailed performance
595measurements, and for a comparison with the `async` package.
599Our goals for [Streamly][] from the very beginning have been:
6011. To achieve simplicity by unifying abstractions.
6022. To offer high performance.
604These goals are hard to achieve simultaneously because they are usually
605inversely related. We have spent many years trying to get the abstractions
606right without compromising performance.
608`Unfold` is an example of an abstraction that we have created to achieve
609high performance when mapping streams on streams. `Unfold` allows stream
610generation to be optimized well by the compiler through stream fusion.
611A `Fold` with termination capability is another example which modularizes
612stream elimination operations through stream fusion. Terminating folds
613can perform many simple parsing tasks that do not require backtracking.
614In Streamly, `Parser`s are a natural extension to terminating `Fold`s;
615`Parser`s add the ability to backtrack to `Fold`s. Unification leads
616to simpler abstractions and lower cognitive overheads while also not
617compromising performance.
621The following authors/libraries have influenced or inspired this library in a
624 * Roman Leshchinskiy ([vector](http://hackage.haskell.org/package/vector))
625 * Gabriel Gonzalez ([foldl](https://hackage.haskell.org/package/foldl))
626 * Alberto G. Corona ([transient](https://hackage.haskell.org/package/transient))
628Please see the [`credits`](./credits/README.md) directory for a full
629list of contributors, credits and licenses.
633Streamly is an [open source](https://github.com/composewell/streamly)
634project available under a liberal [BSD-3-Clause license][LICENSE]
636## Contributing to Streamly
638As an open project we welcome contributions:
640* [Streamly Contributor's Guide][CONTRIBUTING.md]
641* [Contact the streamly development team](mailto:streamly@composewell.com)
645Professional support is available for [Streamly][]: please contact
646[support@composewell.com](mailto:support@composewell.com).
648You can also join our [community chat
649channel](https://gitter.im/composewell/streamly) on Gitter.
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