@@ -15,18 +15,28 @@ module Streamly.Internal.Data.Time.Clock
1515 -- * Async clock
1616 , asyncClock
1717 , readClock
18+
19+ -- * Adjustable Timer
20+ , Timer
21+ , timer
22+ , resetTimer
23+ , extendTimer
24+ , shortenTimer
25+ , readTimer
26+ , waitTimer
1827 )
1928where
2029
2130import Control.Concurrent (threadDelay, ThreadId)
22−import Control.Monad (forever)
31+import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, tryPutMVar)
32+import Control.Monad (forever, when, void)
2333import Streamly.Internal.Data.Time.Clock.Type (Clock(..), getTime)
24−import Streamly.Internal.Data.Time.Units (MicroSecond64(..), fromAbsTime)
34+import Streamly.Internal.Data.Time.Units
35+ (MicroSecond64(..), fromAbsTime, addToAbsTime, toRelTime)
2536import Streamly.Internal.Control.Concurrent (forkManaged)
2637
2738import qualified Streamly.Internal.Data.IORef.Prim as Prim
2839
29−
3040------------------------------------------------------------------------------
3141-- Async clock
3242------------------------------------------------------------------------------
@@ -78,3 +88,93 @@ asyncClock clock g = do
7888{-# INLINE readClock #-}
7989readClock :: (ThreadId, Prim.IORef MicroSecond64) -> IO MicroSecond64
8090readClock (_, timeVar) = Prim.readIORef timeVar
91+
92+------------------------------------------------------------------------------
93+-- Adjustable Timer
94+------------------------------------------------------------------------------
95+
96+-- | Adjustable periodic timer.
97+data Timer = Timer ThreadId (MVar ()) (IO ())
98+
99+-- Set the expiry to current time + timer period
100+{-# INLINE resetTimerExpiry #-}
101+resetTimerExpiry :: Clock -> MicroSecond64 -> Prim.IORef MicroSecond64 -> IO ()
102+resetTimerExpiry clock period timeVar = do
103+ t <- getTime clock
104+ let t1 = addToAbsTime t (toRelTime period)
105+ Prim.modifyIORef' timeVar (const (fromAbsTime t1))
106+
107+{-# INLINE processTimerTick #-}
108+processTimerTick :: RealFrac a =>
109+ Clock -> a -> Prim.IORef MicroSecond64 -> MVar () -> IO () -> IO ()
110+processTimerTick 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
119+
120+ where
121+
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'
128+
129+ where
130+
131+ g' = g * 10 ^ (6 :: Int)
132+
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.
139+timer :: Clock -> Double -> Double -> IO Timer
140+timer 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
150+
151+-- | Blocking wait for a timer tick.
152+{-# INLINE waitTimer #-}
153+waitTimer :: Timer -> IO ()
154+waitTimer (Timer _ mvar _) = takeMVar mvar
155+
156+-- | Resets the current period.
157+{-# INLINE resetTimer #-}
158+resetTimer :: Timer -> IO ()
159+resetTimer (Timer _ _ reset) = reset
160+
161+-- | Elongates the current period by specified amount.
162+--
163+-- /Unimplemented/
164+{-# INLINE extendTimer #-}
165+extendTimer :: Timer -> Double -> IO ()
166+extendTimer = undefined
167+
168+-- | Shortens the current period by specified amount.
169+--
170+-- /Unimplemented/
171+{-# INLINE shortenTimer #-}
172+shortenTimer :: Timer -> Double -> IO ()
173+shortenTimer = undefined
174+
175+-- | Show the remaining time in the current time period.
176+--
177+-- /Unimplemented/
178+{-# INLINE readTimer #-}
179+readTimer :: Timer -> IO Double
180+readTimer = undefined