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

git.iko.soy/streamly

on fc45e9b help sign in

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

521 lines 19 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.

47* An `Unfold IO a b` is a representation of a function that converts a seed

48 value of type `a` into a stream of values of type `b` in the IO Monad.

50## The Examples

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

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

54snippets are omitted from the latter ones.

56### Modular Word Counting

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

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

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

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

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

64program.

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

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

70```haskell ghci

71import Data.Function ((&))

73import qualified Streamly.Data.Fold as Fold

74import qualified Streamly.Data.Stream as Stream

75import qualified Streamly.FileSystem.File as File

77wcb :: String -> IO Int

78wcb file =

79 File.read file -- Stream IO Word8

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

81```

83### Count Lines (wc -l)

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

87```haskell ghci

88import Data.Word (Word8)

89import Streamly.Data.Fold (Fold)

91-- ASCII character 10 is a newline.

92countl :: Int -> Word8 -> Int

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

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

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

97nlines = Fold.foldl' countl 0

99wcl :: String -> IO Int

100wcl file =

101 File.read file -- Stream IO Word8

102 & Stream.fold nlines -- IO Int

103```

105### Count Words (wc -w)

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

108in a stream:

110```haskell ghci

111import Data.Char (chr, isSpace)

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

114countw (n, wasSpace) ch =

115 if isSpace $ chr $ fromIntegral ch

116 then (n, True)

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

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

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

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

123wcw :: String -> IO Int

124wcw file =

125 File.read file -- Stream IO Word8

126 & Stream.fold nwords -- IO Int

127```

129### Counting Bytes, Words and Lines Together

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

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

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

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

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

136function (`(,,)`).

138```haskell ghci

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

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

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

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

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

146wc file =

147 File.read file -- Stream IO Word8

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

149```

151This example demonstrates the excellent modularity offered by

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

154### The Performance of Word Counting

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

157and the other using C.

159The performance of the [Streamly word counting

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

162```

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

16411242220 97050938 574714449 gutenberg-500MB.txt

166real 0m2.033s

167user 0m1.821s

168sys 0m0.209s

169```

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

173```

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

17511242220 97050938 574714449 gutenberg-500MB.txt

177real 0m2.113s

178user 0m1.928s

179sys 0m0.185s

180```

182### Concurrent Word Counting

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

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

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

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

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

190arrays for our input data.

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

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

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

197```haskell ghci

198-- Counts lines words chars lastCharWasSpace

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

201{-# INLINE count #-}

202count :: Counts -> Char -> Counts

203count (Counts l w c wasSpace) ch =

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

205 (w1, wasSpace1) =

206 if isSpace ch

207 then (w, True)

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

209 in Counts l1 w1 (c + 1) wasSpace1

210```

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

214```haskell ghci

215import Streamly.Data.Array (Array)

217import qualified Streamly.Data.Array as Array

218import qualified Streamly.Unicode.Stream as Unicode

220countArray :: Array Word8 -> IO Counts

221countArray arr =

222 Stream.unfold Array.reader arr -- Stream IO Word8

223 & Unicode.decodeLatin1 -- Stream IO Char

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

225```

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

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

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

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

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

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

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

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

237```haskell ghci

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

239partialCounts arr = do

240 let r = Array.getIndex 0 arr

241 case r of

242 Just x -> do

243 counts <- countArray arr

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

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

246```

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

250```haskell ghci

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

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

253 let wcount =

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

255 then w1 + w2 - 1

256 else w1 + w2

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

258```

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

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

262from each chunk.

264```haskell ghci

265{-# LANGUAGE FlexibleContexts #-}

267import GHC.Conc (numCapabilities)

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

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

271wc file = do

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

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

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

276 where

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

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

280```

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

284A benchmark with 2 CPUs:

286```

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

28811242220 97050938 574714449 gutenberg-500MB.txt

290real 0m1.443s

291user 0m2.095s

292sys 0m0.202s

293```

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

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

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

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

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

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

301binary.

303Streamly provides concurrency facilities similar

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

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

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

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

309### A Concurrent Network Server

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

312dictionary lookup server which can serve word meanings to multiple

313clients concurrently.

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

316example.

318```haskell ghci

319import Control.Concurrent (threadDelay)

320import Control.Exception (finally)

321import Network.Socket (Socket, close)

323import qualified Streamly.Data.Parser as Parser

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

325import qualified Streamly.Network.Socket as Socket

326import qualified Streamly.Unicode.Stream as Unicode

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

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

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

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

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

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

335-- connection is closed.

336lookupWords :: Socket -> IO ()

337lookupWords sk =

338 Stream.unfold Socket.reader sk -- Stream IO Word8

339 & Unicode.decodeLatin1 -- Stream IO Char

340 & Stream.parseMany word -- Stream IO String

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

342 & fmap show -- Stream IO String

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

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

345 & Stream.fold (Socket.writeChunks sk)

347 where

349 word = Parser.wordBy isSpace Fold.toList

350 cfg = Stream.ordered True

352serve :: Socket -> IO ()

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

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

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

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

358main :: IO ()

359main =

360 Stream.unfold TCP.acceptorOnPort 8091 -- Stream IO Socket

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

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

363```

365### Merging Incoming Streams

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

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

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

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

371streams concurrently.

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

374including the imports that we have omitted below.

376```haskell ghci

377{-# LANGUAGE FlexibleContexts #-}

379import Streamly.Data.Stream (Stream)

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

382import qualified Streamly.Data.Unfold as Unfold

383import qualified Streamly.Network.Socket as Socket

384import qualified Streamly.FileSystem.Handle as Handle

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

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

388-- buffering for safety.

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

390readLines sk =

391 Stream.unfold Socket.reader sk -- Stream IO Word8

392 & Unicode.decodeLatin1 -- Stream IO Char

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

395 where

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

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

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

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

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

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

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

406server :: Handle -> IO ()

407server file =

408 Stream.unfold TCP.acceptorOnPort 8090 -- Stream IO Socket

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

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

411 & Unicode.encodeLatin1 -- Stream IO Word8

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

414main :: IO ()

415main = withFile "output.txt" AppendMode server

416```

418### Listing Directories Recursively/Concurrently

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

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

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

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

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

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

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

430```haskell ghci

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

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

434main :: IO ()

435main = do

436 hSetBuffering stdout LineBuffering

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

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

439 ls = Stream.parConcatIterate id f start

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

441```

443### Rate Limiting

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

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

448```haskell ghci

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

451main :: IO ()

452main =

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

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

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

456```

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

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

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

462throughputs as high as millions of yields per second.

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

465documentation][Streamly].

467### Reactive Programming

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

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

471module for time-specific and sampling combinators.

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

474reactive programming using [Streamly][].

476### More Examples

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

479Examples][streamly-examples] web page.

481<!--

482### Further Reading

484* [Streaming Benchmarks][streaming-benchmarks]

485* [Concurrency Benchmarks][concurrency-benchmarks]

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

487* [Other Guides](/)

488* [Streamly Homepage][Streamly]

489-->

491<!--

492Link References.

493-->

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

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

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

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

500<!--

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

502links (for online docs) before we release.

503-->

505

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

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

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

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

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

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

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

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

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

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

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

518

519[LICENSE]: /LICENSE

520[CONTRIBUTING.md]: /CONTRIBUTING.md

521[docs]: docs/