6ae6879 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

rename maxRate to yieldRate, update doc

6ae6879Harendra Kumar8 years ago

Changelog.md+1−1

compare with master

unifiedsplit

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

11### Enhancements

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

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

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

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

src/Streamly.hs+9−10

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

112 -- $concurrency

113 , maxThreads

114 , maxBuffer

115− , maxRate

115+ , yieldRate

116

117 -- * Folding Containers of Streams

118 -- $foldutils

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

173import Streamly.Streams.Parallel

174import Streamly.Streams.Zip

175import Streamly.Streams.Prelude

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

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

177import Streamly.SVar (MonadAsync)

178import Data.Semigroup (Semigroup(..))

179

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

305--

306-- $concurrency

307--

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

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

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

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

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

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

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

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

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

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

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

314+-- is always unbounded.

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

315−-- combinators.

316

315

317-- $adapters

316-- $adapters

318--

317--

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

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

32 , maxThreads

33 , maxBuffer

34 , maxYields

35− , maxRate

35+ , yieldRate

36 )

37where

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

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

115 ref <- liftIO $ newIORef ()

116 _ <- 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.

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

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

118 where

121 where

119

122

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

134--

137--

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

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

136--

139--

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

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

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

141+-- any concurrent combinator in a stream.

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.

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

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

144--

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

145-- @since 0.4.0

150-- @since 0.4.0

146{-# INLINE_NORMAL maxThreads #-}

151{-# INLINE_NORMAL maxThreads #-}

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

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

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

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

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

176maxBufferSerial _ = id

181maxBufferSerial _ = id

177

182

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

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

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

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

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

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

181--

195--

182-- @since 0.5.0

196-- @since 0.5.0

183−{-# INLINE_NORMAL maxRate #-}

197+{-# INLINE_NORMAL yieldRate #-}

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

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

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

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

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

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

187

201

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

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

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

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

190−maxRateSerial _ = id

204+yieldRateSerial _ = id

191

205

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

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

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

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

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

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

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

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

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

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

198-- restricted by any other control parameters configured.

212-- restricted by any other control parameters configured.

test/Prop.hs+13−13

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

693 let makeOps t =

694 [ ("default", t)

695#ifndef COVERAGE_BUILD

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

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

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

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

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

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

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

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

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

708 serialOps spec = mapOps spec $ (makeOps serially)

709#ifndef COVERAGE_BUILD

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

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

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

712#endif

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

714 wSerialOps spec = mapOps spec $ makeOps wSerially

715#ifndef COVERAGE_BUILD

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

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

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

718#endif

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

720 asyncOps spec = mapOps spec $ makeOps asyncly

721#ifndef COVERAGE_BUILD

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

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

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

724#endif

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

726 wAsyncOps spec = mapOps spec $ makeOps wAsyncly

727#ifndef COVERAGE_BUILD

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

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

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

730#endif

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

732 aheadOps spec = mapOps spec $ makeOps aheadly

733#ifndef COVERAGE_BUILD

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

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

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

736#endif

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

738 parallelOps spec = mapOps spec $ makeOps parallely

739#ifndef COVERAGE_BUILD

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

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

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

742#endif

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

744 zipSerialOps spec = mapOps spec $ makeOps zipSerially

745#ifndef COVERAGE_BUILD

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

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

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

748#endif

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

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

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

752 zipAsyncOps spec = mapOps spec $ makeOps zipAsyncly

753#ifndef COVERAGE_BUILD

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

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

755#endif

756

757 describe "Construction" $ do

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

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

816

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

819 -- zipAsync.

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

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

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

901 let mkOps t =

902 [ ("default", t)

903#ifndef COVERAGE_BUILD

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

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

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

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

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

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

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