git.iko.soy/streamly
on bffdd45 help sign in
streamly/src/Streamly/Internal/Data/Stream/IsStream/Combinators.hs
231 lines 8.8 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Combinators
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9--
10module Streamly.Internal.Data.Stream.IsStream.Combinators
11 ( maxThreads
12 , maxBuffer
13 , maxYields
14 , rate
15 , avgRate
16 , minRate
17 , maxRate
18 , constRate
19 , inspectMode
20 , printState
21 )
22where
24#include "inline.hs"
26import Control.Monad.IO.Class (MonadIO(liftIO))
27import Data.Int (Int64)
29import Streamly.Internal.Data.SVar
30import Streamly.Internal.Data.Stream.StreamK.Type
31import Streamly.Internal.Data.Stream.Serial (SerialT)
33-------------------------------------------------------------------------------
34-- Concurrency control
35-------------------------------------------------------------------------------
36--
37-- XXX need to write these in direct style otherwise they will break fusion.
38--
39-- | Specify the maximum number of threads that can be spawned concurrently for
40-- any concurrent combinator in a stream.
41-- A value of 0 resets the thread limit to default, a negative value means
42-- there is no limit. The default value is 1500. 'maxThreads' does not affect
43-- 'ParallelT' streams as they can use unbounded number of threads.
44--
45-- When the actions in a stream are IO bound, having blocking IO calls, this
46-- option can be used to control the maximum number of in-flight IO requests.
47-- When the actions are CPU bound this option can be used to
48-- control the amount of CPU used by the stream.
49--
50-- /Since: 0.4.0 ("Streamly")/
51--
52-- @since 0.8.0
53{-# INLINE_NORMAL maxThreads #-}
54maxThreads :: IsStream t => Int -> t m a -> t m a
55maxThreads n m = mkStream $ \st stp sng yld ->
56 foldStreamShared (setMaxThreads n st) stp sng yld m
58{-
59{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}
60maxThreadsSerial :: Int -> SerialT m a -> SerialT m a
61maxThreadsSerial _ = id
62-}
64-- | Specify the maximum size of the buffer for storing the results from
65-- concurrent computations. If the buffer becomes full we stop spawning more
66-- concurrent tasks until there is space in the buffer.
67-- A value of 0 resets the buffer size to default, a negative value means
68-- there is no limit. The default value is 1500.
69--
70-- CAUTION! using an unbounded 'maxBuffer' value (i.e. a negative value)
71-- coupled with an unbounded 'maxThreads' value is a recipe for disaster in
72-- presence of infinite streams, or very large streams. Especially, it must
73-- not be used when 'pure' is used in 'ZipAsyncM' streams as 'pure' in
74-- applicative zip streams generates an infinite stream causing unbounded
75-- concurrent generation with no limit on the buffer or threads.
76--
77-- /Since: 0.4.0 ("Streamly")/
78--
79-- @since 0.8.0
80{-# INLINE_NORMAL maxBuffer #-}
81maxBuffer :: IsStream t => Int -> t m a -> t m a
82maxBuffer n m = mkStream $ \st stp sng yld ->
83 foldStreamShared (setMaxBuffer n st) stp sng yld m
85{-
86{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}
87maxBufferSerial :: Int -> SerialT m a -> SerialT m a
88maxBufferSerial _ = id
89-}
91-- | Specify the pull rate of a stream.
92-- A 'Nothing' value resets the rate to default which is unlimited. When the
93-- rate is specified, concurrent production may be ramped up or down
94-- automatically to achieve the specified yield rate. The specific behavior for
95-- different styles of 'Rate' specifications is documented under 'Rate'. The
96-- effective maximum production rate achieved by a stream is governed by:
97--
98-- * The 'maxThreads' limit
99-- * The 'maxBuffer' limit
100-- * The maximum rate that the stream producer can achieve
101-- * The maximum rate that the stream consumer can achieve
102--
103-- /Since: 0.5.0 ("Streamly")/
104--
105-- @since 0.8.0
106{-# INLINE_NORMAL rate #-}
107rate :: IsStream t => Maybe Rate -> t m a -> t m a
108rate r m = mkStream $ \st stp sng yld ->
109 case r of
110 Just (Rate low goal _ _) | goal < low ->
111 error "rate: Target rate cannot be lower than minimum rate."
112 Just (Rate _ goal high _) | goal > high ->
113 error "rate: Target rate cannot be greater than maximum rate."
114 Just (Rate low _ high _) | low > high ->
115 error "rate: Minimum rate cannot be greater than maximum rate."
116 _ -> foldStreamShared (setStreamRate r st) stp sng yld m
118-- XXX implement for serial streams as well, as a simple delay
120{-
121{-# RULES "rate serial" rate = yieldRateSerial #-}
122yieldRateSerial :: Double -> SerialT m a -> SerialT m a
123yieldRateSerial _ = id
124-}
126-- | Same as @rate (Just $ Rate (r/2) r (2*r) maxBound)@
127--
128-- Specifies the average production rate of a stream in number of yields
129-- per second (i.e. @Hertz@). Concurrent production is ramped up or down
130-- automatically to achieve the specified average yield rate. The rate can
131-- go down to half of the specified rate on the lower side and double of
132-- the specified rate on the higher side.
133--
134-- /Since: 0.5.0 ("Streamly")/
135--
136-- @since 0.8.0
137avgRate :: IsStream t => Double -> t m a -> t m a
138avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)
140-- | Same as @rate (Just $ Rate r r (2*r) maxBound)@
141--
142-- Specifies the minimum rate at which the stream should yield values. As
143-- far as possible the yield rate would never be allowed to go below the
144-- specified rate, even though it may possibly go above it at times, the
145-- upper limit is double of the specified rate.
146--
147-- /Since: 0.5.0 ("Streamly")/
148--
149-- @since 0.8.0
150minRate :: IsStream t => Double -> t m a -> t m a
151minRate r = rate (Just $ Rate r r (2*r) maxBound)
153-- | Same as @rate (Just $ Rate (r/2) r r maxBound)@
154--
155-- Specifies the maximum rate at which the stream should yield values. As
156-- far as possible the yield rate would never be allowed to go above the
157-- specified rate, even though it may possibly go below it at times, the
158-- lower limit is half of the specified rate. This can be useful in
159-- applications where certain resource usage must not be allowed to go
160-- beyond certain limits.
161--
162-- /Since: 0.5.0 ("Streamly")/
163--
164-- @since 0.8.0
165maxRate :: IsStream t => Double -> t m a -> t m a
166maxRate r = rate (Just $ Rate (r/2) r r maxBound)
168-- | Same as @rate (Just $ Rate r r r 0)@
169--
170-- Specifies a constant yield rate. If for some reason the actual rate
171-- goes above or below the specified rate we do not try to recover it by
172-- increasing or decreasing the rate in future. This can be useful in
173-- applications like graphics frame refresh where we need to maintain a
174-- constant refresh rate.
175--
176-- /Since: 0.5.0 ("Streamly")/
177--
178-- @since 0.8.0
179constRate :: IsStream t => Double -> t m a -> t m a
180constRate r = rate (Just $ Rate r r r 0)
182-- | Specify the average latency, in nanoseconds, of a single threaded action
183-- in a concurrent composition. Streamly can measure the latencies, but that is
184-- possible only after at least one task has completed. This combinator can be
185-- used to provide a latency hint so that rate control using 'rate' can take
186-- that into account right from the beginning. When not specified then a
187-- default behavior is chosen which could be too slow or too fast, and would be
188-- restricted by any other control parameters configured.
189-- A value of 0 indicates default behavior, a negative value means there is no
190-- limit i.e. zero latency.
191-- This would normally be useful only in high latency and high throughput
192-- cases.
193--
194{-# INLINE_NORMAL _serialLatency #-}
195_serialLatency :: IsStream t => Int -> t m a -> t m a
196_serialLatency n m = mkStream $ \st stp sng yld ->
197 foldStreamShared (setStreamLatency n st) stp sng yld m
199{-
200{-# RULES "serialLatency serial" _serialLatency = serialLatencySerial #-}
201serialLatencySerial :: Int -> SerialT m a -> SerialT m a
202serialLatencySerial _ = id
203-}
205-- Stop concurrent dispatches after this limit. This is useful in API's like
206-- "take" where we want to dispatch only upto the number of elements "take"
207-- needs. This value applies only to the immediate next level and is not
208-- inherited by everything in enclosed scope.
209{-# INLINE_NORMAL maxYields #-}
210maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a
211maxYields n m = mkStream $ \st stp sng yld ->
212 foldStreamShared (setYieldLimit n st) stp sng yld m
214{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}
215maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a
216maxYieldsSerial _ = id
218printState :: MonadIO m => State Stream m a -> m ()
219printState st = liftIO $ do
220 let msv = streamVar st
221 case msv of
222 Just sv -> dumpSVar sv >>= putStrLn
223 Nothing -> putStrLn "No SVar"
225-- | Print debug information about an SVar when the stream ends
226--
227-- /Internal/
228--
229inspectMode :: IsStream t => t m a -> t m a
230inspectMode m = mkStream $ \st stp sng yld ->
231 foldStreamShared (setInspectMode st) stp sng yld m