diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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` diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -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: diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # `timer-wheel` -[![GitHub CI](https://github.com/mitchellwrosen/timer-wheel/workflows/CI/badge.svg)](https://github.com/mitchellwrosen/timer-wheel/actions) +[![GitHub CI](https://github.com/awkward-squad/timer-wheel/workflows/Haskell-CI/badge.svg)](https://github.com/awkward-squad/timer-wheel/actions) [![Hackage](https://img.shields.io/hackage/v/timer-wheel.svg)](https://hackage.haskell.org/package/timer-wheel) [![Stackage LTS](https://stackage.org/package/timer-wheel/badge/lts)](https://www.stackage.org/lts/package/timer-wheel) [![Stackage Nightly](https://stackage.org/package/timer-wheel/badge/nightly)](https://www.stackage.org/nightly/package/timer-wheel) diff --git a/bench/Main.hs b/bench/Main.hs new file mode 100644 --- /dev/null +++ b/bench/Main.hs @@ -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 ()) diff --git a/cabal.project b/cabal.project deleted file mode 100644 --- a/cabal.project +++ /dev/null @@ -1,1 +0,0 @@ -packages: . diff --git a/src/TimerWheel.hs b/src/TimerWheel.hs --- a/src/TimerWheel.hs +++ b/src/TimerWheel.hs @@ -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 diff --git a/src/TimerWheel/Internal/Bucket.hs b/src/TimerWheel/Internal/Bucket.hs new file mode 100644 --- /dev/null +++ b/src/TimerWheel/Internal/Bucket.hs @@ -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 diff --git a/src/TimerWheel/Internal/Config.hs b/src/TimerWheel/Internal/Config.hs deleted file mode 100644 --- a/src/TimerWheel/Internal/Config.hs +++ /dev/null @@ -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) diff --git a/src/TimerWheel/Internal/Counter.hs b/src/TimerWheel/Internal/Counter.hs new file mode 100644 --- /dev/null +++ b/src/TimerWheel/Internal/Counter.hs @@ -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_ #-} diff --git a/src/TimerWheel/Internal/Entries.hs b/src/TimerWheel/Internal/Entries.hs deleted file mode 100644 --- a/src/TimerWheel/Internal/Entries.hs +++ /dev/null @@ -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 #-} diff --git a/src/TimerWheel/Internal/Micros.hs b/src/TimerWheel/Internal/Micros.hs deleted file mode 100644 --- a/src/TimerWheel/Internal/Micros.hs +++ /dev/null @@ -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) diff --git a/src/TimerWheel/Internal/Nanoseconds.hs b/src/TimerWheel/Internal/Nanoseconds.hs new file mode 100644 --- /dev/null +++ b/src/TimerWheel/Internal/Nanoseconds.hs @@ -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)) diff --git a/src/TimerWheel/Internal/Prelude.hs b/src/TimerWheel/Internal/Prelude.hs new file mode 100644 --- /dev/null +++ b/src/TimerWheel/Internal/Prelude.hs @@ -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 diff --git a/src/TimerWheel/Internal/Supply.hs b/src/TimerWheel/Internal/Supply.hs deleted file mode 100644 --- a/src/TimerWheel/Internal/Supply.hs +++ /dev/null @@ -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 #-} diff --git a/src/TimerWheel/Internal/Timestamp.hs b/src/TimerWheel/Internal/Timestamp.hs --- a/src/TimerWheel/Internal/Timestamp.hs +++ b/src/TimerWheel/Internal/Timestamp.hs @@ -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 diff --git a/src/TimerWheel/Internal/Wheel.hs b/src/TimerWheel/Internal/Wheel.hs deleted file mode 100644 --- a/src/TimerWheel/Internal/Wheel.hs +++ /dev/null @@ -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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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))) diff --git a/timer-wheel.cabal b/timer-wheel.cabal --- a/timer-wheel.cabal +++ b/timer-wheel.cabal @@ -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 -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 , Travis Staton +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