e2c366c · streamly · git.iko.soy

Changelog.md+3

compare with master

unifiedsplit

@@ -7,6 +7,9 @@

7 stream.

8* Leftover threads are now cleaned up as soon as the SVar is garbage

9 collected.

10+* Fix passing of state across parallel, async, wAsync, ahead, serial, wSerial

11+ combinators. Without this fix combinators that rely on state passing e.g.

12+ `maxThreads` and `maxBuffer` won't work across these combinators.

10

13

11### Enhancements

14### Enhancements

12

15

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

@@ -634,7 +634,8 @@ instance IsStream AheadT where

634-- @since 0.3.0

635{-# INLINE ahead #-}

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

637+ahead m1 m2 = fromStream $ Stream $ \st stp sng yld ->

637−ahead m1 m2 = fromStream $ aheadS (toStream m1) (toStream m2)

638+ unStream (aheadS (toStream m1) (toStream m2)) st stp sng yld

638

639

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

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

640 (<>) = ahead

641 (<>) = ahead

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

@@ -582,8 +582,9 @@ asyncS = joinStreamVarAsync AsyncVar

582-- @since 0.2.0

583{-# INLINE async #-}

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

585−async m1 m2 = fromStream $

585+async m1 m2 = fromStream $ Stream $ \st stp sng yld ->

586− joinStreamVarAsync AsyncVar (toStream m1) (toStream m2)

586+ unStream (joinStreamVarAsync AsyncVar (toStream m1) (toStream m2))

587+ st stp sng yld

587

588

588-- | Same as 'async'.

589-- | Same as 'async'.

589--

590--

@@ -729,7 +730,8 @@ consMWAsync m r = K.yieldM m `wAsyncS` r

729-- @since 0.2.0

730-- @since 0.2.0

730{-# INLINE wAsync #-}

731{-# INLINE wAsync #-}

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

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

733+wAsync m1 m2 = fromStream $ Stream $ \st stp sng yld ->

732−wAsync m1 m2 = fromStream $ wAsyncS (toStream m1) (toStream m2)

734+ unStream (wAsyncS (toStream m1) (toStream m2)) st stp sng yld

733

735

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

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

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

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

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

@@ -117,7 +117,9 @@ consMParallel m r = K.yieldM m `parallelStream` r

117-- @since 0.2.0

118{-# INLINE parallel #-}

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

120+parallel m1 m2 = fromStream $ Stream $ \st stp sng yld -> do

120−parallel m1 m2 = fromStream $ parallelStream (toStream m1) (toStream m2)

121+ unStream (parallelStream (toStream m1) (toStream m2))

122+ st stp sng yld

121

123

122------------------------------------------------------------------------------

124------------------------------------------------------------------------------

123-- Convert a stream to parallel

125-- Convert a stream to parallel

src/Streamly/Streams/Serial.hs+6−2

@@ -166,7 +166,9 @@ instance IsStream SerialT where

166-- @since 0.2.0

167{-# INLINE serial #-}

168serial :: IsStream t => t m a -> t m a -> t m a

169+serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->

169−serial m1 m2 = fromStream $ K.serial (toStream m1) (toStream m2)

170+ unStream (K.serial (toStream m1) (toStream m2))

171+ (rstState st) stp sng yld

170

172

171------------------------------------------------------------------------------

173------------------------------------------------------------------------------

172-- Monad

174-- Monad

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

298-- @since 0.2.0

300-- @since 0.2.0

299{-# INLINE wSerial #-}

301{-# INLINE wSerial #-}

300wSerial :: IsStream t => t m a -> t m a -> t m a

302wSerial :: IsStream t => t m a -> t m a -> t m a

303+wSerial m1 m2 = fromStream $ Stream $ \st stp sng yld ->

301−wSerial m1 m2 = fromStream $ interleave (toStream m1) (toStream m2)

304+ unStream (interleave (toStream m1) (toStream m2))

305+ (rstState st) stp sng yld

302

306

303instance Semigroup (WSerialT m a) where

307instance Semigroup (WSerialT m a) where

304 (<>) = wSerial

308 (<>) = wSerial