315abcf · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix some state sharing/clearing safety issues

315abcfHarendra Kumar8 years ago

src/Streamly/Prelude.hs+8−8

compare with master

unifiedsplit

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

363import qualified Prelude

364import qualified System.IO as IO

365

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

366+import Streamly.SVar (MonadAsync, defState)

367import Streamly.Streams.Async (mkAsync')

368import Streamly.Streams.Combinators (maxYields)

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

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

1449 let stop = return ()

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

1451 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

1453

1454------------------------------------------------------------------------------

1455-- Transformation by Folding (Scans)

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

1987mergeAsyncBy :: (IsStream t, MonadAsync m)

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

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

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

1990+ ma <- mkAsync' st m1

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

1991+ mb <- mkAsync' st m2

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

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

1993

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

1995-- merged are generated concurrently).

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

1998mergeAsyncByM :: (IsStream t, MonadAsync m)

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

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

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

2001+ ma <- mkAsync' st m1

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

2002+ mb <- mkAsync' st m2

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

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

31 , Limit (..)

32 , State (streamVar)

33 , defState

34− , rstState

34+ , adaptState

35 , getMaxThreads

36 , setMaxThreads

37 , getMaxBuffer

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

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

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

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

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.

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

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

437−rstState st = st

444+adaptState st = st

438 { streamVar = Nothing

445 { streamVar = Nothing

439 , _yieldLimit = Nothing

446 , _yieldLimit = Nothing

440 }

447 }

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

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

688

689 single a = runIsolated $ f a

690 yieldk a r = runShared $

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

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

692 in unStream g st stp single yieldk

693

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

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

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

57import Streamly.Streams.SVar (fromSVar)

58import Streamly.Streams.Serial (map)

59import Streamly.SVar

60+import Streamly.Streams.StreamK

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

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

61− unStreamShared, runStreamSVar)

62+ runStreamSVar)

62import qualified Streamly.Streams.StreamK as K

63import qualified Streamly.Streams.StreamK as K

64#include "Instances.hs"

65#include "Instances.hs"

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

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

