@@ -48,22 +48,22 @@ module Streamly.Internal.Data.Stream.Concurrent
4848 -- ** Evaluate
4949 -- | Evaluates a stream concurrently using a channel.
5050 , eval
51− , evalWith
51+ , parEval
5252 -- Add unfoldrM/iterateM?
5353
5454 -- ** Generate
5555 -- | Uses a single channel to evaluate all actions.
5656 , repeatM
57− , repeatMWith
57+ , parRepeatM
5858 , replicateM
59− , replicateMWith
59+ , parReplicateM
6060
6161 -- ** Map
6262 -- | Uses a single channel to evaluate all actions.
6363 , mapM
64− , mapMWith
64+ , parMapM
6565 , sequence
66− , sequenceWith
66+ , parSequence
6767
6868 -- ** Combine two
6969 -- | Use a channel for each pair.
@@ -78,28 +78,29 @@ module Streamly.Internal.Data.Stream.Concurrent
7878
7979 -- ** List of streams
8080 -- | Shares a single channel across many streams.
81− , append
82− , interleave
83− , ahead
84− , parallel
85− , parallelFst
86− , parallelMin
87− , concatListWith
88− , zipWithM
89− , zipWith
81+ , parLazy
82+ , parLazyOrdered
83+ , parLazyInterleaved
84+ , parEager
85+ , parEagerFst
86+ , parEagerMin
87+ , parConcatList
88+
89+ , parZipWithM
90+ , parZipWith
9091
9192 -- ** Stream of streams
9293 -- *** Apply
9394 , apply
94− , applyWith
95+ , parApply
9596
9697 -- *** Concat
9798 -- | Shares a single channel across many streams.
9899 , concat
99− , concatWith
100+ , parConcat
100101 , concatMap
101102 , concatMapInterleave
102− , concatMapWith
103+ , parConcatMap
103104
104105 -- ** Reactive
105106 , fromCallback
@@ -144,7 +145,7 @@ import Streamly.Internal.Data.Stream.Concurrent.Channel
144145-- >>> import qualified Streamly.Data.Fold as Fold
145146-- >>> import qualified Streamly.Data.Parser as Parser
146147-- >>> import qualified Streamly.Data.Stream as Stream
147−-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur
148+-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream
148149-- >>> import Prelude hiding (concatMap, concat, zipWith)
149150-- >>> :{
150151-- delay n = do
@@ -158,9 +159,9 @@ import Streamly.Internal.Data.Stream.Concurrent.Channel
158159-------------------------------------------------------------------------------
159160
160161{-
161−{-# INLINE_NORMAL evalWithD #-}
162−evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
163−evalWithD modifier m = D.Stream step Nothing
162+{-# INLINE_NORMAL parEvalD #-}
163+parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
164+parEvalD modifier m = D.Stream step Nothing
164165 where
165166
166167 step _ Nothing = do
@@ -180,18 +181,23 @@ evalWithD modifier m = D.Stream step Nothing
180181-- | Evaluate a stream asynchronously using a channel and serve the consumer
181182-- from the evaluation buffer.
182183--
183−{-# INLINE evalWith #-}
184−evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
185−evalWith modifier input = withChannel modifier input (const id)
186− -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream
184+-- Note that the evaluation requires a single thread as only one stream needs
185+-- to be evaluated. Therefore, the concurrency options that are relevant to
186+-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,
187+-- ordered, stopWhen options won't have any effect.
188+--
189+{-# INLINE parEval #-}
190+parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
191+parEval modifier input = withChannel modifier input (const id)
192+ -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream
187193
188194-- | Definition:
189195--
190−-- >>> eval = Concur.evalWith id
196+-- >>> eval = Stream.parEval id
191197--
192198{-# INLINE eval #-}
193199eval :: MonadAsync m => Stream m a -> Stream m a
194−eval = evalWith id
200+eval = parEval id
195201
196202-------------------------------------------------------------------------------
197203-- combining two streams
@@ -232,18 +238,18 @@ appendWithK :: MonadAsync m =>
232238appendWithK modifier stream1 stream2 =
233239{-
234240 if getOrdered (modifier defaultConfig)
235− then concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
241+ then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
236242 else _appendGeneric Append.newChannel modifier stream1 stream2
237243-}
238− concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
244+ parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)
239245
240246-- | Evaluate the first stream in the current thread and add the second stream
241247-- to the supplied channel. This is to be used by a worker thread.
242248--
243−-- This can be used with concatMapWith:
249+-- This can be used with parConcatMap:
244250--
245251-- @
246−-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream
252+-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream
247253-- @
248254--
249255{-# INLINE _appendWithChanK #-}
@@ -272,13 +278,13 @@ combineWith modifier stream1 stream2 =
272278-- all streams. However, with this operation you can precisely control the
273279-- scheduling by creating arbitrary shape expression trees.
274280--
275−-- >>> append2 = Concur.combineWith id
281+-- >>> append2 = Stream.combineWith id
276282--
277283-- The following code finishes in 4 seconds:
278284--
279285-- >>> stream1 = Stream.fromEffect (delay 4)
280286-- >>> stream2 = Stream.fromEffect (delay 2)
281−-- >>> Stream.fold Fold.toList $ stream1 `Concur.append2` stream2
287+-- >>> Stream.fold Fold.toList $ stream1 `Stream.append2` stream2
282288-- 2 sec
283289-- 4 sec
284290-- [2,4]
@@ -291,7 +297,7 @@ append2 = combineWith id
291297-- the left stream. This schedules all streams in a round robin fashion over
292298-- limited number of threads.
293299--
294−-- >>> interleave2 = Concur.combineWith (Concur.interleaved True)
300+-- >>> interleave2 = Stream.combineWith (Stream.interleaved True)
295301--
296302{-# INLINE interleave2 #-}
297303interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
@@ -299,7 +305,7 @@ interleave2 = combineWith (interleaved True)
299305
300306-- | Like 'append' but with 'ordered' on.
301307--
302−-- >>> ahead2 = Concur.combineWith (Concur.ordered True)
308+-- >>> ahead2 = Stream.combineWith (Stream.ordered True)
303309--
304310{-# INLINE ahead2 #-}
305311ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
@@ -307,7 +313,7 @@ ahead2 = combineWith (ordered True)
307313
308314-- | Like 'append2' but with 'eager' on.
309315--
310−-- >>> parallel2 = Concur.combineWith (Concur.eager True)
316+-- >>> parallel2 = Stream.combineWith (Stream.eager True)
311317--
312318{-# INLINE parallel2 #-}
313319parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
@@ -315,7 +321,7 @@ parallel2 = combineWith (eager True)
315321
316322-- | Like 'parallel2' but stops the output as soon as the first stream stops.
317323--
318−-- >>> parallelFst2 = Concur.combineWith (Concur.eager True . Concur.stopWhen Concur.FirstStops)
324+-- >>> parallelFst2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.FirstStops)
319325--
320326-- /Pre-release/
321327{-# INLINE parallelFst2 #-}
@@ -325,7 +331,7 @@ parallelFst2 = combineWith (eager True . stopWhen FirstStops)
325331-- | Like 'parallel2' but stops the output as soon as any of the two streams
326332-- stops.
327333--
328−-- >>> parallelMin2 = Concur.combineWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)
334+-- >>> parallelMin2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.AnyStops)
329335--
330336-- /Pre-release/
331337{-# INLINE parallelMin2 #-}
@@ -366,26 +372,26 @@ mkEnqueue chan runner = do
366372--
367373-- This function is designed to be used by worker threads on a channel to
368374-- concurrently map and evaluate a stream.
369−{-# INLINE concatMapWithChanK #-}
370−concatMapWithChanK :: MonadAsync m =>
375+{-# INLINE parConcatMapChanK #-}
376+parConcatMapChanK :: MonadAsync m =>
371377 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
372−concatMapWithChanK chan f stream =
378+parConcatMapChanK chan f stream =
373379 let run q = concatMapDivK q f
374380 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
375− -- K.concatMapWith (_appendWithChanK chan) f stream
381+ -- K.parConcatMap (_appendWithChanK chan) f stream
376382
377−{-# INLINE concatMapWithChanKAny #-}
378−concatMapWithChanKAny :: MonadAsync m =>
383+{-# INLINE parConcatMapChanKAny #-}
384+parConcatMapChanKAny :: MonadAsync m =>
379385 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
380−concatMapWithChanKAny chan f stream =
386+parConcatMapChanKAny chan f stream =
381387 let done = K.nilM (stopChannel chan)
382388 run q = concatMapDivK q (\x -> K.serial (f x) done)
383389 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
384390
385−{-# INLINE concatMapWithChanKFirst #-}
386−concatMapWithChanKFirst :: MonadAsync m =>
391+{-# INLINE parConcatMapChanKFirst #-}
392+parConcatMapChanKFirst :: MonadAsync m =>
387393 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
388−concatMapWithChanKFirst chan f stream =
394+parConcatMapChanKFirst chan f stream =
389395 let done = K.nilM (stopChannel chan)
390396 run q = concatMapDivK q f
391397 in K.concatEffect $ do
@@ -397,19 +403,19 @@ concatMapWithChanKFirst chan f stream =
397403 q t
398404 return $ K.serial (f h) done
399405
400−{-# INLINE concatMapWithChanKGeneric #-}
401−concatMapWithChanKGeneric :: MonadAsync m =>
406+{-# INLINE parConcatMapChanKGeneric #-}
407+parConcatMapChanKGeneric :: MonadAsync m =>
402408 (Config -> Config)
403409 -> Channel m b
404410 -> (a -> K.Stream m b)
405411 -> K.Stream m a
406412 -> K.Stream m b
407−concatMapWithChanKGeneric modifier chan f stream = do
413+parConcatMapChanKGeneric modifier chan f stream = do
408414 let cfg = modifier defaultConfig
409415 case getStopWhen cfg of
410− AllStop -> concatMapWithChanK chan f stream
411− FirstStops -> concatMapWithChanKFirst chan f stream
412− AnyStops -> concatMapWithChanKAny chan f stream
416+ AllStop -> parConcatMapChanK chan f stream
417+ FirstStops -> parConcatMapChanKFirst chan f stream
418+ AnyStops -> parConcatMapChanKAny chan f stream
413419
414420-- XXX Add a deep evaluation variant that evaluates individual elements in the
415421-- generated streams in parallel.
@@ -417,11 +423,11 @@ concatMapWithChanKGeneric modifier chan f stream = do
417423-- | Allocate a channel and use it to concurrently evaluate the streams
418424-- generated by the mapped function.
419425--
420−{-# INLINE concatMapWithK #-}
421−concatMapWithK :: MonadAsync m =>
426+{-# INLINE parConcatMapK #-}
427+parConcatMapK :: MonadAsync m =>
422428 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
423−concatMapWithK modifier f input =
424− let g = concatMapWithChanKGeneric modifier
429+parConcatMapK modifier f input =
430+ let g = parConcatMapChanKGeneric modifier
425431 in withChannelK modifier input (`g` f)
426432
427433-- | Map each element of the input to a stream and then concurrently evaluate
@@ -431,11 +437,11 @@ concatMapWithK modifier f input =
431437--
432438-- Definition:
433439--
434−-- >>> concatMapWith modifier f stream = Concur.concatWith modifier $ fmap f stream
440+-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream
435441--
436442-- Examples:
437443--
438−-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith cfg id $ Stream.fromList xs
444+-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs
439445--
440446-- The following streams finish in 4 seconds:
441447--
@@ -451,140 +457,140 @@ concatMapWithK modifier f input =
451457-- Limiting threads to 2 schedules the third stream only after one of the first
452458-- two has finished, releasing a thread:
453459--
454−-- >>> f (Concur.maxThreads 2) [stream1, stream2, stream3]
460+-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]
455461-- ...
456462-- [2,1,4]
457463--
458464-- When used with a Single thread it behaves like serial concatMap:
459465--
460−-- >>> f (Concur.maxThreads 1) [stream1, stream2, stream3]
466+-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]
461467-- ...
462468-- [4,2,1]
463469--
464470-- >>> stream1 = Stream.fromList [1,2,3]
465471-- >>> stream2 = Stream.fromList [4,5,6]
466−-- >>> f (Concur.maxThreads 1) [stream1, stream2]
472+-- >>> f (Stream.maxThreads 1) [stream1, stream2]
467473-- [1,2,3,4,5,6]
468474--
469−{-# INLINE concatMapWith #-}
470−concatMapWith :: MonadAsync m =>
475+{-# INLINE parConcatMap #-}
476+parConcatMap :: MonadAsync m =>
471477 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
472−concatMapWith modifier f stream =
478+parConcatMap modifier f stream =
473479 Stream.fromStreamK
474− $ concatMapWithK
480+ $ parConcatMapK
475481 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
476482
477483-- | Definition:
478484--
479−-- >>> concatMap = Concur.concatMapWith id
485+-- >>> concatMap = Stream.parConcatMap id
480486--
481487{-# INLINE concatMap #-}
482488concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
483−concatMap = concatMapWith id
489+concatMap = parConcatMap id
484490
485491-- | Map each element of the input to a stream and then concurrently evaluate
486492-- and interleave the resulting streams. Unlike 'concatMap' which prefers to
487493-- evaluate the earlier stream first, this schedules all streams in a round
488494-- robin fashion over the available threads.
489495--
490−-- >>> concatMapInterleave = Concur.concatMapWith (Concur.interleaved True)
496+-- >>> concatMapInterleave = Stream.parConcatMap (Stream.interleaved True)
491497--
492498-- When used with a single thread it behaves like serial interleaving:
493499--
494−-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith (Concur.interleaved True . cfg) id $ Stream.fromList xs
500+-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs
495501--
496502-- >>> stream1 = Stream.fromList [1,2,3]
497503-- >>> stream2 = Stream.fromList [4,5,6]
498−-- >>> f (Concur.maxThreads 1) [stream1, stream2]
504+-- >>> f (Stream.maxThreads 1) [stream1, stream2]
499505-- [1,4,2,5,3,6]
500506--
501507-- /Works only on finite number of streams/
502508--
503509{-# INLINE concatMapInterleave #-}
504510concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
505−concatMapInterleave = concatMapWith (interleaved True)
511+concatMapInterleave = parConcatMap (interleaved True)
506512
507513-- | Evaluate the streams in the input stream concurrently and combine them.
508514--
509−-- >>> concatWith modifier = Concur.concatMapWith modifier id
515+-- >>> parConcat modifier = Stream.parConcatMap modifier id
510516--
511−{-# INLINE concatWith #-}
512−concatWith :: MonadAsync m =>
517+{-# INLINE parConcat #-}
518+parConcat :: MonadAsync m =>
513519 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
514−concatWith modifier = concatMapWith modifier id
520+parConcat modifier = parConcatMap modifier id
515521
516522-- | Definition:
517523--
518−-- >>> concat = Concur.concatWith id
524+-- >>> concat = Stream.parConcat id
519525--
520526{-# INLINE concat #-}
521527concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a
522−concat = concatWith id
528+concat = parConcat id
523529
524530-------------------------------------------------------------------------------
525531-- concat Lists
526532-------------------------------------------------------------------------------
527533
528−-- | Like 'concatWith' but works on a list of streams.
534+-- | Like 'parConcat' but works on a list of streams.
529535--
530−-- >>> concatListWith modifier = Concur.concatWith modifier . Stream.fromList
536+-- >>> parConcatList modifier = Stream.parConcat modifier . Stream.fromList
531537--
532−{-# INLINE concatListWith #-}
533−concatListWith :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
534−concatListWith modifier = concatWith modifier . Stream.fromList
538+{-# INLINE parConcatList #-}
539+parConcatList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a
540+parConcatList modifier = parConcat modifier . Stream.fromList
535541
536542-- | Like 'concat' but works on a list of streams.
537543--
538−-- >>> append = Concur.concatListWith id
544+-- >>> parLazy = Stream.parConcatList id
539545--
540−{-# INLINE append #-}
541−append :: MonadAsync m => [Stream m a] -> Stream m a
542−append = concatListWith id
546+{-# INLINE parLazy #-}
547+parLazy :: MonadAsync m => [Stream m a] -> Stream m a
548+parLazy = parConcatList id
543549
544−-- | Like 'append' but interleaves the streams fairly instead of prioritizing
550+-- | Like 'parLazy' but interleaves the streams fairly instead of prioritizing
545551-- the left stream. This schedules all streams in a round robin fashion over
546552-- limited number of threads.
547553--
548−-- >>> interleave = Concur.concatListWith (Concur.interleaved True)
554+-- >>> parLazyInterleaved = Stream.parConcatList (Stream.interleaved True)
549555--
550−{-# INLINE interleave #-}
551−interleave :: MonadAsync m => [Stream m a] -> Stream m a
552−interleave = concatListWith (interleaved True)
556+{-# INLINE parLazyInterleaved #-}
557+parLazyInterleaved :: MonadAsync m => [Stream m a] -> Stream m a
558+parLazyInterleaved = parConcatList (interleaved True)
553559
554−-- | Like 'append' but with 'ordered' on.
560+-- | Like 'parLazy' but with 'ordered' on.
555561--
556−-- >>> ahead = Concur.concatListWith (Concur.ordered True)
562+-- >>> parLazyOrdered = Stream.parConcatList (Stream.ordered True)
557563--
558−{-# INLINE ahead #-}
559−ahead :: MonadAsync m => [Stream m a] -> Stream m a
560−ahead = concatListWith (ordered True)
564+{-# INLINE parLazyOrdered #-}
565+parLazyOrdered :: MonadAsync m => [Stream m a] -> Stream m a
566+parLazyOrdered = parConcatList (ordered True)
561567
562−-- | Like 'append' but with 'eager' on.
568+-- | Like 'parLazy' but with 'eager' on.
563569--
564−-- >>> parallel = Concur.concatListWith (Concur.eager True)
570+-- >>> parEager = Stream.parConcatList (Stream.eager True)
565571--
566−{-# INLINE parallel #-}
567−parallel :: MonadAsync m => [Stream m a] -> Stream m a
568−parallel = concatListWith (eager True)
572+{-# INLINE parEager #-}
573+parEager :: MonadAsync m => [Stream m a] -> Stream m a
574+parEager = parConcatList (eager True)
569575
570−-- | Like 'parallel' but stops the output as soon as the first stream stops.
576+-- | Like 'parEager' but stops the output as soon as the first stream stops.
571577--
572−-- >>> parallelFst = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.FirstStops)
578+-- >>> parEagerFst = Stream.parConcatList (Stream.eager True . Stream.stopWhen Stream.FirstStops)
573579--
574−{-# INLINE parallelFst #-}
575−parallelFst :: MonadAsync m => [Stream m a] -> Stream m a
576−parallelFst = concatListWith (eager True . stopWhen FirstStops)
580+{-# INLINE parEagerFst #-}
581+parEagerFst :: MonadAsync m => [Stream m a] -> Stream m a
582+parEagerFst = parConcatList (eager True . stopWhen FirstStops)
577583
578−-- | Like 'parallel' but stops the output as soon as any of the two streams
584+-- | Like 'parEager' but stops the output as soon as any of the two streams
579585-- stops.
580586--
581587-- Definition:
582588--
583−-- >>> parallelMin = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)
589+-- >>> parEagerMin = Stream.parConcatList (Stream.eager True . Stream.stopWhen Stream.AnyStops)
584590--
585−{-# INLINE parallelMin #-}
586−parallelMin :: MonadAsync m => [Stream m a] -> Stream m a
587−parallelMin = concatListWith (eager True . stopWhen AnyStops)
591+{-# INLINE parEagerMin #-}
592+parEagerMin :: MonadAsync m => [Stream m a] -> Stream m a
593+parEagerMin = parConcatList (eager True . stopWhen AnyStops)
588594
589595-------------------------------------------------------------------------------
590596-- Applicative
@@ -592,21 +598,21 @@ parallelMin = concatListWith (eager True . stopWhen AnyStops)
592598
593599-- | Apply an argument stream to a function stream concurrently. Uses a
594600-- shared channel for all individual applications within a stream application.
595−{-# INLINE applyWith #-}
596−{-# SPECIALIZE applyWith ::
601+{-# INLINE parApply #-}
602+{-# SPECIALIZE parApply ::
597603 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
598−applyWith :: MonadAsync m =>
604+parApply :: MonadAsync m =>
599605 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b
600−applyWith modifier stream1 stream2 =
601− concatMapWith
606+parApply modifier stream1 stream2 =
607+ parConcatMap
602608 modifier
603− (\g -> concatMapWith modifier (pure . g) stream2)
609+ (\g -> parConcatMap modifier (pure . g) stream2)
604610 stream1
605611
606612{-# INLINE apply #-}
607613{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
608614apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b
609−apply = applyWith id
615+apply = parApply id
610616
611617-------------------------------------------------------------------------------
612618-- Map
@@ -615,79 +621,85 @@ apply = applyWith id
615621-- |
616622-- Definition:
617623--
618−-- >>> mapMWith modifier f = Concur.concatMapWith modifier (Stream.fromEffect . f)
624+-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)
619625--
620626-- Example, the following example finishes in 1 second as all actions run in
621627-- parallel. Even though results are available out of order they are ordered
622628-- due to the config option::
623629--
624630-- >>> f x = delay x >> return x
625−-- >>> Stream.fold Fold.toList $ Concur.mapMWith (Concur.ordered True) f $ Stream.fromList [3,2,1]
631+-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]
626632-- 1 sec
627633-- 2 sec
628634-- 3 sec
629635-- [3,2,1]
630636--
631−{-# INLINE mapMWith #-}
632−mapMWith :: MonadAsync m =>
637+{-# INLINE parMapM #-}
638+parMapM :: MonadAsync m =>
633639 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
634−mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)
640+parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)
635641
636642-- |
637643-- Definition:
638644--
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]
645+-- >>> mapM = Stream.parMapM id
650646--
651647{-# INLINE mapM #-}
652648mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
653−mapM = mapMWith id
649+mapM = parMapM id
654650
655651-- |
656−-- >>> sequenceWith modifier = Concur.mapMWith modifier id
652+-- >>> parSequence modifier = Stream.parMapM modifier id
657653--
658−{-# INLINE sequenceWith #-}
659−sequenceWith :: MonadAsync m =>
654+{-# INLINE parSequence #-}
655+parSequence :: MonadAsync m =>
660656 (Config -> Config) -> Stream m (m a) -> Stream m a
661−sequenceWith modifier = mapMWith modifier id
657+parSequence modifier = parMapM modifier id
662658
663659-- |
664−-- >>> sequence = Concur.sequenceWith id
660+-- >>> sequence = Stream.parSequence id
665661--
666662{-# INLINE sequence #-}
667663sequence :: MonadAsync m =>
668664 Stream m (m a) -> Stream m a
669−sequence = sequenceWith id
665+sequence = parSequence id
670666
671−-- |
672−-- >>> zipWithM f m1 m2 = Stream.zipWithM f (Concur.eval m1) (Concur.eval m2)
667+-- | Evaluates the streams being zipped in separate threads than the consumer.
668+-- The zip function is evaluated in the consumer thread.
669+--
670+-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
673671--
674−{-# INLINE zipWithM #-}
675−zipWithM :: MonadAsync m
676− => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
677−zipWithM f m1 m2 = Stream.zipWithM f (eval m1) (eval m2)
672+-- Multi-stream concurrency options won't apply here, see the notes in
673+-- 'parEval'.
674+--
675+-- If you want to evaluate the zip function as well in a separate thread, you
676+-- can use a 'parEval' on 'parZipWithM'.
677+--
678+{-# INLINE parZipWithM #-}
679+parZipWithM :: MonadAsync m
680+ => (Config -> Config)
681+ -> (a -> b -> m c)
682+ -> Stream m a
683+ -> Stream m b
684+ -> Stream m c
685+parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)
678686
679687-- |
680−-- >>> zipWith f = Stream.zipWithM (\a b -> return $ f a b)
688+-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)
681689--
682690-- >>> m1 = Stream.fromList [1,2,3]
683691-- >>> m2 = Stream.fromList [4,5,6]
684−-- >>> Stream.fold Fold.toList $ Concur.zipWith (,) m1 m2
692+-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2
685693-- [(1,4),(2,5),(3,6)]
686694--
687−{-# INLINE zipWith #-}
688−zipWith :: MonadAsync m
689− => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
690−zipWith f = zipWithM (\a b -> return $ f a b)
695+{-# INLINE parZipWith #-}
696+parZipWith :: MonadAsync m
697+ => (Config -> Config)
698+ -> (a -> b -> c)
699+ -> Stream m a
700+ -> Stream m b
701+ -> Stream m c
702+parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
691703
692704-------------------------------------------------------------------------------
693705-- Generate
@@ -696,46 +708,46 @@ zipWith f = zipWithM (\a b -> return $ f a b)
696708-- |
697709-- Definition:
698710--
699−-- >>> repeatMWith cfg = Concur.sequenceWith cfg . Stream.repeat
711+-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat
700712--
701713-- 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
714+{-# INLINE parRepeatM #-}
715+parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a
716+parRepeatM cfg = parSequence cfg . Stream.repeat
705717
706718-- |
707719-- Definition:
708720--
709−-- >>> repeatM = Concur.repeatMWith id
721+-- >>> repeatM = Stream.parRepeatM id
710722--
711723{-# INLINE repeatM #-}
712724repeatM :: MonadAsync m => m a -> Stream m a
713−repeatM = repeatMWith id
725+repeatM = parRepeatM id
714726
715727-- | Generate a stream by concurrently performing a monadic action @n@ times.
716728--
717729-- Definition:
718730--
719−-- >>> replicateMWith cfg n = Concur.sequenceWith cfg . Stream.replicate n
731+-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n
720732--
721−-- Example, 'replicateMWith' in the following example executes all the
733+-- Example, 'parReplicateM' in the following example executes all the
722734-- replicated actions concurrently, thus taking only 1 second:
723735--
724−-- >>> Stream.fold Fold.drain $ Concur.replicateMWith id 10 $ delay 1
736+-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1
725737-- ...
726738--
727−{-# INLINE replicateMWith #-}
728−replicateMWith :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
729−replicateMWith cfg n = sequenceWith cfg . Stream.replicate n
739+{-# INLINE parReplicateM #-}
740+parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a
741+parReplicateM cfg n = parSequence cfg . Stream.replicate n
730742
731743-- |
732744-- Definition:
733745--
734−-- >>> replicateM n = Concur.replicateMWith id
746+-- >>> replicateM n = Stream.parReplicateM id
735747--
736748{-# INLINE replicateM #-}
737749replicateM :: MonadAsync m => Int -> m a -> Stream m a
738−replicateM = replicateMWith id
750+replicateM = parReplicateM id
739751
740752-------------------------------------------------------------------------------
741753-- Reactive
@@ -824,7 +836,7 @@ pollCountsD predicate fld (D.Stream step state) = D.Stream step' Nothing
824836-- For example, to print the count of elements processed every second:
825837--
826838-- @
827−-- > Stream.drain $ Concur.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (Fold.drainBy print)
839+-- > Stream.drain $ Stream.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (Fold.drainBy print)
828840-- $ Stream.enumerateFrom 0
829841-- @
830842--