6ae6879 · streamly · git.iko.soy

git.iko.soy/streamly

rename maxRate to yieldRate, update doc

commit6ae6879887701040038e31b2a549e21d5ead510f authorHarendra Kumar <harendra.kumar@gmail.com> dateSat 18 Aug 2018, 01:51 (8 years ago) parentce90b27

4 files changed+53−40

compare with master

unifiedsplit

Changelog.md+1−1 src/Streamly.hs+9−10 src/Streamly/Streams/SVar.hs+30−16 test/Prop.hs+13−13

Changelog.md+1−1

@@ -10,7 +10,7 @@

1010

1111### Enhancements

1212

13−* Added `maxRate` combinator to control the yield rate of a stream.

13+* Added `yieldRate` combinator to control the yield rate of a stream.

1414* Add `foldl1'`, `foldr1`, `intersperseM`, `find`, `lookup`, `and`, `or`,

1515 `findIndices`, `findIndex`, `elemIndices`, `elemIndex`, `init` to Prelude

1616

src/Streamly.hs+9−10

@@ -112,7 +112,7 @@ module Streamly

112112 -- $concurrency

113113 , maxThreads

114114 , maxBuffer

115− , maxRate

115+ , yieldRate

116116

117117 -- * Folding Containers of Streams

118118 -- $foldutils

@@ -173,7 +173,7 @@ import Streamly.Streams.Ahead

173173import Streamly.Streams.Parallel

174174import Streamly.Streams.Zip

175175import Streamly.Streams.Prelude

176−import Streamly.Streams.SVar (maxThreads, maxBuffer, maxRate)

176+import Streamly.Streams.SVar (maxThreads, maxBuffer, yieldRate)

177177import Streamly.SVar (MonadAsync)

178178import Data.Semigroup (Semigroup(..))

179179

@@ -305,14 +305,13 @@ runZipAsync = runStream . K.adapt

305305--

306306-- $concurrency

307307--

308−-- These combinators can be used at any point in a stream composition to

309−-- control the concurrency of the enclosed stream. When the combinators are

310−-- used in a nested manner, the nearest enclosing combinator overrides the

311−-- outer ones. These combinators have no effect on 'Parallel' streams,

312−-- concurrency for 'Parallel' streams is always unbounded.

313−-- Note that the use of these combinators does not enable concurrency, to

314−-- enable concurrency you have to use one of the concurrent stream type

315−-- combinators.

308+-- These combinators can be used at any point in a stream composition to set

309+-- parameters to control the concurrency of the enclosed stream. A parameter

310+-- set at any point remains effective for any concurrent combinators used

311+-- downstream until it is reset. These control parameters have no effect on

312+-- non-concurrent combinators in the stream, or on non-concurrent streams. They

313+-- also do not affect 'Parallel' streams, as concurrency for 'Parallel' streams

314+-- is always unbounded.

316315

317316-- $adapters

318317--

src/Streamly/Streams/SVar.hs+30−16

@@ -32,7 +32,7 @@ module Streamly.Streams.SVar

3232 , maxThreads

3333 , maxBuffer

3434 , maxYields

35− , maxRate

35+ , yieldRate

3636 )

3737where

3838

@@ -114,6 +114,9 @@ fromSVar sv = do

114114 fromStream $ Stream $ \st stp sng yld -> do

115115 ref <- liftIO $ newIORef ()

116116 _ <- liftIO $ mkWeakIORef ref hook

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

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

119+ -- will get garbage collected and our hook will be called.

117120 unStream (fromStreamVar sv{svarRef = Just ref}) st stp sng yld

118121 where

119122

@@ -134,14 +137,16 @@ toSVar sv m = toStreamVar sv (toStream m)

134137--

135138-- XXX need to write these in direct style otherwise they will break fusion.

136139--

137−-- | Specify the maximum number of threads that can be spawned concurrently

138−-- when using concurrent streams. This value denotes maximum concurrent

139−-- requests, or tasks in progress at any given point of time. Note that this is

140−-- not the grand total number of threads but maximum threads at an individual

141−-- point of concurrency.

140+-- | Specify the maximum number of threads that can be spawned concurrently for

141+-- any concurrent combinator in a stream.

142142-- A value of 0 resets the thread limit to default, a negative value means

143143-- there is no limit. The default value is 1500.

144144--

145+-- When the actions in a stream are IO bound, having blocking IO calls, this

146+-- option can be used to control the maximum number of in-flight IO requests.

