diff --git a/Control/Event.hs b/Control/Event.hs --- a/Control/Event.hs +++ b/Control/Event.hs @@ -18,52 +18,48 @@ -- On the other hand, a shim could be made to provide the -- control-timeout API with Control.Event running under the hood. module Control.Event ( - Event - ,EventId + EventId ,EventSystem ,noEvent ,initEventSystem - ,setEventPreprocessing ,addEvent ,addEventSTM ,cancelEvent ,cancelEventSTM - ,alwaysForkEvents - ,neverForkEvents - ,printMessageOnEvents + ,evtSystemSize ) where +import Prelude hiding (lookup) import Control.Concurrent (forkIO, myThreadId, ThreadId, threadDelay) import Control.Concurrent.STM import Control.Exception (throwDynTo, catchDyn, block, unblock) import Control.Monad (forever) import Data.Dynamic -import Data.List (partition) -import Data.Map (Map, empty) -import Data.Word (Word32) +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) --- |IDs the program can use to cancel previously scheduled events. -data EventId = EvtId ClockTime Word32 deriving (Eq, Ord, Show) -noEvent = EvtId (TOD (-1) (-1)) 0 +import GHC.Conc --- |The basic event structure, exported to allow custom function for setEventPreprocessing -data Event = Evt { - evtId :: EventId, - evtAction :: IO (), - evtUseOwnThread :: Bool - } +type EventNumber = Int +type EventSet = (EventNumber, Map EventNumber (IO ())) +singletonSet :: (IO ()) -> EventSet +singletonSet a = (1, singleton 0 a) -evtExpire :: Event -> ClockTime -evtExpire (Evt (EvtId clk _) _ _) = clk +-- |IDs the program can use to cancel 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) + -- |The event system can either be initilized and passed as state or a global -- system can be declared using gEvtSys = unsafePeformIO initEventSystem data EventSystem = EvtSys { - esEvents :: TVar [Event], -- Pending Events + esEvents :: TVar (Map ClockTime EventSet), -- Pending Events esThread :: TVar (Maybe ThreadId), -- Id of thread for TimerReset exceptions esAlarm :: TVar ClockTime, -- Time of soonest event - esPreProcessing :: TVar (Event -> Event) -- Event preprocessing + esNewAlarm :: TVar Bool, -- An event w/ earlier expiration was added + esExpired :: TVar [[EventSet]] } -- |The only way to get an event system is to initilize one, which sets internal TVars @@ -71,16 +67,16 @@ -- event expiring before the current alarm). initEventSystem :: IO EventSystem initEventSystem = do - evts <- newTVarIO ([] :: [Event]) + evts <- newTVarIO empty tid <- newTVarIO Nothing alm <- newTVarIO (TOD (-1) (-1)) - pp <- newTVarIO (id :: (Event -> Event)) - let evtSys = EvtSys evts tid alm pp + new <- newTVarIO False + exp <- newTVarIO [] + let evtSys = EvtSys evts tid alm new exp forkIO $ forever $ trackAlarm evtSys + forkIO $ forever $ monitorExpiredQueue exp forkIO $ expireEvents evtSys return evtSys - where - maxEvents = show (maxBound :: Word32) -- |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. @@ -96,41 +92,71 @@ -- |Worker function for expireEvents - the parent simply catches the exceptions expireEvents' :: EventSystem -> IO () expireEvents' evtSys = do - usDelay <- determineDelay - threadDelay usDelay - runExpire evtSys + usDelay <- determineDelay + threadDelay usDelay + runExpire evtSys where determineDelay :: IO Int determineDelay = do alm <- atomically (do - alm <- readTVar (esAlarm evtSys) - case alm of - (TOD (-1) (-1)) -> retry - time -> return time ) + evts <- readTVar (esEvents evtSys) + case findMinM evts of + Nothing -> retry + Just (c,_) -> return c ) now <- getClockTime return $ timeDiffToMicroSec $ diffClockTimes alm now + findMinM :: Map ClockTime EventSet -> Maybe (ClockTime,EventSet) + findMinM m | size m == 0 = Nothing + | otherwise = Just $ findMin m + -- |Determines which events are expired, running all their actions runExpire :: EventSystem -> IO () runExpire evtSys = do - now <- getClockTime - evts <- atomically (do - evts <- readTVar (esEvents evtSys) - let (expired, remain) = partition ((< now) . evtExpire) evts - newAlarm = getAlarm remain - writeTVar (esEvents evtSys) remain - writeTVar (esAlarm evtSys) newAlarm - return expired ) - runEvents evts + now <- getClockTime + atomically (do evts <- readTVar (esEvents evtSys) + let (exp, newMap) = getEarlierKeys now evts + 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 - getAlarm [] = TOD (-1) (-1) - getAlarm (e:_) = evtExpire e + getEarlierKeys :: ClockTime -> Map ClockTime EventSet -> ([EventSet], Map ClockTime EventSet) + getEarlierKeys clk m = + case deleteFindMinM m of + Just ((k,es), m') -> + if k < clk + then let (exp, lastMap) = getEarlierKeys clk m' + in (es:exp, lastMap) + else ([], m) + Nothing -> ([], m) + getAlarm m | size m == 0 = TOD (-1) (-1) + | otherwise = fst $ findMin m + + deleteFindMinM :: Map k a -> Maybe ((k, a), Map k a) + deleteFindMinM m = if size m == 0 then Nothing else Just (deleteFindMin m) + +-- |Execute expired events +monitorExpiredQueue :: TVar [[EventSet]] -> IO () +monitorExpiredQueue exp = do + exp <- atomically (do + e <- readTVar exp + case e of + (a:as) -> writeTVar exp [] >> return e + _ -> retry ) + mapM_ (mapM_ runEvents) exp + -- |Runs all provided events (which must have expired) -runEvents :: [Event] -> IO () -runEvents [] = return () -runEvents ((Evt _ act True) :es) = (forkIO act) >> runEvents es -runEvents ((Evt _ act _ ) :es) = act >> runEvents es +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! -- |Add an *action* to be performed at *time* by *system*, returning a unique id. addEvent :: EventSystem -> ClockTime -> IO () -> IO EventId @@ -140,23 +166,25 @@ addEventSTM :: EventSystem -> ClockTime -> IO () -> STM EventId addEventSTM sys clk act = do evts <- readTVar (esEvents sys) - pp <- readTVar (esPreProcessing sys) - let evt = pp $ Evt (EvtId clk 0) act True - (newEvts, eid) = insertEvent evts evt 0 - writeTVar (esEvents sys) newEvts + let (old, newMap) = insertLookupWithKey (\_ _ o -> insertEvent o) clk (singletonSet act) evts + num = case old of + Nothing -> 0 + Just (n,_) -> n + eid = EvtId clk num + writeTVar (esEvents sys) newMap alm <- readTVar (esAlarm sys) if clk < alm || alm == (TOD (-1) (-1)) - then writeTVar (esAlarm sys) clk + 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 () return eid where - insertEvent :: [Event] -> Event -> Word32 -> ([Event],EventId) - insertEvent [] n i = let eid = EvtId clk i in ([n { evtId = EvtId clk i }], eid) - insertEvent (e:es) n i - | i == maxBound = error "Unreasonably large number of events (maxBound::Word32)" - | evtExpire n == evtExpire e = let (lst,eid) = insertEvent es n (i+1) in (e:lst, eid) - | evtExpire n < evtExpire e = let eid = EvtId clk i in (n { evtId = eid } : e : es, eid) - | otherwise = let (lst, eid) = insertEvent es n i in (e:lst, eid) + insertEvent :: EventSet -> EventSet + insertEvent (num,set) | num == maxBound = error "maxBound events at given time, something is broken." + | otherwise = + (num+1, insert num act set) -- |Cancel an event from the system, returning True on success. cancelEvent :: EventSystem -> EventId -> IO Bool @@ -164,39 +192,37 @@ -- |Atomically cancel an event from the system, returning True on success. cancelEventSTM :: EventSystem -> EventId -> STM Bool -cancelEventSTM sys eid = do +cancelEventSTM sys eid@(EvtId clk num) = do evts <- readTVar (esEvents sys) - let (newEvts,ret) = deleteOrd evts eid - writeTVar (esEvents sys) newEvts + let newMap :: Map ClockTime EventSet + prev :: Maybe EventSet + (prev,newMap) = updateLookupWithKey (\_ (num, old) -> Just (num,delete num old)) clk evts + ret = case prev of + Nothing -> False -- error "Canceling an event that never existed." + Just (_,p) -> case lookup clk newMap of + Nothing -> False + Just (_,m) -> (size p /= size m) + writeTVar (esEvents sys) newMap return ret - where - deleteOrd :: [Event] -> EventId -> ([Event],Bool) - deleteOrd [] _ = ([],False) - deleteOrd (e:es) eid@(EvtId clk num) - | evtExpire e > clk = (e : es, False) - | evtId e == eid = (es, True) - | otherwise = let (es',ret) = (deleteOrd es eid) in (e : es', ret) +evtSystemSize :: EventSystem -> STM Int +evtSystemSize sys = do + evts <- readTVar (esEvents sys) + return $ fold (\(_,m) n -> n + size m) 0 evts + -- |Tracks the alarm time and the earliest event. If an earlier event is added -- the alarm time is updated and TimerReset is thrown to the expireEvent thread trackAlarm :: EventSystem -> IO () trackAlarm sys = do tid <- atomically (do + newAlm <- readTVar (esNewAlarm sys) + if newAlm then writeTVar (esNewAlarm sys) False else retry + tid <- readTVar (esThread sys) i <- case tid of Just i -> return i Nothing -> retry - - alm <- readTVar (esAlarm sys) - evts <- readTVar (esEvents sys) - case evts of - [] -> retry - (e:_) -> - if alm > evtExpire e || alm == (TOD (-1) (-1)) - then writeTVar (esAlarm sys) (evtExpire e) - else retry - return i - ) + return i ) throwDynTo tid TimerReset -- |Returns the time difference in microseconds (potentially returning maxBound <= the real difference) @@ -210,27 +236,3 @@ realTime = (fromIntegral sec) * (10^6) + fromIntegral (picosec `div` (10^6)) data TimerReset = TimerReset deriving (Eq, Ord, Show, Typeable) - --- Start extra bloat here -- - --- |You can set a modifier on each event as its added to the event system. --- alwaysForkEvents and neverForkEvents are two such premade functions. --- Default is 'id'. -setEventPreprocessing :: EventSystem -> (Event -> Event) -> STM () -setEventPreprocessing sys pp = writeTVar (esPreProcessing sys) pp - --- |A premade event preprocessor to always fork a new Haskell thread for each event (default) --- If used as an argument to *setEventPreprocessor* it will effect all newly added events. -alwaysForkEvents :: Event -> Event -alwaysForkEvents evt = evt { evtUseOwnThread = True } - --- |A premade event preprocessor to never fork a Haskell thread for new events. --- If used as an argument to *setEventPreprocessor* it will effect all newly added events. --- In testing this harms the event system scalability, but it might save a bit of CPU time --- in some use cases. -neverForkEvents :: Event -> Event -neverForkEvents e = e { evtUseOwnThread = False } - --- |A premade event preprocessor that prints the provided debug message before each event. -printMessageOnEvents :: String -> Event -> Event -printMessageOnEvents str = \evt -> evt { evtAction = putStrLn str >> evtAction evt } diff --git a/Control/Event/Timeout.hs b/Control/Event/Timeout.hs new file mode 100644 --- /dev/null +++ b/Control/Event/Timeout.hs @@ -0,0 +1,39 @@ +-- | +-- This module is a shim, providing the control-timeout api using control-event to run the show + +module Control.Event.Timeout ( + addTimeout + , addTimeoutAtomic + , cancelTimeout + , TimeoutTag) where + +import System.IO.Unsafe +import Control.Concurrent.STM +import Control.Event +import System.Time + +evtSys = unsafePerformIO initEventSystem + +newtype TimeoutTag = TTag EventId + +addTimeout :: Float -> (IO ()) -> IO TimeoutTag +addTimeout delay act = do + clk <- getExpireTime delay + i <- addEvent evtSys clk act + return $ TTag i + +addTimeoutAtomic :: Float -> IO () -> IO (STM TimeoutTag) +addTimeoutAtomic delay act = do + clk <- getExpireTime delay + return $ addEventSTM evtSys clk act >>= return . TTag + +cancelTimeout :: TimeoutTag -> STM Bool +cancelTimeout (TTag eid) = cancelEventSTM evtSys eid + +getExpireTime :: Float -> IO ClockTime +getExpireTime delay = do + (TOD sec ps) <- getClockTime + let dSec = truncate delay + dPS = truncate $ (delay - fromIntegral dSec) * 10^12 + clk = TOD (sec + dSec) (ps + dPS) + return clk diff --git a/Test/Test.hs b/Test/Test.hs --- a/Test/Test.hs +++ b/Test/Test.hs @@ -16,13 +16,13 @@ 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 "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 "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 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.0 +version: 0.1 synopsis: Event scheduling system. description: Allows scheduling and canceling of IO actions to be executed at a specified future time. @@ -11,7 +11,7 @@ build-type: Simple build-Depends: base, old-time, containers >= 0.1, stm ghc-options: DeriveDataTypeable -exposed-modules: Control.Event +exposed-modules: Control.Event, Control.Event.Timeout stability: alpha tested-with: GHC == 6.8.2 extra-source-files: Test/Test.hs