packages feed

timer-wheel 0.3.0 → 0.4.0

raw patch · 17 files changed

+549/−588 lines, 17 filesdep +arraydep +kidep −vectordep ~base

Dependencies added: array, ki

Dependencies removed: vector

Dependency ranges changed: base

Files

CHANGELOG.md view
@@ -1,24 +1,20 @@-# Changelog--All notable changes to this project will be documented in this file.+## [0.4.0] - 2022-11-05 -The format is based on [Keep a Changelog](http://keepachangelog.com/)-and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/).+- Add `create`+- Rename `Data.TimerWheel` to `TimerWheel`+- Swap out `vector` for `array`+- Treat negative delays as 0+- Drop support for GHC < 8.6  ## [0.3.0] - 2020-06-18 -### Added - Add `with` - Add support for GHC 8.8, GHC 8.10--### Changed - Change type of `spokes` from `Natural` to `Int` - Change order of delay and action arguments in `register`, `register_`, `recurring`, and `recurring_` - Simplify `cancel` to return `True` at most once - Throw an error if a negative delay is provided to `register`, `register_`, `recurring`, or `recurring_` - Fix underflow bug that affected recurring timers--### Removed - Remove `create`, `destroy` - Remove `TimerWheelDied` exception. `with` now simply re-throws the exception that the timer wheel thread throws - Remove `InvalidTimerWheelConfig` exception. `error` is used instead@@ -26,16 +22,12 @@  ## [0.2.0.1] - 2019-05-19 -### Changed - Swap out `ghc-prim` and `primitive` for `vector`  ## [0.2.0] - 2019-02-03 -### Added - Add `destroy` function, for reaping the background thread - Add `recurring_` function--### Changed - If the timer wheel reaper thread crashes, it will propagate the exception to the thread that spawned it - `new` may now throw `InvalidTimerWheelConfig`@@ -49,5 +41,4 @@  ## [0.1.0] - 2018-07-18 -### Added - Initial release
− src/Data/TimerWheel.hs 
@@ -1,263 +0,0 @@-{-# LANGUAGE RecursiveDo #-}-{-# LANGUAGE StrictData #-}-{-# OPTIONS_GHC -funbox-strict-fields #-}---- | A simple, hashed timer wheel.-module Data.TimerWheel-  ( -- * Timer wheel-    TimerWheel,-    with,-    Config (..),-    register,-    register_,-    recurring,-    recurring_,-  )-where--import Control.Concurrent-import Control.Exception-import Control.Monad (join, void)-import Data.Bool (bool)-import Data.Fixed (E6, Fixed)-import Data.Function (fix)-import Data.IORef (newIORef, readIORef, writeIORef)-import Data.TimerWheel.Internal.Config (Config)-import qualified Data.TimerWheel.Internal.Config as Config-import Data.TimerWheel.Internal.Micros (Micros (Micros))-import qualified Data.TimerWheel.Internal.Micros as Micros-import Data.TimerWheel.Internal.Supply (Supply)-import qualified Data.TimerWheel.Internal.Supply as Supply-import Data.TimerWheel.Internal.Wheel (Wheel)-import qualified Data.TimerWheel.Internal.Wheel as Wheel---- | A timer wheel is a vector-of-collections-of timers to fire. It is configured with a /spoke count/ and /resolution/.--- Timers may be scheduled arbitrarily far in the future. A timeout thread is spawned to step through the timer wheel--- and fire expired timers at regular intervals.------ * The /spoke count/ determines the size of the timer vector.------     * A __larger spoke count__ will result in __less insert contention__ at each spoke and will require---       __more memory__ to store the timer wheel.------     * A __smaller spoke count__ will result in __more insert contention__ at each spoke and will require---       __less memory__ to store the timer wheel.------ * The /resolution/ determines both the duration of time that each spoke corresponds to, and how often the timeout---   thread wakes. For example, with a resolution of __@1s@__, a timer that expires at __@2.5s@__ will not fire until---   the timeout thread wakes at __@3s@__.------     * A __larger resolution__ will result in __more insert contention__ at each spoke, __less accurate__ timers, and---       will require __fewer wakeups__ by the timeout thread.------     * A __smaller resolution__ will result in __less insert contention__ at each spoke, __more accurate__ timers, and---       will require __more wakeups__ by the timeout thread.------ * The timeout thread has some important properties:------     * There is only one, and it fires expired timers synchronously. If your timer actions execute quicky, 'register'---       them directly. Otherwise, consider registering an action that enqueues the /real/ action to be performed on a---       job queue.------     * Synchronous exceptions thrown by enqueued @IO@ actions will bring the thread down, and the exception will be---       propagated to the thread that created the timer wheel. If you want to catch exceptions and log them, for---       example, you will have to bake this into the registered actions yourself.------ As an example, below is a depiction of a timer wheel with @6@ timers inserted across @8@ spokes, and a resolution of--- @.1s@. It depicts a cursor at @.3s@, which indicates where the timeout thread currently is.------ @---  0       .1      .2      .3      .4      .5      .6      .7--- ┌───────┬───────┬───────┬───────┬───────┬───────┬───────┬───────┐--- │       │ A⁰    │       │ B¹ C⁰ │ D⁰    │       │       │ E² F⁰ │--- └───────┴───────┴───────┴───────┴───────┴───────┴───────┴───────┘---                           ↑--- @------ After @.1s@, the timeout thread will advance to the next spoke and process all of the timers it passed over. In--- this case, __C__ will fire, and __B__ will be put back with its count decremented to @0@. This is how the timer wheel--- can schedule a timer to fire arbitrarily far in the future: its count is simply the number of times its delay wraps--- the entire duration of the timer wheel.------ @---  0       .1      .2      .3      .4      .5      .6      .7--- ┌───────┬───────┬───────┬───────┬───────┬───────┬───────┬───────┐--- │       │ A⁰    │       │ B⁰    │ D⁰    │       │       │ E² F⁰ │--- └───────┴───────┴───────┴───────┴───────┴───────┴───────┴───────┘---                                   ↑--- @-data TimerWheel = TimerWheel-  { -- | A supply of unique ints.-    supply :: Supply,-    -- | The array of collections of timers.-    wheel :: Wheel,-    thread :: ThreadId-  }---- | The timeout thread died.-newtype TimerWheelDied-  = TimerWheelDied SomeException-  deriving stock (Show)--instance Exception TimerWheelDied where-  toException = asyncExceptionToException-  fromException = asyncExceptionFromException---- | Perform an action with a timer wheel.------ /Throws./------   * Calls 'error' if the config is invalid---   * Throws the exception the given action throws, if any---   * Throws the exception the timer wheel thread throws, if any-with :: Config -> (TimerWheel -> IO a) -> IO a-with config action =-  case validateConfig config of-    () -> _with config action--_with :: Config -> (TimerWheel -> IO a) -> IO a-_with config action = do-  wheel <- Wheel.create (Config.spokes config) (Micros.fromFixed (Config.resolution config))-  supply <- Supply.new-  parentThread <- myThreadId-  uninterruptibleMask \restore -> do-    thread <--      forkIOWithUnmask $ \unmask ->-        unmask (Wheel.reap wheel) `catch` \e ->-          case fromException e of-            Just ThreadKilled -> pure ()-            _ -> throwTo parentThread (TimerWheelDied e)-    let cleanup = killThread thread-    let handler :: SomeException -> IO void-        handler ex = do-          cleanup-          case fromException ex of-            Just (TimerWheelDied ex') -> throwIO ex'-            _ -> throwIO ex-    result <- restore (action TimerWheel {supply, wheel, thread}) `catch` handler-    cleanup-    pure result--validateConfig :: Config -> ()-validateConfig config-  | invalid = error ("[timer-wheel] invalid config: " ++ show config)-  | otherwise = ()-  where-    invalid :: Bool-    invalid =-      Config.spokes config <= 0 || Config.resolution config <= 0---- | @register wheel delay action@ registers an action __@action@__ in timer wheel __@wheel@__ to fire after __@delay@__--- seconds.------ Returns an action that, when called, attempts to cancel the timer, and returns whether or not it was successful--- (@False@ means the timer has already fired, or was already cancelled).------ /Throws/.------   * Calls 'error' if the given number of seconds is negative.-register ::-  -- |-  TimerWheel ->-  -- | Delay, in seconds-  Fixed E6 ->-  -- | Action-  IO () ->-  IO (IO Bool)-register wheel (Micros.fromSeconds -> delay) =-  _register wheel delay---- | Like 'register', but for when you don't intend to cancel the timer.------ /Throws/.------   * Calls 'error' if the given number of seconds is negative.-register_ ::-  -- |-  TimerWheel ->-  -- | Delay, in seconds-  Fixed E6 ->-  -- | Action-  IO () ->-  IO ()-register_ wheel delay action =-  void (register wheel delay action)--_register :: TimerWheel -> Micros -> IO () -> IO (IO Bool)-_register TimerWheel {supply, wheel} delay action = do-  key <- Supply.next supply-  Wheel.insert wheel key delay action---- | @recurring wheel action delay@ registers an action __@action@__ in timer wheel __@wheel@__ to fire every--- __@delay@__ seconds (or every /resolution/ seconds, whichever is smaller).------ Returns an action that, when called, cancels the recurring timer.------ /Throws/.------   * Calls 'error' if the given number of seconds is negative.-recurring ::-  TimerWheel ->-  -- | Delay, in seconds-  Fixed E6 ->-  -- | Action-  IO () ->-  IO (IO ())-recurring wheel (Micros.fromSeconds -> delay) action = mdo-  let doAction :: IO ()-      doAction = do-        writeIORef cancelRef =<< _reregister wheel delay doAction-        action-  cancel <- _register wheel delay doAction-  cancelRef <- newIORef cancel-  pure (untilTrue (join (readIORef cancelRef)))-  where-    -- Repeat an IO action until it returns 'True'.-    untilTrue :: IO Bool -> IO ()-    untilTrue m =-      fix \again ->-        m >>= bool again (pure ())---- | Like 'recurring', but for when you don't intend to cancel the timer.------ /Throws/.------   * Calls 'error' if the given number of seconds is negative.-recurring_ ::-  TimerWheel ->-  -- | Delay, in seconds-  Fixed E6 ->-  -- | Action-  IO () ->-  IO ()-recurring_ wheel (Micros.fromSeconds -> delay) action =-  void (_register wheel delay doAction)-  where-    doAction :: IO ()-    doAction = do-      _ <- _reregister wheel delay doAction-      action---- Re-register one bucket early, to account for the fact that timers are--- expired at the *end* of a bucket.------ +---+---+---+---+--- { A |   |   |   }--- +---+---+---+---+---      |---      The reaper thread fires 'A' approximately here, so if it's meant---      to be repeated every two buckets, and we just re-register it at---      this time, three buckets will pass before it's run again. So, we---      act as if it's still "one bucket ago" at the moment we re-register---      it.-_reregister :: TimerWheel -> Micros -> IO () -> IO (IO Bool)-_reregister wheel delay =-  _register wheel (if reso > delay then Micros 0 else delay `Micros.minus` reso)-  where-    reso :: Micros-    reso =-      resolution wheel--resolution :: TimerWheel -> Micros-resolution =-  Wheel.resolution . wheel
− src/Data/TimerWheel/Internal/Config.hs 
@@ -1,19 +0,0 @@-module Data.TimerWheel.Internal.Config-  ( Config (..),-  )-where--import Data.Fixed (E6, Fixed)-import GHC.Generics (Generic)---- | Timer wheel config.------ * @spokes@ must be ∈ @(0, maxBound]@--- * @resolution@ must ∈ @(0, ∞]@-data Config = Config-  { -- | Spoke count.-    spokes :: Int,-    -- | Resolution, in seconds.-    resolution :: Fixed E6-  }-  deriving stock (Generic, Show)
− src/Data/TimerWheel/Internal/Entries.hs 
@@ -1,69 +0,0 @@-{-# LANGUAGE TypeApplications #-}--module Data.TimerWheel.Internal.Entries-  ( Entries,-    empty,-    Data.TimerWheel.Internal.Entries.null,-    size,-    insert,-    delete,-    partition,-  )-where--import Data.Coerce-import Data.IntPSQ (IntPSQ)-import qualified Data.IntPSQ as IntPSQ-import Data.Word (Word64)--newtype Entries-  = Entries (IntPSQ Word64 (IO ()))---- | An empty collection.-empty :: Entries-empty =-  Entries IntPSQ.empty-{-# INLINEABLE empty #-}--null :: Entries -> Bool-null =-  coerce (IntPSQ.null @Word64 @(IO ()))-{-# INLINEABLE null #-}---- | The number of timers in the collection.-size :: Entries -> Int-size =-  coerce (IntPSQ.size @Word64 @(IO ()))-{-# INLINEABLE size #-}---- | @insert i n m x@ inserts callback @m@ into collection @x@ with unique--- identifier @i@ and "count" @n@. The-insert :: Int -> Word64 -> IO () -> Entries -> Entries-insert i n m =-  coerce (IntPSQ.unsafeInsertNew i n m)-{-# INLINEABLE insert #-}---- | Delete a timer by id. Returns 'Nothing' if the timer was not found.-delete :: Int -> Entries -> Maybe Entries-delete =-  coerce delete_-{-# INLINEABLE delete #-}--delete_ :: Int -> IntPSQ Word64 (IO ()) -> Maybe (IntPSQ Word64 (IO ()))-delete_ i xs =-  (\(_, _, ys) -> ys) <$> IntPSQ.deleteView i xs---- | Extract expired timers.-partition :: Entries -> ([IO ()], Entries)-partition (Entries entries) =-  case IntPSQ.atMostView 0 entries of-    (expired, alive) ->-      (map f expired, Entries (IntPSQ.unsafeMapMonotonic g alive))-  where-    f :: (Int, Word64, IO ()) -> IO ()-    f (_, _, m) =-      m-    g :: Int -> Word64 -> IO () -> (Word64, IO ())-    g _ n m =-      (n -1, m)-{-# INLINEABLE partition #-}
− src/Data/TimerWheel/Internal/Micros.hs 
@@ -1,46 +0,0 @@-{-# LANGUAGE TypeApplications #-}--module Data.TimerWheel.Internal.Micros-  ( Micros (..),-    fromFixed,-    fromSeconds,-    Data.TimerWheel.Internal.Micros.div,-    minus,-    scale,-    sleep,-  )-where--import Control.Concurrent (threadDelay)-import Data.Coerce-import Data.Fixed-import Data.Word--newtype Micros = Micros {unMicros :: Word64}-  deriving stock (Eq, Ord)---- | Precondition: input is >= 0-fromFixed :: Fixed E6 -> Micros-fromFixed =-  coerce @(Integer -> Word64) fromIntegral--fromSeconds :: Fixed E6 -> Micros-fromSeconds seconds@(MkFixed micros)-  | micros < 0 = error ("[timer-wheel] invalid seconds: " ++ show seconds)-  | otherwise = Micros (fromIntegral micros)--div :: Micros -> Micros -> Micros-div =-  coerce (Prelude.div @Word64)--minus :: Micros -> Micros -> Micros-minus =-  coerce ((-) @Word64)--scale :: Int -> Micros -> Micros-scale n (Micros w) =-  Micros (fromIntegral n * w)--sleep :: Micros -> IO ()-sleep (Micros micros) =-  threadDelay (fromIntegral micros)
− src/Data/TimerWheel/Internal/Supply.hs 
@@ -1,21 +0,0 @@-module Data.TimerWheel.Internal.Supply-  ( Supply,-    new,-    next,-  )-where--import Data.Atomics.Counter (AtomicCounter, incrCounter, newCounter)-import Data.Coerce (coerce)--newtype Supply-  = Supply AtomicCounter--new :: IO Supply-new =-  coerce (newCounter 0)--next :: Supply -> IO Int-next =-  coerce (incrCounter 1)-{-# INLINE next #-}
− src/Data/TimerWheel/Internal/Timestamp.hs 
@@ -1,42 +0,0 @@-{-# LANGUAGE TypeApplications #-}--module Data.TimerWheel.Internal.Timestamp-  ( Timestamp (..),-    epoch,-    minus,-    now,-    plus,-    Data.TimerWheel.Internal.Timestamp.rem,-  )-where--import Data.Coerce (coerce)-import Data.TimerWheel.Internal.Micros (Micros (..))-import Data.Word (Word64)-import GHC.Clock (getMonotonicTimeNSec)--newtype Timestamp-  = Timestamp Word64-  deriving stock (Eq, Ord)---- Which epoch does this correspond to, if they are measured in chunks of the given number of milliseconds?-epoch :: Micros -> Timestamp -> Word64-epoch (Micros chunk) (Timestamp timestamp) =-  timestamp `div` chunk--minus :: Timestamp -> Timestamp -> Micros-minus =-  coerce ((-) @Word64)--now :: IO Timestamp-now = do-  nanos <- getMonotonicTimeNSec-  pure (Timestamp (nanos `div` 1000))--plus :: Timestamp -> Micros -> Timestamp-plus =-  coerce ((+) @Word64)--rem :: Timestamp -> Micros -> Micros-rem =-  coerce (Prelude.rem @Word64)
− src/Data/TimerWheel/Internal/Wheel.hs 
@@ -1,85 +0,0 @@-{-# LANGUAGE StrictData #-}--module Data.TimerWheel.Internal.Wheel-  ( Wheel (resolution),-    create,-    lenMicros,-    insert,-    reap,-  )-where--import Control.Monad (join, when)-import Data.IORef-import Data.TimerWheel.Internal.Entries (Entries)-import qualified Data.TimerWheel.Internal.Entries as Entries-import Data.TimerWheel.Internal.Micros (Micros (..))-import qualified Data.TimerWheel.Internal.Micros as Micros-import Data.TimerWheel.Internal.Timestamp (Timestamp)-import qualified Data.TimerWheel.Internal.Timestamp as Timestamp-import Data.Vector (Vector)-import qualified Data.Vector as Vector--data Wheel = Wheel-  { buckets :: Vector (IORef Entries),-    resolution :: Micros-  }--create :: Int -> Micros -> IO Wheel-create spokes resolution = do-  buckets <- Vector.replicateM spokes (newIORef Entries.empty)-  pure Wheel {buckets, resolution}--numSpokes :: Wheel -> Int-numSpokes wheel =-  Vector.length (buckets wheel)--lenMicros :: Wheel -> Micros-lenMicros wheel =-  Micros.scale (numSpokes wheel) (resolution wheel)--bucket :: Wheel -> Timestamp -> IORef Entries-bucket wheel timestamp =-  Vector.unsafeIndex (buckets wheel) (index wheel timestamp)--index :: Wheel -> Timestamp -> Int-index wheel@Wheel {resolution} timestamp =-  fromIntegral (Timestamp.epoch resolution timestamp) `rem` numSpokes wheel--insert :: Wheel -> Int -> Micros -> IO () -> IO (IO Bool)-insert wheel key delay action = do-  bucketRef <- do-    now <- Timestamp.now-    pure (bucket wheel (now `Timestamp.plus` delay))--  atomicModifyIORef' bucketRef (\entries -> (insertEntry entries, ()))--  pure do-    atomicModifyIORef' bucketRef \entries ->-      case Entries.delete key entries of-        Nothing -> (entries, False)-        Just entries' -> (entries', True)-  where-    insertEntry :: Entries -> Entries-    insertEntry =-      Entries.insert key (unMicros (delay `Micros.div` lenMicros wheel)) action--reap :: Wheel -> IO ()-reap wheel@Wheel {buckets, resolution} = do-  now <- Timestamp.now-  let remainingBucketMicros = resolution `Micros.minus` (now `Timestamp.rem` resolution)-  Micros.sleep remainingBucketMicros-  loop (now `Timestamp.plus` remainingBucketMicros `Timestamp.plus` resolution) (index wheel now)-  where-    loop :: Timestamp -> Int -> IO ()-    loop nextTime i = do-      join (atomicModifyIORef' (Vector.unsafeIndex buckets i) expire)-      afterTime <- Timestamp.now-      when (afterTime < nextTime) (Micros.sleep (nextTime `Timestamp.minus` afterTime))-      loop (nextTime `Timestamp.plus` resolution) ((i + 1) `rem` numSpokes wheel)-    expire :: Entries -> (Entries, IO ())-    expire entries-      | Entries.null entries = (entries, pure ())-      | otherwise = (alive, sequence_ expired)-      where-        (expired, alive) = Entries.partition entries
+ src/TimerWheel.hs view
@@ -0,0 +1,229 @@+{-# LANGUAGE RecursiveDo #-}++-- | A simple, hashed timer wheel.+module TimerWheel+  ( -- * Timer wheel+    TimerWheel,+    create,+    with,+    Config (..),+    register,+    register_,+    recurring,+    recurring_,+  )+where++import Control.Exception (throwIO)+import Control.Monad (when)+import Data.Bool (bool)+import Data.Fixed (E6, Fixed)+import Data.Function (fix)+import Data.IORef (newIORef, readIORef, writeIORef)+import GHC.Exception (errorCallException)+import qualified Ki+import TimerWheel.Internal.Config (Config)+import qualified TimerWheel.Internal.Config as Config+import TimerWheel.Internal.Micros (Micros (Micros))+import qualified TimerWheel.Internal.Micros as Micros+import TimerWheel.Internal.Supply (Supply)+import qualified TimerWheel.Internal.Supply as Supply+import TimerWheel.Internal.Wheel (Wheel)+import qualified TimerWheel.Internal.Wheel as Wheel++-- | A timer wheel is a vector-of-collections-of timers to fire. It is configured with a /spoke count/ and /resolution/.+-- Timers may be scheduled arbitrarily far in the future. A timeout thread is spawned to step through the timer wheel+-- and fire expired timers at regular intervals.+--+-- * The /spoke count/ determines the size of the timer vector.+--+--     * A __larger spoke count__ will result in __less insert contention__ at each spoke and will require+--       __more memory__ to store the timer wheel.+--+--     * A __smaller spoke count__ will result in __more insert contention__ at each spoke and will require+--       __less memory__ to store the timer wheel.+--+-- * The /resolution/ determines both the duration of time that each spoke corresponds to, and how often the timeout+--   thread wakes. For example, with a resolution of __@1s@__, a timer that expires at __@2.5s@__ will not fire until+--   the timeout thread wakes at __@3s@__.+--+--     * A __larger resolution__ will result in __more insert contention__ at each spoke, __less accurate__ timers, and+--       will require __fewer wakeups__ by the timeout thread.+--+--     * A __smaller resolution__ will result in __less insert contention__ at each spoke, __more accurate__ timers, and+--       will require __more wakeups__ by the timeout thread.+--+-- * The timeout thread has some important properties:+--+--     * There is only one, and it fires expired timers synchronously. If your timer actions execute quicky, 'register'+--       them directly. Otherwise, consider registering an action that enqueues the /real/ action to be performed on a+--       job queue.+--+--     * Synchronous exceptions thrown by enqueued @IO@ actions will bring the thread down, and the exception will be+--       propagated to the thread that created the timer wheel. If you want to catch exceptions and log them, for+--       example, you will have to bake this into the registered actions yourself.+--+-- As an example, below is a depiction of a timer wheel with @6@ timers inserted across @8@ spokes, and a resolution of+-- @.1s@. It depicts a cursor at @.3s@, which indicates where the timeout thread currently is.+--+-- @+--  0       .1      .2      .3      .4      .5      .6      .7+-- ┌───────┬───────┬───────┬───────┬───────┬───────┬───────┬───────┐+-- │       │ A⁰    │       │ B¹ C⁰ │ D⁰    │       │       │ E² F⁰ │+-- └───────┴───────┴───────┴───────┴───────┴───────┴───────┴───────┘+--                           ↑+-- @+--+-- After @.1s@, the timeout thread will advance to the next spoke and process all of the timers it passed over. In+-- this case, __C__ will fire, and __B__ will be put back with its count decremented to @0@. This is how the timer wheel+-- can schedule a timer to fire arbitrarily far in the future: its count is simply the number of times its delay wraps+-- the entire duration of the timer wheel.+--+-- @+--  0       .1      .2      .3      .4      .5      .6      .7+-- ┌───────┬───────┬───────┬───────┬───────┬───────┬───────┬───────┐+-- │       │ A⁰    │       │ B⁰    │ D⁰    │       │       │ E² F⁰ │+-- └───────┴───────┴───────┴───────┴───────┴───────┴───────┴───────┘+--                                   ↑+-- @+data TimerWheel = TimerWheel+  { -- | A supply of unique ints.+    supply :: {-# UNPACK #-} !Supply,+    -- | The array of collections of timers.+    wheel :: {-# UNPACK #-} !Wheel+    -- thread :: {-# UNPACK #-} !ThreadId+  }++-- | Create a timer wheel in a scope.+--+-- /Throws./+--+--   * Calls 'error' if the config is invalid+create :: Ki.Scope -> Config -> IO TimerWheel+create scope config = do+  validateConfig config+  wheel <- Wheel.create (Config.spokes config) (Micros.fromFixed (Config.resolution config))+  supply <- Supply.new+  Ki.fork_ scope (Wheel.reap wheel)+  pure TimerWheel {supply, wheel}++-- | Perform an action with a timer wheel.+--+-- /Throws./+--+--   * Calls 'error' if the config is invalid+--   * Throws the exception the given action throws, if any+--   * Throws the exception the timer wheel thread throws, if any+with :: Config -> (TimerWheel -> IO a) -> IO a+with config action =+  Ki.scoped \scope -> do+    wheel <- create scope config+    action wheel++validateConfig :: Config -> IO ()+validateConfig config =+  when (Config.spokes config <= 0 || Config.resolution config <= 0) do+    throwIO (errorCallException ("timer-wheel: invalid config: " ++ show config))++-- | @register wheel delay action@ registers an action __@action@__ in timer wheel __@wheel@__ to fire after __@delay@__+-- seconds.+--+-- Returns an action that, when called, attempts to cancel the timer, and returns whether or not it was successful+-- (@False@ means the timer has already fired, or was already cancelled).+register ::+  TimerWheel ->+  -- | Delay, in seconds+  Fixed E6 ->+  -- | Action+  IO () ->+  IO (IO Bool)+register wheel delay =+  registerImpl wheel (Micros.fromSeconds (max 0 delay))++-- | Like 'register', but for when you don't intend to cancel the timer.+register_ ::+  TimerWheel ->+  -- | Delay, in seconds+  Fixed E6 ->+  -- | Action+  IO () ->+  IO ()+register_ wheel delay action = do+  _ <- register wheel delay action+  pure ()++registerImpl :: TimerWheel -> Micros -> IO () -> IO (IO Bool)+registerImpl TimerWheel {supply, wheel} delay action = do+  key <- Supply.next supply+  Wheel.insert wheel key delay action++-- | @recurring wheel action delay@ registers an action __@action@__ in timer wheel __@wheel@__ to fire every+-- __@delay@__ seconds (or every /resolution/ seconds, whichever is smaller).+--+-- Returns an action that, when called, cancels the recurring timer.+recurring ::+  TimerWheel ->+  -- | Delay, in seconds+  Fixed E6 ->+  -- | Action+  IO () ->+  IO (IO ())+recurring wheel (Micros.fromSeconds -> delay) action = mdo+  let doAction :: IO ()+      doAction = do+        cancel <- reregister wheel delay doAction+        writeIORef cancelRef cancel+        action+  cancel0 <- registerImpl wheel delay doAction+  cancelRef <- newIORef cancel0+  pure do+    untilTrue do+      cancel <- readIORef cancelRef+      cancel++-- | Like 'recurring', but for when you don't intend to cancel the timer.+recurring_ ::+  TimerWheel ->+  -- | Delay, in seconds+  Fixed E6 ->+  -- | Action+  IO () ->+  IO ()+recurring_ wheel (Micros.fromSeconds -> delay) action = do+  _ <- registerImpl wheel delay doAction+  pure ()+  where+    doAction :: IO ()+    doAction = do+      _ <- reregister wheel delay doAction+      action++-- Re-register one bucket early, to account for the fact that timers are+-- expired at the *end* of a bucket.+--+-- +---+---+---+---++-- { A |   |   |   }+-- +---+---+---+---++--      |+--      The reaper thread fires 'A' approximately here, so if it's meant+--      to be repeated every two buckets, and we just re-register it at+--      this time, three buckets will pass before it's run again. So, we+--      act as if it's still "one bucket ago" at the moment we re-register+--      it.+reregister :: TimerWheel -> Micros -> IO () -> IO (IO Bool)+reregister wheel delay =+  registerImpl wheel (if reso > delay then Micros 0 else delay `Micros.minus` reso)+  where+    reso :: Micros+    reso =+      resolution wheel++resolution :: TimerWheel -> Micros+resolution =+  Wheel.resolution . wheel++-- Repeat an IO action until it returns 'True'.+untilTrue :: IO Bool -> IO ()+untilTrue m =+  fix \again ->+    m >>= bool again (pure ())
+ src/TimerWheel/Internal/Config.hs view
@@ -0,0 +1,19 @@+module TimerWheel.Internal.Config+  ( Config (..),+  )+where++import Data.Fixed (E6, Fixed)+import GHC.Generics (Generic)++-- | Timer wheel config.+--+-- * @spokes@ must be ∈ @[1, maxBound]@+-- * @resolution@ must ∈ @(0, ∞]@+data Config = Config+  { -- | Spoke count.+    spokes :: {-# UNPACK #-} !Int,+    -- | Resolution, in seconds.+    resolution :: !(Fixed E6)+  }+  deriving stock (Generic, Show)
+ src/TimerWheel/Internal/Entries.hs view
@@ -0,0 +1,67 @@+module TimerWheel.Internal.Entries+  ( Entries,+    empty,+    TimerWheel.Internal.Entries.null,+    size,+    insert,+    delete,+    partition,+  )+where++import Data.Coerce+import Data.IntPSQ (IntPSQ)+import qualified Data.IntPSQ as IntPSQ+import Data.Word (Word64)++newtype Entries+  = Entries (IntPSQ Word64 (IO ()))++-- | An empty collection.+empty :: Entries+empty =+  Entries IntPSQ.empty+{-# INLINEABLE empty #-}++null :: Entries -> Bool+null =+  coerce (IntPSQ.null @Word64 @(IO ()))+{-# INLINEABLE null #-}++-- | The number of timers in the collection.+size :: Entries -> Int+size =+  coerce (IntPSQ.size @Word64 @(IO ()))+{-# INLINEABLE size #-}++-- | @insert i n m x@ inserts callback @m@ into collection @x@ with unique+-- identifier @i@ and "count" @n@. The+insert :: Int -> Word64 -> IO () -> Entries -> Entries+insert i n m =+  coerce (IntPSQ.unsafeInsertNew i n m)+{-# INLINEABLE insert #-}++-- | Delete a timer by id. Returns 'Nothing' if the timer was not found.+delete :: Int -> Entries -> Maybe Entries+delete =+  coerce delete_+{-# INLINEABLE delete #-}++delete_ :: Int -> IntPSQ Word64 (IO ()) -> Maybe (IntPSQ Word64 (IO ()))+delete_ i xs =+  (\(_, _, ys) -> ys) <$> IntPSQ.deleteView i xs++-- | Extract expired timers.+partition :: Entries -> ([IO ()], Entries)+partition (Entries entries) =+  case IntPSQ.atMostView 0 entries of+    (expired, alive) ->+      (map f expired, Entries (IntPSQ.unsafeMapMonotonic g alive))+  where+    f :: (Int, Word64, IO ()) -> IO ()+    f (_, _, m) =+      m+    g :: Int -> Word64 -> IO () -> (Word64, IO ())+    g _ n m =+      (n -1, m)+{-# INLINEABLE partition #-}
+ src/TimerWheel/Internal/Micros.hs view
@@ -0,0 +1,44 @@+module TimerWheel.Internal.Micros+  ( Micros (..),+    fromFixed,+    fromSeconds,+    TimerWheel.Internal.Micros.div,+    minus,+    scale,+    sleep,+  )+where++import Control.Concurrent (threadDelay)+import Data.Coerce+import Data.Fixed+import Data.Word++newtype Micros = Micros {unMicros :: Word64}+  deriving stock (Eq, Ord)++-- | Precondition: input is >= 0+fromFixed :: Fixed E6 -> Micros+fromFixed =+  coerce @(Integer -> Word64) fromIntegral++fromSeconds :: Fixed E6 -> Micros+fromSeconds seconds@(MkFixed micros)+  | micros < 0 = error ("[timer-wheel] invalid seconds: " ++ show seconds)+  | otherwise = Micros (fromIntegral micros)++div :: Micros -> Micros -> Micros+div =+  coerce (Prelude.div @Word64)++minus :: Micros -> Micros -> Micros+minus =+  coerce ((-) @Word64)++scale :: Int -> Micros -> Micros+scale n (Micros w) =+  Micros (fromIntegral n * w)++sleep :: Micros -> IO ()+sleep (Micros micros) =+  threadDelay (fromIntegral micros)
+ src/TimerWheel/Internal/Supply.hs view
@@ -0,0 +1,21 @@+module TimerWheel.Internal.Supply+  ( Supply,+    new,+    next,+  )+where++import Data.Atomics.Counter (AtomicCounter, incrCounter, newCounter)+import Data.Coerce (coerce)++newtype Supply+  = Supply AtomicCounter++new :: IO Supply+new =+  coerce (newCounter 0)++next :: Supply -> IO Int+next =+  coerce (incrCounter 1)+{-# INLINE next #-}
+ src/TimerWheel/Internal/Timestamp.hs view
@@ -0,0 +1,40 @@+module TimerWheel.Internal.Timestamp+  ( Timestamp (..),+    epoch,+    minus,+    now,+    plus,+    TimerWheel.Internal.Timestamp.rem,+  )+where++import Data.Coerce (coerce)+import Data.Word (Word64)+import GHC.Clock (getMonotonicTimeNSec)+import TimerWheel.Internal.Micros (Micros (..))++newtype Timestamp+  = Timestamp Word64+  deriving stock (Eq, Ord)++-- Which epoch does this correspond to, if they are measured in chunks of the given number of milliseconds?+epoch :: Micros -> Timestamp -> Word64+epoch (Micros chunk) (Timestamp timestamp) =+  timestamp `div` chunk++minus :: Timestamp -> Timestamp -> Micros+minus =+  coerce ((-) @Word64)++now :: IO Timestamp+now = do+  nanos <- getMonotonicTimeNSec+  pure (Timestamp (nanos `div` 1000))++plus :: Timestamp -> Micros -> Timestamp+plus =+  coerce ((+) @Word64)++rem :: Timestamp -> Micros -> Micros+rem =+  coerce (Prelude.rem @Word64)
+ src/TimerWheel/Internal/Wheel.hs view
@@ -0,0 +1,84 @@+module TimerWheel.Internal.Wheel+  ( Wheel (resolution),+    create,+    lenMicros,+    insert,+    reap,+  )+where++import Control.Monad (join, replicateM, when)+import Data.Array (Array)+import qualified Data.Array as Array+import Data.IORef+import TimerWheel.Internal.Entries (Entries)+import qualified TimerWheel.Internal.Entries as Entries+import TimerWheel.Internal.Micros (Micros (..))+import qualified TimerWheel.Internal.Micros as Micros+import TimerWheel.Internal.Timestamp (Timestamp)+import qualified TimerWheel.Internal.Timestamp as Timestamp++data Wheel = Wheel+  { buckets :: {-# UNPACK #-} !(Array Int (IORef Entries)),+    resolution :: {-# UNPACK #-} !Micros+  }++create :: Int -> Micros -> IO Wheel+create spokes resolution = do+  refs <- replicateM spokes (newIORef Entries.empty)+  let buckets = Array.listArray (0, spokes - 1) refs+  pure Wheel {buckets, resolution}++numSpokes :: Wheel -> Int+numSpokes wheel =+  length (buckets wheel)++lenMicros :: Wheel -> Micros+lenMicros wheel =+  Micros.scale (numSpokes wheel) (resolution wheel)++bucket :: Wheel -> Timestamp -> IORef Entries+bucket wheel timestamp =+  buckets wheel Array.! index wheel timestamp++index :: Wheel -> Timestamp -> Int+index wheel@Wheel {resolution} timestamp =+  fromIntegral (Timestamp.epoch resolution timestamp) `rem` numSpokes wheel++insert :: Wheel -> Int -> Micros -> IO () -> IO (IO Bool)+insert wheel key delay action = do+  bucketRef <- do+    now <- Timestamp.now+    pure (bucket wheel (now `Timestamp.plus` delay))++  atomicModifyIORef' bucketRef (\entries -> (insertEntry entries, ()))++  pure do+    atomicModifyIORef' bucketRef \entries ->+      case Entries.delete key entries of+        Nothing -> (entries, False)+        Just entries' -> (entries', True)+  where+    insertEntry :: Entries -> Entries+    insertEntry =+      Entries.insert key (unMicros (delay `Micros.div` lenMicros wheel)) action++reap :: Wheel -> IO a+reap wheel@Wheel {buckets, resolution} = do+  now <- Timestamp.now+  let remainingBucketMicros = resolution `Micros.minus` (now `Timestamp.rem` resolution)+  Micros.sleep remainingBucketMicros+  loop (now `Timestamp.plus` remainingBucketMicros `Timestamp.plus` resolution) (index wheel now)+  where+    loop :: Timestamp -> Int -> IO a+    loop nextTime i = do+      join (atomicModifyIORef' (buckets Array.! i) expire)+      afterTime <- Timestamp.now+      when (afterTime < nextTime) (Micros.sleep (nextTime `Timestamp.minus` afterTime))+      loop (nextTime `Timestamp.plus` resolution) ((i + 1) `rem` numSpokes wheel)+    expire :: Entries -> (Entries, IO ())+    expire entries+      | Entries.null entries = (entries, pure ())+      | otherwise = (alive, sequence_ expired)+      where+        (expired, alive) = Entries.partition entries
test/Main.hs view
@@ -1,19 +1,23 @@-{-# LANGUAGE BlockArguments, DeriveAnyClass, DerivingStrategies, FlexibleInstances, MultiParamTypeClasses #-}+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}  import Control.Concurrent import Control.Exception import Control.Monad import Data.IORef import Data.Maybe (isJust)-import Data.TimerWheel import System.Mem (performGC) import System.Mem.Weak (deRefWeak)+import TimerWheel  main :: IO () main = do   do     putStrLn "Timer wheel runs scheduled actions"-    with Config { spokes = 4, resolution = 0.05 } \wheel -> do+    with Config {spokes = 4, resolution = 0.05} \wheel -> do       var <- newEmptyMVar       let n = 1000       replicateM_ n (register_ wheel 0 (putMVar var ()))@@ -22,7 +26,7 @@   do     putStrLn "Timers can be canceled"     var <- newEmptyMVar-    with Config { spokes = 4, resolution = 0.05 } \wheel -> do+    with Config {spokes = 4, resolution = 0.05} \wheel -> do       let n = 1000       cancels <- replicateM n (register wheel 0 (putMVar var ()))       successes <- sequence (take (n `div` 2) cancels)@@ -30,14 +34,14 @@    do     putStrLn "Successful `cancel` returns True (then False)"-    with Config { spokes = 4, resolution = 0.05 } \wheel -> do+    with Config {spokes = 4, resolution = 0.05} \wheel -> do       cancel <- register wheel 1 (pure ())       cancel `is` True       cancel `is` False    do     putStrLn "Unsuccessful `cancel` returns False"-    with Config { spokes = 4, resolution = 0.05 } \wheel -> do+    with Config {spokes = 4, resolution = 0.05} \wheel -> do       var <- newEmptyMVar       cancel <- register wheel 0 (putMVar var ())       takeMVar var@@ -45,7 +49,7 @@    do     putStrLn "Recurring timers work"-    with Config { spokes = 4, resolution = 0.05 } \wheel -> do+    with Config {spokes = 4, resolution = 0.05} \wheel -> do       canary <- newIORef () -- kept alive only by timer       weakCanary <- mkWeakIORef canary (pure ())       var <- newEmptyMVar@@ -58,15 +62,17 @@   do     putStrLn "`with` re-throws exception from background thread"     catch-      (with Config { spokes = 4, resolution = 0.05 } \wheel -> do-        var <- newEmptyMVar-        register_ wheel 0 (throwIO Bye >> putMVar var ())-        takeMVar var-        throwIO (userError "fail"))-      (\ex ->-        case fromException ex of-          Just Bye -> pure ()-          _ -> throwIO ex)+      ( with Config {spokes = 4, resolution = 0.05} \wheel -> do+          var <- newEmptyMVar+          register_ wheel 0 (throwIO Bye >> putMVar var ())+          takeMVar var+          throwIO (userError "fail")+      )+      ( \ex ->+          case fromException ex of+            Just Bye -> pure ()+            _ -> throwIO ex+      )  data Bye = Bye   deriving stock (Show)
timer-wheel.cabal view
@@ -1,7 +1,7 @@ cabal-version: 2.0  name: timer-wheel-version: 0.3.0+version: 0.4.0 category: Data description:   This library provides a timer wheel data structure for@@ -9,7 +9,7 @@   * (Almost) /O(1)/ registering @IO@ actions to fire after a given amount of time   * /O(1)/ canceling registered actions   .-  It is similar to @TimerManager@ from "GHC.Event", but supports recurring+  It is similar to @TimerManager@ from @GHC.Event@, but supports recurring   timers, can scale to handle many more registered timers.   . synopsis: A timer wheel@@ -21,6 +21,7 @@ license: BSD3 license-file: LICENSE build-type: Simple+tested-with: GHC ==8.10.7, GHC ==9.0.2, GHC ==9.2.1  extra-source-files:   CHANGELOG.md@@ -33,10 +34,11 @@  library   build-depends:+    array ^>= 0.5.2.0,     atomic-primops ^>= 0.8,-    base ^>= 4.9 || ^>= 4.10 || ^>= 4.11 || ^>= 4.12 || ^>= 4.13 || ^>= 4.14,-    psqueues ^>= 0.2.7,-    vector ^>= 0.10 || ^>= 0.11 || ^>= 0.12+    base ^>= 4.12 || ^>= 4.13 || ^>= 4.14 || ^>= 4.15 || ^>= 4.16 || ^>= 4.17,+    ki ^>= 1.0.0,+    psqueues ^>= 0.2.7    default-extensions:     BlockArguments@@ -56,7 +58,7 @@     Haskell2010    exposed-modules:-    Data.TimerWheel+    TimerWheel    ghc-options:     -Weverything@@ -68,17 +70,20 @@     ghc-options:       -Wno-missing-safe-haskell-mode       -Wno-prepositive-qualified-module+  if impl(ghc >= 9.2)+    ghc-options:+      -Wno-missing-kind-signatures    hs-source-dirs:     src    other-modules:-    Data.TimerWheel.Internal.Config-    Data.TimerWheel.Internal.Entries-    Data.TimerWheel.Internal.Micros-    Data.TimerWheel.Internal.Supply-    Data.TimerWheel.Internal.Timestamp-    Data.TimerWheel.Internal.Wheel+    TimerWheel.Internal.Config+    TimerWheel.Internal.Entries+    TimerWheel.Internal.Micros+    TimerWheel.Internal.Supply+    TimerWheel.Internal.Timestamp+    TimerWheel.Internal.Wheel  test-suite tests   build-depends: