54815c5 · streamly · git.iko.soy

git.iko.soy/streamly

Break Tutorial module into multiple small modules

commit54815c5ee3570b5f57e2f22090808e38f973d095 authorHarendra Kumar <harendra@composewell.com> dateWed 16 Jun 2021, 00:11 (5 years ago) parentb43facf

7 files changed+990−892

compare with master

unifiedsplit

docs/ConcurrentStreams.hs+780 docs/ReactiveProgramming.hs+124 docs/Tutorial.hs+42−888 docs/performance.md+7 docs/streamly-docs.cabal+2 docs/transformers.md+34−1 streamly.cabal+1−3

docs/ConcurrentStreams.hsnew file+780

@@ -0,0 +1,780 @@

1+{-# OPTIONS_GHC -fno-warn-unused-imports #-}

2+-- |

3+-- Module : ConcurrentStreams

5+--

6+-- License : BSD3

7+-- Maintainer : streamly@composewell.com

8+--

9+-- In this tutorial we will show how streamly can be used for idiomatic and

10+-- declarative concurrent programming. Before you go through this tutorial we

11+-- recommend that you take a look at the Streamly serial streams tutorial.

12+

13+module ConcurrentStreams

14+ (

15+ -- * Concurrent Streams

16+ -- $concurrentStreams

17+

18+ -- * Combining Streams

19+ -- $flavors

20+

21+ -- * Imports and Supporting Code

22+ -- $imports

23+

24+ -- * Generating Streams Concurrently

25+ -- $generatingConcurrently

26+

27+ -- * Concurrent Pipeline Stages

28+ -- $concurrentApplication

29+

30+ -- * Mapping Concurrently

31+ -- $concurrentTransformation

32+

33+ -- * Merging Streams

34+

35+ -- ** Semigroup Style

36+

37+ -- *** Deep Speculative Composition ('Ahead')

38+ -- $ahead

39+

40+ -- *** Deep Asynchronous Composition ('Async')

41+ -- $async

42+

43+ -- *** Wide Asynchronous Composition ('WAsync')

44+ -- $wasync

45+

46+ -- *** Parallel Asynchronous Composition ('Parallel')

47+ -- $parallel

48+

49+ -- XXX we should deprecate and remove the mkAsync API

50+ -- Custom composition

51+ -- custom

52+

53+ -- ** Monoid Style

54+ -- $monoid

55+

56+ -- * Nesting Streams

57+

58+ -- ** Monad

59+ -- *** Deep Speculative Nesting ('Ahead')

60+ -- $aheadNesting

61+

62+ -- *** Deep Asynchronous Nesting ('Async')

63+ -- $concurrentNesting

64+

65+ -- *** Wide Asynchronous Nesting ('WAsync')

66+ -- $wasyncNesting

67+

68+ -- *** Parallel Asynchronous Nesting ('Parallel')

69+ -- $parallelNesting

70+

71+ -- ** Applicative

72+ -- $applicative

73+

74+ -- * Zipping Streams

75+

76+ -- ** Parallel Zipping

77+ -- $parallelzip

78+

79+ -- * Concurrent Programming

80+ -- $concurrent

81+

82+ -- * Writing Concurrent Programs

83+ -- $programs

84+

85+ -- * Where to go next?

86+ -- $furtherReading

87+ )

88+where

89+

90+import Streamly.Prelude

91+import Data.Semigroup

92+import Control.Applicative

93+import Control.Monad

94+import Control.Monad.IO.Class (MonadIO(..))

95+import Control.Monad.Trans.Class (MonadTrans (lift))

96+

97+-- CAUTION: please keep setup and imports sections in sync

98+--

99+-- $setup

100+-- >>> :m

101+-- >>> import Data.Function ((&))

102+-- >>> import Streamly.Prelude ((|:), (|&))

103+-- >>> import qualified Streamly.Prelude as Stream

104+-- >>> import qualified Streamly.Data.Fold as Fold

105+--

106+-- >>> import Control.Concurrent (threadDelay, myThreadId)

107+-- >>> :{

108+-- delay n = Stream.fromEffect $ do

109+-- threadDelay (n * 1000000)

110+-- tid <- myThreadId

111+-- putStrLn (show tid ++ ": Delay " ++ show n)

112+-- :}

113+--

114+-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

115+-- >>> hSetBuffering stdout LineBuffering

116+--

117+

118+-- $imports

119+--

120+-- In most of example snippets we do not repeat the imports. Where imports are

121+-- not explicitly specified use the imports shown below.

122+--

123+-- >>> :m

124+-- >>> import Data.Function ((&))

125+-- >>> import Streamly.Prelude ((|:), (|&))

126+-- >>> import qualified Streamly.Prelude as Stream

127+-- >>> import qualified Streamly.Data.Fold as Fold

128+--

129+-- To illustrate concurrent vs serial composition aspects, we will use the

130+-- following @delay@ function to introduce a sleep or delay specified in

131+-- seconds. After the delay it prints the number of seconds it slept.

132+--

133+-- >>> import Control.Concurrent (threadDelay, myThreadId)

134+-- >>> :{

135+-- delay n = Stream.fromEffect $ do

136+-- threadDelay (n * 1000000)

137+-- tid <- myThreadId

138+-- putStrLn (show tid ++ ": Delay " ++ show n)

139+-- :}

140+--

141+-- For concurrent examples, use line buffering, otherwise output from different

142+-- threads may get mixed:

143+--

144+-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

145+-- >>> hSetBuffering stdout LineBuffering

146+--

147+

148+-- $concurrentStreams

149+--

150+-- Many stream operations can be done concurrently:

151+--

152+-- * Streams can be generated concurrently.

153+--

154+-- * Streams can be merged concurrently.

155+--

156+-- * Multiple stages in a streaming pipeline can run concurrently.

157+--

158+-- * Streams can be mapped and zipped concurrently.

159+--

160+-- * In monadic composition they combine like a list transformer,

161+-- providing concurrent non-determinism.

162+--

163+-- There are three basic concurrent stream styles, 'Ahead', 'Async', and

164+-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that

165+-- they can speculatively execute multiple stream actions concurrently in

166+-- advance. 'Ahead' would return exactly the same stream as 'Serial' except

167+-- that it may execute the actions concurrently. The 'Async' style streams,

168+-- like 'Ahead', speculatively execute multiple stream actions in advance but

169+-- return the results in their finishing order rather than in the stream

170+-- traversal order. 'Parallel' is like 'Async' except that it provides

171+-- unbounded parallelism instead of controlled parallelism.

172+--

173+-- For easy reference, we can classify the stream types based on /execution order/,

174+-- /consumption order/, and /bounded or unbounded/ concurrency.

175+-- Execution could be serial (i.e. synchronous) or asynchronous. In serial

176+-- execution we execute the next action in the stream only after the previous

177+-- one has finished executing. In asynchronous execution multiple actions in

178+-- the stream can be executed asynchronously i.e. the next action can start

179+-- executing even before the first one has finished. Consumption order

180+-- determines the order in which the outputs generated by the composition are

181+-- consumed. Consumption could be serial or asynchronous. In serial

182+-- consumption, the outputs are consumed in the traversal order, in

183+-- asynchronous consumption the outputs are consumed as they arrive i.e. first

184+-- come first serve order.

185+--

186+-- +------------+--------------+--------------+--------------+

188+-- +============+==============+==============+==============+

190+-- +------------+--------------+--------------+--------------+

192+-- +------------+--------------+--------------+--------------+

194+-- +------------+--------------+--------------+--------------+

196+-- +------------+--------------+--------------+--------------+

197+--

198+-- All these types can be freely inter-converted using type conversion

199+-- combinators or type annotations, without any cost, to achieve the desired

200+-- composition style. To force a particular type of composition, we coerce the

201+-- stream type using the corresponding type adapting combinator from

202+-- 'fromSerial', 'fromAhead', 'fromAsync', or 'fromParallel'. The default stream type

203+-- is inferred as 'Serial' unless you change it by using one of the combinators

204+-- or by using a type annotation.

205+

206+-- $flavors

207+--

208+-- Streams can be combined using '<>' or 'mappend' to form a

209+-- composite. Composite streams can be interpreted in a depth first or

210+-- breadth first manner using an appropriate type conversion before

211+-- consumption. Deep (e.g. 'Serial') stream type variants traverse a

212+-- composite stream in a depth first manner, such that each stream is

213+-- traversed fully before traversing the next stream. Wide

214+-- (e.g. 'WSerial') stream types traverse it in a breadth first

215+-- manner, such that one element from each stream is traversed before

216+-- coming back to the first stream again.

217+--

218+-- Each stream type has a wide traversal variant prefixed by 'W'. The wide

219+-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad

220+-- compositions of the streams.

221+-- The following table summarizes the basic types and the corresponding wide

222+-- variants:

223+--

224+-- @

225+-- +------------+-----------+

226+-- | Deep | Wide |

227+-- +============+===========+

228+-- | 'Serial' | 'WSerial' |

229+-- +------------+-----------+

230+-- | 'Ahead' | 'WAhead' |

231+-- +------------+-----------+

232+-- | 'Async' | 'WAsync' |

233+-- +------------+-----------+

234+-- @

235+--

236+-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that

237+-- zip streams serially or concurrently using 'Applicative' operation. These

238+-- types are not monads they are only applicatives and they do not differ in

239+-- 'Semigroup' composition.

240+--

241+

242+-- $programs

243+--

244+-- When writing concurrent programs it is advised to not use the concurrent

245+-- style stream combinators blindly at the top level. That might create too

246+-- much concurrency where it is not even required, and can even degrade

247+-- performance in some cases. In some cases it can also lead to surprising

248+-- behavior because of some code that is supposed to be serial becoming

249+-- concurrent. Please be aware that all concurrency capable APIs that you may

250+-- have used under the scope of a concurrent stream combinator will become

251+-- concurrent. For example if you have a 'repeatM' somewhere in your program

252+-- and you use 'fromParallel' on top, the 'repeatM' becomes fully parallel,

253+-- resulting into an infinite parallel execution . Instead, use the

254+-- /Keep It Serial and Stupid/ principle, start with the default serial

255+-- composition and enable concurrent combinators only when and where necessary.

256+-- When you use a concurrent combinator you can use an explicit 'fromSerial'

257+-- combinator to suppress any unnecessary concurrency under the scope of that

258+-- combinator.

259+

260+-- $generatingConcurrently

261+--

262+-- Monadic construction and generation functions like 'consM', 'unfoldrM',

263+-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' work concurrently

264+-- when used with appropriate stream type combinator. The pure versions of

265+-- these APIs are not concurrent, however you can use the monadic versions even

266+-- for pure computations by wrapping the pure value in a monad to get the

267+-- concurrent generation capability where required.

268+--

269+-- The following code finishes in 3 seconds (6 seconds when serial):

270+--

271+-- >>> let p n = threadDelay (n * 1000000) >> return n

272+-- >>> Stream.toList $ Stream.fromParallel $ p 3 |: p 2 |: p 1 |: Stream.nil

273+-- [1,2,3]

274+--

275+-- >>> Stream.toList $ Stream.fromAhead $ p 3 |: p 2 |: p 1 |: Stream.nil

276+-- [3,2,1]

277+--

278+-- The following finishes in 10 seconds (100 seconds when serial):

279+--

280+-- >>> Stream.drain $ Stream.fromAsync $ Stream.replicateM 10 $ p 10

281+--

282+

283+-- $concurrentTransformation

284+--

285+-- Monadic transformation functions 'mapM' and 'sequence' work concurrently

286+-- when used with appropriate stream type combinators. The pure versions do not

287+-- work concurrently, however you can use the monadic versions even for pure

288+-- computations to get the concurrent transformation capability where required.

289+--

290+-- This would print a value every second (2 seconds when serial):

291+--

292+-- >>> let p n = threadDelay (n * 1000000) >> return n

293+-- >>> :{

294+-- parMap =

295+-- Stream.repeatM (p 1)

296+-- & Stream.fromSerial -- repeatM is serial

297+-- & Stream.mapM (\x -> p 1 >> print x)

298+-- & Stream.fromAhead -- mapM is cocnurrent using Ahead style

299+-- & Stream.drain

300+-- :}

301+--

302+

303+-- $concurrentApplication

304+--

305+-- The concurrent function application operators '|$' and '|&' apply a stream

306+-- argument to a stream function concurrently to compose a concurrent pipeline

307+-- of stream processing functions:

308+--

309+-- Because both the stages run concurrently, we would see a delay of only 1

310+-- second instead of 2 seconds in the following:

311+--

312+-- >>> let p n = threadDelay (n * 1000000) >> return n

313+-- >>> :{

314+-- parApp =

315+-- Stream.repeatM (p 1)

316+-- |& Stream.mapM (\x -> p 1 >> print x)

317+-- & Stream.drain

318+-- :}

319+

320+-- $ahead

321+--

322+-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a

323+-- /serial depth first/ manner with concurrent lookahead. We use the 'fromAhead'

324+-- type combinator to effect 'Ahead' style of composition. We can also use an

325+-- explicit 'Ahead' type annotation for the stream to achieve the same effect.

326+--

327+-- When two streams are combined in this manner, the streams are traversed in

328+-- depth first manner just like 'Serial', however it can execute the next

329+-- stream concurrently and keep the results ready when its turn arrives.

330+-- Concurrent execution of the next stream(s) is performed if the first stream

331+-- blocks or if it cannot produce output at the rate that is enough to meet the

332+-- consumer demand. Multiple streams can be executed concurrently to meet the

333+-- demand. The following example would print the result in a second even

334+-- though each action in each stream takes one second:

335+--

336+-- >>> p n = threadDelay 1000000 >> return n

337+-- >>> stream1 = p 1 |: p 2 |: Stream.nil

338+-- >>> stream2 = p 3 |: p 4 |: Stream.nil

339+-- >>> Stream.toList $ Stream.fromAhead $ stream1 <> stream2

340+-- [1,2,3,4]

341+--

342+-- Each stream is constructed 'fromAhead' and then both the streams are merged

343+-- 'fromAhead', therefore, all the actions can run concurrently but the result is

344+-- presented in serial order.

345+--

346+-- You can also use the polymorphic combinator 'ahead' in place of '<>' to

347+-- compose any type of streams in this manner.

348+

349+-- $async

350+--

351+-- The 'Semigroup' operation '<>' of the 'Async' type combines the two

352+-- streams in a depth first manner with parallel look ahead. We use the

353+-- 'fromAsync' type combinator to effect 'Async' style of composition. We

354+-- can also use the 'Async' type annotation for the stream type to achieve

355+-- the same effect.

356+--

357+-- When two streams with multiple elements are combined in this manner, the

358+-- streams are traversed in depth first manner just like 'Serial', however it

359+-- can execute the next stream concurrently and return the results from it

360+-- as they arrive i.e. the results from the next stream may be yielded even

361+-- before the results from the first stream. Concurrent execution of the next

362+-- stream(s) is performed if the first stream blocks or if it cannot produce

363+-- output at the rate that is enough to meet the consumer demand. Multiple

364+-- streams can be executed concurrently to meet the demand.

365+-- In the example below each element in the stream introduces a constant delay

366+-- of 1 second, however, it takes just one second to produce all the results.

367+-- The results are not guaranteed to be in any particular order:

368+--

369+-- >>> p n = threadDelay 1000000 >> return n

370+-- >>> stream1 = p 1 |: p 2 |: Stream.nil

371+-- >>> stream2 = p 3 |: p 4 |: Stream.nil

372+-- >>> Stream.toList $ Stream.fromAsync $ stream1 <> stream2

373+-- ...

374+--

375+-- The constituent streams are also composed in 'Async' manner and the

376+-- composition of streams too. We can compose the constituent streams to run

377+-- serially, in that case it would take 2 seconds to produce all the results.

378+-- The elements in the serial streams would be in serial order in the results:

379+--

380+-- >>> p n = threadDelay 1000000 >> return n

381+-- >>> stream = (Stream.fromSerial stream1) <> (Stream.fromSerial stream2)

382+-- >>> Stream.toList $ Stream.fromAsync stream

383+-- ...

384+--

385+-- In the following example we can see that new threads are started when a

386+-- computation blocks. Notice that the output from the stream with the

387+-- shortest delay is printed first. The whole computation takes @maximum of

388+-- (3, 2, 1) = 3@ seconds:

389+--

390+-- >>> Stream.drain $ Stream.fromAsync $ delay 3 <> delay 2 <> delay 1

391+-- ThreadId ...: Delay 1

392+-- ThreadId ...: Delay 2

393+-- ThreadId ...: Delay 3

394+--

395+-- When we have a tree of computations composed using this style, the tree is

396+-- traversed in DFS style just like the 'Serial' style, the only difference is

397+-- that here we can move on to executing the next stream if a stream blocks.

398+-- However, we will not start new threads if we have sufficient output to

399+-- saturate the consumer. This is why we call it left-biased demand driven or

400+-- adaptive concurrency style, the concurrency tends to stay on the left side

401+-- of the composition as long as possible. More threads are started based on

402+-- the pull rate of the consumer. The following example prints an output every

403+-- second as all of the actions are concurrent.

404+--

405+-- >>> Stream.drain $ Stream.fromAsync $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)

