a285b55 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Change mkParallel and mkAsync signatures

a285b55Harendra Kumar7 years ago

src/Streamly.hs+16−1

compare with master

unifiedsplit

@@ -52,6 +52,7 @@

52-- package.

54{-# LANGUAGE CPP #-}

55+{-# LANGUAGE FlexibleContexts #-}

55{-# LANGUAGE MultiParamTypeClasses #-}

56{-# LANGUAGE MultiParamTypeClasses #-}

57#if __GLASGOW_HASKELL__ >= 800

58#if __GLASGOW_HASKELL__ >= 800

@@ -225,7 +226,7 @@ where

225import Data.Semigroup (Semigroup(..))

226import Data.Semigroup (Semigroup(..))

226import Streamly.Internal.Data.SVar (MonadAsync, Rate(..))

227import Streamly.Internal.Data.SVar (MonadAsync, Rate(..))

227import Streamly.Internal.Data.Stream.Ahead

228import Streamly.Internal.Data.Stream.Ahead

228−import Streamly.Internal.Data.Stream.Async

229+import Streamly.Internal.Data.Stream.Async hiding (mkAsync)

229import Streamly.Internal.Data.Stream.Combinators

230import Streamly.Internal.Data.Stream.Combinators

230import Streamly.Internal.Data.Stream.Parallel

231import Streamly.Internal.Data.Stream.Parallel

231import Streamly.Internal.Data.Stream.Serial

232import Streamly.Internal.Data.Stream.Serial

@@ -235,6 +236,7 @@ import Streamly.Internal.Data.Stream.Zip

235import qualified Streamly.Prelude as P

236import qualified Streamly.Prelude as P

236import qualified Streamly.Internal.Prelude as IP

237import qualified Streamly.Internal.Prelude as IP

237import qualified Streamly.Internal.Data.Stream.StreamK as K

238import qualified Streamly.Internal.Data.Stream.StreamK as K

239+import qualified Streamly.Internal.Data.Stream.Async as Async

238

240

239-- XXX provide good succinct examples of pipelining, merging, splitting etc.

241-- XXX provide good succinct examples of pipelining, merging, splitting etc.

240-- below.

242-- below.

@@ -451,6 +453,19 @@ forEachWith :: (IsStream t, Foldable f)

451forEachWith = P.forEachWith

453forEachWith = P.forEachWith

452-}

454-}

453

455

456+-- XXX Deprecate it in 0.8.0

457+--

458+-- | Make a stream asynchronous, triggers the computation and returns a stream

459+-- in the underlying monad representing the output generated by the original

460+-- computation. The returned action is exhaustible and must be drained once. If

461+-- not drained fully we may have a thread blocked forever and once exhausted it

462+-- will always return 'empty'.

463+--

464+-- @since 0.2.0

465+{-# INLINABLE mkAsync #-}

466+mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)

467+mkAsync = return . Async.mkAsync

468+

454------------------------------------------------------------------------------

469------------------------------------------------------------------------------

455-- Documentation

470-- Documentation

456------------------------------------------------------------------------------

471------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/Async.hs+41−14

@@ -9,6 +9,8 @@

9{-# LANGUAGE ScopedTypeVariables #-}

10{-# LANGUAGE UndecidableInstances #-} -- XXX

12+#include "inline.hs"

13+

12-- |

14-- |

13-- Module : Streamly.Internal.Data.Stream.Async

15-- Module : Streamly.Internal.Data.Stream.Async

@@ -27,7 +29,7 @@ module Streamly.Internal.Data.Stream.Async

27 , async

29 , async

28 , (<|) --deprecated

30 , (<|) --deprecated

29 , mkAsync

31 , mkAsync

30− , mkAsync'

32+ , mkAsyncK

32 , WAsyncT

34 , WAsyncT

33 , WAsync

35 , WAsync

@@ -494,21 +496,46 @@ newAsyncVar st m = do

494 sv <- liftIO $ getLifoSVar st mrun

496 sv <- liftIO $ getLifoSVar st mrun

495 sendFirstWorker sv m

497 sendFirstWorker sv m

496

498

497−-- XXX Get rid of this?

499+-- | Generate a stream asynchronously to keep it buffered, lazily consume

498−-- | Make a stream asynchronous, triggers the computation and returns a stream

500+-- from the buffer.

499−-- in the underlying monad representing the output generated by the original

500−-- computation. The returned action is exhaustible and must be drained once. If

501−-- not drained fully we may have a thread blocked forever and once exhausted it

502−-- will always return 'empty'.

503--

501--

504−-- @since 0.2.0

502+-- /Internal/

503+--

505−{-# INLINABLE mkAsync #-}

504+{-# INLINABLE mkAsyncK #-}

506−mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)

505+mkAsyncK :: (IsStream t, MonadAsync m) => t m a -> t m a

507−mkAsync = mkAsync' defState

506+mkAsyncK m = mkStream $ \st yld sng stp -> do

507+ sv <- newAsyncVar (adaptState st) (toStream m)

508+ foldStream st yld sng stp $ fromSVar sv

509+

510+{-# INLINE_NORMAL mkAsyncD #-}

511+mkAsyncD :: MonadAsync m => D.Stream m a -> D.Stream m a

512+mkAsyncD m = D.Stream step Nothing

513+ where

508

514

509−{-# INLINABLE mkAsync' #-}

515+ step gst Nothing = do

516+ sv <- newAsyncVar gst (D.fromStreamD m)

517+ return $ D.Skip $ Just $ D.fromSVar sv

518+

519+ step gst (Just (D.UnStream step1 st)) = do

520+ r <- step1 gst st

521+ return $ case r of

522+ D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)

523+ D.Skip s -> D.Skip (Just $ D.Stream step1 s)

524+ D.Stop -> D.Stop

525+

526+-- This is slightly faster than the CPS version above

527+--

528+-- | Make the stream producer and consumer run concurrently by introducing a

529+-- buffer between them. The producer thread evaluates the input stream until

530+-- the buffer fills, it terminates if the buffer is full and a worker thread is

531+-- kicked off again to evaluate the remaining stream when there is space in the

532+-- buffer. The consumer consumes the stream lazily from the buffer.

533+--

534+-- /Internal/

535+--

536+{-# INLINE_NORMAL mkAsync #-}

510−mkAsync' :: (IsStream t, MonadAsync m) => State Stream m a -> t m a -> m (t m a)

537+mkAsync :: (K.IsStream t, MonadAsync m) => t m a -> t m a

511−mkAsync' st m = fmap fromSVar (newAsyncVar st (toStream m))

538+mkAsync = D.fromStreamD . mkAsyncD . D.toStreamD

512

539

513-- | Create a new SVar and enqueue one stream computation on it.

540-- | Create a new SVar and enqueue one stream computation on it.

514{-# INLINABLE newWAsyncVar #-}

541{-# INLINABLE newWAsyncVar #-}

src/Streamly/Internal/Data/Stream/Parallel.hs+9−33

@@ -33,13 +33,10 @@ module Streamly.Internal.Data.Stream.Parallel

34 -- * Evaluate Concurrently

35 , mkParallel

36− , applyParallel

37− , foldParallel

39 -- * Tap Concurrently

37 -- * Tap Concurrently

40 , tapAsync

38 , tapAsync

41 , distributeAsync_

39 , distributeAsync_

42−

43 )

40 )

44where

41where

@@ -248,38 +245,17 @@ parallelMin = joinStreamVarPar ParallelVar StopAny

248-- Convert a stream to parallel

245-- Convert a stream to parallel

249------------------------------------------------------------------------------

246------------------------------------------------------------------------------

250

247

248+-- | Generate a stream asynchronously to keep it buffered, lazily consume

249+-- from the buffer.

250+--

251+-- /Internal/

252+--

251−mkParallel :: (IsStream t, MonadAsync m)

253+mkParallel :: (IsStream t, MonadAsync m) => t m a -> t m a

252− => State Stream m a -> t m a -> m (t m a)

253−mkParallel st m = do

254+mkParallel m = mkStream $ \st yld sng stp -> do

254− sv <- newParallelVar StopNone defState

255+ sv <- newParallelVar StopNone (adaptState st)

255 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)

256 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)

256 D.toSVarParallel st sv $ D.toStreamD m

257 D.toSVarParallel st sv $ D.toStreamD m

257− return $ fromSVar sv

258−

259−------------------------------------------------------------------------------

260−-- Stream to stream concurrent function application

261−------------------------------------------------------------------------------

262−

263−{-# INLINE applyParallel #-}

264−applyParallel :: (IsStream t, MonadAsync m)

265− => (t m a -> t m b) -> t m a -> t m b

266−applyParallel f m = mkStream $ \st yld sng stp -> do

267− sv <- newParallelVar StopNone (adaptState st)

268− -- pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))

269− D.toSVarParallel (adaptState st) sv $ D.toStreamD m

270− foldStream st yld sng stp $ f $ fromSVar sv

258+ foldStream st yld sng stp $ fromSVar sv

271−

272−------------------------------------------------------------------------------

273−-- Stream runner concurrent function application

274−------------------------------------------------------------------------------

275−

276−{-# INLINE foldParallel #-}

277−foldParallel :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b

278−foldParallel f m = do

279− sv <- newParallelVar StopNone defState

280− -- pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)

281− D.toSVarParallel defState sv $ D.toStreamD m

282− f $ fromSVar sv

283

259

284------------------------------------------------------------------------------

260------------------------------------------------------------------------------

285-- Clone and distribute a stream in parallel

261-- Clone and distribute a stream in parallel

src/Streamly/Internal/Data/Stream/StreamD.hs+25−56

@@ -277,8 +277,6 @@ module Streamly.Internal.Data.Stream.StreamD

277 , indexedR

278 , zipWith

279 , zipWithM

280− , zipAsyncWith

281− , zipAsyncWithM

282

280

283 -- * Comparisons

281 -- * Comparisons

284 , eqBy

282 , eqBy

@@ -306,8 +304,6 @@ module Streamly.Internal.Data.Stream.StreamD

306

304

307 -- * Concurrent Application

305 -- * Concurrent Application

308 , mkParallel

306 , mkParallel

309− , applyParallel

310− , foldParallel

311

307

312 , lastN

308 , lastN

313

309

@@ -3857,37 +3853,6 @@ zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)

3857zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

3853zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

3858zipWith f = zipWithM (\a b -> return (f a b))

3854zipWith f = zipWithM (\a b -> return (f a b))

3859

3855

3860−-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped

3861−-- are generated concurrently.

3862−--

3863−{-# INLINE_NORMAL zipAsyncWithM #-}

3864−zipAsyncWithM :: MonadAsync m

3865− => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c

3866−zipAsyncWithM f m1 m2 = Stream step Nothing

3867−

3868− where

3869−

3870− {-# INLINE_LATE step #-}

3871− step gst Nothing = do

3872− ma <- mkParallel (adaptState gst) m1

3873− mb <- mkParallel (adaptState gst) m2

3874− return $ Skip (Just $ zipWithM f ma mb)

3875−

3876− step gst (Just (UnStream step1 st)) = do

3877− r <- step1 gst st

3878− return $ case r of

3879− Yield a s -> Yield a (Just $ Stream step1 s)

3880− Skip s -> Skip (Just $ Stream step1 s)

3881− Stop -> Stop

3882−

3883−-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped

3884−-- are generated concurrently.

3885−--

3886−{-# INLINE zipAsyncWith #-}

3887−zipAsyncWith :: MonadAsync m

3888− => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

3889−zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))

3890−

3891------------------------------------------------------------------------------

3856------------------------------------------------------------------------------

3892-- Merging

3857-- Merging

3893------------------------------------------------------------------------------

3858------------------------------------------------------------------------------

@@ -3968,6 +3933,8 @@ runFold (Fold step begin done) = foldlMx' step begin done

3968-- Concurrent application and fold

3933-- Concurrent application and fold

3969-------------------------------------------------------------------------------

3934-------------------------------------------------------------------------------

3970

3935

3936+-- XXX These functions should be moved to Stream/Parallel.hs

3937+--

3971-- Using StreamD the worker stream producing code can fuse with the code to

3938-- Using StreamD the worker stream producing code can fuse with the code to

3972-- queue output to the SVar giving some perf boost.

3939-- queue output to the SVar giving some perf boost.

3973--

3940--

@@ -4034,24 +4001,16 @@ toSVarParallel st sv xs =

4034 (handleChildException sv)

4001 (handleChildException sv)

4035 modifyThread sv tid

4002 modifyThread sv tid

4036

4003

4037−{-# INLINE_NORMAL mkParallel #-}

4004+{-# INLINE_NORMAL mkParallelD #-}

4038−mkParallel :: MonadAsync m => State t m a -> Stream m a -> m (Stream m a)

4005+mkParallelD :: MonadAsync m => Stream m a -> Stream m a

4039−mkParallel st m = do

4040− sv <- newParallelVar StopNone defState

4041− toSVarParallel st sv m

4042− return $ fromSVar sv

4043−

4044−{-# INLINE_NORMAL applyParallel #-}

4045−applyParallel :: MonadAsync m

4046− => (Stream m a -> Stream m b) -> Stream m a -> Stream m b

4047−applyParallel f m = Stream step Nothing

4006+mkParallelD m = Stream step Nothing

4048 where

4007 where

4049

4008

4050− {-# INLINE_LATE step #-}

4051 step gst Nothing = do

4009 step gst Nothing = do

4052− sv <- newParallelVar StopNone (adaptState gst)

4010+ sv <- newParallelVar StopNone gst

4053− toSVarParallel (adaptState gst) sv m

4011+ toSVarParallel gst sv m

4012+ -- XXX use unfold instead?

4054− return $ Skip $ Just $ f (fromSVar sv)

4013+ return $ Skip $ Just $ fromSVar sv

4055

4014

4056 step gst (Just (UnStream step1 st)) = do

4015 step gst (Just (UnStream step1 st)) = do

4057 r <- step1 gst st

4016 r <- step1 gst st

@@ -4060,12 +4019,22 @@ applyParallel f m = Stream step Nothing

4060 Skip s -> Skip (Just $ Stream step1 s)

4019 Skip s -> Skip (Just $ Stream step1 s)

4061 Stop -> Stop

4020 Stop -> Stop

4062

4021

4063−{-# INLINE foldParallel #-}

4022+-- Compare with mkAsync. mkAsync uses an Async style SVar whereas this uses a

4023+-- parallel style SVar for evaluation. Currently, parallel style cannot use

4024+-- rate control whereas Async style can use rate control. In async style SVar

4025+-- the worker thread terminates when the buffer is full whereas in Parallel

4026+-- style it blocks.

4027+--

4028+-- | Make the stream producer and consumer run concurrently by introducing a

4029+-- buffer between them. The producer thread evaluates the input stream until

4030+-- the buffer fills, it blocks if the buffer is full until there is space in

4031+-- the buffer. The consumer consumes the stream lazily from the buffer.

4032+--

4033+-- /Internal/

4034+--

4035+{-# INLINE_NORMAL mkParallel #-}

4064−foldParallel :: (K.IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b

4036+mkParallel :: (K.IsStream t, MonadAsync m) => t m a -> t m a

4065−foldParallel f m = do

4037+mkParallel = fromStreamD . mkParallelD . toStreamD

4066− sv <- newParallelVar StopNone defState

4067− toSVarParallel defState sv (toStreamD m)

4068− f $ fromStreamD $ fromSVar sv

4069

4038

4070-------------------------------------------------------------------------------

4039-------------------------------------------------------------------------------

4071-- Concurrent tap

4040-- Concurrent tap

src/Streamly/Internal/Data/Stream/Zip.hs+2−16

@@ -56,7 +56,7 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,

56import Prelude hiding (map, repeat, zipWith)

58import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream)

59−import Streamly.Internal.Data.SVar (MonadAsync, adaptState)

59+import Streamly.Internal.Data.SVar (MonadAsync)

61import qualified Streamly.Internal.Data.Stream.Prelude as P

62import qualified Streamly.Internal.Data.Stream.StreamK as K

@@ -93,11 +93,6 @@ zipWith f m1 m2 = P.fromStreamS $ S.zipWith f (P.toStreamS m1) (P.toStreamS m2)

93-- Parallel Zipping

94------------------------------------------------------------------------------

96−-- The CPS version and the direct version of zipAsyncWithM below seem to have

97−-- identical performance. However, we need to use the StreamD version of

98−-- mkParallel which uses a direct implementation of fromSVar. In comparison to

99−-- CPS version of fromSVar the direct version gives a 2x improvement.

100−

101-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped

96-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped

102-- are generated concurrently.

97-- are generated concurrently.

103--

98--

@@ -105,16 +100,7 @@ zipWith f m1 m2 = P.fromStreamS $ S.zipWith f (P.toStreamS m1) (P.toStreamS m2)

105{-# INLINE zipAsyncWithM #-}

100{-# INLINE zipAsyncWithM #-}

106zipAsyncWithM :: (IsStream t, MonadAsync m)

101zipAsyncWithM :: (IsStream t, MonadAsync m)

107 => (a -> b -> m c) -> t m a -> t m b -> t m c

102 => (a -> b -> m c) -> t m a -> t m b -> t m c

108−zipAsyncWithM f m1 m2 = K.mkStream $ \st stp sng yld -> do

103+zipAsyncWithM f m1 m2 = zipWithM f (D.mkParallel m1) (D.mkParallel m2)

109− ma <- D.mkParallel (adaptState st) (D.toStreamD m1)

110− mb <- D.mkParallel (adaptState st) (D.toStreamD m2)

111− K.foldStream st stp sng yld $ zipWithM f (D.fromStreamD ma)

112− (D.fromStreamD mb)

113−{-

114−zipAsyncWithM f m1 m2 =

115− fromStreamD $ D.zipAsyncWithM f (toStreamD m1) (toStreamD m2)

116−-}

117−

118-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped

104-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped

119-- are generated concurrently.

105-- are generated concurrently.

120--

106--

src/Streamly/Internal/Prelude.hs+12−33

@@ -172,6 +172,8 @@ module Streamly.Internal.Prelude

172 -- , lprescanlM'

173

174 -- ** Concurrent Transformation

175+ , D.mkParallel

176+ -- Par.mkParallel

175 , applyAsync

177 , applyAsync

176 , (|$)

178 , (|$)

177 , (|&)

179 , (|&)

@@ -413,9 +415,6 @@ module Streamly.Internal.Prelude

413 , usingStateT

415 , usingStateT

414 , runStateT

416 , runStateT

415

417

416− -- * Concurrency

417− , mkParallel

418−

419 -- * MonadFix

418 -- * MonadFix

420 , K.mfix

419 , K.mfix

421

420

@@ -468,8 +467,7 @@ import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))

468import Streamly.Internal.Data.Unfold.Types (Unfold)

467import Streamly.Internal.Data.Unfold.Types (Unfold)

469import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)

468import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)

470-- import Streamly.Memory.Ring (Ring)

469-- import Streamly.Memory.Ring (Ring)

471−import Streamly.Internal.Data.SVar (MonadAsync, State, defState)

470+import Streamly.Internal.Data.SVar (MonadAsync, defState)

472−import Streamly.Internal.Data.Stream.Async (mkAsync')

473import Streamly.Internal.Data.Stream.Combinators (inspectMode, maxYields)

471import Streamly.Internal.Data.Stream.Combinators (inspectMode, maxYields)

474import Streamly.Internal.Data.Stream.Prelude

472import Streamly.Internal.Data.Stream.Prelude

475 (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)

473 (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)

@@ -498,6 +496,7 @@ import qualified Streamly.Internal.Data.Stream.StreamK as S

498import qualified Streamly.Internal.Data.Stream.StreamD as S

496import qualified Streamly.Internal.Data.Stream.StreamD as S

499#endif

497#endif

500

498

499+-- import qualified Streamly.Internal.Data.Stream.Async as Async

501import qualified Streamly.Internal.Data.Stream.Serial as Serial

500import qualified Streamly.Internal.Data.Stream.Serial as Serial

502import qualified Streamly.Internal.Data.Stream.Parallel as Par

501import qualified Streamly.Internal.Data.Stream.Parallel as Par

503import qualified Streamly.Internal.Data.Stream.Zip as Z

502import qualified Streamly.Internal.Data.Stream.Zip as Z

@@ -1541,19 +1540,6 @@ toPureRev = foldl' (flip K.cons) K.nil

1541-- Concurrent Application

1540-- Concurrent Application

1542------------------------------------------------------------------------------

1541------------------------------------------------------------------------------

1543

1542

1544−-- XXX mkParallel is usually used to make serial streams concurrent by

1545−-- evaluating the stream in another thread. Ideally, we should not be creating

1546−-- an SVar if the stream itself is concurrent. For example, if we are zipping

1547−-- two streams that are concurrently composed, we should be able to pull from

1548−-- the SVar of that concurrent stream instead of creating another redundant

1549−-- SVar on top of it.

1550−--

1551−{-# INLINE_NORMAL mkParallel #-}

1552−mkParallel :: (IsStream t, MonadAsync m)

1553− => State K.Stream m a -> t m a -> m (t m a)

1554−mkParallel st m = fmap fromStreamD $ D.mkParallel st (toStreamD m)

1555−-- mkParallel = Par.mkParallel

1556−

1557infixr 0 |$

1543infixr 0 |$

1558infixr 0 |$.

1544infixr 0 |$.

1559

1545

@@ -1586,11 +1572,10 @@ infixl 1 |&.

1586-- @since 0.3.0

1572-- @since 0.3.0

1587{-# INLINE (|$) #-}

1573{-# INLINE (|$) #-}

1588(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)

1574(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)

1589−f |$ x = D.fromStreamD $

1575+-- (|$) f = f . Async.mkAsync

1590− D.applyParallel (D.toStreamD . f . D.fromStreamD) (D.toStreamD x)

1576+(|$) f = f . D.mkParallel

1591−-- (|$) = Par.applyParallel

1592

1577

1593−-- | Same as (|$).

1578+-- | Same as '|$'.

1594--

1579--

1595-- /Internal/

1580-- /Internal/

1596--

1581--

@@ -1639,10 +1624,10 @@ x |& f = f |$ x

1639-- @since 0.3.0

1624-- @since 0.3.0

1640{-# INLINE (|$.) #-}

1625{-# INLINE (|$.) #-}

1641(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)

1626(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)

1642−(|$.) = D.foldParallel

1627+-- (|$.) f = f . Async.mkAsync

1643−-- (|$.) = Par.foldParallel

1628+(|$.) f = f . D.mkParallel

1644

1629

1645−-- | Same as '(|$.)'.

1630+-- | Same as '|$.'.

1646--

1631--

1647-- /Internal/

1632-- /Internal/

1648--

1633--

@@ -2404,10 +2389,7 @@ merge = mergeBy compare

2404-- @since 0.6.0

2389-- @since 0.6.0

2405mergeAsyncBy :: (IsStream t, MonadAsync m)

2390mergeAsyncBy :: (IsStream t, MonadAsync m)

2406 => (a -> a -> Ordering) -> t m a -> t m a -> t m a

2391 => (a -> a -> Ordering) -> t m a -> t m a -> t m a

2407−mergeAsyncBy f m1 m2 = K.mkStream $ \st stp sng yld -> do

2392+mergeAsyncBy f m1 m2 = K.mergeBy f (D.mkParallel m1) (D.mkParallel m2)

2408− ma <- mkAsync' st m1

2409− mb <- mkAsync' st m2

2410− K.foldStream st stp sng yld (K.mergeBy f ma mb)

2411

2393

2412-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being

2394-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being

2413-- merged are generated concurrently).

2395-- merged are generated concurrently).

@@ -2415,10 +2397,7 @@ mergeAsyncBy f m1 m2 = K.mkStream $ \st stp sng yld -> do

2415-- @since 0.6.0

2397-- @since 0.6.0

2416mergeAsyncByM :: (IsStream t, MonadAsync m)

2398mergeAsyncByM :: (IsStream t, MonadAsync m)

2417 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a

2399 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a

2418−mergeAsyncByM f m1 m2 = K.mkStream $ \st stp sng yld -> do

2400+mergeAsyncByM f m1 m2 = K.mergeByM f (D.mkParallel m1) (D.mkParallel m2)

2419− ma <- mkAsync' st m1

2420− mb <- mkAsync' st m2

2421− K.foldStream st stp sng yld (K.mergeByM f ma mb)

2422

2401

2423------------------------------------------------------------------------------

2402------------------------------------------------------------------------------

2424-- Nesting

2403-- Nesting