diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,103 @@ # ChangeLog for time-manager +## 0.5.2 + +* A timeout now fires on Windows under the old I/O manager, which is + still the default there. The timer was `GHC.Event.Windows`, WinIO's + own, so nothing ran the action unless the program used WinIO: warp + timed no connection out, and neither did anything else on top. + [#1139](https://github.com/yesodweb/wai/issues/1139) +* `tickle` extends a timeout by microseconds on Windows as it does + everywhere else. It floored the value to a whole second there, which + meant a timeout under a second was never extended at all. + +## 0.5.1 + +* A renewed timeout compared nanoseconds with microseconds, so a timeout + renewed after its callback had been queued ran at once instead of being + rescheduled for the time it had just been given. Wrong since 0.4.0. + [#1134](https://github.com/yesodweb/wai/pull/1134) + +## 0.5.0 + +* BREAKING CHANGE: a `ThreadManager` no longer carries a `T.Manager`. + `newThreadManager` takes no argument, `forkManagedTimeout` and + `forkManagedTimeoutFinally` take the caller's, and `withHandle` and the + re-exports from `System.TimeManager` are gone. + [#1131](https://github.com/yesodweb/wai/pull/1131) +* Count a thread in `System.ThreadManager` from when it is forked rather + than from when it starts, with `forkManagedBy`, `reserveManaged`, + `takeReservation` and `countManaged`. + [#1128](https://github.com/yesodweb/wai/pull/1128) +* Keep counting the threads `stopAfter` stopped until they end. + [#1128](https://github.com/yesodweb/wai/pull/1128) +* `countManaged` throws `AssertionFailed` when a reservation has been taken + by a manager other than its own. + [#1128](https://github.com/yesodweb/wai/pull/1128) +* `forkManagedFinally` and `forkManagedTimeoutFinally` no longer swallow an + exception from the action; it ends the thread, after the cleanup has run. + [#1128](https://github.com/yesodweb/wai/pull/1128) +* Add `forkManagedUnmaskFinally`, so that every fork has a variant with a + cleanup. + [#1128](https://github.com/yesodweb/wai/pull/1128) + +## 0.4.0 + +* CHANGES IN BEHAVIOUR: + * `tickle` is rate-limited/debounced. The renewal is skipped unless a quarter + of the timeout (capped at one second) has passed since the timeout was last + registered or updated. This does mean a timeout _might_ run a bit earlier + than the last `tickle` would indicate, but never more than the maximum + debounce period. + * `cancel` completely stops the timeout, making it un`resume`able. + `resume` will only resume a timeout that has been `pause`d. + * Prior to this major version, the `Handle` could be reused to run more + timeout actions. Now, a timeout action will only ever run, at most, once. + After a timeout action has run, the `Handle` is turned off and won't be + `resume`able, necessitating a call to `register` to start a new timeout. + + [#1109](https://github.com/yesodweb/wai/pull/1109) + +## 0.3.2 + +* Add `stopAfterWithResult`. [#1069](https://github.com/yesodweb/wai/pull/1069) + +## 0.3.1.1 + +* Added `README.md` file and improved documentation of modules and functions. + [#1057](https://github.com/yesodweb/wai/pull/1057) + +## 0.3.1 + +BUGFIXES: + +* [#1055](https://github.com/yesodweb/wai/pull/1055) + * `resume` now acts as a `tickle` if the `Handle` isn't paused. + This is the same behaviour as before version `0.3.0`. + * `registerKillThread` now throws the `TimeoutThread` in a separate + thread, so as to not block the GHC's System TimerManager. + This does mean that `TimeoutThread` exceptions could technically + be thrown "out of order", but they will be more prompt. + + +## 0.3.0 + +* [#1048](https://github.com/yesodweb/wai/pull/1048) + * New architecture. The backend is switched from the designated thread + to GHC's System TimerManager. From this version, this library is + just wrapper APIs of GHC's System TimerManager. Unlike v0.2 or + earlier, callbacks are executed at the exact time. System + TimerManager uses a PSQ (a tree) while v0.2 or earlier uses a list. + So, this version hopefully scales better. + * Deprecated functions: `stopManager`, `killManager` and `withManager'`. + * `tickle` sets the specified timeout from now. + * `pause` is now identical to `cancel`. + * `resume` is now re-registration of timeout. + * The signature of `withHandle` is changed. + +This change also means that using this package only works with the threaded runtime. +The moment a timeout is registered on a non-threaded runtime, an exception will be thrown. + ## 0.2.4 * Providing `isAllGone`. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,13 @@ +## time-manager + +This package provides module to let you run actions with resettable timeouts +(i.e. `System.TimeManager`) and run actions that will make sure that all +threads forked with the given `ThreadManager` will be killed when the action +finishes. + +## WARNINGS + +Since version `0.3.0`, **the timeout manager relies on GHC internals**. +This change also means that using **this package only works with the threaded** +**runtime**. The moment a timeout is registered on a non-threaded runtime, an +exception will be thrown. diff --git a/System/ThreadManager.hs b/System/ThreadManager.hs --- a/System/ThreadManager.hs +++ b/System/ThreadManager.hs @@ -1,42 +1,58 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} --- | A thread manager including a time manager. +-- | A thread manager. -- The manager has responsibility to kill managed threads. +-- +-- Because this is based on the accompanying "System.TimeManager" module, +-- the same caveats apply: +-- +-- * Only works for GHC. +-- * Only works with a threaded runtime. +-- * Users of older versions should check the current semantics. +-- * Using 32-bit systems means the max timeout is @'maxBound' :: Int@ +-- (2147483647) microseconds, which is less than 36 minutes. +-- * Using the same 'Handle' in different threads might cause issues in some +-- edge cases. (i.e. using cancel/pause in one thread, and resume in another) module System.ThreadManager ( ThreadManager, newThreadManager, stopAfter, + stopAfterWithResult, KilledByThreadManager (..), -- * Fork + Reservation, + reserveManaged, + takeReservation, forkManaged, forkManagedFinally, forkManagedUnmask, + forkManagedUnmaskFinally, + forkManagedBy, + forkManagedByFinally, forkManagedTimeout, forkManagedTimeoutFinally, -- * Synchronization waitUntilAllGone, isAllGone, - - -- * Re-exports - T.Manager, - withHandle, - T.Handle, - T.tickle, - T.pause, - T.resume, + countManaged, ) where import Control.Concurrent import Control.Concurrent.STM -import Control.Exception (Exception (..), SomeException (..)) +import Control.Exception ( + AssertionFailed (..), + Exception (..), + SomeException (..), + ) import qualified Control.Exception as E -import Control.Monad (unless, void) +import Control.Monad (unless, void, when) import Data.Foldable (forM_) -import Data.IORef +import Data.IORef (IORef, atomicModifyIORef', newIORef) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map import Data.Word (Word64) @@ -49,31 +65,89 @@ ---------------------------------------------------------------- --- | Manager to manage the thread and the timer. -data ThreadManager = ThreadManager T.Manager (TVar ManagedThreads) +-- | Manager to own threads. +data ThreadManager = ThreadManager + { managedThreads :: TVar ManagedThreads + , reservedPlaces :: TVar Int + -- ^ Places kept by 'reserveManaged' for threads not started yet. + , managerStopped :: TVar Bool + -- ^ Stopped by 'stopAfter'. A thread which registers after that + -- is killed on arrival: it was forked before the stop and would + -- otherwise run on with nobody owning it. + } type Key = Word64 type ManagedThreads = Map Key ManagedThread ---------------------------------------------------------------- --- 'IORef' prevents race between WAI TimeManager (TimeoutThread) --- and stopAfter (KilledByThreadManager). +-- The 'IORef' prevents a race between the timeout of +-- 'forkManagedTimeout' and 'forkManagedTimeoutFinally', which the time +-- manager runs, and 'stopAfter'. Both throw 'KilledByThreadManager'. -- It is initialized with 'False' and turned into 'True' when locked. -- The winner can throw an asynchronous exception. data ManagedThread = ManagedThread (Weak ThreadId) (IORef Bool) ---------------------------------------------------------------- --- | Starting a thread manager. --- Its action is initially set to 'return ()' and should be set --- by 'setAction'. This allows that the action can include --- the manager itself. -newThreadManager :: T.Manager -> IO ThreadManager -newThreadManager timmgr = ThreadManager timmgr <$> newTVarIO Map.empty +-- | Create a thread manager. +-- +-- A manager owns threads; it does not time them out. A caller which wants +-- both passes its own 'T.Manager' to 'forkManagedTimeout'. +newThreadManager :: IO ThreadManager +newThreadManager = do + managedThreads <- newTVarIO Map.empty + reservedPlaces <- newTVarIO 0 + managerStopped <- newTVarIO False + return ThreadManager{..} ---------------------------------------------------------------- +-- | A place kept for a thread that is not running yet. +-- +-- Registering a thread needs its 'ThreadId', and it is the thread +-- itself which provides it, by asking for its own. A caller which +-- forks through a function that hands no identity back cannot do it on +-- the thread's behalf, so until the thread arrives and registers +-- itself, what is counted is the place kept for it. +newtype Reservation = Reservation (IORef Bool) + +-- | Keeping a place for a thread about to be forked. It counts towards +-- 'countManaged' from here, so that a manager stopped in between does +-- not see a connection that has been accepted as one that is gone. +-- A place that is never taken is never given back, so a caller which +-- reserves and then fails to fork leaves the manager counting it. +reserveManaged :: ThreadManager -> IO Reservation +reserveManaged ThreadManager{..} = do + atomically $ modifyTVar' reservedPlaces (+ 1) + Reservation <$> newIORef False + +-- | Running the action of a thread which was reserved for, taking the +-- place kept for it: the thread records itself, so that 'stopAfter' can +-- reach it, and gives the reservation back. +-- +-- The reservation is given back only once the thread is recorded, so +-- that it is counted throughout. +-- A thread which gets here after 'stopAfter' has stopped the manager is +-- killed rather than registered. +-- +-- The reservation must be one kept by this manager. Taking one kept by +-- another lowers this manager's count instead of that one's: the other +-- manager never reaches zero, and once this count goes negative, +-- 'countManaged', 'isAllGone' and 'waitUntilAllGone' throw 'AssertionFailed'. +takeReservation :: ThreadManager -> Reservation -> IO () -> IO () +takeReservation mgr rsv io = taking mgr rsv (return ()) $ \_ -> io + +-- | Giving a place back, at most once however often this is called: the +-- thread which takes it gives it back, and so does a caller whose fork +-- throws, and only one of them can have it. +release :: ThreadManager -> Reservation -> IO () +release ThreadManager{..} (Reservation ref) = do + taken <- atomicModifyIORef' ref (\b -> (True, b)) + unless taken $ atomically $ modifyTVar' reservedPlaces (subtract 1) + +---------------------------------------------------------------- + -- | An exception used internally to kill a managed thread. data KilledByThreadManager = KilledByThreadManager (Maybe SomeException) deriving (Show) @@ -84,33 +158,76 @@ -- | Stopping the manager. -- +-- @ +-- stopAfter threadManager action cleanup +-- @ +-- -- The action is run in the scope of an exception handler that catches all -- exceptions (including asynchronous ones); this allows the cleanup handler -- to cleanup in all circumstances. If an exception is caught, it is rethrown -- after the cleanup is complete. +-- +-- This tells the managed threads to stop; it does not wait for them to be +-- gone, and neither does the cleanup handler. They are the manager\'s +-- until they end, so 'countManaged' goes on counting them -- which is what +-- a caller waits on with 'waitUntilAllGone' before it frees what those +-- threads were using. +-- +-- A thread whose place is still being kept by 'reserveManaged' is not told +-- but refused when it arrives, and never runs. +-- +-- A thread which swallows the exception and never ends is therefore +-- counted for ever, and 'waitUntilAllGone' on such a manager never +-- returns. Nothing here can make it end; a caller which cannot rule that +-- out should put its own timeout around the wait. stopAfter :: ThreadManager -> IO a -> (Maybe SomeException -> IO ()) -> IO a -stopAfter (ThreadManager _timmgr var) action cleanup = do +stopAfter mgr action cleanup = + stopAfterWithResult mgr action $ \mResult -> + case mResult of + Left err -> do + cleanup (Just err) + E.throwIO err + Right result -> do + cleanup Nothing + return result + +-- | Generalization of 'stopAfter' where the cleanup handler is allowed to construct the final result +-- +-- Unlike in 'stopAfter', if an exception is thrown, it is not re-thrown after the cleanup +-- handler completes; the cleanup handler itself can decide to rethrow it or compute a result. +-- +-- @since 0.3.2 +stopAfterWithResult + :: ThreadManager -> IO a -> (Either SomeException a -> IO b) -> IO b +stopAfterWithResult ThreadManager{..} action cleanup = do E.mask $ \unmask -> do ma <- E.try $ unmask action + -- The threads are left in the map rather than taken out of it: + -- 'throwTo' waits for the exception to be taken, not for the + -- thread to be gone, and a manager which reports that it owns + -- nothing while its threads are still unwinding is no use to a + -- caller which frees what they were using. They should take + -- themselves out as they end, and 'managerStopped' is what keeps + -- new ones from arriving. m <- atomically $ do - m0 <- readTVar var - writeTVar var Map.empty - return m0 + writeTVar managerStopped True + readTVar managedThreads let ths = Map.elems m er = either Just (const Nothing) ma ex = KilledByThreadManager er forM_ ths $ \(ManagedThread wtid ref) -> lockAndKill wtid ref ex - case ma of - Left err -> cleanup (Just err) >> E.throwIO err - Right a -> cleanup Nothing >> return a + cleanup ma ---------------------------------------------------------------- -- | Fork a managed thread. -- --- This guarantees that the thread ID is added to the manager's queue before --- the thread starts, and is removed again when the thread terminates --- (normally or abnormally). +-- A place is kept for the thread before it is forked, so that it counts +-- towards 'countManaged' from this call rather than from the moment the +-- run-time system gives the thread its first step. The thread adds +-- itself to the manager's queue as the first thing it does, takes the +-- place over, and is removed again when it terminates (normally or +-- abnormally). forkManaged :: ThreadManager -> String @@ -118,52 +235,189 @@ -> IO () -- ^ Action -> IO () -forkManaged mgr label io = - forkManagedUnmask mgr label $ \unmask -> unmask io +forkManaged mgr label io = forkManagedFinally mgr label io $ return () --- | Like 'forkManaged', but run action with exceptions masked +-- | Like 'forkManaged', but with a cleanup function, run regardless of how +-- the action ends. An exception from the action is not caught: it ends +-- the thread as it would under 'forkManaged'. +forkManagedFinally + :: ThreadManager + -> String + -- ^ Thread name + -> IO () + -- ^ Action + -> IO () + -- ^ Cleanup function + -> IO () +forkManagedFinally mgr label io final = + forkManagedUnmaskFinally mgr label (\restore -> restore io) final + +-- | Like 'forkManaged', but run the action with exceptions masked +-- until the provided `unmask` is used. (i.e. @(forall x. IO x -> IO x)@) forkManagedUnmask - :: ThreadManager -> String -> ((forall x. IO x -> IO x) -> IO ()) -> IO () -forkManagedUnmask (ThreadManager _timmgr var) label io = - void $ E.mask_ $ forkIOWithUnmask $ \unmask -> E.handle ignore $ do - labelMe label - E.bracket (setup var) (clear var) $ \_ -> io unmask + :: ThreadManager + -> String + -- ^ Thread name + -> ((forall x. IO x -> IO x) -> IO ()) + -- ^ Action with unmask argument + -> IO () +forkManagedUnmask mgr label io = forkManagedUnmaskFinally mgr label io $ return () --- | Fork a managed thread with a handle created by a timeout manager. -forkManagedTimeout :: ThreadManager -> String -> (T.Handle -> IO ()) -> IO () -forkManagedTimeout (ThreadManager timmgr var) label io = - void $ forkIO $ E.handle ignore $ do - labelMe label - E.bracket (setup var) (clear var) $ \(_n, wtid, ref) -> - -- 'TimeoutThread' is ignored by 'withHandle'. - void $ T.withHandle timmgr (lockAndKill wtid ref T.TimeoutThread) io +-- | Like 'forkManagedUnmask', but with a cleanup function, run regardless of +-- how the action ends. An exception from the action is not caught: it +-- ends the thread as it would under 'forkManagedUnmask'. +forkManagedUnmaskFinally + :: ThreadManager + -> String + -- ^ Thread name + -> ((forall x. IO x -> IO x) -> IO ()) + -- ^ Action with unmask argument + -> IO () + -- ^ Cleanup: will always run, regardless of how the thread ends + -> IO () +forkManagedUnmaskFinally mgr label io final = + E.mask_ $ + forkManagedByFinally mgr (\body -> void $ forkIOWithUnmask body) label io final --- | Fork a managed thread with a cleanup function. -forkManagedFinally :: ThreadManager -> String -> IO () -> IO () -> IO () -forkManagedFinally mgr label io final = E.mask $ \restore -> - forkManaged - mgr - label - (E.try (restore io) >>= \(_ :: Either E.SomeException ()) -> final) +-- | Fork a managed thread with a fork of the caller's own, for a caller +-- which does not fork with 'forkIO'. (The @warp@ library, for example, +-- forks the thread serving a connection through a function of its +-- settings, so that the user can replace it.) +-- +-- A place is kept for the thread before the fork, so that it counts +-- towards 'countManaged' from here rather than from the moment the +-- thread starts, and the place is given back whether the fork makes the +-- thread or throws. A caller which forks by itself can do the same +-- with 'reserveManaged' and 'takeReservation', which this is made of. +forkManagedBy + :: ThreadManager + -> (((forall x. IO x -> IO x) -> IO ()) -> IO a) + -- ^ The fork to make the thread with + -> String + -- ^ Thread name + -> ((forall x. IO x -> IO x) -> IO ()) + -- ^ Action with unmask argument + -> IO a +forkManagedBy mgr fork label io = forkManagedByFinally mgr fork label io $ return () --- | Fork a managed thread with a handle created by a timeout manager --- and with a cleanup function. +-- | Like 'forkManagedBy', but with a cleanup function. +forkManagedByFinally + :: ThreadManager + -> (((forall x. IO x -> IO x) -> IO ()) -> IO a) + -- ^ The fork to make the thread with + -> String + -- ^ Thread name + -> ((forall x. IO x -> IO x) -> IO ()) + -- ^ Action with unmask argument + -> IO () + -- ^ Cleanup: will always run, regardless of how the thread ends + -> IO a +forkManagedByFinally mgr fork label io final = + E.bracketOnError (reserveManaged mgr) (release mgr) $ \rsv -> + fork $ \unmask -> managed mgr rsv label final $ \_ -> io unmask + +-- | Fork a managed thread with a handle created by a timeout manager. +-- +-- The time manager is the caller's, since this manager has none: owning a +-- thread and timing it out are separate things, and only this pair of +-- functions needs both. What needs them together is the lock that keeps +-- the timeout and 'stopAfter' from both throwing at the thread. +forkManagedTimeout + :: ThreadManager + -> T.Manager + -- ^ Time manager + -> String + -- ^ Thread name + -> (T.Handle -> IO ()) + -- ^ Action with timeout handle + -> IO () +forkManagedTimeout mgr timmgr label io = + forkManagedTimeoutFinally mgr timmgr label io $ return () + +-- | Like 'forkManagedTimeout', but with a cleanup function, run regardless of +-- how the action ends. An exception from the action is not caught: it +-- ends the thread as it would under 'forkManagedTimeout'. forkManagedTimeoutFinally - :: ThreadManager -> String -> (T.Handle -> IO ()) -> IO () -> IO () -forkManagedTimeoutFinally mgr label io final = E.mask $ \restore -> - forkManagedTimeout - mgr - label - (\th -> E.try (restore $ io th) >>= \(_ :: Either E.SomeException ()) -> final) + :: ThreadManager + -> T.Manager + -- ^ Time manager + -> String + -- ^ Thread name + -> (T.Handle -> IO ()) + -- ^ Action with timeout handle + -> IO () + -- ^ Cleanup: will always run, regardless of how the thread ends + -> IO () +forkManagedTimeoutFinally mgr timmgr label io final = + E.bracketOnError (reserveManaged mgr) (release mgr) $ \rsv -> + void $ forkIO $ + managed mgr rsv label final $ \(_n, ManagedThread wtid ref) -> + T.withHandle timmgr (lockAndKill wtid ref ex) io + where + ex = KilledByThreadManager Nothing -setup :: TVar (Map Key ManagedThread) -> IO (Key, Weak ThreadId, IORef Bool) -setup var = do - (wtid, n) <- myWeakThradId +-- | The body of a managed thread, under its own label, taking the place +-- kept for it. +managed + :: ThreadManager + -> Reservation + -> String + -> IO () + -> ((Key, ManagedThread) -> IO ()) + -> IO () +managed mgr rsv label final body = do + labelMe label + taking mgr rsv final body + +-- | Running the body of a thread which had a place kept for it. The +-- place is given back only once the thread is recorded, so that it is +-- counted throughout. +taking + :: ThreadManager + -> Reservation + -> IO () + -> ((Key, ManagedThread) -> IO ()) + -> IO () +taking mgr rsv final body = + recorded mgr (give >> final) $ \key -> give >> body key + where + give = release mgr rsv + +-- | Running the body of a managed thread with the thread recorded. +-- +-- The cleanup runs regardless of how the body ends, and while the thread is still +-- recorded, so that a manager which has none left has none whose cleanup +-- is still running either. +-- +-- A thread forked before the manager was stopped and scheduled after it +-- is not recorded and does not run its body: there is nobody left to +-- kill it, and running it is what the stop was for. +recorded + :: ThreadManager + -> IO () + -> ((Key, ManagedThread) -> IO ()) + -> IO () +recorded mgr final body = + E.handle ignore $ + E.bracket (setup mgr) (mapM_ $ clear mgr) $ \mkey -> + mapM_ body mkey `E.finally` final + +-- | Recording this thread, or 'Nothing' once the manager has stopped. +setup + :: ThreadManager + -> IO (Maybe (Key, ManagedThread)) +setup ThreadManager{..} = do + (wtid, n) <- myWeakThreadId ref <- newIORef False let ent = ManagedThread wtid ref -- asking to throw KilledByThreadManager to me - atomically $ modifyTVar' var $ Map.insert n ent - return (n, wtid, ref) + atomically $ do + stopped <- readTVar managerStopped + if stopped + then return Nothing + else do + modifyTVar' managedThreads $ Map.insert n ent + return $ Just (n, ent) lockAndKill :: Exception e => Weak ThreadId -> IORef Bool -> e -> IO () lockAndKill wtid ref e = do @@ -175,29 +429,57 @@ Just tid -> E.throwTo tid e clear - :: TVar (Map Key ManagedThread) - -> (Key, Weak ThreadId, IORef Bool) + :: ThreadManager + -> (Key, ManagedThread) -> IO () -clear var (n, _, _) = atomically $ modifyTVar' var $ Map.delete n +clear ThreadManager{..} (n, _) = + atomically $ modifyTVar' managedThreads $ Map.delete n ignore :: KilledByThreadManager -> IO () ignore (KilledByThreadManager _) = return () --- | Wait until all managed thread are finished. +-- | Wait until all managed threads are finished. +-- +-- After 'stopAfter' this waits for the threads it stopped to end, not +-- merely for them to have been told. See there for the thread that +-- never ends. +-- +-- A place kept by 'reserveManaged' and never taken is never given back, +-- so this never returns for a caller which reserves and then fails to +-- fork. waitUntilAllGone :: ThreadManager -> IO () -waitUntilAllGone (ThreadManager _timmgr var) = atomically $ do - m <- readTVar var - check (Map.size m == 0) +waitUntilAllGone tm = + atomically $ + isAllGone tm >>= check +-- | STM action that checks if all managed threads are finished. isAllGone :: ThreadManager -> STM Bool -isAllGone (ThreadManager _timmgr var) = do - m <- readTVar var - return (Map.size m == 0) +isAllGone tm = (== 0) <$> countManaged tm +-- | STM action for how many threads are managed, counting the places +-- kept by 'reserveManaged' for threads which have not started, and the +-- threads 'stopAfter' has stopped which have not ended yet. +-- +-- Throws 'AssertionFailed' if the count of places has gone negative, +-- which happens when a reservation is taken by a manager other than the +-- one that kept it. A mistake of that kind which does not drive the +-- count negative goes unnoticed. +countManaged :: ThreadManager -> STM Int +countManaged ThreadManager{..} = do + places <- readTVar reservedPlaces + when (places < 0) $ + throwSTM $ + AssertionFailed $ + "System.ThreadManager: " + ++ show places + ++ " reserved places; a reservation was taken by a" + ++ " manager other than its own" + (+ places) . Map.size <$> readTVar managedThreads + ---------------------------------------------------------------- -myWeakThradId :: IO (Weak ThreadId, Key) -myWeakThradId = do +myWeakThreadId :: IO (Weak ThreadId, Key) +myWeakThreadId = do tid <- myThreadId wtid <- mkWeakThreadId tid let n = fromThreadId tid @@ -207,10 +489,6 @@ labelMe l = do tid <- myThreadId labelThread tid l - -withHandle - :: ThreadManager -> T.TimeoutAction -> (T.Handle -> IO a) -> IO (Maybe a) -withHandle (ThreadManager timmgr _) = T.withHandle timmgr #if __GLASGOW_HASKELL__ < 908 fromThreadId :: ThreadId -> Word64 diff --git a/System/TimeManager.hs b/System/TimeManager.hs --- a/System/TimeManager.hs +++ b/System/TimeManager.hs @@ -1,7 +1,21 @@ -{-# LANGUAGE BangPatterns #-} -{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE RecordWildCards #-} +-- | Timeout manager. Since @v0.3.0@, timeout manager is a wrapper of +-- GHC System TimerManager, except on Windows, where it has a timer of +-- its own: the one in @GHC.Event.Windows@ belongs to WinIO and never +-- runs under the I\/O manager which is still the default there. +-- +-- Some caveats of using this package: +-- +-- * Only works for GHC +-- * Only works with a threaded runtime +-- * Users of older versions should check the current semantics. +-- * Using 32-bit systems means the max timeout is @'maxBound' :: Int@ +-- (2147483647) microseconds, which is less than 36 minutes. +-- * Using the same 'Handle' in different threads might cause issues in some +-- edge cases. (i.e. using 'cancel'/'pause' in one thread, and 'resume' in another) module System.TimeManager ( -- ** Types Manager, @@ -25,174 +39,396 @@ tickle, pause, resume, + cancel, -- ** Low level register, registerKillThread, - cancel, -- ** Exceptions TimeoutThread (..), ) where -import Control.Concurrent (mkWeakThreadId, myThreadId) +import Control.Concurrent (forkIO, mkWeakThreadId, myThreadId, newMVar) import qualified Control.Exception as E -import Control.Monad (void) -import Control.Reaper -import Data.IORef (IORef) +import Control.Monad (void, when) +import Data.Bits (shiftR) import qualified Data.IORef as I -import Data.Typeable (Typeable) -import System.IO.Unsafe +import Data.Word (Word64) +import GHC.Clock (getMonotonicTimeNSec) +import System.IO.Unsafe (unsafePerformIO) import System.Mem.Weak (deRefWeak) +import System.TimeManager.Internal ----------------------------------------------------------------- +#if defined(mingw32_HOST_OS) +import qualified System.TimeManager.Windows as EV +#else +import qualified GHC.Event as EV +#endif --- | A timeout manager -data Manager = Manager (Reaper [Handle] Handle) | NoManager +---------------------------------------------------------------- --- | No manager. +-- | A manager whose timeout value is 0 (no callbacks are fired). defaultManager :: Manager -defaultManager = NoManager - --- | An action to be performed on timeout. -type TimeoutAction = IO () - --- | A handle used by a timeout manager. -data Handle = Handle - { handleManager :: Manager - , handleActionRef :: IORef TimeoutAction - , handleStateRef :: IORef State - } - -{-# NOINLINE emptyAction #-} -emptyAction :: IORef TimeoutAction -emptyAction = unsafePerformIO $ I.newIORef (return ()) +defaultManager = Manager 0 -{-# NOINLINE emptyState #-} -emptyState :: IORef State -emptyState = unsafePerformIO $ I.newIORef Inactive +---------------------------------------------------------------- +-- | Dummy 'Handle'. emptyHandle :: Handle emptyHandle = Handle - { handleManager = NoManager - , handleActionRef = emptyAction - , handleStateRef = emptyState + { handleTimeout = 0 + , handleAction = pure () + , handleTimerManager = mutError "handleTimerManager" + , handleState = mutError "handleState" + , handleLastRenewed = mutError "handleLastRenewed" + , handleMinRenewGap = 0 + , handleLock = emptyLock } + where + mutError s = error $ "time-manager: Handle." <> s <> " not set" -data State - = Active -- Manager turns it to Inactive. - | Inactive -- Manager removes it with timeout action. - | Paused -- Manager does not change it. +emptyLock :: Lock +emptyLock = unsafePerformIO $ newMVar () +{-# NOINLINE emptyLock #-} ---------------------------------------------------------------- --- | Creating timeout manager which works every N microseconds --- where N is the first argument. +-- | Creating timeout manager with a timeout value in microseconds. +-- +-- Setting the timeout to zero or lower @(<= 0)@ will produce a +-- `defaultManager`. initialize :: Int -> IO Manager -initialize timeout - | timeout <= 0 = return NoManager -initialize timeout = - Manager - <$> mkReaper - defaultReaperSettings - { -- Data.Set cannot be used since 'partition' cannot be used - -- with 'readIORef`. So, let's just use a list. - reaperAction = mkListAction prune - , reaperDelay = timeout - , reaperThreadName = "WAI timeout manager (Reaper)" - } - where - prune m@Handle{..} = do - state <- I.atomicModifyIORef' handleStateRef (\x -> (inactivate x, x)) - case state of - Inactive -> do - onTimeout <- I.readIORef handleActionRef - onTimeout `E.catch` ignoreSync - return Nothing - _ -> return $ Just m - - inactivate Active = Inactive - inactivate x = x +initialize = pure . Manager . max 0 ---------------------------------------------------------------- --- | Stopping timeout manager with onTimeout fired. +-- | Obsoleted since version 0.3.0 +-- Is now equivalent to @pure ()@. stopManager :: Manager -> IO () -stopManager NoManager = return () -stopManager (Manager mgr) = E.mask_ (reaperStop mgr >>= mapM_ fire) - where - fire Handle{..} = do - onTimeout <- I.readIORef handleActionRef - onTimeout `E.catch` ignoreSync +stopManager _ = pure () +{-# DEPRECATED stopManager "This function does nothing since version 0.3.0" #-} --- | Killing timeout manager immediately without firing onTimeout. +-- | Obsoleted since version 0.3.0 +-- Is now equivalent to @pure ()@. killManager :: Manager -> IO () -killManager NoManager = return () -killManager (Manager mgr) = reaperKill mgr +killManager _ = pure () +{-# DEPRECATED killManager "This function does nothing since version 0.3.0" #-} ---------------------------------------------------------------- -- | Registering a timeout action and unregister its handle -- when the body action is finished. --- 'Nothing' is returned on timeout. -withHandle :: Manager -> TimeoutAction -> (Handle -> IO a) -> IO (Maybe a) -withHandle mgr onTimeout action = - E.handle ignore $ E.bracket (register mgr onTimeout) cancel $ \th -> - Just <$> action th - where - ignore TimeoutThread = return Nothing +withHandle :: Manager -> TimeoutAction -> (Handle -> IO a) -> IO a +withHandle mgr onTimeout action + | isNoManager mgr = action emptyHandle + | otherwise = E.bracket (register mgr onTimeout) cancel action -- | Registering a timeout action of killing this thread and -- unregister its handle when the body action is killed or finished. withHandleKillThread :: Manager -> TimeoutAction -> (Handle -> IO ()) -> IO () -withHandleKillThread mgr onTimeout action = - E.handle ignore $ E.bracket (registerKillThread mgr onTimeout) cancel action +withHandleKillThread mgr onTimeout action + | isNoManager mgr = action emptyHandle + | otherwise = + E.handle ignore $ E.bracket (registerKillThread mgr onTimeout) cancel action where - ignore TimeoutThread = return () + ignore TimeoutThread = pure () ---------------------------------------------------------------- +-- ============== NOTE ABOUT THREAD SAFETY ============== +-- +-- The use of 'IORef's are fine in the current situation where +-- the 'TimeManager' is supposed to be used in a single thread. +-- +-- The triggered action, though, is run by the Timer Manager +-- outside of the thread it was registered in. +-- This will potentially cause race conditions if we implement +-- anything that depends on the 'Handle's state. +-- +-- Given the following: +-- - If run in one thread: 'register/tickle/pause/resume/cancel' never +-- overlap, making them devoid of race conditions in the general sense. +-- - We want to hit the Timer Manager as little as possible. +-- - We want to keep the 'resume/pause' surface functionality intact, while +-- not hitting the Timer Manager when we don't have to. This means not +-- cancelling the timeout on a pause, but rather mark the timeout paused. +-- - Not actually stopping the timeout on 'pause' introduces race +-- conditions, because the registered action will need to check the 'Handle' +-- state to see whether it should actually run (Active) or if it should +-- drop the action (Paused/Stopped). +-- - Not hitting the Timer Manager on a 'pause' will increase performance on +-- hot 'resume/pause' loops, like 'warp' has when using a streaming response. +-- - 'tickle' gets a sort of debounce to avoid repeated updates in hot loops. +-- - The debounce is 1/4 of the timeout, but we cap it to a maximum of 1 second. +-- - This means the registered action might run earlier than the timeout +-- would indicate; that difference going up to a maximum of 'handleMinRenewGap'. +-- - The following can happen: +-- - == The "Surprise Active" issue == +-- A 'resume' might get called right after a 'Paused' registered action +-- starts running, and sets the state to 'Active' __before__ the action +-- inspects the 'Handle' state. +-- - == The "Dropped Active" issue == +-- A 'resume' might get called right after a 'Paused' registered action +-- starts running, but inspects the state __before__ the action sets the +-- state to 'Stopped', and the registered action inspects the state +-- __before__ the 'resume' has set it to 'Active'. Essentially missing +-- the 'resume' completely. +-- - == The "Dropped Cancel" issue == +-- A 'cancel' getting called right after a 'Paused' registered action +-- starts running, and cancelling __after__ the action reads the state +-- will have the registered action overwrite the state to 'Stopped', +-- when it shouldn't register a new action, but stop everything. +-- - A 'pause' should technically not be an issue, as it will only run when +-- the state is 'Active', but it is a function that changes the state, so +-- just to be cautious, we let it grab the lock. +-- - A 'tickle' in the same situation doesn't matter, as a 'tickle' +-- shouldn't activate a 'Paused' state. (and doesn't change any state) +-- - The "Surprise Active" issue can be mitigated by checking the +-- 'handleLastRenewed' time and reregistering the timeout action with the +-- remaining amount of microseconds in the case where it has not yet been +-- 'handleTimeout' amount of time. +-- - A 'tickle' could also cause this if the state was 'Active' all along, +-- but we'll accept the 'tickle' as being on time to extend the timeout. +-- - The "Dropped Active" issue is a bit more difficult to mitigate. We'll +-- need a lock to guarantee that either the activation of 'resume' is seen +-- by the registered action, or that the termination of the registered +-- action is seen by the 'resume'. +-- - The "Dropped Cancel" issue will also be avoided when using a lock. +-- - The lock will generally never be contested. It is there only for the +-- off-chance that a state-changing function runs JUST after the registered +-- action triggers. So in general, we don't expect the lock to reduce +-- performance noticeably. + +---------------------------------------------------------------- + -- | Registering a timeout action. register :: Manager -> TimeoutAction -> IO Handle -register NoManager _ = return emptyHandle -register m@(Manager mgr) !onTimeout = do - actionRef <- I.newIORef onTimeout - stateRef <- I.newIORef Active - let h = - Handle - { handleManager = m - , handleActionRef = actionRef - , handleStateRef = stateRef - } - reaperAdd mgr h - return h +register mgr@(Manager timeout) onTimeout + | isNoManager mgr = pure emptyHandle + | otherwise = do + -- The system timer manager is stable for the lifetime of the + -- process (and even if it were replaced, e.g. around a fork, + -- the key registered below would only be meaningful to the + -- manager it was registered with). So fetch it once here and + -- cache it in the 'Handle' instead of re-reading the global + -- IORef on every tickle/pause/resume. + sysmgr <- getTimerManager + stateRef <- I.newIORef Stopped + lock <- newLock + lastRenewedRef <- I.newIORef =<< getMonotonicTimeNSec + let h = + Handle + { handleTimeout = timeout + , handleAction = onTimeout + , handleTimerManager = sysmgr + , handleState = stateRef + , handleLastRenewed = lastRenewedRef + , handleMinRenewGap = minRenewGap timeout + , handleLock = lock + } + -- Just in case the timeout is only 1 microsecond and because of thread + -- scheduling it runs before we can change the state to 'Active' + withLock lock $ do + key <- registerAdjustedTimeout h timeout + now <- getMonotonicTimeNSec + I.writeIORef lastRenewedRef now + I.writeIORef stateRef $ Active key + pure h --- | Removing the 'Handle' from the 'Manager' immediately. +-- | This function needs a separate 'timeout' argument, because we might not +-- register the full amount of time when continuing a timeout that was started +-- a bit earlier. (cf. "Surprise Active" situation) +registerAdjustedTimeout :: Handle -> Int -> IO EV.TimeoutKey +registerAdjustedTimeout h@Handle{..} timeout = do + originalKeyRef <- + I.newIORef $ + error "System.TimeManager.registerAdjustedTimeout: originalKeyRef not filled" + key <- + EV.registerTimeout handleTimerManager timeout $ + adjustOnTimeout originalKeyRef h + I.writeIORef originalKeyRef key + pure key + +-- | Wrapper around a registered action to ensure correct handling. +-- +-- We basically need the 'Handle', but this is used before making the handle, so +adjustOnTimeout :: I.IORef EV.TimeoutKey -> Handle -> TimeoutAction +adjustOnTimeout originalKeyRef h@Handle{..} = do + let writeState = I.atomicWriteIORef handleState + -- Lock ensures we don't get race conditions. + -- We return a boolean so that we don't run the (potentially long) action + -- while holding on to the lock. + shouldRun <- withLock handleLock $ do + st <- I.readIORef handleState + case st of + -- We can check the @now - handleLastRenewed@ diff + -- and 'threadDelay' the diff to make the timing better? + -- @if diff > 'handleTimeout - 'handleMinRenewGap' then runTimeout@ + Active key -> do + -- set state ref to 'Active'? + ifSameKey key $ do + lastRenewed <- I.readIORef handleLastRenewed + now <- getMonotonicTimeNSec + let diff = fromIntegral $ (now - lastRenewed) `div` 1_000 + if diff >= handleTimeout + -- Valid expiration of the timeout, we run the action + then do + -- We're going to run the action, so set the state + -- so that it won't be resumed. + writeState Terminated + pure True + -- "Surprise Active" situation + else do + -- We reschedule, but with only the remaining time + let remainingTimeout = handleTimeout - diff + k <- registerAdjustedTimeout h remainingTimeout + writeState $ Active k + pure False + -- We find this action being run after it's been paused. We write + -- the state to 'Stopped' so that 'resume' knows to reregister the + -- timeout action. + Paused key -> + ifSameKey key $ do + writeState Stopped + pure False + -- 'Stopped' and 'Terminated' mean the action shouldn't run. + _ -> pure False + when shouldRun handleAction + where + -- If the key in the state isn't the same as the one this action + -- was registered with, then this action shouldn't run. + -- (Technically, this situation shouldn't happen. but since the registered + -- action only ever runs once, we can afford to be redundant) + ifSameKey key f = do + originalKey <- I.readIORef originalKeyRef + if key == originalKey + then f + else pure False + +-- | How long 'tickle' waits before actually renewing the timeout: +-- a quarter of the timeout, capped at one second. Skipping a renewal +-- inside this window only shortens the effective timeout by up to +-- this gap, but turns hot 'tickle' loops (one per chunk sent or +-- received) into a clock read and a comparison. +minRenewGap :: Int -> Word64 +minRenewGap timeout = + -- @shiftR 2 === divide by 4@ + min maxRenewDebounce (microToNano timeout `shiftR` 2) + where + microToNano = (* 1_000) . fromIntegral + +-- | One second in nanoseconds +maxRenewDebounce :: Word64 +maxRenewDebounce = 1_000_000_000 + +-- | Run 'f' if the minimum renew gap has been crossed. +whenRenew :: Handle -> IO () -> IO () +whenRenew h f = do + now <- getMonotonicTimeNSec + lastRenewed <- I.readIORef $ handleLastRenewed h + when (now - lastRenewed >= handleMinRenewGap h) f + +-- | Unregistering the timeout. +-- +-- The timeout can not be 'resume'd. To "resume" the timeout, you need to +-- 'register' again. cancel :: Handle -> IO () -cancel Handle{..} = case handleManager of - NoManager -> return () - Manager mgr -> void $ reaperModify mgr filt +cancel h@Handle{..} = + withNonEmptyHandle h $ + -- "Dropped Cancel" remedy + -- + -- We can eat a potential mutex pause here to avoid race conditions, + -- because we don't expect 'cancel' to be called in hot loops. + -- + -- (The race condition being: the 'Terminated' state being overwritten + -- because the 'cancel' runs JUST after the registered action starts + -- running, sets the state to 'Terminated', and then the registered + -- action finishes and overwrites it to 'Stopped') + withLock handleLock $ do + withTimeoutKey h $ EV.unregisterTimeout handleTimerManager + I.atomicWriteIORef handleState Terminated + +-- | Extending the timeout. +-- +-- To keep frequent callers cheap, the renewal is rate-limited: it is +-- skipped unless at least a quarter of the timeout (capped at one +-- second) has passed since the timeout was last registered or updated. +-- +-- Careful: this does NOT reactivate an already 'pause'd 'Handle'! +-- +tickle :: Handle -> IO () +tickle h@Handle{..} = + withNonEmptyHandle h $ + whenRenew h $ + withActiveTimeoutKey h $ \key -> do + updateTheTimeout key + now <- getMonotonicTimeNSec + I.atomicWriteIORef handleLastRenewed now where - -- It's very important that this function forces the whole workload so we - -- don't retain old handles, otherwise disasterous leaks occur. - filt [] = [] - filt (h@(Handle _ _ ref) : hs) - | handleStateRef == ref = hs - | otherwise = - let !hs' = filt hs - in h : hs' + updateTheTimeout key = + EV.updateTimeout handleTimerManager key handleTimeout +-- | Pauses the timeout so you can 'resume' it later. Does not stop it entirely. +-- Use 'cancel' if you want to make sure the action will not be resumed. +-- +-- To resume a timeout with the same 'Handle', 'resume' MUST be called. +-- Don't call 'tickle' for resumption. +pause :: Handle -> IO () +pause h@Handle{..} = + withNonEmptyHandle h $ + withLock handleLock . withActiveTimeoutKey h $ + I.atomicWriteIORef handleState . Paused + +-- | Resuming the timeout. +-- +-- Works like 'tickle' if the 'Handle' wasn't 'pause'd or 'cancel'ed. +resume :: Handle -> IO () +resume h@Handle{..} = + withNonEmptyHandle h $ + -- we ignore the key when paused, because we recheck the state after + -- grabbing the lock. + checkStateWith (\_ -> onPausedOrStopped) onPausedOrStopped + where + -- "Dropped Active" remedy + -- + -- Grabbing the lock ensures 'resume' runs either before or after the + -- registered action changes the state. + onPausedOrStopped = + withLock handleLock $ checkStateWith pausedF stoppedF + checkStateWith onPaused onStopped = do + state <- I.readIORef handleState + case state of + -- 'tickle' doesn't introduce race conditions, so can always be run. + Active{} -> tickle h + -- Abort when terminated. + Terminated -> pure () + Paused k -> onPaused k + Stopped -> onStopped + pausedF k = do + -- Set state to 'Active' before 'tickle'ing, because + -- 'tickle' only runs when the state is 'Active'. + activateTimeout k + tickle h + stoppedF = do + key <- registerAdjustedTimeout h handleTimeout + now <- getMonotonicTimeNSec + I.atomicWriteIORef handleLastRenewed now + activateTimeout key + activateTimeout = + I.atomicWriteIORef handleState . Active + ---------------------------------------------------------------- -- | The asynchronous exception thrown if a thread is registered via -- 'registerKillThread'. data TimeoutThread = TimeoutThread - deriving (Typeable) instance E.Exception TimeoutThread where toException = E.asyncExceptionToException fromException = E.asyncExceptionFromException + instance Show TimeoutThread where show TimeoutThread = "Thread killed by timeout manager" @@ -202,70 +438,31 @@ -- want to leak the asynchronous exception to GHC RTS. registerKillThread :: Manager -> TimeoutAction -> IO Handle registerKillThread m onTimeout = do - tid <- myThreadId - wtid <- mkWeakThreadId tid + wtid <- myThreadId >>= mkWeakThreadId -- First run the timeout action in case the child thread is masked. register m $ onTimeout `E.finally` do mtid <- deRefWeak wtid case mtid of - Nothing -> return () - Just tid' -> E.throwTo tid' TimeoutThread - ----------------------------------------------------------------- - --- | Setting the state to active. --- 'Manager' turns active to inactive repeatedly. -tickle :: Handle -> IO () -tickle Handle{..} = I.writeIORef handleStateRef Active - --- | Setting the state to paused. --- 'Manager' does not change the value. -pause :: Handle -> IO () -pause Handle{..} = I.writeIORef handleStateRef Paused - --- | Setting the paused state to active. --- This is an alias to 'tickle'. -resume :: Handle -> IO () -resume = tickle + Nothing -> pure () + Just tid' -> void . forkIO $ E.throwTo tid' TimeoutThread ---------------------------------------------------------------- -- | Call the inner function with a timeout manager. --- 'stopManager' is used after that. withManager :: Int -- ^ timeout in microseconds -> (Manager -> IO a) -> IO a -withManager timeout f = - E.bracket - (initialize timeout) - stopManager - f +withManager timeout f = initialize timeout >>= f -- | Call the inner function with a timeout manager. --- 'killManager' is used after that. +-- This is identical to 'withManager'. withManager' :: Int -- ^ timeout in microseconds -> (Manager -> IO a) -> IO a -withManager' timeout f = - E.bracket - (initialize timeout) - killManager - f - ----------------------------------------------------------------- - -isAsyncException :: E.Exception e => e -> Bool -isAsyncException e = - case E.fromException (E.toException e) of - Just (E.SomeAsyncException _) -> True - Nothing -> False - -ignoreSync :: E.SomeException -> IO () -ignoreSync se - | isAsyncException se = E.throwIO se - | otherwise = return () +withManager' = withManager +{-# DEPRECATED withManager' "This function is the same as 'withManager' since version 0.3.0" #-} diff --git a/System/TimeManager/Internal.hs b/System/TimeManager/Internal.hs new file mode 100644 --- /dev/null +++ b/System/TimeManager/Internal.hs @@ -0,0 +1,117 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE StrictData #-} + +module System.TimeManager.Internal where + +import Control.Concurrent.MVar (MVar, modifyMVar, newMVar) +import Data.IORef (IORef, readIORef) +import Data.Word (Word64) + +#if defined(mingw32_HOST_OS) +import qualified System.TimeManager.Windows as EV +#else +import qualified GHC.Event as EV +#endif + +---------------------------------------------------------------- + +-- | A timeout manager +newtype Manager = Manager Int + +isNoManager :: Manager -> Bool +isNoManager (Manager 0) = True +isNoManager _ = False + +---------------------------------------------------------------- + +-- | An action (callback) to be performed on timeout. +type TimeoutAction = IO () + +-- | A handle used by a timeout manager. +data Handle = Handle + { handleTimeout :: Int + , handleAction :: TimeoutAction + , handleTimerManager :: ~TimerManager + -- ^ The system timer manager the timeout key was registered with. + -- Cached so that per-request operations don't re-fetch it. + , handleLock :: Lock + -- ^ Used by 'resume', 'pause' and 'cancel' to determine race conditions. + -- + -- /We intentionally do not use an @MVar HandleState@ for performance reasons./ + -- /The lock only has to be grabbed to avoid race conditions./ + , handleState :: ~(IORef HandleState) + -- ^ The current state. Used to decide whether a timeout is still going, + -- paused, or completely terminated. + -- + -- /We intentionally do not use an @MVar HandleState@ for performance reasons./ + , handleLastRenewed :: ~(IORef Word64) + -- ^ Monotonic time (in nanoseconds) when the timeout was last + -- registered or updated. + , handleMinRenewGap :: Word64 + -- ^ 'tickle' is a no-op unless at least this many nanoseconds have + -- passed since the last renewal. + } + +-- | Makes sure the function is only run when there's a key to act on. +withTimeoutKey :: Handle -> (EV.TimeoutKey -> IO ()) -> IO () +withTimeoutKey h keyF = do + st <- readIORef $ handleState h + case st of + Paused key -> keyF key + Active key -> keyF key + _ -> pure () + +-- | Makes sure the function is only run when the state is 'Active'. +withActiveTimeoutKey :: Handle -> (EV.TimeoutKey -> IO ()) -> IO () +withActiveTimeoutKey h keyF = do + st <- readIORef $ handleState h + case st of + Active key -> keyF key + _ -> pure () + +-- | Used to avoid race conditions in situations when the state has to be changed. +type Lock = MVar () + +newLock :: IO Lock +newLock = newMVar () + +withLock :: Lock -> IO a -> IO a +withLock lock action = + -- Not sure whether this should be 'modifyMVarMasked' or not. + modifyMVar lock $ \l -> do + a <- action + pure (l, a) + +-- | Tracking the state of a handle. +data HandleState + = -- Timeout is primed to run + Active EV.TimeoutKey + | -- Timeout is paused, but still running + -- ('resume' will set it back to 'Active' and 'tickle') + Paused EV.TimeoutKey + | -- Action ran, but timeout was paused, so it is resumable + -- ('resume' will reregister the action) + Stopped + | -- Action was cancelled or run. 'register' is needed to start a new timeout. + Terminated + +isEmptyHandle :: Handle -> Bool +isEmptyHandle Handle{..} = handleTimeout == 0 + +withNonEmptyHandle :: Handle -> IO () -> IO () +withNonEmptyHandle h act = + if isEmptyHandle h then pure () else act + +#if defined(mingw32_HOST_OS) +type TimerManager = EV.Manager + +getTimerManager :: IO TimerManager +getTimerManager = EV.getSystemManager +#else +type TimerManager = EV.TimerManager + +getTimerManager :: IO TimerManager +getTimerManager = EV.getSystemTimerManager +#endif diff --git a/System/TimeManager/Windows.hs b/System/TimeManager/Windows.hs new file mode 100644 --- /dev/null +++ b/System/TimeManager/Windows.hs @@ -0,0 +1,64 @@ +{-# LANGUAGE NumericUnderscores #-} + +-- | A timer for Windows which does not depend on the I\/O manager. +-- +-- @GHC.Event.Windows@ is WinIO's manager, so a timeout registered with +-- it is never run when the program uses the old Windows I\/O manager, +-- which is still the default. This does the same job with one thread +-- and 'threadDelay' per timeout, which both managers drive. +module System.TimeManager.Windows ( + Manager, + TimeoutKey, + getSystemManager, + registerTimeout, + unregisterTimeout, + updateTimeout, +) where + +import Control.Concurrent (ThreadId, forkIO, killThread, threadDelay) +import Data.IORef (IORef, newIORef, readIORef, writeIORef) +import Data.Word (Word64) +import GHC.Clock (getMonotonicTimeNSec) + +-- | There is nothing to keep: each timeout is its own thread. +data Manager = Manager + +getSystemManager :: IO Manager +getSystemManager = pure Manager + +-- | A registered timeout. The deadline is read by the thread rather +-- than captured, so that 'updateTimeout' does not have to replace it. +data TimeoutKey = TimeoutKey (IORef Word64) ThreadId + +instance Eq TimeoutKey where + TimeoutKey _ t1 == TimeoutKey _ t2 = t1 == t2 + +-- | Run the action once the given number of microseconds has passed. +registerTimeout :: Manager -> Int -> IO () -> IO TimeoutKey +registerTimeout _ us action = do + ref <- newIORef =<< deadlineIn us + tid <- forkIO $ wait ref + pure $ TimeoutKey ref tid + where + wait ref = do + deadline <- readIORef ref + now <- getMonotonicTimeNSec + if now >= deadline + then action + else do + -- Round up, so that a deadline under a microsecond away + -- is still waited for rather than busy looped on. + threadDelay $ fromIntegral ((deadline - now + 999) `div` 1_000) + wait ref + +-- | Move the deadline the given number of microseconds into the future. +updateTimeout :: Manager -> TimeoutKey -> Int -> IO () +updateTimeout _ (TimeoutKey ref _) us = writeIORef ref =<< deadlineIn us + +unregisterTimeout :: Manager -> TimeoutKey -> IO () +unregisterTimeout _ (TimeoutKey _ tid) = killThread tid + +deadlineIn :: Int -> IO Word64 +deadlineIn us = do + now <- getMonotonicTimeNSec + pure $ now + fromIntegral (max 0 us) * 1_000 diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,380 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE NumericUnderscores #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# OPTIONS_GHC -Wno-orphans #-} + +module Main where + +import Test.Hspec + +#if defined(mingw32_HOST_OS) +import qualified System.TimeManager.Windows as EV +#else +import qualified GHC.Event as EV +#endif + +import Control.Concurrent ( + MVar, + forkIO, + newEmptyMVar, + putMVar, + takeMVar, + threadDelay, + tryPutMVar, + ) +import Control.Concurrent.STM (atomically) +import Control.Exception (AssertionFailed (..), finally, throwIO) +import Control.Monad (forM_, void) +import Data.IORef as I ( + IORef, + atomicModifyIORef', + newIORef, + readIORef, + writeIORef, + ) +import GHC.Clock (getMonotonicTimeNSec) +import GHC.Conc.Sync ( + getUncaughtExceptionHandler, + setUncaughtExceptionHandler, + ) +import System.ThreadManager ( + ThreadManager, + countManaged, + forkManaged, + forkManagedFinally, + forkManagedTimeout, + forkManagedTimeoutFinally, + forkManagedUnmaskFinally, + newThreadManager, + reserveManaged, + stopAfter, + takeReservation, + waitUntilAllGone, + ) +import System.TimeManager +import System.TimeManager.Internal +import qualified System.Timeout as Timeout +import Test.HUnit (assertBool) + +main :: IO () +main = hspec $ do + describe "ThreadManager, once stopped" $ do + -- 'setup' refuses to record a thread forked before the stop and + -- scheduled after it, by throwing. That must not escape the fork, + -- and must not take the cleanup with it. + it "does not let forkManagedTimeout's refusal escape the fork" $ do + tm <- newThreadManager + stopAfter tm (return ()) $ \_ -> return () + ran <- freshRef + escaped <- freshRef + before <- getUncaughtExceptionHandler + setUncaughtExceptionHandler $ \_ -> I.writeIORef escaped True + forkManagedTimeout tm defaultManager "probe" $ \_ -> I.writeIORef ran True + threadDelay 200_000 + setUncaughtExceptionHandler before + ran `refShouldBe` False + escaped `refShouldBe` False + + itRunsTheCleanupOf "forkManagedFinally" $ + \tm label -> forkManagedFinally tm label (return ()) + + itRunsTheCleanupOf "forkManagedTimeoutFinally" $ + \tm label -> forkManagedTimeoutFinally tm defaultManager label (\_ -> return ()) + + itRunsTheCleanupOf "forkManagedUnmaskFinally" $ + \tm label -> forkManagedUnmaskFinally tm label (\_ -> return ()) + + describe "ThreadManager, the action of a fork with a cleanup" $ do + itLetsTheExceptionOutOf "forkManagedFinally" forkManagedFinally + + itLetsTheExceptionOutOf "forkManagedTimeoutFinally" $ + \tm label io -> forkManagedTimeoutFinally tm defaultManager label (\_ -> io) + + itLetsTheExceptionOutOf "forkManagedUnmaskFinally" $ + \tm label io -> forkManagedUnmaskFinally tm label (\unmask -> unmask io) + + describe "ThreadManager" $ do + it "fails loudly when a reservation is taken by another manager" $ do + a <- newThreadManager + b <- newThreadManager + rsv <- reserveManaged a + takeReservation b rsv (return ()) + atomically (countManaged b) + `shouldThrow` (\(AssertionFailed _) -> True) + + it "has none left only once their cleanups are done" $ do + tm <- newThreadManager + ref <- freshRef + forkManagedFinally + tm + "probe" + (threadDelay 100_000) + (threadDelay 300_000 >> I.writeIORef ref True) + -- The action is over and the cleanup is still running. + threadDelay 200_000 + waitUntilAllGone tm + ref `refShouldBe` True + + it "still owns a thread it killed until that thread is gone" $ do + tm <- newThreadManager + ref <- freshRef + started <- newEmptyMVar + -- The cleanup here is the thread's own, not one the manager + -- runs for it, so what is tested is an ordinary managed + -- thread unwinding. + forkManaged tm "probe" $ + (putMVar started () >> threadDelay 10_000_000) + `finally` (threadDelay 300_000 >> I.writeIORef ref True) + takeMVar started + threadDelay 50_000 + -- 'throwTo' waits for the exception to be taken, not for the + -- thread to be gone. + stopAfter tm (return ()) $ \_ -> return () + threadDelay 50_000 + atomically (countManaged tm) `shouldReturn` 1 + waitUntilAllGone tm + ref `refShouldBe` True + + describe "TimeManager" $ do + it "defaultManager == no manager" $ + defaultManager `shouldSatisfy` isNoManager + + it "initializes negative manager" $ do + let check = (`shouldBe` defaultManager) + initialize (-10) >>= check + withManager (-5) check + + it "empty handle is correct" $ + handleTimeout emptyHandle `shouldBe` 0 + + it "empty handle check is consistent" $ do + assertBool "emptyHandle not empty" $ + isEmptyHandle emptyHandle + + it "gives emptyHandle when registering defaultManager" $ do + hndl <- register defaultManager $ pure () + assertBool "got non-empty handle" $ isEmptyHandle hndl + + it "throws TimeoutThread exception" $ + throwsTimeoutThread $ do + mngr <- initialize timeoutAmount + _hndl <- registerKillThread mngr $ pure () + waitLong + + it "defaultManager doesn't kill thread" $ do + _hndl <- registerKillThread defaultManager $ pure () + waitShort + + it "withHandle: registers timeout" $ + withHandleTest mgr1 $ \check _ -> do + waitShort + check True + + it "withHandle: doesn't register timeout" $ + withHandleTest defaultManager $ \check _ -> do + waitShort + check False + + -- We make a ref on the outside, to check that the ref is indeed + -- set before the timeout kills the action inside. + it "withHandleKillThread: registers timeout (and kills)" $ do + ref <- freshRef + withHandleKillTest (Just ref) mgr1 $ \_ _ -> + throwsTimeoutThread waitShort + ref `refShouldBe` True + + it "withHandleKillThread: doesn't register timeout" $ + withHandleKillTest Nothing defaultManager $ \check _ -> do + waitShort >> check False + + it "cancel/pause works as expected" $ do + m <- mkTestManager + let killUnless f = do + hndl <- registerKillThread m (pure ()) + _ <- f hndl + waitLong + throwsTimeoutThread $ killUnless pure + killUnless cancel + killUnless pause + + it "tickle works as expected" $ do + m <- mkTestManager + withHandleTest m $ \check hndl -> do + forM_ [(1 :: Int) .. 20] $ \_ -> do + waitShort + tickle hndl + check False + + let runAndWaitForTimeout f = + runIt $ \hndl -> do + void $ f hndl + waitLong + it "resume works as expected (nothing)" $ do + -- Doing nothing kills the thread + throwsTimeoutThread . runAndWaitForTimeout $ \_ -> pure () + it "resume works as expected (pause)" $ do + -- Pausing stops the kill + runAndWaitForTimeout $ \hndl -> do + waitShort >> pause hndl + it "resume works as expected (pause/resume)" $ do + -- Resuming kills the thread again + throwsTimeoutThread . runAndWaitForTimeout $ \hndl -> do + waitShort >> pause hndl + waitLong >> resume hndl + it "resume works as expected (cancel/resume)" $ do + -- Cancelling is unresumable + runAndWaitForTimeout $ \hndl -> do + waitShort >> cancel hndl + waitLong >> resume hndl + it "resume works as expected (cancel/pause/resume)" $ do + -- Cancelling and then pausing is still unresumable + runAndWaitForTimeout $ \hndl -> do + waitShort >> cancel hndl + waitShort >> pause hndl + waitLong >> resume hndl + -- Pausing, then cancelling doesn't change anything + runAndWaitForTimeout $ \hndl -> do + waitShort >> pause hndl + waitShort >> cancel hndl + waitLong >> resume hndl + it "finished timeout won't resume" $ do + -- If the timeout action runs, resume shouldn't work + counter <- I.newIORef (0 :: Int) + m <- mkTestManager + let increase = I.atomicModifyIORef' counter $ \i -> (i + 1, ()) + withHandle m increase $ \h -> do + let checkCount x = do + i <- I.readIORef counter + i `shouldBe` x + timeoutOnlyRanOnce = waitLong >> checkCount 1 + + checkCount 0 + -- waiting lets the timeout + timeoutOnlyRanOnce + -- resuming should not influence the counter + resume h + timeoutOnlyRanOnce + -- pausing after it runs also doesn't re-arm the timeout + pause h + resume h + timeoutOnlyRanOnce + -- cancel also doesn't re-arm the timeout + cancel h + pause h + resume h + timeoutOnlyRanOnce + + it "reschedules an expired callback for the remaining timeout" $ do + m <- initialize 1_000_000 + fired <- newEmptyMVar + withHandle m (putMVar fired ()) $ \h -> do + threadDelay 500_000 + -- Model a renewal after GHC has queued the original callback. + -- Updating GHC's timer here would remove that stale callback. + withLock (handleLock h) $ do + now <- getMonotonicTimeNSec + I.writeIORef (handleLastRenewed h) now + Timeout.timeout 750_000 (takeMVar fired) + `shouldReturn` Nothing + Timeout.timeout 1_000_000 (takeMVar fired) + `shouldReturn` Just () + + it "resume also works as tickle" $ + testResume resume + + it "resume also works as tickle with pauses" $ + testResume $ \hndl -> do + resume hndl + pause hndl + resume hndl + + it "old resume did NOT work as tickle" $ + throwsTimeoutThread $ + testResume oldResume + where + withHandleTest = withTest withHandle Nothing + withHandleKillTest = withTest withHandleKillThread + -- Test that starts with a 'False' IORef and on timeout sets it to true + withTest withF mRef m f = do + ref <- maybe freshRef pure mRef + withF m (I.writeIORef ref True) . f $ refShouldBe ref + -- run with a 20ms timeout and kill + runIt f = do + m <- mkTestManager + void $ f =<< registerKillThread m (pure ()) + timeoutAmount = 20_000 + mkTestManager = initialize timeoutAmount + -- Waiting a lot less than the timeout takes + waitShort = threadDelay $ timeoutAmount `div` 5 + -- Waiting a lot longer than the timeout takes + waitLong = threadDelay $ timeoutAmount * 5 + -- "resuming" every 2.5ms 20 times + testResume f = do + runIt $ \hndl -> do + forM_ [(1 :: Int) .. 20] $ \_ -> waitShort >> f hndl + +mgr1 :: Manager +mgr1 = Manager 1 + +-- | The cleanup of a fork that takes one runs even when the manager has +-- been stopped and the thread is refused. +itRunsTheCleanupOf + :: String + -> (ThreadManager -> String -> IO () -> IO ()) + -- ^ The fork, with its action already given and its cleanup to come + -> Spec +itRunsTheCleanupOf name fork = + it ("runs the cleanup of " ++ name) $ do + tm <- newThreadManager + stopAfter tm (return ()) $ \_ -> return () + done <- newEmptyMVar + fork tm "probe" $ putMVar done () + ranWithin 500_000 done `shouldReturn` True + +-- | An exception from the action of a fork that takes a cleanup leaves the +-- thread, after the cleanup has run. +itLetsTheExceptionOutOf + :: String + -> (ThreadManager -> String -> IO () -> IO () -> IO ()) + -> Spec +itLetsTheExceptionOutOf name fork = + it ("lets an exception from the action of " ++ name ++ " out") $ do + tm <- newThreadManager + escaped <- newEmptyMVar + cleaned <- freshRef + before <- getUncaughtExceptionHandler + setUncaughtExceptionHandler $ \_ -> void $ tryPutMVar escaped () + fork tm "probe" (throwIO $ userError "probe") (I.writeIORef cleaned True) + out <- ranWithin 500_000 escaped + setUncaughtExceptionHandler before + out `shouldBe` True + cleaned `refShouldBe` True + +-- | Whether the 'MVar' was filled within the time. +ranWithin :: Int -> MVar () -> IO Bool +ranWithin us var = do + r <- newEmptyMVar + void $ forkIO $ takeMVar var >> putMVar r True + void $ forkIO $ threadDelay us >> putMVar r False + takeMVar r + +freshRef :: IO (IORef Bool) +freshRef = I.newIORef False + +refShouldBe :: IORef Bool -> Bool -> IO () +refShouldBe ref expected = + I.readIORef ref >>= (`shouldBe` expected) + +throwsTimeoutThread :: IO () -> Expectation +throwsTimeoutThread t = t `shouldThrow` (const True :: TimeoutThread -> Bool) + +deriving instance Eq Manager +deriving instance Show Manager + +-- copied from time-manager-0.3.0 to check it actually is broken +oldResume :: Handle -> IO () +oldResume h | isEmptyHandle h = return () +oldResume Handle{..} = do + key <- EV.registerTimeout handleTimerManager handleTimeout handleAction + I.writeIORef handleState $ Active key diff --git a/time-manager.cabal b/time-manager.cabal --- a/time-manager.cabal +++ b/time-manager.cabal @@ -1,26 +1,58 @@ -Name: time-manager -Version: 0.2.4 -Synopsis: Scalable timer -License: MIT -License-file: LICENSE -Author: Michael Snoyman and Kazu Yamamoto -Maintainer: kazu@iij.ad.jp -Homepage: http://github.com/yesodweb/wai -Category: System -Build-Type: Simple -Cabal-Version: >=1.10 -Stability: Stable -Description: Scalable timer functions provided by a timer manager - and thread management functions to prevent thread - leak by a thread manager. -Extra-Source-Files: ChangeLog.md +cabal-version: >=1.10 +name: time-manager +version: 0.5.2 +license: MIT +license-file: LICENSE +maintainer: kazu@iij.ad.jp +author: Michael Snoyman and Kazu Yamamoto +stability: Stable +homepage: http://github.com/yesodweb/wai +synopsis: Scalable timer +description: + Scalable timer functions provided by a timer manager + and thread management functions to prevent thread + leak by a thread manager. -Library - Build-Depends: base >= 4.12 && < 5 - , auto-update >= 0.2 && < 0.3 - , containers - , stm - Default-Language: Haskell2010 - Exposed-modules: System.TimeManager - Exposed-modules: System.ThreadManager - Ghc-Options: -Wall +category: System +build-type: Simple +extra-source-files: ChangeLog.md + README.md + +library + exposed-modules: + System.TimeManager + System.ThreadManager + other-modules: + System.TimeManager.Internal + + if os(windows) + other-modules: System.TimeManager.Windows + + default-language: Haskell2010 + default-extensions: Strict StrictData + ghc-options: -Wall + build-depends: + base >=4.12 && <5, + containers, + stm + +test-suite spec + type: exitcode-stdio-1.0 + main-is: test/Spec.hs + other-modules: + System.ThreadManager, + System.TimeManager, + System.TimeManager.Internal + + if os(windows) + other-modules: System.TimeManager.Windows + + build-depends: base + default-language: Haskell2010 + ghc-options: -Wall -threaded + build-depends: + base >=4.12 && <5, + containers, + hspec, + HUnit, + stm