1a89f0c · streamly · git.iko.soy

git.iko.soy/streamly

Review and update the Stream.Concurrent docs/API

commit1a89f0c9eb818f3cefad6220f7198e9fdcf83138 authorHarendra Kumar <harendra@composewell.com> dateSun 13 Nov 2022, 19:51 (4 years ago) parent7475d44

3 files changed+189−54

compare with master

unifiedsplit

src/Streamly/Data/Stream/Concurrent.hs+95−17 src/Streamly/Internal/Data/Stream/Channel/Types.hs+6−6 src/Streamly/Internal/Data/Stream/Concurrent.hs+88−31

src/Streamly/Data/Stream/Concurrent.hs+95−17

@@ -5,6 +5,81 @@

55-- Maintainer : streamly@composewell.com

66-- Stability : released

77-- Portability : GHC

8+--

9+-- This module provides concurrent streaming abstractions.

10+--

11+-- == Concurrency Channels

12+--

13+-- At a lower level, concurrency is implemented using channels that support

14+-- concurrent evaluation of streams. We create a channel, and add one or more

15+-- streams to it. The channel evaluates multiple streams concurrently and then

16+-- generates a single output stream from the results.

17+--

18+-- == Concurrency Primitives

19+--

20+-- There are only two fundamental abstractions for concurrency, all other

21+-- combinators in this module can be expressed in terms of those two.

22+--

23+-- The first one is 'evalWith', it evaluates a single stream asynchronously, a

24+-- worker thread runs the stream and buffers the results, and the consumer of

25+-- the stream runs in another thread consuming it from the buffer, thus

26+-- decoupling the production and consumption of the stream. This can be used to

27+-- run different stages of a pipeline concurrently.

28+--

29+-- The second one is 'concatMapWith', unlike 'evalWith' this is used to

30+-- evaluate multiple streams concurrently and combine the results. A stream

31+-- generator function is mapped to the input stream and all the generated

32+-- streams are then evaluated concurrently, and the results are combined.

33+--

34+-- == Concurrency Configuration

35+--

36+-- Combinators ending with the "With" suffix ('concatMapWith') take a 'Config'

37+-- argument which controls the concurrent behavior. For example, maximum number

38+-- of threads to be used ('maxThreads') or the maxmimum size of the buffer

39+-- ('maxBuffer'), or how the streams are scheduled with respect to each other

40+-- ('interleaved'), or how the results are consumed ('ordered').

41+--

42+-- Configuration is specified as @Config -> Config@ modifier functions that can

43+-- be composed together using function composition. For example, to specify the

44+-- maximum threads we can use @concatMapWith (maxThreads 10)@ if we also want

45+-- to specify the maximum buffer we can compose the two options

46+-- @concatMapwith (maxThreads 10 . maxBuffer 100)@. To use default configuration

47+-- use 'id' as the config modifier e.g. @concatMapWith id@.

48+--

49+-- See the "Configuration" section and individual configuration options'

50+-- documentation for the default behavior and values.

51+--

52+-- == Scheduling behavior

53+--

54+-- The most important option is to control whether the output of the concurrent

55+-- execution is consumed in the same order as the corresponding actions in the

56+-- input stream or as soon as they arrive. The default is the latter, however,

57+-- we can enforce the original order by using the 'ordered' option.

58+--

59+-- Another important option controls whether the number of worker threads are

60+-- automatically increased or decreased based on the consumption rate or

61+-- threads are started as aggresively as possible until the maxThreads or

62+-- maxBuffer limits are hit. The default is the former. However, the 'eager'

63+-- option can be enabled to use the latter behavior. When 'eager' is on,

64+-- consumer thread does not make any impact on scheduling of the available

65+-- tasks.

66+--

67+-- == Concurrent Combinators

68+--

69+-- Using the two fundamental concurrency primitives we can implement all the

70+-- usual streaming combinators with concurrent behavior. Combinators like

71+-- 'unfoldrM', 'iterateM' that are inherently serial can be evaluated

72+-- concurrently with respect to the consumer pipeline using 'eval'. Combinators

73+-- like 'zipWithM', 'mergeByM' can also use 'evalWith' on the input streams to

74+-- make them concurrent.

75+--

76+-- Combinators like 'repeatM', 'replicateM', 'fromListM', 'sequence', 'mapM' in

