a67e688 · streamly · git.iko.soy

benchmark/LinearAsync.hs+2

compare with master

unifiedsplit

@@ -136,6 +136,8 @@ main = do

136 , benchSrcIO serially "zipAsyncAp (2,x/2)"

137 (Ops.zipAsyncAp (value `div` 2))

138 , benchIO value "fmap zipAsyncly" $ Ops.fmap' zipAsyncly 1

139+ , benchSrcIO serially "mergeAsyncBy (2,x/2)"

140+ (Ops.mergeAsyncBy (value `div` 2))

139 , benchSrcIO serially "mergeAsyncByM (2,x/2)"

141 , benchSrcIO serially "mergeAsyncByM (2,x/2)"

140 (Ops.mergeAsyncByM (value `div` 2))

142 (Ops.mergeAsyncByM (value `div` 2))

141 -- Parallel stages in a pipeline

143 -- Parallel stages in a pipeline

benchmark/Streamly/Benchmark/Prelude.hs+7

@@ -744,6 +744,13 @@ mergeAsyncByM count n = do

744 (sourceUnfoldrMN count n)

745 (sourceUnfoldrMN count (n + 1))

746

747+{-# INLINE mergeAsyncBy #-}

748+mergeAsyncBy :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int

749+mergeAsyncBy count n = do

750+ S.mergeAsyncBy compare

751+ (sourceUnfoldrMN count n)

752+ (sourceUnfoldrMN count (n + 1))

753+

747-------------------------------------------------------------------------------

754-------------------------------------------------------------------------------

748-- Multi-stream folds

755-- Multi-stream folds

749-------------------------------------------------------------------------------

756-------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/StreamD.hs+1

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

304

305 -- * Concurrent Application

306 , mkParallel

307+ , mkParallelD

307

308

308 , lastN

309 , lastN

309

310

src/Streamly/Internal/Prelude.hs+5−2

@@ -2387,17 +2387,20 @@ merge = mergeBy compare

2387-- merged are generated concurrently).

2388--

2389-- @since 0.6.0

2390+{-# INLINE mergeAsyncBy #-}

2390mergeAsyncBy :: (IsStream t, MonadAsync m)

2391mergeAsyncBy :: (IsStream t, MonadAsync m)

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

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

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

2393+mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)

2393

2394

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

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

2395-- merged are generated concurrently).

2396-- merged are generated concurrently).

2396--

2397--

2397-- @since 0.6.0

2398-- @since 0.6.0

2399+{-# INLINE mergeAsyncByM #-}

2398mergeAsyncByM :: (IsStream t, MonadAsync m)

2400mergeAsyncByM :: (IsStream t, MonadAsync m)

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

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

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

2402+mergeAsyncByM f m1 m2 = fromStreamD $

2403+ D.mergeByM f (D.mkParallelD $ toStreamD m1) (D.mkParallelD $ toStreamD m2)

2401

2404

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

2405------------------------------------------------------------------------------

2403-- Nesting

2406-- Nesting