406+-- ThreadId ...: Delay 1

407+-- ThreadId ...: Delay 2

408+-- ThreadId ...: Delay 3

409+-- ThreadId ...: Delay 4

410+--

411+-- All the computations may even run in a single thread when more threads are

412+-- not needed. As you can see, in the following example the computations are

413+-- run in a single thread one after another, because none of them blocks.

414+-- However, if the thread consuming the stream were faster than the producer

415+-- then it would have started parallel threads for each computation to keep up

416+-- even if none of them blocks:

417+--

418+-- >>> :{

419+-- traced m = Stream.fromEffect (myThreadId >>= print) >> return m

420+-- stream = traced (sqrt 9) <> traced (sqrt 16) <> traced (sqrt 25)

421+-- main = Stream.drain $ Stream.fromAsync stream

422+-- :}

423+--

424+-- Note that the order of printing in the above examples may change due to

425+-- variations in scheduling latencies for concurrent threads.

426+--

427+-- The polymorphic version of the 'Async' binary operation '<>' is called

428+-- 'async'. We can use 'async' to join streams in a left biased

429+-- adaptively concurrent manner irrespective of the type, notice that we have

430+-- not used the 'fromAsync' combinator in the following example:

431+--

432+-- >>> Stream.drain $ delay 3 `Stream.async` delay 2 `Stream.async` delay 1

433+-- ThreadId ...: Delay 1

434+-- ThreadId ...: Delay 2

435+-- ThreadId ...: Delay 3

436+--

437+-- Since the concurrency provided by this operator is demand driven it cannot

438+-- be used when the composed computations start timers that are relative to

439+-- each other because all computations may not be started at the same time and

440+-- therefore timers in all of them may not start at the same time. When

441+-- relative timing among all computations is important or when we need to start

442+-- all computations at once for any reason 'Parallel' style must be used

443+-- instead.

444+--

445+-- 'Async' style utilizes resources optimally and should be preferred over

446+-- 'Parallel' or 'WAsync' unless you really need those. 'Async' should be used

447+-- when we know that the computations can run in parallel but we do not care if

448+-- they actually run in parallel or not, that decision can be left to the

449+-- scheduler based on demand. Also, note that 'async' operator can be used to fold

450+-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,

451+-- because it does not require us to run all of them at the same time in a fair

452+-- manner.

453+

454+-- $wasync

455+--

456+-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in

457+-- a concurrent manner using /breadth first traversal/. We use the 'fromWAsync'

458+-- type combinator to effect 'WAsync' style of composition. We can also use the

459+-- 'WAsync' type annotation for the stream to achieve the same effect.

460+--

461+-- When streams with multiple elements are combined in this manner, we traverse

462+-- all the streams concurrently in a breadth first manner i.e. one action from

463+-- each stream is performed and yielded to the resulting stream before we come

464+-- back to the first stream again and so on. Even though we execute the actions

465+-- in a breadth first order the outputs are consumed on a first come first

466+-- serve basis.

467+--

468+-- In the following example we can see that outputs are produced in the breadth

469+-- first traversal order but this is not guaranteed.

470+--

471+-- >>> stream1 = print 1 |: print 2 |: Stream.nil

472+-- >>> stream2 = print 3 |: print 4 |: Stream.nil

473+-- >>> Stream.drain $ Stream.fromWAsync $ stream1 <> stream2

474+-- 1

475+-- 3

476+-- 2

477+-- 4

478+--

479+-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is

480+-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first

481+-- concurrent traversal irrespective of the type, notice that we have not used

482+-- the 'fromWAsync' combinator in the following example:

483+--

484+-- >>> Stream.drain $ delay 3 `Stream.wAsync` delay 2 `Stream.wAsync` delay 1

485+-- ThreadId ...: Delay 1

486+-- ThreadId ...: Delay 2

487+-- ThreadId ...: Delay 3

488+--

489+-- Since the concurrency provided by this style is demand driven it may not

490+-- be used when the composed computations start timers that are relative to

491+-- each other because all computations may not be started at the same time and

492+-- therefore timers in all of them may not start at the same time. When

493+-- relative timing among all computations is important or when we need to start

494+-- all computations at once for any reason 'Parallel' style must be used

495+-- instead.

496+--

497+

498+-- $parallel

499+--

500+-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two

501+-- streams in a fairly concurrent manner with round robin scheduling. We use

502+-- the 'fromParallel' type combinator to effect 'Parallel' style of composition.

503+-- We can also use the 'Parallel' type annotation for the stream type to

504+-- achieve the same effect.

505+--

506+-- When two streams with multiple elements are combined in this manner, the

507+-- monadic actions in both the streams are performed concurrently with a fair

508+-- round robin scheduling. The outputs are yielded in the order in which the

509+-- actions complete. This is pretty similar to the 'WAsync' type, the

510+-- difference is that 'WAsync' is adaptive to the consumer demand and may or

511+-- may not execute all actions in parallel depending on the demand, whereas

512+-- 'Parallel' runs all the streams in parallel irrespective of the demand.

513+--

514+-- The polymorphic version of the binary operation '<>' of the 'Parallel' type

515+-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent

516+-- manner irrespective of the type, notice that we have not used the

517+-- 'fromParallel' combinator in the following example:

518+--

519+-- >>> Stream.drain $ delay 3 `Stream.parallel` delay 2 `Stream.wAsync` delay 1