142{-# INLINE applyWith #-}

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

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

145− sv <- newParallelVar (rstState st)

145+ sv <- newParallelVar (adaptState st)

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

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

148

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

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

5{-# LANGUAGE FlexibleContexts #-}

6{-# LANGUAGE FlexibleInstances #-}

7{-# LANGUAGE MultiParamTypeClasses #-}

8+{-# LANGUAGE PatternSynonyms #-}

9+{-# LANGUAGE ViewPatterns #-}

8{-# LANGUAGE RankNTypes #-}

10{-# LANGUAGE RankNTypes #-}

10#include "inline.hs"

12#include "inline.hs"

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

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

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

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

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

199

201

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

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

201

203

202import qualified Streamly.Streams.StreamK as K

204import qualified Streamly.Streams.StreamK as K

203

205

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

219-- gst = global state

221-- gst = global state

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

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

221-- current state, and the current state.

223-- current state, and the current state.

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+

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

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

223

242

224------------------------------------------------------------------------------

243------------------------------------------------------------------------------

225-- Construction

244-- Construction

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

238 {-# INLINE_LATE step1 #-}

257 {-# INLINE_LATE step1 #-}

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

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

240 step1 gst (Just st) = do

259 step1 gst (Just st) = do

241− r <- step (rstState gst) st

260+ r <- step gst st

242 return $

261 return $

243 case r of

262 case r of

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

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

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

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

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

493 step _ [] = return Stop

512 step _ [] = return Stop

494

513

495−-- XXX pass the state to streamD

496{-# INLINE_LATE fromStreamK #-}

514{-# INLINE_LATE fromStreamK #-}

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

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

498fromStreamK = Stream step

516fromStreamK = Stream step

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

773 where

791 where

774 {-# INLINE_LATE step' #-}

792 {-# INLINE_LATE step' #-}

775 step' gst (st, i) = do

793 step' gst (st, i) = do

776− r <- step (rstState gst) st

794+ r <- step (adaptState gst) st

777 return $ case r of

795 return $ case r of

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

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

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

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

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

785 where

803 where

786 {-# INLINE_LATE step' #-}

804 {-# INLINE_LATE step' #-}

787 step' gst (Left st) = do

805 step' gst (Left st) = do

788− r <- step (rstState gst) st

806+ r <- step (adaptState gst) st

789 case r of

807 case r of

790 Yield a s -> do

808 Yield a s -> do

791 b_stream <- f a

809 b_stream <- f a

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

935 where

953 where

936 {-# INLINE_LATE step' #-}

954 {-# INLINE_LATE step' #-}

937 step' gst (st, prev) = do

955 step' gst (st, prev) = do

938− r <- step (rstState gst) st

956+ r <- step (adaptState gst) st

939 case r of

957 case r of

940 Yield x s -> do

958 Yield x s -> do

941 acc <- prev

959 acc <- prev

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

963 where

981 where

964 {-# INLINE_LATE step' #-}

982 {-# INLINE_LATE step' #-}

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

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

966− r <- step (rstState gst) st

984+ r <- step (adaptState gst) st

967 case r of

985 case r of

968 Yield x s -> do

986 Yield x s -> do

969 y <- fstep acc x

987 y <- fstep acc x

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

981 where

999 where

982 {-# INLINE_LATE step' #-}

1000 {-# INLINE_LATE step' #-}

983 step' gst (st, acc) = do

1001 step' gst (st, acc) = do

984− r <- step (rstState gst) st

1002+ r <- step (adaptState gst) st

985 case r of

1003 case r of

986 Yield x s -> do

1004 Yield x s -> do

987 y <- fstep acc x

1005 y <- fstep acc x

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

1015 where

1033 where

1016 {-# INLINE_LATE step' #-}

1034 {-# INLINE_LATE step' #-}

1017 step' gst (st, Nothing) = do

1035 step' gst (st, Nothing) = do

1018− r <- step (rstState gst) st

1036+ r <- step gst st

1019 case r of

1037 case r of

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

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

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

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

1022 Stop -> return Stop

1040 Stop -> return Stop

1023

1041

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

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

1025− r <- step (rstState gst) st

1043+ r <- step gst st

1026 case r of

1044 case r of

1027 Yield y s -> do

1045 Yield y s -> do

1028 z <- fstep acc y

1046 z <- fstep acc y

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

1040 where

1058 where

1041 {-# INLINE_LATE step' #-}

1059 {-# INLINE_LATE step' #-}

1042 step' gst (st, Nothing) = do

1060 step' gst (st, Nothing) = do

1043− r <- step (rstState gst) st

1061+ r <- step gst st

1044 case r of

1062 case r of

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

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

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

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

1047 Stop -> return Stop

1065 Stop -> return Stop

1048

1066

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

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

1050− r <- step (rstState gst) st

1068+ r <- step gst st

1051 case r of

1069 case r of

1052 Yield y s -> do

1070 Yield y s -> do

1053 z <- fstep acc y

1071 z <- fstep acc y

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

1069 where

1087 where

1070 {-# INLINE_LATE step' #-}

1088 {-# INLINE_LATE step' #-}

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

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

1072− r <- step (rstState gst) st

1090+ r <- step gst st

1073 return $ case r of

1091 return $ case r of

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

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

1075 Skip s -> Skip (s, i)

1093 Skip s -> Skip (s, i)

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

1082 where

1100 where

1083 {-# INLINE_LATE step' #-}

1101 {-# INLINE_LATE step' #-}

1084 step' gst st = do

1102 step' gst st = do

1085− r <- step (rstState gst) st

1103+ r <- step gst st

1086 case r of

1104 case r of

1087 Yield x s -> do

1105 Yield x s -> do

1088 b <- f x

1106 b <- f x

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

1101 {-# INLINE_LATE step' #-}

1119 {-# INLINE_LATE step' #-}

1102 step' gst (st, Just i)

1120 step' gst (st, Just i)

1103 | i > 0 = do

1121 | i > 0 = do

1104− r <- step (rstState gst) st

1122+ r <- step gst st

1105 return $

1123 return $

1106 case r of

1124 case r of

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

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

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

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

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

1111

1129

1112 step' gst (st, Nothing) = do

1130 step' gst (st, Nothing) = do

1113− r <- step (rstState gst) st

1131+ r <- step gst st

1114 return $

1132 return $

1115 case r of

1133 case r of

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

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

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

1128 where

1146 where

1129 {-# INLINE_LATE step' #-}

1147 {-# INLINE_LATE step' #-}

1130 step' gst (DropWhileDrop st) = do

1148 step' gst (DropWhileDrop st) = do

1131− r <- step (rstState gst) st

1149+ r <- step gst st

1132 case r of

1150 case r of

1133 Yield x s -> do

1151 Yield x s -> do

1134 b <- f x

1152 b <- f x

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

1139 Stop -> return Stop

1157 Stop -> return Stop

1140

1158

1141 step' gst (DropWhileNext st) = do

1159 step' gst (DropWhileNext st) = do

1142− r <- step (rstState gst) st

1160+ r <- step gst st

1143 case r of

1161 case r of

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

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

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

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

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

1157 where

1175 where

1158 {-# INLINE_LATE step' #-}

1176 {-# INLINE_LATE step' #-}

1159 step' gst st = do

1177 step' gst st = do

1160− r <- step (rstState gst) st

1178+ r <- step gst st

1161 case r of

1179 case r of

1162 Yield x s -> do

1180 Yield x s -> do

1163 b <- f x

1181 b <- f x

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

1177 where

1195 where

1178 {-# INLINE_LATE step' #-}

1196 {-# INLINE_LATE step' #-}

1179 step' gst (Nothing, st) = do

1197 step' gst (Nothing, st) = do

1180− r <- step (rstState gst) st

1198+ r <- step gst st

1181 case r of

1199 case r of

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

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

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

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

1184 Stop -> return Stop

1202 Stop -> return Stop

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

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

1186− r <- step (rstState gst) st

1204+ r <- step gst st

1187 case r of

1205 case r of

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

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

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

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

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

1201 where

1219 where

1202 {-# INLINE_LATE step' #-}

1220 {-# INLINE_LATE step' #-}

1203 step' gst st = do

1221 step' gst st = do

1204− r <- step (rstState gst) st

1222+ r <- step (adaptState gst) st

1205 case r of

1223 case r of

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

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

1207 Skip s -> return $ Skip s

1225 Skip s -> return $ Skip s

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

1217 where

1235 where

1218 {-# INLINE_LATE step' #-}

1236 {-# INLINE_LATE step' #-}

1219 step' gst st = do

1237 step' gst st = do

1220− r <- step (rstState gst) st

1238+ r <- step (adaptState gst) st

1221 case r of

1239 case r of

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

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

1223 Skip s -> return $ Skip s

1241 Skip s -> return $ Skip s

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

1233 where

1251 where

1234 {-# INLINE_LATE step' #-}

1252 {-# INLINE_LATE step' #-}

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

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

1236− r <- step (rstState gst) st

1254+ r <- step gst st

1237 case r of

1255 case r of

1238 Yield x s -> case cmp a x of

1256 Yield x s -> case cmp a x of

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

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

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

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

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

1245

1263

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

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

1247− r <- step (rstState gst) st

1265+ r <- step gst st

1248 case r of

1266 case r of

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

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

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

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

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

1260 where

1278 where

1261 {-# INLINE_LATE step' #-}

1279 {-# INLINE_LATE step' #-}

1262 step' gst (st, False) = do

1280 step' gst (st, False) = do

1263− r <- step (rstState gst) st

1281+ r <- step gst st

1264 case r of

1282 case r of

1265 Yield y s -> return $

1283 Yield y s -> return $

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

1284 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)

1268 Stop -> return Stop

1286 Stop -> return Stop

1269

1287

1270 step' gst (st, True) = do

1288 step' gst (st, True) = do

1271− r <- step (rstState gst) st

1289+ r <- step gst st

1272 case r of

1290 case r of

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

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

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

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

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

1297 where

1315 where

1298 {-# INLINE_LATE step' #-}

1316 {-# INLINE_LATE step' #-}

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

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

1300− r <- step (rstState gst) st

1318+ r <- step (adaptState gst) st

1301 case r of

1319 case r of

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

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

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

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

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

1309 where

1327 where

1310 {-# INLINE_LATE step' #-}

1328 {-# INLINE_LATE step' #-}

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

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

1312− r <- step (rstState gst) st

1330+ r <- step (adaptState gst) st

1313 case r of

1331 case r of

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

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

1315 in

1333 in

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

1324 where

1342 where

1325 {-# INLINE_LATE step #-}

1343 {-# INLINE_LATE step #-}

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

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

1327− r <- stepa (rstState gst) sa

1345+ r <- stepa (adaptState gst) sa

1328 return $

1346 return $

1329 case r of

1347 case r of

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

1348 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)

1332 Stop -> Stop

1350 Stop -> Stop

1333

1351

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

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

1335− r <- stepb (rstState gst) sb

1353+ r <- stepb (adaptState gst) sb

1336 case r of

1354 case r of

1337 Yield y sb' -> do

1355 Yield y sb' -> do

1338 z <- f x y

1356 z <- f x y

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

1423

1441

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

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

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

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

1426− r <- stepa (rstState gst) sa

1444+ r <- stepa gst sa

1427 return $ case r of

1445 return $ case r of

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

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

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

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

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

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

1431

1449

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

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

1433− r <- stepb (rstState gst) sb

1451+ r <- stepb gst sb

1434 return $ case r of

1452 return $ case r of

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

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

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

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

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

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

34 -- * The stream type

35 , Stream

36 , unStream

37− , isolateStream

38 , unStreamShared

37 , unStreamShared

38+ , unShare

39 , runStreamSVar

41 -- * Construction

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

204 -> m r

205 )

206

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

207{-# INLINE unStream #-}

208{-# INLINE unStream #-}

208unStream ::

209unStream ::

209 Stream m a

210 Stream m a

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

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

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

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

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

214 -> m r

215 -> m r

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

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

216

217

217−{-# INLINE isolateStream #-}

218+-- | Detach a stream from an SVar

219+{-# INLINE unShare #-}

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

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

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

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

220 unStream x st stp sng yld

222 unStream x st stp sng yld

221

223

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

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

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

944 where

946 where

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

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

946 let single _ = stp

948 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

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

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

949 in if n1 <= 0

951 in if n1 <= 0

950− then unStreamShared m1 st stp sng yld

952+ then unStream m1 st stp sng yld

951− else unStreamShared m1 st stp single yieldk

953+ else unStream m1 st stp single yieldk

952

954

953{-# INLINE dropWhile #-}

955{-# INLINE dropWhile #-}

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

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

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

1108 where

1110 where

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

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

1110 let mergeWithY a ra =

1112 let mergeWithY a ra =

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

1113+ let stop2 = unStream mx st stp sng yld

1112 single2 b = do

1114 single2 b = do

1113 r <- cmp a b

1115 r <- cmp a b

1114 case r of

1116 case r of

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

1119 case r of

1121 case r of

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

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

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

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

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

1124+ in unStream my st stop2 single2 yield2

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

1125+ let stopX = unStream my st stp sng yld

1124 singleX a = mergeWithY a nil

1126 singleX a = mergeWithY a nil

1125 yieldX = mergeWithY

1127 yieldX = mergeWithY

1126− unStream mx (rstState st) stopX singleX yieldX

1128+ unStream mx st stopX singleX yieldX

1127

1129

1128{-# INLINABLE mergeByM #-}

1130{-# INLINABLE mergeByM #-}

1129mergeByM

1131mergeByM

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

1208 runIsolated x = unStream x st stp sng yld

1210 runIsolated x = unStream x st stp sng yld

1209

1211

1210 single a = runIsolated $ f a

1212 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

1212 in unStream m st stp single yieldk

1214 in unStream m st stp single yieldk

1213

1215

1214------------------------------------------------------------------------------

1216------------------------------------------------------------------------------

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

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

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

52import Streamly.Streams.Async (mkAsync')

53import Streamly.Streams.Serial (map)

54−import Streamly.SVar (MonadAsync, rstState)

54+import Streamly.SVar (MonadAsync, adaptState)

56import qualified Streamly.Streams.Prelude as P

57import qualified Streamly.Streams.StreamK as K

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

143zipAsyncWith :: (IsStream t, MonadAsync m)

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

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

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

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

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

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

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

149

150-- | 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

154zipAsyncWithM :: (IsStream t, MonadAsync m)

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

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

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

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

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

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

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

160

161------------------------------------------------------------------------------