diff --git a/Control/AutoUpdate.hs b/Control/AutoUpdate.hs --- a/Control/AutoUpdate.hs +++ b/Control/AutoUpdate.hs @@ -23,13 +23,12 @@ , mkAutoUpdate ) where -import Control.AutoUpdate.Util (atomicModifyIORef') -import Control.Concurrent (ThreadId, forkIO, myThreadId, threadDelay) -import Control.Exception (Exception, SomeException - ,assert, fromException, handle,throwIO, throwTo) -import Control.Monad (forever, join) -import Data.IORef (IORef, newIORef) -import Data.Typeable (Typeable) +import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent.MVar (newEmptyMVar, putMVar, readMVar, + takeMVar, tryPutMVar, tryTakeMVar) +import Control.Exception (SomeException, catch, throw) +import Control.Monad (forever, void) +import Data.IORef (newIORef, readIORef, writeIORef) -- | Default value for creating an @UpdateSettings@. -- @@ -60,9 +59,12 @@ -- -- Since 0.1.0 , updateSpawnThreshold :: Int - -- ^ How many times the data must be requested before we decide to - -- spawn a dedicated thread. + -- ^ NOTE: This value no longer has any effect, since worker threads are + -- dedicated instead of spawned on demand. -- + -- Previously, this determined: How many times the data must be requested + -- before we decide to spawn a dedicated thread. + -- -- Default: 3 -- -- Since 0.1.0 @@ -74,94 +76,57 @@ -- Since 0.1.0 } -data Status a = AutoUpdated - !a - {-# UNPACK #-} !Int - -- Number of times used since last updated. - {-# UNPACK #-} !ThreadId - -- Worker thread. - | ManualUpdates - {-# UNPACK #-} !Int - -- Number of times used since we started/switched - -- off manual updates. - -- | Generate an action which will either read from an automatically -- updated value, or run the update action in the current thread. -- -- Since 0.1.0 mkAutoUpdate :: UpdateSettings a -> IO (IO a) mkAutoUpdate us = do - istatus <- newIORef $ ManualUpdates 0 - return $! getCurrent us istatus + -- The current value, if available. + currRef <- newIORef Nothing -data Action a = Return a | Manual | Spawn + -- A baton to tell the worker thread to generate a new value. + needsRunning <- newEmptyMVar -data Replaced = Replaced deriving (Show, Typeable) -instance Exception Replaced + -- The last value generated, to allow for blocking semantics when currRef + -- is Nothing. + lastValue <- newEmptyMVar --- | Get the current value, either fed from an auto-update thread, or --- computed manually in the current thread. --- --- Since 0.1.0 -getCurrent :: UpdateSettings a - -> IORef (Status a) -- ^ mutable state - -> IO a -getCurrent settings@UpdateSettings{..} istatus = do - ea <- atomicModifyIORef' istatus increment - case ea of - Return a -> return a - Manual -> updateAction - Spawn -> do - a <- updateAction - tid <- forkIO $ spawn settings istatus - join $ atomicModifyIORef' istatus $ turnToAuto a tid - return a - where - increment (AutoUpdated a cnt tid) = (AutoUpdated a (succ cnt) tid, Return a) - increment (ManualUpdates i) = (ManualUpdates (succ i), act) - where - act = if i > updateSpawnThreshold then Spawn else Manual + -- fork the worker thread immediately... + void $ forkIO $ forever $ do + -- but block until a value is actually needed + takeMVar needsRunning - -- Normal case. - turnToAuto a tid (ManualUpdates cnt) = (AutoUpdated a cnt tid - ,return ()) - -- Race condition: multiple threads were spawned. - -- So, let's kill the previous one by this thread. - turnToAuto a tid (AutoUpdated _ cnt old) = (AutoUpdated a cnt tid - ,throwTo old Replaced) + -- new value requested, so run the updateAction + a <- catchSome $ updateAction us -spawn :: UpdateSettings a -> IORef (Status a) -> IO () -spawn UpdateSettings{..} istatus = handle (onErr istatus) $ forever $ do - threadDelay updateFreq - a <- updateAction - join $ atomicModifyIORef' istatus $ turnToManual a - where - -- Normal case. - turnToManual a (AutoUpdated _ cnt tid) - | cnt >= 1 = (AutoUpdated a 0 tid, return ()) - | otherwise = (ManualUpdates 0, stop) - -- This case must not happen. - turnToManual _ (ManualUpdates i) = assert False (ManualUpdates i, stop) + -- we got a new value, update currRef and lastValue + writeIORef currRef $ Just a + void $ tryTakeMVar lastValue + putMVar lastValue a -onErr :: IORef (Status a) -> SomeException -> IO () -onErr istatus ex = case fromException ex of - Just Replaced -> return () -- this thread is terminated - Nothing -> do - tid <- myThreadId - atomicModifyIORef' istatus $ clear tid - throwIO ex - where - -- In the race condition described above, - -- suppose thread A is running, and is killed by thread B. - -- Thread B then updates the IORef to refer to thread B. - -- Then thread A's exception handler fires. - -- We don't want to modify the IORef at all, - -- since it refers to thread B already. - -- Solution: only switch back to manual updates - -- if the IORef is pointing at the current thread. - clear tid (AutoUpdated _ _ tid') | tid == tid' = (ManualUpdates 0, ()) - clear _ status = (status, ()) + -- delay until we're needed again + threadDelay $ updateFreq us --- | Throw an error to kill a thread. -stop :: IO a -stop = throwIO Replaced + -- delay's over, clear out currRef and lastValue so that demanding the + -- value again forces us to start work + writeIORef currRef Nothing + void $ takeMVar lastValue + + return $ do + mval <- readIORef currRef + case mval of + -- we have a current value, use it + Just val -> return val + Nothing -> do + -- no current value, force the worker thread to run... + void $ tryPutMVar needsRunning () + + -- and block for the result from the worker + readMVar lastValue + +-- | Turn a runtime exception into an impure exception, so that all @IO@ +-- actions will complete successfully. This simply defers the exception until +-- the value is forced. +catchSome :: IO a -> IO a +catchSome act = Control.Exception.catch act $ \e -> return $ throw (e :: SomeException) diff --git a/auto-update.cabal b/auto-update.cabal --- a/auto-update.cabal +++ b/auto-update.cabal @@ -1,5 +1,5 @@ name: auto-update -version: 0.1.1.3 +version: 0.1.1.4 synopsis: Efficiently run periodic, on-demand actions description: A common problem is the desire to have an action run at a scheduled interval, but only if it is needed. For example, instead of having every web request result in a new @getCurrentTime@ call, we'd like to have a single worker thread run every second, updating an @IORef@. However, if the request frequency is less than once per second, this is a pessimization, and worse, kills idle GC.