8adf0c3 · streamly · git.iko.soy

git.iko.soy/streamly

Update the docs and sync with current version

commit8adf0c3c3d5ce7a08bfca314136cf870a5de3b7c authorHarendra Kumar <harendra@composewell.com> dateFri 6 Jan 2023, 10:07 (4 years ago) parent43afc26

2 files changed+102−158

compare with master

unifiedsplit

docs/User/HowTo/faq.md+44−78 docs/User/Tutorials/Introduction.md+58−80

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

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

88## Distribute and Zip Concurrently

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

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

12−results together and consume the resulting stream concurrently.

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

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

12+together to create a single transformed stream.

1313

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

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`)

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

2015

2116```haskell ghci

2217{-# LANGUAGE FlexibleContexts #-}

2318

2419import Data.Function ((&))

2520import qualified Streamly.Data.Fold as Fold

26−import qualified Streamly.Data.Stream as Stream

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

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

2822

2923f1 x =

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

31− & Stream.fromList -- Stream of actions

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

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

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

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

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

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

3428```

3529

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

37−can use `zipWith`:

38−

39−- Create 2 streams of actions.

40−- Sequence the actions concurrently.

41−- Zip the resulting streams.

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

31+streams:

4232

4333```haskell ghci

44−import Data.Maybe (fromJust)

4534

4635f2 x =

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

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

49− actionStream3 = Stream.fromPure (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

56− & fmap fromJust

36+ let app = Stream.parApply id

37+ in (,,)

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

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

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

41+ & Stream.fold Fold.one

5742```

5843

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:

10245

10346```haskell ghci

10447g f xs =

10548 Stream.fromList xs

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

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

10750 & Stream.fold Fold.toList

10851```

10952

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

54+transformations using the distribute/zip operations.

55+

11056```haskell docspec

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

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

11359

11460>>> 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)

11678```

11779

11880## Sliding Window

11981

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

83+

12084```haskell docspec

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

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

12286>>> :{

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

12488& Stream.scan (Array.writeLastN 2)

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

12690:}

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

12892```

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 ((&))

101101

102102import qualified Streamly.Data.Fold as Fold

103103import qualified Streamly.Data.Stream as Stream

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

104+import qualified Streamly.FileSystem.File as File

105105

106106wcb :: String -> IO Int

107107wcb file =

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

165165function (`(,,)`).

166166

167167```haskell ghci

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

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

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

170169

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

172171countAll :: 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

174173

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

176175wc file =

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

296295```haskell ghci

297296{-# LANGUAGE FlexibleContexts #-}

298297

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

300−

301−-- Number of threads we can afford

302−numCapabilities :: Int

303−numCapabilities = 10

298+import GHC.Conc (numCapabilities)

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

304300

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

306302wc file = do

307− Stream.unfold File.chunkReader file

308− & Concur.mapMWith

309− ( Concur.maxThreads numCapabilities

310− . Concur.ordered True )

311− partialCounts

312− & Stream.fold

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

314−```

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

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

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

306+

307+ where

315308

316−Please note that we use the mapping function from

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

318−our stream are concurrent.

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

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

311+```

319312

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

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.

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

324314

325315A benchmark with 2 CPUs:

326316

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

376366-- connection is closed.

377367lookupWords :: Socket -> IO ()

378368lookupWords sk =

379− Stream.unfold Socket.reader sk

380− & Unicode.decodeLatin1

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

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

383− & fmap show

384− & Stream.intersperse "\n"

385− & Unicode.encodeStrings Unicode.encodeLatin1

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

370+ & Unicode.decodeLatin1 -- Stream IO Char

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

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

373+ & fmap show -- Stream IO String

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

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

386376 & Stream.fold (Socket.writeChunks sk)

387377

378+ where

379+

380+ word = Parser.wordBy isSpace Fold.toList

381+ cfg = Stream.ordered True

382+

388383serve :: Socket -> IO ()

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

390385

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

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

394389main :: IO ()

395390main =

396− Stream.unfold TCP.acceptorOnPort 8091

397− & Concur.mapMWith id serve

398− & Stream.fold Fold.drain

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

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

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

399394```

400395

401396### Merging Incoming Streams

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

424419-- buffering for safety.

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

426421readLines sk =

427− Stream.unfold Socket.reader sk

428− & Unicode.decodeLatin1

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

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

423+ & Unicode.decodeLatin1 -- Stream IO Char

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

425+

426+ where

427+

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

430429

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

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

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

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

438437server :: Handle -> IO ()

439438server file =

440− Stream.unfold TCP.acceptorOnPort 8090

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

442− & Stream.unfoldMany Array.reader

443− & Unicode.encodeLatin1

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

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

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

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

442+ & Unicode.encodeLatin1 -- Stream IO Word8

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

445444

446445main :: IO ()

447446main = withFile "output.txt" AppendMode server

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

449448

450449### Listing Directories Recursively/Concurrently

451450

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

453−multiple directories concurrently.

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

454452

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

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

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

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

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

460−making it iterate on the directories concurrently.

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

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

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

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

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

461458

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

463−including the imports that we have omitted below.

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

464460

465461```haskell ghci

466−import Data.Bifunctor (bimap)

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

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

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

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

479464

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

481465main :: IO ()

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

488467 hSetBuffering stdout LineBuffering

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

490− Stream.iterateMapLeftsWith combineOrdered listDir start

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

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

470+ ls = Stream.parConcatIterate id f start

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

492472```

493473

494474### Rate Limiting

495475

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

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

498−write:

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

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

499478

500479```haskell ghci

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

502481

503482main :: IO ()

504483main =

505− Stream.repeat (pure "tick")

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

507− & Stream.timestamped

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

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

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

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

509487```

510488

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

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

520498### Reactive Programming

521499

522500Streamly 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`

524502module for time-specific and sampling combinators.

525503

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