diff --git a/Control/AutoUpdate.hs b/Control/AutoUpdate.hs --- a/Control/AutoUpdate.hs +++ b/Control/AutoUpdate.hs @@ -1,6 +1,5 @@ -{-# LANGUAGE BangPatterns #-} {-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE CPP #-} +{-# LANGUAGE RecordWildCards #-} -- | 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 @@ -12,15 +11,19 @@ -- This library allows you to define actions which will either be -- performed by a dedicated thread or, in times of low volume, will be -- executed by the calling thread. -module Control.AutoUpdate - ( UpdateSettings +module Control.AutoUpdate ( + -- * Type + UpdateSettings , defaultUpdateSettings + -- * Accessors , updateFreq , updateSpawnThreshold , updateAction + -- * Creation , mkAutoUpdate ) where +import Control.AutoUpdate.Util (atomicModifyIORef') import Control.Concurrent (ThreadId, forkIO, myThreadId, threadDelay) import Control.Exception (Exception, SomeException ,assert, fromException, handle,throwIO, throwTo) @@ -28,25 +31,6 @@ import Data.IORef (IORef, newIORef) import Data.Typeable (Typeable) -#ifndef MIN_VERSION_base -#define MIN_VERSION_base(x,y,z) 1 -#endif - -#if MIN_VERSION_base(4,6,0) -import Data.IORef (atomicModifyIORef') -#else -import Data.IORef (atomicModifyIORef) --- | Strict version of 'atomicModifyIORef'. This forces both the value stored --- in the 'IORef' as well as the value returned. -atomicModifyIORef' :: IORef a -> (a -> (a,b)) -> IO b -atomicModifyIORef' ref f = do - c <- atomicModifyIORef ref - (\x -> let (a, b) = f x -- Lazy application of "f" - in (a, a `seq` b)) -- Lazy application of "seq" - -- The following forces "a `seq` b", so it also forces "f x". - c `seq` return c -#endif - -- | Default value for creating an @UpdateSettings@. -- -- Since 0.1.0 @@ -106,9 +90,9 @@ -- -- Since 0.1.0 mkAutoUpdate :: UpdateSettings a -> IO (IO a) -mkAutoUpdate (UpdateSettings !f !t !a) = do +mkAutoUpdate us = do istatus <- newIORef $ ManualUpdates 0 - return $! getCurrent f t a istatus + return $! getCurrent us istatus data Action a = Return a | Manual | Spawn @@ -119,26 +103,24 @@ -- computed manually in the current thread. -- -- Since 0.1.0 -getCurrent :: Int -- ^ frequency - -> Int -- ^ spawn threshold - -> IO a -- ^ internal update action +getCurrent :: UpdateSettings a -> IORef (Status a) -- ^ mutable state -> IO a -getCurrent freq spawnThreshold update istatus = do +getCurrent settings@UpdateSettings{..} istatus = do ea <- atomicModifyIORef' istatus increment case ea of Return a -> return a - Manual -> update + Manual -> updateAction Spawn -> do - a <- update - tid <- forkIO $ spawn freq update istatus + 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 > spawnThreshold then Spawn else Manual + act = if i > updateSpawnThreshold then Spawn else Manual -- Normal case. turnToAuto a tid (ManualUpdates cnt) = (AutoUpdated a cnt tid @@ -148,10 +130,10 @@ turnToAuto a tid (AutoUpdated _ cnt old) = (AutoUpdated a cnt tid ,throwTo old Replaced) -spawn :: Int -> IO a -> IORef (Status a) -> IO () -spawn freq update istatus = handle (onErr istatus) $ forever $ do - threadDelay freq - a <- update +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. @@ -163,7 +145,7 @@ onErr :: IORef (Status a) -> SomeException -> IO () onErr istatus ex = case fromException ex of - Just Replaced -> return () + Just Replaced -> return () -- this thread is terminated Nothing -> do tid <- myThreadId atomicModifyIORef' istatus $ clear tid @@ -180,5 +162,6 @@ clear tid (AutoUpdated _ _ tid') | tid == tid' = (ManualUpdates 0, ()) clear _ status = (status, ()) +-- | Throw an error to kill a thread. stop :: IO a stop = throwIO Replaced diff --git a/Control/AutoUpdate/Util.hs b/Control/AutoUpdate/Util.hs new file mode 100644 --- /dev/null +++ b/Control/AutoUpdate/Util.hs @@ -0,0 +1,23 @@ +{-# LANGUAGE CPP #-} +module Control.AutoUpdate.Util + ( atomicModifyIORef' + ) where + +#ifndef MIN_VERSION_base +#define MIN_VERSION_base(x,y,z) 1 +#endif + +#if MIN_VERSION_base(4,6,0) +import Data.IORef (atomicModifyIORef') +#else +import Data.IORef (atomicModifyIORef) +-- | Strict version of 'atomicModifyIORef'. This forces both the value stored +-- in the 'IORef' as well as the value returned. +atomicModifyIORef' :: IORef a -> (a -> (a,b)) -> IO b +atomicModifyIORef' ref f = do + c <- atomicModifyIORef ref + (\x -> let (a, b) = f x -- Lazy application of "f" + in (a, a `seq` b)) -- Lazy application of "seq" + -- The following forces "a `seq` b", so it also forces "f x". + c `seq` return c +#endif diff --git a/Control/Reaper.hs b/Control/Reaper.hs new file mode 100644 --- /dev/null +++ b/Control/Reaper.hs @@ -0,0 +1,185 @@ +{-# LANGUAGE RecordWildCards #-} + +-- | This module provides the ability to create reapers: dedicated cleanup +-- threads. These threads will automatically spawn and die based on the +-- presence of a workload to process on. +module Control.Reaper ( + -- * Settings + ReaperSettings + , defaultReaperSettings + -- * Accessors + , reaperAction + , reaperDelay + , reaperCons + , reaperNull + , reaperEmpty + -- * Type + , Reaper(..) + -- * Creation + , mkReaper + -- * Helper + , mkListAction + ) where + +import Control.AutoUpdate.Util (atomicModifyIORef') +import Control.Concurrent (forkIO, threadDelay) +import Control.Exception (mask_) +import Control.Monad (join, void) +import Data.IORef (IORef, newIORef, readIORef) + +-- | Settings for creating a reaper. This type has two parameters: +-- @workload@ gives the entire workload, whereas @item@ gives an +-- individual piece of the queue. A common approach is to have @workload@ +-- be a list of @item@s. This is encouraged by 'defaultReaperSettings' and +-- 'mkListAction'. +-- +-- Since 0.1.1 +data ReaperSettings workload item = ReaperSettings + { reaperAction :: workload -> IO (workload -> workload) + -- ^ The action to perform on a workload. The result of this is a + -- \"workload modifying\" function. In the common case of using lists, + -- the result should be a difference list that prepends the remaining + -- workload to the temporary workload. For help with setting up such + -- an action, see 'mkListAction'. + -- + -- Default: do nothing with the workload, and then prepend it to the + -- temporary workload. This is incredibly useless; you should + -- definitely override this default. + -- + -- Since 0.1.1 + , reaperDelay :: {-# UNPACK #-} !Int + -- ^ Number of microseconds to delay between calls of 'reaperAction'. + -- + -- Default: 30 seconds. + -- + -- Since 0.1.1 + , reaperCons :: item -> workload -> workload + -- ^ Add an item onto a workload. + -- + -- Default: list consing. + -- + -- Since 0.1.1 + , reaperNull :: workload -> Bool + -- ^ Check if a workload is empty, in which case the worker thread + -- will shut down. + -- + -- Default: 'null'. + -- + -- Since 0.1.1 + , reaperEmpty :: workload + -- ^ An empty workload. + -- + -- Default: empty list. + -- + -- Since 0.1.1 + } + +-- | Default @ReaperSettings@ value, biased towards having a list of work +-- items. +-- +-- Since 0.1.1 +defaultReaperSettings :: ReaperSettings [item] item +defaultReaperSettings = ReaperSettings + { reaperAction = \wl -> return (wl ++) + , reaperDelay = 30000000 + , reaperCons = (:) + , reaperNull = null + , reaperEmpty = [] + } + +-- | A data structure to hold reaper APIs. +data Reaper workload item = Reaper { + -- | Adding an item to the workload + reaperAdd :: item -> IO () + -- | Reading workload. + , reaperRead :: IO workload + -- | Stopping the reaper thread if exists. + -- The current workload is returned. + , reaperStop :: IO workload + } + +-- | State of reaper. +data State workload = NoReaper -- ^ No reaper thread + | Workload workload -- ^ The current jobs + +-- | Create a reaper addition function. This funciton can be used to add +-- new items to the workload. Spawning of reaper threads will be handled +-- for you automatically. +-- +-- Since 0.1.1 +mkReaper :: ReaperSettings workload item -> IO (Reaper workload item) +mkReaper settings@ReaperSettings{..} = do + stateRef <- newIORef NoReaper + return Reaper { + reaperAdd = update settings stateRef + , reaperRead = readRef stateRef + , reaperStop = stop stateRef + } + where + readRef stateRef = do + mx <- readIORef stateRef + case mx of + NoReaper -> return reaperEmpty + Workload wl -> return wl + stop stateRef = atomicModifyIORef' stateRef $ \mx -> + case mx of + NoReaper -> (NoReaper, reaperEmpty) + Workload x -> (Workload reaperEmpty, x) + +update :: ReaperSettings workload item -> IORef (State workload) -> item + -> IO () +update settings@ReaperSettings{..} stateRef item = + mask_ $ join $ atomicModifyIORef' stateRef cons + where + cons NoReaper = (Workload $ reaperCons item reaperEmpty + ,spawn settings stateRef) + cons (Workload wl) = (Workload $ reaperCons item wl + ,return ()) + +spawn :: ReaperSettings workload item -> IORef (State workload) -> IO () +spawn settings stateRef = void . forkIO $ reaper settings stateRef + +reaper :: ReaperSettings workload item -> IORef (State workload) -> IO () +reaper settings@ReaperSettings{..} stateRef = do + threadDelay reaperDelay + -- Getting the current jobs. Push an empty job to the reference. + wl <- atomicModifyIORef' stateRef swapWithEmpty + -- Do the jobs. A function to merge the left jobs and + -- new jobs is returned. + merge <- reaperAction wl + -- Merging the left jobs and new jobs. + -- If there is no jobs, this thread finishes. + join $ atomicModifyIORef' stateRef (check merge) + where + swapWithEmpty NoReaper = error "Control.Reaper.reaper: unexpected NoReaper (1)" + swapWithEmpty (Workload wl) = (Workload reaperEmpty, wl) + + check _ NoReaper = error "Control.Reaper.reaper: unexpected NoReaper (2)" + check merge (Workload wl) + -- If there is no job, reaper is terminated. + | reaperNull wl' = (NoReaper, return ()) + -- If there are jobs, carry them out. + | otherwise = (Workload wl', reaper settings stateRef) + where + wl' = merge wl + +-- | A helper function for creating 'reaperAction' functions. You would +-- provide this function with a function to process a single work item and +-- return either a new work item, or @Nothing@ if the work item is +-- expired. +-- +-- Since 0.1.1 +mkListAction :: (item -> IO (Maybe item')) + -> [item] + -> IO ([item'] -> [item']) +mkListAction f = + go id + where + go front [] = return front + go front (x:xs) = do + my <- f x + let front' = + case my of + Nothing -> front + Just y -> front . (y:) + go front' xs 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.0.0 +version: 0.1.1.0 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. @@ -17,6 +17,8 @@ library exposed-modules: Control.AutoUpdate + Control.Reaper + other-modules: Control.AutoUpdate.Util build-depends: base >= 4 && < 5 default-language: Haskell2010