diff --git a/Control/Event.hs b/Control/Event.hs --- a/Control/Event.hs +++ b/Control/Event.hs @@ -1,12 +1,8 @@ -- |This module can execute events at specified time. It uses a two thread -- system that allows the STM adding and deleting of new threads without --- requiring later IO actions. An ability to place arbitrary event preprocessing --- when adding each event exists, but in a course grained manner. This --- feature can be expanded on request. +-- requiring later IO actions. -- --- This is very like control-timeout, but was developed separately internal --- operation is similar with a thread sleeping via threadDelay and EventIds --- being based in part on expire time. It differs in that control-event is: +-- This differs from control-timeout in that control-event is: -- * More complex -- * Requires initilization -- * Allows pure STM adding and removing of events (no post STM IO action) @@ -15,8 +11,8 @@ -- (advisable if thread spark is too expensive / computation is cheap) -- * No possible duplication of EventId (theoretical! no real advantage) -- --- On the other hand, a shim could be made to provide the --- control-timeout API with Control.Event running under the hood. +-- A shim has been made providing control-timeout API with +-- Control.Event running under the hood (called Control.Event.Timeout). module Control.Event ( EventId ,EventSystem @@ -33,27 +29,31 @@ import Control.Concurrent (forkIO, myThreadId, ThreadId, threadDelay) import Control.Concurrent.STM import Control.Exception (throwDynTo, catchDyn, block, unblock) -import Control.Monad (forever) +import Control.Monad (forever, when) import Data.Dynamic import Data.List (partition, deleteBy) import Data.Map (Map, empty, findMin, deleteFindMin, insertLookupWithKey, adjust, size, singleton, toList, insert, updateLookupWithKey, delete, lookup, fold) import System.Time (TimeDiff(..), ClockTime(..), diffClockTimes, getClockTime) -import GHC.Conc - type EventNumber = Int type EventSet = (EventNumber, Map EventNumber (IO ())) singletonSet :: (IO ()) -> EventSet singletonSet a = (1, singleton 0 a) --- |IDs the program can use to cancel previously scheduled events. +-- |IDs useful for canceling previously scheduled events. data EventId = EvtId ClockTime EventNumber deriving (Eq, Ord, Show) -- |A value indicating there is no such event. -noEvent = EvtId (TOD (-1) (-1)) (-1) +-- Canceling this event returns True and has no other effect. +noEvent :: EventId +noEvent = EvtId never (-1) --- |The event system can either be initilized and passed as state or a global --- system can be declared using gEvtSys = unsafePeformIO initEventSystem +never :: ClockTime +never = TOD (-1) (-1) + +-- |The event system must be initilized using initEventSystem. +-- More than one event system can be instantiated at once +-- (eg. for non-interference). data EventSystem = EvtSys { esEvents :: TVar (Map ClockTime EventSet), -- Pending Events esThread :: TVar (Maybe ThreadId), -- Id of thread for TimerReset exceptions @@ -62,14 +62,17 @@ esExpired :: TVar [[EventSet]] } --- |The only way to get an event system is to initilize one, which sets internal TVars --- and sparks two threads (one to expire events, one to look if you've added an --- event expiring before the current alarm). +-- |The only way to get an event system is to initilize one + +-- This sets internal TVars and sparks three threads: +-- trackAlarm: Ensured esAlarm remains +-- monitorExpiredQueue: Executes all events on the expired queue +-- expireEvents: When the alarm sounds, moves expired events to expired queue initEventSystem :: IO EventSystem initEventSystem = do evts <- newTVarIO empty tid <- newTVarIO Nothing - alm <- newTVarIO (TOD (-1) (-1)) + alm <- newTVarIO never new <- newTVarIO False exp <- newTVarIO [] let evtSys = EvtSys evts tid alm new exp @@ -78,8 +81,9 @@ forkIO $ expireEvents evtSys return evtSys --- |Main thread that delays till the alarm time then executes any expired events. --- Asynchronous 'TimerReset' exceptions might occur to indicate a new, earlier, alarm time. +-- |Main thread. Delays till the alarm time then executes any expired events. +-- Asynchronous 'TimerReset' exceptions might occur to indicate a new, +-- earlier, alarm time. expireEvents :: EventSystem -> IO () expireEvents es = do block (do @@ -119,9 +123,6 @@ newAlarm = getAlarm newMap writeTVar (esAlarm evtSys) newAlarm writeTVar (esEvents evtSys) newMap - if (fold (\(_,m) n -> size m + n) 0 newMap) + sum (map (\(_,m) -> size m) exp) /= (fold (\(_,m) n -> size m + n) 0 evts) - then unsafeIOToSTM $ putStrLn "Expire is dropping events." - else return () exps <- readTVar (esExpired evtSys) writeTVar (esExpired evtSys) (exp:exps) ) where @@ -135,7 +136,7 @@ else ([], m) Nothing -> ([], m) - getAlarm m | size m == 0 = TOD (-1) (-1) + getAlarm m | size m == 0 = never | otherwise = fst $ findMin m deleteFindMinM :: Map k a -> Maybe ((k, a), Map k a) @@ -155,14 +156,13 @@ runEvents :: EventSet -> IO () runEvents (_,set) = do let actions = map snd (toList set) - -- mapM_ forkIO actions - sequence_ actions -- FIXME why does this line work and not the top! + mapM_ forkIO actions --- |Add an *action* to be performed at *time* by *system*, returning a unique id. +-- |Add an *action* to be performed at *time* by *system*. Returns a unique ID. addEvent :: EventSystem -> ClockTime -> IO () -> IO EventId addEvent sys clk act = atomically (addEventSTM sys clk act) --- |Atomically add an action to be performed at specified time and returning a unique id. +-- |Atomic version of addEvent addEventSTM :: EventSystem -> ClockTime -> IO () -> STM EventId addEventSTM sys clk act = do evts <- readTVar (esEvents sys) @@ -173,12 +173,8 @@ eid = EvtId clk num writeTVar (esEvents sys) newMap alm <- readTVar (esAlarm sys) - if clk < alm || alm == (TOD (-1) (-1)) - then writeTVar (esAlarm sys) clk >> writeTVar (esNewAlarm sys) True - else return () - if fold (\(_,m) n -> n + size m) 0 newMap /= fold (\(_,m) n -> n + size m) 1 evts - then unsafeIOToSTM $ putStrLn "Event mapping has not grown!" - else return () + when (clk < alm || alm == never) + (writeTVar (esAlarm sys) clk >> writeTVar (esNewAlarm sys) True) return eid where insertEvent :: EventSet -> EventSet @@ -190,7 +186,7 @@ cancelEvent :: EventSystem -> EventId -> IO Bool cancelEvent sys eid = atomically (cancelEventSTM sys eid) --- |Atomically cancel an event from the system, returning True on success. +-- |Atomic version of cancelEvent cancelEventSTM :: EventSystem -> EventId -> STM Bool cancelEventSTM sys eid@(EvtId clk num) = do evts <- readTVar (esEvents sys) @@ -202,9 +198,10 @@ Just (_,p) -> case lookup clk newMap of Nothing -> False Just (_,m) -> (size p /= size m) - writeTVar (esEvents sys) newMap - return ret + when (eid /= noEvent) (writeTVar (esEvents sys) newMap) + return (eid == noEvent || ret) +-- |Returns the number of pending events. evtSystemSize :: EventSystem -> STM Int evtSystemSize sys = do evts <- readTVar (esEvents sys) diff --git a/Control/Event/Timeout.hs b/Control/Event/Timeout.hs --- a/Control/Event/Timeout.hs +++ b/Control/Event/Timeout.hs @@ -1,5 +1,5 @@ --- | --- This module is a shim, providing the control-timeout api using control-event to run the show +-- |This module is a shim, providing the control-timeout api using +-- control-event to run the show module Control.Event.Timeout ( addTimeout diff --git a/Test/Test.hs b/Test/Test.hs deleted file mode 100644 --- a/Test/Test.hs +++ /dev/null @@ -1,82 +0,0 @@ -module Main where - -import Control.Event -import Control.Concurrent (threadDelay) -import Control.Concurrent.STM -import System.Time - - -secDelay :: Integer -secDelay = 5 - -waitForEvents :: Int -waitForEvents = 7 - -tol :: Int -tol = 2 -- seconds - -main = do --- runTest "testOneEvent" testOneEvent --- runTest "testManyEvents 50" (testManyEvents 50) --- runTest "testDeletingEvents 50" (testDeletingEvents 50) --- runTest "testOnTime [8,1]" (testOnTime [8,1]) - runTest "testOnTime [8,4,2]" (testOnTime [8,4,2]) --- runTest "testOnTime [2,4,8]" (testOnTime [2,4,8]) --- runTest "testOnTime [1,1,1,1...]" (testOnTime (take 20 (repeat 1))) - -runTest :: String -> (IO Bool) -> IO () -runTest label test = do - putStrLn $ "Running test '" ++ label ++ "'" - res <- test - putStrLn $ "Test completed: " ++ (show res) - -testOneEvent :: IO Bool -testOneEvent = do - tv <- newTVarIO False - sys <- initEventSystem - (TOD sec picosec) <- getClockTime - let clk = TOD (sec + secDelay) picosec - addEvent sys clk (atomically (writeTVar tv True)) - threadDelay $ waitForEvents * 10^6 - atomically (readTVar tv >>= return) - -testManyEvents :: Int -> IO Bool -testManyEvents nrEvts = do - tv <- newTVarIO nrEvts - sys <- initEventSystem - (TOD sec picosec) <- getClockTime - let clks = map (\_ -> TOD (sec + secDelay) picosec) [1..nrEvts] - mapM_ (\clk -> addEvent sys clk (atomically (readTVar tv >>= \x ->writeTVar tv (x - 1)))) clks - threadDelay $ waitForEvents*10^6 - atomically (readTVar tv >>= return . (==0)) - -testDeletingEvents :: Int -> IO Bool -testDeletingEvents nrEvents = do - tv <- newTVarIO nrEvents - sys <- initEventSystem - (TOD sec picosec) <- getClockTime - let clks = map (\_ -> TOD (sec + secDelay) picosec) [1..nrEvents] - ids <- mapM (\clk -> addEvent sys clk (atomically (readTVar tv >>= \x ->writeTVar tv (x - 1)))) clks - mapM_ (cancelEvent sys) ids - threadDelay $ waitForEvents*10^6 - atomically (readTVar tv >>= return . (==nrEvents)) - -testOnTime :: [Integer] -> IO Bool -testOnTime delays = do - tv <- newTVarIO (length delays) - sys <- initEventSystem --- atomically (setEventPreprocessing sys neverForkEvents) - (TOD sec picosec) <- getClockTime - let clks = map (\d -> TOD (sec + d) picosec) delays - mapM_ (\clk -> addEvent sys clk (theTimeIsNow clk tv)) clks - threadDelay $ fromIntegral $ (maximum delays + fromIntegral tol)*10^6 - atomically (readTVar tv >>= return . (==length delays)) - -theTimeIsNow :: ClockTime -> TVar Int -> IO () -theTimeIsNow clk tv = do - now <- getClockTime - let (TimeDiff _ _ _ _ _ sec _) = diffClockTimes now clk - if sec > tol - then do putStrLn ("Time error of " ++ (show sec) ++ " seconds") - atomically (readTVar tv >>= \x -> writeTVar tv (x-1)) - else return () diff --git a/Test/Timeout.hs b/Test/Timeout.hs new file mode 100644 --- /dev/null +++ b/Test/Timeout.hs @@ -0,0 +1,50 @@ +module Main where + +import Control.Concurrent +import Control.Concurrent.STM + +import Data.Time.Clock.POSIX +import Control.Event.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 () diff --git a/Test/TimeoutTest.hs b/Test/TimeoutTest.hs deleted file mode 100644 --- a/Test/TimeoutTest.hs +++ /dev/null @@ -1,50 +0,0 @@ -module Main where - -import Control.Concurrent -import Control.Concurrent.STM - -import Data.Time.Clock.POSIX -import Control.Event.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 () diff --git a/control-event.cabal b/control-event.cabal --- a/control-event.cabal +++ b/control-event.cabal @@ -1,5 +1,5 @@ name: control-event -version: 0.2 +version: 0.3.0 synopsis: Event scheduling system. description: Allows scheduling and canceling of IO actions to be executed at a specified future time. @@ -14,4 +14,4 @@ exposed-modules: Control.Event, Control.Event.Timeout stability: alpha tested-with: GHC == 6.8.2 -extra-source-files: Test/Test.hs +extra-source-files: Test/Event.hs Test/Timeout.hs