520+-- ThreadId ...: Delay 1

521+-- ThreadId ...: Delay 2

522+-- ThreadId ...: Delay 3

523+--

524+-- Note that this style of composition cannot be used to combine infinite

525+-- number of streams, as it will lead to an infinite sized scheduling queue.

526+--

527+

528+-- XXX to be removed

529+-- $custom

530+--

531+-- The 'mkAsync' API can be used to create references to asynchronously running

532+-- stream computations. We can then use 'uncons' to explore the streams

533+-- arbitrarily and then recompose individual elements to create a new stream.

534+-- This way we can dynamically decide which stream to explore at any given

535+-- time. Take an example of a merge sort of two sorted streams. We need to

536+-- keep consuming items from the stream which has the lowest item in the sort

537+-- order. This can be achieved using async references to streams. See

538+-- "MergeSort.hs" in <https://github.com/composewell/streamly-examples Streamly Examples>.

539+

540+-- $monoid

541+--

542+-- All of the following are equivalent and start ten concurrent tasks each with

543+-- a delay from 1 to 10 seconds, resulting in the printing of each number every

544+-- second:

545+--

546+-- >>> :{

547+-- main = do

548+-- Stream.drain $ Stream.fromAsync $ foldMap delay [1..10]

549+-- Stream.drain $ Stream.concatFoldableWith Stream.async (map delay [1..10])

550+-- Stream.drain $ Stream.concatMapFoldableWith Stream.async delay [1..10]

551+-- Stream.drain $ Stream.concatForFoldableWith Stream.async [1..10] delay

552+-- :}

553+--

554+

555+-- $aheadNesting

556+--

557+-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except

558+-- that it can speculatively perform a bounded number of next iterations of a

559+-- loop concurrently.

560+--

561+-- >>> :{

562+-- Stream.toList $ Stream.fromAhead $ do

563+-- x <- Stream.fromFoldable [3,2,1]

564+-- delay x

565+-- return x

566+-- :}

567+-- ThreadId ...: Delay 1

568+-- ThreadId ...: Delay 2

569+-- ThreadId ...: Delay 3

570+-- [3,2,1]

571+--

572+-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can

573+-- see all the three iterations are concurrent and run in different threads,

574+-- however, the results are returned in the serial order.

575+--

576+-- Concurrency is demand driven, when multiple streams are composed using this

577+-- style, the iterations are executed in a depth first manner just like

578+-- 'Serial' i.e. nested iterations are executed before we proceed to the next

579+-- outer iteration. The only difference is that we may execute multiple future

580+-- iterations concurrently and keep the results ready.

581+--

582+-- The 'fromAhead' type combinator can be used to switch to this style of

583+-- composition. Alternatively, a type annotation can be used to specify the

584+-- type of the stream as 'Ahead'.

585+--

586+

587+-- $concurrentNesting

588+--

589+-- The 'Monad' composition of 'Async' type can perform the iterations of a

590+-- loop concurrently.

591+--

592+-- >>> :{

593+-- Stream.drain $ Stream.fromAsync $ do

594+-- x <- Stream.fromFoldable [3,2,1]

595+-- delay x

596+-- :}

597+-- ThreadId ...: Delay 1

598+-- ThreadId ...: Delay 2

599+-- ThreadId ...: Delay 3

600+--

601+-- As we can see the code after the @fromFoldable@ statement is run three

602+-- times, once for each value of @x@. All the three iterations are concurrent

603+-- and run in different threads. The iteration with least delay finishes first.

604+-- When compared to imperative programming, this can be viewed as a @for@ loop

605+-- with three concurrent iterations.

606+--

607+-- Concurrency is demand driven i.e. more concurrent iterations are started

608+-- only if the previous iterations are not able to saturate the consumer of the

609+-- output stream. This works exactly the same way as the merging of two

610+-- streams using 'async' works.

611+--

612+-- The 'fromAsync' type combinator can be used to switch to this style of

613+-- composition. Alternatively, a type annotation can be used to specify the

614+-- type of the stream as 'Async'.

615+--

616+-- When multiple streams are nested using this style, the iterations are

617+-- concurrently evaluated in a depth first manner:

618+--

619+--

620+-- >>> :{

621+-- Stream.drain $ Stream.fromAsync $ do

622+-- x <- Stream.fromFoldable [1,2]

623+-- y <- Stream.fromFoldable [3,4]

624+-- Stream.fromEffect $ putStrLn $ show (x, y)

625+-- :}

626+-- (1,3)

627+-- (1,4)

628+-- (2,3)

629+-- (2,4)

630+--

631+-- Nested iterations are given preference for concurrent evaluation i.e.

632+-- (1,4) will be scheduled in preference to (2,3).

633+

634+-- $wasyncNesting

635+--

636+-- Like 'Async', the 'Monad' composition of 'WAsync' runs the iterations of a

637+-- loop concurrently. It differs from 'Async' in the nested loop behavior. Like

638+-- 'WSerial', the nested loops in this type are traversed and executed in a

639+-- breadth first manner rather than the depth first manner of 'Async' style.

640+--

641+-- >>> :{

642+-- Stream.drain $ Stream.fromWAsync $ do

643+-- x <- Stream.fromSerial $ Stream.fromFoldable [1,2]

644+-- y <- Stream.fromSerial $ Stream.fromFoldable [3,4]

645+-- Stream.fromEffect $ putStrLn $ show (x, y)

646+-- :}

647+-- (1,3)

648+-- (1,4)

649+-- (2,3)

650+-- (2,4)

651+--

652+-- Note that (2,3) is preferred to (1,4) when evaluating the iterations

653+-- concurrently. This works exactly the same way as the merging of two streams

654+-- using 'wAsync' works.

655+--

656+-- The 'fromWAsync' type combinator can be used to switch to this style of

657+-- composition. Alternatively, a type annotation can be used to specify the

658+-- type of the stream as 'WAsync'.

659+--

660+

661+-- $parallelNesting

662+--

663+-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the

664+-- iterations of a loop concurrently.

665+--

666+-- >>> :{

667+-- Stream.drain $ Stream.fromParallel $ do

668+-- x <- Stream.fromFoldable [3,2,1]

669+-- delay x

670+-- :}

671+-- ThreadId ...: Delay 1

672+-- ThreadId ...: Delay 2

673+-- ThreadId ...: Delay 3

674+--

675+-- It differs from 'Async' and 'WAsync' in the nested loop behavior. All

676+-- iterations of the loop are run fully concurrently irrespective of the

677+-- demand. This works exactly the same way as the merging of streams using

678+-- 'parallel' works.

679+--

680+-- The 'fromParallel' type combinator can be used to switch to this style of

681+-- composition. Alternatively, a type annotation can be used to specify the

682+-- type of the stream as 'Parallel'.

683+--

684+

685+-- $applicative

686+--

687+-- 'Async' can run the iterations concurrently, therefore, it takes a total

688+-- of 6 seconds which is max (1, 2) + max (3, 4):

689+--

690+-- >>> (Stream.toList $ Stream.fromAsync $ (,) <$> s1 <*> s2) >>= print

691+-- ...

692+--

693+-- @

694+-- ThreadId 34: Delay 1

695+-- ThreadId 36: Delay 2

696+-- ThreadId 35: Delay 3

697+-- ThreadId 36: Delay 3

698+-- ThreadId 35: Delay 4

699+-- ThreadId 36: Delay 4

700+-- [(1,3),(2,3),(1,4),(2,4)]

701+-- @

702+--

703+-- Similarly, 'WAsync' as well can run the iterations concurrently, but with a

704+-- different style of scheduling than 'Async' as explained in the Monad

705+-- section, therefore, it too takes a total of 6 seconds (2 + 4):

706+--

707+-- >>> (Stream.toList $ Stream.fromWAsync $ (,) <$> s1 <*> s2) >>= print

708+-- ...

709+--

710+-- @

711+-- ThreadId 34: Delay 1

712+-- ThreadId 36: Delay 2

713+-- ThreadId 35: Delay 3

714+-- ThreadId 36: Delay 3

715+-- ThreadId 35: Delay 4

716+-- ThreadId 36: Delay 4

717+-- [(1,3),(2,3),(1,4),(2,4)]

718+-- @

719+

720+-- $parallelzip

721+--

722+-- The applicative instance of 'ZipAsync' type zips streams concurrently.

723+-- 'fromZipAsync' type combinator can be used to switch to parallel applicative

724+-- zip composition:

725+--

726+-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3

727+-- are produced concurrently, and 2 and 4 are produced concurrently:

728+--

729+-- >>> d n = delay n >> return n

730+-- >>> s1 = Stream.fromSerial $ d 1 <> d 2

731+-- >>> s2 = Stream.fromSerial $ d 3 <> d 4

732+-- >>> (Stream.toList $ Stream.fromZipAsync $ (,) <$> s1 <*> s2) >>= print

733+-- ThreadId ...: Delay 1

734+-- ThreadId ...: Delay 2

735+-- ThreadId ...: Delay 3

736+-- ThreadId ...: Delay 4

737+-- [(1,3),(2,4)]

738+--

739+

740+-- $concurrent

741+--

742+-- When writing concurrent programs there are two distinct places where the

743+-- programmer can control the concurrency. First, when /composing/ a stream by

744+-- merging multiple streams we can choose an appropriate sum style operators to

745+-- combine them concurrently or serially. Second, when /processing/ a stream in

746+-- a monadic composition we can choose one of the monad composition types to

747+-- choose the desired type of concurrency.

748+--

749+-- In the following example the squares of @x@ and @y@ are computed

750+-- concurrently using the 'async' operation and the square roots of their

751+-- sum are computed serially because of the 'fromSerial' combinator. We can

752+-- choose different combinators for the monadic processing and the stream

753+-- generation, to control the concurrency. We can also use the 'fromAsync'

754+-- combinator instead of explicitly folding with 'async'.

755+--

756+-- >>> import Data.List (sum)

757+-- >>> :{

758+-- main = do

759+-- z <- Stream.toList

760+-- $ Stream.fromSerial -- Serial monadic processing (sqrt below)

761+-- $ do

762+-- x2 <- Stream.concatForFoldableWith Stream.async [1..100] $ -- Concurrent @"for"@ loop

763+-- \x -> return $ x * x -- body of the loop

764+-- y2 <- Stream.concatForFoldableWith Stream.async [1..100] $

765+-- \y -> return $ y * y

766+-- return $ sqrt (x2 + y2)

767+-- print $ sum z

768+-- :}

769+--

770+-- We can see how this directly maps to the imperative style

771+-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators

772+-- and operators instead of the ugly pragmas.

773+--

774+-- For more concurrent programming examples see,

775+-- <https://github.com/composewell/streamly-examples>.

776+

777+-- $furtherReading

778+--

779+-- * Read the reactive programming tutorial

780+-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.

docs/ReactiveProgramming.hsnew file+124

@@ -0,0 +1,124 @@