147+-- When the actions are CPU bound this option can be used to

148+-- control the amount of CPU used by the stream.

149+--

145150-- @since 0.4.0

146151{-# INLINE_NORMAL maxThreads #-}

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

@@ -175,24 +180,33 @@ maxBuffer n m = fromStream $ Stream $ \st stp sng yld -> do

175180maxBufferSerial :: Int -> SerialT m a -> SerialT m a

176181maxBufferSerial _ = id

177182

178−-- | Specify the maximum rate in number of yields per second at which the

179−-- stream can be generated. A value of 0 resets the rate to default, a negative

180−-- value means there is no limit. The default value is no limit.

183+-- | Specify the maximum average pull rate in number of yields per second (i.e.

184+-- @Hertz@) which the consumer of the stream cannot exceed. A value of 0 resets

185+-- the rate to default, a negative value means there is no limit. The default

186+-- value is no limit.

187+-- Concurrent production is ramped up or down automatically to achieve the

188+-- specified yield rate. However, the effective maximum production rate

189+-- achieved by a stream is governed by:

190+--

191+-- * The 'maxThreads' limit

192+-- * The 'maxBuffer' limit

193+-- * The maximum rate that the stream producer can achieve

194+-- * The maximum rate that the stream consumer can achieve

181195--

182196-- @since 0.5.0

183−{-# INLINE_NORMAL maxRate #-}

184−maxRate :: IsStream t => Double -> t m a -> t m a

185−maxRate n m = fromStream $ Stream $ \st stp sng yld -> do

197+{-# INLINE_NORMAL yieldRate #-}

198+yieldRate :: IsStream t => Double -> t m a -> t m a

199+yieldRate n m = fromStream $ Stream $ \st stp sng yld -> do

186200 unStream (toStream m) (setMaxStreamRate n st) stp sng yld

187201

188−{-# RULES "maxRate serial" maxRate = maxRateSerial #-}

189−maxRateSerial :: Double -> SerialT m a -> SerialT m a

190−maxRateSerial _ = id

202+{-# RULES "yieldRate serial" yieldRate = yieldRateSerial #-}

203+yieldRateSerial :: Double -> SerialT m a -> SerialT m a

204+yieldRateSerial _ = id

191205

192206-- | Specify the average latency, in nanoseconds, of a single threaded action

193207-- in a concurrent composition. Streamly can measure the latencies, but that is

194208-- possible only after at least one task has completed. This combinator can be

195−-- used to provide a latency hint so that rate control using 'maxRate' can take

209+-- used to provide a latency hint so that rate control using 'yieldRate' can take

196210-- that into account right from the beginning. When not specified then a

197211-- default behavior is chosen which could be too slow or too fast, and would be

198212-- restricted by any other control parameters configured.

test/Prop.hs+13−13

@@ -693,8 +693,8 @@ main = hspec

693693 let makeOps t =

694694 [ ("default", t)

695695#ifndef COVERAGE_BUILD

696− , ("maxRate 10000", t . maxRate 10000)

697− , ("maxRate -1", t . maxRate (-1))

696+ , ("yieldRate 10000", t . yieldRate 10000)

697+ , ("yieldRate -1", t . yieldRate (-1))

698698 , ("maxBuffer 0", t . maxBuffer 0)

699699 , ("maxBuffer 1", t . maxBuffer 1)

700700 , ("maxThreads 0", t . maxThreads 0)

@@ -707,43 +707,43 @@ main = hspec

707707 let serialOps :: IsStream t => ((SerialT IO a -> t IO a) -> Spec) -> Spec

708708 serialOps spec = mapOps spec $ (makeOps serially)

709709#ifndef COVERAGE_BUILD

710− ++ [("maxRate 0.00000001", serially . maxRate 0.00000001)]

710+ ++ [("yieldRate 0.00000001", serially . yieldRate 0.00000001)]

711711 ++ [("maxBuffer -1", serially . maxBuffer (-1))]

712712#endif

713713 let wSerialOps :: IsStream t => ((WSerialT IO a -> t IO a) -> Spec) -> Spec

714714 wSerialOps spec = mapOps spec $ makeOps wSerially

715715#ifndef COVERAGE_BUILD

716− ++ [("maxRate 0.00000001", wSerially . maxRate 0.00000001)]

716+ ++ [("yieldRate 0.00000001", wSerially . yieldRate 0.00000001)]

717717 ++ [("maxBuffer (-1)", wSerially . maxBuffer (-1))]

718718#endif

719719 let asyncOps :: IsStream t => ((AsyncT IO a -> t IO a) -> Spec) -> Spec

720720 asyncOps spec = mapOps spec $ makeOps asyncly

721721#ifndef COVERAGE_BUILD

722− ++ [("maxRate 10000", asyncly . maxRate 10000)]

722+ ++ [("yieldRate 10000", asyncly . yieldRate 10000)]

723723 ++ [("maxBuffer (-1)", asyncly . maxBuffer (-1))]

724724#endif

725725 let wAsyncOps :: IsStream t => ((WAsyncT IO a -> t IO a) -> Spec) -> Spec

726726 wAsyncOps spec = mapOps spec $ makeOps wAsyncly

727727#ifndef COVERAGE_BUILD

728− ++ [("maxRate 10000", wAsyncly . maxRate 10000)]

728+ ++ [("yieldRate 10000", wAsyncly . yieldRate 10000)]

729729 ++ [("maxBuffer (-1)", wAsyncly . maxBuffer (-1))]

730730#endif

731731 let aheadOps :: IsStream t => ((AheadT IO a -> t IO a) -> Spec) -> Spec

732732 aheadOps spec = mapOps spec $ makeOps aheadly

733733#ifndef COVERAGE_BUILD

734− -- ++ [("maxRate 10000", aheadly . maxRate 10000)]

734+ -- ++ [("yieldRate 10000", aheadly . yieldRate 10000)]

735735 -- ++ [("maxBuffer (-1)", aheadly . maxBuffer (-1))]

736736#endif

737737 let parallelOps :: IsStream t => ((ParallelT IO a -> t IO a) -> Spec) -> Spec

738738 parallelOps spec = mapOps spec $ makeOps parallely

739739#ifndef COVERAGE_BUILD

740− ++ [("maxRate 0.00000001", parallely . maxRate 0.00000001)]

740+ ++ [("yieldRate 0.00000001", parallely . yieldRate 0.00000001)]

741741 ++ [("maxBuffer (-1)", parallely . maxBuffer (-1))]

742742#endif

743743 let zipSerialOps :: IsStream t => ((ZipSerialM IO a -> t IO a) -> Spec) -> Spec

744744 zipSerialOps spec = mapOps spec $ makeOps zipSerially

745745#ifndef COVERAGE_BUILD

746− ++ [("maxRate 0.00000001", zipSerially . maxRate 0.00000001)]

746+ ++ [("yieldRate 0.00000001", zipSerially . yieldRate 0.00000001)]

747747 ++ [("maxBuffer (-1)", zipSerially . maxBuffer (-1))]

748748#endif

749749 -- Note, the "pure" of applicative Zip streams generates and infinite

@@ -751,7 +751,7 @@ main = hspec

751751 let zipAsyncOps :: IsStream t => ((ZipAsyncM IO a -> t IO a) -> Spec) -> Spec

752752 zipAsyncOps spec = mapOps spec $ makeOps zipAsyncly

753753#ifndef COVERAGE_BUILD

754− ++ [("maxRate 10000", zipAsyncly . maxRate 10000)]

754+ ++ [("yieldRate 10000", zipAsyncly . yieldRate 10000)]

755755#endif

756756

757757 describe "Construction" $ do

@@ -815,7 +815,7 @@ main = hspec

815815 zipAsyncOps $ prop "zipAsyncly applicative folded" . zipApplicative folded (==)

816816

817817 -- We test only the serial zip with serial streams and the parallel

818− -- stream, because the maxRate setting in these streams can slow down

818+ -- stream, because the yieldRate setting in these streams can slow down

819819 -- zipAsync.

820820 serialOps $ prop "zip monadic serially" . zipMonadic S.fromFoldable (==)

821821 serialOps $ prop "zip monadic serially folded" . zipMonadic folded (==)

@@ -901,8 +901,8 @@ main = hspec

901901 let mkOps t =

902902 [ ("default", t)

903903#ifndef COVERAGE_BUILD

904− , ("maxRate 0", t . maxRate 0)

905− , ("maxRate -1", t . maxRate (-1))

904+ , ("yieldRate 0", t . yieldRate 0)

905+ , ("yieldRate -1", t . yieldRate (-1))

906906 , ("maxBuffer 0", t . maxBuffer 0)

907907 , ("maxThreads 0", t . maxThreads 0)

908908 , ("maxThreads 0", t . maxThreads (-1))