8adf0c3 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Update the docs and sync with current version

8adf0c3Harendra Kumar4 years ago

docs/User/HowTo/faq.md+44−78

compare with master

unifiedsplit

@@ -7,118 +7,82 @@ This document provides additional examples.

8## Distribute and Zip Concurrently

10−Say, you have an input stream, and each element of the input stream is to be

10+Transform a stream in multiple ways, generating multiple transformed

11−transformed in multiple ways using monadic functions. Then you want to zip the

11+streams and then zip the corresponding elements of each resulting stream

12−results together and consume the resulting stream concurrently.

12+together to create a single transformed stream.

14−First, distribute an item to multiple effectful consumer concurrently, for

14+Distributing a value to a stream of consumers concurrently:

15−example:

16−

17−- Create a stream of actions.

18−- Sequence the actions concurrently.

19− - Note that we respect the order of input. (`Config.ordered True`)

21```haskell ghci

16```haskell ghci

22{-# LANGUAGE FlexibleContexts #-}

17{-# LANGUAGE FlexibleContexts #-}

24import Data.Function ((&))

19import Data.Function ((&))

25import qualified Streamly.Data.Fold as Fold

20import qualified Streamly.Data.Fold as Fold

26−import qualified Streamly.Data.Stream as Stream

21+import qualified Streamly.Data.Stream.Prelude as Stream

27−import qualified Streamly.Data.Stream.Concurrent as Concur

29f1 x =

23f1 x =

30− Prelude.map ($ x) [return . (+ 1), return . (+ 2)] -- List of actions

24+ Stream.fromList [return . (+ 1), return . (+ 2)] -- Stream of functions

31− & Stream.fromList -- Stream of actions

25+ & fmap ($ x) -- Stream of lazy actions

32− & Concur.sequenceWith (Concur.ordered True) -- Sequence the actions concurrently

26+ & Stream.parSequence (Stream.ordered True) -- Evaluate concurrently

33− & Stream.fold Fold.toList -- Fold the resulting stream into a list

27+ & Stream.fold Fold.toList -- Fold to list

34```

28```

36−Alternatively, if you want to concurrently zip different types of outputs, you

30+Use `parApply` to zip streams concurrently. Here, we zip three singleton

37−can use `zipWith`:

31+streams:

38−

39−- Create 2 streams of actions.

40−- Sequence the actions concurrently.

41−- Zip the resulting streams.

43```haskell ghci

33```haskell ghci

44−import Data.Maybe (fromJust)

46f2 x =

35f2 x =

47− let actionStream1 = Stream.fromPure (return $ show x)

36+ let app = Stream.parApply id

37+ in (,,)

38+ `fmap` Stream.fromEffect (return $ show x)

48− actionStream2 = Stream.fromPure (return $ x + 1)

39+ `app` Stream.fromEffect (return $ x + 1)

49− actionStream3 = Stream.fromPure (return $ fromIntegral x / 2)

40+ `app` Stream.fromEffect (return $ fromIntegral x / 2)

50− in Stream.zipWith (\a (b, c) -> (a, b, c))

51− (Concur.sequence actionStream1)

52− (Stream.zipWith (,)

53− (Concur.sequence actionStream2)

54− (Concur.sequence actionStream3))

55− & Stream.fold Fold.one

41+ & Stream.fold Fold.one

56− & fmap fromJust

57```

42```

59−Alternatively, generate a type with concurrent zipping property to use the

60−`Applicative` instance,

61−

62−```haskell

63−{-# LANGUAGE UndecidableInstances #-}

64−{-# LANGUAGE FlexibleInstances #-}

65−{-# LANGUAGE MultiParamTypeClasses #-}

66−{-# LANGUAGE TemplateHaskell #-}

67−

68−import Control.Monad.Catch (MonadThrow(..))

69−import Control.Monad.IO.Class (MonadIO(..))

70−import Control.Monad.Reader.Class (MonadReader(..))

71−import Control.Monad.Trans.Class (MonadTrans(..))

72−import Data.Function ((&))

73−import Data.Maybe (fromJust)

74−import Streamly.Data.Stream (Stream)

75−import Streamly.Data.Stream.Concurrent (MonadAsync)

76−

77−import qualified Streamly.Data.Fold as Fold

78−import qualified Streamly.Data.Stream

79−import qualified Streamly.Data.Stream as Stream

80−import qualified Streamly.Data.Stream.Concurrent

81−import qualified Streamly.Data.Stream.Concurrent as Concur

82−import qualified Streamly.Internal.Data.Stream.TypeGen as StreamTypeGen

83−

84−applyConcurrent :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b

85−applyConcurrent = Concur.zipWith ($)

86−

87−$(StreamTypeGen.mkZippingType "ZipConcurrent" "applyConcurrent" True)

88−

89−f2 x =

90− let actionStream1 = Stream.fromEffect (return $ show x)

91− actionStream2 = Stream.fromEffect (return $ x + 1)

92− actionStream3 = Stream.fromEffect (return $ fromIntegral x / 2)

93− in ((,,,)

94− <$> actionStream1

95− <*> actionStream2

96− <*> actionStream3)

97− & Stream.fold Fold.one

98− & fmap fromJust

99−```

100−

101−Then apply the function `f` concurrently to your input stream:

44+Applying a function concurrently to your input stream:

102

103```haskell ghci

46```haskell ghci

104g f xs =

47g f xs =

105 Stream.fromList xs

48 Stream.fromList xs

106− & Concur.mapMWith (Concur.ordered True) f -- Map the actions concurrently

49+ & Stream.parMapM (Stream.ordered True) f

107 & Stream.fold Fold.toList

50 & Stream.fold Fold.toList

108```

51```

109

53+You can now use the concurrent map to pipe each element through multiple

54+transformations using the distribute/zip operations.

55+

110```haskell docspec

56```haskell docspec

111>>> g f1 [1,2,3,4::Int]

57>>> g f1 [1,2,3,4::Int]

112[[2,3],[3,4],[4,5],[5,6]]

58[[2,3],[3,4],[4,5],[5,6]]

113

114>>> g f2 [1,2,3,4::Int]

60>>> g f2 [1,2,3,4::Int]

115−[("1",2,0.5),("2",3,1.0),("3",4,1.5),("4",5,2.0)]

61+[Just ("1",2,0.5),Just ("2",3,1.0),Just ("3",4,1.5),Just ("4",5,2.0)]

62+```

63+

64+Instead of using `parApply` directly, you can use `mkZipType` to

65+create a zip Applicative newtype so that you can use the `Applicative`

66+instance.

67+

68+```haskell

69+{-# LANGUAGE UndecidableInstances #-}

70+{-# LANGUAGE FlexibleInstances #-}

71+{-# LANGUAGE MultiParamTypeClasses #-}

72+{-# LANGUAGE TemplateHaskell #-}

73+

74+import Streamly.Internal.Data.Stream.TypeGen

75+

76+app = parApply id

77+$(mkZippingType "ZipConcurrent" "app" True)

116```

78```

117

118## Sliding Window

80## Sliding Window

119

82+The `writeLastN` fold can be used to create a stream of sliding windows.

83+

120```haskell docspec

84```haskell docspec

121−>>> import qualified Streamly.Internal.Data.Array as Array

85+>>> import qualified Streamly.Data.Array as Array

122>>> :{

86>>> :{

123 Stream.fromList [1,2,3,4,5::Int]

87 Stream.fromList [1,2,3,4,5::Int]

124& Stream.scan (Array.writeLastN 2)

88& Stream.scan (Array.writeLastN 2)

@@ -126,3 +90,5 @@ g f xs =

126:}

90:}

127[fromList [],fromList [1],fromList [1,2],fromList [2,3],fromList [3,4],fromList [4,5]]

91[fromList [],fromList [1],fromList [1,2],fromList [2,3],fromList [3,4],fromList [4,5]]

128```

92```

93+

94+Also see the "Streamly.Internal.Data.Fold.Window" module for widnow based folds.

docs/User/Tutorials/Introduction.md+58−80

@@ -101,7 +101,7 @@ import Data.Function ((&))

101

102import qualified Streamly.Data.Fold as Fold

103import qualified Streamly.Data.Stream as Stream

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

104+import qualified Streamly.FileSystem.File as File

105

106wcb :: String -> IO Int

107wcb file =

@@ -165,12 +165,11 @@ then combines the outputs from the folds using the supplied combiner

165function (`(,,)`).

166

167```haskell ghci

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

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

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

170

169

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

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

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

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

173−countAll = Tee.toFold $ (,,) <$> Tee Fold.length <*> Tee nlines <*> Tee nwords

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

174

173

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

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

176wc file =

175wc file =

@@ -296,31 +295,22 @@ from each chunk.

296```haskell ghci

295```haskell ghci

297{-# LANGUAGE FlexibleContexts #-}

296{-# LANGUAGE FlexibleContexts #-}

298

297

298+import GHC.Conc (numCapabilities)

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

299+import qualified Streamly.Data.Stream.Prelude as Stream

300−

301−-- Number of threads we can afford

302−numCapabilities :: Int

303−numCapabilities = 10

304

300

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

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

306wc file = do

302wc file = do

307− Stream.unfold File.chunkReader file

303+ File.readChunks file -- Stream IO (Array Word8)

308− & Concur.mapMWith

309− ( Concur.maxThreads numCapabilities

310− . Concur.ordered True )

311− partialCounts

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

312− & Stream.fold

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

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

306+

314−```

307+ where

315

308

316−Please note that we use the mapping function from

309+ cfg = Stream.maxThreads numCapabilities . Stream.ordered True

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

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

318−our stream are concurrent.

311+```

319

312

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

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

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

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

323−have expressed the concurrent word counting problem.

324

314

325A benchmark with 2 CPUs:

315A benchmark with 2 CPUs:

326

316

@@ -376,15 +366,20 @@ fetch w = threadDelay 1000000 >> return (w,w)

376-- connection is closed.

366-- connection is closed.

377lookupWords :: Socket -> IO ()

367lookupWords :: Socket -> IO ()

378lookupWords sk =

368lookupWords sk =

379− Stream.unfold Socket.reader sk

369+ Stream.unfold Socket.reader sk -- Stream IO Word8

380− & Unicode.decodeLatin1

370+ & Unicode.decodeLatin1 -- Stream IO Char

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

371+ & Stream.parseMany word -- Stream IO String

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

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

383− & fmap show

373+ & fmap show -- Stream IO String

384− & Stream.intersperse "\n"

374+ & Stream.intersperse "\n" -- Stream IO String

385− & Unicode.encodeStrings Unicode.encodeLatin1

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

386 & Stream.fold (Socket.writeChunks sk)

376 & Stream.fold (Socket.writeChunks sk)

387

377

378+ where

379+

380+ word = Parser.wordBy isSpace Fold.toList

381+ cfg = Stream.ordered True

382+

388serve :: Socket -> IO ()

383serve :: Socket -> IO ()

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

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

390

385

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

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

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

394main :: IO ()

389main :: IO ()

395main =

390main =

396− Stream.unfold TCP.acceptorOnPort 8091

391+ Stream.unfold TCP.acceptorOnPort 8091 -- Stream IO Socket

397− & Concur.mapMWith id serve

392+ & Stream.parMapM id serve -- Stream IO ()

398− & Stream.fold Fold.drain

393+ & Stream.fold Fold.drain -- IO ()

399```

394```

400

395

401### Merging Incoming Streams

396### Merging Incoming Streams

@@ -424,9 +419,13 @@ import qualified Streamly.FileSystem.Handle as Handle

424-- buffering for safety.

419-- buffering for safety.

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

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

426readLines sk =

421readLines sk =

427− Stream.unfold Socket.reader sk

422+ Stream.unfold Socket.reader sk -- Stream IO Word8

428− & Unicode.decodeLatin1

423+ & Unicode.decodeLatin1 -- Stream IO Char

424+ & Stream.foldMany line -- Stream IO (Array Char)

425+

426+ where

427+

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

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

430

429

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

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

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

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

@@ -437,11 +436,11 @@ recv sk = Stream.finallyIO (close sk) (readLines sk)

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

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

438server :: Handle -> IO ()

437server :: Handle -> IO ()

439server file =

438server file =

440− Stream.unfold TCP.acceptorOnPort 8090

439+ Stream.unfold TCP.acceptorOnPort 8090 -- Stream IO Socket

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

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

442− & Stream.unfoldMany Array.reader

441+ & Stream.unfoldMany Array.reader -- Stream IO Char

443− & Unicode.encodeLatin1

442+ & Unicode.encodeLatin1 -- Stream IO Word8

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

443+ & Stream.fold (Handle.write file) -- IO ()

445

444

446main :: IO ()

445main :: IO ()

447main = withFile "output.txt" AppendMode server

446main = withFile "output.txt" AppendMode server

@@ -449,63 +448,42 @@ main = withFile "output.txt" AppendMode server

449

448

450### Listing Directories Recursively/Concurrently

449### Listing Directories Recursively/Concurrently

451

450

452−Our next example lists a directory tree recursively, reading

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

453−multiple directories concurrently.

454

452

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

453+This example uses the tree traversing combinator `parConcatIterate`. This

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

454+combinator maps a stream generator function on the input stream and then

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

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

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

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

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

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

460−making it iterate on the directories concurrently.

461

458

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

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

463−including the imports that we have omitted below.

464

460

465```haskell ghci

461```haskell ghci

466−import Data.Bifunctor (bimap)

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

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

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

463+import qualified Streamly.Internal.FileSystem.Dir as Dir (readEitherPaths)

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

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

471−

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

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

474−listDir dir =

475− Dir.readEither dir

476− & fmap (bimap mkAbs mkAbs)

477−

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

479

464

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

481main :: IO ()

465main :: IO ()

482main = do

466main = 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

488 hSetBuffering stdout LineBuffering

467 hSetBuffering stdout LineBuffering

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

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

490− Stream.iterateMapLeftsWith combineOrdered listDir start

469+ f = either Dir.readEitherPaths (const Stream.nil)

470+ ls = Stream.parConcatIterate id f start

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

471+ in Stream.fold (Fold.drainMapM print) ls

492```

472```

493

473

494### Rate Limiting

474### Rate Limiting

495

475

496−For bounded concurrent streams, a stream yield rate can be specified

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

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

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

498−write:

499

478

500```haskell ghci

479```haskell ghci

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

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

502

481

503main :: IO ()

482main :: IO ()

504main =

483main =

505− Stream.repeat (pure "tick")

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

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

507− & Stream.timestamped

485+ & Stream.timestamped -- Stream IO (AbsTime, String)

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

486+ & Stream.fold (Fold.drainMapM print) -- IO ()

509```

487```

510

488

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

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

@@ -520,7 +498,7 @@ documentation][Streamly].

520### Reactive Programming

498### Reactive Programming

521

499

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

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

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

501+declarative concurrency. Please see the `Streamly.Data.Stream.Prelude`

524module for time-specific and sampling combinators.

502module for time-specific and sampling combinators.

525

503

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

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