1+{-# OPTIONS_GHC -fno-warn-unused-imports #-}

2+-- |

3+-- Module : ReactiveProgramming

5+--

6+-- License : BSD3

7+-- Maintainer : streamly@composewell.com

8+--

9+-- In this tutorial we will show how Streamly can be used for reactive

10+-- programming. Before you go through this tutorial we recommend that you take

11+-- a look at the Streamly concurrent programming tutorial.

12+

13+module ReactiveProgramming

14+ (

15+ -- * Reactive Programming

16+ -- $reactive

17+

18+ -- * Where to go next?

19+ -- $furtherReading

20+ )

21+where

22+

23+import Streamly.Prelude

24+import Data.Semigroup

25+import Control.Applicative

26+import Control.Monad

27+import Control.Monad.IO.Class (MonadIO(..))

28+import Control.Monad.Trans.Class (MonadTrans (lift))

29+

30+-- $reactive

31+--

32+-- Reactive programming is nothing but concurrent streaming which is what

33+-- streamly is all about. With streamly we can generate streams of events,

34+-- merge streams that are generated concurrently and process events

35+-- concurrently. We can do all this without any knowledge about the specifics

36+-- of the implementation of concurrency. In the following example you will see

37+-- that the code is just regular Haskell code without much streamly APIs used

38+-- (active hyperlinks are the streamly APIs) and yet it is a reactive

39+-- application.

40+--

41+-- This application has two independent and concurrent sources of event

42+-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates

43+-- events that deteriorate the health of the character in the game.

44+-- @userAction@ can be "potion" or "quit". When the user types "potion" the

45+-- health improves and the game continues.

46+--

47+-- @

48+-- {-\# LANGUAGE FlexibleContexts \#-}

49+--

50+-- import "Streamly.Prelude" (MonadAsync, SerialT)

51+-- import "Streamly.Prelude" as Stream

52+-- import Control.Monad (void)

53+-- import Control.Monad.IO.Class (MonadIO(liftIO))

54+-- import Control.Monad.State (MonadState, get, modify, runStateT)

55+--

56+-- data Event = Quit | Harm Int | Heal Int deriving (Show)

57+--

58+-- userAction :: MonadAsync m => 'SerialT' m Event

59+-- userAction = Stream.'repeatM' $ liftIO askUser

60+-- where

61+-- askUser = do

62+-- command <- getLine

63+-- case command of

64+-- "potion" -> return (Heal 10)

65+-- "harm" -> return (Harm 10)

66+-- "quit" -> return Quit

67+-- _ -> putStrLn "Type potion or harm or quit" >> askUser

68+--

69+-- acidRain :: MonadAsync m => 'SerialT' m Event

70+-- acidRain = Stream.'fromAsync' $ Stream.'constRate' 1 $ Stream.'repeatM' $ liftIO $ return $ Harm 1

71+--

72+-- data Result = Check | Done

73+--

74+-- runEvents :: (MonadAsync m, MonadState Int m) => 'SerialT' m Result

75+-- runEvents = do

76+-- event \<- userAction \`Stream.'parallel'` acidRain

77+-- case event of

78+-- Harm n -> modify (\\h -> h - n) >> return Check

79+-- Heal n -> modify (\\h -> h + n) >> return Check

80+-- Quit -> return Done

81+--

82+-- data Status = Alive | GameOver deriving Eq

83+--

84+-- getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status

85+-- getStatus result =

86+-- case result of

87+-- Done -> liftIO $ putStrLn "You quit!" >> return GameOver

88+-- Check -> do

89+-- h <- get

90+-- liftIO $ if (h <= 0)

91+-- then putStrLn "You die!" >> return GameOver

92+-- else putStrLn ("Health = " <> show h) >> return Alive

93+--

94+-- main :: IO ()

95+-- main = do

96+-- putStrLn "Your health is deteriorating due to acid rain, type \\\"potion\\\" or \\\"quit\\\""

97+-- let runGame = Stream.'drainWhile' (== Alive) $ Stream.'mapM' getStatus runEvents

98+-- void $ runStateT runGame 60

99+-- @

100+--

101+-- You can also find the source of this example in the streamly-examples repo

102+-- as <https://github.com/composewell/streamly-examples/tree/master/AcidRain.hs AcidRain.hs>.

103+-- It has been adapted from Gabriel's

104+-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>

105+-- package.

106+-- This is much simpler compared to the pipes version because of the builtin

107+-- concurrency in streamly. You can also find a SDL based reactive programming

108+-- example adapted from Yampa in

109+-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare.hs>.

110+

111+-- $performance

112+--

113+-- Streamly is highly optimized for performance, it is designed for serious

114+-- high performing, concurrent and scalable applications. We have created the

115+-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>

116+-- package which is specifically and carefully designed to measure the

117+-- performance of Haskell streaming libraries fairly and squarely in the right

118+-- way. Streamly performs at par or even better than most streaming libraries

119+-- for serial operations even though it needs to deal with the concurrency

120+-- capability.

121+

122+-- $furtherReading

123+--

124+-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.

docs/Tutorial.hs+42−888

@@ -6,45 +6,28 @@

66-- License : BSD3

77-- Maintainer : streamly@composewell.com

88--

9−-- In this tutorial we will go over the basic concepts and how to use the

10−-- Streamly library. Before you go through this tutorial we recommend that you

11−-- take a look at:

12−--

13−-- * The quick overview of streamly in the <README.md README file>.

14−-- * The overview of streams in "Streamly.Prelude" module.

9+-- In this tutorial we will show how Streamly can be used for idiomatic

10+-- dataflow programming. Before you go through this tutorial we recommend that

11+-- you take a look at the Streamly Getting Started guide so that you are ready

12+-- to try out the examples.

1513

1614module Tutorial

1715 (

1816 -- * Stream Types

1917 -- $streams

2018

21− -- * Concurrent Streams

22− -- $concurrentStreams

23−

24− -- * Combining Streams

25− -- $flavors

26−

2719 -- * Imports and Supporting Code

2820 -- $imports

2921

3022 -- * Generating Streams

3123 -- $generating

3224

33− -- * Generating Streams Concurrently

34− -- $generatingConcurrently

35−

3625 -- * Eliminating Streams

3726 -- $eliminating

3827

39− -- * Concurrent Pipeline Stages

40− -- $concurrentApplication

41−

4228 -- * Transforming Streams

4329 -- $transformation

4430

45− -- * Mapping Concurrently

46− -- $concurrentTransformation

47−

4831 -- * Merging Streams

4932

5033 -- ** Semigroup Style

@@ -56,22 +39,6 @@ module Tutorial

5639 -- *** Wide Serial Composition ('WSerial')

5740 -- $interleaved

5841

59− -- *** Deep Speculative Composition ('Ahead')

60− -- $ahead

61−

62− -- *** Deep Asynchronous Composition ('Async')

63− -- $async

64−

65− -- *** Wide Asynchronous Composition ('WAsync')

66− -- $wasync

67−

68− -- *** Parallel Asynchronous Composition ('Parallel')

69− -- $parallel

70−

71− -- XXX we should deprecate and remove the mkAsync API

72− -- Custom composition

73− -- custom

74−

7542 -- ** Monoid Style

7643 -- $monoid

7744

@@ -87,18 +54,6 @@ module Tutorial

8754 -- *** Wide Serial Nesting ('WSerial')

8855 -- $interleavedNesting

8956

90− -- *** Deep Speculative Nesting ('Ahead')

91− -- $aheadNesting

92−

93− -- *** Deep Asynchronous Nesting ('Async')

94− -- $concurrentNesting

95−

96− -- *** Wide Asynchronous Nesting ('WAsync')

97− -- $wasyncNesting

98−

99− -- *** Parallel Asynchronous Nesting ('Parallel')

100− -- $parallelNesting

101−

10257 -- *** Exercise

10358 -- $monadExercise

10459

@@ -114,30 +69,6 @@ module Tutorial

11469 -- ** Serial Zipping

11570 -- $serialzip

11671

117− -- ** Parallel Zipping

118− -- $parallelzip

119−

120− -- * Monad transformers

121− -- $monadtransformers

122−

123− -- * Concurrent Programming

124− -- $concurrent

125−

126− -- * Reactive Programming

127− -- $reactive

128−

129− -- * Writing Concurrent Programs

130− -- $programs

131−

132− -- * Performance

133− -- $performance

134−

135− -- * Interoperation with Streaming Libraries

136− -- $interop

137−

138− -- * Comparison with Existing Packages

139− -- $comparison

140−

14172 -- * Where to go next?

14273 -- $furtherReading

14374 )

@@ -159,17 +90,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

15990-- >>> import qualified Streamly.Prelude as Stream

16091-- >>> import qualified Streamly.Data.Fold as Fold

16192--

162−-- >>> import Control.Concurrent (threadDelay, myThreadId)

163−-- >>> :{

164−-- delay n = Stream.fromEffect $ do

165−-- threadDelay (n * 1000000)

166−-- tid <- myThreadId

167−-- putStrLn (show tid ++ ": Delay " ++ show n)

168−-- :}

169−--

170−-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

171−-- >>> hSetBuffering stdout LineBuffering

172−--

17393

17494-- $imports

17595--

@@ -182,24 +102,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

182102-- >>> import qualified Streamly.Prelude as Stream

183103-- >>> import qualified Streamly.Data.Fold as Fold

184104--

185−-- To illustrate concurrent vs serial composition aspects, we will use the

186−-- following @delay@ function to introduce a sleep or delay specified in

187−-- seconds. After the delay it prints the number of seconds it slept.

188−--

189−-- >>> import Control.Concurrent (threadDelay, myThreadId)

190−-- >>> :{

191−-- delay n = Stream.fromEffect $ do

192−-- threadDelay (n * 1000000)

193−-- tid <- myThreadId

194−-- putStrLn (show tid ++ ": Delay " ++ show n)

195−-- :}

196−--

197−-- For concurrent examples, use line buffering, otherwise output from different

198−-- threads may get mixed:

199−--

200−-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

201−-- >>> hSetBuffering stdout LineBuffering

202−--

203105

204106-- $streams

205107--

@@ -227,142 +129,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

227129-- on the streams we must use the corresponding functions from

228130-- "Streamly.Prelude" instead of using the base "Prelude".

229131

230−-- $concurrentStreams

231−--

232−-- Many stream operations can be done concurrently:

233−--

234−-- * Streams can be generated concurrently.

235−--

236−-- * Streams can be merged concurrently.

237−--

238−-- * Multiple stages in a streaming pipeline can run concurrently.

239−--

240−-- * Streams can be mapped and zipped concurrently.

241−--

242−-- * In monadic composition they combine like a list transformer,

243−-- providing concurrent non-determinism.

244−--

245−-- There are three basic concurrent stream styles, 'Ahead', 'Async', and

246−-- 'Parallel'. The 'Ahead' style streams are similar to 'Serial' except that

247−-- they can speculatively execute multiple stream actions concurrently in

248−-- advance. 'Ahead' would return exactly the same stream as 'Serial' except

249−-- that it may execute the actions concurrently. The 'Async' style streams,

250−-- like 'Ahead', speculatively execute multiple stream actions in advance but

251−-- return the results in their finishing order rather than in the stream

252−-- traversal order. 'Parallel' is like 'Async' except that it provides

253−-- unbounded parallelism instead of controlled parallelism.

254−--

255−-- For easy reference, we can classify the stream types based on /execution order/,

256−-- /consumption order/, and /bounded or unbounded/ concurrency.

257−-- Execution could be serial (i.e. synchronous) or asynchronous. In serial

258−-- execution we execute the next action in the stream only after the previous

259−-- one has finished executing. In asynchronous execution multiple actions in

260−-- the stream can be executed asynchronously i.e. the next action can start

261−-- executing even before the first one has finished. Consumption order

262−-- determines the order in which the outputs generated by the composition are

263−-- consumed. Consumption could be serial or asynchronous. In serial

264−-- consumption, the outputs are consumed in the traversal order, in

265−-- asynchronous consumption the outputs are consumed as they arrive i.e. first

266−-- come first serve order.

267−--

268−-- +------------+--------------+--------------+--------------+

270−-- +============+==============+==============+==============+

272−-- +------------+--------------+--------------+--------------+

274−-- +------------+--------------+--------------+--------------+

276−-- +------------+--------------+--------------+--------------+

278−-- +------------+--------------+--------------+--------------+

279−--

280−-- All these types can be freely inter-converted using type conversion

281−-- combinators or type annotations, without any cost, to achieve the desired

282−-- composition style. To force a particular type of composition, we coerce the

283−-- stream type using the corresponding type adapting combinator from

284−-- 'fromSerial', 'fromAhead', 'fromAsync', or 'fromParallel'. The default stream type

285−-- is inferred as 'Serial' unless you change it by using one of the combinators

286−-- or by using a type annotation.

287−

288−-- $flavors

289−--

290−-- Streams can be combined using '<>' or 'mappend' to form a

291−-- composite. Composite streams can be interpreted in a depth first or

292−-- breadth first manner using an appropriate type conversion before

293−-- consumption. Deep (e.g. 'Serial') stream type variants traverse a

294−-- composite stream in a depth first manner, such that each stream is

295−-- traversed fully before traversing the next stream. Wide

296−-- (e.g. 'WSerial') stream types traverse it in a breadth first

297−-- manner, such that one element from each stream is traversed before

298−-- coming back to the first stream again.

299−--

300−-- Each stream type has a wide traversal variant prefixed by 'W'. The wide

301−-- variant differs only in the Semigroup\/Monoid, Applicative\/Monad

302−-- compositions of the streams.

303−-- The following table summarizes the basic types and the corresponding wide

304−-- variants:

305−--

306−-- @

307−-- +------------+-----------+

308−-- | Deep | Wide |

309−-- +============+===========+

310−-- | 'Serial' | 'WSerial' |

311−-- +------------+-----------+

312−-- | 'Ahead' | 'WAhead' |

313−-- +------------+-----------+

314−-- | 'Async' | 'WAsync' |

315−-- +------------+-----------+

316−-- @

317−--

318−-- Other than these types there are also 'ZipSerial' and 'ZipAsync' types that

319−-- zip streams serially or concurrently using 'Applicative' operation. These

320−-- types are not monads they are only applicatives and they do not differ in

321−-- 'Semigroup' composition.

322−--

323−

324−-- $programs

325−--

326−-- When writing concurrent programs it is advised to not use the concurrent

327−-- style stream combinators blindly at the top level. That might create too

328−-- much concurrency where it is not even required, and can even degrade

329−-- performance in some cases. In some cases it can also lead to surprising

330−-- behavior because of some code that is supposed to be serial becoming

331−-- concurrent. Please be aware that all concurrency capable APIs that you may

332−-- have used under the scope of a concurrent stream combinator will become

333−-- concurrent. For example if you have a 'repeatM' somewhere in your program

334−-- and you use 'fromParallel' on top, the 'repeatM' becomes fully parallel,

335−-- resulting into an infinite parallel execution . Instead, use the

336−-- /Keep It Serial and Stupid/ principle, start with the default serial

337−-- composition and enable concurrent combinators only when and where necessary.

338−-- When you use a concurrent combinator you can use an explicit 'fromSerial'

339−-- combinator to suppress any unnecessary concurrency under the scope of that

340−-- combinator.

341−

342−-- $monadtransformers

343−--

344−-- To represent streams in an arbitrary monad use the more general monad

345−-- transformer types for example the monad transformer type corresponding to

346−-- the 'Serial' type is 'SerialT'. @SerialT m a@ represents a stream of values

347−-- of type 'a' in some underlying monad 'm'. For example, @SerialT IO Int@ is a

348−-- stream of 'Int' in 'IO' monad. In fact, the type 'Serial' is a synonym for

349−-- @SerialT IO@.

350−--

351−-- Similarly we have monad transformer types for other stream types as well viz.

352−-- 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.

353−--

354−-- To lift a value from an underlying monad in a monad transformer stack into a

355−-- singleton stream use 'lift' and to lift from an IO action use 'liftIO'.

356−--

357−-- >>> import Control.Monad.IO.Class (liftIO)

358−-- >>> Stream.drain $ liftIO $ putStrLn "Hello world!"

359−-- Hello world!

360−--

361−-- >>> import Control.Monad.Trans.Class (MonadTrans(lift))

362−-- >>> Stream.drain $ lift $ putStrLn "Hello world!"

363−-- Hello world!

364−--

365−

366132-- $generating

367133--

368134-- 'nil' represents an empty stream and 'consM' or its operator form '|:' adds

@@ -408,29 +174,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

408174--

409175-- For more ways to construct a stream see the module "Streamly.Prelude".

410176

411−-- $generatingConcurrently

412−--

413−-- Monadic construction and generation functions like 'consM', 'unfoldrM',

414−-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' work concurrently

415−-- when used with appropriate stream type combinator. The pure versions of

416−-- these APIs are not concurrent, however you can use the monadic versions even

417−-- for pure computations by wrapping the pure value in a monad to get the

418−-- concurrent generation capability where required.

419−--

420−-- The following code finishes in 3 seconds (6 seconds when serial):

421−--

422−-- >>> let p n = threadDelay (n * 1000000) >> return n

423−-- >>> Stream.toList $ Stream.fromParallel $ p 3 |: p 2 |: p 1 |: Stream.nil

424−-- [1,2,3]

425−--

426−-- >>> Stream.toList $ Stream.fromAhead $ p 3 |: p 2 |: p 1 |: Stream.nil

427−-- [3,2,1]

428−--

429−-- The following finishes in 10 seconds (100 seconds when serial):

430−--

431−-- >>> Stream.drain $ Stream.fromAsync $ Stream.replicateM 10 $ p 10

432−--

433−

434177-- $eliminating

435178--

436179-- We have already seen 'drain' and toList to eliminate a stream in the

@@ -478,43 +221,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

478221-- :}

479222--

480223

481−-- $concurrentTransformation

482−--

483−-- Monadic transformation functions 'mapM' and 'sequence' work concurrently

484−-- when used with appropriate stream type combinators. The pure versions do not

485−-- work concurrently, however you can use the monadic versions even for pure

486−-- computations to get the concurrent transformation capability where required.

487−--

488−-- This would print a value every second (2 seconds when serial):

489−--

490−-- >>> let p n = threadDelay (n * 1000000) >> return n

491−-- >>> :{

492−-- parMap =

493−-- Stream.repeatM (p 1)

494−-- & Stream.fromSerial -- repeatM is serial

495−-- & Stream.mapM (\x -> p 1 >> print x)

496−-- & Stream.fromAhead -- mapM is cocnurrent using Ahead style

497−-- & Stream.drain

498−-- :}

499−--

500−

501−-- $concurrentApplication

502−--

503−-- The concurrent function application operators '|$' and '|&' apply a stream

504−-- argument to a stream function concurrently to compose a concurrent pipeline

505−-- of stream processing functions:

506−--

507−-- Because both the stages run concurrently, we would see a delay of only 1

508−-- second instead of 2 seconds in the following:

509−--

510−-- >>> let p n = threadDelay (n * 1000000) >> return n

511−-- >>> :{

512−-- parApp =

513−-- Stream.repeatM (p 1)

514−-- |& Stream.mapM (\x -> p 1 >> print x)

515−-- & Stream.drain

516−-- :}

517−

518224-- $semigroup

519225--

520226-- We can combine two streams into a single stream using semigroup composition

@@ -551,13 +257,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

551257-- 4

552258--

553259-- All actions in both the streams are performed serially in the same thread.

554−-- In the following example we can see that all actions are performed in the

555−-- same thread and take a combined total of @3 + 2 + 1 = 6@ seconds:

556−--

557−-- >>> Stream.drain $ delay 3 <> delay 2 <> delay 1

558−-- ThreadId ... Delay 3

559−-- ThreadId ... Delay 2

560−-- ThreadId ... Delay 1

561260--

562261-- The polymorphic version of the binary operation '<>' of the 'Serial' type is

563262-- 'serial'. We can use 'serial' to join streams in a sequential manner

@@ -592,14 +291,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

592291-- 4

593292--

594293-- Even though the monadic actions of the two streams are performed in an

595−-- interleaved manner they are all performed serially in the same thread. In

596−-- the following example we can see that all actions are performed in the same

597−-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:

598−--

599−-- >>> Stream.drain $ Stream.fromWSerial $ delay 3 <> delay 2 <> delay 1

600−-- ThreadId ...: Delay 3

601−-- ThreadId ...: Delay 2

602−-- ThreadId ...: Delay 1

294+-- interleaved manner they are all performed serially in the same thread.

603295--

604296-- The polymorphic version of the 'WSerial' binary operation '<>' is called

605297-- 'wSerial'. We can use 'wSerial' to join streams in an interleaved manner

@@ -615,226 +307,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

615307-- Note that this composition cannot be used to fold infinite number of streams

616308-- since it requires preserving the state until a stream is finished.

617309

618−-- $ahead

619−--

620−-- The 'Semigroup' operation '<>' of the 'Ahead' type combines two streams in a

621−-- /serial depth first/ manner with concurrent lookahead. We use the 'fromAhead'

622−-- type combinator to effect 'Ahead' style of composition. We can also use an

623−-- explicit 'Ahead' type annotation for the stream to achieve the same effect.

624−--

625−-- When two streams are combined in this manner, the streams are traversed in

626−-- depth first manner just like 'Serial', however it can execute the next

627−-- stream concurrently and keep the results ready when its turn arrives.

628−-- Concurrent execution of the next stream(s) is performed if the first stream

629−-- blocks or if it cannot produce output at the rate that is enough to meet the

630−-- consumer demand. Multiple streams can be executed concurrently to meet the

631−-- demand. The following example would print the result in a second even

632−-- though each action in each stream takes one second:

633−--

634−-- >>> p n = threadDelay 1000000 >> return n

635−-- >>> stream1 = p 1 |: p 2 |: Stream.nil

636−-- >>> stream2 = p 3 |: p 4 |: Stream.nil

637−-- >>> Stream.toList $ Stream.fromAhead $ stream1 <> stream2

638−-- [1,2,3,4]

639−--

640−-- Each stream is constructed 'fromAhead' and then both the streams are merged

641−-- 'fromAhead', therefore, all the actions can run concurrently but the result is

642−-- presented in serial order.

643−--

644−-- You can also use the polymorphic combinator 'ahead' in place of '<>' to

645−-- compose any type of streams in this manner.

646−

647−-- $async

648−--

649−-- The 'Semigroup' operation '<>' of the 'Async' type combines the two

650−-- streams in a depth first manner with parallel look ahead. We use the

651−-- 'fromAsync' type combinator to effect 'Async' style of composition. We

652−-- can also use the 'Async' type annotation for the stream type to achieve

653−-- the same effect.

654−--

655−-- When two streams with multiple elements are combined in this manner, the

656−-- streams are traversed in depth first manner just like 'Serial', however it

657−-- can execute the next stream concurrently and return the results from it

658−-- as they arrive i.e. the results from the next stream may be yielded even

659−-- before the results from the first stream. Concurrent execution of the next

660−-- stream(s) is performed if the first stream blocks or if it cannot produce

661−-- output at the rate that is enough to meet the consumer demand. Multiple

662−-- streams can be executed concurrently to meet the demand.

663−-- In the example below each element in the stream introduces a constant delay

664−-- of 1 second, however, it takes just one second to produce all the results.

665−-- The results are not guaranteed to be in any particular order:

666−--

667−-- >>> p n = threadDelay 1000000 >> return n

668−-- >>> stream1 = p 1 |: p 2 |: Stream.nil

669−-- >>> stream2 = p 3 |: p 4 |: Stream.nil

670−-- >>> Stream.toList $ Stream.fromAsync $ stream1 <> stream2

671−-- ...

672−--

673−-- The constituent streams are also composed in 'Async' manner and the

674−-- composition of streams too. We can compose the constituent streams to run

675−-- serially, in that case it would take 2 seconds to produce all the results.

676−-- The elements in the serial streams would be in serial order in the results:

677−--

678−-- >>> p n = threadDelay 1000000 >> return n

679−-- >>> stream = (Stream.fromSerial stream1) <> (Stream.fromSerial stream2)

680−-- >>> Stream.toList $ Stream.fromAsync stream

681−-- ...

682−--

683−-- In the following example we can see that new threads are started when a

684−-- computation blocks. Notice that the output from the stream with the

685−-- shortest delay is printed first. The whole computation takes @maximum of

686−-- (3, 2, 1) = 3@ seconds:

687−--

688−-- >>> Stream.drain $ Stream.fromAsync $ delay 3 <> delay 2 <> delay 1

689−-- ThreadId ...: Delay 1

690−-- ThreadId ...: Delay 2

691−-- ThreadId ...: Delay 3

692−--

693−-- When we have a tree of computations composed using this style, the tree is

694−-- traversed in DFS style just like the 'Serial' style, the only difference is

695−-- that here we can move on to executing the next stream if a stream blocks.

696−-- However, we will not start new threads if we have sufficient output to

697−-- saturate the consumer. This is why we call it left-biased demand driven or

698−-- adaptive concurrency style, the concurrency tends to stay on the left side

699−-- of the composition as long as possible. More threads are started based on

700−-- the pull rate of the consumer. The following example prints an output every

701−-- second as all of the actions are concurrent.

702−--

703−-- >>> Stream.drain $ Stream.fromAsync $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)

