git.iko.soy/streamly
on master help sign in
Use StreamD for parallel zip

improves concurrent zip perf by 3x

commit36bfc45842833df15cf3054f78763ac7c8e40adb authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 21 Nov 2019, 03:23 (7 years ago) parent2b31b3d
2 files changed+17−7
compare with master
unifiedsplit
src/Streamly/Streams/Parallel.hs+12−2 src/Streamly/Streams/Zip.hs+5−5
src/Streamly/Streams/Parallel.hs+12−2
@@ -276,12 +276,22 @@ parallelMin = joinStreamVarPar ParallelVar StopAny
276276-- Convert a stream to parallel
277277------------------------------------------------------------------------------
278278
279−mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
280−mkParallel m = do
279+_mkParallelK :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
280+_mkParallelK m = do
281281 sv <- newParallelVar StopNone defState
282282 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
283283 return $ fromSVar sv
284284
285+-- We may have to use it in higher order functions like concatMap, so use the
286+-- Normal phase for inlining.
287+{-# INLINE_NORMAL mkParallel #-}
288+mkParallel :: (IsStream t, MonadAsync m)
289+ => State Stream m a -> t m a -> m (t m a)
290+mkParallel st m = do
291+ sv <- newParallelVar StopNone defState
292+ pushWorkerParD st sv (D.toStreamD m)
293+ return $ fromSVar sv
294+
285295------------------------------------------------------------------------------
286296-- Stream to stream concurrent function application
287297------------------------------------------------------------------------------
src/Streamly/Streams/Zip.hs+5−5
@@ -54,7 +54,7 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
5454import Prelude hiding (map, repeat, zipWith)
5555
5656import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream)
57−import Streamly.Streams.Async (mkAsync')
57+import Streamly.Streams.Parallel (mkParallel)
5858import Streamly.Streams.Serial (map)
5959import Streamly.Internal.Data.SVar (MonadAsync, adaptState)
6060
@@ -153,8 +153,8 @@ TRAVERSABLE_INSTANCE(ZipSerialM)
153153zipAsyncWith :: (IsStream t, MonadAsync m)
154154 => (a -> b -> c) -> t m a -> t m b -> t m c
155155zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do
156− ma <- mkAsync' (adaptState st) m1
157− mb <- mkAsync' (adaptState st) m2
156+ ma <- mkParallel (adaptState st) m1
157+ mb <- mkParallel (adaptState st) m2
158158 foldStream st stp sng yld (K.zipWith f ma mb)
159159
160160-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
@@ -165,8 +165,8 @@ zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do
165165zipAsyncWithM :: (IsStream t, MonadAsync m)
166166 => (a -> b -> m c) -> t m a -> t m b -> t m c
167167zipAsyncWithM f m1 m2 = mkStream $ \st stp sng yld -> do
168− ma <- mkAsync' (adaptState st) m1
169− mb <- mkAsync' (adaptState st) m2
168+ ma <- mkParallel (adaptState st) m1
169+ mb <- mkParallel (adaptState st) m2
170170 foldStream st stp sng yld (K.zipWithM f ma mb)
171171
172172------------------------------------------------------------------------------