924856b · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Make the default unStream safe, #96.

924856bPranay Sashank8 years ago

src/Streamly/Prelude.hs+9−9

compare with master

unifiedsplit

@@ -840,9 +840,9 @@ iterate step = fromStream . go

840iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a

841iterateM step = go

842 where

843− go s = fromStream $ K.Stream $ \svr stp sng yld -> do

843+ go s = fromStream $ K.mkStream $ \svr stp sng yld -> do

844 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

846

847------------------------------------------------------------------------------

848-- Conversions

@@ -896,7 +896,7 @@ each = K.fromFoldable

896fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String

897fromHandle h = fromStream go

898 where

899− go = K.Stream $ \_ stp _ yld -> do

899+ go = K.mkStream $ \_ stp _ yld -> do

900 eof <- liftIO $ IO.hIsEOF h

901 if eof

902 then stp

@@ -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.unStream m1 defState stop single yieldk

1452+ in K.unStreamShared m1 defState stop single yieldk

1453

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

1455-- Transformation by Folding (Scans)

@@ -1770,12 +1770,12 @@ mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m

1770reverse :: (IsStream t) => t m a -> t m a

1771reverse m = fromStream $ go K.nil (toStream m)

1772 where

1773− go rev rest = K.Stream $ \st stp sng yld ->

1773+ go rev rest = K.mkStream $ \st stp sng yld ->

1774− let runIt x = K.unStream x (rstState st) stp sng yld

1774+ let runIt x = K.unStream x st stp sng yld

1775 stop = runIt rev

1776 single a = runIt $ a `K.cons` rev

1777 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

1779

1780------------------------------------------------------------------------------

1781-- Transformation by Inserting

@@ -1986,7 +1986,7 @@ merge = mergeBy compare

1986-- @since 0.6.0

1987mergeAsyncBy :: (IsStream t, MonadAsync m)