704−-- ThreadId ...: Delay 1

705−-- ThreadId ...: Delay 2

706−-- ThreadId ...: Delay 3

707−-- ThreadId ...: Delay 4

708−--

709−-- All the computations may even run in a single thread when more threads are

710−-- not needed. As you can see, in the following example the computations are

711−-- run in a single thread one after another, because none of them blocks.

712−-- However, if the thread consuming the stream were faster than the producer

713−-- then it would have started parallel threads for each computation to keep up

714−-- even if none of them blocks:

715−--

716−-- >>> :{

717−-- traced m = Stream.fromEffect (myThreadId >>= print) >> return m

718−-- stream = traced (sqrt 9) <> traced (sqrt 16) <> traced (sqrt 25)

719−-- main = Stream.drain $ Stream.fromAsync stream

720−-- :}

721−--

722−-- Note that the order of printing in the above examples may change due to

723−-- variations in scheduling latencies for concurrent threads.

724−--

725−-- The polymorphic version of the 'Async' binary operation '<>' is called

726−-- 'async'. We can use 'async' to join streams in a left biased

727−-- adaptively concurrent manner irrespective of the type, notice that we have

728−-- not used the 'fromAsync' combinator in the following example:

729−--

730−-- >>> Stream.drain $ delay 3 `Stream.async` delay 2 `Stream.async` delay 1

