git.iko.soy/streamly
on master help sign in
Replace "clock" with time

Implements convenient conversions between units and uses different representations for Absolute and Relative time. We depend on "time" only yo get system time, which too we can replace in future.

commitd22a24b12f009a519fcde6ee09c9cdaabd63f7b0 authorHarendra Kumar <harendra.kumar@gmail.com> dateTue 8 Jan 2019, 20:07 (8 years ago) parent84d1b0a
4 files changed+474−68
compare with master
unifiedsplit
src/Streamly/Clock.hs+419 src/Streamly/SVar.hs+48−63 src/Streamly/Streams/SVar.hs+2−2 streamly.cabal+5−3
src/Streamly/Clock.hsnew file+419
@@ -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
100100
101101 , toStreamVar
102102 , SVarStats (..)
103− , NanoSecs (..)
104103 , dumpSVar
105104 )
106105where
@@ -133,27 +132,15 @@ import Data.Set (Set)
133132import GHC.Conc (ThreadId(..))
134133import GHC.Exts
135134import 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)
137138import System.IO (hPutStrLn, stderr)
138139import Text.Printf (printf)
139140
140141import qualified Data.Heap as H
141142import qualified Data.Set as S
142143
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−
157144newtype Count = Count Int64
158145 deriving ( Eq
159146 , Read
@@ -230,7 +217,7 @@ data WorkerInfo = WorkerInfo
230217 -- total number of yields by the worker till now
231218 , workerYieldCount :: IORef Count
232219 -- yieldCount at start, timestamp
233− , workerLatencyStart :: IORef (Count, TimeSpec)
220+ , workerLatencyStart :: IORef (Count, AbsTime)
234221 }
235222
236223
@@ -262,13 +249,13 @@ data Rate = Rate
262249 }
263250
264251data LatencyRange = LatencyRange
265− { minLatency :: NanoSecs
266− , maxLatency :: NanoSecs
252+ { minLatency :: NanoSecond64
253+ , maxLatency :: NanoSecond64
267254 } deriving Show
268255
269256-- Rate control.
270257data YieldRateInfo = YieldRateInfo
271− { svarLatencyTarget :: NanoSecs
258+ { svarLatencyTarget :: NanoSecond64
272259 , svarLatencyRange :: LatencyRange
273260 , svarRateBuffer :: Int
274261 , svarGainedLostYields :: IORef Count
@@ -279,11 +266,11 @@ data YieldRateInfo = YieldRateInfo
279266 -- rate. The idle time of workers is adjusted in this, so that we only
280267 -- account for the rate when the consumer actually demands data.
281268 -- XXX interval latency is enough, we can move this under diagnostics build
282− , svarAllTimeLatency :: IORef (Count, TimeSpec)
269+ , svarAllTimeLatency :: IORef (Count, AbsTime)
283270
284271 -- XXX Worker latency specified by the user to be used before the first
285272 -- actual measurement arrives. Not yet implemented
286− , workerBootstrapLatency :: Maybe NanoSecs
273+ , workerBootstrapLatency :: Maybe NanoSecond64
287274
288275 -- After how many yields the worker should update the latency information.
289276 -- If the latency is high, this count is kept lower and vice-versa. XXX If
@@ -297,7 +284,7 @@ data YieldRateInfo = YieldRateInfo
297284 -- empty into workerCollectedLatency stats at certain intervals - whenever
298285 -- we process the stream elements yielded in this period.
299286 -- (yieldCount, timeTaken)
300− , workerPendingLatency :: IORef (Count, NanoSecs)
287+ , workerPendingLatency :: IORef (Count, NanoSecond64)
301288
302289 -- This is the second level stat which is an accmulation from
303290 -- workerPendingLatency stats. We keep accumulating latencies in this
@@ -305,10 +292,10 @@ data YieldRateInfo = YieldRateInfo
305292 -- to start collecting for the next period and retain the computed average
306293 -- latency for the last period in workerMeasuredLatency.
307294 -- (yieldCount, timeTaken)
308− , workerCollectedLatency :: IORef (Count, NanoSecs)
295+ , workerCollectedLatency :: IORef (Count, NanoSecond64)
309296
310297 -- Latency as measured by workers, aggregated for the last period.
311− , workerMeasuredLatency :: IORef NanoSecs
298+ , workerMeasuredLatency :: IORef NanoSecond64
312299 }
313300
314301data SVarStats = SVarStats {
@@ -317,10 +304,10 @@ data SVarStats = SVarStats {
317304 , maxOutQSize :: IORef Int
318305 , maxHeapSize :: IORef Int
319306 , maxWorkQSize :: IORef Int
320− , avgWorkerLatency :: IORef (Count, NanoSecs)
321− , minWorkerLatency :: IORef NanoSecs
322− , maxWorkerLatency :: IORef NanoSecs
323− , svarStopTime :: IORef (Maybe TimeSpec)
307+ , avgWorkerLatency :: IORef (Count, NanoSecond64)
308+ , minWorkerLatency :: IORef NanoSecond64
309+ , maxWorkerLatency :: IORef NanoSecond64
310+ , svarStopTime :: IORef (Maybe AbsTime)
324311}
325312
326313data Limit = Unlimited | Limited Word deriving Show
@@ -393,7 +380,7 @@ data State t m a = State
393380 , _threadsHigh :: Limit
394381 , _bufferHigh :: Limit
395382 -- XXX these two can be collapsed into a single type
396− , _streamLatency :: Maybe NanoSecs -- bootstrap latency
383+ , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
397384 , _maxStreamRate :: Maybe Rate
398385 , _inspectMode :: Bool
399386 }
@@ -505,7 +492,7 @@ setStreamLatency n st =
505492 else Just (fromIntegral n)
506493 }
507494
508−getStreamLatency :: State t m a -> Maybe NanoSecs
495+getStreamLatency :: State t m a -> Maybe NanoSecond64
509496getStreamLatency = _streamLatency
510497
511498setInspectMode :: State t m a -> State t m a
@@ -561,7 +548,7 @@ secs k
561548 | otherwise = printf "%.3f %s" t u
562549
563550-- 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)
565552drainLatency sv yinfo = do
566553 let cur = workerPendingLatency yinfo
567554 col = workerCollectedLatency yinfo
@@ -624,11 +611,11 @@ dumpSVarStats sv ss style = do
624611 gl <- readIORef (svarGainedLostYields yinfo)
625612 case t of
626613 Nothing -> do
627− now <- getTime Monotonic
628− let interval = toNanoSecs (now - startTime)
614+ now <- getTime
615+ let interval = diffAbsTime64 now startTime
629616 return (cnt, gl, interval `div` fromIntegral cnt)
630617 Just stopTime -> do
631− let interval = toNanoSecs (stopTime - startTime)
618+ let interval = diffAbsTime64 stopTime startTime
632619 return (cnt, gl, interval `div` fromIntegral cnt)
633620 else return (0, 0, 0)
634621
@@ -898,7 +885,7 @@ send sv msg = do
898885 active <- readIORef (workerCount sv)
899886 return $ len < (fromIntegral lim - active)
900887
901−workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecs))
888+workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
902889workerCollectLatency winfo = do
903890 (cnt0, t0) <- readIORef (workerLatencyStart winfo)
904891 cnt1 <- readIORef (workerYieldCount winfo)
@@ -906,8 +893,8 @@ workerCollectLatency winfo = do
906893
907894 if cnt > 0
908895 then do
909− t1 <- getTime Monotonic
910− let period = fromInteger $ toNanoSecs (t1 - t0)
896+ t1 <- getTime
897+ let period = fromRelTime64 $ diffAbsTime64 t1 t0
911898 writeIORef (workerLatencyStart winfo) (cnt1, t1)
912899 return $ Just (cnt, period)
913900 else return Nothing
@@ -1305,7 +1292,7 @@ pushWorker yieldMax sv = do
13051292 Nothing -> return Nothing
13061293 Just _ -> liftIO $ do
13071294 cntRef <- newIORef 0
1308− t <- getTime Monotonic
1295+ t <- getTime
13091296 lat <- newIORef (0, t)
13101297 return $ Just WorkerInfo
13111298 { workerYieldMax = yieldMax
@@ -1348,7 +1335,7 @@ pushWorkerPar sv wloop =
13481335 Nothing -> return Nothing
13491336 Just _ -> liftIO $ do
13501337 cntRef <- newIORef 0
1351− t <- getTime Monotonic
1338+ t <- getTime
13521339 lat <- newIORef (0, t)
13531340 return $ Just WorkerInfo
13541341 { workerYieldMax = 0
@@ -1420,24 +1407,21 @@ dispatchWorker yieldCount sv = do
14201407-- sleep time is accumulated.
14211408-- 3. Collected latencies are computed and transferred to measured latency
14221409-- after a minimum of this period.
1423−minThreadDelay :: NanoSecs
1424−minThreadDelay = 10^(6 :: Int)
1410+minThreadDelay :: NanoSecond64
1411+minThreadDelay = 1000000
14251412
14261413-- | Another magic number! When we have to start more workers to cover up a
14271414-- number of yields that we are lagging by then we cannot start one worker for
14281415-- each yield because that may be a very big number and if the latency of the
14291416-- workers is low these number of yields could be very high. We assume that we
14301417-- run each extra worker for at least this much time.
1431−rateRecoveryTime :: NanoSecs
1418+rateRecoveryTime :: NanoSecond64
14321419rateRecoveryTime = 1000000
14331420
1434−nanoToMicroSecs :: NanoSecs -> Int
1435−nanoToMicroSecs s = fromIntegral s `div` 1000
1436−
14371421-- We either block, or send one worker with limited yield count or one or more
14381422-- workers with unlimited yield count.
14391423data Work
1440− = BlockWait NanoSecs
1424+ = BlockWait NanoSecond64
14411425 | PartialWorker Count
14421426 | ManyWorkers Int Count
14431427 deriving Show
@@ -1447,9 +1431,9 @@ estimateWorkers
14471431 :: Limit
14481432 -> Count
14491433 -> Count
1450− -> NanoSecs
1451− -> NanoSecs
1452− -> NanoSecs
1434+ -> NanoSecond64
1435+ -> NanoSecond64
1436+ -> NanoSecond64
14531437 -> LatencyRange
14541438 -> Work
14551439estimateWorkers workerLimit svarYields gainLossYields
@@ -1500,7 +1484,7 @@ estimateWorkers workerLimit svarYields gainLossYields
15001484 fromIntegral rateRecoveryTime
15011485 yieldsFreq = 1.0 / fromIntegral targetLat
15021486 totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1503− requiredLat = NanoSecs $ round $ 1.0 / totalYieldsFreq
1487+ requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
15041488 adjustedLat = min (max requiredLat (minLatency range))
15051489 (maxLatency range)
15061490 in assert (adjustedLat > 0) $
@@ -1527,7 +1511,7 @@ estimateWorkers workerLimit svarYields gainLossYields
15271511-- | Get the worker latency without resetting workerPendingLatency
15281512-- Returns (total yield count, base time, measured latency)
15291513-- CAUTION! keep it in sync with collectLatency
1530−getWorkerLatency :: YieldRateInfo -> IO (Count, TimeSpec, NanoSecs)
1514+getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
15311515getWorkerLatency yinfo = do
15321516 let cur = workerPendingLatency yinfo
15331517 col = workerCollectedLatency yinfo
@@ -1552,8 +1536,8 @@ getWorkerLatency yinfo = do
15521536isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
15531537isBeyondMaxRate sv yinfo = do
15541538 (count, tstamp, wLatency) <- getWorkerLatency yinfo
1555− now <- getTime Monotonic
1556− let duration = fromInteger $ toNanoSecs $ now - tstamp
1539+ now <- getTime
1540+ let duration = fromRelTime64 $ diffAbsTime64 now tstamp
15571541 let targetLat = svarLatencyTarget yinfo
15581542 gainLoss <- readIORef (svarGainedLostYields yinfo)
15591543 let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
@@ -1575,7 +1559,7 @@ isBeyondMaxRate sv yinfo = do
15751559-- value is updated an existing worker must check it again on the next yield.
15761560-- Otherwise it is possible that we may keep updating it and because of the mod
15771561-- worker keeps skipping it.
1578−updateWorkerPollingInterval :: YieldRateInfo -> NanoSecs -> IO ()
1562+updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
15791563updateWorkerPollingInterval yinfo latency = do
15801564 let periodRef = workerPollingInterval yinfo
15811565 cnt = max 1 $ minThreadDelay `div` latency
@@ -1589,7 +1573,7 @@ updateWorkerPollingInterval yinfo latency = do
15891573-- whereas the average is used for future estimates e.g. how many workers
15901574-- should be maintained to maintain the rate.
15911575-- 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)
15931577collectLatency sv yinfo = do
15941578 let cur = workerPendingLatency yinfo
15951579 col = workerCollectedLatency yinfo
@@ -1653,10 +1637,10 @@ dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
16531637dispatchWorkerPaced sv = do
16541638 let yinfo = fromJust $ yieldRateInfo sv
16551639 (svarYields, svarElapsed, wLatency) <- do
1656− now <- liftIO $ getTime Monotonic
1640+ now <- liftIO $ getTime
16571641 (yieldCount, baseTime, lat) <-
16581642 liftIO $ collectLatency sv yinfo
1659− let elapsed = fromInteger $ toNanoSecs $ now - baseTime
1643+ let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
16601644 let latency =
16611645 if lat == 0
16621646 then
@@ -1698,7 +1682,8 @@ dispatchWorkerPaced sv = do
16981682 -- still have a Stop event waiting in the outputQueue.
16991683 done <- allThreadsDone sv
17001684 when done $ void $ do
1701− liftIO $ threadDelay $ nanoToMicroSecs s
1685+ let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1686+ liftIO $ threadDelay (fromIntegral us)
17021687 dispatchWorker 1 sv
17031688 return False
17041689 PartialWorker yields -> do
@@ -1974,7 +1959,7 @@ getYieldRateInfo st = do
19741959 measured <- newIORef 0
19751960 wcur <- newIORef (0,0)
19761961 wcol <- newIORef (0,0)
1977− now <- getTime Monotonic
1962+ now <- getTime
19781963 wlong <- newIORef (0,now)
19791964 period <- newIORef 1
19801965 gainLoss <- newIORef (Count 0)
@@ -1999,9 +1984,9 @@ newSVarStats = do
19991984 maxOq <- newIORef 0
20001985 maxHs <- newIORef 0
20011986 maxWq <- newIORef 0
2002− avgLat <- newIORef (0, NanoSecs 0)
2003− maxLat <- newIORef (NanoSecs 0)
2004− minLat <- newIORef (NanoSecs 0)
1987+ avgLat <- newIORef (0, NanoSecond64 0)
1988+ maxLat <- newIORef (NanoSecond64 0)
1989+ minLat <- newIORef (NanoSecond64 0)
20051990 stpTime <- newIORef Nothing
20061991
20071992 return SVarStats
src/Streamly/Streams/SVar.hs+2−2
@@ -24,7 +24,7 @@ import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
2424import Data.Maybe (isNothing)
2525import Data.Semigroup ((<>))
2626import System.IO (hPutStrLn, stderr)
27−import System.Clock (Clock(Monotonic), getTime)
27+import Streamly.Clock (getTime)
2828import System.Mem (performMajorGC)
2929
3030import Streamly.SVar
@@ -49,7 +49,7 @@ fromStreamVar sv = mkStream $ \st yld sng stp -> do
4949
5050 allDone stp = do
5151 when (svarInspectMode sv) $ do
52− t <- liftIO $ getTime Monotonic
52+ t <- liftIO $ getTime
5353 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
5454 liftIO $ printSVar sv "SVar Done"
5555 stp
streamly.cabal+5−3
@@ -128,6 +128,7 @@ flag examples-sdl
128128library
129129 hs-source-dirs: src
130130 other-modules: Streamly.SVar
131+ , Streamly.Clock
131132
132133 -- Base streams
133134 , Streamly.Streams.StreamK.Type
@@ -182,11 +183,11 @@ library
182183 , deepseq >= 1.4.3 && < 1.5
183184 , containers >= 0.5 && < 0.7
184185 , heaps >= 0.3 && < 0.4
186+ , time >= 1.8 && < 1.10
185187
186188 -- concurrency
187189 , atomic-primops >= 0.8 && < 0.9
188190 , lockfree-queue >= 0.2.3 && < 0.3
189− , clock >= 0.7.1 && < 0.8
190191
191192 -- transfomers
192193 , exceptions >= 0.8 && < 0.11
@@ -524,7 +525,7 @@ benchmark base
524525 -- concurrency
525526 , atomic-primops >= 0.8 && < 0.9
526527 , lockfree-queue >= 0.2.3 && < 0.3
527− , clock >= 0.7.1 && < 0.8
528+ , time >= 1.8 && < 1.10
528529
529530 , exceptions >= 0.8 && < 0.11
530531 , monad-control >= 1.0 && < 2
@@ -542,6 +543,7 @@ executable nano-bench
542543 hs-source-dirs: benchmark, src
543544 main-is: NanoBenchmarks.hs
544545 other-modules: Streamly.SVar
546+ , Streamly.Clock
545547 , Streamly.Streams.StreamK.Type
546548 , Streamly.Streams.StreamK
547549 , Streamly.Streams.StreamD.Type
@@ -562,7 +564,7 @@ executable nano-bench
562564 -- concurrency
563565 , atomic-primops >= 0.8 && < 0.9
564566 , lockfree-queue >= 0.2.3 && < 0.3
565− , clock >= 0.7.1 && < 0.8
567+ , time >= 1.8 && < 1.10
566568
567569 , exceptions >= 0.8 && < 0.11
568570 , monad-control >= 1.0 && < 2