43afc26 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Update Introduction in user docs

43afc26Adithya Kumar4 years ago

docs/User/Tutorials/Introduction.md+138−92

compare with master

unifiedsplit

@@ -55,9 +55,7 @@ As an expository device, we have indicated the types at the intermediate

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

56The meaning of these types are:

58−* A `SerialT IO a` is a serial stream of values of type `a` in the IO Monad.

58+* A `Stream IO a` is a serial stream of values of type `a` in the IO Monad.

59−* An `AsyncT IO a` is a concurrent (asynchronous) stream of values of type

60− `a` in the IO Monad.

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

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

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

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

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

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

@@ -98,14 +96,16 @@ program, including the imports that we have omitted here.

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

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

100

101−```haskell

99+```haskell ghci

100+import Data.Function ((&))

101+

102import qualified Streamly.Data.Fold as Fold

103+import qualified Streamly.Data.Stream as Stream

103import qualified Streamly.Internal.FileSystem.File as File

104import qualified Streamly.Internal.FileSystem.File as File

104−import qualified Streamly.Prelude as Stream

105

106wcb :: String -> IO Int

107wcb file =

108− File.toBytes file -- SerialT IO Word8

108+ File.read file -- Stream IO Word8

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

110```

111

@@ -113,7 +113,10 @@ wcb file =

113

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

115

116−```haskell

116+```haskell ghci

117+import Data.Word (Word8)

118+import Streamly.Data.Fold (Fold)

119+

117-- ASCII character 10 is a newline.

120-- ASCII character 10 is a newline.

118countl :: Int -> Word8 -> Int

121countl :: Int -> Word8 -> Int

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

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

@@ -124,7 +127,7 @@ nlines = Fold.foldl' countl 0

124

127

125wcl :: String -> IO Int

128wcl :: String -> IO Int

126wcl file =

129wcl file =

127− File.toBytes file -- SerialT IO Word8

130+ File.read file -- Stream IO Word8

128 & Stream.fold nlines -- IO Int

131 & Stream.fold nlines -- IO Int

129```

132```

130

133

@@ -133,7 +136,9 @@ wcl file =

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

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

134in a stream:

137in a stream:

135

138

136−```haskell

139+```haskell ghci

140+import Data.Char (chr, isSpace)

141+

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

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

138countw (n, wasSpace) ch =

143countw (n, wasSpace) ch =

139 if isSpace $ chr $ fromIntegral ch

144 if isSpace $ chr $ fromIntegral ch

@@ -146,7 +151,7 @@ nwords = fst <$> Fold.foldl' countw (0, True)

146

151

147wcw :: String -> IO Int

152wcw :: String -> IO Int

148wcw file =

153wcw file =

149− File.toBytes file -- SerialT IO Word8

154+ File.read file -- Stream IO Word8

150 & Stream.fold nwords -- IO Int

155 & Stream.fold nwords -- IO Int

151```

156```

152

157

@@ -159,7 +164,8 @@ to all the supplied folds (`Fold.length`, `nlines`, and `nwords`) and

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

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

160function (`(,,)`).

165function (`(,,)`).

161

166

162−```haskell

167+```haskell ghci

168+import Streamly.Internal.Data.Fold.Tee (Tee(Tee))

163import qualified Streamly.Internal.Data.Fold.Tee as Tee

169import qualified Streamly.Internal.Data.Fold.Tee as Tee

164

170

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

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

@@ -168,7 +174,7 @@ countAll = Tee.toFold $ (,,) <$> Tee Fold.length <*> Tee nlines <*> Tee nwords

168

174

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

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

170wc file =

176wc file =

171− File.toBytes file -- SerialT IO Word8

177+ File.read file -- Stream IO Word8

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

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

173```

179```

174

180

@@ -218,16 +224,36 @@ arrays for our input data.

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

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

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

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

220

226

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

228+

229+```haskell ghci

230+-- Counts lines words chars lastCharWasSpace

231+data Counts = Counts !Int !Int !Int !Bool deriving Show

232+

