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