docs/User/Tutorials/Introduction.md · streamly · git.iko.soy

git.iko.soy/streamly

on b4c007e help sign in

streamly/docs/User/Tutorials/Introduction.md

537 lines 19 kB history raw

1# [Streamly][] Quick Tutorial

2## About This Document

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

5examples:

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

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

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

10 use multiprocessing hardware.

11* Next, we will create a [concurrent network

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

13 how to write a network server that [merges multiple

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

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

16 by reading [multiple directories in

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

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

19 processing.

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

23## Getting Started

25### Installing Streamly

27If you wish to follow along with this guide, you will need to have

28[Streamly][] installed.

30Please see the [Getting Started With The Streamly Package](/docs/User/Tutorials/getting-started.md)

31guide for instructions on how to install [Streamly][].

33If you wish to run benchmarks, please be sure to build your

34application using the instructions in the [Build Guide](/docs/User/HowTo/Compiling.md).

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

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

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

40The meaning of these types are:

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

43 `a` in the IO Monad.

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

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

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

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

49### A Note on Module Naming

51Some of the examples below use modules from the `Internal` Streamly package

52hierarchy. These are not really internal to the library. We classify

53`Streamly` modules into two categories:

55* _Released modules and APIs_: These modules and APIs are

56 stable. Significant changes to these modules and APIs will cause

57 Streamly's version number to change according to the package versioning

58 policy.

59* _Pre-release modules and APIs_: These modules and APIs have not been

60 formally released yet. They may change in the near future, and such

61 changes will not necessarily be reflected in Streamly's package

62 version number. As yet unreleased modules and APIs reside in the

63 `Internal` namespace.

65Please use a minor release upper bound to adhere to the Haskell PVP when

66using the pre-release (internal) modules.

68## The Examples

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

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

72snippets are omitted from the latter ones.

74### Modular Word Counting

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

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

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

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

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

82program.

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

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

88```haskell ghci

89import Data.Function ((&))

91import qualified Streamly.Data.Fold as Fold

92import qualified Streamly.Data.Stream as Stream

93import qualified Streamly.FileSystem.File as File

95wcb :: String -> IO Int

96wcb file =

97 File.read file -- Stream IO Word8

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

99```

101### Count Lines (wc -l)

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

105```haskell ghci

106import Data.Word (Word8)

107import Streamly.Data.Fold (Fold)

109-- ASCII character 10 is a newline.

110countl :: Int -> Word8 -> Int

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

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

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

115nlines = Fold.foldl' countl 0

117wcl :: String -> IO Int

118wcl file =

119 File.read file -- Stream IO Word8

120 & Stream.fold nlines -- IO Int

121```

123### Count Words (wc -w)

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

126in a stream:

128```haskell ghci

129import Data.Char (chr, isSpace)

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

132countw (n, wasSpace) ch =

133 if isSpace $ chr $ fromIntegral ch

134 then (n, True)

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

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

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

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

141wcw :: String -> IO Int

142wcw file =

143 File.read file -- Stream IO Word8

144 & Stream.fold nwords -- IO Int

145```

147### Counting Bytes, Words and Lines Together

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

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

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

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

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

154function (`(,,)`).

156```haskell ghci

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

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

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

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

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

164wc file =

165 File.read file -- Stream IO Word8

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

167```

169This example demonstrates the excellent modularity offered by

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

172### The Performance of Word Counting

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

175and the other using C.

177The performance of the [Streamly word counting

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

180```

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

18211242220 97050938 574714449 gutenberg-500MB.txt

184real 0m2.033s

185user 0m1.821s

186sys 0m0.209s

187```

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

191```

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

19311242220 97050938 574714449 gutenberg-500MB.txt

195real 0m2.113s

196user 0m1.928s

197sys 0m0.185s

198```

200### Concurrent Word Counting

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

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

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

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

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

208arrays for our input data.

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

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

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

215```haskell ghci

216-- Counts lines words chars lastCharWasSpace

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

219{-# INLINE count #-}

220count :: Counts -> Char -> Counts

221count (Counts l w c wasSpace) ch =

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

223 (w1, wasSpace1) =

224 if isSpace ch

225 then (w, True)

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

227 in Counts l1 w1 (c + 1) wasSpace1

228```

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

232```haskell ghci

233import Streamly.Data.Array (Array)

235import qualified Streamly.Data.Array as Array

236import qualified Streamly.Unicode.Stream as Unicode

238countArray :: Array Word8 -> IO Counts

239countArray arr =

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

241 & Unicode.decodeLatin1 -- Stream IO Char

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

243```

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

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

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

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

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

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

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

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

255```haskell ghci

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

257partialCounts arr = do

258 let r = Array.getIndex 0 arr

259 case r of

260 Just x -> do

261 counts <- countArray arr

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

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

264```

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

268```haskell ghci

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

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

271 let wcount =

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

273 then w1 + w2 - 1

274 else w1 + w2

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

276```

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

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

280from each chunk.

282```haskell ghci

283{-# LANGUAGE FlexibleContexts #-}

285import GHC.Conc (numCapabilities)

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

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

289wc file = do

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

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

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

294 where

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

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

298```

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

302A benchmark with 2 CPUs:

304```

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

30611242220 97050938 574714449 gutenberg-500MB.txt

308real 0m1.443s

309user 0m2.095s

310sys 0m0.202s

311```

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

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

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

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

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

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

319binary.

321Streamly provides concurrency facilities similar

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

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

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

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

327### A Concurrent Network Server

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

330dictionary lookup server which can serve word meanings to multiple

331clients concurrently.

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

334example.

336```haskell ghci

337import Control.Concurrent (threadDelay)

338import Control.Exception (finally)

339import Network.Socket (Socket, close)

341import qualified Streamly.Data.Parser as Parser

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

343import qualified Streamly.Network.Socket as Socket

344import qualified Streamly.Unicode.Stream as Unicode

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

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

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

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

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

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

353-- connection is closed.

354lookupWords :: Socket -> IO ()

355lookupWords sk =

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

357 & Unicode.decodeLatin1 -- Stream IO Char

358 & Stream.parseMany word -- Stream IO String

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

360 & fmap show -- Stream IO String

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

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

363 & Stream.fold (Socket.writeChunks sk)

365 where

367 word = Parser.wordBy isSpace Fold.toList

368 cfg = Stream.ordered True

370serve :: Socket -> IO ()

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

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

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

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

376main :: IO ()

377main =

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

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

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

381```

383### Merging Incoming Streams

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

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

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

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

389streams concurrently.

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

392including the imports that we have omitted below.

394```haskell ghci

395{-# LANGUAGE FlexibleContexts #-}

397import Streamly.Data.Stream (Stream)

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

400import qualified Streamly.Data.Unfold as Unfold

401import qualified Streamly.Network.Socket as Socket

402import qualified Streamly.FileSystem.Handle as Handle

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

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

406-- buffering for safety.

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

408readLines sk =

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

410 & Unicode.decodeLatin1 -- Stream IO Char

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

413 where

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

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

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

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

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

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

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

424server :: Handle -> IO ()

425server file =

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

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

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

429 & Unicode.encodeLatin1 -- Stream IO Word8

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

432main :: IO ()

433main = withFile "output.txt" AppendMode server

434```

436### Listing Directories Recursively/Concurrently

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

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

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

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

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

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

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

448```haskell ghci

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

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

452main :: IO ()

453main = do

454 hSetBuffering stdout LineBuffering

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

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

457 ls = Stream.parConcatIterate id f start

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

459```

461### Rate Limiting

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

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

466```haskell ghci

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

469main :: IO ()

470main =

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

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

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

474```

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

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

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

480throughputs as high as millions of yields per second.

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

483documentation][Streamly].

485### Reactive Programming

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

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

489module for time-specific and sampling combinators.

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

492reactive programming using [Streamly][].

494### More Examples

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

497Examples][streamly-examples] web page.

499### Further Reading

501* [Streaming Benchmarks][streaming-benchmarks]

502* [Concurrency Benchmarks][concurrency-benchmarks]

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

504* [Other Guides](/)

505* [Streamly Homepage][Streamly]

507<!--

508Link References.

509-->

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

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

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

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

516<!--

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

518links (for online docs) before we release.

519-->

521

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

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

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

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

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

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

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

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

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

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

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

534

535[LICENSE]: /LICENSE

536[CONTRIBUTING.md]: /CONTRIBUTING.md

537[docs]: docs/