@@ -277,8 +277,6 @@ module Streamly.Internal.Data.Stream.StreamD
277277 , indexedR
278278 , zipWith
279279 , zipWithM
280− , zipAsyncWith
281− , zipAsyncWithM
282280
283281 -- * Comparisons
284282 , eqBy
@@ -306,8 +304,6 @@ module Streamly.Internal.Data.Stream.StreamD
306304
307305 -- * Concurrent Application
308306 , mkParallel
309− , applyParallel
310− , foldParallel
311307
312308 , lastN
313309
@@ -3857,37 +3853,6 @@ zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)
38573853zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
38583854zipWith f = zipWithM (\a b -> return (f a b))
38593855
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−
38913856------------------------------------------------------------------------------
38923857-- Merging
38933858------------------------------------------------------------------------------
@@ -3968,6 +3933,8 @@ runFold (Fold step begin done) = foldlMx' step begin done
39683933-- Concurrent application and fold
39693934-------------------------------------------------------------------------------
39703935
3936+-- XXX These functions should be moved to Stream/Parallel.hs
3937+--
39713938-- Using StreamD the worker stream producing code can fuse with the code to
39723939-- queue output to the SVar giving some perf boost.
39733940--
@@ -4034,24 +4001,16 @@ toSVarParallel st sv xs =
40344001 (handleChildException sv)
40354002 modifyThread sv tid
40364003
4037−{-# INLINE_NORMAL mkParallel #-}
4038−mkParallel :: MonadAsync m => State t m a -> Stream m a -> m (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
4004+{-# INLINE_NORMAL mkParallelD #-}
4005+mkParallelD :: MonadAsync m => Stream m a -> Stream m a
4006+mkParallelD m = Stream step Nothing
40484007 where
40494008
4050− {-# INLINE_LATE step #-}
40514009 step gst Nothing = do
4052− sv <- newParallelVar StopNone (adaptState gst)
4053− toSVarParallel (adaptState gst) sv m
4054− return $ Skip $ Just $ f (fromSVar sv)
4010+ sv <- newParallelVar StopNone gst
4011+ toSVarParallel gst sv m
4012+ -- XXX use unfold instead?
4013+ return $ Skip $ Just $ fromSVar sv
40554014
40564015 step gst (Just (UnStream step1 st)) = do
40574016 r <- step1 gst st
@@ -4060,12 +4019,22 @@ applyParallel f m = Stream step Nothing
40604019 Skip s -> Skip (Just $ Stream step1 s)
40614020 Stop -> Stop
40624021
4063−{-# INLINE foldParallel #-}
4064−foldParallel :: (K.IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
4065−foldParallel f m = do
4066− sv <- newParallelVar StopNone defState
4067− toSVarParallel defState sv (toStreamD m)
4068− f $ fromStreamD $ fromSVar sv
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 #-}
4036+mkParallel :: (K.IsStream t, MonadAsync m) => t m a -> t m a
4037+mkParallel = fromStreamD . mkParallelD . toStreamD
40694038
40704039-------------------------------------------------------------------------------
40714040-- Concurrent tap