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