docs/User/Tutorials/quick-overview.md · streamly · git.iko.soy

git.iko.soy/streamly

on 8927d11 help sign in

streamly/docs/User/Tutorials/quick-overview.md

517 lines 18 kB history raw

1<!--

2(c) 2017, Composewell Technologies and Contributors

3SPDX-License-Identifer: BSD-3-Clause

4-->

6# [Streamly][] Quick Tutorial

7## About This Document

9This guide introduces programming with [Streamly][] using a few practical

10examples:

12* We will start with a simple program that [counts the number of words

13 in a text](#modular-word-counting). We will then transform this program

14 into a [concurrent](#concurrent-word-counting) program that can efficiently

15 use multiprocessing hardware.

16* Next, we will create a [concurrent network

17 server](#a-concurrent-network-server). We then show

18 how to write a network server that [merges multiple

19 streams](#merging-incoming-streams) concurrently.

20* Our third example shows how to list a directory tree concurrently,

21 by reading [multiple directories in

22 parallel](#listing-directories-recursivelyconcurrently).

23* Finally, we will look at how to [rate limit](#rate-limiting) stream

24 processing.

26It [concludes](#further-reading) with suggestions for further reading.

28## Getting Started

30### Installing Streamly

32If you wish to follow along and run examples in this guide, please see

33the [Before You Begin](/docs/User/Tutorials/before-you-begin.md) guide

34for instructions on how to use the `streamly` package interactively or

35in a project.

37### An overview of the types used in these examples

39As an expository device, we have indicated the types at the intermediate

40stages of stream computations as comments in the examples below.

41The meaning of these types are:

43* A `Stream IO a` is a representation of a sequence of values of type

44 `a` in the IO Monad.

45* A `Fold IO a b` is a representation of a function that converts a stream of

46 type `a` to a final accumulator of type `b` in the IO Monad.

48## The Examples

50The code snippets below should work in GHCi if all of those are typed

51in sequence. For brevity, imports that are already used in earlier

52snippets are omitted from the latter ones.

54### Modular Word Counting

56A `Fold` in Streamly is a composable stream consumer. For our first

57example, we will use `Fold`s to count the number of bytes, words and lines

58present in a file. We will then compose individual `Fold`s together to

59count words, bytes and lines at the same time.

61Please see the file [WordCountModular.hs][] for the complete example

62program.

64#### Count Bytes (wc -c)

66We start with a code fragment that counts the number of bytes in a file:

68```haskell ghci

69import Data.Function ((&))

71import qualified Streamly.Data.Fold as Fold

72import qualified Streamly.Data.Stream as Stream

73import qualified Streamly.FileSystem.File as File

75wcb :: String -> IO Int

76wcb file =

77 File.read file -- Stream IO Word8

78 & Stream.fold Fold.length -- IO Int

79```

81### Count Lines (wc -l)

83The next code fragment shows how to count the number of lines in a file:

85```haskell ghci

86import Data.Word (Word8)

87import Streamly.Data.Fold (Fold)

89-- ASCII character 10 is a newline.

90countl :: Int -> Word8 -> Int

91countl n ch = if ch == 10 then n + 1 else n

93-- The `nlines` fold accepts a stream of `Word8` and returns a line count (`Int`).

94nlines :: Monad m => Fold m Word8 Int

95nlines = Fold.foldl' countl 0

97wcl :: String -> IO Int

98wcl file =

99 File.read file -- Stream IO Word8

100 & Stream.fold nlines -- IO Int

101```

103### Count Words (wc -w)

105Our final code fragment counts the number of whitespace-separated words

106in a stream:

108```haskell ghci

109import Data.Char (chr, isSpace)

111countw :: (Int, Bool) -> Word8 -> (Int, Bool)

112countw (n, wasSpace) ch =

113 if isSpace $ chr $ fromIntegral ch

114 then (n, True)

115 else (if wasSpace then n + 1 else n, False)

117-- The `nwords` fold accepts a stream of `Word8` and returns a word count (`Int`).

118nwords :: Monad m => Fold m Word8 Int

119nwords = fst <$> Fold.foldl' countw (0, True)

121wcw :: String -> IO Int

122wcw file =

123 File.read file -- Stream IO Word8

124 & Stream.fold nwords -- IO Int

125```

127### Counting Bytes, Words and Lines Together

129By using the `Tee` combinator we can compose the three folds that count

130bytes, lines and words individually into a single fold that counts all

131three at once. The applicative instance of `Tee` distributes its input

132to all the supplied folds (`Fold.length`, `nlines`, and `nwords`) and

133then combines the outputs from the folds using the supplied combiner

134function (`(,,)`).

136```haskell ghci

137import Streamly.Data.Fold (Tee(..))

139-- The fold accepts a stream of `Word8` and returns the three counts.

140countAll :: Fold IO Word8 (Int, Int, Int)

141countAll = unTee $ (,,) <$> Tee Fold.length <*> Tee nlines <*> Tee nwords

143wc :: String -> IO (Int, Int, Int)

144wc file =

145 File.read file -- Stream IO Word8

146 & Stream.fold countAll -- IO (Int, Int, Int)

147```

149This example demonstrates the excellent modularity offered by

150[Streamly][]'s simple and concise API.

152### The Performance of Word Counting

154We compare two equivalent implementations: one using [Streamly][],

155and the other using C.

157The performance of the [Streamly word counting

158implementation][WordCount.hs] (using ghc-9.4.4 and fusion-plugin) is:

160```

161$ time WordCount-hs gutenberg-500MB.txt

16211242220 97050938 574714449 gutenberg-500MB.txt

164real 0m2.033s

165user 0m1.821s

166sys 0m0.209s

167```

169The performance of an equivalent [wc implementation in C][WordCount.c] is:

171```

172$ time WordCount-c gutenberg-500MB.txt

17311242220 97050938 574714449 gutenberg-500MB.txt

175real 0m2.113s

176user 0m1.928s

177sys 0m0.185s

178```

180### Concurrent Word Counting

182In our next example we show how the task of counting words, lines,

183and bytes could be done in parallel on multiprocessor hardware.

185To count words in parallel we first divide the stream into chunks

186(arrays), do the counting within each chunk, and then add all the

187counts across chunks. We use the same code as above except that we use

188arrays for our input data.

190Please see the file [WordCountParallel.hs][] for the complete working

191code for this example, including the imports that we have omitted below.

193First we create a new data type `Counts` that holds all the context.

195```haskell ghci

196-- Counts lines words chars lastCharWasSpace

197data Counts = Counts !Int !Int !Int !Bool deriving Show

199{-# INLINE count #-}

200count :: Counts -> Char -> Counts

201count (Counts l w c wasSpace) ch =

202 let l1 = if ch == '\n' then l + 1 else l

203 (w1, wasSpace1) =

204 if isSpace ch

205 then (w, True)

206 else (if wasSpace then w + 1 else w, False)

207 in Counts l1 w1 (c + 1) wasSpace1

208```

210The `countArray` function counts the line, word, char counts in one chunk:

212```haskell ghci

213import Streamly.Data.Array (Array)

215import qualified Streamly.Data.Array as Array

216import qualified Streamly.Unicode.Stream as Unicode

218countArray :: Array Word8 -> IO Counts

219countArray arr =

220 Array.read arr -- Stream IO Word8

221 & Unicode.decodeLatin1 -- Stream IO Char

222 & Stream.fold (Fold.foldl' count (Counts 0 0 0 True)) -- IO Counts

223```

225Here the function `count` and the `Counts` data type are defined in the

226`WordCount` helper module defined in [WordCount.hs][].

228When combining the counts in two contiguous chunks, we need to check

229whether the first element of the next chunk is a whitespace character

230in order to determine if the same word continues in the next chunk or

231whether the chunk starts with a new word. The `partialCounts` function

232adds a `Bool` flag to `Counts` returned by `countArray` to indicate

233whether the first character in the chunk is a space.

235```haskell ghci

236partialCounts :: Array Word8 -> IO (Bool, Counts)

237partialCounts arr = do

238 let r = Array.getIndex 0 arr

239 case r of

240 Just x -> do

241 counts <- countArray arr

242 return (isSpace (chr (fromIntegral x)), counts)

243 Nothing -> return (False, Counts 0 0 0 True)

244```

246`addCounts` then adds the counts from two consecutive chunks:

248```haskell ghci

249addCounts :: (Bool, Counts) -> (Bool, Counts) -> (Bool, Counts)

250addCounts (sp1, Counts l1 w1 c1 ws1) (sp2, Counts l2 w2 c2 ws2) =

251 let wcount =

252 if not ws1 && not sp2 -- No space between two chunks.

253 then w1 + w2 - 1

254 else w1 + w2

255 in (sp1, Counts (l1 + l2) wcount (c1 + c2) ws2)

256```

258To count in parallel we now only need to divide the stream into arrays,

259apply our counting function to each array, and then combine the counts

260from each chunk.

262```haskell ghci

263{-# LANGUAGE FlexibleContexts #-}

265import GHC.Conc (numCapabilities)

266import qualified Streamly.Data.Stream.Prelude as Stream

268wc :: String -> IO (Bool, Counts)

269wc file = do

270 File.readChunks file -- Stream IO (Array Word8)

271 & Stream.parMapM cfg partialCounts -- Stream IO (Bool, Counts)

272 & Stream.fold add -- IO (Bool, Counts)

274 where

276 cfg = Stream.maxThreads numCapabilities . Stream.ordered True

277 add = Fold.foldl' addCounts (False, Counts 0 0 0 True)

278```

280We can replace `parMapM` with `mapM` to get a serial version of the program.

282A benchmark with 2 CPUs:

284```

285$ time WordCount-hs-parallel gutenberg-500MB.txt

28611242220 97050938 574714449 gutenberg-500MB.txt

288real 0m1.443s

289user 0m2.095s

290sys 0m0.202s

291```

293These example programs have assumed ASCII encoded input data. For UTF-8

294streams, we have a [concurrent wc implementation][WordCountParallelUTF8.hs]

295with UTF-8 decoding. This concurrent implementation performs as well

296as the standard `wc` program in serial benchmarks. In concurrent mode

297[Streamly][]'s implementation can utilise multiple processing cores if

298these are present, and can thereby run much faster than the standard

299binary.

301Streamly provides concurrency facilities similar

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

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

304style of expression. With Streamly you can write concurrent programs

305with ease, with support for different types of concurrent scheduling.

307### A Concurrent Network Server

309We now move to a slightly more complicated example: we simulate a

310dictionary lookup server which can serve word meanings to multiple

311clients concurrently.

313Please see the file [WordServer.hs][] for the complete code for this

314example.

316```haskell ghci

317import Control.Concurrent (threadDelay)

318import Control.Exception (finally)

319import Network.Socket (Socket, close)

321import qualified Streamly.Data.Parser as Parser

322import qualified Streamly.Network.Inet.TCP as TCP

323import qualified Streamly.Network.Socket as Socket

324import qualified Streamly.Unicode.Stream as Unicode

326-- Simulate network/db query by adding a delay.

327fetch :: String -> IO (String, String)

328fetch w = threadDelay 1000000 >> return (w,w)

330-- Read lines of whitespace separated list of words from a socket, fetch the

331-- meanings of each word concurrently and return the meanings separated by

332-- newlines, in same order as the words were received. Repeat until the

333-- connection is closed.

334lookupWords :: Socket -> IO ()

335lookupWords sk =

336 Socket.read sk -- Stream IO Word8

337 & Unicode.decodeLatin1 -- Stream IO Char

338 & Stream.wordsBy isSpace Fold.toList -- Stream IO String

339 & Stream.parMapM cfg fetch -- Stream IO (String, String)

340 & fmap show -- Stream IO String

341 & Stream.intersperse "\n" -- Stream IO String

342 & Unicode.encodeStrings Unicode.encodeLatin1 -- Stream IO (Array Word8)

343 & Stream.fold (Socket.writeChunks sk)

345 where

347 cfg = Stream.ordered True

349serve :: Socket -> IO ()

350serve sk = finally (lookupWords sk) (close sk)

352-- | Run a server on port 8091. Accept and handle connections concurrently. The

353-- connection handler is "serve" (i.e. lookupWords). You can use "telnet" or

354-- "nc" as a client to try it out.

355main :: IO ()

356main =

357 TCP.accept 8091 -- Stream IO Socket

358 & Stream.parMapM id serve -- Stream IO ()

359 & Stream.fold Fold.drain -- IO ()

360```

362### Merging Incoming Streams

364In the next example, we show how to merge logs coming from multiple

365nodes in your network. These logs are merged at line boundaries and

366the merged logs are written to a file or to a network destination.

367This example uses the `concatMapWith` combinator to merge multiple

368streams concurrently.

370Please see the file [MergeServer.hs][] for the complete working code,

371including the imports that we have omitted below.

373```haskell ghci

374{-# LANGUAGE FlexibleContexts #-}

376import Streamly.Data.Stream (Stream)

377import System.IO (IOMode(AppendMode), Handle, withFile)

379import qualified Streamly.Network.Socket as Socket

380import qualified Streamly.FileSystem.Handle as Handle

382-- | Read a line stream from a socket.

383-- Note: lines are buffered, and we could add a limit to the

384-- buffering for safety.

385readLines :: Socket -> Stream IO (Array Char)

386readLines sk =

387 Socket.read sk -- Stream IO Word8

388 & Unicode.decodeLatin1 -- Stream IO Char

389 & Stream.foldMany line -- Stream IO (Array Char)

391 where

393 line = Fold.takeEndBy (== '\n') Array.write

395recv :: Socket -> Stream IO (Array Char)

396recv sk = Stream.finallyIO (close sk) (readLines sk)

398-- | Starts a server at port 8091 listening for lines with space separated

399-- words. Multiple clients can connect to the server and send streams of lines.

400-- The server handles all the connections concurrently, merges the incoming

401-- streams at line boundaries and writes the merged stream to a file.

402server :: Handle -> IO ()

403server file =

404 TCP.accept 8090 -- Stream IO Socket

405 & Stream.parConcatMap (Stream.eager True) recv -- Stream IO (Array Char)

406 & Stream.unfoldMany Array.reader -- Stream IO Char

407 & Unicode.encodeLatin1 -- Stream IO Word8

408 & Stream.fold (Handle.write file) -- IO ()

410main :: IO ()

411main = withFile "output.txt" AppendMode server

412```

414### Listing Directories Recursively/Concurrently

416Our next example lists a directory tree recursively, and concurrently.

418This example uses the tree traversing combinator `parConcatIterate`. This

419combinator maps a stream generator function on the input stream and then

420recursively on the generated stream as well and flattens the results. We map a

421directory to a stream generating its children and a file to a nil stream. This

422results in a concurrent recursive depth first traversal of the directory tree.

424Please see [ListDir.hs][] for the complete working code.

426```haskell ghci

427import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

428import qualified Streamly.Internal.FileSystem.Dir as Dir (readEitherPaths)

430main :: IO ()

431main = do

432 hSetBuffering stdout LineBuffering

433 let start = Stream.fromPure (Left ".")

434 f = either Dir.readEitherPaths (const Stream.nil)

435 ls = Stream.parConcatIterate id f start

436 in Stream.fold (Fold.drainMapM print) ls

437```

439### Rate Limiting

441For concurrent streams, a stream evaluation rate can be specified. For

442example, to print "tick" once every second you can simply write:

444```haskell ghci

445import qualified Streamly.Internal.Data.Stream as Stream (timestamped)

447main :: IO ()

448main =

449 Stream.parRepeatM (Stream.avgRate 1) (pure "tick") -- Stream IO String

450 & Stream.timestamped -- Stream IO (AbsTime, String)

451 & Stream.fold (Fold.drainMapM print) -- IO ()

452```

454Please see the file [Rate.hs][] for the complete working code.

456The concurrency of the stream is automatically controlled to match the

457specified rate. [Streamly][]'s rate control works precisely even at

458throughputs as high as millions of yields per second.

460For more sophisticated rate control needs please see the Streamly [reference

461documentation][Streamly].

463### Reactive Programming

465Streamly supports reactive (time domain) programming because of its support for

466declarative concurrency. Please see the `Streamly.Data.Stream.Prelude`

467module for time-specific and sampling combinators.

469The examples [AcidRain.hs][] and [CirclingSquare.hs][] demonstrate

470reactive programming using [Streamly][].

472### More Examples

474If you would like to view more examples, please visit the [Streamly

475Examples][streamly-examples] web page.

477<!--

478### Further Reading

480* [Streaming Benchmarks][streaming-benchmarks]

481* [Concurrency Benchmarks][concurrency-benchmarks]

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

483* [Other Guides](/)

484* [Streamly Homepage][Streamly]

485-->

487<!--

488Link References.

489-->

491[Streamly]: https://streamly.composewell.com/

492[streamly-examples]: https://github.com/composewell/streamly-examples

493[streaming-benchmarks]: https://github.com/composewell/streaming-benchmarks

494[concurrency-benchmarks]: https://github.com/composewell/concurrency-benchmarks

496<!--

497Keep all the unstable links here so that they can be updated to stable

498links (for online docs) before we release.

499-->

501

502[WordCountModular.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCountModular.hs

503[WordCount.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCount.hs

504[WordCount.c]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCount.c

505[WordCountParallel.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCountParallel.hs

506[WordCountParallelUTF8.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordCountParallelUTF8.hs

507[WordServer.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/WordServer.hs

508[MergeServer.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/MergeServer.hs

509[ListDir.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/ListDir.hs

510[Rate.hs]: https://github.com/composewell/streamly-examples/blob/master/examples/Rate.hs

511[AcidRain.hs]: https://github.com/composewell/streamly-examples/tree/master/examples/AcidRain.hs

512[CirclingSquare.hs]: https://github.com/composewell/streamly-examples/tree/master/examples/CirclingSquare.hs

514

515[LICENSE]: /LICENSE

516[CONTRIBUTING.md]: /CONTRIBUTING.md

517[docs]: docs/