731−-- ThreadId ...: Delay 1

732−-- ThreadId ...: Delay 2

733−-- ThreadId ...: Delay 3

734−--

735−-- Since the concurrency provided by this operator is demand driven it cannot

736−-- be used when the composed computations start timers that are relative to

737−-- each other because all computations may not be started at the same time and

738−-- therefore timers in all of them may not start at the same time. When

739−-- relative timing among all computations is important or when we need to start

740−-- all computations at once for any reason 'Parallel' style must be used

741−-- instead.

742−--

743−-- 'Async' style utilizes resources optimally and should be preferred over

744−-- 'Parallel' or 'WAsync' unless you really need those. 'Async' should be used

745−-- when we know that the computations can run in parallel but we do not care if

746−-- they actually run in parallel or not, that decision can be left to the

747−-- scheduler based on demand. Also, note that 'async' operator can be used to fold

748−-- infinite number of streams in contrast to the 'Parallel' or 'WAsync' styles,

749−-- because it does not require us to run all of them at the same time in a fair

750−-- manner.

751−

752−-- $wasync

753−--

754−-- The 'Semigroup' operation '<>' of the 'WAsync' type combines two streams in

755−-- a concurrent manner using /breadth first traversal/. We use the 'fromWAsync'

756−-- type combinator to effect 'WAsync' style of composition. We can also use the

757−-- 'WAsync' type annotation for the stream to achieve the same effect.

758−--

759−-- When streams with multiple elements are combined in this manner, we traverse

760−-- all the streams concurrently in a breadth first manner i.e. one action from

761−-- each stream is performed and yielded to the resulting stream before we come

762−-- back to the first stream again and so on. Even though we execute the actions

763−-- in a breadth first order the outputs are consumed on a first come first

764−-- serve basis.

765−--

766−-- In the following example we can see that outputs are produced in the breadth

767−-- first traversal order but this is not guaranteed.

768−--

769−-- >>> stream1 = print 1 |: print 2 |: Stream.nil

770−-- >>> stream2 = print 3 |: print 4 |: Stream.nil

771−-- >>> Stream.drain $ Stream.fromWAsync $ stream1 <> stream2

772−-- 1

773−-- 3

774−-- 2

775−-- 4

776−--

777−-- The polymorphic version of the binary operation '<>' of the 'WAsync' type is

778−-- 'wAsync'. We can use 'wAsync' to join streams using a breadth first

779−-- concurrent traversal irrespective of the type, notice that we have not used

780−-- the 'fromWAsync' combinator in the following example:

781−--

782−-- >>> Stream.drain $ delay 3 `Stream.wAsync` delay 2 `Stream.wAsync` delay 1

783−-- ThreadId ...: Delay 1

784−-- ThreadId ...: Delay 2

785−-- ThreadId ...: Delay 3

786−--

787−-- Since the concurrency provided by this style is demand driven it may not

788−-- be used when the composed computations start timers that are relative to

789−-- each other because all computations may not be started at the same time and

790−-- therefore timers in all of them may not start at the same time. When

791−-- relative timing among all computations is important or when we need to start

792−-- all computations at once for any reason 'Parallel' style must be used

793−-- instead.

794−--

795−

796−-- $parallel

797−--

798−-- The 'Semigroup' operation '<>' of the 'Parallel' type combines the two

799−-- streams in a fairly concurrent manner with round robin scheduling. We use

800−-- the 'fromParallel' type combinator to effect 'Parallel' style of composition.

801−-- We can also use the 'Parallel' type annotation for the stream type to

802−-- achieve the same effect.

803−--

804−-- When two streams with multiple elements are combined in this manner, the

805−-- monadic actions in both the streams are performed concurrently with a fair

806−-- round robin scheduling. The outputs are yielded in the order in which the

807−-- actions complete. This is pretty similar to the 'WAsync' type, the

808−-- difference is that 'WAsync' is adaptive to the consumer demand and may or

809−-- may not execute all actions in parallel depending on the demand, whereas

810−-- 'Parallel' runs all the streams in parallel irrespective of the demand.

811−--

812−-- The polymorphic version of the binary operation '<>' of the 'Parallel' type

813−-- is 'parallel'. We can use 'parallel' to join streams in a fairly concurrent

814−-- manner irrespective of the type, notice that we have not used the

815−-- 'fromParallel' combinator in the following example:

816−--

817−-- >>> Stream.drain $ delay 3 `Stream.parallel` delay 2 `Stream.wAsync` delay 1

818−-- ThreadId ...: Delay 1

819−-- ThreadId ...: Delay 2

820−-- ThreadId ...: Delay 3

821−--

822−-- Note that this style of composition cannot be used to combine infinite

823−-- number of streams, as it will lead to an infinite sized scheduling queue.

824−--

825−

826−-- XXX to be removed

827−-- $custom

828−--

829−-- The 'mkAsync' API can be used to create references to asynchronously running

830−-- stream computations. We can then use 'uncons' to explore the streams

831−-- arbitrarily and then recompose individual elements to create a new stream.

832−-- This way we can dynamically decide which stream to explore at any given

833−-- time. Take an example of a merge sort of two sorted streams. We need to

834−-- keep consuming items from the stream which has the lowest item in the sort

835−-- order. This can be achieved using async references to streams. See

836−-- "MergeSort.hs" in <https://github.com/composewell/streamly-examples Streamly Examples>.

837−

838310-- $monoid

839311--

840312-- We can use 'Monoid' instances to fold a container of streams in the desired

@@ -848,16 +320,15 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

848320-- * 'concatForFoldableWith' is like @concatMapFoldableWith@ but the container

849321-- argument comes before the function argument.

850322--

851−-- All of the following are equivalent and start ten concurrent tasks each with

852−-- a delay from 1 to 10 seconds, resulting in the printing of each number every

853−-- second:

323+-- All of the following are equivalent:

854324--

855325-- >>> :{

326+-- traced = Stream.fromEffect . print

856327-- main = do

857−-- Stream.drain $ Stream.fromAsync $ foldMap delay [1..10]

858−-- Stream.drain $ Stream.concatFoldableWith Stream.async (map delay [1..10])

859−-- Stream.drain $ Stream.concatMapFoldableWith Stream.async delay [1..10]

860−-- Stream.drain $ Stream.concatForFoldableWith Stream.async [1..10] delay

328+-- Stream.drain $ foldMap traced [1..10]

329+-- Stream.drain $ Stream.concatFoldableWith Stream.serial (map traced [1..10])

330+-- Stream.drain $ Stream.concatMapFoldableWith Stream.serial traced [1..10]

331+-- Stream.drain $ Stream.concatForFoldableWith Stream.serial [1..10] traced

861332-- :}

862333--

863334

@@ -905,11 +376,11 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

905376-- >>> :{

906377-- Stream.drain $ do

907378-- x <- Stream.fromFoldable [3,2,1]

908−-- delay x

379+-- Stream.fromEffect $ print x

909380-- :}

910−-- ThreadId ...: Delay 3

911−-- ThreadId ...: Delay 2

912−-- ThreadId ...: Delay 1

381+-- 3

382+-- 2

383+-- 1

913384--

914385-- As we can see, the code after the @fromFoldable@ statement is run three

915386-- times, once for each value of @x@ drawn from the stream. All the three

@@ -933,7 +404,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

933404-- Stream.drain $ do

934405-- x <- Stream.fromFoldable [1,2]

935406-- y <- Stream.fromFoldable [3,4]

936−-- Stream.fromEffect $ putStrLn $ show (x, y)

