git.iko.soy/streamly
on master help sign in
Replace "clock" with time
d22a24bHarendra Kumar
src/Streamly/Clock.hsnew file+419
compare with master
unifiedsplit
@@ -0,0 +1,419 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
3+
4+#include "Streams/inline.hs"
5+
6+-- |
7+-- Module : Streamly.Clock
8+-- Copyright : (c) 2019 Harendra Kumar
9+--
10+-- License : BSD3
11+-- Maintainer : harendra.kumar@gmail.com
12+-- Stability : experimental
13+-- Portability : GHC
14+
15+module Streamly.Clock
16+ (
17+ -- * Time Unit Conversions
18+ TimeUnit()
19+ -- , TimeUnitWide()
20+ , TimeUnit64()
21+
22+ -- * Time Units
23+ , TimeSpec(..)
24+ , NanoSecond64(..)
25+ , MicroSecond64(..)
26+ , MilliSecond64(..)
27+
28+ -- * Absolute times (using TimeSpec)
29+ , AbsTime
30+ , toAbsTime
31+ , fromAbsTime
32+
33+ -- * Relative times (using TimeSpec)
34+ , RelTime
35+ , toRelTime
36+ , fromRelTime
37+ , diffAbsTime
38+ , addToAbsTime
39+
40+ -- * Relative times (using NanoSecond64)
41+ , RelTime64
42+ , toRelTime64
43+ , fromRelTime64
44+ , diffAbsTime64
45+ , addToAbsTime64
46+
47+ -- * get time from the system clock
48+ , getTime
49+ )
50+where
51+
52+import Data.Time.Clock.System (SystemTime(..), getSystemTime)
53+import Data.Int
54+
55+-------------------------------------------------------------------------------
56+-- Some constants
57+-------------------------------------------------------------------------------
58+
59+{-# INLINE tenPower3 #-}
60+tenPower3 :: Int64
61+tenPower3 = 1000
62+
63+{-# INLINE tenPower6 #-}
64+tenPower6 :: Int64
65+tenPower6 = 1000000
66+
67+{-# INLINE tenPower9 #-}
68+tenPower9 :: Int64
69+tenPower9 = 1000000000
70+
71+-------------------------------------------------------------------------------
72+-- Time Unit Representations
73+-------------------------------------------------------------------------------
74+
75+-- Double or Fixed would be a much better representation so that we do not lose
76+-- information between conversions. However, for faster arithmetic operations
77+-- we use an 'Int64' here. When we need convservation of values we can use a
78+-- different system of units with a Fixed precision.
79+--
80+-- | An 'Int64' time representation with a nanosecond resolution. It can
81+-- represent time up to ~292 years.
82+newtype NanoSecond64 = NanoSecond64 Int64
83+ deriving ( Eq
84+ , Read
85+ , Show
86+ , Enum
87+ , Bounded
88+ , Num
89+ , Real
90+ , Integral
91+ , Ord
92+ )
93+
94+-- | An 'Int64' time representation with a microsecond resolution.
95+-- It can represent time up to ~292,000 years.
96+newtype MicroSecond64 = MicroSecond64 Int64
97+ deriving ( Eq
98+ , Read
99+ , Show
100+ , Enum
101+ , Bounded
102+ , Num
103+ , Real
104+ , Integral
105+ , Ord
106+ )
107+
108+-- | An 'Int64' time representation with a millisecond resolution.
109+-- It can represent time up to ~292 million years.
110+newtype MilliSecond64 = MilliSecond64 Int64
111+ deriving ( Eq
112+ , Read
113+ , Show
114+ , Enum
115+ , Bounded
116+ , Num
117+ , Real
118+ , Integral
119+ , Ord
120+ )
121+
122+-------------------------------------------------------------------------------
123+-- TimeSpec representation
124+-------------------------------------------------------------------------------
125+
126+-- A structure storing seconds and nanoseconds as 'Int64' is the simplest and
127+-- fastest way to store practically large quantities of time with efficient
128+-- arithmetic operations. If we store nanoseconds using 'Integer' it can store
129+-- practically unbounded quantities but it may not be as efficient to
130+-- manipulate in performance critical applications. XXX need to measure the
131+-- performance.
132+--
133+-- | Data type to represent practically large quantities of time efficiently.
134+-- It can represent time up to ~292 billion years at nanosecond resolution.
135+data TimeSpec = TimeSpec
136+ { sec :: {-# UNPACK #-} !Int64 -- ^ seconds
137+ , nsec :: {-# UNPACK #-} !Int64 -- ^ nanoseconds
138+ } deriving (Eq, Read, Show)
139+
140+-- We assume that nsec is always less than 10^9. When TimeSpec is negative then
141+-- both sec and nsec are negative.
142+instance Ord TimeSpec where
143+ compare (TimeSpec s1 ns1) (TimeSpec s2 ns2) =
144+ if s1 == s2
145+ then compare ns1 ns2
146+ else compare s1 s2
147+
148+-- make sure nsec is less than 10^9
149+{-# INLINE addWithOverflow #-}
150+addWithOverflow :: TimeSpec -> TimeSpec -> TimeSpec
151+addWithOverflow (TimeSpec s1 ns1) (TimeSpec s2 ns2) =
152+ let nsum = ns1 + ns2
153+ (s', ns) = if (nsum > tenPower9 || nsum < negate tenPower9)
154+ then nsum `divMod` tenPower9
155+ else (0, nsum)
156+ in TimeSpec (s1 + s2 + s') ns
157+
158+-- make sure both sec and nsec have the same sign
159+{-# INLINE adjustSign #-}
160+adjustSign :: TimeSpec -> TimeSpec
161+adjustSign (t@(TimeSpec s ns)) =
162+ if (s > 0 && ns < 0)
163+ then TimeSpec (s - 1) (ns + tenPower9)
164+ else if (s < 0 && ns > 0)
165+ then TimeSpec (s + 1) (ns - tenPower9)
166+ else t
167+
168+{-# INLINE timeSpecToInteger #-}
169+timeSpecToInteger :: TimeSpec -> Integer
170+timeSpecToInteger (TimeSpec s ns) = toInteger $ s * tenPower9 + ns
171+
172+instance Num TimeSpec where
173+ {-# INLINE (+) #-}
174+ t1 + t2 = adjustSign (addWithOverflow t1 t2)
175+
176+ -- XXX will this be more optimal if imlemented without "negate"?
177+ {-# INLINE (-) #-}
178+ t1 - t2 = t1 + (negate t2)
179+ t1 * t2 = fromInteger $ timeSpecToInteger t1 * timeSpecToInteger t2
180+
181+ {-# INLINE negate #-}
182+ negate (TimeSpec s ns) = TimeSpec (negate s) (negate ns)
183+ {-# INLINE abs #-}
184+ abs (TimeSpec s ns) = TimeSpec (abs s) (abs ns)
185+ {-# INLINE signum #-}
186+ signum (TimeSpec s ns) | s == 0 = TimeSpec (signum ns) 0
187+ | otherwise = TimeSpec (signum s) 0
188+ -- This is fromNanoSecond64 Integer
189+ {-# INLINE fromInteger #-}
190+ fromInteger nanosec = TimeSpec (fromInteger s) (fromInteger ns)
191+ where (s, ns) = nanosec `divMod` toInteger tenPower9
192+
193+-------------------------------------------------------------------------------
194+-- Time unit conversions
195+-------------------------------------------------------------------------------
196+
197+-- TODO: compare whether using TimeSpec instead of Integer provides significant
198+-- performance boost. If not then we can just use Integer nanoseconds and get
199+-- rid of TimeUnitWide.
200+--
201+-- | A type class for converting between time units using 'Integer' as the
202+-- intermediate and the widest representation with a nanosecond resolution.
203+-- This system of units can represent arbitrarily large times but provides
204+-- least efficient arithmetic operations due to 'Integer' arithmetic.
205+--
206+-- NOTE: Converting to and from units may truncate the value depending on the
207+-- original value and the size and resolution of the destination unit.
208+{-
209+class TimeUnitWide a where
210+ toTimeInteger :: a -> Integer
211+ fromTimeInteger :: Integer -> a
212+-}
213+
214+-- | A type class for converting between units of time using 'TimeSpec' as the
215+-- intermediate representation. This system of units can represent up to ~292
216+-- billion years at nanosecond resolution with reasonably efficient arithmetic
217+-- operations.
218+--
219+-- NOTE: Converting to and from units may truncate the value depending on the
220+-- original value and the size and resolution of the destination unit.
221+class TimeUnit a where
222+ toTimeSpec :: a -> TimeSpec
223+ fromTimeSpec :: TimeSpec -> a
224+
225+-- XXX we can use a fromNanoSecond64 for conversion with overflow check and
226+-- fromNanoSecond64Unsafe for conversion without overflow check.
227+--
228+-- | A type class for converting between units of time using 'Int64' as the
229+-- intermediate representation with a nanosecond resolution. This system of
230+-- units can represent up to ~292 years at nanosecond resolution with fast
231+-- arithmetic operations.
232+--
233+-- NOTE: Converting to and from units may truncate the value depending on the
234+-- original value and the size and resolution of the destination unit.
235+class TimeUnit64 a where
236+ toNanoSecond64 :: a -> NanoSecond64
237+ fromNanoSecond64 :: NanoSecond64 -> a
238+
239+-------------------------------------------------------------------------------
240+-- Time units
241+-------------------------------------------------------------------------------
242+
243+instance TimeUnit TimeSpec where
244+ toTimeSpec = id
245+ fromTimeSpec = id
246+
247+instance TimeUnit NanoSecond64 where
248+ {-# INLINE toTimeSpec #-}
249+ toTimeSpec (NanoSecond64 t) = TimeSpec s ns
250+ where (s, ns) = t `divMod` tenPower9
251+
252+ {-# INLINE fromTimeSpec #-}
253+ fromTimeSpec (TimeSpec s ns) =
254+ NanoSecond64 $ s * tenPower9 + ns
255+
256+instance TimeUnit64 NanoSecond64 where
257+ {-# INLINE toNanoSecond64 #-}
258+ toNanoSecond64 = id
259+
260+ {-# INLINE fromNanoSecond64 #-}
261+ fromNanoSecond64 = id
262+
263+instance TimeUnit MicroSecond64 where
264+ {-# INLINE toTimeSpec #-}
265+ toTimeSpec (MicroSecond64 t) = TimeSpec s us
266+ where (s, us) = t `divMod` tenPower6
267+
268+ {-# INLINE fromTimeSpec #-}
269+ fromTimeSpec (TimeSpec s us) =
270+ MicroSecond64 $ s * tenPower6 + us
271+
272+instance TimeUnit64 MicroSecond64 where
273+ {-# INLINE toNanoSecond64 #-}
274+ toNanoSecond64 (MicroSecond64 us) = NanoSecond64 $ us * tenPower3
275+
276+ {-# INLINE fromNanoSecond64 #-}
277+ fromNanoSecond64 (NanoSecond64 ns) = MicroSecond64 $ ns `div` tenPower3
278+
279+instance TimeUnit MilliSecond64 where
280+ {-# INLINE toTimeSpec #-}
281+ toTimeSpec (MilliSecond64 t) = TimeSpec s us
282+ where (s, us) = t `divMod` tenPower3
283+
284+ {-# INLINE fromTimeSpec #-}
285+ fromTimeSpec (TimeSpec s us) =
286+ MilliSecond64 $ s * tenPower3 + us
287+
288+instance TimeUnit64 MilliSecond64 where
289+ {-# INLINE toNanoSecond64 #-}
290+ toNanoSecond64 (MilliSecond64 us) = NanoSecond64 $ us * tenPower6
291+
292+ {-# INLINE fromNanoSecond64 #-}
293+ fromNanoSecond64 (NanoSecond64 ns) = MilliSecond64 $ ns `div` tenPower6
294+
295+-------------------------------------------------------------------------------
296+-- Absolute time
297+-------------------------------------------------------------------------------
298+
299+-- | Absolute times are relative to a predefined epoch in time. 'AbsTime'
300+-- represents times using 'TimeSpec' which can represent times up to ~292
301+-- billion years at a nanosecond resolution.
302+newtype AbsTime = AbsTime TimeSpec
303+ deriving (Eq, Ord, Show)
304+
305+-- | Convert a 'TimeUnit' to an absolute time.
306+{-# INLINE_NORMAL toAbsTime #-}
307+toAbsTime :: TimeUnit a => a -> AbsTime
308+toAbsTime = AbsTime . toTimeSpec
309+
310+-- | Convert absolute time to a 'TimeUnit'.
311+{-# INLINE_NORMAL fromAbsTime #-}
312+fromAbsTime :: TimeUnit a => AbsTime -> a
313+fromAbsTime (AbsTime t) = fromTimeSpec t
314+
315+-- XXX We can also write rewrite rules to simplify divisions multiplications
316+-- and additions when manipulating units. Though, that might get simplified at
317+-- the assembly (llvm) level as well. Note to/from conversions may be lossy and
318+-- therefore this equation may not hold, but that's ok.
319+{-# RULES "fromAbsTime/toAbsTime" forall a. toAbsTime (fromAbsTime a) = a #-}
320+{-# RULES "toAbsTime/fromAbsTime" forall a. fromAbsTime (toAbsTime a) = a #-}
321+
322+-------------------------------------------------------------------------------
323+-- Relative time using NaonoSecond64 as the underlying representation
324+-------------------------------------------------------------------------------
325+
326+-- We use a separate type to represent relative time for safety and speed.
327+-- RelTime has a Num instance, absolute time doesn't. Relative times are
328+-- usually shorter and for our purposes an Int64 nanoseconds can hold close to
329+-- thousand year duration. It is also faster to manipulate. We do not check for
330+-- overflows during manipulations so use it only when you know the time cannot
331+-- be too big. If you need a bigger RelTime representation then use RelTimeBig.
332+
333+-- | Relative times are relative to some arbitrary point of time. Unlike
334+-- 'AbsTime' they are not relative to a predefined epoch.
335+newtype RelTime64 = RelTime64 NanoSecond64
336+ deriving ( Eq
337+ , Read
338+ , Show
339+ , Enum
340+ , Bounded
341+ , Num
342+ , Real
343+ , Integral
344+ , Ord
345+ )
346+
347+-- | Convert a 'TimeUnit' to a relative time.
348+{-# INLINE_NORMAL toRelTime64 #-}
349+toRelTime64 :: TimeUnit64 a => a -> RelTime64
350+toRelTime64 = RelTime64 . toNanoSecond64
351+
352+-- | Convert relative time to a 'TimeUnit'.
353+{-# INLINE_NORMAL fromRelTime64 #-}
354+fromRelTime64 :: TimeUnit64 a => RelTime64 -> a
355+fromRelTime64 (RelTime64 t) = fromNanoSecond64 t
356+
357+{-# RULES "fromRelTime64/toRelTime64" forall a .
358+ toRelTime64 (fromRelTime64 a) = a #-}
359+
360+{-# RULES "toRelTime64/fromRelTime64" forall a .
361+ fromRelTime64 (toRelTime64 a) = a #-}
362+
363+-- | Difference between two absolute points of time.
364+{-# INLINE diffAbsTime64 #-}
365+diffAbsTime64 :: AbsTime -> AbsTime -> RelTime64
366+diffAbsTime64 (AbsTime (TimeSpec s1 ns1)) (AbsTime (TimeSpec s2 ns2)) =
367+ RelTime64 $ NanoSecond64 $ ((s1 - s2) * tenPower9) + (ns1 - ns2)
368+
369+{-# INLINE addToAbsTime64 #-}
370+addToAbsTime64 :: AbsTime -> RelTime64 -> AbsTime
371+addToAbsTime64 (AbsTime (TimeSpec s1 ns1)) (RelTime64 (NanoSecond64 ns2)) =
372+ AbsTime $ TimeSpec (s1 + s) ns
373+ where (s, ns) = (ns1 + ns2) `divMod` tenPower9
374+
375+-------------------------------------------------------------------------------
376+-- Relative time using TimeSpec as the underlying representation
377+-------------------------------------------------------------------------------
378+
379+newtype RelTime = RelTime TimeSpec
380+ deriving ( Eq
381+ , Read
382+ , Show
383+ -- , Enum
384+ -- , Bounded
385+ , Num
386+ -- , Real
387+ -- , Integral
388+ , Ord
389+ )
390+
391+{-# INLINE_NORMAL toRelTime #-}
392+toRelTime :: TimeUnit a => a -> RelTime
393+toRelTime = RelTime . toTimeSpec
394+
395+{-# INLINE_NORMAL fromRelTime #-}
396+fromRelTime :: TimeUnit a => RelTime -> a
397+fromRelTime (RelTime t) = fromTimeSpec t
398+
399+{-# RULES "fromRelTime/toRelTime" forall a. toRelTime (fromRelTime a) = a #-}
400+{-# RULES "toRelTime/fromRelTime" forall a. fromRelTime (toRelTime a) = a #-}
401+
402+{-# INLINE diffAbsTime #-}
403+diffAbsTime :: AbsTime -> AbsTime -> RelTime
404+diffAbsTime (AbsTime t1) (AbsTime t2) = RelTime (t1 - t2)
405+
406+{-# INLINE addToAbsTime #-}
407+addToAbsTime :: AbsTime -> RelTime -> AbsTime
408+addToAbsTime (AbsTime t1) (RelTime t2) = AbsTime $ t1 + t2
409+
410+-------------------------------------------------------------------------------
411+-- Clock time
412+-------------------------------------------------------------------------------
413+
414+-- XXX Use our own implementation using clock_gettime and remove the dependency
415+-- on "time".
416+getTime :: IO AbsTime
417+getTime = do
418+ MkSystemTime s ns <- getSystemTime
419+ return $ AbsTime $ TimeSpec s (fromIntegral ns)
src/Streamly/SVar.hs+48−63
@@ -100,7 +100,6 @@ module Streamly.SVar
100
100
101 , toStreamVar
101 , toStreamVar
102 , SVarStats (..)
102 , SVarStats (..)
103− , NanoSecs (..)
104 , dumpSVar
103 , dumpSVar
105 )
104 )
106where
105where
@@ -133,27 +132,15 @@ import Data.Set (Set)
133import GHC.Conc (ThreadId(..))
132import GHC.Conc (ThreadId(..))
134import GHC.Exts
133import GHC.Exts
135import GHC.IO (IO(..))
134import GHC.IO (IO(..))
136−import System.Clock (TimeSpec, Clock(Monotonic), getTime, toNanoSecs)
135+import Streamly.Clock
136+ (AbsTime, NanoSecond64(..), MicroSecond64(..), getTime, diffAbsTime64,
137+ fromRelTime64, toRelTime64)
137import System.IO (hPutStrLn, stderr)
138import System.IO (hPutStrLn, stderr)
138import Text.Printf (printf)
139import Text.Printf (printf)
139
140
140import qualified Data.Heap as H
141import qualified Data.Heap as H
141import qualified Data.Set as S
142import qualified Data.Set as S
142
143
143−-- Always use signed arithmetic to avoid inadvertant overflows of signed values
144−-- on conversion when comparing unsigned quantities with signed.
145−newtype NanoSecs = NanoSecs Int64
146− deriving ( Eq
147− , Read
148− , Show
149− , Enum
150− , Bounded
151− , Num
152− , Real
153− , Integral
154− , Ord
155− )
156−
157newtype Count = Count Int64
144newtype Count = Count Int64
158 deriving ( Eq
145 deriving ( Eq
159 , Read
146 , Read
@@ -230,7 +217,7 @@ data WorkerInfo = WorkerInfo
230 -- total number of yields by the worker till now
217 -- total number of yields by the worker till now
231 , workerYieldCount :: IORef Count
218 , workerYieldCount :: IORef Count
232 -- yieldCount at start, timestamp
219 -- yieldCount at start, timestamp
233− , workerLatencyStart :: IORef (Count, TimeSpec)
220+ , workerLatencyStart :: IORef (Count, AbsTime)
234 }
221 }
235
222
236
223
@@ -262,13 +249,13 @@ data Rate = Rate
262 }
249 }
263
250
264data LatencyRange = LatencyRange
251data LatencyRange = LatencyRange
265− { minLatency :: NanoSecs
252+ { minLatency :: NanoSecond64
266− , maxLatency :: NanoSecs
253+ , maxLatency :: NanoSecond64
267 } deriving Show
254 } deriving Show
268
255
269-- Rate control.
256-- Rate control.
270data YieldRateInfo = YieldRateInfo
257data YieldRateInfo = YieldRateInfo
271− { svarLatencyTarget :: NanoSecs
258+ { svarLatencyTarget :: NanoSecond64
272 , svarLatencyRange :: LatencyRange
259 , svarLatencyRange :: LatencyRange
273 , svarRateBuffer :: Int
260 , svarRateBuffer :: Int
274 , svarGainedLostYields :: IORef Count
261 , svarGainedLostYields :: IORef Count
@@ -279,11 +266,11 @@ data YieldRateInfo = YieldRateInfo
279 -- rate. The idle time of workers is adjusted in this, so that we only
266 -- rate. The idle time of workers is adjusted in this, so that we only
280 -- account for the rate when the consumer actually demands data.
267 -- account for the rate when the consumer actually demands data.
281 -- XXX interval latency is enough, we can move this under diagnostics build
268 -- XXX interval latency is enough, we can move this under diagnostics build
282− , svarAllTimeLatency :: IORef (Count, TimeSpec)
269+ , svarAllTimeLatency :: IORef (Count, AbsTime)
283
270
284 -- XXX Worker latency specified by the user to be used before the first
271 -- XXX Worker latency specified by the user to be used before the first
285 -- actual measurement arrives. Not yet implemented
272 -- actual measurement arrives. Not yet implemented
286− , workerBootstrapLatency :: Maybe NanoSecs
273+ , workerBootstrapLatency :: Maybe NanoSecond64
287
274
288 -- After how many yields the worker should update the latency information.
275 -- After how many yields the worker should update the latency information.
289 -- If the latency is high, this count is kept lower and vice-versa. XXX If
276 -- If the latency is high, this count is kept lower and vice-versa. XXX If
@@ -297,7 +284,7 @@ data YieldRateInfo = YieldRateInfo
297 -- empty into workerCollectedLatency stats at certain intervals - whenever
284 -- empty into workerCollectedLatency stats at certain intervals - whenever
298 -- we process the stream elements yielded in this period.
285 -- we process the stream elements yielded in this period.
299 -- (yieldCount, timeTaken)
286 -- (yieldCount, timeTaken)
300− , workerPendingLatency :: IORef (Count, NanoSecs)
287+ , workerPendingLatency :: IORef (Count, NanoSecond64)
301
288
302 -- This is the second level stat which is an accmulation from
289 -- This is the second level stat which is an accmulation from
303 -- workerPendingLatency stats. We keep accumulating latencies in this
290 -- workerPendingLatency stats. We keep accumulating latencies in this
@@ -305,10 +292,10 @@ data YieldRateInfo = YieldRateInfo
305 -- to start collecting for the next period and retain the computed average
292 -- to start collecting for the next period and retain the computed average
306 -- latency for the last period in workerMeasuredLatency.
293 -- latency for the last period in workerMeasuredLatency.
307 -- (yieldCount, timeTaken)
294 -- (yieldCount, timeTaken)
308− , workerCollectedLatency :: IORef (Count, NanoSecs)
295+ , workerCollectedLatency :: IORef (Count, NanoSecond64)
309
296
310 -- Latency as measured by workers, aggregated for the last period.
297 -- Latency as measured by workers, aggregated for the last period.
311− , workerMeasuredLatency :: IORef NanoSecs
298+ , workerMeasuredLatency :: IORef NanoSecond64
312 }
299 }
313
300
314data SVarStats = SVarStats {
301data SVarStats = SVarStats {
@@ -317,10 +304,10 @@ data SVarStats = SVarStats {
317 , maxOutQSize :: IORef Int
304 , maxOutQSize :: IORef Int
318 , maxHeapSize :: IORef Int
305 , maxHeapSize :: IORef Int
319 , maxWorkQSize :: IORef Int
306 , maxWorkQSize :: IORef Int
320− , avgWorkerLatency :: IORef (Count, NanoSecs)
307+ , avgWorkerLatency :: IORef (Count, NanoSecond64)
321− , minWorkerLatency :: IORef NanoSecs
308+ , minWorkerLatency :: IORef NanoSecond64
322− , maxWorkerLatency :: IORef NanoSecs
309+ , maxWorkerLatency :: IORef NanoSecond64
323− , svarStopTime :: IORef (Maybe TimeSpec)
310+ , svarStopTime :: IORef (Maybe AbsTime)
324}
311}
325
312
326data Limit = Unlimited | Limited Word deriving Show
313data Limit = Unlimited | Limited Word deriving Show
@@ -393,7 +380,7 @@ data State t m a = State
393 , _threadsHigh :: Limit
380 , _threadsHigh :: Limit
394 , _bufferHigh :: Limit
381 , _bufferHigh :: Limit
395 -- XXX these two can be collapsed into a single type
382 -- XXX these two can be collapsed into a single type
396− , _streamLatency :: Maybe NanoSecs -- bootstrap latency
383+ , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
397 , _maxStreamRate :: Maybe Rate
384 , _maxStreamRate :: Maybe Rate
398 , _inspectMode :: Bool
385 , _inspectMode :: Bool
399 }
386 }
@@ -505,7 +492,7 @@ setStreamLatency n st =
505 else Just (fromIntegral n)
492 else Just (fromIntegral n)
506 }
493 }
507
494
508−getStreamLatency :: State t m a -> Maybe NanoSecs
495+getStreamLatency :: State t m a -> Maybe NanoSecond64
509getStreamLatency = _streamLatency
496getStreamLatency = _streamLatency
510
497
511setInspectMode :: State t m a -> State t m a
498setInspectMode :: State t m a -> State t m a
@@ -561,7 +548,7 @@ secs k
561 | otherwise = printf "%.3f %s" t u
548 | otherwise = printf "%.3f %s" t u
562
549
563-- XXX Code duplicated from collectLatency
550-- XXX Code duplicated from collectLatency
564−drainLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
551+drainLatency :: SVar t m a -> YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
565drainLatency sv yinfo = do
552drainLatency sv yinfo = do
566 let cur = workerPendingLatency yinfo
553 let cur = workerPendingLatency yinfo
567 col = workerCollectedLatency yinfo
554 col = workerCollectedLatency yinfo
@@ -624,11 +611,11 @@ dumpSVarStats sv ss style = do
624 gl <- readIORef (svarGainedLostYields yinfo)
611 gl <- readIORef (svarGainedLostYields yinfo)
625 case t of
612 case t of
626 Nothing -> do
613 Nothing -> do
627− now <- getTime Monotonic
614+ now <- getTime
628− let interval = toNanoSecs (now - startTime)
615+ let interval = diffAbsTime64 now startTime
629 return (cnt, gl, interval `div` fromIntegral cnt)
616 return (cnt, gl, interval `div` fromIntegral cnt)
630 Just stopTime -> do
617 Just stopTime -> do
631− let interval = toNanoSecs (stopTime - startTime)
618+ let interval = diffAbsTime64 stopTime startTime
632 return (cnt, gl, interval `div` fromIntegral cnt)
619 return (cnt, gl, interval `div` fromIntegral cnt)
633 else return (0, 0, 0)
620 else return (0, 0, 0)
634
621
@@ -898,7 +885,7 @@ send sv msg = do
898 active <- readIORef (workerCount sv)
885 active <- readIORef (workerCount sv)
899 return $ len < (fromIntegral lim - active)
886 return $ len < (fromIntegral lim - active)
900
887
901−workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecs))
888+workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
902workerCollectLatency winfo = do
889workerCollectLatency winfo = do
903 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
890 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
904 cnt1 <- readIORef (workerYieldCount winfo)
891 cnt1 <- readIORef (workerYieldCount winfo)
@@ -906,8 +893,8 @@ workerCollectLatency winfo = do
906
893
907 if cnt > 0
894 if cnt > 0
908 then do
895 then do
909− t1 <- getTime Monotonic
896+ t1 <- getTime
910− let period = fromInteger $ toNanoSecs (t1 - t0)
897+ let period = fromRelTime64 $ diffAbsTime64 t1 t0
911 writeIORef (workerLatencyStart winfo) (cnt1, t1)
898 writeIORef (workerLatencyStart winfo) (cnt1, t1)
912 return $ Just (cnt, period)
899 return $ Just (cnt, period)
913 else return Nothing
900 else return Nothing
@@ -1305,7 +1292,7 @@ pushWorker yieldMax sv = do
1305 Nothing -> return Nothing
1292 Nothing -> return Nothing
1306 Just _ -> liftIO $ do
1293 Just _ -> liftIO $ do
1307 cntRef <- newIORef 0
1294 cntRef <- newIORef 0
1308− t <- getTime Monotonic
1295+ t <- getTime
1309 lat <- newIORef (0, t)
1296 lat <- newIORef (0, t)
1310 return $ Just WorkerInfo
1297 return $ Just WorkerInfo
1311 { workerYieldMax = yieldMax
1298 { workerYieldMax = yieldMax
@@ -1348,7 +1335,7 @@ pushWorkerPar sv wloop =
1348 Nothing -> return Nothing
1335 Nothing -> return Nothing
1349 Just _ -> liftIO $ do
1336 Just _ -> liftIO $ do
1350 cntRef <- newIORef 0
1337 cntRef <- newIORef 0
1351− t <- getTime Monotonic
1338+ t <- getTime
1352 lat <- newIORef (0, t)
1339 lat <- newIORef (0, t)
1353 return $ Just WorkerInfo
1340 return $ Just WorkerInfo
1354 { workerYieldMax = 0
1341 { workerYieldMax = 0
@@ -1420,24 +1407,21 @@ dispatchWorker yieldCount sv = do
1420-- sleep time is accumulated.
1407-- sleep time is accumulated.
1421-- 3. Collected latencies are computed and transferred to measured latency
1408-- 3. Collected latencies are computed and transferred to measured latency
1422-- after a minimum of this period.
1409-- after a minimum of this period.
1423−minThreadDelay :: NanoSecs
1410+minThreadDelay :: NanoSecond64
1424−minThreadDelay = 10^(6 :: Int)
1411+minThreadDelay = 1000000
1425
1412
1426-- | Another magic number! When we have to start more workers to cover up a
1413-- | Another magic number! When we have to start more workers to cover up a
1427-- number of yields that we are lagging by then we cannot start one worker for
1414-- number of yields that we are lagging by then we cannot start one worker for
1428-- each yield because that may be a very big number and if the latency of the
1415-- each yield because that may be a very big number and if the latency of the
1429-- workers is low these number of yields could be very high. We assume that we
1416-- workers is low these number of yields could be very high. We assume that we
1430-- run each extra worker for at least this much time.
1417-- run each extra worker for at least this much time.
1431−rateRecoveryTime :: NanoSecs
1418+rateRecoveryTime :: NanoSecond64
1432rateRecoveryTime = 1000000
1419rateRecoveryTime = 1000000
1433
1420
1434−nanoToMicroSecs :: NanoSecs -> Int
1435−nanoToMicroSecs s = fromIntegral s `div` 1000
1436−
1437-- We either block, or send one worker with limited yield count or one or more
1421-- We either block, or send one worker with limited yield count or one or more
1438-- workers with unlimited yield count.
1422-- workers with unlimited yield count.
1439data Work
1423data Work
1440− = BlockWait NanoSecs
1424+ = BlockWait NanoSecond64
1441 | PartialWorker Count
1425 | PartialWorker Count
1442 | ManyWorkers Int Count
1426 | ManyWorkers Int Count
1443 deriving Show
1427 deriving Show
@@ -1447,9 +1431,9 @@ estimateWorkers
1447 :: Limit
1431 :: Limit
1448 -> Count
1432 -> Count
1449 -> Count
1433 -> Count
1450− -> NanoSecs
1434+ -> NanoSecond64
1451− -> NanoSecs
1435+ -> NanoSecond64
1452− -> NanoSecs
1436+ -> NanoSecond64
1453 -> LatencyRange
1437 -> LatencyRange
1454 -> Work
1438 -> Work
1455estimateWorkers workerLimit svarYields gainLossYields
1439estimateWorkers workerLimit svarYields gainLossYields
@@ -1500,7 +1484,7 @@ estimateWorkers workerLimit svarYields gainLossYields
1500 fromIntegral rateRecoveryTime
1484 fromIntegral rateRecoveryTime
1501 yieldsFreq = 1.0 / fromIntegral targetLat
1485 yieldsFreq = 1.0 / fromIntegral targetLat
1502 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1486 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1503− requiredLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1487+ requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1504 adjustedLat = min (max requiredLat (minLatency range))
1488 adjustedLat = min (max requiredLat (minLatency range))
1505 (maxLatency range)
1489 (maxLatency range)
1506 in assert (adjustedLat > 0) $
1490 in assert (adjustedLat > 0) $
@@ -1527,7 +1511,7 @@ estimateWorkers workerLimit svarYields gainLossYields
1527-- | Get the worker latency without resetting workerPendingLatency
1511-- | Get the worker latency without resetting workerPendingLatency
1528-- Returns (total yield count, base time, measured latency)
1512-- Returns (total yield count, base time, measured latency)
1529-- CAUTION! keep it in sync with collectLatency
1513-- CAUTION! keep it in sync with collectLatency
1530−getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1514+getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1531getWorkerLatency yinfo = do
1515getWorkerLatency yinfo = do
1532 let cur = workerPendingLatency yinfo
1516 let cur = workerPendingLatency yinfo
1533 col = workerCollectedLatency yinfo
1517 col = workerCollectedLatency yinfo
@@ -1552,8 +1536,8 @@ getWorkerLatency yinfo = do
1552isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1536isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1553isBeyondMaxRate sv yinfo = do
1537isBeyondMaxRate sv yinfo = do
1554 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1538 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1555− now <- getTime Monotonic
1539+ now <- getTime
1556− let duration = fromInteger $ toNanoSecs $ now - tstamp
1540+ let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1557 let targetLat = svarLatencyTarget yinfo
1541 let targetLat = svarLatencyTarget yinfo
1558 gainLoss <- readIORef (svarGainedLostYields yinfo)
1542 gainLoss <- readIORef (svarGainedLostYields yinfo)
1559 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1543 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
@@ -1575,7 +1559,7 @@ isBeyondMaxRate sv yinfo = do
1575-- value is updated an existing worker must check it again on the next yield.
1559-- value is updated an existing worker must check it again on the next yield.
1576-- Otherwise it is possible that we may keep updating it and because of the mod
1560-- Otherwise it is possible that we may keep updating it and because of the mod
1577-- worker keeps skipping it.
1561-- worker keeps skipping it.
1578−updateWorkerPollingInterval :: YieldRateInfo -> NanoSecs -> IO ()
1562+updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
1579updateWorkerPollingInterval yinfo latency = do
1563updateWorkerPollingInterval yinfo latency = do
1580 let periodRef = workerPollingInterval yinfo
1564 let periodRef = workerPollingInterval yinfo
1581 cnt = max 1 $ minThreadDelay `div` latency
1565 cnt = max 1 $ minThreadDelay `div` latency
@@ -1589,7 +1573,7 @@ updateWorkerPollingInterval yinfo latency = do
1589-- whereas the average is used for future estimates e.g. how many workers
1573-- whereas the average is used for future estimates e.g. how many workers
1590-- should be maintained to maintain the rate.
1574-- should be maintained to maintain the rate.
1591-- CAUTION! keep it in sync with getWorkerLatency
1575-- CAUTION! keep it in sync with getWorkerLatency
1592−collectLatency :: SVar t m a -> YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1576+collectLatency :: SVar t m a -> YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1593collectLatency sv yinfo = do
1577collectLatency sv yinfo = do
1594 let cur = workerPendingLatency yinfo
1578 let cur = workerPendingLatency yinfo
1595 col = workerCollectedLatency yinfo
1579 col = workerCollectedLatency yinfo
@@ -1653,10 +1637,10 @@ dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1653dispatchWorkerPaced sv = do
1637dispatchWorkerPaced sv = do
1654 let yinfo = fromJust $ yieldRateInfo sv
1638 let yinfo = fromJust $ yieldRateInfo sv
1655 (svarYields, svarElapsed, wLatency) <- do
1639 (svarYields, svarElapsed, wLatency) <- do
1656− now <- liftIO $ getTime Monotonic
1640+ now <- liftIO $ getTime
1657 (yieldCount, baseTime, lat) <-
1641 (yieldCount, baseTime, lat) <-
1658 liftIO $ collectLatency sv yinfo
1642 liftIO $ collectLatency sv yinfo
1659− let elapsed = fromInteger $ toNanoSecs $ now - baseTime
1643+ let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1660 let latency =
1644 let latency =
1661 if lat == 0
1645 if lat == 0
1662 then
1646 then
@@ -1698,7 +1682,8 @@ dispatchWorkerPaced sv = do
1698 -- still have a Stop event waiting in the outputQueue.
1682 -- still have a Stop event waiting in the outputQueue.
1699 done <- allThreadsDone sv
1683 done <- allThreadsDone sv
1700 when done $ void $ do
1684 when done $ void $ do
1685+ let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1701− liftIO $ threadDelay $ nanoToMicroSecs s
1686+ liftIO $ threadDelay (fromIntegral us)
1702 dispatchWorker 1 sv
1687 dispatchWorker 1 sv
1703 return False
1688 return False
1704 PartialWorker yields -> do
1689 PartialWorker yields -> do
@@ -1974,7 +1959,7 @@ getYieldRateInfo st = do
1974 measured <- newIORef 0
1959 measured <- newIORef 0
1975 wcur <- newIORef (0,0)
1960 wcur <- newIORef (0,0)
1976 wcol <- newIORef (0,0)
1961 wcol <- newIORef (0,0)
1977− now <- getTime Monotonic
1962+ now <- getTime
1978 wlong <- newIORef (0,now)
1963 wlong <- newIORef (0,now)
1979 period <- newIORef 1
1964 period <- newIORef 1
1980 gainLoss <- newIORef (Count 0)
1965 gainLoss <- newIORef (Count 0)
@@ -1999,9 +1984,9 @@ newSVarStats = do
1999 maxOq <- newIORef 0
1984 maxOq <- newIORef 0
2000 maxHs <- newIORef 0
1985 maxHs <- newIORef 0
2001 maxWq <- newIORef 0
1986 maxWq <- newIORef 0
2002− avgLat <- newIORef (0, NanoSecs 0)
1987+ avgLat <- newIORef (0, NanoSecond64 0)
2003− maxLat <- newIORef (NanoSecs 0)
1988+ maxLat <- newIORef (NanoSecond64 0)
2004− minLat <- newIORef (NanoSecs 0)
1989+ minLat <- newIORef (NanoSecond64 0)
2005 stpTime <- newIORef Nothing
1990 stpTime <- newIORef Nothing
2006
1991
2007 return SVarStats
1992 return SVarStats
src/Streamly/Streams/SVar.hs+2−2
@@ -24,7 +24,7 @@ import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
24import Data.Maybe (isNothing)
24import Data.Maybe (isNothing)
25import Data.Semigroup ((<>))
25import Data.Semigroup ((<>))
26import System.IO (hPutStrLn, stderr)
26import System.IO (hPutStrLn, stderr)
27−import System.Clock (Clock(Monotonic), getTime)
27+import Streamly.Clock (getTime)
28import System.Mem (performMajorGC)
28import System.Mem (performMajorGC)
29
29
30import Streamly.SVar
30import Streamly.SVar
@@ -49,7 +49,7 @@ fromStreamVar sv = mkStream $ \st yld sng stp -> do
49
49
50 allDone stp = do
50 allDone stp = do
51 when (svarInspectMode sv) $ do
51 when (svarInspectMode sv) $ do
52− t <- liftIO $ getTime Monotonic
52+ t <- liftIO $ getTime
53 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
53 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
54 liftIO $ printSVar sv "SVar Done"
54 liftIO $ printSVar sv "SVar Done"
55 stp
55 stp
streamly.cabal+5−3
@@ -128,6 +128,7 @@ flag examples-sdl
128library
128library
129 hs-source-dirs: src
129 hs-source-dirs: src
130 other-modules: Streamly.SVar
130 other-modules: Streamly.SVar
131+ , Streamly.Clock
131
132
132 -- Base streams
133 -- Base streams
133 , Streamly.Streams.StreamK.Type
134 , Streamly.Streams.StreamK.Type
@@ -182,11 +183,11 @@ library
182 , deepseq >= 1.4.3 && < 1.5
183 , deepseq >= 1.4.3 && < 1.5
183 , containers >= 0.5 && < 0.7
184 , containers >= 0.5 && < 0.7
184 , heaps >= 0.3 && < 0.4
185 , heaps >= 0.3 && < 0.4
186+ , time >= 1.8 && < 1.10
185
187
186 -- concurrency
188 -- concurrency
187 , atomic-primops >= 0.8 && < 0.9
189 , atomic-primops >= 0.8 && < 0.9
188 , lockfree-queue >= 0.2.3 && < 0.3
190 , lockfree-queue >= 0.2.3 && < 0.3
189− , clock >= 0.7.1 && < 0.8
190
191
191 -- transfomers
192 -- transfomers
192 , exceptions >= 0.8 && < 0.11
193 , exceptions >= 0.8 && < 0.11
@@ -524,7 +525,7 @@ benchmark base
524 -- concurrency
525 -- concurrency
525 , atomic-primops >= 0.8 && < 0.9
526 , atomic-primops >= 0.8 && < 0.9
526 , lockfree-queue >= 0.2.3 && < 0.3
527 , lockfree-queue >= 0.2.3 && < 0.3
527− , clock >= 0.7.1 && < 0.8
528+ , time >= 1.8 && < 1.10
528
529
529 , exceptions >= 0.8 && < 0.11
530 , exceptions >= 0.8 && < 0.11
530 , monad-control >= 1.0 && < 2
531 , monad-control >= 1.0 && < 2
@@ -542,6 +543,7 @@ executable nano-bench
542 hs-source-dirs: benchmark, src
543 hs-source-dirs: benchmark, src
543 main-is: NanoBenchmarks.hs
544 main-is: NanoBenchmarks.hs
544 other-modules: Streamly.SVar
545 other-modules: Streamly.SVar
546+ , Streamly.Clock
545 , Streamly.Streams.StreamK.Type
547 , Streamly.Streams.StreamK.Type
546 , Streamly.Streams.StreamK
548 , Streamly.Streams.StreamK
547 , Streamly.Streams.StreamD.Type
549 , Streamly.Streams.StreamD.Type
@@ -562,7 +564,7 @@ executable nano-bench
562 -- concurrency
564 -- concurrency
563 , atomic-primops >= 0.8 && < 0.9
565 , atomic-primops >= 0.8 && < 0.9
564 , lockfree-queue >= 0.2.3 && < 0.3
566 , lockfree-queue >= 0.2.3 && < 0.3
565− , clock >= 0.7.1 && < 0.8
567+ , time >= 1.8 && < 1.10
566
568
567 , exceptions >= 0.8 && < 0.11
569 , exceptions >= 0.8 && < 0.11
568 , monad-control >= 1.0 && < 2
570 , monad-control >= 1.0 && < 2