@@ -47,22 +47,17 @@ module Streamly.Internal.Data.Stream.Concurrent
4747
4848 -- ** Evaluate
4949 -- | Evaluates a stream concurrently using a channel.
50− , eval
5150 , parEval
5251 -- Add unfoldrM/iterateM?
5352
5453 -- ** Generate
5554 -- | Uses a single channel to evaluate all actions.
56− , repeatM
5755 , parRepeatM
58− , replicateM
5956 , parReplicateM
6057
6158 -- ** Map
6259 -- | Uses a single channel to evaluate all actions.
63− , mapM
6460 , parMapM
65− , sequence
6661 , parSequence
6762
6863 -- ** Combine two
@@ -91,15 +86,11 @@ module Streamly.Internal.Data.Stream.Concurrent
9186
9287 -- ** Stream of streams
9388 -- *** Apply
94− , apply
9589 , parApply
9690
9791 -- *** Concat
9892 -- | Shares a single channel across many streams.
99− , concat
10093 , parConcat
101− , concatMap
102− , concatMapInterleave
10394 , parConcatMap
10495
10596 -- *** ConcatIterate
@@ -196,14 +187,6 @@ parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
196187parEval modifier input = withChannel modifier input (const id)
197188 -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
198189
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−
207190-------------------------------------------------------------------------------
208191-- combining two streams
209192-------------------------------------------------------------------------------
@@ -477,30 +460,7 @@ parConcatMapK modifier f input =
477460-- >>> f (Stream.maxThreads 1) [stream1, stream2]
478461-- [1,2,3,4,5,6]
479462--
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.
500−--
501−-- >>> concatMapInterleave = Stream.parConcatMap (Stream.interleaved True)
502−--
503−-- When used with a single thread it behaves like serial interleaving:
463+-- Schedule all streams in a round robin fashion over the available threads:
504464--
505465-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
506466--
@@ -509,11 +469,13 @@ concatMap = parConcatMap id
509469-- >>> f (Stream.maxThreads 1) [stream1, stream2]
510470-- [1,4,2,5,3,6]
511471--
512−-- /Works only on finite number of streams/
513−--
514−{-# INLINE concatMapInterleave #-}
515−concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
516−concatMapInterleave = parConcatMap (interleaved True)
472+{-# INLINE parConcatMap #-}
473+parConcatMap :: MonadAsync m =>
474+ (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
475+parConcatMap modifier f stream =
476+ Stream.fromStreamK
477+ $ parConcatMapK
478+ modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
517479
518480-- | Evaluate the streams in the input stream concurrently and combine them.
519481--
@@ -524,14 +486,6 @@ parConcat :: MonadAsync m =>
524486 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
525487parConcat modifier = parConcatMap modifier id
526488
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−
535489-------------------------------------------------------------------------------
536490-- concat Lists
537491-------------------------------------------------------------------------------
@@ -614,11 +568,6 @@ parApply modifier stream1 stream2 =
614568 (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)
615569 stream1
616570
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−
622571-------------------------------------------------------------------------------
623572-- Map
624573-------------------------------------------------------------------------------
@@ -644,15 +593,6 @@ parMapM :: MonadAsync m =>
644593 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
645594parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
646595
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−
656596-- |
657597-- >>> parSequence modifier = Stream.parMapM modifier id
658598--
@@ -661,14 +601,6 @@ parSequence :: MonadAsync m =>
661601 (Config -> Config) -> Stream m (m a) -> Stream m a
662602parSequence modifier = parMapM modifier id
663603
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−
672604-- | Evaluates the streams being zipped in separate threads than the consumer.
673605-- The zip function is evaluated in the consumer thread.
674606--
@@ -746,15 +678,6 @@ parConcatIterate modifier f input =
746678parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
747679parRepeatM cfg = parSequence cfg . Stream.repeat
748680
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−
758681-- | Generate a stream by concurrently performing a monadic action @n@ times.
759682--
760683-- Definition:
@@ -771,15 +694,6 @@ repeatM = parRepeatM id
771694parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
772695parReplicateM cfg n = parSequence cfg . Stream.replicate n
773696
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−
783697-------------------------------------------------------------------------------
784698-- Reactive
785699-------------------------------------------------------------------------------