407+-- Stream.fromEffect $ print (x, y)

937408-- :}

938409-- (1,3)

939410-- (1,4)

@@ -947,85 +418,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

947418-- Notice that this is analogous to merging streams of type 'Serial' or merging

948419-- streams using 'serial'.

949420

950−-- $aheadNesting

951−--

952−-- The 'Monad' composition of 'Ahead' type behaves just like 'Serial' except

953−-- that it can speculatively perform a bounded number of next iterations of a

954−-- loop concurrently.

955−--

956−-- >>> :{

957−-- Stream.toList $ Stream.fromAhead $ do

958−-- x <- Stream.fromFoldable [3,2,1]

959−-- delay x

960−-- return x

961−-- :}

962−-- ThreadId ...: Delay 1

963−-- ThreadId ...: Delay 2

964−-- ThreadId ...: Delay 3

965−-- [3,2,1]

966−--

967−-- This code finishes in 3 seconds, 'Serial' would take 6 seconds. As we can

968−-- see all the three iterations are concurrent and run in different threads,

969−-- however, the results are returned in the serial order.

970−--

971−-- Concurrency is demand driven, when multiple streams are composed using this

972−-- style, the iterations are executed in a depth first manner just like

973−-- 'Serial' i.e. nested iterations are executed before we proceed to the next

974−-- outer iteration. The only difference is that we may execute multiple future

975−-- iterations concurrently and keep the results ready.

976−--

977−-- The 'fromAhead' type combinator can be used to switch to this style of

978−-- composition. Alternatively, a type annotation can be used to specify the

979−-- type of the stream as 'Ahead'.

980−--

981−

982−-- $concurrentNesting

983−--

984−-- The 'Monad' composition of 'Async' type can perform the iterations of a

985−-- loop concurrently.

986−--

987−-- >>> :{

988−-- Stream.drain $ Stream.fromAsync $ do

989−-- x <- Stream.fromFoldable [3,2,1]

990−-- delay x

991−-- :}

992−-- ThreadId ...: Delay 1

993−-- ThreadId ...: Delay 2

994−-- ThreadId ...: Delay 3

995−--

996−-- As we can see the code after the @fromFoldable@ statement is run three

997−-- times, once for each value of @x@. All the three iterations are concurrent

998−-- and run in different threads. The iteration with least delay finishes first.

999−-- When compared to imperative programming, this can be viewed as a @for@ loop

1000−-- with three concurrent iterations.

1001−--

1002−-- Concurrency is demand driven i.e. more concurrent iterations are started

1003−-- only if the previous iterations are not able to saturate the consumer of the

1004−-- output stream. This works exactly the same way as the merging of two

1005−-- streams using 'async' works.

1006−--

1007−-- The 'fromAsync' type combinator can be used to switch to this style of

1008−-- composition. Alternatively, a type annotation can be used to specify the

1009−-- type of the stream as 'Async'.

1010−--

1011−-- When multiple streams are nested using this style, the iterations are

1012−-- concurrently evaluated in a depth first manner:

1013−--

1014−--

1015−-- >>> :{

1016−-- Stream.drain $ Stream.fromAsync $ do

1017−-- x <- Stream.fromFoldable [1,2]

1018−-- y <- Stream.fromFoldable [3,4]

1019−-- Stream.fromEffect $ putStrLn $ show (x, y)

1020−-- :}

1021−-- (1,3)

1022−-- (1,4)

1023−-- (2,3)

1024−-- (2,4)

1025−--

1026−-- Nested iterations are given preference for concurrent evaluation i.e.

1027−-- (1,4) will be scheduled in preference to (2,3).

1028−

1029421-- $interleavedNesting

1030422--

1031423-- The 'Monad' composition of 'WSerial' type interleaves the iterations of

@@ -1035,7 +427,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

1035427-- Stream.drain $ Stream.fromWSerial $ do

1036428-- x <- Stream.fromFoldable [1,2]

1037429-- y <- Stream.fromFoldable [3,4]

1038−-- Stream.fromEffect $ putStrLn $ show (x, y)

430+-- Stream.fromEffect $ print (x, y)

1039431-- :}

1040432-- (1,3)

1041433-- (2,3)

@@ -1050,57 +442,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

1050442-- type of the stream as 'WSerial'.

1051443--

1052444

1053−-- $wasyncNesting

1054−--

1055−-- Like 'Async', the 'Monad' composition of 'WAsync' runs the iterations of a

1056−-- loop concurrently. It differs from 'Async' in the nested loop behavior. Like

1057−-- 'WSerial', the nested loops in this type are traversed and executed in a

1058−-- breadth first manner rather than the depth first manner of 'Async' style.

1059−--

1060−-- >>> :{

1061−-- Stream.drain $ Stream.fromWAsync $ do

1062−-- x <- Stream.fromSerial $ Stream.fromFoldable [1,2]

1063−-- y <- Stream.fromSerial $ Stream.fromFoldable [3,4]

1064−-- Stream.fromEffect $ putStrLn $ show (x, y)

1065−-- :}

1066−-- (1,3)

1067−-- (1,4)

1068−-- (2,3)

1069−-- (2,4)

1070−--

1071−-- Note that (2,3) is preferred to (1,4) when evaluating the iterations

1072−-- concurrently. This works exactly the same way as the merging of two streams

1073−-- using 'wAsync' works.

1074−--

1075−-- The 'fromWAsync' type combinator can be used to switch to this style of

1076−-- composition. Alternatively, a type annotation can be used to specify the

1077−-- type of the stream as 'WAsync'.

1078−--

1079−

1080−-- $parallelNesting

1081−--

1082−-- Just like 'Async' or 'WAsync' the 'Monad' composition of 'Parallel' runs the

1083−-- iterations of a loop concurrently.

1084−--

1085−-- >>> :{

1086−-- Stream.drain $ Stream.fromParallel $ do

1087−-- x <- Stream.fromFoldable [3,2,1]

1088−-- delay x

1089−-- :}

1090−-- ThreadId ...: Delay 1

1091−-- ThreadId ...: Delay 2

1092−-- ThreadId ...: Delay 3

1093−--

1094−-- It differs from 'Async' and 'WAsync' in the nested loop behavior. All

1095−-- iterations of the loop are run fully concurrently irrespective of the

1096−-- demand. This works exactly the same way as the merging of streams using

1097−-- 'parallel' works.

1098−--

1099−-- The 'fromParallel' type combinator can be used to switch to this style of

1100−-- composition. Alternatively, a type annotation can be used to specify the

1101−-- type of the stream as 'Parallel'.

1102−--

1103−

1104445-- $monadExercise

1105446--

1106447-- Streamly code is usually written in a way that is agnostic of the

@@ -1114,7 +455,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

1114455-- sz <- sizes

1115456-- cl <- colors

1116457-- sh <- shapes

1117−-- Stream.fromEffect $ putStrLn $ show (sz, cl, sh)

458+-- Stream.fromEffect $ print (sz, cl, sh)

1118459-- where

1119460-- sizes = Stream.fromFoldable [1, 2, 3]

1120461-- colors = Stream.fromFoldable ["red", "green", "blue"]

@@ -1153,65 +494,32 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

1153494-- provided.

1154495--

1155496-- The following is an example of 'Serial' applicative, it runs all iterations

1156−-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):

497+-- serially:

1157498--

1158−-- >>> d n = delay n >> return n

1159−-- >>> s1 = d 1 <> d 2

1160−-- >>> s2 = d 3 <> d 4

499+-- >>> p n = Stream.fromEffect (print n) >> return n

500+-- >>> s1 = p 1 <> p 2

501+-- >>> s2 = p 3 <> p 4

1161502--

1162503-- >>> (Stream.toList $ Stream.fromSerial $ (,) <$> s1 <*> s2) >>= print

1163−-- ThreadId ...: Delay 1

1164−-- ThreadId ...: Delay 3

1165−-- ThreadId ...: Delay 4

1166−-- ThreadId ...: Delay 2

1167−-- ThreadId ...: Delay 3

1168−-- ThreadId ...: Delay 4

504+-- 1

505+-- 3

506+-- 4

507+-- 2

508+-- 3

509+-- 4

1169510-- [(1,3),(1,4),(2,3),(2,4)]

1170511--

1171512-- Similarly, 'WSerial' applicative runs the iterations in an interleaved

1172513-- order but being serial it too takes a total of 17 seconds:

1173514--

1174515-- >>> (Stream.toList $ Stream.fromWSerial $ (,) <$> s1 <*> s2) >>= print

1175−-- ThreadId ...: Delay 1

1176−-- ThreadId ...: Delay 3

1177−-- ThreadId ...: Delay 2

1178−-- ThreadId ...: Delay 3

1179−-- ThreadId ...: Delay 4

1180−-- ThreadId ...: Delay 4

1181−-- [(1,3),(2,3),(1,4),(2,4)]

1182−--

1183−-- 'Async' can run the iterations concurrently, therefore, it takes a total

1184−-- of 6 seconds which is max (1, 2) + max (3, 4):

1185−--

1186−-- >>> (Stream.toList $ Stream.fromAsync $ (,) <$> s1 <*> s2) >>= print

1187−-- ...

1188−--

1189−-- @

1190−-- ThreadId 34: Delay 1

1191−-- ThreadId 36: Delay 2

1192−-- ThreadId 35: Delay 3

1193−-- ThreadId 36: Delay 3

1194−-- ThreadId 35: Delay 4

1195−-- ThreadId 36: Delay 4

1196−-- [(1,3),(2,3),(1,4),(2,4)]

1197−-- @

1198−--

1199−-- Similarly, 'WAsync' as well can run the iterations concurrently, but with a

1200−-- different style of scheduling than 'Async' as explained in the Monad

1201−-- section, therefore, it too takes a total of 6 seconds (2 + 4):

1202−--

1203−-- >>> (Stream.toList $ Stream.fromWAsync $ (,) <$> s1 <*> s2) >>= print

1204−-- ...

1205−--

1206−-- @

1207−-- ThreadId 34: Delay 1

1208−-- ThreadId 36: Delay 2

1209−-- ThreadId 35: Delay 3

1210−-- ThreadId 36: Delay 3

1211−-- ThreadId 35: Delay 4

1212−-- ThreadId 36: Delay 4

516+-- 1

517+-- 3

518+-- 2

519+-- 3

520+-- 4

521+-- 4

1213522-- [(1,3),(2,3),(1,4),(2,4)]

1214−-- @

1215523

1216524-- $zipping

1217525--

@@ -1228,172 +536,18 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

1228536-- 'fromZipSerial' type combinator can be used to switch to serial applicative

1229537-- zip composition:

1230538--

1231−-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since

1232−-- everything runs serially:

1233−--

1234−-- >>> d n = delay n >> return n

1235−-- >>> s1 = Stream.fromSerial $ d 1 <> d 2

1236−-- >>> s2 = Stream.fromSerial $ d 3 <> d 4

539+-- >>> p n = Stream.fromEffect (print n) >> return n

540+-- >>> s1 = Stream.fromSerial $ p 1 <> p 2

541+-- >>> s2 = Stream.fromSerial $ p 3 <> p 4

1237542-- >>> (Stream.toList $ Stream.fromZipSerial $ (,) <$> s1 <*> s2) >>= print

1238−-- ThreadId ...: Delay 1