77+-- which all actions are independent of each other can be made concurrent using

78+-- the concurrent 'concatMapWith' operation.

79+--

80+-- A concurrent 'repeatM' would repeat an action using multiple concurrent

81+-- executions of the action. Similarly, a concurrent 'mapM' would perform the

82+-- mapped action in independent threads.

883

984module Streamly.Data.Stream.Concurrent

1085 (

@@ -42,44 +117,47 @@ module Streamly.Data.Stream.Concurrent

42117

43118 -- ** Evaluate

44119 -- | Evaluates a stream concurrently using a channel.

45− , eval

120+

121+ -- , eval

46122 , evalWith

47123

48124 -- ** Generate

49125 -- | Uses a single channel to evaluate mapped actions concurrently.

50− , repeatM

51− , replicateM

126+ , repeatMWith

127+ , replicateMWith

52128

53129 -- ** Map

54130 -- | Uses a single channel to evaluate mapped actions concurrently.

55− , mapM

131+

132+ -- , mapM

56133 , mapMWith

57− , sequence

134+ -- , sequence

58135 , sequenceWith

59136

60137 -- ** List of streams

61138 -- | Shares a single channel across many streams.

62− , append

63− , ahead

64− , parallel

139+

140+ -- , append

141+ -- , ahead

142+ -- , parallel

65143 , concatListWith

66− , parallelFst

67− , parallelMin

68− , zipWithM

69− , zipWith

144+ -- , parallelFst

145+ -- , parallelMin

146+ -- , zipWithM

147+ -- , zipWith

70148

71149 -- ** Stream of streams

72150 -- *** Apply

73− -- | Apply argument streams to a function concurrently. Uses a separate

74− -- channel for each application.

75− , apply

151+

152+ -- , apply

76153 , applyWith

77154

78155 -- *** Concat

79156 -- | Shares a single channel across many streams.

80− , concat

157+

158+ -- , concat

81159 , concatWith

82− , concatMap

160+ -- , concatMap

83161 , concatMapWith

84162 )

85163where

src/Streamly/Internal/Data/Stream/Channel/Types.hs+6−6

@@ -306,11 +306,9 @@ data Rate = Rate

306306 , rateBuffer :: Int -- ^ Maximum slack from the goal

307307 }

308308

309−-- | Specify when the 'Channel' should stop. Note that for ordered streams only

310−-- 'FirstStops' and 'AnyStop' would be the same, and usually only 'AllStop'

311−-- would make sense. This option is most useful for eager unordered streams.

309+-- | Specify when the 'Channel' should stop.

312310data StopWhen =

313− FirstStops -- ^ Stop when the first stream/action ends.

311+ FirstStops -- ^ Stop when the first stream ends.

314312 | AllStop -- ^ Stop when all the streams end.

315313 | AnyStops -- ^ Stop when any one stream ends.

316314

@@ -321,8 +319,8 @@ data StopWhen =

321319-- XXX make all these Limited types and use phantom types to distinguish them

322320

323321-- | An abstract type for specifying the configuration parameters of a

324−-- 'Channel'. Use @Config -> Config@ modifier functions to modify the

325−-- default configuration. See the modifiers for default values.

322+-- 'Channel'. Use @Config -> Config@ modifier functions to modify the default

323+-- configuration. See the individual modifier documentation for default values.

326324--

327325data Config = Config

328326 { -- one shot configuration, automatically reset for each API call

@@ -503,6 +501,7 @@ getInspectMode = _inspect

503501-- 'maxBuffer' is reached.

504502--

505503-- /Note:/ This option has no effect when rate has been specified.

504+--

506505-- /Note:/ Not supported with 'interleaved'.

507506--

508507eager :: Bool -> Config -> Config

@@ -534,6 +533,7 @@ getOrdered = _ordered

534533-- threads.

535534--

536535-- /Note:/ Can only be used on finite number of streams.

536+--

537537-- /Note:/ Not supported with 'ordered'.

538538--

539539interleaved :: Bool -> Config -> Config

src/Streamly/Internal/Data/Stream/Concurrent.hs+88−31

@@ -54,7 +54,9 @@ module Streamly.Internal.Data.Stream.Concurrent

5454 -- ** Generate

5555 -- | Uses a single channel to evaluate all actions.

5656 , repeatM

57+ , repeatMWith

5758 , replicateM

59+ , replicateMWith

5860

5961 -- ** Map

6062 -- | Uses a single channel to evaluate all actions.

@@ -88,7 +90,6 @@ module Streamly.Internal.Data.Stream.Concurrent

8890

8991 -- ** Stream of streams

9092 -- *** Apply

91− -- | Uses a separate channel for each application.

9293 , apply

9394 , applyWith

9495

@@ -176,18 +177,15 @@ evalWithD modifier m = D.Stream step Nothing

176177 D.Stop -> D.Stop

177178-}

