README.md · streamly · git.iko.soy

git.iko.soy/streamly

on ec19422 help sign in

streamly/README.md

606 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* Documentation: [Quick](#streaming-concurrently) | [Tutorial](https://hackage.haskell.org/package/streamly/docs/Streamly-Tutorial.html) | [Reference (Hackage)](https://hackage.haskell.org/package/streamly) | [Reference (Latest)](https://composewell.github.io/streamly) | [Guides](docs)

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

14* Examples: [streamly](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.Prelude` provides the core stream types and combinators

110for type casting, controlling concurrency, stream construction, transformation,

111folding, merging, and zipping.

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 qualified Streamly.Prelude as S

121import Data.Function ((&))

123main = S.drain $

124 S.repeatM getLine

125 & fmap read

126 & S.filter even

127 & S.takeWhile (<= 9)

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

129 & S.mapM print

130```

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

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

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

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

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

137application (`&`) operator is enough.

139## Concurrent Stream Generation

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

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

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

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

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

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

148each action is finished (asyncly):

150``` haskell

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

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

153[1,2,3]

154```

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

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

159``` haskell

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

161[3,2,1]

162```

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

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

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

169``` haskell

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

171```

173## Concurrency Auto Scaling

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

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

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

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

179documentation in the `Streamly.Prelude` module.

181## Concurrent Streaming Pipelines

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

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

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

186application.

188``` haskell

189main = S.drain $

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

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

192```

194## Mapping Concurrently

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

198``` haskell

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

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

201```

203## Serial and Concurrent Merging

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

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

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

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

210``` haskell

211import qualified Streamly.Prelude as S

212import Control.Concurrent (threadDelay)

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

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

216```

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

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

220function in the examples to demonstrate the concurrency aspects:

222``` haskell

223import qualified Streamly.Prelude as S

224import Control.Concurrent

226delay n = S.yieldM $ do

227 threadDelay (n * 1000000)

228 tid <- myThreadId

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

230```

231### Serial

233``` haskell

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

235```

236```

237ThreadId 36: Delay 3

238ThreadId 36: Delay 2

239ThreadId 36: Delay 1

240```

242### Parallel

244``` haskell

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

246```

247```

248ThreadId 42: Delay 1

249ThreadId 41: Delay 2

250ThreadId 40: Delay 3

251```

253## Nested Loops (aka List Transformer)

255The monad instance composes like a list monad.

257``` haskell

258import qualified Streamly.Prelude as S

260loops = do

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

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

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

265main = S.drain loops

266```

267```

268(1,3)

269(1,4)

270(2,3)

271(2,4)

272```

274## Concurrent Nested Loops

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

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

278output in the same order as serial execution:

280``` haskell

281main = S.drain $ S.aheadly $ loops

282```

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

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

286concurrent stream styles to execute the loop iterations concurrently in

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

289## Magical Concurrency

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

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

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

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

296``` haskell

297import qualified Streamly.Prelude as S

299main = do

300 s <- S.sum $ S.asyncly $ do

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

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

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

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

305 return $ sqrt (x2 + y2)

306 print s

307```

309The concurrency facilities provided by streamly can be compared with

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

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

312expression.

314## Example: Listing Directories Recursively/Concurrently

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

317directories concurrently:

319```haskell

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

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

322import Streamly.Prelude (AsyncT, adapt)

323import qualified Streamly.Prelude as S

325listDirRecursive :: AsyncT IO ()

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

327 where

328 readdir dir = do

329 (dirs, files) <- listDir dir

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

332main :: IO ()

333main = S.drain $ adapt $ listDirRecursive

334```

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

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

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

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

340serial, exactly equivalent to a `ListT`.

342## Rate Limiting

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

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

347``` haskell

348import Streamly.Prelude as S

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

351```

353For some practical uses of rate control, see

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

355and

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

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

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

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

361documentation.

363## Arrays

365The `Streamly.Data.Array.Storable.Foreign` module provides immutable arrays. Arrays are the

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

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

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

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

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

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

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

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

374storage.

376`Streamly.Data.Array.Storable.Foreign` uses pinned memory outside GC and therefore avoid any

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

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

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

381## Folds

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

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

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

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

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

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

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

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

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

393powerful ways of composing and applying them.

395## Unfolds

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

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

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

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

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

402stream merging combinators.

404## File IO

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

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

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

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

410to keep them short and elegant. Source file

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

412in the examples directory includes these examples.

414``` haskell

415module Main where

417import qualified Streamly.Prelude as S

418import qualified Streamly.Data.Fold as FL

419import qualified Streamly.Data.Array.Storable.Foreign as A

420import qualified Streamly.FileSystem.Handle as FH

421import qualified System.IO as FH

423import Data.Char (ord)

424import System.Environment (getArgs)

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

427withArg f = do

428 (name : _) <- getArgs

429 src <- openFile name ReadMode

430 f src

432withArg2 f = do

433 (sname : dname : _) <- getArgs

434 src <- openFile sname ReadMode

435 dst <- openFile dname WriteMode

436 f src dst

437```

439### cat

441``` haskell

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

443main = withArg cat

444```

446### cp

448``` haskell

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

450main = withArg2 cp

451```

453### wc -l

455``` haskell

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

457main = withArg wcl >>= print

458```

460### Average Line Length

462``` haskell

463avgll =

464 S.fold avg

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

466 . S.unfold FH.read

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

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

471main = withArg avgll >>= print

472```

474### Line Length Histogram

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

477`Streamly.Internal.Data.Fold`

479``` haskell

480llhisto =

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

482 . S.map bucket

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

484 . S.unfold FH.read

486 where

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

489main = withArg llhisto >>= print

490```

492## Socket IO

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

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

496and client programs using streamly. See

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

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

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

500in the examples directory.

502## Exceptions

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

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

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

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

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

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

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

512streams other concurrent streams are promptly terminated.

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

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

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

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

519## Reactive Programming (FRP)

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

522of integrating concurrency and streaming. See

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

524for a console based FRP game example and

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

526for an SDL based animation example.

528## Conclusion

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

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

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

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

534consumed concurrently – providing a generalized high level programming

535framework unifying streaming and concurrency. Controlled concurrency allows

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

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

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

539programs.

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

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

543key points in favor of streamly are:

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

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

547 and ease of use as a design goal.

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

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

550 same way as non-concurrent ones.

551 * _Generality_: Unifies functionality provided by several disparate packages

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

553 programming) in a concise API.

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

555 fusion optimizations for best possible performance. Serial peformance is

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

557 in some cases. Concurrent performance is unbeatable. See

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

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

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

562streaming libraries like

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

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

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

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

567In addition to providing streaming functionality, streamly subsumes

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

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

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

571the concurrency side, it subsumes the functionality of the

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

573higher level concurrent composition. Because it supports

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

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

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

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

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

581## Support

583Please feel free to ask questions on the

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

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

586enhancements to the library please contact

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

589## Credits

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

592significant way:

594 * Roman Leshchinskiy ([vector](http://hackage.haskell.org/package/vector))

595 * Gabriel Gonzalez ([foldl](https://hackage.haskell.org/package/foldl))

596 * Alberto G. Corona ([transient](https://hackage.haskell.org/package/transient))

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

600## Contributing

602The code is available under BSD-3 license

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

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

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

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