on master help sign in
make custom rate combinators functions

instead of constructors

commit482b971ebb460872d1d8981e667c44708bbc8719 authorHarendra Kumar <harendra.kumar@gmail.com> dateFri 24 Aug 2018, 06:11 (8 years ago) parent4a546a8
8 files changed+87−71
benchmark/Linear.hs+2−2 examples/AcidRain.hs+1−2 examples/CirclingSquare.hs+1−1 src/Streamly.hs+7−1 src/Streamly/SVar.hs+15−56 src/Streamly/Streams/SVar.hs+55−3 test/MaxRate.hs+1−1 test/Prop.hs+5−5
benchmark/Linear.hs+2−2
@@ -130,7 +130,7 @@ main = do
130130 (maxThreads 1 . Ops.sourceUnfoldrM)
131131 -- XXX arbitrarily large maxRate should be the same as maxRate -1
132132 , benchSrcIO asyncly "unfoldrM rate AvgRate 1000000"
133− (rate (Just $ MaxRate 1000000) . Ops.sourceUnfoldrM)
133+ (avgRate 1000000 . Ops.sourceUnfoldrM)
134134 , benchSrcIO asyncly "unfoldrM maxBuffer 1 (1000 ops)"
135135 (maxBuffer 1 . Ops.sourceUnfoldrMN 1000)
136136 -- , benchSrcIO asyncly "fromFoldable" Ops.sourceFromFoldable
@@ -157,7 +157,7 @@ main = do
157157 (maxThreads 1 . Ops.sourceUnfoldrM)
158158 -- XXX arbitrarily large maxRate should be the same as maxRate -1
159159 , benchSrcIO aheadly "unfoldrM rate AvgRate 1000000"
160− (rate (Just $ AvgRate 1000000) . Ops.sourceUnfoldrM)
160+ (avgRate 1000000 . Ops.sourceUnfoldrM)
161161 , benchSrcIO aheadly "unfoldrM maxBuffer 1 (1000 ops)"
162162 (maxBuffer 1 . Ops.sourceUnfoldrMN 1000)
163163 -- , benchSrcIO aheadly "fromFoldable" Ops.sourceFromFoldable
examples/AcidRain.hs+1−2
@@ -22,8 +22,7 @@ userAction = S.repeatM $ liftIO askUser
2222 _ -> putStrLn "What?" >> askUser
2323
2424acidRain :: MonadAsync m => SerialT m Event
25−acidRain = asyncly $ rate (Just $ ConstRate 1) $
26− S.repeatM $ liftIO $ return $ Harm 1
25+acidRain = asyncly $ constRate 1 $ S.repeatM $ liftIO $ return $ Harm 1
2726
2827game :: (MonadAsync m, MonadState Int m) => SerialT m ()
2928game = do
examples/CirclingSquare.hs+1−1
@@ -79,6 +79,6 @@ main :: IO ()
7979main = do
8080 sdlInit
8181 cref <- newIORef (0,0)
82− runStream $ asyncly $ rate (Just $ ConstRate 40)
82+ runStream $ asyncly $ constRate 40
8383 $ S.repeatM (updateController cref)
8484 `parallel` S.repeatM (updateDisplay cref)
src/Streamly.hs+7−1
@@ -112,8 +112,14 @@ module Streamly
112112 -- $concurrency
113113 , maxThreads
114114 , maxBuffer
115+
116+ -- * Rate Limiting
115117 , Rate (..)
116118 , rate
119+ , avgRate
120+ , minRate
121+ , maxRate
122+ , constRate
117123
118124 -- * Folding Containers of Streams
119125 -- $foldutils
@@ -174,7 +180,7 @@ import Streamly.Streams.Ahead
174180import Streamly.Streams.Parallel
175181import Streamly.Streams.Zip
176182import Streamly.Streams.Prelude
177−import Streamly.Streams.SVar (maxThreads, maxBuffer, rate)
183+import Streamly.Streams.SVar
178184import Streamly.SVar (MonadAsync, Rate (..))
179185import Data.Semigroup (Semigroup(..))
180186
src/Streamly/SVar.hs+15−56
@@ -235,42 +235,17 @@ data WorkerInfo = WorkerInfo
235235 , workerLatencyStart :: IORef (Count, TimeSpec)
236236 }
237237
238−-- | Specify the stream yield rate.
238+-- | Specify the stream yield rate in yields per second (@Hertz@). We try to
239+-- keep the average rate at 'rateGoal', if the actual average rate goes above
240+-- or below the goal we try to recover it by increasing or decreasing the
241+-- instantaneous rate but keeping it within 'rateLow' and 'rateHigh' limits.
239242--
240243-- @since 0.5.0
241−data Rate =
242− AvgRate Double
243− -- ^ Specifies the average production rate of a stream in number of yields
244− -- per second (i.e. @Hertz@). Concurrent production is ramped up or down
245− -- automatically to achieve the specified average yield rate. The rate can
246− -- go down to half of the specified rate on the lower side and double of
247− -- the specified rate on the higher side.
248− | MinRate Double
249− -- ^ Specifies the minimum rate at which the stream should yield values. As
250− -- far as possible the yield rate would never be allowed to go below the
251− -- specified rate, even though it may possibly go above it at times, the
252− -- upper limit is double of the specified rate.
253− | MaxRate Double
254− -- ^ Specifies the maximum rate at which the stream should yield values. As
255− -- far as possible the yield rate would never be allowed to go above the
256− -- specified rate, even though it may possibly go below it at times, the
257− -- lower limit is half of the specified rate. This can be useful in
258− -- applications where certain resource usage must not be allowed to go
259− -- beyond certain limits.
260− | ConstRate Double
261− -- ^ Specifies a constant yield rate. If for some reason the actual rate
262− -- goes above or below the specified rate we do not try to recover it by
263− -- increasing or decreasing the rate in future. This can be useful in
264− -- applications like graphics frame refresh where we need to maintain a
265− -- constant refresh rate.
266− | Rate Double Double Double
267− -- ^ This is the most general specification of the stream yield rate and
268− -- all the other specifications can be expressed in terms of this. The
269− -- first argument is the target rate, the second argument is the lower
270− -- limit and the third argument is the upper limit. We try to keep the
271− -- average rate at the target rate specified, if the actual rate goes above
272− -- or below we try to recover the rate by increasing or decreasing it in
273− -- future while keeping the maximum or minimum within the specified range.
244+data Rate = Rate
245+ { rateLow :: Double -- ^ The lower rate limit
246+ , rateGoal :: Double -- ^ The target rate we want to achieve
247+ , rateHigh :: Double -- ^ The upper rate limit
248+ }
274249
275250data LatencyRange = LatencyRange
276251 { minLatency :: NanoSecs
@@ -1828,29 +1803,13 @@ postProcessPaced sv = do
18281803
18291804getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
18301805getYieldRateInfo st = do
1831− let -- convert rate in Hertz to latency in Nanoseconds
1832− rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
1833− -- handle overflowed latency value
1834− toMaxLatency l = if l <= 0 then maxBound else l
1806+ -- convert rate in Hertz to latency in Nanoseconds
1807+ let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
18351808 case getStreamRate st of
1836− Just (AvgRate rate) ->
1837− let l = rateToLatency rate
1838− mx = toMaxLatency (l * 2)
1839− in mkYieldRateInfo l (LatencyRange (l `div` 2) mx)
1840− Just (MinRate rate) ->
1841− let l = rateToLatency rate
1842− in mkYieldRateInfo l (LatencyRange (l `div` 2) l)
1843− Just (MaxRate rate) ->
1844− let l = rateToLatency rate
1845− mx = toMaxLatency (l * 2)
1846− in mkYieldRateInfo l (LatencyRange l mx)
1847− Just (ConstRate rate) ->
1848− let l = rateToLatency rate
1849− in mkYieldRateInfo l (LatencyRange l l)
1850− Just (Rate rate minr maxr) ->
1851− let l = rateToLatency rate
1852− minl = rateToLatency maxr
1853− maxl = rateToLatency minr
1809+ Just (Rate low goal high) ->
1810+ let l = rateToLatency goal
1811+ minl = rateToLatency high
1812+ maxl = rateToLatency low
18541813 in mkYieldRateInfo l (LatencyRange minl maxl)
18551814 Nothing -> return Nothing
18561815
src/Streamly/Streams/SVar.hs+55−3
@@ -33,6 +33,10 @@ module Streamly.Streams.SVar
3333 , maxBuffer
3434 , maxYields
3535 , rate
36+ , avgRate
37+ , minRate
38+ , maxRate
39+ , constRate
3640 )
3741where
3842
@@ -203,11 +207,11 @@ maxBufferSerial _ = id
203207rate :: IsStream t => Maybe Rate -> t m a -> t m a
204208rate r m = fromStream $ Stream $ \st stp sng yld -> do
205209 case r of
206− Just (Rate rt minr _) | rt < minr ->
210+ Just (Rate low goal _) | goal < low ->
207211 error "rate: Target rate cannot be lower than minimum rate."
208− Just (Rate rt _ maxr) | rt > maxr ->
212+ Just (Rate _ goal high) | goal > high ->
209213 error "rate: Target rate cannot be greater than maximum rate."
210− Just (Rate _ minr maxr) | minr > maxr ->
214+ Just (Rate low _ high) | low > high ->
211215 error "rate: Minimum rate cannot be greater than maximum rate."
212216 _ -> unStream (toStream m) (setStreamRate r st) stp sng yld
213217
@@ -217,6 +221,54 @@ yieldRateSerial :: Double -> SerialT m a -> SerialT m a
217221yieldRateSerial _ = id
218222-}
219223
224+-- | Same as @rate (Just $ Rate (r/2) r (2*r))@
225+--
226+-- Specifies the average production rate of a stream in number of yields
227+-- per second (i.e. @Hertz@). Concurrent production is ramped up or down
228+-- automatically to achieve the specified average yield rate. The rate can
229+-- go down to half of the specified rate on the lower side and double of
230+-- the specified rate on the higher side.
231+--
232+-- @since 0.5.0
233+avgRate :: IsStream t => Double -> t m a -> t m a
234+avgRate r = rate (Just $ Rate (r/2) r (2*r))
235+
236+-- | Same as @rate (Just $ Rate r r (2*r))@
237+--
238+-- Specifies the minimum rate at which the stream should yield values. As
239+-- far as possible the yield rate would never be allowed to go below the
240+-- specified rate, even though it may possibly go above it at times, the
241+-- upper limit is double of the specified rate.
242+--
243+-- @since 0.5.0
244+minRate :: IsStream t => Double -> t m a -> t m a
245+minRate r = rate (Just $ Rate r r (2*r))
246+
247+-- | Same as @rate (Just $ Rate (r/2) r r)@
248+--
249+-- Specifies the maximum rate at which the stream should yield values. As
250+-- far as possible the yield rate would never be allowed to go above the
251+-- specified rate, even though it may possibly go below it at times, the
252+-- lower limit is half of the specified rate. This can be useful in
253+-- applications where certain resource usage must not be allowed to go
254+-- beyond certain limits.
255+--
256+-- @since 0.5.0
257+maxRate :: IsStream t => Double -> t m a -> t m a
258+maxRate r = rate (Just $ Rate (r/2) r r)
259+
260+-- | Same as @rate (Just $ Rate r r r)@
261+--
262+-- Specifies a constant yield rate. If for some reason the actual rate
263+-- goes above or below the specified rate we do not try to recover it by
264+-- increasing or decreasing the rate in future. This can be useful in
265+-- applications like graphics frame refresh where we need to maintain a
266+-- constant refresh rate.
267+--
268+-- @since 0.5.0
269+constRate :: IsStream t => Double -> t m a -> t m a
270+constRate r = rate (Just $ Rate r r r)
271+
220272-- | Specify the average latency, in nanoseconds, of a single threaded action
221273-- in a concurrent composition. Streamly can measure the latencies, but that is
222274-- possible only after at least one task has completed. This combinator can be
test/MaxRate.hs+1−1
@@ -43,7 +43,7 @@ measureRate' desc t rval consumerDelay producerDelay dur = do
4343 $ t
4444 $ maxBuffer (-1)
4545 $ maxThreads (-1)
46− $ rate (Just $ AvgRate rval)
46+ $ avgRate rval
4747 $ S.take (round $ rval * 10)
4848 $ S.repeatM $ do
4949 let (t1, t2) = producerDelay
test/Prop.hs+5−5
@@ -693,7 +693,7 @@ main = hspec
693693 let makeOps t =
694694 [ ("default", t)
695695#ifndef COVERAGE_BUILD
696− , ("rate AvgRate 10000", t . rate (Just $ AvgRate 10000))
696+ , ("rate AvgRate 10000", t . avgRate 10000)
697697 , ("rate Nothing", t . rate Nothing)
698698 , ("maxBuffer 0", t . maxBuffer 0)
699699 , ("maxBuffer 1", t . maxBuffer 1)
@@ -707,13 +707,13 @@ 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− ++ [("rate AvgRate 0.00000001", serially . rate (Just $ AvgRate 0.00000001))]
710+ ++ [("rate AvgRate 0.00000001", serially . avgRate 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− ++ [("rate AvgRate 0.00000001", wSerially . rate (Just $ AvgRate 0.00000001))]
716+ ++ [("rate AvgRate 0.00000001", wSerially . avgRate 0.00000001)]
717717 ++ [("maxBuffer (-1)", wSerially . maxBuffer (-1))]
718718#endif
719719 let asyncOps :: IsStream t => ((AsyncT IO a -> t IO a) -> Spec) -> Spec
@@ -734,13 +734,13 @@ main = hspec
734734 let parallelOps :: IsStream t => ((ParallelT IO a -> t IO a) -> Spec) -> Spec
735735 parallelOps spec = mapOps spec $ makeOps parallely
736736#ifndef COVERAGE_BUILD
737− ++ [("rate AvgRate 0.00000001", parallely . rate (Just $ AvgRate 0.00000001))]
737+ ++ [("rate AvgRate 0.00000001", parallely . avgRate 0.00000001)]
738738 ++ [("maxBuffer (-1)", parallely . maxBuffer (-1))]
739739#endif
740740 let zipSerialOps :: IsStream t => ((ZipSerialM IO a -> t IO a) -> Spec) -> Spec
741741 zipSerialOps spec = mapOps spec $ makeOps zipSerially
742742#ifndef COVERAGE_BUILD
743− ++ [("rate AvgRate 0.00000001", zipSerially . rate (Just $ AvgRate 0.00000001))]
743+ ++ [("rate AvgRate 0.00000001", zipSerially . avgRate 0.00000001)]
744744 ++ [("maxBuffer (-1)", zipSerially . maxBuffer (-1))]
745745#endif
746746 -- Note, the "pure" of applicative Zip streams generates and infinite