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