21c25f8 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Remove some obsolete ops from Concurrent module

21c25f8Harendra Kumar4 years ago

src/Streamly/Internal/Data/Stream/Concurrent.hs+8−94

compare with master

unifiedsplit

@@ -47,22 +47,17 @@ module Streamly.Internal.Data.Stream.Concurrent

48 -- ** Evaluate

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

50− , eval

51 , parEval

50 , parEval

52 -- Add unfoldrM/iterateM?

51 -- Add unfoldrM/iterateM?

54 -- ** Generate

53 -- ** Generate

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

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

56− , repeatM

57 , parRepeatM

55 , parRepeatM

58− , replicateM

59 , parReplicateM

56 , parReplicateM

61 -- ** Map

58 -- ** Map

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

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

63− , mapM

64 , parMapM

60 , parMapM

65− , sequence

66 , parSequence

61 , parSequence

68 -- ** Combine two

63 -- ** Combine two

@@ -91,15 +86,11 @@ module Streamly.Internal.Data.Stream.Concurrent

92 -- ** Stream of streams

87 -- ** Stream of streams

93 -- *** Apply

88 -- *** Apply

94− , apply

95 , parApply

89 , parApply

97 -- *** Concat

91 -- *** Concat

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

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

99− , concat

100 , parConcat

93 , parConcat

101− , concatMap

102− , concatMapInterleave

103 , parConcatMap

94 , parConcatMap

104

105 -- *** ConcatIterate

96 -- *** ConcatIterate

@@ -196,14 +187,6 @@ parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a

196parEval modifier input = withChannel modifier input (const id)

187parEval modifier input = withChannel modifier input (const id)

197 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream

188 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream

198

189

199−-- | Definition:

200−--

201−-- >>> eval = Stream.parEval id

202−--

203−{-# INLINE eval #-}

204−eval :: MonadAsync m => Stream m a -> Stream m a

205−eval = parEval id

206−

207-------------------------------------------------------------------------------

190-------------------------------------------------------------------------------

208-- combining two streams

191-- combining two streams

209-------------------------------------------------------------------------------

192-------------------------------------------------------------------------------

@@ -477,30 +460,7 @@ parConcatMapK modifier f input =

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

460-- >>> f (Stream.maxThreads 1) [stream1, stream2]

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

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

479--

462--

480−{-# INLINE parConcatMap #-}

481−parConcatMap :: MonadAsync m =>

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

483−parConcatMap modifier f stream =

484− Stream.fromStreamK

485− $ parConcatMapK

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

487−

488−-- | Definition:

489−--

490−-- >>> concatMap = Stream.parConcatMap id

491−--

492−{-# INLINE concatMap #-}

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

494−concatMap = parConcatMap id

495−

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

497−-- and interleave the resulting streams. Unlike 'concatMap' which prefers to

498−-- evaluate the earlier stream first, this schedules all streams in a round

499−-- robin fashion over the available threads.

463+-- Schedule all streams in a round robin fashion over the available threads:

500−--

501−-- >>> concatMapInterleave = Stream.parConcatMap (Stream.interleaved True)

502−--

503−-- When used with a single thread it behaves like serial interleaving:

504--

464--

505-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs

465-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs

506--

466--

@@ -509,11 +469,13 @@ concatMap = parConcatMap id

509-- >>> f (Stream.maxThreads 1) [stream1, stream2]

469-- >>> f (Stream.maxThreads 1) [stream1, stream2]

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

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

511--

471--

512−-- /Works only on finite number of streams/

472+{-# INLINE parConcatMap #-}

513−--

473+parConcatMap :: MonadAsync m =>

514−{-# INLINE concatMapInterleave #-}

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

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

516−concatMapInterleave = parConcatMap (interleaved True)

475+parConcatMap modifier f stream =

476+ Stream.fromStreamK

477+ $ parConcatMapK

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

517

479

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

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

519--

481--

@@ -524,14 +486,6 @@ parConcat :: MonadAsync m =>

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

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

525parConcat modifier = parConcatMap modifier id

487parConcat modifier = parConcatMap modifier id

526

488

527−-- | Definition:

528−--

529−-- >>> concat = Stream.parConcat id

530−--

531−{-# INLINE concat #-}

532−concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a

533−concat = parConcat id

534−

535-------------------------------------------------------------------------------

489-------------------------------------------------------------------------------

536-- concat Lists

490-- concat Lists

537-------------------------------------------------------------------------------

491-------------------------------------------------------------------------------

@@ -614,11 +568,6 @@ parApply modifier stream1 stream2 =

614 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)

568 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)

615 stream1

569 stream1

616

570

617−{-# INLINE apply #-}

618−{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}

619−apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b

620−apply = parApply id

621−

622-------------------------------------------------------------------------------

571-------------------------------------------------------------------------------

623-- Map

572-- Map

624-------------------------------------------------------------------------------

573-------------------------------------------------------------------------------

@@ -644,15 +593,6 @@ parMapM :: MonadAsync m =>

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

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

645parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)

594parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)

646

595

647−-- |

648−-- Definition:

649−--

650−-- >>> mapM = Stream.parMapM id

651−--

652−{-# INLINE mapM #-}

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

654−mapM = parMapM id

655−

656-- |

596-- |

657-- >>> parSequence modifier = Stream.parMapM modifier id

597-- >>> parSequence modifier = Stream.parMapM modifier id

658--

598--

@@ -661,14 +601,6 @@ parSequence :: MonadAsync m =>

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

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

662parSequence modifier = parMapM modifier id

602parSequence modifier = parMapM modifier id

663

603

664−-- |

665−-- >>> sequence = Stream.parSequence id

666−--

667−{-# INLINE sequence #-}

668−sequence :: MonadAsync m =>

669− Stream m (m a) -> Stream m a

670−sequence = parSequence id

671−

672-- | Evaluates the streams being zipped in separate threads than the consumer.

604-- | Evaluates the streams being zipped in separate threads than the consumer.

673-- The zip function is evaluated in the consumer thread.

605-- The zip function is evaluated in the consumer thread.

674--

606--

@@ -746,15 +678,6 @@ parConcatIterate modifier f input =

746parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a

678parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a

747parRepeatM cfg = parSequence cfg . Stream.repeat

679parRepeatM cfg = parSequence cfg . Stream.repeat

748

680

749−-- |

750−-- Definition:

751−--

752−-- >>> repeatM = Stream.parRepeatM id

753−--

754−{-# INLINE repeatM #-}

755−repeatM :: MonadAsync m => m a -> Stream m a

756−repeatM = parRepeatM id

757−

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

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

759--

682--

760-- Definition:

683-- Definition:

@@ -771,15 +694,6 @@ repeatM = parRepeatM id

771parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a

694parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a

772parReplicateM cfg n = parSequence cfg . Stream.replicate n

695parReplicateM cfg n = parSequence cfg . Stream.replicate n

773

696

774−-- |

775−-- Definition:

776−--

777−-- >>> replicateM n = Stream.parReplicateM id

778−--

779−{-# INLINE replicateM #-}

780−replicateM :: MonadAsync m => Int -> m a -> Stream m a

781−replicateM = parReplicateM id

782−

783-------------------------------------------------------------------------------

697-------------------------------------------------------------------------------

784-- Reactive

698-- Reactive

785-------------------------------------------------------------------------------

699-------------------------------------------------------------------------------