178179

179−-- | Like 'eval' but can specify a config modifier to change the concurrent

180−-- channel parameters.

181−--

182−-- > evalWith = withChannel (const id)

180+-- | Evaluate a stream asynchronously using a channel and serve the consumer

181+-- from the evaluation buffer.

183182--

184183{-# INLINE evalWith #-}

185184evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a

186185evalWith modifier input = withChannel modifier input (const id)

187186 -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream

188187

189−-- | Evaluate a stream asynchronously using a channel and serve the consumer

190−-- from the evaluation buffer.

188+-- | Definition:

191189--

192190-- >>> eval = Concur.evalWith id

193191--

@@ -426,25 +424,16 @@ concatMapWithK modifier f input =

426424 let g = concatMapWithChanKGeneric modifier

427425 in withChannelK modifier input (`g` f)

428426

429−-- concatMapWith modifier f stream = concatWith modifier $ fmap f stream

430−

431−-- | Like 'concatMap' but we can also specify the concurrent channel's

432−-- configuration parameters using Config modifiers.

433−--

434−{-# INLINE concatMapWith #-}

435−concatMapWith :: MonadAsync m =>

436− (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b

437−concatMapWith modifier f stream =

438− Stream.fromStreamK

439− $ concatMapWithK

440− modifier (Stream.toStreamK . f) (Stream.toStreamK stream)

441−

442427-- | Map each element of the input to a stream and then concurrently evaluate

443428-- and concatenate the resulting streams. Multiple streams may be evaluated

444429-- concurrently but earlier streams are perferred. Output from the streams are

445430-- used as they arrive.

446431--

447−-- >>> concatMap = Concur.concatMapWith id

432+-- Definition:

433+--

434+-- >>> concatMapWith modifier f stream = Concur.concatWith modifier $ fmap f stream

435+--

436+-- Examples:

448437--

449438-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith cfg id $ Stream.fromList xs

450439--

@@ -477,6 +466,18 @@ concatMapWith modifier f stream =

477466-- >>> f (Concur.maxThreads 1) [stream1, stream2]

478467-- [1,2,3,4,5,6]

479468--

469+{-# INLINE concatMapWith #-}

470+concatMapWith :: MonadAsync m =>

471+ (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b

472+concatMapWith modifier f stream =

473+ Stream.fromStreamK

474+ $ concatMapWithK

475+ modifier (Stream.toStreamK . f) (Stream.toStreamK stream)

476+

477+-- | Definition:

478+--

479+-- >>> concatMap = Concur.concatMapWith id

480+--

480481{-# INLINE concatMap #-}

481482concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b

482483concatMap = concatMapWith id

@@ -503,8 +504,7 @@ concatMap = concatMapWith id

503504concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b

504505concatMapInterleave = concatMapWith (interleaved True)

505506

506−-- | Like 'concat' but we can also specify the concurrent channel's

507−-- configuration parameters using Config modifiers.

507+-- | Evaluate the streams in the input stream concurrently and combine them.

508508--

509509-- >>> concatWith modifier = Concur.concatMapWith modifier id

510510--

@@ -513,7 +513,7 @@ concatWith :: MonadAsync m =>

513513 (Config -> Config) -> Stream m (Stream m a) -> Stream m a

514514concatWith modifier = concatMapWith modifier id

515515

516−-- | Evaluate the streams in the input stream concurrently and combine them.

516+-- | Definition:

517517--

518518-- >>> concat = Concur.concatWith id

519519--

@@ -578,6 +578,8 @@ parallelFst = concatListWith (eager True . stopWhen FirstStops)

578578-- | Like 'parallel' but stops the output as soon as any of the two streams

579579-- stops.

580580--

581+-- Definition:

582+--

581583-- >>> parallelMin = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)

582584--

583585{-# INLINE parallelMin #-}

@@ -588,6 +590,8 @@ parallelMin = concatListWith (eager True . stopWhen AnyStops)

588590-- Applicative

589591-------------------------------------------------------------------------------

590592

593+-- | Apply an argument stream to a function stream concurrently. Uses a

594+-- shared channel for all individual applications within a stream application.

591595{-# INLINE applyWith #-}

592596{-# SPECIALIZE applyWith ::

593597 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}

@@ -609,15 +613,41 @@ apply = applyWith id

609613-------------------------------------------------------------------------------

610614

611615-- |

616+-- Definition:

617+--

612618-- >>> mapMWith modifier f = Concur.concatMapWith modifier (Stream.fromEffect . f)

613619--

620+-- Example, the following example finishes in 1 second as all actions run in

621+-- parallel. Even though results are available out of order they are ordered

622+-- due to the config option::

623+--

624+-- >>> f x = delay x >> return x

625+-- >>> Stream.fold Fold.toList $ Concur.mapMWith (Concur.ordered True) f $ Stream.fromList [3,2,1]

626+-- 1 sec

627+-- 2 sec

628+-- 3 sec

629+-- [3,2,1]

630+--

614631{-# INLINE mapMWith #-}

615632mapMWith :: MonadAsync m =>

616633 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b

617634mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)

618635

619636-- |

637+-- Definition:

638+--

620639-- >>> mapM = Concur.mapMWith id

640+--

641+-- Example, the following example finishes in 1 second as all actions run in

642+-- parallel, by default, the results are output as soon as they are available::

643+--

644+-- >>> f x = delay x >> return x

645+-- >>> Stream.fold Fold.toList $ Concur.mapM f $ Stream.fromList [3,2,1]

646+-- 1 sec

647+-- 2 sec

648+-- 3 sec

649+-- [1,2,3]

650+--

621651{-# INLINE mapM #-}

622652mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b

623653mapM = mapMWith id

@@ -664,21 +694,48 @@ zipWith f = zipWithM (\a b -> return $ f a b)

664694-------------------------------------------------------------------------------

665695

666696-- |

667−-- >>> repeatM = Concur.sequence . Stream.repeat

697+-- Definition:

698+--

699+-- >>> repeatMWith cfg = Concur.sequenceWith cfg . Stream.repeat

700+--

701+-- Generate a stream by repeatedly executing a monadic action forever.

702+{-# INLINE repeatMWith #-}

703+repeatMWith :: MonadAsync m => (Config -> Config) -> m a -> Stream m a

704+repeatMWith cfg = sequenceWith cfg . Stream.repeat

705+

706+-- |

707+-- Definition:

708+--

709+-- >>> repeatM = Concur.repeatMWith id

668710--

669−-- Generate a stream by repeatedly executing a monadic action forever. The

670−-- number of actions that are executed concurrently is limited by 'maxThreads'.

671711{-# INLINE repeatM #-}

672712repeatM :: MonadAsync m => m a -> Stream m a

673−repeatM = sequence . Stream.repeat

713+repeatM = repeatMWith id

714+

715+-- | Generate a stream by concurrently performing a monadic action @n@ times.

716+--

717+-- Definition:

718+--

719+-- >>> replicateMWith cfg n = Concur.sequenceWith cfg . Stream.replicate n

720+--

721+-- Example, 'replicateMWith' in the following example executes all the

722+-- replicated actions concurrently, thus taking only 1 second:

723+--

724+-- >>> Stream.fold Fold.drain $ Concur.replicateMWith id 10 $ delay 1

725+-- ...

726+--

727+{-# INLINE replicateMWith #-}

728+replicateMWith :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a

729+replicateMWith cfg n = sequenceWith cfg . Stream.replicate n

674730

675731-- |

676−-- >>> replicateM n = Concur.sequence . Stream.replicate n

732+-- Definition:

733+--

734+-- >>> replicateM n = Concur.replicateMWith id

677735--

678−-- Generate a stream by concurrently performing a monadic action @n@ times.

679736{-# INLINE replicateM #-}

680737replicateM :: MonadAsync m => Int -> m a -> Stream m a

681−replicateM n = sequence . Stream.replicate n

738+replicateM = replicateMWith id

682739

683740-------------------------------------------------------------------------------

684741-- Reactive