a67e688 · streamly · git.iko.soy

git.iko.soy/streamly

Use StreamD version of mergeBy in async merge

Now that we have StreamD version of mkParallel we should be able to use StreamD version of merge efficiently. Performance improved by 6x.

commita67e688c5d8fa059dea1a811d68abaccd9f11801 authorHarendra Kumar <harendra.kumar@gmail.com> dateFri 24 Jan 2020, 01:12 (7 years ago) parenta285b55

4 files changed+15−2

compare with master

unifiedsplit

benchmark/LinearAsync.hs+2 benchmark/Streamly/Benchmark/Prelude.hs+7 src/Streamly/Internal/Data/Stream/StreamD.hs+1 src/Streamly/Internal/Prelude.hs+5−2

benchmark/LinearAsync.hs+2

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

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

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

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

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

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

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

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

141143 -- Parallel stages in a pipeline

benchmark/Streamly/Benchmark/Prelude.hs+7

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

744744 (sourceUnfoldrMN count n)

745745 (sourceUnfoldrMN count (n + 1))

746746

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+

747754-------------------------------------------------------------------------------

748755-- Multi-stream folds

749756-------------------------------------------------------------------------------

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

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

304304

305305 -- * Concurrent Application

306306 , mkParallel

307+ , mkParallelD

307308

308309 , lastN

309310

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

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

23872387-- merged are generated concurrently).

23882388--

23892389-- @since 0.6.0

2390+{-# INLINE mergeAsyncBy #-}

23902391mergeAsyncBy :: (IsStream t, MonadAsync m)

23912392 => (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)

23932394

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

23952396-- merged are generated concurrently).

23962397--

23972398-- @since 0.6.0

2399+{-# INLINE mergeAsyncByM #-}

23982400mergeAsyncByM :: (IsStream t, MonadAsync m)

23992401 => (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)

24012404

24022405------------------------------------------------------------------------------

24032406-- Nesting