1a89f0c · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Review and update the Stream.Concurrent docs/API

1a89f0cHarendra Kumar4 years ago

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

compare with master

unifiedsplit

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

5-- Maintainer : streamly@composewell.com

6-- Stability : released

7-- 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.

9module Streamly.Data.Stream.Concurrent

84module Streamly.Data.Stream.Concurrent

10 (

85 (

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

117

43 -- ** Evaluate

118 -- ** Evaluate

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

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

120+

45− , eval

121+ -- , eval

46 , evalWith

122 , evalWith

123

48 -- ** Generate

124 -- ** Generate

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

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

50− , repeatM

126+ , repeatMWith

51− , replicateM

127+ , replicateMWith

128

53 -- ** Map

129 -- ** Map

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

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

131+

55− , mapM

132+ -- , mapM

56 , mapMWith

133 , mapMWith

57− , sequence

134+ -- , sequence

58 , sequenceWith

135 , sequenceWith

136

60 -- ** List of streams

137 -- ** List of streams

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

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

139+

62− , append

140+ -- , append

63− , ahead

141+ -- , ahead

64− , parallel

142+ -- , parallel

65 , concatListWith

143 , concatListWith

66− , parallelFst

144+ -- , parallelFst

67− , parallelMin

145+ -- , parallelMin

68− , zipWithM

146+ -- , zipWithM

69− , zipWith

147+ -- , zipWith

148

71 -- ** Stream of streams

149 -- ** Stream of streams

72 -- *** Apply

150 -- *** Apply

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

151+

74− -- channel for each application.

75− , apply

152+ -- , apply

76 , applyWith

153 , applyWith

154

78 -- *** Concat

155 -- *** Concat

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

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

157+

80− , concat

158+ -- , concat

81 , concatWith

159 , concatWith

82− , concatMap

160+ -- , concatMap

83 , concatMapWith

161 , concatMapWith

84 )

162 )

85where

163where

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

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

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

307 }

308

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

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

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.

312data StopWhen =

310data StopWhen =

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

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

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

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

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

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

316

314

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

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

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

322

320

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

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

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

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

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

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

326--

324--

327data Config = Config

325data Config = Config

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

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

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

503-- 'maxBuffer' is reached.

501-- 'maxBuffer' is reached.

504--

502--

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

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

504+--

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

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

507--

506--

508eager :: Bool -> Config -> Config

507eager :: Bool -> Config -> Config

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

534-- threads.

533-- threads.

535--

534--

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

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

536+--

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

538--

539interleaved :: Bool -> Config -> Config

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

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

54 -- ** Generate

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

56 , repeatM

57+ , repeatMWith

57 , replicateM

58 , replicateM

59+ , replicateMWith

59 -- ** Map

61 -- ** Map

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

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

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

89 -- ** Stream of streams

91 -- ** Stream of streams

90 -- *** Apply

92 -- *** Apply

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

92 , apply

93 , apply

93 , applyWith

94 , applyWith

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

176 D.Stop -> D.Stop

177 D.Stop -> D.Stop

177-}

178-}

178

179

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

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

180−-- channel parameters.

181+-- from the evaluation buffer.

181−--

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

183--

182--

184{-# INLINE evalWith #-}

183{-# INLINE evalWith #-}

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

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

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

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

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

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

188

187

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

188+-- | Definition:

190−-- from the evaluation buffer.

191--

189--

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

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

193--

191--

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

426 let g = concatMapWithChanKGeneric modifier

424 let g = concatMapWithChanKGeneric modifier

427 in withChannelK modifier input (`g` f)

425 in withChannelK modifier input (`g` f)

428

426

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−

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

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

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

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

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

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

445-- used as they arrive.

430-- used as they arrive.

446--

431--

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

432+-- Definition:

433+--

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

435+--

436+-- Examples:

448--

437--

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

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

450--

439--

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

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

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

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

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

479--

468--

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+--

480{-# INLINE concatMap #-}

481{-# INLINE concatMap #-}

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

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

482concatMap = concatMapWith id

483concatMap = concatMapWith id

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

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

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

504concatMapInterleave = concatMapWith (interleaved True)

505concatMapInterleave = concatMapWith (interleaved True)

505

506

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

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

507−-- configuration parameters using Config modifiers.

508--

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

510--

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

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

514concatWith modifier = concatMapWith modifier id

515

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

516+-- | Definition:

517--

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

519--

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

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

579-- stops.

580--

581+-- Definition:

582+--

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

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

582--

584--

583{-# INLINE parallelMin #-}

585{-# INLINE parallelMin #-}

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

588-- Applicative

590-- Applicative

589-------------------------------------------------------------------------------

591-------------------------------------------------------------------------------

590

592

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

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

591{-# INLINE applyWith #-}

595{-# INLINE applyWith #-}

592{-# SPECIALIZE applyWith ::

596{-# SPECIALIZE applyWith ::

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

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

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

609-------------------------------------------------------------------------------

613-------------------------------------------------------------------------------

610

614

611-- |

615-- |

616+-- Definition:

617+--

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

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

613--

619--

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+--

614{-# INLINE mapMWith #-}

631{-# INLINE mapMWith #-}

615mapMWith :: MonadAsync m =>

632mapMWith :: MonadAsync m =>

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

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

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

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

618

635

619-- |

636-- |

637+-- Definition:

638+--

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

639-- >>> 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+--

621{-# INLINE mapM #-}

651{-# INLINE mapM #-}

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

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

623mapM = mapMWith id

653mapM = mapMWith id

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

664-------------------------------------------------------------------------------

694-------------------------------------------------------------------------------

665

695

666-- |

696-- |

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+--

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

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

668--

710--

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

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

671{-# INLINE repeatM #-}

711{-# INLINE repeatM #-}

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

712repeatM :: 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

674

730

675-- |

731-- |

732+-- Definition:

733+--

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

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

677--

735--

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

679{-# INLINE replicateM #-}

736{-# INLINE replicateM #-}

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

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

681−replicateM n = sequence . Stream.replicate n

738+replicateM = replicateMWith id

682

739

683-------------------------------------------------------------------------------

740-------------------------------------------------------------------------------

684-- Reactive

741-- Reactive