233+{-# INLINE count #-}

234+count :: Counts -> Char -> Counts

235+count (Counts l w c wasSpace) ch =

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

237+ (w1, wasSpace1) =

238+ if isSpace ch

239+ then (w, True)

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

241+ in Counts l1 w1 (c + 1) wasSpace1

242+```

243+

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

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

222

245

223−```haskell

246+```haskell ghci

247+import Streamly.Data.Array (Array)

248+

224−import qualified Streamly.Data.Array.Foreign as Array

249+import qualified Streamly.Data.Array as Array

250+import qualified Streamly.Unicode.Stream as Unicode

225

251

226countArray :: Array Word8 -> IO Counts

252countArray :: Array Word8 -> IO Counts

227countArray arr =

253countArray arr =

228− Stream.unfold Array.read arr -- SerialT IO Word8

254+ Stream.unfold Array.reader arr -- Stream IO Word8

229− & Stream.decodeLatin1 -- SerialT IO Char

255+ & Unicode.decodeLatin1 -- Stream IO Char

230− & Stream.foldl' count (Counts 0 0 0 True) -- IO Counts

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

231```

257```

232

258

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

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

@@ -240,10 +266,10 @@ whether the chunk starts with a new word. The `partialCounts` function

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

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

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

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

242

268

243−```haskell

269+```haskell ghci

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

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

245partialCounts arr = do

271partialCounts arr = do

246− let r = Array.getIndex arr 0

272+ let r = Array.getIndex 0 arr

247 case r of

273 case r of

248 Just x -> do

274 Just x -> do

249 counts <- countArray arr

275 counts <- countArray arr

@@ -253,7 +279,7 @@ partialCounts arr = do

253

279

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

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

255

281

256−```haskell

282+```haskell ghci

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

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

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

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

259 let wcount =

285 let wcount =

@@ -267,22 +293,34 @@ To count in parallel we now only need to divide the stream into arrays,

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

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

268from each chunk.

294from each chunk.

269

295

270−```haskell

296+```haskell ghci

297+{-# LANGUAGE FlexibleContexts #-}

298+

299+import qualified Streamly.Data.Stream.Concurrent as Concur

300+

301+-- Number of threads we can afford

302+numCapabilities :: Int

303+numCapabilities = 10

304+

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

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

272wc file = do

306wc file = do

273− Stream.unfold File.readChunks file -- AheadT IO (Array Word8)

307+ Stream.unfold File.chunkReader file

274− & Stream.mapM partialCounts -- AheadT IO (Bool, Counts)

308+ & Concur.mapMWith

275− & Stream.maxThreads numCapabilities -- AheadT IO (Bool, Counts)

309+ ( Concur.maxThreads numCapabilities

276− & Stream.fromAhead -- SerialT IO (Bool, Counts)

310+ . Concur.ordered True )

311+ partialCounts

312+ & Stream.fold

277− & Stream.foldl' addCounts (False, Counts 0 0 0 True) -- IO (Bool, Counts)

313+ (Fold.foldl' addCounts (False, Counts 0 0 0 True))

278```

314```

279

315

280−Please note that the only difference between a concurrent and a

316+Please note that we use the mapping function from

281−non-concurrent program lies in the use of the `Stream.fromAhead`

317+`Streamly.Data.Stream.Concurrent` module. We explicitly specify what parts of

282−combinator. If we remove the call to `Stream.fromAhead`, we would

318+our stream are concurrent.

319+

320+If we replace `Concur.mapMWith (Concur.maxThreads numCapabilities

321+. Concur.ordered True)` with `Stream.mapM`, we would still have a perfectly

283−still have a perfectly valid and performant serial program. Notice

322+valid and performant serial program. Notice how succinctly and idiomatically we

284−how succinctly and idiomatically we have expressed the concurrent word

323+have expressed the concurrent word counting problem.

285−counting problem.

286

324

287A benchmark with 2 CPUs:

325A benchmark with 2 CPUs:

288

326

@@ -313,14 +351,17 @@ with ease, with support for different types of concurrent scheduling.

313

351

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

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

315dictionary lookup server which can serve word meanings to multiple

353dictionary lookup server which can serve word meanings to multiple

316−clients concurrently. This example demonstrates the use of the concurrent

354+clients concurrently.

317−`mapM` combinator.

318

355

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

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

320example, including the imports that we have omitted below.

357example, including the imports that we have omitted below.

321

358

322−```haskell

359+```haskell ghci

360+import Control.Concurrent (threadDelay)

361+import Control.Exception (finally)

362+import Network.Socket (Socket, close)

363+

323−import qualified Streamly.Data.Fold as Fold

364+import qualified Streamly.Data.Parser as Parser

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

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

325import qualified Streamly.Network.Socket as Socket

366import qualified Streamly.Network.Socket as Socket

326import qualified Streamly.Unicode.Stream as Unicode

367import qualified Streamly.Unicode.Stream as Unicode

@@ -335,16 +376,14 @@ fetch w = threadDelay 1000000 >> return (w,w)

335-- connection is closed.

376-- connection is closed.

336lookupWords :: Socket -> IO ()

377lookupWords :: Socket -> IO ()

337lookupWords sk =

378lookupWords sk =

338− Stream.unfold Socket.read sk -- SerialT IO Word8

379+ Stream.unfold Socket.reader sk

339− & Unicode.decodeLatin1 -- SerialT IO Char

380+ & Unicode.decodeLatin1

340− & Stream.wordsBy isSpace Fold.toList -- SerialT IO String

381+ & Stream.parseMany (Parser.wordBy isSpace Fold.toList)

341− & Stream.fromSerial -- AheadT IO String

382+ & Concur.mapMWith (Concur.ordered True) fetch

342− & Stream.mapM fetch -- AheadT IO (String, String)

383+ & fmap show

343− & Stream.fromAhead -- SerialT IO (String, String)

344− & Stream.map show -- SerialT IO String

345− & Stream.intersperse "\n" -- SerialT IO String

384+ & Stream.intersperse "\n"

346− & Unicode.encodeStrings Unicode.encodeLatin1 -- SerialT IO (Array Word8)

385+ & Unicode.encodeStrings Unicode.encodeLatin1

347− & Stream.fold (Socket.writeChunks sk) -- IO ()

386+ & Stream.fold (Socket.writeChunks sk)

348

387

349serve :: Socket -> IO ()

388serve :: Socket -> IO ()

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

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

@@ -354,11 +393,9 @@ serve sk = finally (lookupWords sk) (close sk)

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

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

355main :: IO ()

394main :: IO ()

356main =

395main =

357− Stream.unfold TCP.acceptOnPort 8091 -- SerialT IO Socket

396+ Stream.unfold TCP.acceptorOnPort 8091

358− & Stream.fromSerial -- AsyncT IO ()

397+ & Concur.mapMWith id serve

359− & Stream.mapM serve -- AsyncT IO ()

398+ & Stream.fold Fold.drain

360− & Stream.fromAsync -- SerialT IO ()

361− & Stream.drain -- IO ()

362```

399```

363

400

364### Merging Incoming Streams

401### Merging Incoming Streams

@@ -372,21 +409,27 @@ streams concurrently.

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

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

373including the imports that we have omitted below.

410including the imports that we have omitted below.

374

411

375−```haskell

412+```haskell ghci

413+{-# LANGUAGE FlexibleContexts #-}

414+

415+import Streamly.Data.Stream (Stream)

416+import System.IO (IOMode(AppendMode), Handle, withFile)

417+

376import qualified Streamly.Data.Unfold as Unfold

418import qualified Streamly.Data.Unfold as Unfold

377import qualified Streamly.Network.Socket as Socket

419import qualified Streamly.Network.Socket as Socket

420+import qualified Streamly.FileSystem.Handle as Handle

378

421

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

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

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

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

381-- buffering for safety.

424-- buffering for safety.

382−readLines :: Socket -> SerialT IO (Array Char)

425+readLines :: Socket -> Stream IO (Array Char)

383readLines sk =

426readLines sk =

384− Stream.unfold Socket.read sk -- SerialT IO Word8

427+ Stream.unfold Socket.reader sk

385− & Unicode.decodeLatin1 -- SerialT IO Char

428+ & Unicode.decodeLatin1

386− & Stream.splitWithSuffix (== '\n') Array.write -- SerialT IO String

429+ & Stream.foldMany (Fold.takeEndBy (== '\n') Array.write)

387

430

388−recv :: Socket -> SerialT IO (Array Char)

431+recv :: Socket -> Stream IO (Array Char)

389−recv sk = Stream.finally (liftIO $ close sk) (readLines sk)

432+recv sk = Stream.finallyIO (close sk) (readLines sk)

390

433

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

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

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

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

@@ -394,11 +437,11 @@ recv sk = Stream.finally (liftIO $ close sk) (readLines sk)

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

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

395server :: Handle -> IO ()

438server :: Handle -> IO ()

396server file =

439server file =

397− Stream.unfold TCP.acceptOnPort 8090 -- SerialT IO Socket

440+ Stream.unfold TCP.acceptorOnPort 8090

398− & Stream.concatMapWith Stream.parallel recv -- SerialT IO (Array Char)

441+ & Concur.concatMapWith (Concur.eager True) recv

399− & Stream.unfoldMany Array.read -- SerialT IO Char

442+ & Stream.unfoldMany Array.reader

400− & Unicode.encodeLatin1 -- SerialT IO Word8

443+ & Unicode.encodeLatin1

401− & Stream.fold (Handle.write file) -- IO ()

444+ & Stream.fold (Handle.write file)

402

445

403main :: IO ()

446main :: IO ()

404main = withFile "output.txt" AppendMode server

447main = withFile "output.txt" AppendMode server

@@ -409,37 +452,43 @@ main = withFile "output.txt" AppendMode server

409Our next example lists a directory tree recursively, reading

452Our next example lists a directory tree recursively, reading

410multiple directories concurrently.

453multiple directories concurrently.

411

454

412−This example uses the tree traversing combinator `iterateMapLeftsWith`.

455+This example uses the tree traversing combinator `iterateMapLeftsWith`. This

413−This combinator maps a stream generator on the `Left` values in its

456+combinator maps a stream generator on the `Left` values in its input stream

414−input stream (directory names in this case), feeding the resulting `Left`

457+(directory names in this case), feeding the resulting `Left` values back to the

415−values back to the input, while it lets the `Right` values (file names

458+input, while it lets the `Right` values (file names in this case) pass through

416−in this case) pass through to the output. The `Stream.ahead` stream

459+to the output. The joining combinator explicitly sends streams to a channel

417−joining combinator then makes it iterate on the directories concurrently.

460+making it iterate on the directories concurrently.

418

461

419Please see the file [ListDir.hs][] for the complete working code,

462Please see the file [ListDir.hs][] for the complete working code,

420including the imports that we have omitted below.

463including the imports that we have omitted below.

421

464

422−```haskell

465+```haskell ghci

466+import Data.Bifunctor (bimap)

467+import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

423−import Streamly.Internal.Data.Stream.IsStream (iterateMapLeftsWith)

468+import qualified Streamly.Internal.Data.Stream as Stream (iterateMapLeftsWith)

424−

425−import qualified Streamly.Prelude as Stream

426−import qualified Streamly.Internal.FileSystem.Dir as Dir (toEither)

469+import qualified Streamly.Internal.FileSystem.Dir as Dir (readEither)

470+import qualified Streamly.Internal.Data.Stream.Concurrent.Channel as Concur

427

471

428-- Lists a directory as a stream of (Either Dir File).

472-- Lists a directory as a stream of (Either Dir File).

429−listDir :: String -> SerialT IO (Either String String)

473+listDir :: String -> Stream IO (Either String String)

430listDir dir =

474listDir dir =

431− Dir.toEither dir -- SerialT IO (Either String String)

475+ Dir.readEither dir

432− & Stream.map (bimap mkAbs mkAbs) -- SerialT IO (Either String String)

476+ & fmap (bimap mkAbs mkAbs)

433

477

434 where mkAbs x = dir ++ "/" ++ x

478 where mkAbs x = dir ++ "/" ++ x

435

479

436-- | List the current directory recursively using concurrent processing.

480-- | List the current directory recursively using concurrent processing.

437main :: IO ()

481main :: IO ()

438main = do

482main = do

483+ chan <- Concur.newChannel (Concur.ordered True)

484+ let combineOrdered s1 s2 = do

485+ Stream.fromEffect $ Concur.toChannel chan s1

486+ Stream.fromEffect $ Concur.toChannel chan s2

487+ Concur.fromChannel chan

439 hSetBuffering stdout LineBuffering

488 hSetBuffering stdout LineBuffering

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

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

441− Stream.iterateMapLeftsWith Stream.ahead listDir start

490+ Stream.iterateMapLeftsWith combineOrdered listDir start

442− & Stream.mapM_ print

491+ & Stream.fold (Fold.drainMapM print)

443```

492```

444

493

445### Rate Limiting

494### Rate Limiting

@@ -448,14 +497,15 @@ For bounded concurrent streams, a stream yield rate can be specified

448easily. For example, to print "tick" once every second you can simply

497easily. For example, to print "tick" once every second you can simply

449write:

498write:

450

499

451−```haskell

500+```haskell ghci

501+import qualified Streamly.Internal.Data.Stream as Stream (timestamped)

502+

452main :: IO ()

503main :: IO ()

453main =

504main =

454− Stream.repeatM (pure "tick") -- AsyncT IO String

505+ Stream.repeat (pure "tick")

455− & Stream.timestamped -- AsyncT IO (AbsTime, String)

506+ & Concur.sequenceWith (Concur.avgRate 1)

456− & Stream.avgRate 1 -- AsyncT IO (AbsTime, String)

507+ & Stream.timestamped

457− & Stream.fromAsync -- SerialT IO (AbsTime, String)

458− & Stream.mapM_ print -- IO ()

508+ & Stream.fold (Fold.drainMapM print)

459```

509```

460

510

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

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

@@ -469,13 +519,9 @@ documentation][Streamly].

469

519

470### Reactive Programming

520### Reactive Programming

471

521

472−Streamly supports reactive (time domain) programming because of its

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

473−support for declarative concurrency. Please see the `Streamly.Prelude`

523+declarative concurrency. Please see the `Streamly.Internal.Data.Stream.Time`

474−module for time-specific combinators like `intervalsOf`, and

524+module for time-specific and sampling combinators.

475−folds like `takeInterval` in `Streamly.Internal.Data.Fold`.

476−Please also see the pre-release sampling combinators in the

477−`Streamly.Internal.Data.Stream.IsStream.Top` module for `throttle` and

478−`debounce` like operations.

479

525

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

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

481reactive programming using [Streamly][].

527reactive programming using [Streamly][].