git.iko.soy/streamly
on 5cbcd5b help sign in
streamly/src/Streamly/Streams/SVar.hs
306 lines 12 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE MultiParamTypeClasses #-}
6{-# LANGUAGE RankNTypes #-}
7{-# LANGUAGE UndecidableInstances #-} -- XXX
9#include "inline.h"
11-- |
12-- Module : Streamly.Streams.SVar
13-- Copyright : (c) 2017 Harendra Kumar
14--
15-- License : BSD3
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
18-- Portability : GHC
19--
20--
21module Streamly.Streams.SVar
22 (
23 fromSVar
24 , toSVar
25 , maxThreads
26 , maxBuffer
27 , maxYields
28 , rate
29 , avgRate
30 , minRate
31 , maxRate
32 , constRate
33 , inspectMode
34 , printState
35 )
36where
38import Control.Exception (fromException)
39import Control.Monad (when)
40import Control.Monad.Catch (throwM)
41import Control.Monad.IO.Class (MonadIO(liftIO))
42import Data.Int (Int64)
43import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
44import Data.Maybe (isNothing)
45import Data.Semigroup ((<>))
46import System.IO (hPutStrLn, stderr)
47import System.Clock (Clock(Monotonic), getTime)
48import System.Mem (performMajorGC)
50import Streamly.SVar
51import Streamly.Streams.StreamK
52import Streamly.Streams.Serial (SerialT)
54printSVar :: SVar t m a -> String -> IO ()
55printSVar sv how = do
56 svInfo <- dumpSVar sv
57 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
59printState :: MonadIO m => State Stream m a -> m ()
60printState st = liftIO $ do
61 let msv = streamVar st
62 case msv of
63 Just sv -> dumpSVar sv >>= putStrLn
64 Nothing -> putStrLn "No SVar"
66-- | Pull a stream from an SVar.
67{-# NOINLINE fromStreamVar #-}
68fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
69fromStreamVar sv = Stream $ \st stp sng yld -> do
70 list <- readOutputQ sv
71 -- Reversing the output is important to guarantee that we process the
72 -- outputs in the same order as they were generated by the constituent
73 -- streams.
74 unStream (processEvents $ reverse list) (rstState st) stp sng yld
76 where
78 allDone stp = do
79 when (svarInspectMode sv) $ do
80 t <- liftIO $ getTime Monotonic
81 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
82 liftIO $ printSVar sv "SVar Done"
83 stp
85 {-# INLINE processEvents #-}
86 processEvents [] = Stream $ \st stp sng yld -> do
87 done <- postProcess sv
88 if done
89 then allDone stp
90 else unStream (fromStreamVar sv) (rstState st) stp sng yld
92 processEvents (ev : es) = Stream $ \st stp sng yld -> do
93 let rest = processEvents es
94 case ev of
95 ChildYield a -> yld a rest
96 ChildStop tid e -> do
97 accountThread sv tid
98 case e of
99 Nothing -> unStream rest (rstState st) stp sng yld
100 Just ex ->
101 case fromException ex of
102 Just ThreadAbort ->
103 unStream rest (rstState st) stp sng yld
104 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
106{-# INLINE fromSVar #-}
107fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
108fromSVar sv =
109 fromStream $ Stream $ \st stp sng yld -> do
110 ref <- liftIO $ newIORef ()
111 _ <- liftIO $ mkWeakIORef ref hook
112 -- We pass a copy of sv to fromStreamVar, so that we know that it has
113 -- no other references, when that copy gets garbage collected "ref"
114 -- will get garbage collected and our hook will be called.
115 unStream (fromStreamVar sv{svarRef = Just ref}) st stp sng yld
116 where
118 hook = do
119 when (svarInspectMode sv) $ do
120 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
121 when (isNothing r) $
122 printSVar sv "SVar Garbage Collected"
123 cleanupSVar sv
124 -- If there are any SVars referenced by this SVar a GC will prompt
125 -- them to be cleaned up quickly.
126 when (svarInspectMode sv) performMajorGC
128-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
129-- be read back from the SVar using 'fromSVar'.
130toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
131toSVar sv m = toStreamVar sv (toStream m)
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 #-}
151maxThreads :: IsStream t => Int -> t m a -> t m a
152maxThreads n m = fromStream $ Stream $ \st stp sng yld ->
153 unStream (toStream m) (setMaxThreads n st) stp sng yld
155{-
156{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}
157maxThreadsSerial :: Int -> SerialT m a -> SerialT m a
158maxThreadsSerial _ = id
159-}
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 #-}
176maxBuffer :: IsStream t => Int -> t m a -> t m a
177maxBuffer n m = fromStream $ Stream $ \st stp sng yld ->
178 unStream (toStream m) (setMaxBuffer n st) stp sng yld
180{-
181{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}
182maxBufferSerial :: Int -> SerialT m a -> SerialT m a
183maxBufferSerial _ = id
184-}
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 #-}
200rate :: IsStream t => Maybe Rate -> t m a -> t m a
201rate 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
211-- XXX implement for serial streams as well, as a simple delay
213{-
214{-# RULES "rate serial" rate = yieldRateSerial #-}
215yieldRateSerial :: Double -> SerialT m a -> SerialT m a
216yieldRateSerial _ = id
217-}
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
228avgRate :: IsStream t => Double -> t m a -> t m a
229avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)
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
239minRate :: IsStream t => Double -> t m a -> t m a
240minRate r = rate (Just $ Rate r r (2*r) maxBound)
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
252maxRate :: IsStream t => Double -> t m a -> t m a
253maxRate r = rate (Just $ Rate (r/2) r r maxBound)
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
264constRate :: IsStream t => Double -> t m a -> t m a
265constRate r = rate (Just $ Rate r r r 0)
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
284{-
285{-# RULES "serialLatency serial" _serialLatency = serialLatencySerial #-}
286serialLatencySerial :: Int -> SerialT m a -> SerialT m a
287serialLatencySerial _ = id
288-}
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 #-}
295maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a
296maxYields n m = fromStream $ Stream $ \st stp sng yld ->
297 unStream (toStream m) (setYieldLimit n st) stp sng yld
299{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}
300maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a
301maxYieldsSerial _ = id
303-- | Print debug information about an SVar when the stream ends
304inspectMode :: IsStream t => t m a -> t m a
305inspectMode m = fromStream $ Stream $ \st stp sng yld ->
306 unStream (toStream m) (setInspectMode st) stp sng yld