diff --git a/Control/Timeout.hs b/Control/Timeout.hs new file mode 100644 --- /dev/null +++ b/Control/Timeout.hs @@ -0,0 +1,197 @@ +-- | +-- This module handles timeouts by using a (single) thread sitting in threadDelay +-- and the STM. One can request an IO action be performed after some number of +-- seconds and later cancel that request if need be. +-- +-- The number of threads used is constant. +module Control.Timeout ( + addTimeout + , cancelTimeout + , TimeoutTag) where + +import Control.Concurrent +import Control.Concurrent.STM +import Control.Monad + +import System.IO.Unsafe + +import qualified Data.Map as Map +import Data.Time.Clock.POSIX + +-- | This is set, atomically, to true when the manager thread is started. +-- This thread isn't started unless someone actually creates a timeout +managerThreadStarted :: TVar Bool +managerThreadStarted = unsafePerformIO $ newTVarIO False + +-- | When a timeout thread times out, it compares the first element of this +-- tuple against the value that it was created with. If they don't match then +-- it's no longer the current timeout thread and it exits. Otherwise, it sets +-- the second element to true and exits. +signal :: TVar (Int, Bool) +signal = unsafePerformIO $ newTVarIO (0, False) + +-- | This is a map of all the timeouts. It maps the absolute time +-- that the timeout expires at to a list of tagged actions +-- to perform at that time. For a given value in the map, the Ints of +-- every element in the list must be unique. +timeouts :: TVar (Map.Map POSIXTime [(Int, IO ())]) +timeouts = unsafePerformIO $ newTVarIO Map.empty + +-- Here's how everything works. The addTimeout and cancelTimeout functions +-- alter the above globals using the STM. The first call to addTimeout will +-- start a manager thread which watches @signal@ and @timeouts@ for changes: +-- either the timeoutThread has completed or someone has added/removed the +-- least element of timeouts. +-- +-- In the first case, @timeouts@ is updated by removing all the expired +-- timeouts and their actions are performed. In both cases, the time to the +-- next timeout is calculated and a timeoutThread is forked to sleep for that +-- length of time before signaling via @signal@. +-- +-- Timeouts are identified by their absolute time value and the unique tag +-- number for their action at that time. When creating a timeout, that pair +-- is wrapped in the (opaque) TimeoutTag type and returned. When canceling a +-- timeout, the list of actions for the given absolute time is filtered to +-- remove the indicated action. Because of this, the tags for a given absolute +-- time must be unique. This is achieved by giving the first element a tag of 0 +-- and giving subsequent elements a tag one greater than the current max. +-- +-- Each timeoutThread is given a tag (of a different type to the tags in the +-- last paragraph) so that @signal@ is never set by an old timeoutThread +-- which hasn't died yet. + +-- | Get the list of expired timers from a map of timeouts +expiredTimers :: POSIXTime -- ^ the current time + -> Map.Map POSIXTime a -- ^ the timeouts map + -> ([a], Map.Map POSIXTime a) -- ^ the list of actions and a new map +expiredTimers curtime m = do + unfoldrWithValue f m where + f m = + case Map.minViewWithKey m of + Nothing -> Nothing + Just ((time, action), m') -> + if time <= curtime + then Just (action, m') + else Nothing + +-- | Run the actions for all expired timers in the @timeouts@ global. Update +-- that global with a new Map, less the expired timeouts. +runExpiredTimers :: (Monad m) + => POSIXTime -- ^ the current time + -> TVar (Map.Map POSIXTime [(Int, m ())]) -- ^ the timeouts map + -> STM (m ()) +runExpiredTimers currentTime tm = do + m <- readTVar tm + let (actions, m') = expiredTimers currentTime m + when (length actions > 0) $ writeTVar tm m' + return $ (do sequence $ map (sequence . map snd) actions; return ()) + +-- | A version of unfoldr which returns the final value as well. Note that +-- the resulting list comes off in reverse order +unfoldrWithValue :: (b -> Maybe (a, b)) -> b -> ([a], b) +unfoldrWithValue f i = inner f i [] where + inner f i acc = + case f i of + Nothing -> (acc, i) + Just (v, i') -> inner f i' $ v : acc + +-- | This is a thread which waits for the given number of milliseconds +-- and tries to set the snd element of the global @signal@ to true, iff +-- the first element of that global is equal to its tag number. +timeoutThread :: Int -- ^ the id of this timeout, see @signal@ + -> POSIXTime -- ^ the time to wakeup + -> IO () +timeoutThread id targetTime = do + currentTime <- getPOSIXTime + let deltausecs = truncate $ fromRational $ toRational ((currentTime - targetTime) * 1000000) + when (deltausecs > 0) $ threadDelay deltausecs + atomically (do + (id', _) <- readTVar signal + when (id' == id) $ writeTVar signal (id, True)) + +-- | This is an opaque type of timeouts. A value of this type is returned +-- when creating a timeout and can be used to cancel the same timeout. +newtype TimeoutTag = TimeoutTag (POSIXTime, Int) + +-- | Add an action to be performed at some point in the future. The action will +-- occur inside a thread which is dedicated to performing them so it should +-- run quickly and certainly should not block on IO etc. +addTimeout :: Float -- ^ the number of seconds in the future to perform the action + -> (IO ()) -- ^ the action to perform + -> IO TimeoutTag +addTimeout delta action = do + currentTime <- getPOSIXTime + let future = currentTime + (fromRational $ toRational delta) + tag <- atomically (do + m <- readTVar timeouts + case Map.lookup future m of + Nothing -> do writeTVar timeouts $ Map.insert future [(0, action)] m + return $ TimeoutTag (future, 0) + Just xs -> do let magic = 1 + (maximum $ map fst xs) + writeTVar timeouts $ Map.insert future ((magic, action) : xs) m + return $ TimeoutTag (future, magic)) + -- If the manager thread isn't running, start it now. + startp <- atomically (do + started <- readTVar managerThreadStarted + when (not started) $ writeTVar managerThreadStarted True + return $ not started) + when startp $ forkIO (timeoutManagerThread timeouts signal 0 Nothing) >> return () + + return tag + +-- | Remove a timeout. This function never fails, but will return False if the +-- given timeout couldn't be found. This may be because cancelTimeout has +-- already been called with this tag, or because the timeout has already +-- fired. Note that, since timeouts are IO actions, they don't run atomically. +-- Thus it's possible that this call returns False and that the timeout is +-- currently in the process of running. +-- +-- Note that one should never call cancelTimeout twice with the same tag since +-- it's possible that the tag will be reused and thus the second call could +-- cancel a different timeout. +cancelTimeout :: TimeoutTag -- ^ the tag returned by addTimeout + -> STM Bool -- ^ returns False if the timeout didn't exist +cancelTimeout (TimeoutTag (future, n)) = do + m <- readTVar timeouts + case Map.lookup future m of + Nothing -> return False + Just xs -> do + let xs' = filter (\(t, _) -> t /= n) xs + m' = Map.insert future xs' m + writeTVar timeouts m' + return $ length xs' /= length xs + +timeoutManagerThread :: TVar (Map.Map POSIXTime [(Int, IO ())]) + -> TVar (Int, Bool) + -> POSIXTime -- ^ the current minimum time + -> Maybe ThreadId -- ^ the id of the current timeoutThread + -> IO () +timeoutManagerThread tm signal currentMin mthid = do + -- the event is either a signal from the timeoutThread (True) or that a + -- timeout has been added to the timeouts map (False) + (event, currentTag) <- atomically (do + s <- readTVar signal + let currentTag = fst s + if snd s + then return (True, currentTag) + else (do + m <- readTVar tm + if (Map.null m && currentMin /= 0) || + (not (Map.null m) && fst (Map.findMin m) /= currentMin) + then return (False, currentTag) + else retry)) + + currentTime <- getPOSIXTime + when event $ atomically (runExpiredTimers currentTime tm) >>= id + minTimeout <- atomically (do + m <- readTVar timeouts + if Map.null m + then return 0 + else return $ fst $ Map.findMin m) + let nextTag = currentTag + 1 + atomically $ writeTVar signal (nextTag, False) + case mthid of + Nothing -> return () + Just x -> killThread x + tid <- forkIO $ timeoutThread nextTag minTimeout + timeoutManagerThread tm signal minTimeout $ Just tid diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) Adam Langley + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions +are met: + +1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in the + documentation and/or other materials provided with the distribution. + +3. Neither the name of the author nor the names of his contributors + may be used to endorse or promote products derived from this software + without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE CONTRIBUTORS ``AS IS'' AND ANY EXPRESS +OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE FOR +ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS +OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) +HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN +ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +POSSIBILITY OF SUCH DAMAGE. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,3 @@ +#!/usr/bin/env runhaskell +> import Distribution.Simple +> main = defaultMain diff --git a/control-timeout.cabal b/control-timeout.cabal new file mode 100644 --- /dev/null +++ b/control-timeout.cabal @@ -0,0 +1,14 @@ +name: control-timeout +version: 0.1 +license: BSD3 +license-file: LICENSE +author: Adam Langley +description: This package provides functions for running timeouts +synopsis: Timeout handling +category: Control +build-depends: base, containers, time, stm +stability: provisional +tested-with: GHC == 6.8.1 +exposed-modules: Control.Timeout +ghc-options: -O2 +extra-source-files: test/TimeoutTest.hs diff --git a/test/TimeoutTest.hs b/test/TimeoutTest.hs new file mode 100644 --- /dev/null +++ b/test/TimeoutTest.hs @@ -0,0 +1,50 @@ +module Main where + +import Control.Concurrent +import Control.Concurrent.STM + +import Data.Time.Clock.POSIX +import Control.Timeout + +import Text.Printf (printf) + +singleTimeout secs = do + a <- atomically $ newTVar 0 + startTime <- getPOSIXTime + addTimeout secs (getPOSIXTime >>= atomically . writeTVar a) + endTime <- atomically (do + t <- readTVar a + if t == 0 + then retry + else return t) + printf "%dms sleep took: %s\n" ((truncate $ secs * 1000) :: Int) (show (endTime - startTime)) + +multiTimeout :: Int -> Float -> IO () +multiTimeout n secs = do + a <- atomically $ newTVar n + startTime <- getPOSIXTime + mapM_ (const $ addTimeout secs (atomically $ readTVar a >>= writeTVar a . ((flip (-)) 1))) [1..n] + atomically (do + n' <- readTVar a + if n' == 0 + then return () + else retry) + endTime <- getPOSIXTime + printf "%d timeouts for %d ms took %s\n" n ((truncate $ secs * 1000) :: Int) (show $ endTime - startTime) + +setAndCancelTimeout :: Int -> Float -> IO () +setAndCancelTimeout n secs = do + a <- atomically $ newTVar (0 :: Int) + startTime <- getPOSIXTime + tags <- mapM (const $ addTimeout secs (atomically $ readTVar a >>= writeTVar a . ((+) 1))) [1..n] + mapM (atomically . cancelTimeout) tags + endTime <- getPOSIXTime + expired <- atomically $ readTVar a + + printf "Setting and killing %d timeouts for %d ms took %s and %d expired\n" n ((truncate $ secs * 1000) :: Int) (show $ endTime - startTime) expired + +main = do + sequence [singleTimeout 0, singleTimeout 0.1, singleTimeout 0.5, singleTimeout 1.0] + sequence [multiTimeout 10 0.1, multiTimeout 100 0.1, multiTimeout 1000 0.1, multiTimeout 10000 0.1, multiTimeout 100000 0.1] + sequence [setAndCancelTimeout 10 0.1, setAndCancelTimeout 100 0.1, setAndCancelTimeout 1000 0.1, setAndCancelTimeout 10000 0.1, setAndCancelTimeout 100000 0.1] + return ()