control-event 0.0 → 0.1
raw patch · 4 files changed
+155/−114 lines, 4 files
Files
- Control/Event.hs +108/−106
- Control/Event/Timeout.hs +39/−0
- Test/Test.hs +6/−6
- control-event.cabal +2/−2
Control/Event.hs view
@@ -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 }
+ Control/Event/Timeout.hs view
@@ -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
Test/Test.hs view
@@ -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
control-event.cabal view
@@ -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