@@ -89,10 +89,9 @@ where
90#include "inline.hs"
90#include "inline.hs"
92−import Control.Concurrent (threadDelay)
92+import Control.Concurrent (threadDelay, ThreadId)
93import Control.Monad (forever)
93import Control.Monad (forever)
94import Control.Monad.IO.Class (MonadIO(..))
94import Control.Monad.IO.Class (MonadIO(..))
95−import Data.Int (Int64)
96import Streamly.Internal.Data.Time.Units (toRelTime64, RelTime64)
95import Streamly.Internal.Data.Time.Units (toRelTime64, RelTime64)
97import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
96import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
98import Streamly.Internal.Data.Time.Units
97import Streamly.Internal.Data.Time.Units
@@ -326,16 +325,17 @@ numFromThen from next = enumerateFromStepNum from (next - from)
326------------------------------------------------------------------------------
325------------------------------------------------------------------------------
328{-# INLINE updateTimeVar #-}
327{-# INLINE updateTimeVar #-}
329−updateTimeVar :: Prim.IORef Int64 -> IO ()
328+updateTimeVar :: Clock -> Prim.IORef MicroSecond64 -> IO ()
330−updateTimeVar timeVar = do
329+updateTimeVar clock timeVar = do
331− MicroSecond64 t <- fromAbsTime <$> getTime Monotonic
330+ t <- fromAbsTime <$> getTime clock
332 Prim.modifyIORef' timeVar (const t)
331 Prim.modifyIORef' timeVar (const t)
334{-# INLINE updateWithDelay #-}
333{-# INLINE updateWithDelay #-}
335−updateWithDelay :: RealFrac a => a -> Prim.IORef Int64 -> IO ()
334+updateWithDelay :: RealFrac a =>
335+ Clock -> a -> Prim.IORef MicroSecond64 -> IO ()
336−updateWithDelay precision timeVar = do
336+updateWithDelay clock precision timeVar = do
337 threadDelay (delayTime precision)
337 threadDelay (delayTime precision)
338− updateTimeVar timeVar
338+ updateTimeVar clock timeVar
@@ -344,12 +344,36 @@ updateWithDelay precision timeVar = do
344 delayTime g
344 delayTime g
345 | g' >= fromIntegral (maxBound :: Int) = maxBound
345 | g' >= fromIntegral (maxBound :: Int) = maxBound
346 | g' < 1000 = 1000
346 | g' < 1000 = 1000
347− | otherwise = round g'
347+ | otherwise = round g'
351 g' = g * 10 ^ (6 :: Int)
351 g' = g * 10 ^ (6 :: Int)
353+-- XXX Move this to the time/clock module?
355+-- | @asyncClock g@ starts a clock thread that updates an IORef with current
356+-- time as a 64-bit value in microseconds, every 'g' seconds. The IORef can be
357+-- read asynchronously. The thread exits automatically when the reference to
358+-- the returned 'ThreadId' is lost.
360+-- Minimum granularity of clock update is 1 ms. Higher is better for
363+-- CAUTION! This is safe only on a 64-bit machine. On a 32-bit machine a 64-bit
364+-- 'Var' cannot be read consistently without a lock while another thread is
366+asyncClock :: Clock -> Double -> IO (ThreadId, Prim.IORef MicroSecond64)
367+asyncClock clock g = do
368+ timeVar <- Prim.newIORef undefined
369+ updateTimeVar clock timeVar
370+ tid <- forkManaged $ forever (updateWithDelay clock g timeVar)
371+ return (tid, timeVar)
373+{-# INLINE readClock #-}
374+readClock :: (ThreadId, Prim.IORef MicroSecond64) -> IO MicroSecond64
375+readClock (_, timeVar) = Prim.readIORef timeVar
353{-# INLINE_NORMAL times #-}
377{-# INLINE_NORMAL times #-}
354times :: MonadAsync m => Double -> Stream m (AbsTime, RelTime64)
378times :: MonadAsync m => Double -> Stream m (AbsTime, RelTime64)
355times g = Stream step Nothing
379times g = Stream step Nothing
@@ -358,22 +382,16 @@ times g = Stream step Nothing
359 {-# INLINE_LATE step #-}
383 {-# INLINE_LATE step #-}
360 step _ Nothing = do
384 step _ Nothing = do
361− -- XXX note that this is safe only on a 64-bit machine. On a 32-bit
385+ clock <- liftIO $ asyncClock Monotonic g
362− -- machine a 64-bit 'Var' cannot be read consistently without a lock
386+ a <- liftIO $ readClock clock
363− -- while another thread is writing to it.
364− timeVar <- liftIO $ Prim.newIORef (0 :: Int64)
365− liftIO $ updateTimeVar timeVar
366− tid <- forkManaged $ liftIO $ forever (updateWithDelay g timeVar)
367− a <- liftIO $ Prim.readIORef timeVar
368− return $ Skip $ Just (timeVar, tid, a)
387+ return $ Skip $ Just (clock, a)
370− step _ s@(Just (timeVar, _, t0)) = do
389+ step _ s@(Just (clock, t0)) = do
371− a <- liftIO $ Prim.readIORef timeVar
390+ a <- liftIO $ readClock clock
372 -- XXX we can perhaps use an AbsTime64 using a 64 bit Int for
391 -- XXX we can perhaps use an AbsTime64 using a 64 bit Int for
373 -- efficiency. or maybe we can use a representation using Double for
392 -- efficiency. or maybe we can use a representation using Double for
374 -- floating precision time
393 -- floating precision time
375− return $ Yield (toAbsTime (MicroSecond64 t0),
394+ return $ Yield (toAbsTime t0, toRelTime64 (a - t0)) s
376− (toRelTime64 (MicroSecond64 (a - t0)))) s
378-------------------------------------------------------------------------------
396-------------------------------------------------------------------------------
379-- From Generators
397-- From Generators