git.iko.soy/streamly
on master help sign in
Use StreamD for parallel zip
36bfc45Harendra Kumar
src/Streamly/Streams/Parallel.hs+12−2
compare with master
unifiedsplit
@@ -276,12 +276,22 @@ parallelMin = joinStreamVarPar ParallelVar StopAny
276-- Convert a stream to parallel
276-- Convert a stream to parallel
277------------------------------------------------------------------------------
277------------------------------------------------------------------------------
278
278
279−mkParallel :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
279+_mkParallelK :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
280−mkParallel m = do
280+_mkParallelK m = do
281 sv <- newParallelVar StopNone defState
281 sv <- newParallelVar StopNone defState
282 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
282 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
283 return $ fromSVar sv
283 return $ fromSVar sv
284
284
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+
285------------------------------------------------------------------------------
295------------------------------------------------------------------------------
286-- Stream to stream concurrent function application
296-- Stream to stream concurrent function application
287------------------------------------------------------------------------------
297------------------------------------------------------------------------------
src/Streamly/Streams/Zip.hs+5−5
@@ -54,7 +54,7 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
54import Prelude hiding (map, repeat, zipWith)
54import Prelude hiding (map, repeat, zipWith)
55
55
56import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream)
56import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream)
57−import Streamly.Streams.Async (mkAsync')
57+import Streamly.Streams.Parallel (mkParallel)
58import Streamly.Streams.Serial (map)
58import Streamly.Streams.Serial (map)
59import Streamly.Internal.Data.SVar (MonadAsync, adaptState)
59import Streamly.Internal.Data.SVar (MonadAsync, adaptState)
60
60
@@ -153,8 +153,8 @@ TRAVERSABLE_INSTANCE(ZipSerialM)
153zipAsyncWith :: (IsStream t, MonadAsync m)
153zipAsyncWith :: (IsStream t, MonadAsync m)
154 => (a -> b -> c) -> t m a -> t m b -> t m c
154 => (a -> b -> c) -> t m a -> t m b -> t m c
155zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do
155zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do
156− ma <- mkAsync' (adaptState st) m1
156+ ma <- mkParallel (adaptState st) m1
157− mb <- mkAsync' (adaptState st) m2
157+ mb <- mkParallel (adaptState st) m2
158 foldStream st stp sng yld (K.zipWith f ma mb)
158 foldStream st stp sng yld (K.zipWith f ma mb)
159
159
160-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
160-- | 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
165zipAsyncWithM :: (IsStream t, MonadAsync m)
165zipAsyncWithM :: (IsStream t, MonadAsync m)
166 => (a -> b -> m c) -> t m a -> t m b -> t m c
166 => (a -> b -> m c) -> t m a -> t m b -> t m c
167zipAsyncWithM f m1 m2 = mkStream $ \st stp sng yld -> do
167zipAsyncWithM f m1 m2 = mkStream $ \st stp sng yld -> do
168− ma <- mkAsync' (adaptState st) m1
168+ ma <- mkParallel (adaptState st) m1
169− mb <- mkAsync' (adaptState st) m2
169+ mb <- mkParallel (adaptState st) m2
170 foldStream st stp sng yld (K.zipWithM f ma mb)
170 foldStream st stp sng yld (K.zipWithM f ma mb)
171
171
172------------------------------------------------------------------------------
172------------------------------------------------------------------------------