README.md · streamly · git.iko.soy

git.iko.soy/streamly

on e08ff86 help sign in

streamly/README.md

468 lines 18 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[![Build Status](https://travis-ci.org/composewell/streamly.svg?branch=master)](https://travis-ci.org/composewell/streamly)

6[![Windows Build status](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)

11## Streaming Concurrently

13Haskell lists express pure computations using composable stream operations like

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

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

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

17with concurrent execution semantics without introducing any additional APIs.

19Streamly expresses concurrency using standard, well known abstractions.

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

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

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

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

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

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

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

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

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

29and declarative monadic concurrency.

31## Where to use streamly?

33Streamly is a general purpose programming framwework. It can be used equally

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

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

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

37Streamly generalizes lists to monadic streams, and the `IO` monad to

38non-deterministic and concurrent stream composition. The `IO` monad is a

39special case of streamly; if we use single element streams the behavior of

40streamly becomes identical to the IO monad. The IO monad code can be replaced

41with streamly by just prefixing the IO actions with `liftIO`, without any other

42changes, and without any loss of performance. Pure lists too are a special

43case of streamly; if we use `Identity` as the underlying monad, streamly

44streams turn into pure lists. Non-concurrent programs are just a special case

45of concurrent ones, simply adding a combinator turns a non-concurrent program

46into a concurrent one.

48In other words, streamly combines the functionality of lists and IO, with

49builtin concurrency. If you want to write a program that involves IO,

50concurrent or not, then you can just use streamly as the base monad, in fact,

51you could even use streamly for pure computations, as streamly performs at par

52with pure lists or `vector`.

54## Why data flow programming?

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

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

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

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

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

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

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

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

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

66## Show me an example

68Here is an IO monad code to list a directory recursively:

70```haskell

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

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

74listDirRecursive = getCurrentDir >>= readdir

75 where

76 readdir dir = do

77 (dirs, files) <- listDir dir

78 liftIO $ mapM_ putStrLn

79 $ map show dirs ++ map show files

80 foldMap readdir dirs

81```

83This is your usual IO monad code, with no streamly specific code whatsoever.

84This is how you can run this:

86``` haskell

87main :: IO ()

88main = listDirRecursive

89```

91And, this is how you can run exactly the same code using streamly with

92lookahead style concurrency, the only difference is that this time multiple

93directories are read concurrently:

95``` haskell

96import Streamly (runStream, aheadly)

98main :: IO ()

99main = runStream $ aheadly $ listDirRecursive

100```

102Isn't that magical? What's going on here? Streamly does not introduce any new

103abstractions, it just uses standard abstractions like `Semigroup` or

104`Monoid` to combine monadic streams concurrently, the way lists combine a

105sequence of pure values non-concurrently. The `foldMap` in the code

106above turns into a concurrent monoidal composition of a stream of `readdir`

107computations.

109## How does it perform?

111Providing monadic streaming and high level declarative concurrency does not

112mean that `streamly` compromises with performance in any way. The

113non-concurrent performance of `streamly` competes with lists and the `vector`

114library. The concurrent performance is as good as it gets, see [concurrency

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

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

118The following chart shows a summary of the cost of key streaming operations

119processing a million elements. The timings for `streamly` and `vector` are in

120the 600-700 microseconds range and therefore can barely be seen in the graph.

121For more details, see [streaming

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

124![Streaming Operations at a Glance](charts-0/KeyOperations-time.svg)

126## Streaming Pipelines

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

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

130number more than 9 is entered.

132``` haskell

133import Streamly

134import qualified Streamly.Prelude as S

135import Data.Function ((&))

137main = runStream $

138 S.repeatM getLine

139 & fmap read

140 & S.filter even

141 & S.takeWhile (<= 9)

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

143 & S.mapM print

144```

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

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

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

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

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

151application (`&`) operator is enough. Combinators are provided in

152`Streamly.Prelude` to transform or fold streams.

154## Concurrent Stream Generation

156Monadic construction and generation functions e.g. `consM`, `unfoldrM`,

157`replicateM`, `repeatM`, `iterateM` and `fromFoldableM` etc. work concurrently

158when used with appropriate stream type combinator (e.g. `asyncly`, `aheadly` or

159`parallely`).

161The following code finishes in 3 seconds (6 seconds when serial):

163``` haskell

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

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

166[3,2,1]

168> S.toList $ parallely $ p 3 |: p 2 |: p 1 |: S.nil

169[1,2,3]

170```

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

174``` haskell

175runStream $ asyncly $ S.replicateM 10 $ p 10

176```

178## Concurrent Streaming Pipelines

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

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

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

183application.

185``` haskell

186main = runStream $

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

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

189```

191## Mapping Concurrently

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

195``` haskell

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

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

198```

200## Serial and Concurrent Merging

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

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

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

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

207``` haskell

208import Streamly

209import qualified Streamly.Prelude as S

210import Control.Concurrent (threadDelay)

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

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

214```

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

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

218function in the examples to demonstrate the concurrency aspects:

220``` haskell

221import Streamly

222import qualified Streamly.Prelude as S

223import Control.Concurrent

225delay n = S.yieldM $ do

226 threadDelay (n * 1000000)

227 tid <- myThreadId

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

229```

230### Serial

232``` haskell

233main = runStream $ delay 3 <> delay 2 <> delay 1

234```

235```

236ThreadId 36: Delay 3

237ThreadId 36: Delay 2

238ThreadId 36: Delay 1

239```

241### Parallel

243``` haskell

244main = runStream . parallely $ delay 3 <> delay 2 <> delay 1

245```

246```

247ThreadId 42: Delay 1

248ThreadId 41: Delay 2

249ThreadId 40: Delay 3

250```

252## Nested Loops (aka List Transformer)

254The monad instance composes like a list monad.

256``` haskell

257import Streamly

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 = runStream 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, lookahead style concurrency i.e. each iteration in

277the loop can run run concurrently by but the results are presented in the same

278order as serial execution:

280``` haskell

281main = runStream $ aheadly $ loops

282```

284To run it with depth first concurrency yielding results asynchronously in the

285same order as they become available (deep async composition):

287``` haskell

288main = runStream $ asyncly $ loops

289```

291To run it with breadth first concurrency and yeilding results asynchronously

292(wide async composition):

294``` haskell

295main = runStream $ wAsyncly $ loops

296```

298The above streams provide lazy/demand-driven concurrency which is automatically

299scaled as per demand and is controlled/bounded so that it can be used on

300infinite streams. The following combinator provides strict, unbounded

301concurrency irrespective of demand:

303``` haskell

304main = runStream $ parallely $ loops

305```

307To run it serially but interleaving the outer and inner loop iterations

308(breadth first serial):

310``` haskell

311main = runStream $ wSerially $ loops

312```

314## Magical Concurrency

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

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

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

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

321``` haskell

322import Streamly

323import qualified Streamly.Prelude as S

325main = do

326 s <- S.sum $ asyncly $ do

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

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

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

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

331 return $ sqrt (x2 + y2)

332 print s

333```

335The concurrency facilities provided by streamly can be compared with

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

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

338expression.

340## Rate Limiting

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

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

345``` haskell

346import Streamly

347import Streamly.Prelude as S

349main = runStream $ asyncly $ avgRate 1 $ S.repeatM $ putStrLn "hello"

350```

352For some practical uses of rate control, see

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

354and

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

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

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

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

360documentation.

362## Exceptions

364From a library user point of view, there is nothing much to learn or talk about

365exceptions. Synchronous exceptions work just the way they are supposed to work

366in any standard non-concurrent code. When concurrent streams are combined

367together, exceptions from the constituent streams are propagated to the

368consumer stream. When an exception occurs in any of the constituent streams

369other concurrent streams are promptly terminated. Exceptions can be thrown

370using the `MonadThrow` instance.

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

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

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

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

378## Reactive Programming (FRP)

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

381of integrating concurrency and streaming. See

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

383for a console based FRP game example and

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

385for an SDL based animation example.

387## Conclusion

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

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

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

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

393consumed concurrently – providing a generalized high level programming

394framework unifying streaming and concurrency. Controlled concurrency allows

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

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

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

398programs.

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

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

402key points in favor of streamly are:

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

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

406 and ease of use as a design goal.

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

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

409 same way as non-concurrent ones.

410 * _Generality_: Unifies functionality provided by several disparate packages

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

412 programming) in a concise API.

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

414 fusion optimizations for best possible performance. Serial peformance is

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

416 in some cases. Concurrent performance is unbeatable. See

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

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

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

421streaming libraries like

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

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

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

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

426In addition to providing streaming functionality, streamly subsumes

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

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

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

430the concurrency side, it subsumes the functionality of the

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

432higher level concurrent composition. Because it supports

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

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

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

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

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

440## Further Reading

442For more information, see:

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

445 * [Reference documentation](https://hackage.haskell.org/package/streamly)

446 * [Examples](https://github.com/composewell/streamly/tree/master/examples)

447 * [Guides](https://github.com/composewell/streamly/blob/master/docs)

448 * [Streaming benchmarks](https://github.com/composewell/streaming-benchmarks)

449 * [Concurrency benchmarks](https://github.com/composewell/concurrency-benchmarks)

451## Support

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

454enhancements to the library please contact

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

457## Contributing

459The code is available under BSD-3 license

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

461[gitter chat](https://gitter.im/composewell/streamly) channel for discussions.

462You can find some of the

463[todo items on the github wiki](https://github.com/composewell/streamly/wiki/Things-To-Do).

464Please ask on the gitter channel or [contact the maintainer directly](mailto:harendra.kumar@gmail.com)

465for more details on each item. All contributions are welcome!

467This library was originally inspired by the `transient` package authored by

468Alberto G. Corona.