@@ -1,12 +1,4 @@
1−{-# LANGUAGE CPP #-}
2−{-# LANGUAGE ConstraintKinds #-}
31{-# LANGUAGE FlexibleContexts #-}
4−{-# LANGUAGE FlexibleInstances #-}
5−{-# LANGUAGE MultiParamTypeClasses #-}
6−{-# LANGUAGE RankNTypes #-}
7−{-# LANGUAGE UndecidableInstances #-} -- XXX
8−
9−#include "inline.hs"
102
113-- |
124-- Module : Streamly.Streams.SVar
@@ -19,19 +11,8 @@
1911--
2012--
2113module Streamly.Streams.SVar
22− (
23− fromSVar
14+ ( fromSVar
2415 , toSVar
25− , maxThreads
26− , maxBuffer
27− , maxYields
28− , rate
29− , avgRate
30− , minRate
31− , maxRate
32− , constRate
33− , inspectMode
34− , printState
3516 )
3617where
3718
@@ -39,7 +20,6 @@ import Control.Exception (fromException)
3920import Control.Monad (when)
4021import Control.Monad.Catch (throwM)
4122import Control.Monad.IO.Class (MonadIO(liftIO))
42−import Data.Int (Int64)
4323import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
4424import Data.Maybe (isNothing)
4525import Data.Semigroup ((<>))
@@ -49,20 +29,12 @@ import System.Mem (performMajorGC)
4929
5030import Streamly.SVar
5131import Streamly.Streams.StreamK
52−import Streamly.Streams.Serial (SerialT)
5332
5433printSVar :: SVar t m a -> String -> IO ()
5534printSVar sv how = do
5635 svInfo <- dumpSVar sv
5736 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
5837
59−printState :: MonadIO m => State Stream m a -> m ()
60−printState st = liftIO $ do
61− let msv = streamVar st
62− case msv of
63− Just sv -> dumpSVar sv >>= putStrLn
64− Nothing -> putStrLn "No SVar"
65−
6638-- | Pull a stream from an SVar.
6739{-# NOINLINE fromStreamVar #-}
6840fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
@@ -129,178 +101,3 @@ fromSVar sv =
129101-- be read back from the SVar using 'fromSVar'.
130102toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
131103toSVar sv m = toStreamVar sv (toStream m)
132−
133−-------------------------------------------------------------------------------
134−-- Concurrency control
135−-------------------------------------------------------------------------------
136−--
137−-- XXX need to write these in direct style otherwise they will break fusion.
138−--
139−-- | Specify the maximum number of threads that can be spawned concurrently for
140−-- any concurrent combinator in a stream.
141−-- A value of 0 resets the thread limit to default, a negative value means
142−-- there is no limit. The default value is 1500.
143−--
144−-- When the actions in a stream are IO bound, having blocking IO calls, this
145−-- option can be used to control the maximum number of in-flight IO requests.
146−-- When the actions are CPU bound this option can be used to
147−-- control the amount of CPU used by the stream.
148−--
149−-- @since 0.4.0
150−{-# INLINE_NORMAL maxThreads #-}
151−maxThreads :: IsStream t => Int -> t m a -> t m a
152−maxThreads n m = fromStream $ Stream $ \st stp sng yld ->
153− unStream (toStream m) (setMaxThreads n st) stp sng yld
154−
155−{-
156−{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}
157−maxThreadsSerial :: Int -> SerialT m a -> SerialT m a
158−maxThreadsSerial _ = id
159−-}
160−
161−-- | Specify the maximum size of the buffer for storing the results from
162−-- concurrent computations. If the buffer becomes full we stop spawning more
163−-- concurrent tasks until there is space in the buffer.
164−-- A value of 0 resets the buffer size to default, a negative value means
165−-- there is no limit. The default value is 1500.
166−--
167−-- CAUTION! using an unbounded 'maxBuffer' value (i.e. a negative value)
168−-- coupled with an unbounded 'maxThreads' value is a recipe for disaster in
169−-- presence of infinite streams, or very large streams. Especially, it must
170−-- not be used when 'pure' is used in 'ZipAsyncM' streams as 'pure' in
171−-- applicative zip streams generates an infinite stream causing unbounded
172−-- concurrent generation with no limit on the buffer or threads.
173−--
174−-- @since 0.4.0
175−{-# INLINE_NORMAL maxBuffer #-}
176−maxBuffer :: IsStream t => Int -> t m a -> t m a
177−maxBuffer n m = fromStream $ Stream $ \st stp sng yld ->
178− unStream (toStream m) (setMaxBuffer n st) stp sng yld
179−
180−{-
181−{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}
182−maxBufferSerial :: Int -> SerialT m a -> SerialT m a
183−maxBufferSerial _ = id
184−-}
185−
186−-- | Specify the pull rate of a stream.
187−-- A 'Nothing' value resets the rate to default which is unlimited. When the
188−-- rate is specified, concurrent production may be ramped up or down
189−-- automatically to achieve the specified yield rate. The specific behavior for
190−-- different styles of 'Rate' specifications is documented under 'Rate'. The
191−-- effective maximum production rate achieved by a stream is governed by:
192−--
193−-- * The 'maxThreads' limit
194−-- * The 'maxBuffer' limit
195−-- * The maximum rate that the stream producer can achieve
196−-- * The maximum rate that the stream consumer can achieve
197−--
198−-- @since 0.5.0
199−{-# INLINE_NORMAL rate #-}
200−rate :: IsStream t => Maybe Rate -> t m a -> t m a
201−rate r m = fromStream $ Stream $ \st stp sng yld ->
202− case r of
203− Just (Rate low goal _ _) | goal < low ->
204− error "rate: Target rate cannot be lower than minimum rate."
205− Just (Rate _ goal high _) | goal > high ->
206− error "rate: Target rate cannot be greater than maximum rate."
207− Just (Rate low _ high _) | low > high ->
208− error "rate: Minimum rate cannot be greater than maximum rate."
209− _ -> unStream (toStream m) (setStreamRate r st) stp sng yld
210−
211−-- XXX implement for serial streams as well, as a simple delay
212−
213−{-
214−{-# RULES "rate serial" rate = yieldRateSerial #-}
215−yieldRateSerial :: Double -> SerialT m a -> SerialT m a
216−yieldRateSerial _ = id
217−-}
218−
219−-- | Same as @rate (Just $ Rate (r/2) r (2*r) maxBound)@
220−--
221−-- Specifies the average production rate of a stream in number of yields
222−-- per second (i.e. @Hertz@). Concurrent production is ramped up or down
223−-- automatically to achieve the specified average yield rate. The rate can
224−-- go down to half of the specified rate on the lower side and double of
225−-- the specified rate on the higher side.
226−--
227−-- @since 0.5.0
228−avgRate :: IsStream t => Double -> t m a -> t m a
229−avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)
230−
231−-- | Same as @rate (Just $ Rate r r (2*r) maxBound)@
232−--
233−-- Specifies the minimum rate at which the stream should yield values. As
234−-- far as possible the yield rate would never be allowed to go below the
235−-- specified rate, even though it may possibly go above it at times, the
236−-- upper limit is double of the specified rate.
237−--
238−-- @since 0.5.0
239−minRate :: IsStream t => Double -> t m a -> t m a
240−minRate r = rate (Just $ Rate r r (2*r) maxBound)
241−
242−-- | Same as @rate (Just $ Rate (r/2) r r maxBound)@
243−--
244−-- Specifies the maximum rate at which the stream should yield values. As
245−-- far as possible the yield rate would never be allowed to go above the
246−-- specified rate, even though it may possibly go below it at times, the
247−-- lower limit is half of the specified rate. This can be useful in
248−-- applications where certain resource usage must not be allowed to go
249−-- beyond certain limits.
250−--
251−-- @since 0.5.0
252−maxRate :: IsStream t => Double -> t m a -> t m a
253−maxRate r = rate (Just $ Rate (r/2) r r maxBound)
254−
255−-- | Same as @rate (Just $ Rate r r r 0)@
256−--
257−-- Specifies a constant yield rate. If for some reason the actual rate
258−-- goes above or below the specified rate we do not try to recover it by
259−-- increasing or decreasing the rate in future. This can be useful in
260−-- applications like graphics frame refresh where we need to maintain a
261−-- constant refresh rate.
262−--
263−-- @since 0.5.0
264−constRate :: IsStream t => Double -> t m a -> t m a
265−constRate r = rate (Just $ Rate r r r 0)
266−
267−-- | Specify the average latency, in nanoseconds, of a single threaded action
268−-- in a concurrent composition. Streamly can measure the latencies, but that is
269−-- possible only after at least one task has completed. This combinator can be
270−-- used to provide a latency hint so that rate control using 'rate' can take
271−-- that into account right from the beginning. When not specified then a
272−-- default behavior is chosen which could be too slow or too fast, and would be
273−-- restricted by any other control parameters configured.
274−-- A value of 0 indicates default behavior, a negative value means there is no
275−-- limit i.e. zero latency.
276−-- This would normally be useful only in high latency and high throughput
277−-- cases.
278−--
279−{-# INLINE_NORMAL _serialLatency #-}
280−_serialLatency :: IsStream t => Int -> t m a -> t m a
281−_serialLatency n m = fromStream $ Stream $ \st stp sng yld ->
282− unStream (toStream m) (setStreamLatency n st) stp sng yld
283−
284−{-
285−{-# RULES "serialLatency serial" _serialLatency = serialLatencySerial #-}
286−serialLatencySerial :: Int -> SerialT m a -> SerialT m a
287−serialLatencySerial _ = id
288−-}
289−
290−-- Stop concurrent dispatches after this limit. This is useful in API's like
291−-- "take" where we want to dispatch only upto the number of elements "take"
292−-- needs. This value applies only to the immediate next level and is not
293−-- inherited by everything in enclosed scope.
294−{-# INLINE_NORMAL maxYields #-}
295−maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a
296−maxYields n m = fromStream $ Stream $ \st stp sng yld ->
297− unStream (toStream m) (setYieldLimit n st) stp sng yld
298−
299−{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}
300−maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a
301−maxYieldsSerial _ = id
302−
303−-- | Print debug information about an SVar when the stream ends
304−inspectMode :: IsStream t => t m a -> t m a
305−inspectMode m = fromStream $ Stream $ \st stp sng yld ->
306− unStream (toStream m) (setInspectMode st) stp sng yld