README.md · streamly · git.iko.soy

git.iko.soy/streamly

on 5054813 help sign in

streamly/README.md

612 lines 25 kB history raw

1# Streamly

3[![Hackage](https://img.shields.io/hackage/v/streamly.svg?style=flat)](https://hackage.haskell.org/package/streamly)

4[![Gitter chat](https://badges.gitter.im/composewell/gitter.svg)](https://gitter.im/composewell/streamly)

5[![Travis](https://travis-ci.com/composewell/streamly.svg?branch=master)](https://travis-ci.com/composewell/streamly)

6[![Appveyor](https://ci.appveyor.com/api/projects/status/ajxg0c79raou9ned?svg=true)](https://ci.appveyor.com/project/harendra-kumar/streamly)

7[![CircleCI](https://circleci.com/gh/composewell/streamly/tree/master.svg?style=svg)](https://circleci.com/gh/composewell/streamly/tree/master)

8[![Coverage Status](https://coveralls.io/repos/composewell/streamly/badge.svg?branch=master&service=github)](https://coveralls.io/github/composewell/streamly?branch=master)

10## Learning Materials

12* [Overview: this readme](#streaming-concurrently) | [Tutorial](https://hackage.haskell.org/package/streamly/docs/Streamly-Tutorial.html) | [Reference](https://hackage.haskell.org/package/streamly) | [Guides](docs)

13* [Installing](./INSTALL.md) | [Building for optimal performance](docs/Build.md)

14* Examples: [in this repo](examples) | [streamly-examples](https://github.com/composewell/streamly-examples)

15* Benchmarks: [Streaming](https://github.com/composewell/streaming-benchmarks) | [Concurrency](https://github.com/composewell/concurrency-benchmarks)

16* Talks: [Functional Conf 2019 Video](https://www.youtube.com/watch?v=uzsqgdMMgtk) | [Functional Conf 2019 Slides](https://www.slideshare.net/HarendraKumar10/streamly-concurrent-data-flow-programming)

18## Streaming Concurrently

20Haskell lists express pure computations using composable stream operations like

21`:`, `unfold`, `map`, `filter`, `zip` and `fold`. Streamly is exactly like

22lists except that it can express sequences of pure as well as monadic

23computations aka streams. More importantly, it can express monadic sequences

24with concurrent execution semantics without introducing any additional APIs.

26Streamly expresses concurrency using standard, well known abstractions.

27Concurrency semantics are defined for list operations, semigroup, applicative

28and monadic compositions. Programmer does not need to know any low level

29notions of concurrency like threads, locking or synchronization. Concurrent

30and non-concurrent programs are fundamentally the same. A chosen segment of

31the program can be made concurrent by annotating it with an appropriate

32combinator. We can choose a combinator for lookahead style or asynchronous

33concurrency. Concurrency is automatically scaled up or down based on the

34demand from the consumer application, we can finally say goodbye to managing

35thread pools and associated sizing issues. The result is truly fearless

36and declarative monadic concurrency.

38## Where to use streamly?

40Streamly is a general purpose programming framework. It can be used equally

41efficiently from a simple `Hello World!` program to a massively concurrent

42application. The answer to the question, "where to use streamly?" - would be

43similar to the answer to - "Where to use Haskell lists or the IO monad?".

45Streamly simplifies streaming and makes it as intuitive as plain lists. Unlike

46other streaming libraries, no fancy types are required. Streamly is simply a

47generalization of Haskell lists to monadic streaming optionally with concurrent

48composition. The basic stream type in streamly `SerialT m a` can be considered

49as a list type `[a]` parameterized by the monad `m`. For example, `SerialT IO

50a` is a moral equivalent of `[a]` in the IO monad. `SerialT Identity a`, is

51equivalent to pure lists. Streams are constructed very much like lists, except

52that they use `nil` and `cons` instead of `[]` and `:`. Unlike lists, streams

53can be constructed from monadic effects, not just pure elements. Streams are

54processed just like lists, with list like combinators, except that they are

55monadic and work in a streaming fashion. In other words streamly just completes

56what lists lack, you do not need to learn anything new. Please see [streamly vs

57lists](docs/streamly-vs-lists.md) for a detailed comparison.

59Not surprisingly, the monad instance of streamly is a list transformer, with

60concurrency capability.

62## Why data flow programming?

64If you need some convincing for using streaming or data flow programming

65paradigm itself then try to answer this question - why do we use lists in

66Haskell? It boils down to why we use functional programming in the first place.

67Haskell is successful in enforcing the functional data flow paradigm for pure

68computations using lists, but not for monadic computations. In the absence of a

69standard and easy to use data flow programming paradigm for monadic

70computations, and the IO monad providing an escape hatch to an imperative

71model, we just love to fall into the imperative trap, and start asking the same

72fundamental question again - why do we have to use the streaming data model?

74## Comparative Performance

76High performance and simplicity are the two primary goals of streamly.

77`Streamly` employs two different stream representations (CPS and direct style)

78and interconverts between the two to get the best of both worlds on different

79operations. It uses both foldr/build (for CPS style) and stream fusion (for

80direct style) techniques to fuse operations. In terms of performance,

81Streamly's goal is to compete with equivalent C programs. Streamly redefines

82"blazing fast" for streaming libraries, it competes with lists and `vector`.

83Other streaming libraries like "streaming", "pipes" and "conduit" are orders of

84magnitude slower on most microbenchmarks. See [streaming

85benchmarks](https://github.com/composewell/streaming-benchmarks) for detailed

86comparison.

88The following chart shows a comparison of those streamly and list operations

89where performance of the two differs by more than 10%. Positive y-axis displays

90how many times worse is a list operation compared to the same streamly

91operation, negative y-axis shows where streamly is worse compared to lists.

93![Streamly vs Lists (time) comparison](charts-0/streamly-vs-list-time.svg)

95Streamly uses lock-free synchronization for concurrent operations. It employs

96auto-scaling of the degree of concurrency based on demand. For CPU bound tasks

97it tries to keep the threads close to the number of CPUs available whereas for

98IO bound tasks more threads can be utilized. Parallelism can be utilized with

99little overhead even if the task size is very small. See [concurrency

100benchmarks](https://github.com/composewell/concurrency-benchmarks) for detailed

101performance results and a comparison with the `async` package.

103## Installing and using

105Please see [INSTALL.md](./INSTALL.md) for instructions on how to use streamly

106with your Haskell build tool or package manager. You may want to go through it

107before jumping to run the examples below.

109The module `Streamly` provides just the core stream types, type casting and

110concurrency control combinators. Stream construction, transformation, folding,

111merging, zipping combinators are found in `Streamly.Prelude`.

113## Streaming Pipelines

115The following snippet provides a simple stream composition example that reads

116numbers from stdin, prints the squares of even numbers and exits if an even

117number more than 9 is entered.

119``` haskell

120import Streamly

121import qualified Streamly.Prelude as S

122import Data.Function ((&))

124main = S.drain $

125 S.repeatM getLine

126 & fmap read

127 & S.filter even

128 & S.takeWhile (<= 9)

129 & fmap (\x -> x * x)

130 & S.mapM print

131```

133Unlike `pipes` or `conduit` and like `vector` and `streaming`, `streamly`

134composes stream data instead of stream processors (functions). A stream is

135just like a list and is explicitly passed around to functions that process the

136stream. Therefore, no special operator is needed to join stages in a streaming

137pipeline, just the standard function application (`$`) or reverse function

138application (`&`) operator is enough.

140## Concurrent Stream Generation

142`consM` or its operator form `|:` can be used to construct a stream from

143monadic actions. A stream constructed with `consM` can run the monadic actions

144in the stream concurrently when used with appropriate stream type combinator

145(e.g. `asyncly`, `aheadly` or `parallely`).

147The following code finishes in 3 seconds (6 seconds when serial), note the

148order of elements in the resulting output, the outputs are consumed as soon as

149each action is finished (asyncly):

151``` haskell

152> let p n = threadDelay (n * 1000000) >> return n

153> S.toList $ asyncly $ p 3 |: p 2 |: p 1 |: S.nil

154[1,2,3]

155```

157Use `aheadly` if you want speculative concurrency i.e. execute the actions in

158the stream concurrently but consume the results in the specified order:

160``` haskell

161> S.toList $ aheadly $ p 3 |: p 2 |: p 1 |: S.nil

162[3,2,1]

163```

165Monadic stream generation functions e.g. `unfoldrM`, `replicateM`, `repeatM`,

166`iterateM` and `fromFoldableM` etc. can work concurrently.

168The following finishes in 10 seconds (100 seconds when serial):

170``` haskell

171S.drain $ asyncly $ S.replicateM 10 $ p 10

172```

174## Concurrency Auto Scaling

176Concurrency is auto-scaled i.e. more actions are executed concurrently if the

177consumer is consuming the stream at a higher speed. How many tasks are executed

178concurrently can be controlled by `maxThreads` and how many results are

179buffered ahead of consumption can be controlled by `maxBuffer`. See the

180documentation in the `Streamly` module.

182## Concurrent Streaming Pipelines

184Use `|&` or `|$` to apply stream processing functions concurrently. The

185following example prints a "hello" every second; if you use `&` instead of

186`|&` you will see that the delay doubles to 2 seconds instead because of serial

187application.

189``` haskell

190main = S.drain $

191 S.repeatM (threadDelay 1000000 >> return "hello")

192 |& S.mapM (\x -> threadDelay 1000000 >> putStrLn x)

193```

195## Mapping Concurrently

197We can use `mapM` or `sequence` functions concurrently on a stream.

199``` haskell

200> let p n = threadDelay (n * 1000000) >> return n

201> S.drain $ aheadly $ S.mapM (\x -> p 1 >> print x) (serially $ repeatM (p 1))

202```

204## Serial and Concurrent Merging

206Semigroup and Monoid instances can be used to fold streams serially or

207concurrently. In the following example we compose ten actions in the

208stream, each with a delay of 1 to 10 seconds, respectively. Since all the

209actions are concurrent we see one output printed every second:

211``` haskell

212import Streamly

213import qualified Streamly.Prelude as S

214import Control.Concurrent (threadDelay)

216main = S.toList $ parallely $ foldMap delay [1..10]

217 where delay n = S.yieldM $ threadDelay (n * 1000000) >> print n

218```

220Streams can be combined together in many ways. We provide some examples

221below, see the tutorial for more ways. We use the following `delay`

222function in the examples to demonstrate the concurrency aspects:

224``` haskell

225import Streamly

226import qualified Streamly.Prelude as S

227import Control.Concurrent

229delay n = S.yieldM $ do

230 threadDelay (n * 1000000)

231 tid <- myThreadId

232 putStrLn (show tid ++ ": Delay " ++ show n)

233```

234### Serial

236``` haskell

237main = S.drain $ delay 3 <> delay 2 <> delay 1

238```

239```

240ThreadId 36: Delay 3

241ThreadId 36: Delay 2

242ThreadId 36: Delay 1

243```

245### Parallel

247``` haskell

248main = S.drain . parallely $ delay 3 <> delay 2 <> delay 1

249```

250```

251ThreadId 42: Delay 1

252ThreadId 41: Delay 2

253ThreadId 40: Delay 3

254```

256## Nested Loops (aka List Transformer)

258The monad instance composes like a list monad.

260``` haskell

261import Streamly

262import qualified Streamly.Prelude as S

264loops = do

265 x <- S.fromFoldable [1,2]

266 y <- S.fromFoldable [3,4]

267 S.yieldM $ putStrLn $ show (x, y)

269main = S.drain loops

270```

271```

272(1,3)

273(1,4)

274(2,3)

275(2,4)

276```

278## Concurrent Nested Loops

280To run the above code with speculative concurrency i.e. each iteration in the

281loop can run concurrently but the results are presented to the consumer of the

282output in the same order as serial execution:

284``` haskell

285main = S.drain $ aheadly $ loops

286```

288Different stream types execute the loop iterations in different ways. For

289example, `wSerially` interleaves the loop iterations. There are several

290concurrent stream styles to execute the loop iterations concurrently in

291different ways, see the `Streamly.Tutorial` module for a detailed treatment.

293## Magical Concurrency

295Streams can perform semigroup (<>) and monadic bind (>>=) operations

296concurrently using combinators like `asyncly`, `parallelly`. For example,

297to concurrently generate squares of a stream of numbers and then concurrently

298sum the square roots of all combinations of two streams:

300``` haskell

301import Streamly

302import qualified Streamly.Prelude as S

304main = do

305 s <- S.sum $ asyncly $ do

306 -- Each square is performed concurrently, (<>) is concurrent

307 x2 <- foldMap (\x -> return $ x * x) [1..100]

308 y2 <- foldMap (\y -> return $ y * y) [1..100]

309 -- Each addition is performed concurrently, monadic bind is concurrent

310 return $ sqrt (x2 + y2)

311 print s

312```

314The concurrency facilities provided by streamly can be compared with

315[OpenMP](https://en.wikipedia.org/wiki/OpenMP) and

316[Cilk](https://en.wikipedia.org/wiki/Cilk) but with a more declarative

317expression.

319## Example: Listing Directories Recursively/Concurrently

321The following code snippet lists a directory tree recursively, reading multiple

322directories concurrently:

324```haskell

325import Control.Monad.IO.Class (liftIO)

326import Path.IO (listDir, getCurrentDir) -- from path-io package

327import Streamly (AsyncT, adapt)

328import qualified Streamly.Prelude as S

330listDirRecursive :: AsyncT IO ()

331listDirRecursive = getCurrentDir >>= readdir >>= liftIO . mapM_ putStrLn

332 where

333 readdir dir = do

334 (dirs, files) <- listDir dir

335 S.yield (map show dirs ++ map show files) <> foldMap readdir dirs

337main :: IO ()

338main = S.drain $ adapt $ listDirRecursive

339```

341`AsyncT` is a stream monad transformer. If you are familiar with a list

342transformer, it is nothing but `ListT` with concurrency semantics. For example,

343the semigroup operation `<>` is concurrent. This makes `foldMap` concurrent

344too. You can replace `AsyncT` with `SerialT` and the above code will become

345serial, exactly equivalent to a `ListT`.

347## Rate Limiting

349For bounded concurrent streams, stream yield rate can be specified. For

350example, to print hello once every second you can simply write this:

352``` haskell

353import Streamly

354import Streamly.Prelude as S

356main = S.drain $ asyncly $ avgRate 1 $ S.repeatM $ putStrLn "hello"

357```

359For some practical uses of rate control, see

360[AcidRain.hs](https://github.com/composewell/streamly/tree/master/examples/AcidRain.hs)

361and

362[CirclingSquare.hs](https://github.com/composewell/streamly/tree/master/examples/CirclingSquare.hs)

364Concurrency of the stream is automatically controlled to match the specified

365rate. Rate control works precisely even at throughputs as high as millions of

366yields per second. For more sophisticated rate control see the haddock

367documentation.

369## Arrays

371The `Streamly.Memory.Array` module provides immutable arrays. Arrays are the

372computing duals of streams. Streams are good at sequential access and immutable

373transformations of in-transit data whereas arrays are good at random access and

374in-place transformations of buffered data. Unlike streams which are potentially

375infinite, arrays are necessarily finite. Arrays can be used as an efficient

376interface between streams and external storage systems like memory, files and

377network. Streams and arrays complete each other to provide a general purpose

378computing system. The design of streamly as a general purpose computing

379framework is centered around these two fundamental aspects of computing and

380storage.

382`Streamly.Memory.Array` uses pinned memory outside GC and therefore avoid any

383GC overhead for the storage in arrays. Streamly allows efficient

384transformations over arrays using streams. It uses arrays to transfer data to

385and from the operating system and to store data in memory.

387## Folds

389Folds are consumers of streams. `Streamly.Data.Fold` module provides a `Fold`

390type that represents a `foldl'`. Such folds can be efficiently composed

391allowing the compiler to perform stream fusion and therefore implement high

392performance combinators for consuming streams. A stream can be distributed to

393multiple folds, or it can be partitioned across multiple folds, or

394demultiplexed over multiple folds, or unzipped to two folds. We can also use

395folds to fold segments of stream generating a stream of the folded results.

397If you are familiar with the `foldl` library, these are the same composable

398left folds but simpler and better integrated with streamly, and with many more

399powerful ways of composing and applying them.

401## Unfolds

403Unfolds are duals of folds. Folds help us compose consumers of streams

404efficiently and unfolds help us compose producers of streams efficiently.

405`Streamly.Data.Unfold` provides an `Unfold` type that represents an `unfoldr`

406or a stream generator. Such generators can be combined together efficiently

407allowing the compiler to perform stream fusion and implement high performance

408stream merging combinators.

410## File IO

412The following code snippets implement some common Unix command line utilities

413using streamly. You can compile these with `ghc -O2 -fspec-constr-recursive=16

414-fmax-worker-args=16` and compare the performance with regular GNU coreutils

415available on your system. Though many of these are not most optimal solutions

416to keep them short and elegant. Source file

417[HandleIO.hs](https://github.com/composewell/streamly/tree/master/examples/HandleIO.hs)

418in the examples directory includes these examples.

420``` haskell

421module Main where

423import qualified Streamly.Prelude as S

424import qualified Streamly.Data.Fold as FL

425import qualified Streamly.Memory.Array as A

426import qualified Streamly.FileSystem.Handle as FH

427import qualified System.IO as FH

429import Data.Char (ord)

430import System.Environment (getArgs)

431import System.IO (openFile, IOMode(..), stdout)

433withArg f = do

434 (name : _) <- getArgs

435 src <- openFile name ReadMode

436 f src

438withArg2 f = do

439 (sname : dname : _) <- getArgs

440 src <- openFile sname ReadMode

441 dst <- openFile dname WriteMode

442 f src dst

443```

445### cat

447``` haskell

448cat = S.fold (FH.writeChunks stdout) . S.unfold FH.readChunks

449main = withArg cat

450```

452### cp

454``` haskell

455cp src dst = S.fold (FH.writeChunks dst) $ S.unfold FH.readChunks src

456main = withArg2 cp

457```

459### wc -l

461``` haskell

462wcl = S.length . S.splitOn (== 10) FL.drain . S.unfold FH.read

463main = withArg wcl >>= print

464```

466### Average Line Length

468``` haskell

469avgll =

470 S.fold avg

471 . S.splitOn (== 10) FL.length

472 . S.unfold FH.read

474 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length

475 toDouble = fmap (fromIntegral :: Int -> Double)

477main = withArg avgll >>= print

478```

480### Line Length Histogram

482`classify` is not released yet, and is available in

483`Streamly.Internal.Data.Fold`

485``` haskell

486llhisto =

487 S.fold (FL.classify FL.length)

488 . S.map bucket

489 . S.splitOn (== 10) FL.length

490 . S.unfold FH.read

492 where

493 bucket n = let i = n `mod` 10 in if i > 9 then (9,n) else (i,n)

495main = withArg llhisto >>= print

496```

498## Socket IO

500Its easy to build concurrent client and server programs using streamly.

501`Streamly.Network.*` modules provide easy combinators to build network servers

502and client programs using streamly. See

503[FromFileClient.hs](https://github.com/composewell/streamly/tree/master/examples/FromFileClient.hs),

504[EchoServer.hs](https://github.com/composewell/streamly/tree/master/examples/EchoServer.hs),

505[FileSinkServer.hs](https://github.com/composewell/streamly/tree/master/examples/FileSinkServer.hs)

506in the examples directory.

508## Exceptions

510Exceptions can be thrown at any point using the `MonadThrow` instance. Standard

511exception handling combinators like `bracket`, `finally`, `handle`,

512`onException` are provided in `Streamly.Prelude` module.

514In presence of concurrency, synchronous exceptions work just the way they are

515supposed to work in non-concurrent code. When concurrent streams

516are combined together, exceptions from the constituent streams are propagated

517to the consumer stream. When an exception occurs in any of the constituent

518streams other concurrent streams are promptly terminated.

520There is no notion of explicit threads in streamly, therefore, no

521asynchronous exceptions to deal with. You can just ignore the zillions of

522blogs, talks, caveats about async exceptions. Async exceptions just don't

523exist. Please don't use things like `myThreadId` and `throwTo` just for fun!

525## Reactive Programming (FRP)

527Streamly is a foundation for first class reactive programming as well by virtue

528of integrating concurrency and streaming. See

529[AcidRain.hs](https://github.com/composewell/streamly/tree/master/examples/AcidRain.hs)

530for a console based FRP game example and

531[CirclingSquare.hs](https://github.com/composewell/streamly/tree/master/examples/CirclingSquare.hs)

532for an SDL based animation example.

534## Conclusion

536Streamly, short for streaming concurrently, provides monadic streams, with a

537simple API, almost identical to standard lists, and an in-built

538support for concurrency. By using stream-style combinators on stream

539composition, streams can be generated, merged, chained, mapped, zipped, and

540consumed concurrently – providing a generalized high level programming

541framework unifying streaming and concurrency. Controlled concurrency allows

542even infinite streams to be evaluated concurrently. Concurrency is auto scaled

543based on feedback from the stream consumer. The programmer does not have to be

544aware of threads, locking or synchronization to write scalable concurrent

545programs.

547Streamly is a programmer first library, designed to be useful and friendly to

548programmers for solving practical problems in a simple and concise manner. Some

549key points in favor of streamly are:

551 * _Simplicity_: Simple list like streaming API, if you know how to use lists

552 then you know how to use streamly. This library is built with simplicity

553 and ease of use as a design goal.

554 * _Concurrency_: Simple, powerful, and scalable concurrency. Concurrency is

555 built-in, and not intrusive, concurrent programs are written exactly the

556 same way as non-concurrent ones.

557 * _Generality_: Unifies functionality provided by several disparate packages

558 (streaming, concurrency, list transformer, logic programming, reactive

559 programming) in a concise API.

560 * _Performance_: Streamly is designed for high performance. It employs stream

561 fusion optimizations for best possible performance. Serial peformance is

562 equivalent to the venerable `vector` library in most cases and even better

563 in some cases. Concurrent performance is unbeatable. See

564 [streaming-benchmarks](https://github.com/composewell/streaming-benchmarks)

565 for a comparison of popular streaming libraries on micro-benchmarks.

567The basic streaming functionality of streamly is equivalent to that provided by

568streaming libraries like

569[vector](https://hackage.haskell.org/package/vector),

570[streaming](https://hackage.haskell.org/package/streaming),

571[pipes](https://hackage.haskell.org/package/pipes), and

572[conduit](https://hackage.haskell.org/package/conduit).

573In addition to providing streaming functionality, streamly subsumes

574the functionality of list transformer libraries like `pipes` or

575[list-t](https://hackage.haskell.org/package/list-t), and also the logic

576programming library [logict](https://hackage.haskell.org/package/logict). On

577the concurrency side, it subsumes the functionality of the

578[async](https://hackage.haskell.org/package/async) package, and provides even

579higher level concurrent composition. Because it supports

580streaming with concurrency we can write FRP applications similar in concept to

581[Yampa](https://hackage.haskell.org/package/Yampa) or

582[reflex](https://hackage.haskell.org/package/reflex).

584See the `Comparison with existing packages` section at the end of the

585[tutorial](https://hackage.haskell.org/package/streamly/docs/Streamly-Tutorial.html).

587## Support

589Please feel free to ask questions on the

590[streamly gitter channel](https://gitter.im/composewell/streamly).

591If you require professional support, consulting, training or timely

592enhancements to the library please contact

593[support@composewell.com](mailto:support@composewell.com).

595## Credits

597The following authors/libraries have influenced or inspired this library in a

598significant way:

600 * Roman Leshchinskiy (vector)

601 * Gabriel Gonzalez (foldl)

602 * Alberto G. Corona (transient)

604See the `credits` directory for full list of contributors, credits and licenses.

606## Contributing

608The code is available under BSD-3 license

609[on github](https://github.com/composewell/streamly). Join the [gitter

610chat](https://gitter.im/composewell/streamly) channel for discussions. Please

611ask any questions on the gitter channel or [contact the maintainer

612directly](mailto:streamly@composewell.com). All contributions are welcome!