315abcf · streamly · git.iko.soy

git.iko.soy/streamly

Fix some state sharing/clearing safety issues

1. removed redundant rstState from mergeAsyncBy and mergeByS 2. renamed rstState to adaptState, as it is now used only coerce one state type to another. 3. renamed isolateStream to unShare 4. made StreamD as well safer by clearing the SVar sharing during pattern match

commit315abcf41039d570cd63373b9e1257695e50ba79 authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 10 Dec 2018, 21:18 (8 years ago) parent769ce84

8 files changed+95−67

compare with master

unifiedsplit

src/Streamly/Prelude.hs+8−8 src/Streamly/SVar.hs+10−3 src/Streamly/Streams/Ahead.hs+1−1 src/Streamly/Streams/Async.hs+3−2 src/Streamly/Streams/Parallel.hs+1−1 src/Streamly/Streams/StreamD.hs+52−34 src/Streamly/Streams/StreamK.hs+15−13 src/Streamly/Streams/Zip.hs+5−5

src/Streamly/Prelude.hs+8−8

@@ -363,7 +363,7 @@ import Prelude

363363import qualified Prelude

364364import qualified System.IO as IO

365365

366−import Streamly.SVar (MonadAsync, defState, rstState)

366+import Streamly.SVar (MonadAsync, defState)

367367import Streamly.Streams.Async (mkAsync')

368368import Streamly.Streams.Combinators (maxYields)

369369import Streamly.Streams.Prelude (fromStreamS, toStreamS)

@@ -1449,7 +1449,7 @@ toHandle h m = go (toStream m)

14491449 let stop = return ()

14501450 single a = liftIO (IO.hPutStrLn h a)

14511451 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r

1452− in K.unStreamShared m1 defState stop single yieldk

1452+ in K.unStream m1 defState stop single yieldk

14531453

14541454------------------------------------------------------------------------------

14551455-- Transformation by Folding (Scans)

@@ -1987,9 +1987,9 @@ merge = mergeBy compare

19871987mergeAsyncBy :: (IsStream t, MonadAsync m)

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

19891989mergeAsyncBy f m1 m2 = K.fromStream $ K.mkStream $ \st stp sng yld -> do

1990− ma <- mkAsync' (rstState st) m1

1991− mb <- mkAsync' (rstState st) m2

1992− K.unStream (K.toStream (K.mergeBy f ma mb)) (rstState st) stp sng yld

1990+ ma <- mkAsync' st m1

1991+ mb <- mkAsync' st m2

1992+ K.unStream (K.toStream (K.mergeBy f ma mb)) st stp sng yld

19931993

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

19951995-- merged are generated concurrently).

@@ -1998,6 +1998,6 @@ mergeAsyncBy f m1 m2 = K.fromStream $ K.mkStream $ \st stp sng yld -> do

19981998mergeAsyncByM :: (IsStream t, MonadAsync m)

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

20002000mergeAsyncByM f m1 m2 = K.fromStream $ K.mkStream $ \st stp sng yld -> do

2001− ma <- mkAsync' (rstState st) m1

2002− mb <- mkAsync' (rstState st) m2

2003− K.unStream (K.toStream (K.mergeByM f ma mb)) (rstState st) stp sng yld

2001+ ma <- mkAsync' st m1

2002+ mb <- mkAsync' st m2

2003+ K.unStream (K.toStream (K.mergeByM f ma mb)) st stp sng yld

src/Streamly/SVar.hs+10−3

@@ -31,7 +31,7 @@ module Streamly.SVar

3131 , Limit (..)

3232 , State (streamVar)

3333 , defState

34− , rstState

34+ , adaptState

3535 , getMaxThreads

3636 , setMaxThreads

3737 , getMaxBuffer

@@ -433,8 +433,15 @@ defState = State

433433-- We can optimize this so that we clear it only if it is a Just value, it

434434-- results in slightly better perf for zip/zipM but the performance of scan

435435-- worsens a lot, it does not fuse.

436−rstState :: State t m a -> State t m b

437−rstState st = st

436+--

437+-- This has a side effect of clearing the SVar and yieldLimit, therefore it

438+-- should not be used to convert from the same type to the same type, unless

439+-- you want to clear the SVar. For clearing the SVar you should be using the

440+-- appropriate unStream functions instead.

441+--

442+-- | Adapt the stream state from one type to another.

443+adaptState :: State t m a -> State t m b

