git.iko.soy/typeable/wai
on master help sign in
typeable/wai/auto-update/Control/Reaper.hs
280 lines 10 kB history raw
1{-# LANGUAGE RecordWildCards #-}
2{-# LANGUAGE BangPatterns #-}
4-- | This module provides the ability to create reapers: dedicated cleanup
5-- threads. These threads will automatically spawn and die based on the
6-- presence of a workload to process on. Example uses include:
7--
8-- * Killing long-running jobs
9-- * Closing unused connections in a connection pool
10-- * Pruning a cache of old items (see example below)
11--
12-- For real-world usage, search the <https://github.com/yesodweb/wai WAI family of packages>
13-- for imports of "Control.Reaper".
14module Control.Reaper (
15 -- * Example: Regularly cleaning a cache
16 -- $example1
18 -- * Settings
19 ReaperSettings
20 , defaultReaperSettings
21 -- * Accessors
22 , reaperAction
23 , reaperDelay
24 , reaperCons
25 , reaperNull
26 , reaperEmpty
27 -- * Type
28 , Reaper(..)
29 -- * Creation
30 , mkReaper
31 -- * Helper
32 , mkListAction
33 ) where
35import Control.AutoUpdate.Util (atomicModifyIORef')
36import Control.Concurrent (forkIO, threadDelay, killThread, ThreadId)
37import Control.Exception (mask_)
38import Data.IORef (IORef, newIORef, readIORef, writeIORef)
40-- | Settings for creating a reaper. This type has two parameters:
41-- @workload@ gives the entire workload, whereas @item@ gives an
42-- individual piece of the queue. A common approach is to have @workload@
43-- be a list of @item@s. This is encouraged by 'defaultReaperSettings' and
44-- 'mkListAction'.
45--
46-- @since 0.1.1
47data ReaperSettings workload item = ReaperSettings
48 { reaperAction :: workload -> IO (workload -> workload)
49 -- ^ The action to perform on a workload. The result of this is a
50 -- \"workload modifying\" function. In the common case of using lists,
51 -- the result should be a difference list that prepends the remaining
52 -- workload to the temporary workload. The temporary workload here
53 -- refers to items added to the workload while the reaper action is
54 -- running. For help with setting up such an action, see 'mkListAction'.
55 --
56 -- Default: do nothing with the workload, and then prepend it to the
57 -- temporary workload. This is incredibly useless; you should
58 -- definitely override this default.
59 --
60 -- @since 0.1.1
61 , reaperDelay :: {-# UNPACK #-} !Int
62 -- ^ Number of microseconds to delay between calls of 'reaperAction'.
63 --
64 -- Default: 30 seconds.
65 --
66 -- @since 0.1.1
67 , reaperCons :: item -> workload -> workload
68 -- ^ Add an item onto a workload.
69 --
70 -- Default: list consing.
71 --
72 -- @since 0.1.1
73 , reaperNull :: workload -> Bool
74 -- ^ Check if a workload is empty, in which case the worker thread
75 -- will shut down.
76 --
77 -- Default: 'null'.
78 --
79 -- @since 0.1.1
80 , reaperEmpty :: workload
81 -- ^ An empty workload.
82 --
83 -- Default: empty list.
84 --
85 -- @since 0.1.1
86 }
88-- | Default @ReaperSettings@ value, biased towards having a list of work
89-- items.
90--
91-- @since 0.1.1
92defaultReaperSettings :: ReaperSettings [item] item
93defaultReaperSettings = ReaperSettings
94 { reaperAction = \wl -> return (wl ++)
95 , reaperDelay = 30000000
96 , reaperCons = (:)
97 , reaperNull = null
98 , reaperEmpty = []
99 }
101-- | A data structure to hold reaper APIs.
102data Reaper workload item = Reaper {
103 -- | Adding an item to the workload
104 reaperAdd :: item -> IO ()
105 -- | Reading workload.
106 , reaperRead :: IO workload
107 -- | Stopping the reaper thread if exists.
108 -- The current workload is returned.
109 , reaperStop :: IO workload
110 -- | Killing the reaper thread immediately if exists.
111 , reaperKill :: IO ()
112 }
114-- | State of reaper.
115data State workload = NoReaper -- ^ No reaper thread
116 | Workload !workload -- ^ The current jobs
118-- | Create a reaper addition function. This function can be used to add
119-- new items to the workload. Spawning of reaper threads will be handled
120-- for you automatically.
121--
122-- @since 0.1.1
123mkReaper :: ReaperSettings workload item -> IO (Reaper workload item)
124mkReaper settings@ReaperSettings{..} = do
125 stateRef <- newIORef NoReaper
126 tidRef <- newIORef Nothing
127 return Reaper {
128 reaperAdd = add settings stateRef tidRef
129 , reaperRead = readRef stateRef
130 , reaperStop = stop stateRef
131 , reaperKill = kill tidRef
132 }
133 where
134 readRef stateRef = do
135 mx <- readIORef stateRef
136 case mx of
137 NoReaper -> return reaperEmpty
138 Workload wl -> return wl
139 stop stateRef = atomicModifyIORef' stateRef $ \mx ->
140 case mx of
141 NoReaper -> (NoReaper, reaperEmpty)
142 Workload x -> (Workload reaperEmpty, x)
143 kill tidRef = do
144 mtid <- readIORef tidRef
145 case mtid of
146 Nothing -> return ()
147 Just tid -> killThread tid
149add :: ReaperSettings workload item
150 -> IORef (State workload) -> IORef (Maybe ThreadId)
151 -> item -> IO ()
152add settings@ReaperSettings{..} stateRef tidRef item =
153 mask_ $ do
154 next <- atomicModifyIORef' stateRef cons
155 next
156 where
157 cons NoReaper = let wl = reaperCons item reaperEmpty
158 in (Workload wl, spawn settings stateRef tidRef)
159 cons (Workload wl) = let wl' = reaperCons item wl
160 in (Workload wl', return ())
162spawn :: ReaperSettings workload item
163 -> IORef (State workload) -> IORef (Maybe ThreadId)
164 -> IO ()
165spawn settings stateRef tidRef = do
166 tid <- forkIO $ reaper settings stateRef tidRef
167 writeIORef tidRef $ Just tid
169reaper :: ReaperSettings workload item
170 -> IORef (State workload) -> IORef (Maybe ThreadId)
171 -> IO ()
172reaper settings@ReaperSettings{..} stateRef tidRef = do
173 threadDelay reaperDelay
174 -- Getting the current jobs. Push an empty job to the reference.
175 wl <- atomicModifyIORef' stateRef swapWithEmpty
176 -- Do the jobs. A function to merge the left jobs and
177 -- new jobs is returned.
178 !merge <- reaperAction wl
179 -- Merging the left jobs and new jobs.
180 -- If there is no jobs, this thread finishes.
181 next <- atomicModifyIORef' stateRef (check merge)
182 next
183 where
184 swapWithEmpty NoReaper = error "Control.Reaper.reaper: unexpected NoReaper (1)"
185 swapWithEmpty (Workload wl) = (Workload reaperEmpty, wl)
187 check _ NoReaper = error "Control.Reaper.reaper: unexpected NoReaper (2)"
188 check merge (Workload wl)
189 -- If there is no job, reaper is terminated.
190 | reaperNull wl' = (NoReaper, writeIORef tidRef Nothing)
191 -- If there are jobs, carry them out.
192 | otherwise = (Workload wl', reaper settings stateRef tidRef)
193 where
194 wl' = merge wl
196-- | A helper function for creating 'reaperAction' functions. You would
197-- provide this function with a function to process a single work item and
198-- return either a new work item, or @Nothing@ if the work item is
199-- expired.
200--
201-- @since 0.1.1
202mkListAction :: (item -> IO (Maybe item'))
203 -> [item]
204 -> IO ([item'] -> [item'])
205mkListAction f =
206 go id
207 where
208 go !front [] = return front
209 go !front (x:xs) = do
210 my <- f x
211 let front' =
212 case my of
213 Nothing -> front
214 Just y -> front . (y:)
215 go front' xs
217-- $example1
218-- In this example code, we use a 'Data.Map.Strict.Map' to cache fibonacci numbers, and a 'Reaper' to prune the cache.
219--
220-- The @main@ function first creates a 'Reaper', with fields to initialize the
221-- cache ('reaperEmpty'), add items to it ('reaperCons'), and prune it ('reaperAction').
222-- The reaper will run every two seconds ('reaperDelay'), but will stop running while
223-- 'reaperNull' is true.
224--
225-- @main@ then loops infinitely ('Control.Monad.forever'). Each second it calculates the fibonacci number
226-- for a value between 30 and 34, first trying the cache ('reaperRead' and 'Data.Map.Strict.lookup'),
227-- then falling back to manually calculating it (@fib@)
228-- and updating the cache with the result ('reaperAdd')
229--
230-- @clean@ simply removes items cached for more than 10 seconds.
231-- This function is where you would perform IO-related cleanup,
232-- like killing threads or closing connections, if that was the purpose of your reaper.
233--
234-- @
235-- module Main where
236--
237-- import "Data.Time" (UTCTime, getCurrentTime, diffUTCTime)
238-- import "Control.Reaper"
239-- import "Control.Concurrent" (threadDelay)
240-- import "Data.Map.Strict" (Map)
241-- import qualified "Data.Map.Strict" as Map
242-- import "Control.Monad" (forever)
243-- import "System.Random" (getStdRandom, randomR)
244--
245-- fib :: 'Int' -> 'Int'
246-- fib 0 = 0
247-- fib 1 = 1
248-- fib n = fib (n-1) + fib (n-2)
249--
250-- type Cache = 'Data.Map.Strict.Map' 'Int' ('Int', 'Data.Time.Clock.UTCTime')
251--
252-- main :: IO ()
253-- main = do
254-- reaper <- 'mkReaper' 'defaultReaperSettings'
255-- { 'reaperEmpty' = Map.'Data.Map.Strict.empty'
256-- , 'reaperCons' = \\(k, v, time) workload -> Map.'Data.Map.Strict.insert' k (v, time) workload
257-- , 'reaperAction' = clean
258-- , 'reaperDelay' = 1000000 * 2 -- Clean every 2 seconds
259-- , 'reaperNull' = Map.'Data.Map.Strict.null'
260-- }
261-- forever $ do
262-- fibArg <- 'System.Random.getStdRandom' ('System.Random.randomR' (30,34))
263-- cache <- 'reaperRead' reaper
264-- let cachedResult = Map.'Data.Map.Strict.lookup' fibArg cache
265-- case cachedResult of
266-- 'Just' (fibResult, _createdAt) -> 'putStrLn' $ "Found in cache: `fib " ++ 'show' fibArg ++ "` " ++ 'show' fibResult
267-- 'Nothing' -> do
268-- let fibResult = fib fibArg
269-- 'putStrLn' $ "Calculating `fib " ++ 'show' fibArg ++ "` " ++ 'show' fibResult
270-- time <- 'Data.Time.Clock.getCurrentTime'
271-- ('reaperAdd' reaper) (fibArg, fibResult, time)
272-- 'threadDelay' 1000000 -- 1 second
273--
274-- -- Remove items > 10 seconds old
275-- clean :: Cache -> IO (Cache -> Cache)
276-- clean oldMap = do
277-- currentTime <- 'Data.Time.Clock.getCurrentTime'
278-- let pruned = Map.'Data.Map.Strict.filter' (\\(_, createdAt) -> currentTime \`diffUTCTime\` createdAt < 10.0) oldMap
279-- return (\\newData -> Map.'Data.Map.Strict.union' pruned newData)
280-- @