timer-wheel 0.4.0.1 → 1.0.0
raw patch · 18 files changed
+1153/−547 lines, 18 filesdep +primitivedep +randomdep +tasty-benchdep −arraydep −psqueuesdep ~basedep ~ki
Dependencies added: primitive, random, tasty-bench
Dependencies removed: array, psqueues
Dependency ranges changed: base, ki
Files
- CHANGELOG.md +14/−0
- LICENSE +1/−1
- README.md +1/−1
- bench/Main.hs +35/−0
- cabal.project +0/−1
- src/TimerWheel.hs +438/−165
- src/TimerWheel/Internal/Bucket.hs +308/−0
- src/TimerWheel/Internal/Config.hs +0/−19
- src/TimerWheel/Internal/Counter.hs +85/−0
- src/TimerWheel/Internal/Entries.hs +0/−67
- src/TimerWheel/Internal/Micros.hs +0/−44
- src/TimerWheel/Internal/Nanoseconds.hs +45/−0
- src/TimerWheel/Internal/Prelude.hs +22/−0
- src/TimerWheel/Internal/Supply.hs +0/−21
- src/TimerWheel/Internal/Timestamp.hs +22/−22
- src/TimerWheel/Internal/Wheel.hs +0/−84
- test/Main.hs +115/−60
- timer-wheel.cabal +67/−62
CHANGELOG.md view
@@ -1,3 +1,17 @@+## [1.0.0] - 2023-10-10++- Add `count`, which returns the number of timers in a timer wheel+- Add `Seconds` type alias for readability+- Add `Timer` newtype for readability+- Make `create` / `with` no longer throw an exception if given an invalid config; rather, the config's invalid values are+ replaced with sensible defaults+- Make `recurring` / `recurring_` handle delays that are shorter than the wheel resolution more correctly+- Make `recurring` / `recurring_` no longer throw an exception if given a negative delay+- Make calling `cancel` more than once on a recurring timer not enter an infinite loop+- Make timers that expire in the same batch no longer fire in an arbitrary order+- Improve the resolution of timers from microseconds to nanoseconds+- Simplify and optimize internals+ ## [0.4.0.1] - 2022-11-05 - Fix inaccurate haddock on `recurring`
LICENSE view
@@ -1,4 +1,4 @@-Copyright 2018 Mitchell Rosen+Copyright 2018-2023 Mitchell Rosen, Travis Staton Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
README.md view
@@ -1,6 +1,6 @@ # `timer-wheel` -[](https://github.com/mitchellwrosen/timer-wheel/actions)+[](https://github.com/awkward-squad/timer-wheel/actions) [](https://hackage.haskell.org/package/timer-wheel) [](https://www.stackage.org/lts/package/timer-wheel) [](https://www.stackage.org/nightly/package/timer-wheel)
+ bench/Main.hs view
@@ -0,0 +1,35 @@+import Control.Exception (evaluate)+import Control.Monad (replicateM)+import Data.Foldable (for_)+import GHC.Conc (atomically)+import qualified Ki+import qualified System.Random as Random+import qualified System.Random.Stateful as Random+import Test.Tasty.Bench+import TimerWheel+import Prelude++main :: IO ()+main = do+ Ki.scoped \scope -> do+ mainThread <- Ki.fork scope main1+ atomically (Ki.await mainThread)++main1 :: IO ()+main1 = do+ let delays :: [Double]+ delays =+ Random.runStateGen_+ (Random.mkStdGen 0)+ (replicateM 1_000_000 . Random.uniformRM (0, 60 * 5))+ _ <- evaluate (sum delays)++ defaultMain+ [ bench "insert 1m" (whnfIO (insertN (map (realToFrac @Double @Seconds) delays)))+ ]++insertN :: [Seconds] -> IO ()+insertN delays =+ with Config {spokes = 1024, resolution = 1} \wheel -> do+ for_ delays \delay -> do+ register wheel delay (pure ())
− cabal.project
@@ -1,1 +0,0 @@-packages: .
src/TimerWheel.hs view
@@ -1,229 +1,502 @@-{-# LANGUAGE RecursiveDo #-}---- | A simple, hashed timer wheel.+-- | This module is intended to be imported qualified:+--+-- > import TimerWheel (TimerWheel)+-- > import TimerWheel qualified module TimerWheel ( -- * Timer wheel TimerWheel,++ -- * Timer wheel configuration+ Config (..),+ Seconds,++ -- * Timer+ Timer,++ -- * Constructing a timer wheel create, with,- Config (..),++ -- * Querying a timer wheel+ count,++ -- * Registering timers in a timer wheel register, register_, recurring, recurring_,++ -- * Canceling timers+ cancel, ) 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 Control.Exception (mask_)+import qualified Data.Atomics as Atomics+import Data.Functor (void)+import Data.Primitive.Array (MutableArray)+import qualified Data.Primitive.Array as Array+import GHC.Base (RealWorld) 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+import TimerWheel.Internal.Bucket (Bucket)+import qualified TimerWheel.Internal.Bucket as Bucket+import TimerWheel.Internal.Counter (Counter, decrCounter_, incrCounter, incrCounter_, newCounter, readCounter)+import TimerWheel.Internal.Nanoseconds (Nanoseconds (..))+import qualified TimerWheel.Internal.Nanoseconds as Nanoseconds+import TimerWheel.Internal.Prelude+import TimerWheel.Internal.Timestamp (Timestamp)+import qualified TimerWheel.Internal.Timestamp as Timestamp --- | 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.+-- | A timer wheel is a vector-of-collections-of timers to fire. Timers may be one-shot or recurring, and may be+-- scheduled arbitrarily far in the future. --+-- A timer wheel is configured with a /spoke count/ and /resolution/:+-- -- * 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 __larger spoke count__ will require __more memory__, but will result in __less insert contention__. ----- * 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 the duration of time that each spoke corresponds to, and thus how often timers are+-- checked for expiry. ----- * 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@__.+-- For example, in a timer wheel with a /resolution/ of __@1 second@__, a timer that is scheduled to fire at+-- __@8.4 o'clock@__ will end up firing around __@9.0 o'clock@__ instead (that is, on the+-- __@1 second@__-boundary). ----- * 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 __larger resolution__ will result in __more insert contention__ and __less accurate timers__, but 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: ----- * The timeout thread has some important properties:+-- * There is only one, and it fires expired timers synchronously. If your timer actions execute quicky, you can+-- 'register' them directly. Otherwise, consider registering an action that enqueues the real action to be+-- performed on a job queue. ----- * 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.+-- * A synchronous exception thrown by a registered timer will bring the timeout thread down, and the exception will+-- be propagated to the thread that created the timer wheel. If you want to log and ignore exceptions, for example,+-- you will have to bake this into the registered actions yourself. ----- * 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.+-- __API summary__ ----- 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.+-- +----------+---------+----------------++-- | Create | Query | Modify |+-- +==========+=========+================++-- | 'create' | 'count' | 'register' |+-- +----------+---------+----------------++-- | 'with' | | 'register_' |+-- +----------+ +----------------++-- | | | 'recurring' |+-- | | +----------------++-- | | | 'recurring_' |+-- +----------+---------+----------------++data TimerWheel = TimerWheel+ { buckets :: {-# UNPACK #-} !(MutableArray RealWorld (Bucket Timer0)),+ resolution :: {-# UNPACK #-} !Nanoseconds,+ numTimers :: {-# UNPACK #-} !Counter,+ -- A counter to generate unique ints that identify registered actions, so they can be canceled.+ timerIdSupply :: {-# UNPACK #-} !Counter+ }++-- | A timer wheel config. ----- @--- 0 .1 .2 .3 .4 .5 .6 .7--- ┌───────┬───────┬───────┬───────┬───────┬───────┬───────┬───────┐--- │ │ A⁰ │ │ B¹ C⁰ │ D⁰ │ │ │ E² F⁰ │--- └───────┴───────┴───────┴───────┴───────┴───────┴───────┴───────┘--- ↑--- @+-- * @spokes@ must be ∈ @[1, maxBound]@, and is set to @1024@ if invalid.+-- * @resolution@ must be ∈ @(0, ∞]@, and is set to @1@ if invalid. ----- 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.+-- __API summary__ ----- @--- 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 |+-- +==========++-- | 'Config' |+-- +----------++data Config = Config+ { -- | Spoke count+ spokes :: {-# UNPACK #-} !Int,+ -- | Resolution+ resolution :: !Seconds }+ deriving stock (Generic, Show) -- | 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}+create ::+ -- | + Ki.Scope ->+ -- | + Config ->+ -- | + IO TimerWheel+create scope (Config spokes0 resolution0) = do+ buckets <- Array.newArray spokes Bucket.empty+ numTimers <- newCounter+ timerIdSupply <- newCounter+ Ki.fork_ scope (runTimerReaperThread buckets numTimers resolution)+ pure TimerWheel {buckets, numTimers, resolution, timerIdSupply}+ where+ spokes = if spokes0 <= 0 then 1024 else spokes0+ resolution = Nanoseconds.fromNonNegativeSeconds (if resolution0 <= 0 then 1 else resolution0) -- | 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 ->+ -- | + (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))+-- | Get the number of timers in a timer wheel.+--+-- /O(1)/.+count :: TimerWheel -> IO Int+count TimerWheel {numTimers} =+ readCounter numTimers --- | @register wheel delay action@ registers an action __@action@__ in timer wheel __@wheel@__ to fire after __@delay@__--- seconds.+-- | @register wheel delay action@ registers __@action@__ in __@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).+-- When canceled, the timer returns whether or not the cancelation was successful; @False@ means the timer had either+-- already fired, or had already been canceled. register ::+ -- | The timer wheel TimerWheel ->- -- | Delay, in seconds- Fixed E6 ->- -- | Action+ -- | The delay before the action is fired+ Seconds ->+ -- | The action to fire IO () ->- IO (IO Bool)-register wheel delay =- registerImpl wheel (Micros.fromSeconds (max 0 delay))+ -- | The timer+ IO (Timer Bool)+register TimerWheel {buckets, numTimers, resolution, timerIdSupply} delay action = do+ now <- Timestamp.now+ let timestamp = now `Timestamp.plus` Nanoseconds.fromSeconds delay+ let index = timestampToIndex buckets resolution timestamp+ timerId <- incrCounter timerIdSupply+ mask_ do+ atomicModifyArray buckets index (Bucket.insert timerId timestamp (OneShot1 action))+ incrCounter_ numTimers+ coerce @(IO (IO Bool)) @(IO (Timer Bool)) do+ pure do+ mask_ do+ deleted <- atomicMaybeModifyArray buckets index (Bucket.deleteExpectingHit timerId)+ when deleted (decrCounter_ numTimers)+ pure deleted -- | Like 'register', but for when you don't intend to cancel the timer. register_ ::+ -- | The timer wheel TimerWheel ->- -- | Delay, in seconds- Fixed E6 ->- -- | Action+ -- | The delay before the action is fired+ Seconds ->+ -- | The action to fire 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+register_ wheel delay action =+ void (register wheel delay action) --- | @recurring wheel action delay@ registers an action __@action@__ in timer wheel __@wheel@__ to fire every--- __@delay@__ seconds.------ Returns an action that, when called, cancels the recurring timer.+-- | @recurring wheel action delay@ registers __@action@__ in __@wheel@__ to fire in __@delay@__ seconds, and every+-- __@delay@__ seconds thereafter. recurring ::+ -- | The timer wheel TimerWheel ->- -- | Delay, in seconds- Fixed E6 ->- -- | Action+ -- | The delay before each action is fired+ Seconds ->+ -- | The action to fire repeatedly 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+ -- | The timer+ IO (Timer ())+recurring TimerWheel {buckets, numTimers, resolution, timerIdSupply} (Nanoseconds.fromSeconds -> delay) action = do+ now <- Timestamp.now+ let timestamp = now `Timestamp.plus` delay+ let index = timestampToIndex buckets resolution timestamp+ timerId <- incrCounter timerIdSupply+ canceledRef <- newIORef False+ mask_ do+ atomicModifyArray buckets index (Bucket.insert timerId timestamp (Recurring1 action delay canceledRef))+ incrCounter_ numTimers+ coerce @(IO (IO ())) @(IO (Timer ())) do+ pure do+ mask_ do+ writeIORef canceledRef True+ decrCounter_ numTimers -- | Like 'recurring', but for when you don't intend to cancel the timer. recurring_ :: TimerWheel ->- -- | Delay, in seconds- Fixed E6 ->- -- | Action+ -- | The delay before each action is fired+ Seconds ->+ -- | The action to fire repeatedly IO () -> IO ()-recurring_ wheel (Micros.fromSeconds -> delay) action = do- _ <- registerImpl wheel delay doAction- pure ()- where- doAction :: IO ()- doAction = do- _ <- reregister wheel delay doAction- action+recurring_ TimerWheel {buckets, numTimers, resolution, timerIdSupply} (Nanoseconds.fromSeconds -> delay) action = do+ now <- Timestamp.now+ let timestamp = now `Timestamp.plus` delay+ let index = timestampToIndex buckets resolution timestamp+ timerId <- incrCounter timerIdSupply+ mask_ do+ atomicModifyArray buckets index (Bucket.insert timerId timestamp (Recurring1_ action delay))+ incrCounter_ numTimers --- Re-register one bucket early, to account for the fact that timers are--- expired at the *end* of a bucket.+-- | A registered timer, parameterized by the result of attempting to cancel it: ----- +---+---+---+---+--- { 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)+-- * A one-shot timer may only be canceled if it has not already fired.+-- * A recurring timer can always be canceled.+--+-- __API summary__+--+-- +-------------+----------++-- | Create | Modify |+-- +=============+==========++-- | 'register' | 'cancel' |+-- +-------------+----------++-- | 'recurring' | |+-- +-------------+----------++newtype Timer a+ = Timer (IO a)++-- | Cancel a timer.+cancel :: Timer a -> IO a+cancel =+ coerce++-- `timestampToIndex buckets resolution timestamp` figures out which index `timestamp` corresponds to in `buckets`,+-- where each bucket corresponds to `resolution` nanoseconds.+--+-- For example, consider a three-element `buckets` with resolution `1000000000`.+--+-- +--------------------------------------++-- | 1000000000 | 1000000000 | 1000000000 |+-- +--------------------------------------++--+-- Some timestamp like `1053298012387` gets binned to one of the three indices 0, 1, or 2, with quick and easy maffs:+--+-- 1. Figure out which index the timestamp corresponds to, if there were infinitely many:+--+-- 1053298012387 `div` 1000000000 = 1053+--+-- 2. Wrap around per the actual length of the array:+--+-- 1053 `rem` 3 = 0+timestampToIndex :: MutableArray RealWorld bucket -> Nanoseconds -> Timestamp -> Int+timestampToIndex buckets resolution timestamp =+ -- This downcast is safe because there are at most `maxBound :: Int` buckets (not that anyone would ever have that+ -- many...)+ fromIntegral @Word64 @Int+ (Timestamp.epoch resolution timestamp `rem` fromIntegral @Int @Word64 (Array.sizeofMutableArray buckets))++data Timer0+ = OneShot1 !(IO ())+ | Recurring1 !(IO ()) !Nanoseconds !(IORef Bool)+ | Recurring1_ !(IO ()) !Nanoseconds++type TimerId =+ Int++------------------------------------------------------------------------------------------------------------------------+-- Atomic operations on arrays++atomicModifyArray :: forall a. MutableArray RealWorld a -> Int -> (a -> a) -> IO ()+atomicModifyArray array index f = do+ ticket0 <- Atomics.readArrayElem array index+ loop ticket0 where- reso :: Micros- reso =- resolution wheel+ loop :: Atomics.Ticket a -> IO ()+ loop ticket = do+ (success, ticket1) <- Atomics.casArrayElem array index ticket (f (Atomics.peekTicket ticket))+ if success then pure () else loop ticket1 -resolution :: TimerWheel -> Micros-resolution =- Wheel.resolution . wheel+atomicMaybeModifyArray :: forall a. MutableArray RealWorld a -> Int -> (a -> Maybe a) -> IO Bool+atomicMaybeModifyArray buckets index doDelete = do+ ticket0 <- Atomics.readArrayElem buckets index+ loop ticket0+ where+ loop :: Atomics.Ticket a -> IO Bool+ loop ticket =+ case doDelete (Atomics.peekTicket ticket) of+ Nothing -> pure False+ Just bucket -> do+ (success, ticket1) <- Atomics.casArrayElem buckets index ticket bucket+ if success then pure True else loop ticket1 --- Repeat an IO action until it returns 'True'.-untilTrue :: IO Bool -> IO ()-untilTrue m =- fix \again ->- m >>= bool again (pure ())+atomicExtractExpiredTimersFromBucket :: MutableArray RealWorld (Bucket Timer0) -> Int -> Timestamp -> IO (Bucket Timer0)+atomicExtractExpiredTimersFromBucket buckets index now = do+ ticket0 <- Atomics.readArrayElem buckets index+ loop ticket0+ where+ loop :: Atomics.Ticket (Bucket Timer0) -> IO (Bucket Timer0)+ loop ticket = do+ let Bucket.Pair expired bucket1 = Bucket.partition now (Atomics.peekTicket ticket)+ if Bucket.isEmpty expired+ then pure Bucket.empty+ else do+ (success, ticket1) <- Atomics.casArrayElem buckets index ticket bucket1+ if success then pure expired else loop ticket1++------------------------------------------------------------------------------------------------------------------------+-- Timer reaper thread+--+-- The main loop is rather simple, but the code is somewhat fiddly. In brief, the reaper thread wakes up to fire all of+-- the expired timers in bucket N, then sleeps, then wakes up to fire all of the expired timers in bucket N+1, then+-- sleeps, and so on, forever.+--+-- It wakes up on the "bucket boundaries", that is,+--+-- +------+------+------+------+------+------+------+------+------+------++-- | | | | | | | | | | |+-- | | | | | | | | | | |+-- +------+------+------+------+------+------+------+------+------+------++-- ^ ^+-- | we wake up around here+-- |+-- to fire all of the expired timers stored here+--+-- It's entirely possible the reaper thread gets hopelessly behind, that is, it's taken so long to expire all of the+-- timers in previous buckets that we're behind schedule an entire bucket or more. That might look like this:+--+-- +------+------+------+------+------+------+------+------+------+------++-- | | | | | | | | | | |+-- | | | | | | | | | | |+-- +------+------+------+------+------+------+------+------+------+------++-- ^ ^+-- | we are very behind, and enter the loop around here+-- |+-- yet we nonetheless fire all of the expired timers stored here, as if we were on time+--+-- That's accomplished simplly by maintaining in the loop state the "ideal" time that we wake up, ignoring reality. We+-- only ultimately check the *actual* current time when determining how long to *sleep* after expiring all of the timers+-- in the current bucket. If we're behind schedule, we won't sleep at all.+--+-- +------+------+------+------+------+------+------+------+------+------++-- | | | | | | | | | | |+-- | | | | | | | | | | |+-- +------+------+------+------+------+------+------+------+------+------++-- ^ ^ ^+-- | | |+-- | we enter the loop with this "ideal" time+-- | |+-- to fire timers in here |+-- |+-- not caring how far ahead the actual current time is+--+-- On to expiring timers: a "bucket" of timers is stored at each array index, which can be partitioned into "expired"+-- (meant to fire at or before the ideal time) and "not expired" (to expire on a subsequent wrap around the bucket+-- array).+--+-- +-----------------------++-- | / |+-- | expired / |+-- | / not expired |+-- | / |+-- +-----------------------++--+-- The reaper thread simply atomically partitions the bucket, keeping the expired collection for itself, and putting the+-- not-expired collection back in the array.+--+-- Next, the timers are carefully fired one-by-one, in timestamp order. It's possible that two or more timers are+-- scheduled to expire concurrently (i.e. on the same nanosecond); that's fine: we fire them in the order they were+-- scheduled.+--+-- Let's say this is our set of timers to fire.+--+-- Ideal time Timers to fire+-- +--------------+ +-----------------------------++-- | 700 | | Expiry | Type |+-- +--------------+ +--------+--------------------++-- | 630 | One-shot |+-- Next ideal time | 643 | Recurring every 10 |+-- +--------------+ | 643 | One-shot |+-- | 800 | | 689 | Recurring every 80 |+-- +--------------+ +--------+--------------------++--+-- Expiring a one-shot timer is simple: call the IO action and move on.+--+-- Expiring a recurring timer is less simple (but still simple): call the IO action, then schedule the next occurrence.+-- There are two possibilities.+--+-- 1. The next occurrence is *at or before* the ideal time, which means it ought to fire along with the other timers+-- in the queue, right now. So, insert it into the collection of timers to fire.+--+-- 2. The next occurrence is *after* the ideal time, so enqueue it in the array of buckets wherever it belongs.+--+-- After all expired timers are fired, the reaper thread has one last decision to make: how long should we sleep? We+-- get the current timestamp, and if it's still before the next ideal time (i.e. the current ideal time plus the wheel+-- resolution), then we sleep for the difference.+--+-- If the actual time is at or after the next ideal time, that's kind of bad - it means the reaper thread is behind+-- schedule. The user's enqueued actions have taken too long, or their wheel resolution is too short. Anyway, it's not+-- our problem, our behavior doesn't change per whether we are behind schedule or not.+runTimerReaperThread :: MutableArray RealWorld (Bucket Timer0) -> Counter -> Nanoseconds -> IO void+runTimerReaperThread buckets numTimers resolution = do+ -- Sleep until the very first bucket of timers expires+ --+ -- resolution = 100+ -- now = 184070+ -- progress = now % resolution = 70+ -- remaining = resolution - progress = 30+ -- idealTime = now + remaining = 184100+ --+ -- +-------------------------+----------------+---------+ -- | progress = 70 | remaining = 30 |+ -- +-------------------------+----------------++ -- | resolution = 100 |+ -- +------------------------------------------+---------+ -- ^ ^+ -- now idealTime+ now <- Timestamp.now+ let progress = now `Timestamp.intoEpoch` resolution+ let remaining = resolution `Nanoseconds.unsafeMinus` progress+ Nanoseconds.sleep remaining+ -- Enter the Loop™+ let idealTime = now `Timestamp.plus` remaining+ theLoop idealTime (timestampToIndex buckets resolution now)+ where+ -- `index` could be derived from `thisTime`, but it's cheaper to just store it separately and bump by 1 as we go+ theLoop :: Timestamp -> Int -> IO void+ theLoop !idealTime !index = do+ expired2 <- atomicExtractExpiredTimersFromBucket buckets index idealTime+ fireTimerBucket expired2+ let !nextIdealTime = idealTime `Timestamp.plus` resolution+ now <- Timestamp.now+ when (nextIdealTime > now) (Nanoseconds.sleep (nextIdealTime `Timestamp.unsafeMinus` now))+ theLoop nextIdealTime ((index + 1) `rem` Array.sizeofMutableArray buckets)+ where+ fireTimerBucket :: Bucket Timer0 -> IO ()+ fireTimerBucket bucket0 =+ case Bucket.pop bucket0 of+ Bucket.PopNada -> pure ()+ Bucket.PopAlgo timerId timestamp timer bucket1 -> do+ expired2 <- fireTimer bucket1 timerId timestamp timer+ fireTimerBucket expired2++ fireTimer :: Bucket Timer0 -> TimerId -> Timestamp -> Timer0 -> IO (Bucket Timer0)+ fireTimer bucket timerId timestamp timer =+ case timer of+ OneShot1 action -> do+ action+ decrCounter_ numTimers+ pure bucket+ Recurring1 action delay canceledRef ->+ readIORef canceledRef >>= \case+ True -> pure bucket+ False -> do+ action+ scheduleNextOccurrence (timestamp `Timestamp.plus` delay)+ Recurring1_ action delay -> do+ action+ scheduleNextOccurrence (timestamp `Timestamp.plus` delay)+ where+ scheduleNextOccurrence :: Timestamp -> IO (Bucket Timer0)+ scheduleNextOccurrence nextOccurrence =+ if nextOccurrence < idealTime+ then pure $! insertNextOccurrence bucket+ else do+ atomicModifyArray+ buckets+ (timestampToIndex buckets resolution nextOccurrence)+ insertNextOccurrence+ pure bucket+ where+ insertNextOccurrence :: Bucket Timer0 -> Bucket Timer0+ insertNextOccurrence =+ Bucket.insert timerId nextOccurrence timer
+ src/TimerWheel/Internal/Bucket.hs view
@@ -0,0 +1,308 @@+{-# LANGUAGE CPP #-}++-- Code from this implementation was cribbed from `psqueues`, whose license is copied below.+--+-- The Glasgow Haskell Compiler License+--+-- Copyright 2004, The University Court of the University of Glasgow.+-- All rights reserved.+--+-- Redistribution and use in source and binary forms, with or without+-- modification, are permitted provided that the following conditions are met:+--+-- - Redistributions of source code must retain the above copyright notice,+-- this list of conditions and the following disclaimer.+--+-- - Redistributions in binary form must reproduce the above copyright notice,+-- this list of conditions and the following disclaimer in the documentation+-- and/or other materials provided with the distribution.+--+-- - Neither name of the University nor the names of its contributors may be+-- used to endorse or promote products derived from this software without+-- specific prior written permission.+--+-- THIS SOFTWARE IS PROVIDED BY THE UNIVERSITY COURT OF THE UNIVERSITY OF+-- GLASGOW AND THE CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,+-- INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND+-- FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE+-- UNIVERSITY COURT OF THE UNIVERSITY OF GLASGOW OR THE CONTRIBUTORS BE LIABLE+-- FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+-- DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR+-- SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER+-- CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT+-- LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY+-- OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH+-- DAMAGE.++module TimerWheel.Internal.Bucket+ ( Bucket,+ empty,++ -- * Queries+ isEmpty,+ partition,++ -- * Modifications+ insert,+ Pop (..),+ pop,+ deleteExpectingHit,++ -- * Strict pair+ Pair (..),+ )+where++import Data.Bits+import TimerWheel.Internal.Prelude+import TimerWheel.Internal.Timestamp (Timestamp)++#include "MachDeps.h"++data Bucket a+ = -- Invariants on `Bin k p v m l r`:+ -- 1. `l` and `r` can't both be Nil+ -- 2. `p` is <= all `p` in `l` and `r`+ -- 3. `k` is not an element of `l` nor `r`+ -- 4. `m` has one 1-bit, which is (usually) the highest bit position at which any two keys in `k`+`l`+`r` differ+ -- (henceforth referred to as the diffbit). It can get out-of-date by deletions and pops. Thus, `m` represents+ -- a left-bound on the diffbit, that is, the true diffbit can be no left-er than `m`.+ -- 5. No key in `l` has the `m` bit set+ -- 6. All keys in `r` have the `m` bit set+ Bin {-# UNPACK #-} !TimerId {-# UNPACK #-} !Timestamp !a {-# UNPACK #-} !Mask !(Bucket a) !(Bucket a)+ | Tip {-# UNPACK #-} !TimerId {-# UNPACK #-} !Timestamp !a+ | Nil++type Mask = Word64++type TimerId = Int++-- | An empty bucket.+empty :: Bucket a+empty =+ Nil++isEmpty :: Bucket a -> Bool+isEmpty = \case+ Nil -> True+ _ -> False++-- | Partition a bucket by timestamp (less-than-or-equal-to, greater-than).+partition :: forall a. Timestamp -> Bucket a -> Pair (Bucket a) (Bucket a)+partition q =+ go empty+ where+ go :: Bucket a -> Bucket a -> Pair (Bucket a) (Bucket a)+ go acc t =+ case t of+ Nil -> Pair acc t+ Tip i p x+ | p > q -> Pair acc t+ | otherwise -> Pair (insert i p x acc) Nil+ Bin i p x m l r+ | p > q -> Pair acc t+ | otherwise ->+ case go acc l of+ Pair acc1 l1 ->+ case go acc1 r of+ Pair acc2 r1 -> Pair (insert i p x acc2) (merge m l1 r1)++-- | Insert a new timer into a bucket.+--+-- If a timer with the given id is already in the bucket, behavior is undefined.+insert :: forall a. TimerId -> Timestamp -> a -> Bucket a -> Bucket a+insert i p x bucket =+ case bucket of+ Nil -> Tip i p x+ Tip j q y+ | betteri -> linki j bucket Nil+ | otherwise -> linkj i (Tip i p x) Nil+ where+ betteri = (p, i) < (q, j)+ linkj = link j q y+ Bin j q y m l r+ | betteri ->+ if+ | outsider -> linki j bucket Nil+ | goleft j m -> bini (insertj l) r+ | otherwise -> bini l (insertj r)+ | outsider -> linkj i (Tip i p x) (merge m l r)+ | goleft i m -> binj (inserti l) r+ | otherwise -> binj l (inserti r)+ where+ outsider = prefixNotEqual m i j+ betteri = (p, i) < (q, j)+ bini = Bin i p x m+ binj = Bin j q y m+ inserti = insert i p x+ insertj = insert j q y+ linkj = link j q y+ where+ linki = link i p x++data Pop a+ = PopAlgo {-# UNPACK #-} !TimerId {-# UNPACK #-} !Timestamp !a !(Bucket a)+ | PopNada++pop :: Bucket a -> Pop a+pop = \case+ Nil -> PopNada+ Tip k p x -> PopAlgo k p x Nil+ Bin k p x m l r -> PopAlgo k p x (merge m l r)+{-# INLINE pop #-}++-- | Delete a timer from a bucket, expecting it to be there.+deleteExpectingHit :: TimerId -> Bucket v -> Maybe (Bucket v)+deleteExpectingHit i =+ go+ where+ go :: Bucket v -> Maybe (Bucket v)+ go = \case+ Nil -> Nothing+ Tip j _ _+ | i == j -> Just Nil+ | otherwise -> Nothing+ Bin j p x m l r+ -- This commented out short-circuit is what makes this delete variant "expecting a hit"+ -- | prefixNotEqual m i j -> Nothing+ | i == j -> Just $! merge m l r+ | goleft i m -> (\l1 -> bin j p x m l1 r) <$> go l+ | otherwise -> bin j p x m l <$> go r++-- | 'Bin' smart constructor, respecting the invariant that both children can't be 'Nil'.+bin :: TimerId -> Timestamp -> v -> Mask -> Bucket v -> Bucket v -> Bucket v+bin i p x _ Nil Nil = Tip i p x+bin i p x m l r = Bin i p x m l r+{-# INLINE bin #-}++link :: TimerId -> Timestamp -> v -> TimerId -> Bucket v -> Bucket v -> Bucket v+link i p x j t u+ | goleft j m = Bin i p x m t u+ | otherwise = Bin i p x m u t+ where+ m = onlyHighestBit (i2w i `xor` i2w j)+{-# INLINE link #-}++-- Merge two disjoint buckets that have the same mask.+merge :: Mask -> Bucket v -> Bucket v -> Bucket v+merge m l r =+ case (l, r) of+ (Nil, _) -> r+ (_, Nil) -> l+ --+ -- ip jq+ --+ (Tip i p x, Tip j q y)+ --+ -- ip+ -- / \+ -- nil jq+ --+ | (p, i) < (q, j) -> Bin i p x m Nil r+ --+ -- jq+ -- / \+ -- ip nil+ --+ | otherwise -> Bin j q y m l Nil+ --+ -- ip jq+ -- / \+ -- rl rr+ --+ (Tip i p x, Bin j q y n rl rr)+ --+ -- ip+ -- / \+ -- nil jq+ -- / \+ -- rl rr+ --+ | (p, i) < (q, j) -> Bin i p x m Nil r+ --+ -- jq+ -- / \+ -- ip rl+rr+ --+ | otherwise -> Bin j q y m l (merge n rl rr)+ --+ -- ip jq+ -- / \+ -- ll lr+ --+ (Bin i p x n ll lr, Tip j q y)+ --+ -- ip+ -- / \+ -- ll+lr jq+ --+ | (p, i) < (q, j) -> Bin i p x m (merge n ll lr) r+ --+ -- jq+ -- / \+ -- ip nil+ -- / \+ -- ll lr+ --+ | otherwise -> Bin j q y m l Nil+ --+ -- ip jq+ -- / \ / \+ -- ll lr rl rr+ --+ (Bin i p x n ll lr, Bin j q y o rl rr)+ --+ -- ip+ -- / \+ -- ll+lr jq+ -- / \+ -- rl rr+ --+ | (p, i) < (q, j) -> Bin i p x m (merge n ll lr) r+ --+ -- jq+ -- / \+ -- ip rl+rr+ -- / \+ -- ll lr+ --+ | otherwise -> Bin j q y m l (merge o rl rr)++------------------------------------------------------------------------------------------------------------------------+-- Bit fiddling++-- | Is (or should) this timer be stored on the left of this bin, given its mask?+goleft :: TimerId -> Mask -> Bool+goleft i m =+ i2w i .&. m == 0+{-# INLINE goleft #-}++-- m = 00001000000000000000000+-- i = IIII???????????????????+-- j = JJJJ???????????????????+--+-- prefixNotEqual m i j answers, is IIII not equal to JJJJ?+prefixNotEqual :: Mask -> TimerId -> TimerId -> Bool+prefixNotEqual (prefixMask -> e) i j =+ i2w i .&. e /= i2w j .&. e+{-# INLINE prefixNotEqual #-}++-- m = 0000000000100000+-- prefixMask m = 1111111111000000+prefixMask :: Word64 -> Word64+prefixMask m = -m `xor` m+{-# INLINE prefixMask #-}++onlyHighestBit :: Word64 -> Mask+onlyHighestBit w = unsafeShiftL 1 (WORD_SIZE_IN_BITS - 1 - countLeadingZeros w)+{-# INLINE onlyHighestBit #-}++i2w :: TimerId -> Word64+i2w = fromIntegral+{-# INLINE i2w #-}++------------------------------------------------------------------------------------------------------------------------+-- Strict pair++data Pair a b+ = Pair !a !b
− src/TimerWheel/Internal/Config.hs
@@ -1,19 +0,0 @@-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/Counter.hs view
@@ -0,0 +1,85 @@+-- Some code modified from the atomic-primops library; license included below.+--+-- Copyright (c)2012-2013, Ryan R. Newton+--+-- All rights reserved.+--+-- Redistribution and use in source and binary forms, with or without+-- modification, are permitted provided that the following conditions are met:+--+-- * Redistributions of source code must retain the above copyright+-- notice, this list of conditions and the following disclaimer.+--+-- * Redistributions in binary form must reproduce the above+-- copyright notice, this list of conditions and the following+-- disclaimer in the documentation and/or other materials provided+-- with the distribution.+--+-- * Neither the name of Ryan R. Newton nor the names of other+-- contributors may be used to endorse or promote products derived+-- from this software without specific prior written permission.+{-# LANGUAGE MagicHash #-}+{-# LANGUAGE UnboxedTuples #-}++module TimerWheel.Internal.Counter+ ( Counter,+ newCounter,+ readCounter,+ incrCounter,+ incrCounter_,+ decrCounter_,+ )+where++import Data.Bits+import GHC.Base+import Prelude++-- | A thread-safe counter implemented with atomic fetch-and-add.+data Counter+ = Counter (MutableByteArray# RealWorld)++-- | Create a new counter initialized to 0.+newCounter :: IO Counter+newCounter =+ IO \s0# ->+ case newByteArray# size s0# of+ (# s1#, arr# #) ->+ case writeIntArray# arr# 0# 0# s1# of+ s2# -> (# s2#, Counter arr# #)+ where+ !(I# size) =+ finiteBitSize (undefined :: Int) `div` 8+{-# INLINE newCounter #-}++-- | Get the value of a counter.+readCounter :: Counter -> IO Int+readCounter (Counter arr#) =+ IO \s0# ->+ case readIntArray# arr# 0# s0# of+ (# s1#, n# #) -> (# s1#, I# n# #)+{-# INLINE readCounter #-}++-- | Increment a counter and return the value prior to incrementing.+incrCounter :: Counter -> IO Int+incrCounter (Counter arr#) =+ IO \s0# ->+ case fetchAddIntArray# arr# 0# 1# s0# of+ (# s1#, n# #) -> (# s1#, I# n# #)+{-# INLINE incrCounter #-}++-- | Increment a counter.+incrCounter_ :: Counter -> IO ()+incrCounter_ (Counter arr#) =+ IO \s0# ->+ case fetchAddIntArray# arr# 0# 1# s0# of+ (# s1#, _ #) -> (# s1#, () #)+{-# INLINE incrCounter_ #-}++-- | Decrement a counter.+decrCounter_ :: Counter -> IO ()+decrCounter_ (Counter arr#) =+ IO \s0# ->+ case fetchSubIntArray# arr# 0# 1# s0# of+ (# s1#, _ #) -> (# s1#, () #)+{-# INLINE decrCounter_ #-}
− src/TimerWheel/Internal/Entries.hs
@@ -1,67 +0,0 @@-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
@@ -1,44 +0,0 @@-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/Nanoseconds.hs view
@@ -0,0 +1,45 @@+module TimerWheel.Internal.Nanoseconds+ ( Nanoseconds (..),+ fromSeconds,+ fromNonNegativeSeconds,+ div,+ unsafeMinus,+ sleep,+ )+where++import Control.Concurrent (threadDelay)+import Data.Fixed (Fixed (..))+import TimerWheel.Internal.Prelude hiding (div)+import qualified Prelude++-- Some positive number of nanoseconds+newtype Nanoseconds = Nanoseconds {unNanoseconds :: Word64}+ deriving stock (Eq, Ord)++-- | Convert a number of seconds into a number of nanoseconds.+--+-- Negative values are converted to 0.+fromSeconds :: Seconds -> Nanoseconds+fromSeconds =+ fromNonNegativeSeconds . max 0++-- | Like 'fromNonNegativeSeconds', but with an unchecked precondition: the given seconds is non-negative.+--+-- What you get for your troubles: one puny fewer int comparisons.+fromNonNegativeSeconds :: Seconds -> Nanoseconds+fromNonNegativeSeconds seconds =+ Nanoseconds (coerce @(Integer -> Word64) fromIntegral seconds)++div :: Nanoseconds -> Nanoseconds -> Nanoseconds+div =+ coerce (Prelude.div @Word64)++-- `unsafeMinus n m` subtracts `m` from `n`, but does something wild if `m` is bigger than `n`+unsafeMinus :: Nanoseconds -> Nanoseconds -> Nanoseconds+unsafeMinus =+ coerce ((-) @Word64)++sleep :: Nanoseconds -> IO ()+sleep (Nanoseconds nanos) =+ threadDelay (fromIntegral (nanos `Prelude.div` 1000))
+ src/TimerWheel/Internal/Prelude.hs view
@@ -0,0 +1,22 @@+module TimerWheel.Internal.Prelude+ ( Seconds,+ module X,+ )+where++import Control.Monad as X (when)+import Data.Coerce as X (coerce)+import Data.Fixed (E9, Fixed)+import Data.IORef as X (IORef, newIORef, readIORef, writeIORef)+import Data.Word as X (Word64)+import GHC.Generics as X (Generic)+import Prelude as X hiding (lookup, null)++-- | A number of seconds, with nanosecond precision.+--+-- You can use numeric literals to construct a value of this type, e.g. @0.5@.+--+-- Otherwise, to convert from a type like @Int@ or @Double@, you can use the generic numeric conversion function+-- @realToFrac@.+type Seconds =+ Fixed E9
− src/TimerWheel/Internal/Supply.hs
@@ -1,21 +0,0 @@-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
@@ -1,40 +1,40 @@ module TimerWheel.Internal.Timestamp ( Timestamp (..), epoch,- minus,- now,+ intoEpoch,+ unsafeMinus, plus,- TimerWheel.Internal.Timestamp.rem,+ now, ) where -import Data.Coerce (coerce)-import Data.Word (Word64) import GHC.Clock (getMonotonicTimeNSec)-import TimerWheel.Internal.Micros (Micros (..))+import TimerWheel.Internal.Nanoseconds (Nanoseconds (..))+import TimerWheel.Internal.Prelude+import qualified Prelude +-- Monotonic time, in nanoseconds newtype Timestamp- = Timestamp Word64+ = Timestamp Nanoseconds 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+-- Which epoch does this correspond to, if they are measured in chunks of the given number of nanoseconds?+epoch :: Nanoseconds -> Timestamp -> Word64+epoch x y =+ coerce @_ @Word64 y `div` coerce x -minus :: Timestamp -> Timestamp -> Micros-minus =- coerce ((-) @Word64)+intoEpoch :: Timestamp -> Nanoseconds -> Nanoseconds+intoEpoch =+ coerce (Prelude.rem @Word64) -now :: IO Timestamp-now = do- nanos <- getMonotonicTimeNSec- pure (Timestamp (nanos `div` 1000))+unsafeMinus :: Timestamp -> Timestamp -> Nanoseconds+unsafeMinus =+ coerce ((-) @Word64) -plus :: Timestamp -> Micros -> Timestamp+plus :: Timestamp -> Nanoseconds -> Timestamp plus = coerce ((+) @Word64) -rem :: Timestamp -> Micros -> Micros-rem =- coerce (Prelude.rem @Word64)+now :: IO Timestamp+now =+ coerce getMonotonicTimeNSec
− src/TimerWheel/Internal/Wheel.hs
@@ -1,84 +0,0 @@-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,87 +1,142 @@-{-# 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 System.Mem (performGC)-import System.Mem.Weak (deRefWeak)+import GHC.Conc (atomically)+import qualified Ki+import qualified System.Random as Random import TimerWheel+import Prelude+import Data.Foldable (traverse_) main :: IO () main = do- do- putStrLn "Timer wheel runs scheduled actions"- with Config {spokes = 4, resolution = 0.05} \wheel -> do- var <- newEmptyMVar- let n = 1000- replicateM_ n (register_ wheel 0 (putMVar var ()))- replicateM_ n (takeMVar var)+ Ki.scoped \scope -> do+ mainThread <- Ki.fork scope main1+ atomically (Ki.await mainThread) - do- putStrLn "Timers can be canceled"+main1 :: IO ()+main1 = do+ putStrLn "timer wheel runs scheduled actions"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do var <- newEmptyMVar- 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)- replicateM_ (n - length (filter id successes)) (takeMVar var)+ let n = 1000 :: Int+ replicateM_ n (register_ wheel 0 (putMVar var ()))+ replicateM_ n (takeMVar var) - do- putStrLn "Successful `cancel` returns True (then False)"- with Config {spokes = 4, resolution = 0.05} \wheel -> do- cancel <- register wheel 1 (pure ())- cancel `is` True- cancel `is` False+ putStrLn "timers can be canceled"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do+ var <- newEmptyMVar+ let n = 1000 :: Int+ timers <- replicateM n (register wheel 0 (putMVar var ()))+ successes <- traverse cancel (take (n `div` 2) timers)+ replicateM_ (n - length (filter id successes)) (takeMVar var) - do- putStrLn "Unsuccessful `cancel` returns False"- with Config {spokes = 4, resolution = 0.05} \wheel -> do+ putStrLn "successful `cancel` returns true, then false"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do+ timer <- register wheel 1 (pure ())+ cancel timer `is` True+ cancel timer `is` False++ putStrLn "unsuccessful `cancel` returns false"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do+ var <- newEmptyMVar+ timer <- register wheel 0 (putMVar var ())+ takeMVar var+ cancel timer `is` False++ putStrLn "recurring timers work with delay > resolution work"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do+ ref <- newIORef (0 :: Int)+ recurring_ wheel 0.2 (modifyIORef' ref (+ 1))+ threadDelay 1_100_000+ readIORef ref `is` (5 :: Int)++ putStrLn "calling `cancel` more than once on a recurring timer is ok"+ with Config {spokes = 4, resolution = 0.05} \wheel -> do+ timer <- recurring wheel 1 (pure ())+ cancel timer+ -- At one time, this one would loop indefinitely :grimace:+ cancel timer++ putStrLn "`with` re-throws exception from background thread"+ ( with Config {spokes = 16, resolution = 0.05} \wheel -> do var <- newEmptyMVar- cancel <- register wheel 0 (putMVar var ())+ register_ wheel 0 (throwIO Bye >> putMVar var ()) takeMVar var- cancel `is` False+ )+ `is` Exn Bye - do- putStrLn "Recurring timers work"- with Config {spokes = 4, resolution = 0.05} \wheel -> do- canary <- newIORef () -- kept alive only by timer- weakCanary <- mkWeakIORef canary (pure ())- var <- newEmptyMVar- cancel <- recurring wheel 0 (readIORef canary >> putMVar var ())- replicateM_ 2 (takeMVar var)- cancel -- should drop reference canary after a GC- performGC- (isJust <$> deRefWeak weakCanary) `is` False+ putStrLn "`count` increments on `register`"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do+ let n = 10 :: Int+ replicateM_ n (register wheel 1 (pure ()))+ count wheel `is` n - 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- )+ putStrLn "`count` increments on `recurring`"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do+ let n = 10 :: Int+ replicateM_ n (recurring_ wheel 1 (pure ()))+ count wheel `is` n + putStrLn "`count` decrements on `cancel` (registered)"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do+ let n = 10 :: Int+ timers <- replicateM n (register wheel 1 (pure ()))+ traverse cancel timers `is` replicate n True+ count wheel `is` (0 :: Int)++ putStrLn "`count` decrements on `cancel` (recurring)"+ with Config {spokes = 16, resolution = 0.05} \wheel -> do+ let n = 10 :: Int+ timers <- replicateM n (recurring wheel 1 (pure ()))+ traverse_ cancel timers+ count wheel `is` (0 :: Int)++ putStrLn "stress test: register 1m timers into 1k spokes, then cancel them all"+ with Config {spokes = 1000, resolution = 1} \wheel -> do+ firedRef <- newIORef (0 :: Int)+ let registerLoop :: [Timer Bool] -> Random.StdGen -> Int -> IO [Timer Bool]+ registerLoop timers gen0 !i+ | i >= 1_000_000 = pure timers+ | otherwise = do+ let (delay, gen1) = Random.uniformR (0 :: Double, 10_000) gen0+ timer <- register wheel (realToFrac @Double @Seconds delay) (modifyIORef' firedRef (+ 1))+ registerLoop (timer : timers) gen1 (i + 1)+ let cancelLoop :: Int -> [Timer Bool] -> IO Int+ cancelLoop !n = \case+ [] -> pure n+ timer : timers -> do+ success <- cancel timer+ cancelLoop (if success then n + 1 else n) timers+ timers <- registerLoop [] (Random.mkStdGen 0) 0+ canceled <- cancelLoop 0 timers+ fired <- readIORef firedRef+ (fired + canceled) `is` (1_000_000 :: Int)+ data Bye = Bye- deriving stock (Show)+ deriving stock (Eq, Show) deriving anyclass (Exception) +newtype Exn a = Exn a+ deriving stock (Eq)+ class Assert a b where is :: a -> b -> IO () +instance (Eq e, Exception e) => Assert (IO void) (Exn e) where+ is mx y = do+ try (void mx) >>= \case+ Left ex+ | Exn ex == y -> pure ()+ | otherwise -> throwIO (userError ("expected different exception, but got: " ++ show ex))+ Right _ -> throwIO (userError "expected exception")+ instance (Eq a, Show a) => Assert (IO a) a where is mx y = do x <- mx+ unless (x == y) (throwIO (userError (show x ++ " /= " ++ show y)))++instance (Eq a, Show a) => Assert a a where+ is x y = unless (x == y) (throwIO (userError (show x ++ " /= " ++ show y)))
timer-wheel.cabal view
@@ -1,71 +1,67 @@-cabal-version: 2.0+cabal-version: 3.0 -name: timer-wheel-version: 0.4.0.1+author: Mitchell Rosen+bug-reports: https://github.com/awkward-squad/timer-wheel/issues+build-type: Simple category: Data description:- This library provides a timer wheel data structure for- .- * (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- timers, can scale to handle many more registered timers.+ This library provides a timer wheel data structure for registering one-shot or recurring @IO@ actions to fire after a+ given amount of time. .-synopsis: A timer wheel-author: Mitchell Rosen-maintainer: Mitchell Rosen <mitchellwrosen@gmail.com>-homepage: https://github.com/mitchellwrosen/timer-wheel-bug-reports: https://github.com/mitchellwrosen/timer-wheel/issues-copyright: (c) 2018-2020, Mitchell Rosen-license: BSD3+ It is similar to @TimerManager@ from @GHC.Event@, but supports recurring actions, and can scale to handle many more+ registered actions.+copyright: (c) 2018-2023 Mitchell Rosen, Travis Staton+homepage: https://github.com/awkward-squad/timer-wheel license-file: LICENSE-build-type: Simple-tested-with: GHC == 9.0.2, GHC == 9.2.4, GHC == 9.4.2+license: BSD-3-Clause+maintainer: Mitchell Rosen <mitchellwrosen@gmail.com>, Travis Staton <hello@travisstaton.com>+name: timer-wheel+synopsis: A timer wheel+tested-with: GHC == 9.4.7, GHC == 9.6.3, GHC == 9.8.1+version: 1.0.0 extra-source-files: CHANGELOG.md README.md- cabal.project source-repository head type: git- location: git://github.com/mitchellwrosen/timer-wheel.git+ location: git://github.com/awkward-squad/timer-wheel.git -library+common component build-depends:- array ^>= 0.5.2.0,- atomic-primops ^>= 0.8,- base ^>= 4.12 || ^>= 4.13 || ^>= 4.14 || ^>= 4.15 || ^>= 4.16 || ^>= 4.17,- ki ^>= 1.0.0,- psqueues ^>= 0.2.7-+ base ^>= 4.12 || ^>= 4.13 || ^>= 4.14 || ^>= 4.15 || ^>= 4.16 || ^>= 4.17 || ^>= 4.18 || ^>= 4.19, default-extensions:+ BangPatterns BlockArguments DeriveAnyClass DeriveGeneric DerivingStrategies+ DuplicateRecordFields+ FlexibleInstances GeneralizedNewtypeDeriving LambdaCase+ MultiParamTypeClasses+ MultiWayIf NamedFieldPuns- RecursiveDo+ NoImplicitPrelude+ NumericUnderscores+ OverloadedStrings ScopedTypeVariables TupleSections TypeApplications ViewPatterns-- default-language:- Haskell2010-- exposed-modules:- TimerWheel-+ default-language: Haskell2010 ghc-options: -Weverything -Wno-all-missed-specialisations -Wno-implicit-prelude -Wno-missing-import-lists -Wno-unsafe+ -- Buggy false-positives on unused-top-binds+ if impl(ghc == 8.6.*) || impl(ghc == 8.8.*)+ ghc-options:+ -Wno-unused-top-binds if impl(ghc >= 8.10) ghc-options: -Wno-missing-safe-haskell-mode@@ -73,36 +69,45 @@ if impl(ghc >= 9.2) ghc-options: -Wno-missing-kind-signatures-- hs-source-dirs:- src+ if impl(ghc >= 9.8)+ ghc-options:+ -Wno-missing-role-annotations +library+ import: component+ build-depends:+ atomic-primops ^>= 0.8,+ ki ^>= 1.0.0,+ primitive ^>= 0.7 || ^>= 0.8 || ^>= 0.9,+ exposed-modules:+ TimerWheel+ hs-source-dirs: src other-modules:- TimerWheel.Internal.Config- TimerWheel.Internal.Entries- TimerWheel.Internal.Micros- TimerWheel.Internal.Supply+ TimerWheel.Internal.Bucket+ TimerWheel.Internal.Counter+ TimerWheel.Internal.Nanoseconds+ TimerWheel.Internal.Prelude TimerWheel.Internal.Timestamp- TimerWheel.Internal.Wheel test-suite tests+ import: component build-depends:- base,- timer-wheel-- default-language:- Haskell2010-- ghc-options:- -threaded- -with-rtsopts=-N2- -Wall-- hs-source-dirs:- test-- main-is:- Main.hs+ ki,+ random ^>= 1.2,+ timer-wheel,+ ghc-options: -threaded -with-rtsopts=-N2+ hs-source-dirs: test+ main-is: Main.hs+ type: exitcode-stdio-1.0 - type:- exitcode-stdio-1.0+benchmark bench+ import: component+ build-depends:+ ki,+ random ^>= 1.2,+ tasty-bench,+ timer-wheel,+ ghc-options: -fproc-alignment=64 -rtsopts -threaded "-with-rtsopts=-N2 -T -A32m"+ hs-source-dirs: bench+ main-is: Main.hs+ type: exitcode-stdio-1.0