time-manager 0.3.2 → 0.5.2
raw patch · 7 files changed
Files
- ChangeLog.md +61/−0
- System/ThreadManager.hs +292/−73
- System/TimeManager.hs +280/−41
- System/TimeManager/Internal.hs +70/−8
- System/TimeManager/Windows.hs +64/−0
- test/Spec.hs +264/−49
- time-manager.cabal +12/−2
ChangeLog.md view
@@ -1,5 +1,63 @@ # 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)@@ -36,6 +94,9 @@ * `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
System/ThreadManager.hs view
@@ -1,8 +1,9 @@ {-# 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,@@ -23,30 +24,33 @@ 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 (IORef, atomicModifyIORef', newIORef) import Data.Map.Strict (Map)@@ -61,16 +65,25 @@ ---------------------------------------------------------------- --- | 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)@@ -79,13 +92,60 @@ -- | Create a thread manager. ----- To create a 'ThreadManager', you will first have to create a--- 'T.Manager' from the "System.TimeManager" module.+-- 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. ----- You can use either 'System.TimeManager.initialize' or 'System.TimeManager.withManager'.-newThreadManager :: T.Manager -> IO ThreadManager-newThreadManager timmgr = ThreadManager timmgr <$> newTVarIO Map.empty+-- 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.@@ -106,6 +166,20 @@ -- 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 mgr action cleanup = stopAfterWithResult mgr action $ \mResult ->@@ -125,13 +199,19 @@ -- @since 0.3.2 stopAfterWithResult :: ThreadManager -> IO a -> (Either SomeException a -> IO b) -> IO b-stopAfterWithResult (ThreadManager _timmgr var) action cleanup = do+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@@ -142,9 +222,12 @@ -- | 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@@ -152,10 +235,25 @@ -> 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@@ -163,64 +261,163 @@ -> ((forall x. IO x -> IO x) -> IO ()) -- ^ Action with unmask argument -> 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+forkManagedUnmask mgr label io = forkManagedUnmaskFinally mgr label io $ return () +-- | 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 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 ()++-- | 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 (ThreadManager timmgr var) label io =- void $ forkIO $ do- labelMe label- E.bracket (setup var) (clear var) $ \(_n, wtid, ref) ->- E.handle ignore $ T.withHandle timmgr (lockAndKill wtid ref ex) 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+ -> 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 --- | Fork a managed thread with a cleanup function.-forkManagedFinally+-- | The body of a managed thread, under its own label, taking the place+-- kept for it.+managed :: ThreadManager+ -> Reservation -> String- -- ^ Thread name -> IO ()- -- ^ Action+ -> ((Key, ManagedThread) -> IO ()) -> IO ()- -- ^ Cleanup function+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 ()-forkManagedFinally mgr label io final =- forkManagedUnmask mgr label $ \restore ->- E.try (restore io) >>= \(_ :: Either E.SomeException ()) -> final+ -> ((Key, ManagedThread) -> IO ())+ -> IO ()+taking mgr rsv final body =+ recorded mgr (give >> final) $ \key -> give >> body key+ where+ give = release mgr rsv --- | Fork a managed thread with a handle created by a timeout manager--- and with a cleanup function.-forkManagedTimeoutFinally+-- | 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- -> String- -- ^ Thread name- -> (T.Handle -> IO ())- -- ^ Action with timeout handle -> IO ()- -- ^ Cleanup function+ -> ((Key, ManagedThread) -> IO ()) -> IO ()-forkManagedTimeoutFinally mgr label io final = E.mask $ \restore ->- forkManagedTimeout mgr label $ \th ->- E.try (restore $ io th) >>= \(_ :: Either E.SomeException ()) -> final+recorded mgr final body =+ E.handle ignore $+ E.bracket (setup mgr) (mapM_ $ clear mgr) $ \mkey ->+ mapM_ body mkey `E.finally` final -setup :: TVar (Map Key ManagedThread) -> IO (Key, Weak ThreadId, IORef Bool)-setup var = do+-- | 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@@ -232,15 +429,24 @@ 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 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 tm = atomically $@@ -248,9 +454,28 @@ -- | STM action that checks if all managed threads are finished. isAllGone :: ThreadManager -> STM Bool-isAllGone (ThreadManager _timmgr var) =- Map.null <$> readTVar var+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+ ---------------------------------------------------------------- myWeakThreadId :: IO (Weak ThreadId, Key)@@ -264,12 +489,6 @@ labelMe l = do tid <- myThreadId labelThread tid l---- | Registering a 'T.TimeoutAction' and unregister its 'T.Handle'--- when the body action is finished.-withHandle- :: ThreadManager -> T.TimeoutAction -> (T.Handle -> IO a) -> IO a-withHandle (ThreadManager timmgr _) = T.withHandle timmgr #if __GLASGOW_HASKELL__ < 908 fromThreadId :: ThreadId -> Word64
System/TimeManager.hs view
@@ -1,8 +1,11 @@ {-# LANGUAGE CPP #-}+{-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE RecordWildCards #-} -- | Timeout manager. Since @v0.3.0@, timeout manager is a wrapper of--- GHC System TimerManager.+-- 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: --@@ -12,7 +15,7 @@ -- * 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)+-- edge cases. (i.e. using 'cancel'/'pause' in one thread, and 'resume' in another) module System.TimeManager ( -- ** Types Manager,@@ -36,25 +39,29 @@ tickle, pause, resume,+ cancel, -- ** Low level register, registerKillThread,- cancel, -- ** Exceptions TimeoutThread (..), ) where -import Control.Concurrent (forkIO, mkWeakThreadId, myThreadId)+import Control.Concurrent (forkIO, mkWeakThreadId, myThreadId, newMVar) import qualified Control.Exception as E-import Control.Monad (void)+import Control.Monad (void, when)+import Data.Bits (shiftR) import qualified Data.IORef as I+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 GHC.Event.Windows as EV+import qualified System.TimeManager.Windows as EV #else import qualified GHC.Event as EV #endif@@ -73,15 +80,24 @@ Handle { handleTimeout = 0 , handleAction = pure ()- , handleKeyRef = error "time-manager: Handle.handleKeyRef not set"- , handleState = error "time-manager: Handle.handleState not set"+ , handleTimerManager = mutError "handleTimerManager"+ , handleState = mutError "handleState"+ , handleLastRenewed = mutError "handleLastRenewed"+ , handleMinRenewGap = 0+ , handleLock = emptyLock }+ where+ mutError s = error $ "time-manager: Handle." <> s <> " not set" +emptyLock :: Lock+emptyLock = unsafePerformIO $ newMVar ()+{-# NOINLINE emptyLock #-}+ ---------------------------------------------------------------- -- | Creating timeout manager with a timeout value in microseconds. ----- Setting the timeout to zero or lower (<= 0) will produce a+-- Setting the timeout to zero or lower @(<= 0)@ will produce a -- `defaultManager`. initialize :: Int -> IO Manager initialize = pure . Manager . max 0@@ -121,64 +137,287 @@ ---------------------------------------------------------------- +-- ============== 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 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- key <- EV.registerTimeout sysmgr timeout onTimeout- keyref <- I.newIORef key- state <- I.newIORef Active+ stateRef <- I.newIORef Stopped+ lock <- newLock+ lastRenewedRef <- I.newIORef =<< getMonotonicTimeNSec let h = Handle { handleTimeout = timeout , handleAction = onTimeout- , handleKeyRef = keyref- , handleState = state+ , 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 +-- | 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 h@Handle{..} = withNonEmptyHandle h $ do- mgr <- getTimerManager- key <- I.readIORef handleKeyRef- EV.unregisterTimeout mgr key- I.atomicWriteIORef handleState Stopped+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. ----- Careful: this does NOT reactivate an already paused 'Handle'!+-- 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 $ do- mgr <- getTimerManager- key <- I.readIORef handleKeyRef-#if defined(mingw32_HOST_OS)- EV.updateTimeout mgr key $ fromIntegral (handleTimeout `div` 1000000)-#else- EV.updateTimeout mgr key handleTimeout-#endif+tickle h@Handle{..} =+ withNonEmptyHandle h $+ whenRenew h $+ withActiveTimeoutKey h $ \key -> do+ updateTheTimeout key+ now <- getMonotonicTimeNSec+ I.atomicWriteIORef handleLastRenewed now+ where+ updateTheTimeout key =+ EV.updateTimeout handleTimerManager key handleTimeout --- | This is identical to 'cancel'.--- To resume timeout with the same 'Handle', 'resume' MUST be called.--- Don't call 'tickle' for resumption.+-- | 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 = cancel+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 $ do- state <- I.readIORef handleState- case state of- Active -> tickle h- Stopped -> do- mgr <- getTimerManager- key <- EV.registerTimeout mgr handleTimeout handleAction- I.atomicWriteIORef handleKeyRef key- I.atomicWriteIORef handleState Active+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 ----------------------------------------------------------------
System/TimeManager/Internal.hs view
@@ -5,10 +5,12 @@ module System.TimeManager.Internal where -import Data.IORef (IORef)+import Control.Concurrent.MVar (MVar, modifyMVar, newMVar)+import Data.IORef (IORef, readIORef)+import Data.Word (Word64) #if defined(mingw32_HOST_OS)-import qualified GHC.Event.Windows as EV+import qualified System.TimeManager.Windows as EV #else import qualified GHC.Event as EV #endif@@ -31,14 +33,70 @@ data Handle = Handle { handleTimeout :: Int , handleAction :: TimeoutAction- , handleKeyRef :: ~(IORef EV.TimeoutKey)+ , 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. } --- | Tracking the state of a handle, to be able to have 'resume'--- act like a 'register' or 'tickle'.-data HandleState = Active | Stopped+-- | 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 @@ -47,9 +105,13 @@ if isEmptyHandle h then pure () else act #if defined(mingw32_HOST_OS)-getTimerManager :: IO EV.Manager+type TimerManager = EV.Manager++getTimerManager :: IO TimerManager getTimerManager = EV.getSystemManager #else-getTimerManager :: IO EV.TimerManager+type TimerManager = EV.TimerManager++getTimerManager :: IO TimerManager getTimerManager = EV.getSystemTimerManager #endif
+ System/TimeManager/Windows.hs view
@@ -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
test/Spec.hs view
@@ -6,22 +6,136 @@ module Main where -import Control.Concurrent (threadDelay)-import Control.Monad (forM_, void)-import Data.IORef as I (IORef, newIORef, readIORef, writeIORef)-import System.TimeManager-import System.TimeManager.Internal-import Test.HUnit (assertBool) import Test.Hspec #if defined(mingw32_HOST_OS)-import qualified GHC.Event.Windows as EV+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@@ -44,22 +158,22 @@ it "throws TimeoutThread exception" $ throwsTimeoutThread $ do- mngr <- initialize 1+ mngr <- initialize timeoutAmount _hndl <- registerKillThread mngr $ pure ()- threadDelay 100+ waitLong it "defaultManager doesn't kill thread" $ do _hndl <- registerKillThread defaultManager $ pure ()- threadDelay 2000+ waitShort it "withHandle: registers timeout" $ withHandleTest mgr1 $ \check _ -> do- threadDelay 2000+ waitShort check True it "withHandle: doesn't register timeout" $ withHandleTest defaultManager $ \check _ -> do- threadDelay 2000+ waitShort check False -- We make a ref on the outside, to check that the ref is indeed@@ -67,59 +181,105 @@ it "withHandleKillThread: registers timeout (and kills)" $ do ref <- freshRef withHandleKillTest (Just ref) mgr1 $ \_ _ ->- throwsTimeoutThread $ threadDelay 100+ throwsTimeoutThread waitShort ref `refShouldBe` True it "withHandleKillThread: doesn't register timeout" $ withHandleKillTest Nothing defaultManager $ \check _ -> do- threadDelay 200- check False+ waitShort >> check False it "cancel/pause works as expected" $ do- m <- initialize 100- let runIt f = do+ m <- mkTestManager+ let killUnless f = do hndl <- registerKillThread m (pure ()) _ <- f hndl- threadDelay 1000- throwsTimeoutThread $ runIt pure- runIt cancel- runIt pause+ waitLong+ throwsTimeoutThread $ killUnless pure+ killUnless cancel+ killUnless pause it "tickle works as expected" $ do- m <- initialize 10_000+ m <- mkTestManager withHandleTest m $ \check hndl -> do forM_ [(1 :: Int) .. 20] $ \_ -> do- threadDelay 1000+ waitShort tickle hndl check False - let runIt f = do- m <- initialize 10_000- void $ f =<< registerKillThread m (pure ())- it "resume works as expected" $ do- let runAndWaitForTimeout f =- runIt $ \hndl -> do- void $ f hndl- threadDelay 20_000+ 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- threadDelay 2500- pause hndl+ waitShort >> pause hndl+ it "resume works as expected (pause/resume)" $ do -- Resuming kills the thread again throwsTimeoutThread . runAndWaitForTimeout $ \hndl -> do- threadDelay 2500- pause hndl- threadDelay 20_000- resume hndl+ 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 - -- "resuming" every 2.5ms 20 times- let testResume f = do- runIt $ \hndl -> do- forM_ [(1 :: Int) .. 20] $ \_ -> do- threadDelay 2500- f hndl+ 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 @@ -138,17 +298,73 @@ -- 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 (writeIORef ref True) . f $ refShouldBe ref+ 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 = newIORef False+freshRef = I.newIORef False refShouldBe :: IORef Bool -> Bool -> IO () refShouldBe ref expected =- readIORef ref >>= (`shouldBe` expected)+ I.readIORef ref >>= (`shouldBe` expected) throwsTimeoutThread :: IO () -> Expectation throwsTimeoutThread t = t `shouldThrow` (const True :: TimeoutThread -> Bool)@@ -160,6 +376,5 @@ oldResume :: Handle -> IO () oldResume h | isEmptyHandle h = return () oldResume Handle{..} = do- mgr <- getTimerManager- key <- EV.registerTimeout mgr handleTimeout handleAction- I.writeIORef handleKeyRef key+ key <- EV.registerTimeout handleTimerManager handleTimeout handleAction+ I.writeIORef handleState $ Active key
time-manager.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: time-manager-version: 0.3.2+version: 0.5.2 license: MIT license-file: LICENSE maintainer: kazu@iij.ad.jp@@ -25,6 +25,9 @@ other-modules: System.TimeManager.Internal + if os(windows)+ other-modules: System.TimeManager.Windows+ default-language: Haskell2010 default-extensions: Strict StrictData ghc-options: -Wall@@ -37,12 +40,19 @@ 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+ HUnit,+ stm