git.iko.soy/streamly
on master help sign in
Make the default unStream safe, #96.

1) Hide the `Stream` constructor, use `mkStream` instead. 2) Change `unStream` to use `rstState`. 3) Use `unStreamShared` to share the `SVar`. 4) Remove `unStreamIsolated`, use `unStream` instead. 5) Change all `unStream m (rstState st)` to `unStream m st` 6) Change all `unStream m st` to `unStreamShared m st`

commit924856bbbef1ce722b8680dded43b8ca0904319c authorPranay Sashank <pranaysashank@gmail.com> dateSun 2 Dec 2018, 11:07 (8 years ago) parent7c2b9b7
10 files changed+171−173
compare with master
unifiedsplit
src/Streamly/Prelude.hs+9−9 src/Streamly/Streams/Ahead.hs+14−14 src/Streamly/Streams/Async.hs+13−12 src/Streamly/Streams/Combinators.hs+12−12 src/Streamly/Streams/Parallel.hs+12−11 src/Streamly/Streams/SVar.hs+9−9 src/Streamly/Streams/Serial.hs+15−15 src/Streamly/Streams/StreamD.hs+3−3 src/Streamly/Streams/StreamK.hs+79−83 src/Streamly/Streams/Zip.hs+5−5
src/Streamly/Prelude.hs+9−9
@@ -840,9 +840,9 @@ iterate step = fromStream . go
840840iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a
841841iterateM step = go
842842 where
843− go s = fromStream $ K.Stream $ \svr stp sng yld -> do
843+ go s = fromStream $ K.mkStream $ \svr stp sng yld -> do
844844 next <- step s
845− K.unStream (toStream (return s |: go next)) svr stp sng yld
845+ K.unStreamShared (toStream (return s |: go next)) svr stp sng yld
846846
847847------------------------------------------------------------------------------
848848-- Conversions
@@ -896,7 +896,7 @@ each = K.fromFoldable
896896fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
897897fromHandle h = fromStream go
898898 where
899− go = K.Stream $ \_ stp _ yld -> do
899+ go = K.mkStream $ \_ stp _ yld -> do
900900 eof <- liftIO $ IO.hIsEOF h
901901 if eof
902902 then stp
@@ -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.unStream m1 defState stop single yieldk
1452+ in K.unStreamShared m1 defState stop single yieldk
14531453
14541454------------------------------------------------------------------------------
14551455-- Transformation by Folding (Scans)
@@ -1770,12 +1770,12 @@ mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
17701770reverse :: (IsStream t) => t m a -> t m a
17711771reverse m = fromStream $ go K.nil (toStream m)
17721772 where
1773− go rev rest = K.Stream $ \st stp sng yld ->
1774− let runIt x = K.unStream x (rstState st) stp sng yld
1773+ go rev rest = K.mkStream $ \st stp sng yld ->
1774+ let runIt x = K.unStream x st stp sng yld
17751775 stop = runIt rev
17761776 single a = runIt $ a `K.cons` rev
17771777 yieldk a r = runIt $ go (a `K.cons` rev) r
1778− in K.unStream rest (rstState st) stop single yieldk
1778+ in K.unStream rest st stop single yieldk
17791779
17801780------------------------------------------------------------------------------
17811781-- Transformation by Inserting
@@ -1986,7 +1986,7 @@ merge = mergeBy compare
19861986-- @since 0.6.0
19871987mergeAsyncBy :: (IsStream t, MonadAsync m)
19881988 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
1989−mergeAsyncBy f m1 m2 = K.fromStream $ K.Stream $ \st stp sng yld -> do
1989+mergeAsyncBy f m1 m2 = K.fromStream $ K.mkStream $ \st stp sng yld -> do
19901990 ma <- mkAsync' (rstState st) m1
19911991 mb <- mkAsync' (rstState st) m2
19921992 K.unStream (K.toStream (K.mergeBy f ma mb)) (rstState st) stp sng yld
@@ -1997,7 +1997,7 @@ mergeAsyncBy f m1 m2 = K.fromStream $ K.Stream $ \st stp sng yld -> do
19971997-- @since 0.6.0
19981998mergeAsyncByM :: (IsStream t, MonadAsync m)
19991999 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2000−mergeAsyncByM f m1 m2 = K.fromStream $ K.Stream $ \st stp sng yld -> do
2000+mergeAsyncByM f m1 m2 = K.fromStream $ K.mkStream $ \st stp sng yld -> do
20012001 ma <- mkAsync' (rstState st) m1
20022002 mb <- mkAsync' (rstState st) m2
20032003 K.unStream (K.toStream (K.mergeByM f ma mb)) (rstState st) stp sng yld
src/Streamly/Streams/Ahead.hs+14−14
@@ -48,7 +48,7 @@ import Streamly.Streams.SVar (fromSVar)
4848import Streamly.Streams.Serial (map)
4949import Streamly.SVar
5050import Streamly.Streams.StreamK
51− (IsStream(..), Stream(..), unStreamShared, unStreamIsolated,
51+ (IsStream(..), Stream, mkStream, unStream, unStreamShared,
5252 runStreamSVar)
5353import qualified Streamly.Streams.StreamK as K
5454
@@ -544,25 +544,25 @@ workLoopAhead q heap st sv winfo = do
544544-- The only difference between forkSVarAsync and this is that we run the left
545545-- computation without a shared SVar.
546546forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
547−forkSVarAhead m1 m2 = Stream $ \st stp sng yld -> do
547+forkSVarAhead m1 m2 = mkStream $ \st stp sng yld -> do
548548 sv <- newAheadVar st (concurrently m1 m2) workLoopAhead
549− unStream (fromSVar sv) (rstState st) stp sng yld
549+ unStream (fromSVar sv) st stp sng yld
550550 where
551− concurrently ma mb = Stream $ \st stp sng yld -> do
551+ concurrently ma mb = mkStream $ \st stp sng yld -> do
552552 liftIO $ enqueue (fromJust $ streamVar st) mb
553− unStream ma (rstState st) stp sng yld
553+ unStream ma st stp sng yld
554554
555555{-# INLINE aheadS #-}
556556aheadS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
557−aheadS m1 m2 = Stream $ \st stp sng yld ->
557+aheadS m1 m2 = mkStream $ \st stp sng yld ->
558558 case streamVar st of
559559 Just sv | svarStyle sv == AheadVar -> do
560560 liftIO $ enqueue sv m2
561561 -- Always run the left side on a new SVar to avoid complexity in
562562 -- sequencing results. This means the left side cannot further
563563 -- split into more ahead computations on the same SVar.
564− unStream m1 (rstState st) stp sng yld
565− _ -> unStream (forkSVarAhead m1 m2) st stp sng yld
564+ unStream m1 st stp sng yld
565+ _ -> unStreamShared (forkSVarAhead m1 m2) st stp sng yld
566566
567567-- | XXX we can implement it more efficienty by directly implementing instead
568568-- of combining streams using ahead.
@@ -655,8 +655,8 @@ instance IsStream AheadT where
655655-- @since 0.3.0
656656{-# INLINE ahead #-}
657657ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
658−ahead m1 m2 = fromStream $ Stream $ \st stp sng yld ->
659− unStream (aheadS (toStream m1) (toStream m2)) st stp sng yld
658+ahead m1 m2 = fromStream $ mkStream $ \st stp sng yld ->
659+ unStreamShared (aheadS (toStream m1) (toStream m2)) st stp sng yld
660660
661661instance MonadAsync m => Semigroup (AheadT m a) where
662662 (<>) = ahead
@@ -681,15 +681,15 @@ aheadbind
681681 -> Stream m b
682682aheadbind m f = go m
683683 where
684− go (Stream g) =
685− Stream $ \st stp sng yld ->
684+ go g =
685+ mkStream $ \st stp sng yld ->
686686 let runShared x = unStreamShared x st stp sng yld
687− runIsolated x = unStreamIsolated x st stp sng yld
687+ runIsolated x = unStream x st stp sng yld
688688
689689 single a = runIsolated $ f a
690690 yieldk a r = runShared $
691691 K.isolateStream (f a) `aheadS` go r
692− in g (rstState st) stp single yieldk
692+ in unStream g st stp single yieldk
693693
694694instance MonadAsync m => Monad (AheadT m) where
695695 return = pure
src/Streamly/Streams/Async.hs+13−12
@@ -57,7 +57,8 @@ 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(..), adapt, runStreamSVar)
60+import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, unStream, adapt,
61+ unStreamShared, runStreamSVar)
6162import qualified Streamly.Streams.StreamK as K
6263
6364#include "Instances.hs"
@@ -528,25 +529,25 @@ newWAsyncVar st m = do
528529
529530forkSVarAsync :: MonadAsync m
530531 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
531−forkSVarAsync style m1 m2 = Stream $ \st stp sng yld -> do
532+forkSVarAsync style m1 m2 = mkStream $ \st stp sng yld -> do
532533 sv <- case style of
533534 AsyncVar -> newAsyncVar st (concurrently m1 m2)
534535 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)
535536 _ -> error "illegal svar type"
536− unStream (fromSVar sv) (rstState st) stp sng yld
537+ unStream (fromSVar sv) st stp sng yld
537538 where
538− concurrently ma mb = Stream $ \st stp sng yld -> do
539+ concurrently ma mb = mkStream $ \st stp sng yld -> do
539540 liftIO $ enqueue (fromJust $ streamVar st) mb
540− unStream ma st stp sng yld
541+ unStreamShared ma st stp sng yld
541542
542543{-# INLINE joinStreamVarAsync #-}
543544joinStreamVarAsync :: MonadAsync m
544545 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
545−joinStreamVarAsync style m1 m2 = Stream $ \st stp sng yld ->
546+joinStreamVarAsync style m1 m2 = mkStream $ \st stp sng yld ->
546547 case streamVar st of
547548 Just sv | svarStyle sv == style ->
548− liftIO (enqueue sv m2) >> unStream m1 st stp sng yld
549− _ -> unStream (forkSVarAsync style m1 m2) st stp sng yld
549+ liftIO (enqueue sv m2) >> unStreamShared m1 st stp sng yld
550+ _ -> unStreamShared (forkSVarAsync style m1 m2) st stp sng yld
550551
551552------------------------------------------------------------------------------
552553-- Semigroup and Monoid style compositions for parallel actions
@@ -563,8 +564,8 @@ asyncS = joinStreamVarAsync AsyncVar
563564-- @since 0.2.0
564565{-# INLINE async #-}
565566async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
566−async m1 m2 = fromStream $ Stream $ \st stp sng yld ->
567− unStream (joinStreamVarAsync AsyncVar (toStream m1) (toStream m2))
567+async m1 m2 = fromStream $ mkStream $ \st stp sng yld ->
568+ unStreamShared (joinStreamVarAsync AsyncVar (toStream m1) (toStream m2))
568569 st stp sng yld
569570
570571-- | Same as 'async'.
@@ -711,8 +712,8 @@ consMWAsync m r = K.yieldM m `wAsyncS` r
711712-- @since 0.2.0
712713{-# INLINE wAsync #-}
713714wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
714−wAsync m1 m2 = fromStream $ Stream $ \st stp sng yld ->
715− unStream (wAsyncS (toStream m1) (toStream m2)) st stp sng yld
715+wAsync m1 m2 = fromStream $ mkStream $ \st stp sng yld ->
716+ unStreamShared (wAsyncS (toStream m1) (toStream m2)) st stp sng yld
716717
717718-- | Wide async composition or async composition with breadth first traversal.
718719-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
src/Streamly/Streams/Combinators.hs+12−12
@@ -52,8 +52,8 @@ import Streamly.Streams.Serial (SerialT)
5252-- @since 0.4.0
5353{-# INLINE_NORMAL maxThreads #-}
5454maxThreads :: IsStream t => Int -> t m a -> t m a
55−maxThreads n m = fromStream $ Stream $ \st stp sng yld ->
56− unStream (toStream m) (setMaxThreads n st) stp sng yld
55+maxThreads n m = fromStream $ mkStream $ \st stp sng yld ->
56+ unStreamShared (toStream m) (setMaxThreads n st) stp sng yld
5757
5858{-
5959{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}
@@ -77,8 +77,8 @@ maxThreadsSerial _ = id
7777-- @since 0.4.0
7878{-# INLINE_NORMAL maxBuffer #-}
7979maxBuffer :: IsStream t => Int -> t m a -> t m a
80−maxBuffer n m = fromStream $ Stream $ \st stp sng yld ->
81− unStream (toStream m) (setMaxBuffer n st) stp sng yld
80+maxBuffer n m = fromStream $ mkStream $ \st stp sng yld ->
81+ unStreamShared (toStream m) (setMaxBuffer n st) stp sng yld
8282
8383{-
8484{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}
@@ -101,7 +101,7 @@ maxBufferSerial _ = id
101101-- @since 0.5.0
102102{-# INLINE_NORMAL rate #-}
103103rate :: IsStream t => Maybe Rate -> t m a -> t m a
104−rate r m = fromStream $ Stream $ \st stp sng yld ->
104+rate r m = fromStream $ mkStream $ \st stp sng yld ->
105105 case r of
106106 Just (Rate low goal _ _) | goal < low ->
107107 error "rate: Target rate cannot be lower than minimum rate."
@@ -109,7 +109,7 @@ rate r m = fromStream $ Stream $ \st stp sng yld ->
109109 error "rate: Target rate cannot be greater than maximum rate."
110110 Just (Rate low _ high _) | low > high ->
111111 error "rate: Minimum rate cannot be greater than maximum rate."
112− _ -> unStream (toStream m) (setStreamRate r st) stp sng yld
112+ _ -> unStreamShared (toStream m) (setStreamRate r st) stp sng yld
113113
114114-- XXX implement for serial streams as well, as a simple delay
115115
@@ -181,8 +181,8 @@ constRate r = rate (Just $ Rate r r r 0)
181181--
182182{-# INLINE_NORMAL _serialLatency #-}
183183_serialLatency :: IsStream t => Int -> t m a -> t m a
184−_serialLatency n m = fromStream $ Stream $ \st stp sng yld ->
185− unStream (toStream m) (setStreamLatency n st) stp sng yld
184+_serialLatency n m = fromStream $ mkStream $ \st stp sng yld ->
185+ unStreamShared (toStream m) (setStreamLatency n st) stp sng yld
186186
187187{-
188188{-# RULES "serialLatency serial" _serialLatency = serialLatencySerial #-}
@@ -196,8 +196,8 @@ serialLatencySerial _ = id
196196-- inherited by everything in enclosed scope.
197197{-# INLINE_NORMAL maxYields #-}
198198maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a
199−maxYields n m = fromStream $ Stream $ \st stp sng yld ->
200− unStream (toStream m) (setYieldLimit n st) stp sng yld
199+maxYields n m = fromStream $ mkStream $ \st stp sng yld ->
200+ unStreamShared (toStream m) (setYieldLimit n st) stp sng yld
201201
202202{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}
203203maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a
@@ -212,5 +212,5 @@ printState st = liftIO $ do
212212
213213-- | Print debug information about an SVar when the stream ends
214214inspectMode :: IsStream t => t m a -> t m a
215−inspectMode m = fromStream $ Stream $ \st stp sng yld ->
216− unStream (toStream m) (setInspectMode st) stp sng yld
215+inspectMode m = fromStream $ mkStream $ \st stp sng yld ->
216+ unStreamShared (toStream m) (setInspectMode st) stp sng yld
src/Streamly/Streams/Parallel.hs+12−11
@@ -49,7 +49,8 @@ import Prelude hiding (map)
4949import Streamly.Streams.SVar (fromSVar)
5050import Streamly.Streams.Serial (map)
5151import Streamly.SVar
52−import Streamly.Streams.StreamK (IsStream(..), Stream(..), adapt)
52+import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, unStream,
53+ unStreamShared, adapt)
5354import qualified Streamly.Streams.StreamK as K
5455
5556#include "Instances.hs"
@@ -62,7 +63,7 @@ import qualified Streamly.Streams.StreamK as K
6263runOne
6364 :: MonadIO m
6465 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
65−runOne st m winfo = unStream m st stop single yieldk
66+runOne st m winfo = unStreamShared m st stop single yieldk
6667
6768 where
6869
@@ -88,21 +89,21 @@ runOne st m winfo = unStream m st stop single yieldk
8889
8990{-# NOINLINE forkSVarPar #-}
9091forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
91−forkSVarPar m r = Stream $ \st stp sng yld -> do
92+forkSVarPar m r = mkStream $ \st stp sng yld -> do
9293 sv <- newParallelVar st
9394 pushWorkerPar sv (runOne st{streamVar = Just sv} m)
9495 pushWorkerPar sv (runOne st{streamVar = Just sv} r)
95− unStream (fromSVar sv) (rstState st) stp sng yld
96+ unStream (fromSVar sv) st stp sng yld
9697
9798{-# INLINE joinStreamVarPar #-}
9899joinStreamVarPar :: MonadAsync m
99100 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
100−joinStreamVarPar style m1 m2 = Stream $ \st stp sng yld ->
101+joinStreamVarPar style m1 m2 = mkStream $ \st stp sng yld ->
101102 case streamVar st of
102103 Just sv | svarStyle sv == style -> do
103104 pushWorkerPar sv (runOne st m1)
104− unStream m2 st stp sng yld
105− _ -> unStream (forkSVarPar m1 m2) st stp sng yld
105+ unStreamShared m2 st stp sng yld
106+ _ -> unStreamShared (forkSVarPar m1 m2) st stp sng yld
106107
107108{-# INLINE parallelStream #-}
108109parallelStream :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
@@ -120,8 +121,8 @@ consMParallel m r = K.yieldM m `parallelStream` r
120121-- @since 0.2.0
121122{-# INLINE parallel #-}
122123parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
123−parallel m1 m2 = fromStream $ Stream $ \st stp sng yld ->
124− unStream (parallelStream (toStream m1) (toStream m2))
124+parallel m1 m2 = fromStream $ mkStream $ \st stp sng yld ->
125+ unStreamShared (parallelStream (toStream m1) (toStream m2))
125126 st stp sng yld
126127
127128------------------------------------------------------------------------------
@@ -140,10 +141,10 @@ mkParallel m = do
140141
141142{-# INLINE applyWith #-}
142143applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
143−applyWith f m = fromStream $ Stream $ \st stp sng yld -> do
144+applyWith f m = fromStream $ mkStream $ \st stp sng yld -> do
144145 sv <- newParallelVar (rstState st)
145146 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
146− unStream (toStream $ f $ fromSVar sv) (rstState st) stp sng yld
147+ unStream (toStream $ f $ fromSVar sv) st stp sng yld
147148
148149------------------------------------------------------------------------------
149150-- Stream runner concurrent function application
src/Streamly/Streams/SVar.hs+9−9
@@ -38,12 +38,12 @@ printSVar sv how = do
3838-- | Pull a stream from an SVar.
3939{-# NOINLINE fromStreamVar #-}
4040fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
41−fromStreamVar sv = Stream $ \st stp sng yld -> do
41+fromStreamVar sv = mkStream $ \st stp sng yld -> do
4242 list <- readOutputQ sv
4343 -- Reversing the output is important to guarantee that we process the
4444 -- outputs in the same order as they were generated by the constituent
4545 -- streams.
46− unStream (processEvents $ reverse list) (rstState st) stp sng yld
46+ unStream (processEvents $ reverse list) st stp sng yld
4747
4848 where
4949
@@ -55,36 +55,36 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do
5555 stp
5656
5757 {-# INLINE processEvents #-}
58− processEvents [] = Stream $ \st stp sng yld -> do
58+ processEvents [] = mkStream $ \st stp sng yld -> do
5959 done <- postProcess sv
6060 if done
6161 then allDone stp
62− else unStream (fromStreamVar sv) (rstState st) stp sng yld
62+ else unStream (fromStreamVar sv) st stp sng yld
6363
64− processEvents (ev : es) = Stream $ \st stp sng yld -> do
64+ processEvents (ev : es) = mkStream $ \st stp sng yld -> do
6565 let rest = processEvents es
6666 case ev of
6767 ChildYield a -> yld a rest
6868 ChildStop tid e -> do
6969 accountThread sv tid
7070 case e of
71− Nothing -> unStream rest (rstState st) stp sng yld
71+ Nothing -> unStream rest st stp sng yld
7272 Just ex ->
7373 case fromException ex of
7474 Just ThreadAbort ->
75− unStream rest (rstState st) stp sng yld
75+ unStream rest st stp sng yld
7676 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
7777
7878{-# INLINE fromSVar #-}
7979fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
8080fromSVar sv =
81− fromStream $ Stream $ \st stp sng yld -> do
81+ fromStream $ mkStream $ \st stp sng yld -> do
8282 ref <- liftIO $ newIORef ()
8383 _ <- liftIO $ mkWeakIORef ref hook
8484 -- We pass a copy of sv to fromStreamVar, so that we know that it has
8585 -- no other references, when that copy gets garbage collected "ref"
8686 -- will get garbage collected and our hook will be called.
87− unStream (fromStreamVar sv{svarRef = Just ref}) st stp sng yld
87+ unStreamShared (fromStreamVar sv{svarRef = Just ref}) st stp sng yld
8888 where
8989
9090 hook = do
src/Streamly/Streams/Serial.hs+15−15
@@ -60,8 +60,8 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
6060 readListPrecDefault)
6161import Prelude hiding (map, mapM)
6262
63−import Streamly.SVar (rstState)
64−import Streamly.Streams.StreamK (IsStream(..), adapt, Stream(..))
63+import Streamly.Streams.StreamK (IsStream(..), adapt, Stream, mkStream,
64+ unStream)
6565import qualified Streamly.Streams.Prelude as P
6666import qualified Streamly.Streams.StreamK as K
6767import qualified Streamly.Streams.StreamD as D
@@ -174,9 +174,9 @@ instance IsStream SerialT where
174174-- @since 0.2.0
175175{-# INLINE serial #-}
176176serial :: IsStream t => t m a -> t m a -> t m a
177−serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
177+serial m1 m2 = fromStream $ mkStream $ \st stp sng yld ->
178178 unStream (K.serial (toStream m1) (toStream m2))
179− (rstState st) stp sng yld
179+ st stp sng yld
180180
181181------------------------------------------------------------------------------
182182-- Monad
@@ -184,11 +184,11 @@ serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
184184
185185instance Monad m => Monad (SerialT m) where
186186 return = pure
187− (SerialT (Stream m)) >>= f = SerialT $ Stream $ \st stp sng yld ->
188− let run x = unStream x (rstState st) stp sng yld
187+ (SerialT m) >>= f = SerialT $ mkStream $ \st stp sng yld ->
188+ let run x = unStream x st stp sng yld
189189 single a = run $ toStream (f a)
190190 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)
191− in m (rstState st) stp single yieldk
191+ in unStream m st stp single yieldk
192192
193193------------------------------------------------------------------------------
194194-- Other instances
@@ -306,11 +306,11 @@ instance IsStream WSerialT where
306306
307307{-# INLINE interleave #-}
308308interleave :: Stream m a -> Stream m a -> Stream m a
309−interleave m1 m2 = Stream $ \st stp sng yld -> do
310− let stop = unStream m2 (rstState st) stp sng yld
309+interleave m1 m2 = mkStream $ \st stp sng yld -> do
310+ let stop = unStream m2 st stp sng yld
311311 single a = yld a m2
312312 yieldk a r = yld a (interleave m2 r)
313− unStream m1 (rstState st) stop single yieldk
313+ unStream m1 st stop single yieldk
314314
315315-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
316316-- Interleaves two streams, yielding one element from each stream alternately.
@@ -318,9 +318,9 @@ interleave m1 m2 = Stream $ \st stp sng yld -> do
318318-- @since 0.2.0
319319{-# INLINE wSerial #-}
320320wSerial :: IsStream t => t m a -> t m a -> t m a
321−wSerial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
321+wSerial m1 m2 = fromStream $ mkStream $ \st stp sng yld ->
322322 unStream (interleave (toStream m1) (toStream m2))
323− (rstState st) stp sng yld
323+ st stp sng yld
324324
325325instance Semigroup (WSerialT m a) where
326326 (<>) = wSerial
@@ -349,11 +349,11 @@ instance Monoid (WSerialT m a) where
349349
350350instance Monad m => Monad (WSerialT m) where
351351 return = pure
352− (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \st stp sng yld ->
353− let run x = unStream x (rstState st) stp sng yld
352+ (WSerialT m) >>= f = WSerialT $ mkStream $ \st stp sng yld ->
353+ let run x = unStream x st stp sng yld
354354 single a = run $ toStream (f a)
355355 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)
356− in m (rstState st) stp single yieldk
356+ in unStream m st stp single yieldk
357357
358358------------------------------------------------------------------------------
359359-- Other instances
src/Streamly/Streams/StreamD.hs+3−3
@@ -501,7 +501,7 @@ fromStreamK = Stream step
501501 let stop = return Stop
502502 single a = return $ Yield a K.nil
503503 yieldk a r = return $ Yield a r
504− in K.unStream m1 gst stop single yieldk
504+ in K.unStreamShared m1 gst stop single yieldk
505505
506506{-# INLINE toStreamD #-}
507507toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a
@@ -897,11 +897,11 @@ toList = foldr (:) []
897897toStreamK :: Monad m => Stream m a -> K.Stream m a
898898toStreamK (Stream step state) = go state
899899 where
900− go st = K.Stream $ \gst stp sng yld -> do
900+ go st = K.mkStream $ \gst stp sng yld -> do
901901 r <- step gst st
902902 case r of
903903 Yield x s -> yld x (go s)
904− Skip s -> K.unStream (go s) gst stp sng yld
904+ Skip s -> K.unStreamShared (go s) gst stp sng yld
905905 Stop -> stp
906906
907907#ifndef DISABLE_FUSION
src/Streamly/Streams/StreamK.hs+79−83
@@ -32,8 +32,8 @@ module Streamly.Streams.StreamK
3232 , adapt
3333
3434 -- * The stream type
35− , Stream (..)
36− , unStreamIsolated
35+ , Stream
36+ , unStream
3737 , isolateStream
3838 , unStreamShared
3939 , runStreamSVar
@@ -196,32 +196,28 @@ import Streamly.SVar
196196-- for Applicative and Monad, and in 'lift' for MonadTrans.
197197--
198198newtype Stream m a =
199− Stream {
200− unStream :: forall r.
199+ Stream (forall r.
201200 State Stream m a -- state
202201 -> m r -- stop
203202 -> (a -> m r) -- singleton
204203 -> (a -> Stream m a -> m r) -- yield
205204 -> m r
206− }
205+ )
207206
208−-- XXX make this the default "unStream"
209−-- | unwraps the Stream type producing the stream function that can be run with
210−-- continuations.
211−{-# INLINE unStreamIsolated #-}
212−unStreamIsolated ::
207+{-# INLINE unStream #-}
208+unStream ::
213209 Stream m a
214− -> State Stream m a -- state
210+ -> State Stream m b -- state
215211 -> m r -- stop
216212 -> (a -> m r) -- singleton
217213 -> (a -> Stream m a -> m r) -- yield
218214 -> m r
219−unStreamIsolated x st = unStream x (rstState st)
215+unStream (Stream runner) st = runner (rstState st)
220216
221217{-# INLINE isolateStream #-}
222218isolateStream :: Stream m a -> Stream m a
223219isolateStream x = Stream $ \st stp sng yld ->
224− unStreamIsolated x st stp sng yld
220+ unStream x st stp sng yld
225221
226222-- | Like unstream, but passes a shared SVar across continuations.
227223{-# INLINE unStreamShared #-}
@@ -232,7 +228,7 @@ unStreamShared ::
232228 -> (a -> m r) -- singleton
233229 -> (a -> Stream m a -> m r) -- yield
234230 -> m r
235−unStreamShared = unStream
231+unStreamShared (Stream runner) st stp sng yld = runner st stp sng yld
236232
237233-- Run the stream using a run function associated with the SVar that runs the
238234-- streams with a captured snapshot of the monadic state.
@@ -248,7 +244,7 @@ runStreamSVar
248244 -> m ()
249245runStreamSVar sv m st stp sng yld =
250246 let mrun = runInIO $ svarMrun sv
251− in void $ liftIO $ mrun $ unStream m st stp sng yld
247+ in void $ liftIO $ mrun $ unStreamShared m st stp sng yld
252248
253249------------------------------------------------------------------------------
254250-- Types that can behave as a Stream
@@ -332,7 +328,7 @@ mkStream:: IsStream t
332328 -> t m a
333329mkStream k = fromStream $ Stream $ \st stp sng yld ->
334330 let yieldk a r = yld a (toStream r)
335− in k (rstState st) stp sng yieldk
331+ in k st stp sng yieldk
336332
337333------------------------------------------------------------------------------
338334-- Construction
@@ -430,7 +426,7 @@ uncons m =
430426 let stop = return Nothing
431427 single a = return (Just (a, nil))
432428 yieldk a r = return (Just (a, fromStream r))
433− in unStream (toStream m) defState stop single yieldk
429+ in unStreamShared (toStream m) defState stop single yieldk
434430
435431-------------------------------------------------------------------------------
436432-- Generation
@@ -454,7 +450,7 @@ unfoldrM step = go
454450 case mayb of
455451 Nothing -> stp
456452 Just (a, b) ->
457− unStream (toStream (return a |: go b)) svr stp sng yld
453+ unStreamShared (toStream (return a |: go b)) svr stp sng yld
458454
459455-------------------------------------------------------------------------------
460456-- Special generation
@@ -537,7 +533,7 @@ foldStream
537533 -> m r
538534foldStream st blank single step m =
539535 let yieldk a x = step a (fromStream x)
540− in unStream (toStream m) st blank single yieldk
536+ in unStreamShared (toStream m) st blank single yieldk
541537
542538-- | Lazy right associative fold.
543539foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b
@@ -547,7 +543,7 @@ foldr step acc m = go (toStream m)
547543 let stop = return acc
548544 single a = return (step a acc)
549545 yieldk a r = go r >>= \b -> return (step a b)
550− in unStream m1 defState stop single yieldk
546+ in unStreamShared m1 defState stop single yieldk
551547
552548-- | Lazy right fold with a monadic step function.
553549{-# INLINE foldrM #-}
@@ -558,7 +554,7 @@ foldrM step acc m = go (toStream m)
558554 let stop = return acc
559555 single a = step a acc
560556 yieldk a r = go r >>= step a
561− in unStream m1 defState stop single yieldk
557+ in unStreamShared m1 defState stop single yieldk
562558
563559{-# INLINE foldr1 #-}
564560foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)
@@ -572,7 +568,7 @@ foldr1 step m = do
572568 let stp = return p
573569 single a = return $ step a p
574570 yieldk a r = fmap (step p) (go a r)
575− in unStream m1 defState stp single yieldk
571+ in unStreamShared m1 defState stp single yieldk
576572
577573-- | Strict left fold with an extraction function. Like the standard strict
578574-- left fold, but applies a user supplied extraction function (the third
@@ -590,7 +586,7 @@ foldx step begin done m = get $ go (toStream m) begin
590586 -- XXX we are not strictly evaluating the accumulator here. Is this
591587 -- okay?
592588 let single = return . done
593− in unStream m1 undefined undefined single undefined
589+ in unStreamShared m1 undefined undefined single undefined
594590
595591 -- Note, this can be implemented by making a recursive call to "go",
596592 -- however that is more expensive because of unnecessary recursion
@@ -601,8 +597,8 @@ foldx step begin done m = get $ go (toStream m) begin
601597 single a = sng $ step acc a
602598 yieldk a r =
603599 let stream = go r (step acc a)
604− in unStream stream defState undefined sng yld
605− in unStream m1 defState stop single yieldk
600+ in unStreamShared stream defState undefined sng yld
601+ in unStreamShared m1 defState stop single yieldk
606602
607603-- | Strict left associative fold.
608604{-# INLINE foldl' #-}
@@ -619,7 +615,7 @@ foldxM step begin done m = go begin (toStream m)
619615 let stop = acc >>= done
620616 single a = acc >>= \b -> step b a >>= done
621617 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r
622− in unStream m1 defState stop single yieldk
618+ in unStreamShared m1 defState stop single yieldk
623619
624620-- | Like 'foldl'' but with a monadic step function.
625621foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b
@@ -637,7 +633,7 @@ runStream m = go (toStream m)
637633 let stop = return ()
638634 single _ = return ()
639635 yieldk _ r = go (toStream r)
640− in unStream m1 defState stop single yieldk
636+ in unStreamShared m1 defState stop single yieldk
641637
642638{-# INLINE null #-}
643639null :: (IsStream t, Monad m) => t m a -> m Bool
@@ -645,7 +641,7 @@ null m =
645641 let stop = return True
646642 single _ = return False
647643 yieldk _ _ = return False
648− in unStream (toStream m) defState stop single yieldk
644+ in unStreamShared (toStream m) defState stop single yieldk
649645
650646{-# INLINE head #-}
651647head :: (IsStream t, Monad m) => t m a -> m (Maybe a)
@@ -653,7 +649,7 @@ head m =
653649 let stop = return Nothing
654650 single a = return (Just a)
655651 yieldk a _ = return (Just a)
656− in unStream (toStream m) defState stop single yieldk
652+ in unStreamShared (toStream m) defState stop single yieldk
657653
658654{-# INLINE tail #-}
659655tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
@@ -661,7 +657,7 @@ tail m =
661657 let stop = return Nothing
662658 single _ = return $ Just nil
663659 yieldk _ r = return $ Just $ fromStream r
664− in unStream (toStream m) defState stop single yieldk
660+ in unStreamShared (toStream m) defState stop single yieldk
665661
666662{-# INLINE init #-}
667663init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))
@@ -675,7 +671,7 @@ init m = go1 (toStream m)
675671 go p m1 = Stream $ \_ stp sng yld ->
676672 let single _ = sng p
677673 yieldk a x = yld p $ go a x
678− in unStream m1 defState stp single yieldk
674+ in unStreamShared m1 defState stp single yieldk
679675
680676{-# INLINE elem #-}
681677elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
@@ -685,7 +681,7 @@ elem e m = go (toStream m)
685681 let stop = return False
686682 single a = return (a == e)
687683 yieldk a r = if a == e then return True else go r
688− in unStream m1 defState stop single yieldk
684+ in unStreamShared m1 defState stop single yieldk
689685
690686{-# INLINE notElem #-}
691687notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool
@@ -695,7 +691,7 @@ notElem e m = go (toStream m)
695691 let stop = return True
696692 single a = return (a /= e)
697693 yieldk a r = if a == e then return False else go r
698− in unStream m1 defState stop single yieldk
694+ in unStreamShared m1 defState stop single yieldk
699695
700696all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
701697all p m = go (toStream m)
@@ -705,7 +701,7 @@ all p m = go (toStream m)
705701 | otherwise = return False
706702 yieldk a r | p a = go r
707703 | otherwise = return False
708− in unStream m1 defState (return True) single yieldk
704+ in unStreamShared m1 defState (return True) single yieldk
709705
710706any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool
711707any p m = go (toStream m)
@@ -715,7 +711,7 @@ any p m = go (toStream m)
715711 | otherwise = return False
716712 yieldk a r | p a = return True
717713 | otherwise = go r
718− in unStream m1 defState (return False) single yieldk
714+ in unStreamShared m1 defState (return False) single yieldk
719715
720716-- | Extract the last element of the stream, if any.
721717{-# INLINE last #-}
@@ -730,7 +726,7 @@ minimum m = go Nothing (toStream m)
730726 let stop = return Nothing
731727 single a = return (Just a)
732728 yieldk a r = go (Just a) r
733− in unStream m1 defState stop single yieldk
729+ in unStreamShared m1 defState stop single yieldk
734730
735731 go (Just res) m1 =
736732 let stop = return (Just res)
@@ -742,7 +738,7 @@ minimum m = go Nothing (toStream m)
742738 if res <= a
743739 then go (Just res) r
744740 else go (Just a) r
745− in unStream m1 defState stop single yieldk
741+ in unStreamShared m1 defState stop single yieldk
746742
747743{-# INLINE minimumBy #-}
748744minimumBy
@@ -754,7 +750,7 @@ minimumBy cmp m = go Nothing (toStream m)
754750 let stop = return Nothing
755751 single a = return (Just a)
756752 yieldk a r = go (Just a) r
757− in unStream m1 defState stop single yieldk
753+ in unStreamShared m1 defState stop single yieldk
758754
759755 go (Just res) m1 =
760756 let stop = return (Just res)
@@ -764,7 +760,7 @@ minimumBy cmp m = go Nothing (toStream m)
764760 yieldk a r = case cmp res a of
765761 GT -> go (Just a) r
766762 _ -> go (Just res) r
767− in unStream m1 defState stop single yieldk
763+ in unStreamShared m1 defState stop single yieldk
768764
769765{-# INLINE maximum #-}
770766maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)
@@ -774,7 +770,7 @@ maximum m = go Nothing (toStream m)
774770 let stop = return Nothing
775771 single a = return (Just a)
776772 yieldk a r = go (Just a) r
777− in unStream m1 defState stop single yieldk
773+ in unStreamShared m1 defState stop single yieldk
778774
779775 go (Just res) m1 =
780776 let stop = return (Just res)
@@ -786,7 +782,7 @@ maximum m = go Nothing (toStream m)
786782 if res <= a
787783 then go (Just a) r
788784 else go (Just res) r
789− in unStream m1 defState stop single yieldk
785+ in unStreamShared m1 defState stop single yieldk
790786
791787{-# INLINE maximumBy #-}
792788maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)
@@ -796,7 +792,7 @@ maximumBy cmp m = go Nothing (toStream m)
796792 let stop = return Nothing
797793 single a = return (Just a)
798794 yieldk a r = go (Just a) r
799− in unStream m1 defState stop single yieldk
795+ in unStreamShared m1 defState stop single yieldk
800796
801797 go (Just res) m1 =
802798 let stop = return (Just res)
@@ -806,7 +802,7 @@ maximumBy cmp m = go Nothing (toStream m)
806802 yieldk a r = case cmp res a of
807803 GT -> go (Just res) r
808804 _ -> go (Just a) r
809− in unStream m1 defState stop single yieldk
805+ in unStreamShared m1 defState stop single yieldk
810806
811807{-# INLINE (!!) #-}
812808(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)
@@ -818,7 +814,7 @@ m !! i = go i (toStream m)
818814 yieldk a x | n < 0 = return Nothing
819815 | n == 0 = return $ Just a
820816 | otherwise = go (n - 1) x
821− in unStream m1 defState (return Nothing) single yieldk
817+ in unStreamShared m1 defState (return Nothing) single yieldk
822818
823819{-# INLINE lookup #-}
824820lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)
@@ -829,7 +825,7 @@ lookup e m = go (toStream m)
829825 | otherwise = return Nothing
830826 yieldk (a, b) x | a == e = return $ Just b
831827 | otherwise = go x
832− in unStream m1 defState (return Nothing) single yieldk
828+ in unStreamShared m1 defState (return Nothing) single yieldk
833829
834830{-# INLINE findM #-}
835831findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)
@@ -842,7 +838,7 @@ findM p m = go (toStream m)
842838 yieldk a x = do
843839 b <- p a
844840 if b then return $ Just a else go x
845− in unStream m1 defState (return Nothing) single yieldk
841+ in unStreamShared m1 defState (return Nothing) single yieldk
846842
847843{-# INLINE find #-}
848844find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)
@@ -857,7 +853,7 @@ findIndices p = fromStream . go 0 . toStream
857853 | otherwise = stp
858854 yieldk a x | p a = yld offset $ go (offset + 1) x
859855 | otherwise = unStream (go (offset + 1) x) st stp sng yld
860− in unStream m1 (rstState st) stp single yieldk
856+ in unStream m1 st stp single yieldk
861857
862858------------------------------------------------------------------------------
863859-- Map and Fold
@@ -872,7 +868,7 @@ mapM_ f m = go (toStream m)
872868 let stop = return ()
873869 single a = void (f a)
874870 yieldk a r = f a >> go r
875− in unStream m1 defState stop single yieldk
871+ in unStreamShared m1 defState stop single yieldk
876872
877873------------------------------------------------------------------------------
878874-- Converting folds
@@ -900,7 +896,7 @@ scanx step begin done m =
900896 yieldk a r =
901897 let s = step acc a
902898 in yld (done s) (go r s)
903− in unStream m1 (rstState st) stp single yieldk
899+ in unStream m1 st stp single yieldk
904900
905901{-# INLINE scanl' #-}
906902scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b
@@ -918,8 +914,8 @@ filter p m = fromStream $ go (toStream m)
918914 let single a | p a = sng a
919915 | otherwise = stp
920916 yieldk a r | p a = yld a (go r)
921− | otherwise = unStream r (rstState st) stp single yieldk
922− in unStream m1 (rstState st) stp single yieldk
917+ | otherwise = unStream r st stp single yieldk
918+ in unStream m1 st stp single yieldk
923919
924920{-# INLINE take #-}
925921take :: IsStream t => Int -> t m a -> t m a
@@ -929,7 +925,7 @@ take n m = fromStream $ go n (toStream m)
929925 let yieldk a r = yld a (go (n1 - 1) r)
930926 in if n1 <= 0
931927 then stp
932− else unStream m1 (rstState st) stp sng yieldk
928+ else unStream m1 st stp sng yieldk
933929
934930{-# INLINE takeWhile #-}
935931takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
@@ -940,7 +936,7 @@ takeWhile p m = fromStream $ go (toStream m)
940936 | otherwise = stp
941937 yieldk a r | p a = yld a (go r)
942938 | otherwise = stp
943− in unStream m1 (rstState st) stp single yieldk
939+ in unStream m1 st stp single yieldk
944940
945941drop :: IsStream t => Int -> t m a -> t m a
946942drop n m = fromStream $ Stream $ \st stp sng yld ->
@@ -951,8 +947,8 @@ drop n m = fromStream $ Stream $ \st stp sng yld ->
951947 yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld
952948 -- Somehow "<=" check performs better than a ">"
953949 in if n1 <= 0
954− then unStream m1 st stp sng yld
955− else unStream m1 st stp single yieldk
950+ then unStreamShared m1 st stp sng yld
951+ else unStreamShared m1 st stp single yieldk
956952
957953{-# INLINE dropWhile #-}
958954dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
@@ -961,9 +957,9 @@ dropWhile p m = fromStream $ go (toStream m)
961957 go m1 = Stream $ \st stp sng yld ->
962958 let single a | p a = stp
963959 | otherwise = sng a
964− yieldk a r | p a = unStream r (rstState st) stp single yieldk
960+ yieldk a r | p a = unStream r st stp single yieldk
965961 | otherwise = yld a r
966− in unStream m1 (rstState st) stp single yieldk
962+ in unStream m1 st stp single yieldk
967963
968964-------------------------------------------------------------------------------
969965-- Mapping
@@ -974,7 +970,7 @@ map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
974970map f m = fromStream $ Stream $ \st stp sng yld ->
975971 let single = sng . f
976972 yieldk a r = yld (f a) (fmap f r)
977− in unStream (toStream m) (rstState st) stp single yieldk
973+ in unStream (toStream m) st stp single yieldk
978974
979975-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
980976-- other stream types to overload it.
@@ -984,8 +980,8 @@ mapM f m = go (toStream m)
984980 where
985981 go m1 = fromStream $ Stream $ \st stp sng yld ->
986982 let single a = f a >>= sng
987− yieldk a r = unStream (toStream (f a |: go r)) st stp sng yld
988− in unStream m1 (rstState st) stp single yieldk
983+ yieldk a r = unStreamShared (toStream (f a |: go r)) st stp sng yld
984+ in unStream m1 st stp single yieldk
989985
990986-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
991987-- other stream types to overload it.
@@ -995,8 +991,8 @@ sequence m = go (toStream m)
995991 where
996992 go m1 = fromStream $ Stream $ \st stp sng yld ->
997993 let single ma = ma >>= sng
998− yieldk ma r = unStream (toStream $ ma |: go r) st stp sng yld
999− in unStream m1 (rstState st) stp single yieldk
994+ yieldk ma r = unStreamShared (toStream $ ma |: go r) st stp sng yld
995+ in unStream m1 st stp single yieldk
1000996
1001997-------------------------------------------------------------------------------
1002998-- Inserting
@@ -1007,12 +1003,12 @@ intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
10071003intersperseM a m = fromStream $ prependingStart (toStream m)
10081004 where
10091005 prependingStart m1 = Stream $ \st stp sng yld ->
1010− let yieldk i x = unStream (return i |: go x) st stp sng yld
1011− in unStream m1 (rstState st) stp sng yieldk
1006+ let yieldk i x = unStreamShared (return i |: go x) st stp sng yld
1007+ in unStream m1 st stp sng yieldk
10121008 go m2 = fromStream $ Stream $ \st stp sng yld ->
1013− let single i = unStream (a |: yield i) st stp sng yld
1014− yieldk i x = unStream (a |: return i |: go x) st stp sng yld
1015− in unStream m2 (rstState st) stp single yieldk
1009+ let single i = unStreamShared (a |: yield i) st stp sng yld
1010+ yieldk i x = unStreamShared (a |: return i |: go x) st stp sng yld
1011+ in unStream m2 st stp single yieldk
10161012
10171013{-# INLINE insertBy #-}
10181014insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a
@@ -1026,7 +1022,7 @@ insertBy cmp x m = fromStream $ go (toStream m)
10261022 yieldk a r = case cmp x a of
10271023 GT -> yld a (go r)
10281024 _ -> yld x (a `cons` r)
1029− in unStream m1 (rstState st) stop single yieldk
1025+ in unStream m1 st stop single yieldk
10301026
10311027------------------------------------------------------------------------------
10321028-- Deleting
@@ -1039,9 +1035,9 @@ deleteBy eq x m = fromStream $ go (toStream m)
10391035 go m1 = Stream $ \st stp sng yld ->
10401036 let single a = if eq x a then stp else sng a
10411037 yieldk a r = if eq x a
1042− then unStream r (rstState st) stp sng yld
1038+ then unStream r st stp sng yld
10431039 else yld a (go r)
1044− in unStream m1 (rstState st) stp single yieldk
1040+ in unStream m1 st stp single yieldk
10451041
10461042-------------------------------------------------------------------------------
10471043-- Map and Filter
@@ -1057,8 +1053,8 @@ mapMaybe f m = go (toStream m)
10571053 Nothing -> stp
10581054 yieldk a r = case f a of
10591055 Just b -> yld b (toStream $ go r)
1060− Nothing -> unStream r (rstState st) stp single yieldk
1061− in unStream m1 (rstState st) stp single yieldk
1056+ Nothing -> unStream r st stp single yieldk
1057+ in unStream m1 st stp single yieldk
10621058
10631059------------------------------------------------------------------------------
10641060-- Serial Zipping
@@ -1072,10 +1068,10 @@ zipWithS f = go
10721068 let merge a ra =
10731069 let single2 b = sng (f a b)
10741070 yield2 b rb = yld (f a b) (go ra rb)
1075− in unStream my (rstState st) stp single2 yield2
1071+ in unStream my st stp single2 yield2
10761072 let single1 a = merge a nil
10771073 yield1 = merge
1078− unStream mx (rstState st) stp single1 yield1
1074+ unStream mx st stp single1 yield1
10791075
10801076-- | Zip two streams serially using a pure zipping function.
10811077--
@@ -1092,13 +1088,13 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
10921088 where
10931089 go mx my = Stream $ \st stp sng yld -> do
10941090 let merge a ra =
1095− let runIt x = unStream x (rstState st) stp sng yld
1091+ let runIt x = unStream x st stp sng yld
10961092 single2 b = f a b >>= sng
10971093 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)
1098− in unStream my (rstState st) stp single2 yield2
1094+ in unStream my st stp single2 yield2
10991095 let single1 a = merge a nil
11001096 yield1 = merge
1101− unStream mx (rstState st) stp single1 yield1
1097+ unStream mx st stp single1 yield1
11021098
11031099------------------------------------------------------------------------------
11041100-- Merging
@@ -1158,7 +1154,7 @@ the m = do
11581154 | otherwise = return Nothing
11591155 yieldk a r | h == a = go h r
11601156 | otherwise = return Nothing
1161− in unStream m1 defState (return $ Just h) single yieldk
1157+ in unStreamShared m1 defState (return $ Just h) single yieldk
11621158
11631159------------------------------------------------------------------------------
11641160-- Semigroup
@@ -1171,7 +1167,7 @@ serial :: Stream m a -> Stream m a -> Stream m a
11711167serial m1 m2 = go m1
11721168 where
11731169 go (Stream m) = Stream $ \st stp sng yld ->
1174− let stop = unStream m2 (rstState st) stp sng yld
1170+ let stop = unStream m2 st stp sng yld
11751171 single a = yld a m2
11761172 yieldk a r = yld a (go r)
11771173 in m (rstState st) stop single yieldk
@@ -1209,7 +1205,7 @@ bindWith par m f = go m
12091205 go (Stream g) =
12101206 Stream $ \st stp sng yld ->
12111207 let runShared x = unStreamShared x st stp sng yld
1212− runIsolated x = unStreamIsolated x st stp sng yld
1208+ runIsolated x = unStream x st stp sng yld
12131209
12141210 single a = runIsolated $ f a
12151211 yieldk a r = runShared $ isolateStream (f a) `par` go r
@@ -1221,8 +1217,8 @@ bindWith par m f = go m
12211217
12221218_alt :: Stream m a -> Stream m a -> Stream m a
12231219_alt m1 m2 = Stream $ \st stp sng yld ->
1224− let stop = unStream m2 (rstState st) stp sng yld
1225− in unStream m1 (rstState st) stop sng yld
1220+ let stop = unStream m2 st stp sng yld
1221+ in unStream m1 st stop sng yld
12261222
12271223------------------------------------------------------------------------------
12281224-- MonadReader
@@ -1233,7 +1229,7 @@ withLocal f m =
12331229 Stream $ \st stp sng yld ->
12341230 let single = local f . sng
12351231 yieldk a r = local f $ yld a (withLocal f r)
1236− in unStream m (rstState st) (local f stp) single yieldk
1232+ in unStream m st (local f stp) single yieldk
12371233
12381234------------------------------------------------------------------------------
12391235-- MonadError
src/Streamly/Streams/Zip.hs+5−5
@@ -48,7 +48,7 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
4848 readListPrecDefault)
4949import Prelude hiding (map, repeat, zipWith)
5050
51−import Streamly.Streams.StreamK (IsStream(..), Stream(..))
51+import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, unStream)
5252import Streamly.Streams.Async (mkAsync')
5353import Streamly.Streams.Serial (map)
5454import Streamly.SVar (MonadAsync, rstState)
@@ -142,10 +142,10 @@ TRAVERSABLE_INSTANCE(ZipSerialM)
142142-- @since 0.1.0
143143zipAsyncWith :: (IsStream t, MonadAsync m)
144144 => (a -> b -> c) -> t m a -> t m b -> t m c
145−zipAsyncWith f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do
145+zipAsyncWith f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do
146146 ma <- mkAsync' (rstState st) m1
147147 mb <- mkAsync' (rstState st) m2
148− unStream (toStream (K.zipWith f ma mb)) (rstState st) stp sng yld
148+ 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
151151-- are generated concurrently.
@@ -153,10 +153,10 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do
153153-- @since 0.4.0
154154zipAsyncWithM :: (IsStream t, MonadAsync m)
155155 => (a -> b -> m c) -> t m a -> t m b -> t m c
156−zipAsyncWithM f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do
156+zipAsyncWithM f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do
157157 ma <- mkAsync' (rstState st) m1
158158 mb <- mkAsync' (rstState st) m2
159− unStream (toStream (K.zipWithM f ma mb)) (rstState st) stp sng yld
159+ unStream (toStream (K.zipWithM f ma mb)) st stp sng yld
160160
161161------------------------------------------------------------------------------
162162-- Parallely Zipping Streams