control-timeout (empty) → 0.1

raw patch · 5 files changed

+294/−0 lines, 5 filesdep +basedep +containersdep +stmbuild-type:Customsetup-changed

Dependencies added: base, containers, stm, time

Files

• Control/Timeout.hs +197/−0
• LICENSE +30/−0
• Setup.lhs +3/−0
• control-timeout.cabal +14/−0
• test/TimeoutTest.hs +50/−0

@@ -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

@@ -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.

@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain

@@ -0,0 +1,14 @@+name:            control-timeout+version:         0.1+license:         BSD3+license-file:    LICENSE+author:          Adam Langley <agl@imperialviolet.org>+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

@@ -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 ()