1988 => (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

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

@@ -1997,7 +1997,7 @@ mergeAsyncBy f m1 m2 = K.fromStream $ K.Stream $ \st stp sng yld -> do

1997-- @since 0.6.0

1998mergeAsyncByM :: (IsStream t, MonadAsync m)

1999 => (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

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

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

@@ -48,7 +48,7 @@ import Streamly.Streams.SVar (fromSVar)

48import Streamly.Streams.Serial (map)

49import Streamly.SVar

50import Streamly.Streams.StreamK

51− (IsStream(..), Stream(..), unStreamShared, unStreamIsolated,

51+ (IsStream(..), Stream, mkStream, unStream, unStreamShared,

52 runStreamSVar)

53import qualified Streamly.Streams.StreamK as K

@@ -544,25 +544,25 @@ workLoopAhead q heap st sv winfo = do

544-- The only difference between forkSVarAsync and this is that we run the left

545-- computation without a shared SVar.

546forkSVarAhead :: 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

548 sv <- newAheadVar st (concurrently m1 m2) workLoopAhead

549− unStream (fromSVar sv) (rstState st) stp sng yld

549+ unStream (fromSVar sv) st stp sng yld

550 where

551− concurrently ma mb = Stream $ \st stp sng yld -> do

551+ concurrently ma mb = mkStream $ \st stp sng yld -> do

552 liftIO $ enqueue (fromJust $ streamVar st) mb

553− unStream ma (rstState st) stp sng yld

553+ unStream ma st stp sng yld

554

555{-# INLINE aheadS #-}

556aheadS :: 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 ->

558 case streamVar st of

559 Just sv | svarStyle sv == AheadVar -> do

560 liftIO $ enqueue sv m2

561 -- Always run the left side on a new SVar to avoid complexity in

562 -- sequencing results. This means the left side cannot further

563 -- split into more ahead computations on the same SVar.

564− unStream m1 (rstState st) stp sng yld

564+ unStream m1 st stp sng yld

565− _ -> unStream (forkSVarAhead m1 m2) st stp sng yld

565+ _ -> unStreamShared (forkSVarAhead m1 m2) st stp sng yld

566

567-- | XXX we can implement it more efficienty by directly implementing instead

568-- of combining streams using ahead.

@@ -655,8 +655,8 @@ instance IsStream AheadT where

655-- @since 0.3.0

656{-# INLINE ahead #-}

657ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

658−ahead m1 m2 = fromStream $ Stream $ \st stp sng yld ->

658+ahead m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

659− unStream (aheadS (toStream m1) (toStream m2)) st stp sng yld

659+ unStreamShared (aheadS (toStream m1) (toStream m2)) st stp sng yld

660

661instance MonadAsync m => Semigroup (AheadT m a) where

662 (<>) = ahead

@@ -681,15 +681,15 @@ aheadbind

681 -> Stream m b

682aheadbind m f = go m

683 where

684− go (Stream g) =

684+ go g =

685− Stream $ \st stp sng yld ->

685+ mkStream $ \st stp sng yld ->

686 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

688

689 single a = runIsolated $ f a

690 yieldk a r = runShared $

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

692− in g (rstState st) stp single yieldk

692+ in unStream g st stp single yieldk

693

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

695 return = pure

src/Streamly/Streams/Async.hs+13−12

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

57import Streamly.Streams.SVar (fromSVar)

58import Streamly.Streams.Serial (map)

59import Streamly.SVar

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

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

61+ unStreamShared, runStreamSVar)

61import qualified Streamly.Streams.StreamK as K

62import qualified Streamly.Streams.StreamK as K

63#include "Instances.hs"

64#include "Instances.hs"

@@ -528,25 +529,25 @@ newWAsyncVar st m = do

528

529

529forkSVarAsync :: MonadAsync m

530forkSVarAsync :: MonadAsync m

530 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

531 => 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

532 sv <- case style of

533 sv <- case style of

533 AsyncVar -> newAsyncVar st (concurrently m1 m2)

534 AsyncVar -> newAsyncVar st (concurrently m1 m2)

534 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)

535 WAsyncVar -> newWAsyncVar st (concurrently m1 m2)

535 _ -> error "illegal svar type"

536 _ -> error "illegal svar type"

536− unStream (fromSVar sv) (rstState st) stp sng yld

537+ unStream (fromSVar sv) st stp sng yld

537 where

538 where

538− concurrently ma mb = Stream $ \st stp sng yld -> do

539+ concurrently ma mb = mkStream $ \st stp sng yld -> do

539 liftIO $ enqueue (fromJust $ streamVar st) mb

540 liftIO $ enqueue (fromJust $ streamVar st) mb

540− unStream ma st stp sng yld

541+ unStreamShared ma st stp sng yld

541

542

542{-# INLINE joinStreamVarAsync #-}

543{-# INLINE joinStreamVarAsync #-}

543joinStreamVarAsync :: MonadAsync m

544joinStreamVarAsync :: MonadAsync m

544 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

545 => 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 ->

546 case streamVar st of

547 case streamVar st of

547 Just sv | svarStyle sv == style ->

548 Just sv | svarStyle sv == style ->

548− liftIO (enqueue sv m2) >> unStream m1 st stp sng yld

549+ liftIO (enqueue sv m2) >> unStreamShared m1 st stp sng yld

549− _ -> unStream (forkSVarAsync style m1 m2) st stp sng yld

550+ _ -> unStreamShared (forkSVarAsync style m1 m2) st stp sng yld

550

551

551------------------------------------------------------------------------------

552------------------------------------------------------------------------------

552-- Semigroup and Monoid style compositions for parallel actions

553-- Semigroup and Monoid style compositions for parallel actions

@@ -563,8 +564,8 @@ asyncS = joinStreamVarAsync AsyncVar

563-- @since 0.2.0

564-- @since 0.2.0

564{-# INLINE async #-}

565{-# INLINE async #-}

565async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

566async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

566−async m1 m2 = fromStream $ Stream $ \st stp sng yld ->

567+async m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

567− unStream (joinStreamVarAsync AsyncVar (toStream m1) (toStream m2))

568+ unStreamShared (joinStreamVarAsync AsyncVar (toStream m1) (toStream m2))

568 st stp sng yld

569 st stp sng yld

569

570

570-- | Same as 'async'.

571-- | Same as 'async'.

@@ -711,8 +712,8 @@ consMWAsync m r = K.yieldM m `wAsyncS` r

711-- @since 0.2.0

712-- @since 0.2.0

712{-# INLINE wAsync #-}

713{-# INLINE wAsync #-}

713wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

714wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

714−wAsync m1 m2 = fromStream $ Stream $ \st stp sng yld ->

715+wAsync m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

715− unStream (wAsyncS (toStream m1) (toStream m2)) st stp sng yld

716+ unStreamShared (wAsyncS (toStream m1) (toStream m2)) st stp sng yld

716

717

717-- | Wide async composition or async composition with breadth first traversal.

718-- | Wide async composition or async composition with breadth first traversal.

718-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed

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

52-- @since 0.4.0

53{-# INLINE_NORMAL maxThreads #-}

54maxThreads :: IsStream t => Int -> t m a -> t m a

55−maxThreads n m = fromStream $ Stream $ \st stp sng yld ->

55+maxThreads n m = fromStream $ mkStream $ \st stp sng yld ->

56− unStream (toStream m) (setMaxThreads n st) stp sng yld

56+ unStreamShared (toStream m) (setMaxThreads n st) stp sng yld

58{-

59{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}

@@ -77,8 +77,8 @@ maxThreadsSerial _ = id

77-- @since 0.4.0

78{-# INLINE_NORMAL maxBuffer #-}

79maxBuffer :: IsStream t => Int -> t m a -> t m a

80−maxBuffer n m = fromStream $ Stream $ \st stp sng yld ->

80+maxBuffer n m = fromStream $ mkStream $ \st stp sng yld ->

81− unStream (toStream m) (setMaxBuffer n st) stp sng yld

81+ unStreamShared (toStream m) (setMaxBuffer n st) stp sng yld

83{-

84{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}

@@ -101,7 +101,7 @@ maxBufferSerial _ = id

101-- @since 0.5.0

102{-# INLINE_NORMAL rate #-}

103rate :: 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 ->

105 case r of

106 Just (Rate low goal _ _) | goal < low ->

107 error "rate: Target rate cannot be lower than minimum rate."

@@ -109,7 +109,7 @@ rate r m = fromStream $ Stream $ \st stp sng yld ->

109 error "rate: Target rate cannot be greater than maximum rate."

110 Just (Rate low _ high _) | low > high ->

111 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

113

114-- XXX implement for serial streams as well, as a simple delay

115

@@ -181,8 +181,8 @@ constRate r = rate (Just $ Rate r r r 0)

181--

182{-# INLINE_NORMAL _serialLatency #-}

183_serialLatency :: IsStream t => Int -> t m a -> t m a

184−_serialLatency n m = fromStream $ Stream $ \st stp sng yld ->

184+_serialLatency n m = fromStream $ mkStream $ \st stp sng yld ->

185− unStream (toStream m) (setStreamLatency n st) stp sng yld

185+ unStreamShared (toStream m) (setStreamLatency n st) stp sng yld

186

187{-

188{-# RULES "serialLatency serial" _serialLatency = serialLatencySerial #-}

@@ -196,8 +196,8 @@ serialLatencySerial _ = id

196-- inherited by everything in enclosed scope.

197{-# INLINE_NORMAL maxYields #-}

198maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a

199−maxYields n m = fromStream $ Stream $ \st stp sng yld ->

199+maxYields n m = fromStream $ mkStream $ \st stp sng yld ->

200− unStream (toStream m) (setYieldLimit n st) stp sng yld

200+ unStreamShared (toStream m) (setYieldLimit n st) stp sng yld

201

202{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}

203maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a

@@ -212,5 +212,5 @@ printState st = liftIO $ do

212

213-- | Print debug information about an SVar when the stream ends

214inspectMode :: IsStream t => t m a -> t m a

215−inspectMode m = fromStream $ Stream $ \st stp sng yld ->

215+inspectMode m = fromStream $ mkStream $ \st stp sng yld ->

216− unStream (toStream m) (setInspectMode 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)

49import Streamly.Streams.SVar (fromSVar)

50import Streamly.Streams.Serial (map)

51import Streamly.SVar

52−import Streamly.Streams.StreamK (IsStream(..), Stream(..), adapt)

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

53+ unStreamShared, adapt)

53import qualified Streamly.Streams.StreamK as K

54import qualified Streamly.Streams.StreamK as K

55#include "Instances.hs"

56#include "Instances.hs"

@@ -62,7 +63,7 @@ import qualified Streamly.Streams.StreamK as K

62runOne

63runOne

63 :: MonadIO m

64 :: MonadIO m

64 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()

65 => 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

67 where

68 where

@@ -88,21 +89,21 @@ runOne st m winfo = unStream m st stop single yieldk

89{-# NOINLINE forkSVarPar #-}

90{-# NOINLINE forkSVarPar #-}

90forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

91forkSVarPar :: 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

92 sv <- newParallelVar st

93 sv <- newParallelVar st

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

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

94 pushWorkerPar sv (runOne st{streamVar = Just sv} r)

95 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

97{-# INLINE joinStreamVarPar #-}

98{-# INLINE joinStreamVarPar #-}

98joinStreamVarPar :: MonadAsync m

99joinStreamVarPar :: MonadAsync m

99 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

100 => 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 ->

101 case streamVar st of

102 case streamVar st of

102 Just sv | svarStyle sv == style -> do

103 Just sv | svarStyle sv == style -> do

103 pushWorkerPar sv (runOne st m1)

104 pushWorkerPar sv (runOne st m1)

104− unStream m2 st stp sng yld

105+ unStreamShared m2 st stp sng yld

105− _ -> unStream (forkSVarPar m1 m2) st stp sng yld

106+ _ -> unStreamShared (forkSVarPar m1 m2) st stp sng yld

106

107

107{-# INLINE parallelStream #-}

108{-# INLINE parallelStream #-}

108parallelStream :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

109parallelStream :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -120,8 +121,8 @@ consMParallel m r = K.yieldM m `parallelStream` r

120-- @since 0.2.0

121-- @since 0.2.0

121{-# INLINE parallel #-}

122{-# INLINE parallel #-}

122parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

123parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

123−parallel m1 m2 = fromStream $ Stream $ \st stp sng yld ->

124+parallel m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

124− unStream (parallelStream (toStream m1) (toStream m2))

125+ unStreamShared (parallelStream (toStream m1) (toStream m2))

125 st stp sng yld

126 st stp sng yld

126

127

127------------------------------------------------------------------------------

128------------------------------------------------------------------------------

@@ -140,10 +141,10 @@ mkParallel m = do

140

141

141{-# INLINE applyWith #-}

142{-# INLINE applyWith #-}

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

143applyWith :: (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

144 sv <- newParallelVar (rstState st)

145 sv <- newParallelVar (rstState st)

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

146 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

147

148

148------------------------------------------------------------------------------

149------------------------------------------------------------------------------

149-- Stream runner concurrent function application

150-- Stream runner concurrent function application

src/Streamly/Streams/SVar.hs+9−9

@@ -38,12 +38,12 @@ printSVar sv how = do

38-- | Pull a stream from an SVar.

39{-# NOINLINE fromStreamVar #-}

40fromStreamVar :: 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

42 list <- readOutputQ sv

43 -- Reversing the output is important to guarantee that we process the

44 -- outputs in the same order as they were generated by the constituent

45 -- streams.

46− unStream (processEvents $ reverse list) (rstState st) stp sng yld

46+ unStream (processEvents $ reverse list) st stp sng yld

48 where

@@ -55,36 +55,36 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do

55 stp

57 {-# INLINE processEvents #-}

58− processEvents [] = Stream $ \st stp sng yld -> do

58+ processEvents [] = mkStream $ \st stp sng yld -> do

59 done <- postProcess sv

60 if done

61 then allDone stp

62− else unStream (fromStreamVar sv) (rstState st) stp sng yld

62+ else unStream (fromStreamVar sv) st stp sng yld

64− processEvents (ev : es) = Stream $ \st stp sng yld -> do

64+ processEvents (ev : es) = mkStream $ \st stp sng yld -> do

65 let rest = processEvents es

66 case ev of

67 ChildYield a -> yld a rest

68 ChildStop tid e -> do

69 accountThread sv tid

70 case e of

71− Nothing -> unStream rest (rstState st) stp sng yld

71+ Nothing -> unStream rest st stp sng yld

72 Just ex ->

73 case fromException ex of

74 Just ThreadAbort ->

75− unStream rest (rstState st) stp sng yld

75+ unStream rest st stp sng yld

76 Nothing -> liftIO (cleanupSVar sv) >> throwM ex

78{-# INLINE fromSVar #-}

79fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a

80fromSVar sv =

81− fromStream $ Stream $ \st stp sng yld -> do

81+ fromStream $ mkStream $ \st stp sng yld -> do

82 ref <- liftIO $ newIORef ()

83 _ <- liftIO $ mkWeakIORef ref hook

84 -- We pass a copy of sv to fromStreamVar, so that we know that it has

85 -- no other references, when that copy gets garbage collected "ref"

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

88 where

90 hook = do

src/Streamly/Streams/Serial.hs+15−15

@@ -60,8 +60,8 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,

60 readListPrecDefault)

61import Prelude hiding (map, mapM)

63−import Streamly.SVar (rstState)

64−import Streamly.Streams.StreamK (IsStream(..), adapt, Stream(..))

63+import Streamly.Streams.StreamK (IsStream(..), adapt, Stream, mkStream,

64+ unStream)

65import qualified Streamly.Streams.Prelude as P

66import qualified Streamly.Streams.StreamK as K

67import qualified Streamly.Streams.StreamD as D

@@ -174,9 +174,9 @@ instance IsStream SerialT where

174-- @since 0.2.0

175{-# INLINE serial #-}

176serial :: 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 ->

178 unStream (K.serial (toStream m1) (toStream m2))

179− (rstState st) stp sng yld

179+ st stp sng yld

180

181------------------------------------------------------------------------------

182-- Monad

@@ -184,11 +184,11 @@ serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->

184

185instance Monad m => Monad (SerialT m) where

186 return = pure

187− (SerialT (Stream m)) >>= f = SerialT $ Stream $ \st stp sng yld ->

187+ (SerialT m) >>= f = SerialT $ mkStream $ \st stp sng yld ->

188− let run x = unStream x (rstState st) stp sng yld

188+ let run x = unStream x st stp sng yld

189 single a = run $ toStream (f a)

190 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

192

193------------------------------------------------------------------------------

194-- Other instances

@@ -306,11 +306,11 @@ instance IsStream WSerialT where

306

307{-# INLINE interleave #-}

308interleave :: Stream m a -> Stream m a -> Stream m a

309−interleave m1 m2 = Stream $ \st stp sng yld -> do

309+interleave m1 m2 = mkStream $ \st stp sng yld -> do

310− let stop = unStream m2 (rstState st) stp sng yld

310+ let stop = unStream m2 st stp sng yld

311 single a = yld a m2

312 yieldk a r = yld a (interleave m2 r)

313− unStream m1 (rstState st) stop single yieldk

313+ unStream m1 st stop single yieldk

314

315-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.

316-- Interleaves two streams, yielding one element from each stream alternately.

@@ -318,9 +318,9 @@ interleave m1 m2 = Stream $ \st stp sng yld -> do

318-- @since 0.2.0

319{-# INLINE wSerial #-}

320wSerial :: 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 ->

322 unStream (interleave (toStream m1) (toStream m2))

323− (rstState st) stp sng yld

323+ st stp sng yld

324

325instance Semigroup (WSerialT m a) where

326 (<>) = wSerial

@@ -349,11 +349,11 @@ instance Monoid (WSerialT m a) where

349

350instance Monad m => Monad (WSerialT m) where

351 return = pure

352− (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \st stp sng yld ->

352+ (WSerialT m) >>= f = WSerialT $ mkStream $ \st stp sng yld ->

353− let run x = unStream x (rstState st) stp sng yld

353+ let run x = unStream x st stp sng yld

354 single a = run $ toStream (f a)

355 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

357

358------------------------------------------------------------------------------

359-- Other instances

src/Streamly/Streams/StreamD.hs+3−3

@@ -501,7 +501,7 @@ fromStreamK = Stream step

501 let stop = return Stop

502 single a = return $ Yield a K.nil

503 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

505

506{-# INLINE toStreamD #-}

507toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a

@@ -897,11 +897,11 @@ toList = foldr (:) []

897toStreamK :: Monad m => Stream m a -> K.Stream m a

898toStreamK (Stream step state) = go state

899 where

900− go st = K.Stream $ \gst stp sng yld -> do

900+ go st = K.mkStream $ \gst stp sng yld -> do

901 r <- step gst st

902 case r of

903 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

905 Stop -> stp

906

907#ifndef DISABLE_FUSION

src/Streamly/Streams/StreamK.hs+79−83

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

32 , adapt

34 -- * The stream type

35− , Stream (..)

35+ , Stream

36− , unStreamIsolated

36+ , unStream

37 , isolateStream

38 , unStreamShared

39 , runStreamSVar

@@ -196,32 +196,28 @@ import Streamly.SVar

196-- for Applicative and Monad, and in 'lift' for MonadTrans.

197--

198newtype Stream m a =

199− Stream {

199+ Stream (forall r.

200− unStream :: forall r.

201 State Stream m a -- state

200 State Stream m a -- state

202 -> m r -- stop

201 -> m r -- stop

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

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

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

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

205 -> m r

204 -> m r

206− }

205+ )

207

206

208−-- XXX make this the default "unStream"

207+{-# INLINE unStream #-}

209−-- | unwraps the Stream type producing the stream function that can be run with

208+unStream ::

210−-- continuations.

211−{-# INLINE unStreamIsolated #-}

212−unStreamIsolated ::

213 Stream m a

209 Stream m a

214− -> State Stream m a -- state

210+ -> State Stream m b -- state

215 -> m r -- stop

211 -> m r -- stop

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

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

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

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

218 -> m r

214 -> m r

219−unStreamIsolated x st = unStream x (rstState st)

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

220

216

221{-# INLINE isolateStream #-}

217{-# INLINE isolateStream #-}

222isolateStream :: Stream m a -> Stream m a

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

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

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

224− unStreamIsolated x st stp sng yld

220+ unStream x st stp sng yld

225

221

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

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

227{-# INLINE unStreamShared #-}

223{-# INLINE unStreamShared #-}

@@ -232,7 +228,7 @@ unStreamShared ::

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

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

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

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

234 -> m r

230 -> m r

235−unStreamShared = unStream

231+unStreamShared (Stream runner) st stp sng yld = runner st stp sng yld

236

232

237-- Run the stream using a run function associated with the SVar that runs the

233-- Run the stream using a run function associated with the SVar that runs the

238-- streams with a captured snapshot of the monadic state.

234-- streams with a captured snapshot of the monadic state.

@@ -248,7 +244,7 @@ runStreamSVar

248 -> m ()

244 -> m ()

249runStreamSVar sv m st stp sng yld =

245runStreamSVar sv m st stp sng yld =

250 let mrun = runInIO $ svarMrun sv

246 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

252

248

253------------------------------------------------------------------------------

249------------------------------------------------------------------------------

254-- Types that can behave as a Stream

250-- Types that can behave as a Stream

@@ -332,7 +328,7 @@ mkStream:: IsStream t

332 -> t m a

328 -> t m a

333mkStream k = fromStream $ Stream $ \st stp sng yld ->

329mkStream k = fromStream $ Stream $ \st stp sng yld ->

334 let yieldk a r = yld a (toStream r)

330 let yieldk a r = yld a (toStream r)

335− in k (rstState st) stp sng yieldk

331+ in k st stp sng yieldk

336

332

337------------------------------------------------------------------------------

333------------------------------------------------------------------------------

338-- Construction

334-- Construction

@@ -430,7 +426,7 @@ uncons m =

430 let stop = return Nothing

426 let stop = return Nothing

431 single a = return (Just (a, nil))

427 single a = return (Just (a, nil))

432 yieldk a r = return (Just (a, fromStream r))

428 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

434

430

435-------------------------------------------------------------------------------

431-------------------------------------------------------------------------------

436-- Generation

432-- Generation

@@ -454,7 +450,7 @@ unfoldrM step = go

454 case mayb of

450 case mayb of

455 Nothing -> stp

451 Nothing -> stp

456 Just (a, b) ->

452 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

458

454

459-------------------------------------------------------------------------------

455-------------------------------------------------------------------------------

460-- Special generation

456-- Special generation

@@ -537,7 +533,7 @@ foldStream

537 -> m r

533 -> m r

538foldStream st blank single step m =

534foldStream st blank single step m =

539 let yieldk a x = step a (fromStream x)

535 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

541

537

542-- | Lazy right associative fold.

538-- | Lazy right associative fold.

543foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b

539foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b

@@ -547,7 +543,7 @@ foldr step acc m = go (toStream m)

547 let stop = return acc

543 let stop = return acc

548 single a = return (step a acc)

544 single a = return (step a acc)

549 yieldk a r = go r >>= \b -> return (step a b)

545 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

551

547

552-- | Lazy right fold with a monadic step function.

548-- | Lazy right fold with a monadic step function.

553{-# INLINE foldrM #-}

549{-# INLINE foldrM #-}

@@ -558,7 +554,7 @@ foldrM step acc m = go (toStream m)

558 let stop = return acc

554 let stop = return acc

559 single a = step a acc

555 single a = step a acc

560 yieldk a r = go r >>= step a

556 yieldk a r = go r >>= step a

561− in unStream m1 defState stop single yieldk

557+ in unStreamShared m1 defState stop single yieldk

562

558

563{-# INLINE foldr1 #-}

559{-# INLINE foldr1 #-}

564foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)

560foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)

@@ -572,7 +568,7 @@ foldr1 step m = do

572 let stp = return p

568 let stp = return p

573 single a = return $ step a p

569 single a = return $ step a p

574 yieldk a r = fmap (step p) (go a r)

570 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

576

572

577-- | Strict left fold with an extraction function. Like the standard strict

573-- | Strict left fold with an extraction function. Like the standard strict

578-- left fold, but applies a user supplied extraction function (the third

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

590 -- XXX we are not strictly evaluating the accumulator here. Is this

586 -- XXX we are not strictly evaluating the accumulator here. Is this

591 -- okay?

587 -- okay?

592 let single = return . done

588 let single = return . done

593− in unStream m1 undefined undefined single undefined

589+ in unStreamShared m1 undefined undefined single undefined

594

590

595 -- Note, this can be implemented by making a recursive call to "go",

591 -- Note, this can be implemented by making a recursive call to "go",

596 -- however that is more expensive because of unnecessary recursion

592 -- however that is more expensive because of unnecessary recursion

@@ -601,8 +597,8 @@ foldx step begin done m = get $ go (toStream m) begin

601 single a = sng $ step acc a

597 single a = sng $ step acc a

602 yieldk a r =

598 yieldk a r =

603 let stream = go r (step acc a)

599 let stream = go r (step acc a)

604− in unStream stream defState undefined sng yld

600+ in unStreamShared stream defState undefined sng yld

605− in unStream m1 defState stop single yieldk

601+ in unStreamShared m1 defState stop single yieldk

606

602

607-- | Strict left associative fold.

603-- | Strict left associative fold.

608{-# INLINE foldl' #-}

604{-# INLINE foldl' #-}

@@ -619,7 +615,7 @@ foldxM step begin done m = go begin (toStream m)

619 let stop = acc >>= done

615 let stop = acc >>= done

620 single a = acc >>= \b -> step b a >>= done

616 single a = acc >>= \b -> step b a >>= done

621 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r

617 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

623

619

624-- | Like 'foldl'' but with a monadic step function.

620-- | Like 'foldl'' but with a monadic step function.

625foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b

621foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b

@@ -637,7 +633,7 @@ runStream m = go (toStream m)

637 let stop = return ()

633 let stop = return ()

638 single _ = return ()

634 single _ = return ()

639 yieldk _ r = go (toStream r)

635 yieldk _ r = go (toStream r)

640− in unStream m1 defState stop single yieldk

636+ in unStreamShared m1 defState stop single yieldk

641

637

642{-# INLINE null #-}

638{-# INLINE null #-}

643null :: (IsStream t, Monad m) => t m a -> m Bool

639null :: (IsStream t, Monad m) => t m a -> m Bool

@@ -645,7 +641,7 @@ null m =

645 let stop = return True

641 let stop = return True

646 single _ = return False

642 single _ = return False

647 yieldk _ _ = return False

643 yieldk _ _ = return False

648− in unStream (toStream m) defState stop single yieldk

644+ in unStreamShared (toStream m) defState stop single yieldk

649

645

650{-# INLINE head #-}

646{-# INLINE head #-}

651head :: (IsStream t, Monad m) => t m a -> m (Maybe a)

647head :: (IsStream t, Monad m) => t m a -> m (Maybe a)

@@ -653,7 +649,7 @@ head m =

653 let stop = return Nothing

649 let stop = return Nothing

654 single a = return (Just a)

650 single a = return (Just a)

655 yieldk a _ = return (Just a)

651 yieldk a _ = return (Just a)

656− in unStream (toStream m) defState stop single yieldk

652+ in unStreamShared (toStream m) defState stop single yieldk

657

653

658{-# INLINE tail #-}

654{-# INLINE tail #-}

659tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))

655tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))

@@ -661,7 +657,7 @@ tail m =

661 let stop = return Nothing

657 let stop = return Nothing

662 single _ = return $ Just nil

658 single _ = return $ Just nil

663 yieldk _ r = return $ Just $ fromStream r

659 yieldk _ r = return $ Just $ fromStream r

664− in unStream (toStream m) defState stop single yieldk

660+ in unStreamShared (toStream m) defState stop single yieldk

665

661

666{-# INLINE init #-}

662{-# INLINE init #-}

667init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))

663init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))

@@ -675,7 +671,7 @@ init m = go1 (toStream m)

675 go p m1 = Stream $ \_ stp sng yld ->

671 go p m1 = Stream $ \_ stp sng yld ->

676 let single _ = sng p

672 let single _ = sng p

677 yieldk a x = yld p $ go a x

673 yieldk a x = yld p $ go a x

678− in unStream m1 defState stp single yieldk

674+ in unStreamShared m1 defState stp single yieldk

679

675

680{-# INLINE elem #-}

676{-# INLINE elem #-}

681elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

677elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

@@ -685,7 +681,7 @@ elem e m = go (toStream m)

685 let stop = return False

681 let stop = return False

686 single a = return (a == e)

682 single a = return (a == e)

687 yieldk a r = if a == e then return True else go r

683 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

689

685

690{-# INLINE notElem #-}

686{-# INLINE notElem #-}

691notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

687notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

@@ -695,7 +691,7 @@ notElem e m = go (toStream m)

695 let stop = return True

691 let stop = return True

696 single a = return (a /= e)

692 single a = return (a /= e)

697 yieldk a r = if a == e then return False else go r

693 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

699

695

700all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

696all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

701all p m = go (toStream m)

697all p m = go (toStream m)

@@ -705,7 +701,7 @@ all p m = go (toStream m)

705 | otherwise = return False

701 | otherwise = return False

706 yieldk a r | p a = go r

702 yieldk a r | p a = go r

707 | otherwise = return False

703 | otherwise = return False

708− in unStream m1 defState (return True) single yieldk

704+ in unStreamShared m1 defState (return True) single yieldk

709

705

710any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

706any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

711any p m = go (toStream m)

707any p m = go (toStream m)

@@ -715,7 +711,7 @@ any p m = go (toStream m)

715 | otherwise = return False

711 | otherwise = return False

716 yieldk a r | p a = return True

712 yieldk a r | p a = return True

717 | otherwise = go r

713 | otherwise = go r

718− in unStream m1 defState (return False) single yieldk

714+ in unStreamShared m1 defState (return False) single yieldk

719

715

720-- | Extract the last element of the stream, if any.

716-- | Extract the last element of the stream, if any.

721{-# INLINE last #-}

717{-# INLINE last #-}

@@ -730,7 +726,7 @@ minimum m = go Nothing (toStream m)

730 let stop = return Nothing

726 let stop = return Nothing

731 single a = return (Just a)

727 single a = return (Just a)

732 yieldk a r = go (Just a) r

728 yieldk a r = go (Just a) r

733− in unStream m1 defState stop single yieldk

729+ in unStreamShared m1 defState stop single yieldk

734

730

735 go (Just res) m1 =

731 go (Just res) m1 =

736 let stop = return (Just res)

732 let stop = return (Just res)

@@ -742,7 +738,7 @@ minimum m = go Nothing (toStream m)

742 if res <= a

738 if res <= a

743 then go (Just res) r

739 then go (Just res) r

744 else go (Just a) r

740 else go (Just a) r

745− in unStream m1 defState stop single yieldk

741+ in unStreamShared m1 defState stop single yieldk

746

742

747{-# INLINE minimumBy #-}

743{-# INLINE minimumBy #-}

748minimumBy

744minimumBy

@@ -754,7 +750,7 @@ minimumBy cmp m = go Nothing (toStream m)

754 let stop = return Nothing

750 let stop = return Nothing

755 single a = return (Just a)

751 single a = return (Just a)

756 yieldk a r = go (Just a) r

752 yieldk a r = go (Just a) r

757− in unStream m1 defState stop single yieldk

753+ in unStreamShared m1 defState stop single yieldk

758

754

759 go (Just res) m1 =

755 go (Just res) m1 =

760 let stop = return (Just res)

756 let stop = return (Just res)

@@ -764,7 +760,7 @@ minimumBy cmp m = go Nothing (toStream m)

764 yieldk a r = case cmp res a of

760 yieldk a r = case cmp res a of

765 GT -> go (Just a) r

761 GT -> go (Just a) r

766 _ -> go (Just res) r

762 _ -> go (Just res) r

767− in unStream m1 defState stop single yieldk

763+ in unStreamShared m1 defState stop single yieldk

768

764

769{-# INLINE maximum #-}

765{-# INLINE maximum #-}

770maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)

766maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)

@@ -774,7 +770,7 @@ maximum m = go Nothing (toStream m)

774 let stop = return Nothing

770 let stop = return Nothing

775 single a = return (Just a)

771 single a = return (Just a)

776 yieldk a r = go (Just a) r

772 yieldk a r = go (Just a) r

777− in unStream m1 defState stop single yieldk

773+ in unStreamShared m1 defState stop single yieldk

778

774

779 go (Just res) m1 =

775 go (Just res) m1 =

780 let stop = return (Just res)

776 let stop = return (Just res)

@@ -786,7 +782,7 @@ maximum m = go Nothing (toStream m)

786 if res <= a

782 if res <= a

787 then go (Just a) r

783 then go (Just a) r

788 else go (Just res) r

784 else go (Just res) r

789− in unStream m1 defState stop single yieldk

785+ in unStreamShared m1 defState stop single yieldk

790

786

791{-# INLINE maximumBy #-}

787{-# INLINE maximumBy #-}

792maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)

788maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)

@@ -796,7 +792,7 @@ maximumBy cmp m = go Nothing (toStream m)

796 let stop = return Nothing

792 let stop = return Nothing

797 single a = return (Just a)

793 single a = return (Just a)

798 yieldk a r = go (Just a) r

794 yieldk a r = go (Just a) r

799− in unStream m1 defState stop single yieldk

795+ in unStreamShared m1 defState stop single yieldk

800

796

801 go (Just res) m1 =

797 go (Just res) m1 =

802 let stop = return (Just res)

798 let stop = return (Just res)

@@ -806,7 +802,7 @@ maximumBy cmp m = go Nothing (toStream m)

806 yieldk a r = case cmp res a of

802 yieldk a r = case cmp res a of

807 GT -> go (Just res) r

803 GT -> go (Just res) r

808 _ -> go (Just a) r

804 _ -> go (Just a) r

809− in unStream m1 defState stop single yieldk

805+ in unStreamShared m1 defState stop single yieldk

810

806

811{-# INLINE (!!) #-}

807{-# INLINE (!!) #-}

812(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)

808(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)

@@ -818,7 +814,7 @@ m !! i = go i (toStream m)

818 yieldk a x | n < 0 = return Nothing

814 yieldk a x | n < 0 = return Nothing

819 | n == 0 = return $ Just a

815 | n == 0 = return $ Just a

820 | otherwise = go (n - 1) x

816 | otherwise = go (n - 1) x

821− in unStream m1 defState (return Nothing) single yieldk

817+ in unStreamShared m1 defState (return Nothing) single yieldk

822

818

823{-# INLINE lookup #-}

819{-# INLINE lookup #-}

824lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)

820lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)

@@ -829,7 +825,7 @@ lookup e m = go (toStream m)

829 | otherwise = return Nothing

825 | otherwise = return Nothing

830 yieldk (a, b) x | a == e = return $ Just b

826 yieldk (a, b) x | a == e = return $ Just b

831 | otherwise = go x

827 | otherwise = go x

832− in unStream m1 defState (return Nothing) single yieldk

828+ in unStreamShared m1 defState (return Nothing) single yieldk

833

829

834{-# INLINE findM #-}

830{-# INLINE findM #-}

835findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)

831findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)

@@ -842,7 +838,7 @@ findM p m = go (toStream m)

842 yieldk a x = do

838 yieldk a x = do

843 b <- p a

839 b <- p a

844 if b then return $ Just a else go x

840 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

846

842

847{-# INLINE find #-}

843{-# INLINE find #-}

848find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)

844find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)

@@ -857,7 +853,7 @@ findIndices p = fromStream . go 0 . toStream

857 | otherwise = stp

853 | otherwise = stp

858 yieldk a x | p a = yld offset $ go (offset + 1) x

854 yieldk a x | p a = yld offset $ go (offset + 1) x

859 | otherwise = unStream (go (offset + 1) x) st stp sng yld

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

861

857

862------------------------------------------------------------------------------

858------------------------------------------------------------------------------

863-- Map and Fold

859-- Map and Fold

@@ -872,7 +868,7 @@ mapM_ f m = go (toStream m)

872 let stop = return ()

868 let stop = return ()

873 single a = void (f a)

869 single a = void (f a)

874 yieldk a r = f a >> go r

870 yieldk a r = f a >> go r

875− in unStream m1 defState stop single yieldk

871+ in unStreamShared m1 defState stop single yieldk

876

872

877------------------------------------------------------------------------------

873------------------------------------------------------------------------------

878-- Converting folds

874-- Converting folds

@@ -900,7 +896,7 @@ scanx step begin done m =

900 yieldk a r =

896 yieldk a r =

901 let s = step acc a

897 let s = step acc a

902 in yld (done s) (go r s)

898 in yld (done s) (go r s)

903− in unStream m1 (rstState st) stp single yieldk

899+ in unStream m1 st stp single yieldk

904

900

905{-# INLINE scanl' #-}

901{-# INLINE scanl' #-}

906scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b

902scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b

@@ -918,8 +914,8 @@ filter p m = fromStream $ go (toStream m)

918 let single a | p a = sng a

914 let single a | p a = sng a

919 | otherwise = stp

915 | otherwise = stp

920 yieldk a r | p a = yld a (go r)

916 yieldk a r | p a = yld a (go r)

921− | otherwise = unStream r (rstState st) stp single yieldk

917+ | otherwise = unStream r st stp single yieldk

922− in unStream m1 (rstState st) stp single yieldk

918+ in unStream m1 st stp single yieldk

923

919

924{-# INLINE take #-}

920{-# INLINE take #-}

925take :: IsStream t => Int -> t m a -> t m a

921take :: IsStream t => Int -> t m a -> t m a

@@ -929,7 +925,7 @@ take n m = fromStream $ go n (toStream m)

929 let yieldk a r = yld a (go (n1 - 1) r)

925 let yieldk a r = yld a (go (n1 - 1) r)

930 in if n1 <= 0

926 in if n1 <= 0

931 then stp

927 then stp

932− else unStream m1 (rstState st) stp sng yieldk

928+ else unStream m1 st stp sng yieldk

933

929

934{-# INLINE takeWhile #-}

930{-# INLINE takeWhile #-}

935takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

931takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

@@ -940,7 +936,7 @@ takeWhile p m = fromStream $ go (toStream m)

940 | otherwise = stp

936 | otherwise = stp

941 yieldk a r | p a = yld a (go r)

937 yieldk a r | p a = yld a (go r)

942 | otherwise = stp

938 | otherwise = stp

943− in unStream m1 (rstState st) stp single yieldk

939+ in unStream m1 st stp single yieldk

944

940

945drop :: IsStream t => Int -> t m a -> t m a

941drop :: IsStream t => Int -> t m a -> t m a

946drop n m = fromStream $ Stream $ \st stp sng yld ->

942drop n m = fromStream $ Stream $ \st stp sng yld ->

@@ -951,8 +947,8 @@ drop n m = fromStream $ Stream $ \st stp sng yld ->

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

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

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

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

953 in if n1 <= 0

949 in if n1 <= 0

954− then unStream m1 st stp sng yld

950+ then unStreamShared m1 st stp sng yld

955− else unStream m1 st stp single yieldk

951+ else unStreamShared m1 st stp single yieldk

956

952

957{-# INLINE dropWhile #-}

953{-# INLINE dropWhile #-}

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

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

@@ -961,9 +957,9 @@ dropWhile p m = fromStream $ go (toStream m)

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

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

962 let single a | p a = stp

958 let single a | p a = stp

963 | otherwise = sng a

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

965 | otherwise = yld a r

961 | otherwise = yld a r

966− in unStream m1 (rstState st) stp single yieldk

962+ in unStream m1 st stp single yieldk

967

963

968-------------------------------------------------------------------------------

964-------------------------------------------------------------------------------

969-- Mapping

965-- Mapping

@@ -974,7 +970,7 @@ map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b

974map f m = fromStream $ Stream $ \st stp sng yld ->

970map f m = fromStream $ Stream $ \st stp sng yld ->

975 let single = sng . f

971 let single = sng . f

976 yieldk a r = yld (f a) (fmap f r)

972 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

978

974

979-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

975-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

980-- other stream types to overload it.

976-- other stream types to overload it.

@@ -984,8 +980,8 @@ mapM f m = go (toStream m)

984 where

980 where

985 go m1 = fromStream $ Stream $ \st stp sng yld ->

981 go m1 = fromStream $ Stream $ \st stp sng yld ->

986 let single a = f a >>= sng

982 let single a = f a >>= sng

987− yieldk a r = unStream (toStream (f a |: go r)) st stp sng yld

983+ yieldk a r = unStreamShared (toStream (f a |: go r)) st stp sng yld

988− in unStream m1 (rstState st) stp single yieldk

984+ in unStream m1 st stp single yieldk

989

985

990-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

986-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

991-- other stream types to overload it.

987-- other stream types to overload it.

@@ -995,8 +991,8 @@ sequence m = go (toStream m)

995 where

991 where

996 go m1 = fromStream $ Stream $ \st stp sng yld ->

992 go m1 = fromStream $ Stream $ \st stp sng yld ->

997 let single ma = ma >>= sng

993 let single ma = ma >>= sng

998− yieldk ma r = unStream (toStream $ ma |: go r) st stp sng yld

994+ yieldk ma r = unStreamShared (toStream $ ma |: go r) st stp sng yld

999− in unStream m1 (rstState st) stp single yieldk

995+ in unStream m1 st stp single yieldk

1000

996

1001-------------------------------------------------------------------------------

997-------------------------------------------------------------------------------

1002-- Inserting

998-- Inserting

@@ -1007,12 +1003,12 @@ intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

1007intersperseM a m = fromStream $ prependingStart (toStream m)

1003intersperseM a m = fromStream $ prependingStart (toStream m)

1008 where

1004 where

1009 prependingStart m1 = Stream $ \st stp sng yld ->

1005 prependingStart m1 = Stream $ \st stp sng yld ->

1010− let yieldk i x = unStream (return i |: go x) st stp sng yld

1006+ let yieldk i x = unStreamShared (return i |: go x) st stp sng yld

1011− in unStream m1 (rstState st) stp sng yieldk

1007+ in unStream m1 st stp sng yieldk

1012 go m2 = fromStream $ Stream $ \st stp sng yld ->

1008 go m2 = fromStream $ Stream $ \st stp sng yld ->

1013− let single i = unStream (a |: yield i) st stp sng yld

1009+ let single i = unStreamShared (a |: yield i) st stp sng yld

1014− yieldk i x = unStream (a |: return i |: go x) st stp sng yld

1010+ yieldk i x = unStreamShared (a |: return i |: go x) st stp sng yld

1015− in unStream m2 (rstState st) stp single yieldk

1011+ in unStream m2 st stp single yieldk

1016

1012

1017{-# INLINE insertBy #-}

1013{-# INLINE insertBy #-}

1018insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a

1014insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a

@@ -1026,7 +1022,7 @@ insertBy cmp x m = fromStream $ go (toStream m)

1026 yieldk a r = case cmp x a of

1022 yieldk a r = case cmp x a of

1027 GT -> yld a (go r)

1023 GT -> yld a (go r)

1028 _ -> yld x (a `cons` r)

1024 _ -> yld x (a `cons` r)

1029− in unStream m1 (rstState st) stop single yieldk

1025+ in unStream m1 st stop single yieldk

1030

1026

1031------------------------------------------------------------------------------

1027------------------------------------------------------------------------------

1032-- Deleting

1028-- Deleting

@@ -1039,9 +1035,9 @@ deleteBy eq x m = fromStream $ go (toStream m)

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

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

1040 let single a = if eq x a then stp else sng a

1036 let single a = if eq x a then stp else sng a

1041 yieldk a r = if eq x a

1037 yieldk a r = if eq x a

1042− then unStream r (rstState st) stp sng yld

1038+ then unStream r st stp sng yld

1043 else yld a (go r)

1039 else yld a (go r)

1044− in unStream m1 (rstState st) stp single yieldk

1040+ in unStream m1 st stp single yieldk

1045

1041

1046-------------------------------------------------------------------------------

1042-------------------------------------------------------------------------------

1047-- Map and Filter

1043-- Map and Filter

@@ -1057,8 +1053,8 @@ mapMaybe f m = go (toStream m)

1057 Nothing -> stp

1053 Nothing -> stp

1058 yieldk a r = case f a of

1054 yieldk a r = case f a of

1059 Just b -> yld b (toStream $ go r)

1055 Just b -> yld b (toStream $ go r)

1060− Nothing -> unStream r (rstState st) stp single yieldk

1056+ Nothing -> unStream r st stp single yieldk

1061− in unStream m1 (rstState st) stp single yieldk

1057+ in unStream m1 st stp single yieldk

1062

1058

1063------------------------------------------------------------------------------

1059------------------------------------------------------------------------------

1064-- Serial Zipping

1060-- Serial Zipping

@@ -1072,10 +1068,10 @@ zipWithS f = go

1072 let merge a ra =

1068 let merge a ra =

1073 let single2 b = sng (f a b)

1069 let single2 b = sng (f a b)

1074 yield2 b rb = yld (f a b) (go ra rb)

1070 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

1076 let single1 a = merge a nil

1072 let single1 a = merge a nil

1077 yield1 = merge

1073 yield1 = merge

1078− unStream mx (rstState st) stp single1 yield1

1074+ unStream mx st stp single1 yield1

1079

1075

1080-- | Zip two streams serially using a pure zipping function.

1076-- | Zip two streams serially using a pure zipping function.

1081--

1077--

@@ -1092,13 +1088,13 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

1092 where

1088 where

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

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

1094 let merge a ra =

1090 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

1096 single2 b = f a b >>= sng

1092 single2 b = f a b >>= sng

1097 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)

1093 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

1099 let single1 a = merge a nil

1095 let single1 a = merge a nil

1100 yield1 = merge

1096 yield1 = merge

1101− unStream mx (rstState st) stp single1 yield1

1097+ unStream mx st stp single1 yield1

1102

1098

1103------------------------------------------------------------------------------

1099------------------------------------------------------------------------------

1104-- Merging

1100-- Merging

@@ -1158,7 +1154,7 @@ the m = do

1158 | otherwise = return Nothing

1154 | otherwise = return Nothing

1159 yieldk a r | h == a = go h r

1155 yieldk a r | h == a = go h r

1160 | otherwise = return Nothing

1156 | otherwise = return Nothing

1161− in unStream m1 defState (return $ Just h) single yieldk

1157+ in unStreamShared m1 defState (return $ Just h) single yieldk

1162

1158

1163------------------------------------------------------------------------------

1159------------------------------------------------------------------------------

1164-- Semigroup

1160-- Semigroup

@@ -1171,7 +1167,7 @@ serial :: Stream m a -> Stream m a -> Stream m a

1171serial m1 m2 = go m1

1167serial m1 m2 = go m1

1172 where

1168 where

1173 go (Stream m) = Stream $ \st stp sng yld ->

1169 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

1175 single a = yld a m2

1171 single a = yld a m2

1176 yieldk a r = yld a (go r)

1172 yieldk a r = yld a (go r)

1177 in m (rstState st) stop single yieldk

1173 in m (rstState st) stop single yieldk

@@ -1209,7 +1205,7 @@ bindWith par m f = go m

1209 go (Stream g) =

1205 go (Stream g) =

1210 Stream $ \st stp sng yld ->

1206 Stream $ \st stp sng yld ->

1211 let runShared x = unStreamShared x st stp sng yld

1207 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

1213

1209

1214 single a = runIsolated $ f a

1210 single a = runIsolated $ f a

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

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

@@ -1221,8 +1217,8 @@ bindWith par m f = go m

1221

1217

1222_alt :: Stream m a -> Stream m a -> Stream m a

1218_alt :: Stream m a -> Stream m a -> Stream m a

1223_alt m1 m2 = Stream $ \st stp sng yld ->

1219_alt m1 m2 = Stream $ \st stp sng yld ->

1224− let stop = unStream m2 (rstState st) stp sng yld

1220+ let stop = unStream m2 st stp sng yld

1225− in unStream m1 (rstState st) stop sng yld

1221+ in unStream m1 st stop sng yld

1226

1222

1227------------------------------------------------------------------------------

1223------------------------------------------------------------------------------

1228-- MonadReader

1224-- MonadReader

@@ -1233,7 +1229,7 @@ withLocal f m =

1233 Stream $ \st stp sng yld ->

1229 Stream $ \st stp sng yld ->

1234 let single = local f . sng

1230 let single = local f . sng

1235 yieldk a r = local f $ yld a (withLocal f r)

1231 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

1237

1233

1238------------------------------------------------------------------------------

1234------------------------------------------------------------------------------

1239-- MonadError

1235-- MonadError

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

@@ -48,7 +48,7 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,

48 readListPrecDefault)

49import Prelude hiding (map, repeat, zipWith)

51−import Streamly.Streams.StreamK (IsStream(..), Stream(..))

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

52import Streamly.Streams.Async (mkAsync')

53import Streamly.Streams.Serial (map)

54import Streamly.SVar (MonadAsync, rstState)

@@ -142,10 +142,10 @@ TRAVERSABLE_INSTANCE(ZipSerialM)

142-- @since 0.1.0

143zipAsyncWith :: (IsStream t, MonadAsync m)

144 => (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

146 ma <- mkAsync' (rstState st) m1

147 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

149

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

151-- are generated concurrently.

@@ -153,10 +153,10 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do

153-- @since 0.4.0

154zipAsyncWithM :: (IsStream t, MonadAsync m)

155 => (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

157 ma <- mkAsync' (rstState st) m1

158 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

160

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

162-- Parallely Zipping Streams