444+adaptState st = st

438445 { streamVar = Nothing

439446 , _yieldLimit = Nothing

440447 }

src/Streamly/Streams/Ahead.hs+1−1

@@ -688,7 +688,7 @@ aheadbind m f = go m

688688

689689 single a = runIsolated $ f a

690690 yieldk a r = runShared $

691− K.isolateStream (f a) `aheadS` go r

691+ K.unShare (f a) `aheadS` go r

692692 in unStream g st stp single yieldk

693693

694694instance MonadAsync m => Monad (AheadT m) where

src/Streamly/Streams/Async.hs+3−2

@@ -57,8 +57,9 @@ import qualified Data.Set as S

5757import Streamly.Streams.SVar (fromSVar)

5858import Streamly.Streams.Serial (map)

5959import Streamly.SVar

60−import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, unStream, adapt,

61− unStreamShared, runStreamSVar)

60+import Streamly.Streams.StreamK

61+ (IsStream(..), Stream, mkStream, unStream, adapt, unStreamShared,

62+ runStreamSVar)

6263import qualified Streamly.Streams.StreamK as K

6364

6465#include "Instances.hs"

src/Streamly/Streams/Parallel.hs+1−1

@@ -142,7 +142,7 @@ mkParallel m = do

142142{-# INLINE applyWith #-}

143143applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b

144144applyWith f m = fromStream $ mkStream $ \st stp sng yld -> do

145− sv <- newParallelVar (rstState st)

145+ sv <- newParallelVar (adaptState st)

146146 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))

147147 unStream (toStream $ f $ fromSVar sv) st stp sng yld

148148

src/Streamly/Streams/StreamD.hs+52−34

@@ -5,6 +5,8 @@