1239−-- ThreadId ...: Delay 3

1240−-- ThreadId ...: Delay 2

1241−-- ThreadId ...: Delay 4

1242−-- [(1,3),(2,4)]

1243−--

1244−

1245−-- $parallelzip

1246−--

1247−-- The applicative instance of 'ZipAsync' type zips streams concurrently.

1248−-- 'fromZipAsync' type combinator can be used to switch to parallel applicative

1249−-- zip composition:

1250−--

1251−-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3

1252−-- are produced concurrently, and 2 and 4 are produced concurrently:

1253−--

1254−-- >>> d n = delay n >> return n

1255−-- >>> s1 = Stream.fromSerial $ d 1 <> d 2

1256−-- >>> s2 = Stream.fromSerial $ d 3 <> d 4

1257−-- >>> (Stream.toList $ Stream.fromZipAsync $ (,) <$> s1 <*> s2) >>= print

1258−-- ThreadId ...: Delay 1

1259−-- ThreadId ...: Delay 2

1260−-- ThreadId ...: Delay 3

1261−-- ThreadId ...: Delay 4

543+-- 1

544+-- 3

545+-- 2

546+-- 4

1262547-- [(1,3),(2,4)]

1263548--

1264549

1265−-- $concurrent

1266−--

1267−-- When writing concurrent programs there are two distinct places where the

1268−-- programmer can control the concurrency. First, when /composing/ a stream by

1269−-- merging multiple streams we can choose an appropriate sum style operators to

1270−-- combine them concurrently or serially. Second, when /processing/ a stream in

1271−-- a monadic composition we can choose one of the monad composition types to

1272−-- choose the desired type of concurrency.

1273−--

1274−-- In the following example the squares of @x@ and @y@ are computed

1275−-- concurrently using the 'async' operation and the square roots of their

1276−-- sum are computed serially because of the 'fromSerial' combinator. We can

1277−-- choose different combinators for the monadic processing and the stream

1278−-- generation, to control the concurrency. We can also use the 'fromAsync'

1279−-- combinator instead of explicitly folding with 'async'.

1280−--

1281−-- >>> import Data.List (sum)

1282−-- >>> :{

1283−-- main = do

1284−-- z <- Stream.toList

1285−-- $ Stream.fromSerial -- Serial monadic processing (sqrt below)

1286−-- $ do

1287−-- x2 <- Stream.concatForFoldableWith Stream.async [1..100] $ -- Concurrent @"for"@ loop

1288−-- \x -> return $ x * x -- body of the loop

1289−-- y2 <- Stream.concatForFoldableWith Stream.async [1..100] $

1290−-- \y -> return $ y * y

1291−-- return $ sqrt (x2 + y2)

1292−-- print $ sum z

1293−-- :}

1294−--

1295−-- We can see how this directly maps to the imperative style

1296−-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> model, we use combinators

1297−-- and operators instead of the ugly pragmas.

1298−--

1299−-- For more concurrent programming examples see,

1300−-- <https://github.com/composewell/streamly-examples>.

1301−

1302−-- $reactive

1303−--

1304−-- Reactive programming is nothing but concurrent streaming which is what

1305−-- streamly is all about. With streamly we can generate streams of events,

1306−-- merge streams that are generated concurrently and process events

1307−-- concurrently. We can do all this without any knowledge about the specifics

1308−-- of the implementation of concurrency. In the following example you will see

1309−-- that the code is just regular Haskell code without much streamly APIs used

1310−-- (active hyperlinks are the streamly APIs) and yet it is a reactive

1311−-- application.

1312−--

1313−-- This application has two independent and concurrent sources of event

1314−-- streams, @acidRain@ and @userAction@. @acidRain@ continuously generates

1315−-- events that deteriorate the health of the character in the game.

1316−-- @userAction@ can be "potion" or "quit". When the user types "potion" the

1317−-- health improves and the game continues.

1318−--

1319−-- @

1320−-- {-\# LANGUAGE FlexibleContexts \#-}

1321−--

1322−-- import "Streamly.Prelude" (MonadAsync, SerialT)

1323−-- import "Streamly.Prelude" as Stream

1324−-- import Control.Monad (void)

1325−-- import Control.Monad.IO.Class (MonadIO(liftIO))

1326−-- import Control.Monad.State (MonadState, get, modify, runStateT)

1327−--

1328−-- data Event = Quit | Harm Int | Heal Int deriving (Show)

1329−--

1330−-- userAction :: MonadAsync m => 'SerialT' m Event

1331−-- userAction = Stream.'repeatM' $ liftIO askUser

1332−-- where

1333−-- askUser = do

1334−-- command <- getLine

1335−-- case command of

1336−-- "potion" -> return (Heal 10)

1337−-- "harm" -> return (Harm 10)

1338−-- "quit" -> return Quit

1339−-- _ -> putStrLn "Type potion or harm or quit" >> askUser

1340−--

1341−-- acidRain :: MonadAsync m => 'SerialT' m Event

1342−-- acidRain = Stream.'fromAsync' $ Stream.'constRate' 1 $ Stream.'repeatM' $ liftIO $ return $ Harm 1

1343−--

1344−-- data Result = Check | Done

1345−--

1346−-- runEvents :: (MonadAsync m, MonadState Int m) => 'SerialT' m Result

1347−-- runEvents = do

1348−-- event \<- userAction \`Stream.'parallel'` acidRain

1349−-- case event of

1350−-- Harm n -> modify (\\h -> h - n) >> return Check

1351−-- Heal n -> modify (\\h -> h + n) >> return Check

1352−-- Quit -> return Done

1353−--

1354−-- data Status = Alive | GameOver deriving Eq

1355−--

1356−-- getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status

1357−-- getStatus result =

1358−-- case result of

1359−-- Done -> liftIO $ putStrLn "You quit!" >> return GameOver

1360−-- Check -> do

1361−-- h <- get

1362−-- liftIO $ if (h <= 0)

1363−-- then putStrLn "You die!" >> return GameOver

1364−-- else putStrLn ("Health = " <> show h) >> return Alive

1365−--

1366−-- main :: IO ()

1367−-- main = do

1368−-- putStrLn "Your health is deteriorating due to acid rain, type \\\"potion\\\" or \\\"quit\\\""

1369−-- let runGame = Stream.'drainWhile' (== Alive) $ Stream.'mapM' getStatus runEvents

1370−-- void $ runStateT runGame 60

1371−-- @

1372−--

1373−-- You can also find the source of this example in the streamly-examples repo

1374−-- as <https://github.com/composewell/streamly-examples/tree/master/AcidRain.hs AcidRain.hs>.

1375−-- It has been adapted from Gabriel's

1376−-- <https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html pipes-concurrency>

1377−-- package.

1378−-- This is much simpler compared to the pipes version because of the builtin

1379−-- concurrency in streamly. You can also find a SDL based reactive programming

1380−-- example adapted from Yampa in

1381−-- <https://github.com/composewell/streamly-examples/tree/master/CirclingSquare.hs CirclingSquare.hs>.

1382−

1383−-- $performance

1384−--

1385−-- Streamly is highly optimized for performance, it is designed for serious

1386−-- high performing, concurrent and scalable applications. We have created the

1387−-- <https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks>

1388−-- package which is specifically and carefully designed to measure the

1389−-- performance of Haskell streaming libraries fairly and squarely in the right

1390−-- way. Streamly performs at par or even better than most streaming libraries

1391−-- for serial operations even though it needs to deal with the concurrency

1392−-- capability.

1393−

1394550-- $furtherReading

1395551--

1396−-- * Read the documentation of "Streamly" module

1397−-- * Read the documentation of "Streamly.Prelude" module

552+-- * Read the concurrent streams tutorial

1398553-- * See the examples in <https://github.com/composewell/streamly-examples streamly-examples> repo.

1399−-- * See the tests in the "test" directory of the package

docs/performance.mdnew file+7

@@ -0,0 +1,7 @@

1+# Performance

3+Streamly is highly optimized for performance, it is designed for

4+high performing, concurrent and scalable applications. Please see the

5+[streaming-benchmarks](https://hackage.haskell.org/package/streaming-benchmarks streaming-benchmarks)

6+package for microbencmarks measuring the performance of streamly and comparing

7+it with other streaming libraries.

docs/streamly-docs.cabal+2

@@ -26,6 +26,8 @@ library

2626 hs-source-dirs: .

2727 exposed-modules:

2828 Tutorial

29+ ConcurrentStreams

30+ ReactiveProgramming

2931

3032 build-depends:

3133 base >= 4.9 && < 5

docs/transformers.md+34−1

@@ -1,3 +1,28 @@

1+# Monad Transformers

3+In the stream tutorials we mostly used streams in the IO monad. In

4+general, the type `SerialT` is a monad transformer, @SerialT m a@

5+represents a stream of values of type 'a' in some underlying monad

6+'m'. For example, @SerialT IO Int@ is a stream of 'Int' in 'IO'

7+monad. Similarly, `SerialT Identity Int` would be a pure stream

8+equivalent to `[a]`.

10+Similarly we have monad transformer types for other stream types as well

11+viz. 'WSerialT', 'AsyncT', 'WAsyncT' and 'ParallelT'.

12+

13+To lift a value from an underlying monad in a monad transformer stack into a

14+singleton stream use 'lift' and to lift from an IO action use 'liftIO'.

15+

16+```

17+>>> import Control.Monad.IO.Class (liftIO)

18+>>> Stream.drain $ liftIO $ putStrLn "Hello world!"

19+Hello world!

20+

21+>>> import Control.Monad.Trans.Class (MonadTrans(lift))

22+>>> Stream.drain $ lift $ putStrLn "Hello world!"

23+Hello world!

24+```

25+

126## Using Monad Transformers

227

328Common monad transformers can be used with streamly serial streams, without any

@@ -5,12 +30,16 @@ issues. `ReaderT` can be used with concurrent streams as well without any

530issues.

631

732The semantics of monads other than `ReaderT` with concurrent streams are

8−not yet finalized and will change in future, therefore as of now they are not

33+not yet finalized and will change in future, therefore, as of now they are not

934recommended to be used with concurrent streams.

1035

1136## Ordering of Monad Transformers

1237

1338In most cases it is a good idea to keep streamly as the top level monad.

39+[This

40+example](https://github.com/composewell/streamly-examples/blob/master/examples/ControlFlow.hs)

41+demonstrates how various control flow modifying monads can be combined

42+with streamly stream monads.

1443

1544## State Sharing

1645### Serial Applications

@@ -30,3 +59,7 @@ monad or just share the `IORef` by passing it to the required functions. The

3059`IORef` can be updated atomically using `atomicModifyIORef`.

3160

3261The `CirclingSquare.hs` example shares an `IORef` across parallel tasks.

62+

63+## See also

64+

65+* [Examples of control flow monads with Streamly](https://github.com/composewell/streamly-examples/blob/master/examples/ControlFlow.hs)

streamly.cabal+1−3

@@ -103,9 +103,7 @@ extra-source-files:

103103 design/*.png

104104 design/*.rst

105105 docs/*.md

106− docs/Setup.hs

107− docs/Tutorial.hs

108− docs/doctests.hs

106+ docs/*.hs

109107 docs/streamly-docs.cabal

110108 examples/README.md

111109 src/Streamly/Internal/Data/Stream/Instances.hs