git.iko.soy/streamly
on 9ccee33 help sign in
streamly/src/Streamly/Streams/Combinators.hs
216 lines 8.5 kB history raw
1{-# LANGUAGE CPP #-}
3#include "inline.hs"
5-- |
6-- Module : Streamly.Streams.Combinators
7-- Copyright : (c) 2017 Harendra Kumar
8--
9-- License : BSD3
10-- Maintainer : harendra.kumar@gmail.com
11-- Stability : experimental
12-- Portability : GHC
13--
14--
15module Streamly.Streams.Combinators
16 ( maxThreads
17 , maxBuffer
18 , maxYields
19 , rate
20 , avgRate
21 , minRate
22 , maxRate
23 , constRate
24 , inspectMode
25 , printState
26 )
27where
29import Control.Monad.IO.Class (MonadIO(liftIO))
30import Data.Int (Int64)
32import Streamly.SVar
33import Streamly.Streams.StreamK
34import Streamly.Streams.Serial (SerialT)
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.
46--
47-- When the actions in a stream are IO bound, having blocking IO calls, this
48-- option can be used to control the maximum number of in-flight IO requests.
49-- When the actions are CPU bound this option can be used to
50-- control the amount of CPU used by the stream.
51--
52-- @since 0.4.0
53{-# INLINE_NORMAL maxThreads #-}
54maxThreads :: IsStream t => Int -> t m a -> t m a
55maxThreads n m = fromStream $ Stream $ \st stp sng yld ->
56 unStream (toStream m) (setMaxThreads n st) stp sng yld
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
78{-# INLINE_NORMAL maxBuffer #-}
79maxBuffer :: IsStream t => Int -> t m a -> t m a
80maxBuffer n m = fromStream $ Stream $ \st stp sng yld ->
81 unStream (toStream m) (setMaxBuffer n st) stp sng yld
83{-
84{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}
85maxBufferSerial :: Int -> SerialT m a -> SerialT m a
86maxBufferSerial _ = id
87-}
89-- | Specify the pull rate of a stream.
90-- A 'Nothing' value resets the rate to default which is unlimited. When the
91-- rate is specified, concurrent production may be ramped up or down
92-- automatically to achieve the specified yield rate. The specific behavior for
93-- different styles of 'Rate' specifications is documented under 'Rate'. The
94-- effective maximum production rate achieved by a stream is governed by:
95--
96-- * The 'maxThreads' limit
97-- * The 'maxBuffer' limit
98-- * The maximum rate that the stream producer can achieve
99-- * The maximum rate that the stream consumer can achieve
100--
101-- @since 0.5.0
102{-# INLINE_NORMAL rate #-}
103rate :: IsStream t => Maybe Rate -> t m a -> t m a
104rate r m = fromStream $ Stream $ \st stp sng yld ->
105 case r of
106 Just (Rate low goal _ _) | goal < low ->
107 error "rate: Target rate cannot be lower than minimum rate."
108 Just (Rate _ goal high _) | goal > high ->
109 error "rate: Target rate cannot be greater than maximum rate."
110 Just (Rate low _ high _) | low > high ->
111 error "rate: Minimum rate cannot be greater than maximum rate."
112 _ -> unStream (toStream m) (setStreamRate r st) stp sng yld
114-- XXX implement for serial streams as well, as a simple delay
116{-
117{-# RULES "rate serial" rate = yieldRateSerial #-}
118yieldRateSerial :: Double -> SerialT m a -> SerialT m a
119yieldRateSerial _ = id
120-}
122-- | Same as @rate (Just $ Rate (r/2) r (2*r) maxBound)@
123--
124-- Specifies the average production rate of a stream in number of yields
125-- per second (i.e. @Hertz@). Concurrent production is ramped up or down
126-- automatically to achieve the specified average yield rate. The rate can
127-- go down to half of the specified rate on the lower side and double of
128-- the specified rate on the higher side.
129--
130-- @since 0.5.0
131avgRate :: IsStream t => Double -> t m a -> t m a
132avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)
134-- | Same as @rate (Just $ Rate r r (2*r) maxBound)@
135--
136-- Specifies the minimum rate at which the stream should yield values. As
137-- far as possible the yield rate would never be allowed to go below the
138-- specified rate, even though it may possibly go above it at times, the
139-- upper limit is double of the specified rate.
140--
141-- @since 0.5.0
142minRate :: IsStream t => Double -> t m a -> t m a
143minRate r = rate (Just $ Rate r r (2*r) maxBound)
145-- | Same as @rate (Just $ Rate (r/2) r r maxBound)@
146--
147-- Specifies the maximum rate at which the stream should yield values. As
148-- far as possible the yield rate would never be allowed to go above the
149-- specified rate, even though it may possibly go below it at times, the
150-- lower limit is half of the specified rate. This can be useful in
151-- applications where certain resource usage must not be allowed to go
152-- beyond certain limits.
153--
154-- @since 0.5.0
155maxRate :: IsStream t => Double -> t m a -> t m a
156maxRate r = rate (Just $ Rate (r/2) r r maxBound)
158-- | Same as @rate (Just $ Rate r r r 0)@
159--
160-- Specifies a constant yield rate. If for some reason the actual rate
161-- goes above or below the specified rate we do not try to recover it by
162-- increasing or decreasing the rate in future. This can be useful in
163-- applications like graphics frame refresh where we need to maintain a
164-- constant refresh rate.
165--
166-- @since 0.5.0
167constRate :: IsStream t => Double -> t m a -> t m a
168constRate r = rate (Just $ Rate r r r 0)
170-- | Specify the average latency, in nanoseconds, of a single threaded action
171-- in a concurrent composition. Streamly can measure the latencies, but that is
172-- possible only after at least one task has completed. This combinator can be
173-- used to provide a latency hint so that rate control using 'rate' can take
174-- that into account right from the beginning. When not specified then a
175-- default behavior is chosen which could be too slow or too fast, and would be
176-- restricted by any other control parameters configured.
177-- A value of 0 indicates default behavior, a negative value means there is no
178-- limit i.e. zero latency.
179-- This would normally be useful only in high latency and high throughput
180-- cases.
181--
182{-# INLINE_NORMAL _serialLatency #-}
183_serialLatency :: IsStream t => Int -> t m a -> t m a
184_serialLatency n m = fromStream $ Stream $ \st stp sng yld ->
185 unStream (toStream m) (setStreamLatency n st) stp sng yld
187{-
188{-# RULES "serialLatency serial" _serialLatency = serialLatencySerial #-}
189serialLatencySerial :: Int -> SerialT m a -> SerialT m a
190serialLatencySerial _ = id
191-}
193-- Stop concurrent dispatches after this limit. This is useful in API's like
194-- "take" where we want to dispatch only upto the number of elements "take"
195-- needs. This value applies only to the immediate next level and is not
196-- inherited by everything in enclosed scope.
197{-# INLINE_NORMAL maxYields #-}
198maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a
199maxYields n m = fromStream $ Stream $ \st stp sng yld ->
200 unStream (toStream m) (setYieldLimit n st) stp sng yld
202{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}
203maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a
204maxYieldsSerial _ = id
206printState :: MonadIO m => State Stream m a -> m ()
207printState st = liftIO $ do
208 let msv = streamVar st
209 case msv of
210 Just sv -> dumpSVar sv >>= putStrLn
211 Nothing -> putStrLn "No SVar"
213-- | Print debug information about an SVar when the stream ends
214inspectMode :: IsStream t => t m a -> t m a
215inspectMode m = fromStream $ Stream $ \st stp sng yld ->
216 unStream (toStream m) (setInspectMode st) stp sng yld