55{-# LANGUAGE FlexibleContexts #-}

66{-# LANGUAGE FlexibleInstances #-}

77{-# LANGUAGE MultiParamTypeClasses #-}

8+{-# LANGUAGE PatternSynonyms #-}

9+{-# LANGUAGE ViewPatterns #-}

810{-# LANGUAGE RankNTypes #-}

911

1012#include "inline.hs"

@@ -197,7 +199,7 @@ import Prelude

197199 notElem, null, head, tail, zipWith, lookup, foldr1, sequence,

198200 (!!), scanl, scanl1, concatMap, replicate, enumFromTo)

199201

200−import Streamly.SVar (MonadAsync, State(..), defState, rstState)

202+import Streamly.SVar (MonadAsync, State(..), defState, adaptState)

201203

202204import qualified Streamly.Streams.StreamK as K

203205

@@ -219,7 +221,24 @@ instance Functor (Step s) where

219221-- gst = global state

220222-- | A stream consists of a step function that generates the next step given a

221223-- current state, and the current state.

222−data Stream m a = forall s. Stream (State K.Stream m a -> s -> m (Step s a)) s

224+data Stream m a =

225+ forall s. UnStream (State K.Stream m a -> s -> m (Step s a)) s

226+

227+-- Do not use the UnStream constructor directly, use the pattern synonym

228+-- instead. XXX we should perhaps put the type in a separate module and do not

229+-- export the constructor.

230+

231+detachStream :: Stream m a -> Stream m a

232+detachStream (UnStream step state) = UnStream step' state

233+ where step' gst = step (adaptState gst)

234+

235+pattern Stream :: (State K.Stream m a -> s -> m (Step s a)) -> s -> Stream m a

236+pattern Stream step state <- (detachStream -> UnStream step state)

237+ where Stream = UnStream

238+

239+#if __GLASGOW_HASKELL__ >= 802

240+{-# COMPLETE Stream #-}

241+#endif

223242

224243------------------------------------------------------------------------------

225244-- Construction

@@ -238,7 +257,7 @@ cons x (Stream step state) = Stream step1 Nothing

238257 {-# INLINE_LATE step1 #-}

239258 step1 _ Nothing = return $ Yield x (Just state)

240259 step1 gst (Just st) = do

241− r <- step (rstState gst) st

260+ r <- step gst st

242261 return $

243262 case r of

244263 Yield a s -> Yield a (Just s)

@@ -492,7 +511,6 @@ fromList = Stream step

492511 step _ (x:xs) = return $ Yield x xs

493512 step _ [] = return Stop

494513

495−-- XXX pass the state to streamD

496514{-# INLINE_LATE fromStreamK #-}

497515fromStreamK :: Monad m => K.Stream m a -> Stream m a

498516fromStreamK = Stream step

@@ -773,7 +791,7 @@ findIndices p (Stream step state) = Stream step' (state, 0)

773791 where

774792 {-# INLINE_LATE step' #-}

775793 step' gst (st, i) = do

776− r <- step (rstState gst) st

794+ r <- step (adaptState gst) st

777795 return $ case r of

778796 Yield x s -> if p x then Yield i (s, i+1) else Skip (s, i+1)

779797 Skip s -> Skip (s, i+1)

@@ -785,7 +803,7 @@ concatMapM f (Stream step state) = Stream step' (Left state)

785803 where

786804 {-# INLINE_LATE step' #-}

787805 step' gst (Left st) = do

788− r <- step (rstState gst) st

806+ r <- step (adaptState gst) st

789807 case r of

790808 Yield a s -> do

791809 b_stream <- f a

@@ -935,7 +953,7 @@ prescanlM' f mz (Stream step state) = Stream step' (state, mz)

935953 where

936954 {-# INLINE_LATE step' #-}

937955 step' gst (st, prev) = do

938− r <- step (rstState gst) st

956+ r <- step (adaptState gst) st

939957 case r of

940958 Yield x s -> do

941959 acc <- prev

@@ -963,7 +981,7 @@ postscanlM' fstep begin (Stream step state) =

963981 where

964982 {-# INLINE_LATE step' #-}

965983 step' gst (st, acc) = acc `seq` do

966− r <- step (rstState gst) st

984+ r <- step (adaptState gst) st

967985 case r of

968986 Yield x s -> do

969987 y <- fstep acc x

@@ -981,7 +999,7 @@ postscanlM fstep begin (Stream step state) = Stream step' (state, begin)

981999 where

9821000 {-# INLINE_LATE step' #-}

9831001 step' gst (st, acc) = do

984− r <- step (rstState gst) st

1002+ r <- step (adaptState gst) st

9851003 case r of

9861004 Yield x s -> do

9871005 y <- fstep acc x

@@ -1015,14 +1033,14 @@ scanl1M fstep (Stream step state) = Stream step' (state, Nothing)

10151033 where

10161034 {-# INLINE_LATE step' #-}

10171035 step' gst (st, Nothing) = do

1018− r <- step (rstState gst) st

1036+ r <- step gst st

10191037 case r of

10201038 Yield x s -> return $ Yield x (s, Just x)

10211039 Skip s -> return $ Skip (s, Nothing)

10221040 Stop -> return Stop

10231041

10241042 step' gst (st, Just acc) = do

1025− r <- step (rstState gst) st

1043+ r <- step gst st

10261044 case r of

10271045 Yield y s -> do

10281046 z <- fstep acc y

@@ -1040,14 +1058,14 @@ scanl1M' fstep (Stream step state) = Stream step' (state, Nothing)

10401058 where

10411059 {-# INLINE_LATE step' #-}

10421060 step' gst (st, Nothing) = do

1043− r <- step (rstState gst) st

1061+ r <- step gst st

10441062 case r of

10451063 Yield x s -> x `seq` return $ Yield x (s, Just x)

10461064 Skip s -> return $ Skip (s, Nothing)

10471065 Stop -> return Stop

10481066

10491067 step' gst (st, Just acc) = acc `seq` do

1050− r <- step (rstState gst) st

1068+ r <- step gst st

10511069 case r of

10521070 Yield y s -> do

10531071 z <- fstep acc y

@@ -1069,7 +1087,7 @@ take n (Stream step state) = n `seq` Stream step' (state, 0)

10691087 where

10701088 {-# INLINE_LATE step' #-}

10711089 step' gst (st, i) | i < n = do

1072− r <- step (rstState gst) st

1090+ r <- step gst st

10731091 return $ case r of

10741092 Yield x s -> Yield x (s, i + 1)

10751093 Skip s -> Skip (s, i)

@@ -1082,7 +1100,7 @@ takeWhileM f (Stream step state) = Stream step' state

10821100 where

10831101 {-# INLINE_LATE step' #-}

10841102 step' gst st = do

1085− r <- step (rstState gst) st

1103+ r <- step gst st

10861104 case r of

10871105 Yield x s -> do

10881106 b <- f x

@@ -1101,7 +1119,7 @@ drop n (Stream step state) = Stream step' (state, Just n)

11011119 {-# INLINE_LATE step' #-}

11021120 step' gst (st, Just i)

11031121 | i > 0 = do

1104− r <- step (rstState gst) st

1122+ r <- step gst st

11051123 return $

11061124 case r of

11071125 Yield _ s -> Skip (s, Just (i - 1))

@@ -1110,7 +1128,7 @@ drop n (Stream step state) = Stream step' (state, Just n)

11101128 | otherwise = return $ Skip (st, Nothing)

11111129

11121130 step' gst (st, Nothing) = do

1113− r <- step (rstState gst) st

1131+ r <- step gst st

11141132 return $

11151133 case r of

11161134 Yield x s -> Yield x (s, Nothing)

@@ -1128,7 +1146,7 @@ dropWhileM f (Stream step state) = Stream step' (DropWhileDrop state)

11281146 where

11291147 {-# INLINE_LATE step' #-}

11301148 step' gst (DropWhileDrop st) = do

1131− r <- step (rstState gst) st

1149+ r <- step gst st

11321150 case r of

11331151 Yield x s -> do

11341152 b <- f x

@@ -1139,7 +1157,7 @@ dropWhileM f (Stream step state) = Stream step' (DropWhileDrop state)

11391157 Stop -> return Stop

11401158

11411159 step' gst (DropWhileNext st) = do

1142− r <- step (rstState gst) st

1160+ r <- step gst st

11431161 case r of

11441162 Yield x s -> return $ Skip (DropWhileYield x s)

11451163 Skip s -> return $ Skip (DropWhileNext s)

@@ -1157,7 +1175,7 @@ filterM f (Stream step state) = Stream step' state

11571175 where

11581176 {-# INLINE_LATE step' #-}

11591177 step' gst st = do

1160− r <- step (rstState gst) st

1178+ r <- step gst st

11611179 case r of

11621180 Yield x s -> do

11631181 b <- f x

@@ -1177,13 +1195,13 @@ uniq (Stream step state) = Stream step' (Nothing, state)

11771195 where

11781196 {-# INLINE_LATE step' #-}

11791197 step' gst (Nothing, st) = do

1180− r <- step (rstState gst) st

1198+ r <- step gst st

11811199 case r of

11821200 Yield x s -> return $ Yield x (Just x, s)

11831201 Skip s -> return $ Skip (Nothing, s)

11841202 Stop -> return Stop

11851203 step' gst (Just x, st) = do

1186− r <- step (rstState gst) st

1204+ r <- step gst st

11871205 case r of

11881206 Yield y s | x == y -> return $ Skip (Just x, s)

11891207 | otherwise -> return $ Yield x (Just y, s)

@@ -1201,7 +1219,7 @@ mapM f (Stream step state) = Stream step' state

12011219 where

12021220 {-# INLINE_LATE step' #-}

12031221 step' gst st = do

1204− r <- step (rstState gst) st

1222+ r <- step (adaptState gst) st

12051223 case r of

12061224 Yield x s -> f x >>= \a -> return $ Yield a s

12071225 Skip s -> return $ Skip s

@@ -1217,7 +1235,7 @@ sequence (Stream step state) = Stream step' state

12171235 where

12181236 {-# INLINE_LATE step' #-}

12191237 step' gst st = do

1220− r <- step (rstState gst) st

1238+ r <- step (adaptState gst) st

12211239 case r of

12221240 Yield x s -> x >>= \a -> return (Yield a s)

12231241 Skip s -> return $ Skip s

@@ -1233,7 +1251,7 @@ insertBy cmp a (Stream step state) = Stream step' (state, False, Nothing)

12331251 where

12341252 {-# INLINE_LATE step' #-}

12351253 step' gst (st, False, _) = do

1236− r <- step (rstState gst) st

1254+ r <- step gst st

12371255 case r of

12381256 Yield x s -> case cmp a x of

12391257 GT -> return $ Yield x (s, False, Nothing)

@@ -1244,7 +1262,7 @@ insertBy cmp a (Stream step state) = Stream step' (state, False, Nothing)

12441262 step' _ (_, True, Nothing) = return Stop

12451263

12461264 step' gst (st, True, Just prev) = do

1247− r <- step (rstState gst) st

1265+ r <- step gst st

12481266 case r of

12491267 Yield x s -> return $ Yield prev (s, True, Just x)

12501268 Skip s -> return $ Skip (s, True, Just prev)

@@ -1260,7 +1278,7 @@ deleteBy eq x (Stream step state) = Stream step' (state, False)

12601278 where

12611279 {-# INLINE_LATE step' #-}

12621280 step' gst (st, False) = do

1263− r <- step (rstState gst) st

1281+ r <- step gst st

12641282 case r of

12651283 Yield y s -> return $

12661284 if eq x y then Skip (s, True) else Yield y (s, False)

@@ -1268,7 +1286,7 @@ deleteBy eq x (Stream step state) = Stream step' (state, False)

12681286 Stop -> return Stop

12691287

12701288 step' gst (st, True) = do

1271− r <- step (rstState gst) st

1289+ r <- step gst st

12721290 case r of

12731291 Yield y s -> return $ Yield y (s, True)

12741292 Skip s -> return $ Skip (s, True)

@@ -1297,7 +1315,7 @@ indexed (Stream step state) = Stream step' (state, 0)

12971315 where

12981316 {-# INLINE_LATE step' #-}

12991317 step' gst (st, i) = i `seq` do

1300− r <- step (rstState gst) st

1318+ r <- step (adaptState gst) st

13011319 case r of

13021320 Yield x s -> return $ Yield (i, x) (s, i+1)

13031321 Skip s -> return $ Skip (s, i)

@@ -1309,7 +1327,7 @@ indexedR m (Stream step state) = Stream step' (state, m)

13091327 where

13101328 {-# INLINE_LATE step' #-}

13111329 step' gst (st, i) = i `seq` do

1312− r <- step (rstState gst) st

1330+ r <- step (adaptState gst) st

13131331 case r of

13141332 Yield x s -> let i' = i - 1

13151333 in

@@ -1324,7 +1342,7 @@ zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)

13241342 where

13251343 {-# INLINE_LATE step #-}

13261344 step gst (sa, sb, Nothing) = do

1327− r <- stepa (rstState gst) sa

1345+ r <- stepa (adaptState gst) sa

13281346 return $

13291347 case r of

13301348 Yield x sa' -> Skip (sa', sb, Just x)

@@ -1332,7 +1350,7 @@ zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)

13321350 Stop -> Stop

13331351

13341352 step gst (sa, sb, Just x) = do

1335− r <- stepb (rstState gst) sb

1353+ r <- stepb (adaptState gst) sb

13361354 case r of

13371355 Yield y sb' -> do

13381356 z <- f x y

@@ -1423,14 +1441,14 @@ mergeByM cmp (Stream stepa ta) (Stream stepb tb) =

14231441

14241442 -- one of the values is missing, and the corresponding stream is running

14251443 step gst (Just sa, sb, Nothing, b) = do

1426− r <- stepa (rstState gst) sa

1444+ r <- stepa gst sa

14271445 return $ case r of

14281446 Yield a sa' -> Skip (Just sa', sb, Just a, b)

14291447 Skip sa' -> Skip (Just sa', sb, Nothing, b)

14301448 Stop -> Skip (Nothing, sb, Nothing, b)

14311449

14321450 step gst (sa, Just sb, a, Nothing) = do

1433− r <- stepb (rstState gst) sb

1451+ r <- stepb gst sb

14341452 return $ case r of

14351453 Yield b sb' -> Skip (sa, Just sb', a, Just b)

14361454 Skip sb' -> Skip (sa, Just sb', a, Nothing)

src/Streamly/Streams/StreamK.hs+15−13

@@ -34,8 +34,8 @@ module Streamly.Streams.StreamK

3434 -- * The stream type

3535 , Stream

3636 , unStream

37− , isolateStream

3837 , unStreamShared

38+ , unShare

3939 , runStreamSVar

4040

4141 -- * Construction

@@ -204,6 +204,7 @@ newtype Stream m a =

204204 -> m r

205205 )

206206

207+-- Run a stream in detached mode i.e. not joining an SVar

207208{-# INLINE unStream #-}

208209unStream ::

209210 Stream m a

@@ -212,11 +213,12 @@ unStream ::

212213 -> (a -> m r) -- singleton

213214 -> (a -> Stream m a -> m r) -- yield

214215 -> m r

215−unStream (Stream runner) st = runner (rstState st)

216+unStream (Stream runner) st = runner (adaptState st)

216217

217−{-# INLINE isolateStream #-}

218−isolateStream :: Stream m a -> Stream m a

219−isolateStream x = Stream $ \st stp sng yld ->

218+-- | Detach a stream from an SVar

219+{-# INLINE unShare #-}

220+unShare :: Stream m a -> Stream m a

221+unShare x = Stream $ \st stp sng yld ->

220222 unStream x st stp sng yld

221223

222224-- | Like unstream, but passes a shared SVar across continuations.

@@ -944,11 +946,11 @@ drop n m = fromStream $ Stream $ \st stp sng yld ->

944946 where

945947 go n1 m1 = Stream $ \st stp sng yld ->

946948 let single _ = stp

947− yieldk _ r = (unStreamShared $ go (n1 - 1) r) st stp sng yld

949+ yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld

948950 -- Somehow "<=" check performs better than a ">"

949951 in if n1 <= 0

950− then unStreamShared m1 st stp sng yld

951− else unStreamShared m1 st stp single yieldk

952+ then unStream m1 st stp sng yld

953+ else unStream m1 st stp single yieldk

952954

953955{-# INLINE dropWhile #-}

954956dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

@@ -1108,7 +1110,7 @@ mergeByS cmp = go

11081110 where

11091111 go mx my = Stream $ \st stp sng yld -> do

11101112 let mergeWithY a ra =

1111− let stop2 = unStream mx (rstState st) stp sng yld

1113+ let stop2 = unStream mx st stp sng yld

11121114 single2 b = do

11131115 r <- cmp a b

11141116 case r of

@@ -1119,11 +1121,11 @@ mergeByS cmp = go

11191121 case r of

11201122 GT -> yld b (go (a `cons` ra) rb)

11211123 _ -> yld a (go ra (b `cons` rb))

1122− in unStream my (rstState st) stop2 single2 yield2

1123− let stopX = unStream my (rstState st) stp sng yld

1124+ in unStream my st stop2 single2 yield2

1125+ let stopX = unStream my st stp sng yld

11241126 singleX a = mergeWithY a nil

11251127 yieldX = mergeWithY

1126− unStream mx (rstState st) stopX singleX yieldX

1128+ unStream mx st stopX singleX yieldX

11271129

11281130{-# INLINABLE mergeByM #-}

11291131mergeByM

@@ -1208,7 +1210,7 @@ bindWith par m1 f = go m1

12081210 runIsolated x = unStream x st stp sng yld

12091211

12101212 single a = runIsolated $ f a

1211− yieldk a r = runShared $ isolateStream (f a) `par` go r

1213+ yieldk a r = runShared $ unShare (f a) `par` go r

12121214 in unStream m st stp single yieldk

12131215

12141216------------------------------------------------------------------------------

src/Streamly/Streams/Zip.hs+5−5

@@ -51,7 +51,7 @@ import Prelude hiding (map, repeat, zipWith)

5151import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, unStream)

5252import Streamly.Streams.Async (mkAsync')

5353import Streamly.Streams.Serial (map)

54−import Streamly.SVar (MonadAsync, rstState)

54+import Streamly.SVar (MonadAsync, adaptState)

5555

5656import qualified Streamly.Streams.Prelude as P

5757import qualified Streamly.Streams.StreamK as K

@@ -143,8 +143,8 @@ TRAVERSABLE_INSTANCE(ZipSerialM)

143143zipAsyncWith :: (IsStream t, MonadAsync m)

144144 => (a -> b -> c) -> t m a -> t m b -> t m c

145145zipAsyncWith f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do

146− ma <- mkAsync' (rstState st) m1

147− mb <- mkAsync' (rstState st) m2

146+ ma <- mkAsync' (adaptState st) m1

147+ mb <- mkAsync' (adaptState st) m2

148148 unStream (toStream (K.zipWith f ma mb)) st stp sng yld

149149

150150-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped

@@ -154,8 +154,8 @@ zipAsyncWith f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do

154154zipAsyncWithM :: (IsStream t, MonadAsync m)

155155 => (a -> b -> m c) -> t m a -> t m b -> t m c

156156zipAsyncWithM f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do

157− ma <- mkAsync' (rstState st) m1

158− mb <- mkAsync' (rstState st) m2

157+ ma <- mkAsync' (adaptState st) m1

158+ mb <- mkAsync' (adaptState st) m2

159159 unStream (toStream (K.zipWithM f ma mb)) st stp sng yld

160160

161161------------------------------------------------------------------------------