git.iko.soy/streamly
on cc44d97 help sign in
streamly/src/Streamly/Internal/Data/Time/Clock.hs
180 lines 5.7 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Time.Clock
3-- Copyright : (c) 2021 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : pre-release
7-- Portability : GHC
9module Streamly.Internal.Data.Time.Clock
10 (
11 -- * System clock
12 Clock(..)
13 , getTime
15 -- * Async clock
16 , asyncClock
17 , readClock
19 -- * Adjustable Timer
20 , Timer
21 , timer
22 , resetTimer
23 , extendTimer
24 , shortenTimer
25 , readTimer
26 , waitTimer
27 )
28where
30import Control.Concurrent (threadDelay, ThreadId)
31import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, tryPutMVar)
32import Control.Monad (forever, when, void)
33import Streamly.Internal.Data.Time.Clock.Type (Clock(..), getTime)
34import Streamly.Internal.Data.Time.Units
35 (MicroSecond64(..), fromAbsTime, addToAbsTime, toRelTime)
36import Streamly.Internal.Control.Concurrent (forkManaged)
38import qualified Streamly.Internal.Data.IORef.Prim as Prim
40------------------------------------------------------------------------------
41-- Async clock
42------------------------------------------------------------------------------
44{-# INLINE updateTimeVar #-}
45updateTimeVar :: Clock -> Prim.IORef MicroSecond64 -> IO ()
46updateTimeVar clock timeVar = do
47 t <- fromAbsTime <$> getTime clock
48 Prim.modifyIORef' timeVar (const t)
50{-# INLINE updateWithDelay #-}
51updateWithDelay :: RealFrac a =>
52 Clock -> a -> Prim.IORef MicroSecond64 -> IO ()
53updateWithDelay clock precision timeVar = do
54 threadDelay (delayTime precision)
55 updateTimeVar clock timeVar
57 where
59 -- Keep the minimum at least a millisecond to avoid high CPU usage
60 {-# INLINE delayTime #-}
61 delayTime g
62 | g' >= fromIntegral (maxBound :: Int) = maxBound
63 | g' < 1000 = 1000
64 | otherwise = round g'
66 where
68 g' = g * 10 ^ (6 :: Int)
70-- | @asyncClock g@ starts a clock thread that updates an IORef with current
71-- time as a 64-bit value in microseconds, every 'g' seconds. The IORef can be
72-- read asynchronously. The thread exits automatically when the reference to
73-- the returned 'ThreadId' is lost.
74--
75-- Minimum granularity of clock update is 1 ms. Higher is better for
76-- performance.
77--
78-- CAUTION! This is safe only on a 64-bit machine. On a 32-bit machine a 64-bit
79-- 'Var' cannot be read consistently without a lock while another thread is
80-- writing to it.
81asyncClock :: Clock -> Double -> IO (ThreadId, Prim.IORef MicroSecond64)
82asyncClock clock g = do
83 timeVar <- Prim.newIORef 0
84 updateTimeVar clock timeVar
85 tid <- forkManaged $ forever (updateWithDelay clock g timeVar)
86 return (tid, timeVar)
88{-# INLINE readClock #-}
89readClock :: (ThreadId, Prim.IORef MicroSecond64) -> IO MicroSecond64
90readClock (_, timeVar) = Prim.readIORef timeVar
92------------------------------------------------------------------------------
93-- Adjustable Timer
94------------------------------------------------------------------------------
96-- | Adjustable periodic timer.
97data Timer = Timer ThreadId (MVar ()) (IO ())
99-- Set the expiry to current time + timer period
100{-# INLINE resetTimerExpiry #-}
101resetTimerExpiry :: Clock -> MicroSecond64 -> Prim.IORef MicroSecond64 -> IO ()
102resetTimerExpiry clock period timeVar = do
103 t <- getTime clock
104 let t1 = addToAbsTime t (toRelTime period)
105 Prim.modifyIORef' timeVar (const (fromAbsTime t1))
107{-# INLINE processTimerTick #-}
108processTimerTick :: RealFrac a =>
109 Clock -> a -> Prim.IORef MicroSecond64 -> MVar () -> IO () -> IO ()
110processTimerTick clock precision timeVar mvar reset = do
111 threadDelay (delayTime precision)
112 t <- fromAbsTime <$> getTime clock
113 expiry <- Prim.readIORef timeVar
114 when (t >= expiry) $ do
115 -- non-blocking put so that we can process multiple timers in a
116 -- non-blocking manner in future.
117 void $ tryPutMVar mvar ()
118 reset
120 where
122 -- Keep the minimum at least a millisecond to avoid high CPU usage
123 {-# INLINE delayTime #-}
124 delayTime g
125 | g' >= fromIntegral (maxBound :: Int) = maxBound
126 | g' < 1000 = 1000
127 | otherwise = round g'
129 where
131 g' = g * 10 ^ (6 :: Int)
133-- XXX In future we can add a timer in a heap of timers.
134--
135-- | @timer clockType granularity period@ creates a timer. The timer produces
136-- timer ticks at specified time intervals that can be waited upon using
137-- 'waitTimer'. If the previous tick is not yet processed, the new tick is
138-- lost.
139timer :: Clock -> Double -> Double -> IO Timer
140timer clock g period = do
141 mvar <- newEmptyMVar
142 timeVar <- Prim.newIORef 0
143 let p = round (period * 1e6) :: Int
144 p1 = fromIntegral p :: MicroSecond64
145 reset = resetTimerExpiry clock p1 timeVar
146 process = processTimerTick clock g timeVar mvar reset
147 reset
148 tid <- forkManaged $ forever process
149 return $ Timer tid mvar reset
151-- | Blocking wait for a timer tick.
152{-# INLINE waitTimer #-}
153waitTimer :: Timer -> IO ()
154waitTimer (Timer _ mvar _) = takeMVar mvar
156-- | Resets the current period.
157{-# INLINE resetTimer #-}
158resetTimer :: Timer -> IO ()
159resetTimer (Timer _ _ reset) = reset
161-- | Elongates the current period by specified amount.
162--
163-- /Unimplemented/
164{-# INLINE extendTimer #-}
165extendTimer :: Timer -> Double -> IO ()
166extendTimer = undefined
168-- | Shortens the current period by specified amount.
169--
170-- /Unimplemented/
171{-# INLINE shortenTimer #-}
172shortenTimer :: Timer -> Double -> IO ()
173shortenTimer = undefined
175-- | Show the remaining time in the current time period.
176--
177-- /Unimplemented/
178{-# INLINE readTimer #-}
179readTimer :: Timer -> IO Double
180readTimer = undefined