git.iko.soy/typeable/wai
on master help sign in
Merging the fix for "Weak ThreadId still leaks threads" (#488)
0cd3664Kazu Yamamoto
auto-update/Control/Reaper.hs+41−22
compare with master
unifiedsplit
@@ -1,4 +1,5 @@
1{-# LANGUAGE RecordWildCards #-}
1{-# LANGUAGE RecordWildCards #-}
2+{-# LANGUAGE BangPatterns #-}
2
3
3-- | This module provides the ability to create reapers: dedicated cleanup
4-- | This module provides the ability to create reapers: dedicated cleanup
4-- threads. These threads will automatically spawn and die based on the
5-- threads. These threads will automatically spawn and die based on the
@@ -22,10 +23,9 @@ module Control.Reaper (
22 ) where
23 ) where
23
24
24import Control.AutoUpdate.Util (atomicModifyIORef')
25import Control.AutoUpdate.Util (atomicModifyIORef')
25−import Control.Concurrent (forkIO, threadDelay)
26+import Control.Concurrent (forkIO, threadDelay, killThread, ThreadId)
26import Control.Exception (mask_)
27import Control.Exception (mask_)
27−import Control.Monad (join, void)
28−import Data.IORef (IORef, newIORef, readIORef)
28+import Data.IORef (IORef, newIORef, readIORef, writeIORef)
29
29
30-- | Settings for creating a reaper. This type has two parameters:
30-- | Settings for creating a reaper. This type has two parameters:
31-- @workload@ gives the entire workload, whereas @item@ gives an
31-- @workload@ gives the entire workload, whereas @item@ gives an
@@ -96,6 +96,8 @@ data Reaper workload item = Reaper {
96 -- | Stopping the reaper thread if exists.
96 -- | Stopping the reaper thread if exists.
97 -- The current workload is returned.
97 -- The current workload is returned.
98 , reaperStop :: IO workload
98 , reaperStop :: IO workload
99+ -- | Killing the reaper thread immediately if exists.
100+ , reaperKill :: IO ()
99 }
101 }
100
102
101-- | State of reaper.
103-- | State of reaper.
@@ -110,10 +112,12 @@ data State workload = NoReaper -- ^ No reaper thread
110mkReaper :: ReaperSettings workload item -> IO (Reaper workload item)
112mkReaper :: ReaperSettings workload item -> IO (Reaper workload item)
111mkReaper settings@ReaperSettings{..} = do
113mkReaper settings@ReaperSettings{..} = do
112 stateRef <- newIORef NoReaper
114 stateRef <- newIORef NoReaper
115+ tidRef <- newIORef Nothing
113 return Reaper {
116 return Reaper {
114− reaperAdd = update settings stateRef
117+ reaperAdd = add settings stateRef tidRef
115 , reaperRead = readRef stateRef
118 , reaperRead = readRef stateRef
116 , reaperStop = stop stateRef
119 , reaperStop = stop stateRef
120+ , reaperKill = kill tidRef
117 }
121 }
118 where
122 where
119 readRef stateRef = do
123 readRef stateRef = do
@@ -125,31 +129,46 @@ mkReaper settings@ReaperSettings{..} = do
125 case mx of
129 case mx of
126 NoReaper -> (NoReaper, reaperEmpty)
130 NoReaper -> (NoReaper, reaperEmpty)
127 Workload x -> (Workload reaperEmpty, x)
131 Workload x -> (Workload reaperEmpty, x)
132+ kill tidRef = do
133+ mtid <- readIORef tidRef
134+ case mtid of
135+ Nothing -> return ()
136+ Just tid -> killThread tid
128
137
129−update :: ReaperSettings workload item -> IORef (State workload) -> item
138+add :: ReaperSettings workload item
139+ -> IORef (State workload) -> IORef (Maybe ThreadId)
130− -> IO ()
140+ -> item -> IO ()
131−update settings@ReaperSettings{..} stateRef item =
141+add settings@ReaperSettings{..} stateRef tidRef item =
142+ mask_ $ do
132− mask_ $ join $ atomicModifyIORef' stateRef cons
143+ next <- atomicModifyIORef' stateRef cons
144+ next
133 where
145 where
134− cons NoReaper = (Workload $ reaperCons item reaperEmpty
146+ cons NoReaper = let !wl = reaperCons item reaperEmpty
135− ,spawn settings stateRef)
147+ in (Workload wl, spawn settings stateRef tidRef)
136− cons (Workload wl) = (Workload $ reaperCons item wl
148+ cons (Workload wl) = let wl' = reaperCons item wl
137− ,return ())
149+ in (Workload wl', return ())
138
150
139−spawn :: ReaperSettings workload item -> IORef (State workload) -> IO ()
151+spawn :: ReaperSettings workload item
152+ -> IORef (State workload) -> IORef (Maybe ThreadId)
153+ -> IO ()
154+spawn settings stateRef tidRef = do
140−spawn settings stateRef = void . forkIO $ reaper settings stateRef
155+ tid <- forkIO $ reaper settings stateRef tidRef
156+ writeIORef tidRef $ Just tid
141
157
142−reaper :: ReaperSettings workload item -> IORef (State workload) -> IO ()
158+reaper :: ReaperSettings workload item
159+ -> IORef (State workload) -> IORef (Maybe ThreadId)
160+ -> IO ()
143−reaper settings@ReaperSettings{..} stateRef = do
161+reaper settings@ReaperSettings{..} stateRef tidRef = do
144 threadDelay reaperDelay
162 threadDelay reaperDelay
145 -- Getting the current jobs. Push an empty job to the reference.
163 -- Getting the current jobs. Push an empty job to the reference.
146 wl <- atomicModifyIORef' stateRef swapWithEmpty
164 wl <- atomicModifyIORef' stateRef swapWithEmpty
147 -- Do the jobs. A function to merge the left jobs and
165 -- Do the jobs. A function to merge the left jobs and
148 -- new jobs is returned.
166 -- new jobs is returned.
149− merge <- reaperAction wl
167+ !merge <- reaperAction wl
150 -- Merging the left jobs and new jobs.
168 -- Merging the left jobs and new jobs.
151 -- If there is no jobs, this thread finishes.
169 -- If there is no jobs, this thread finishes.
152− join $ atomicModifyIORef' stateRef (check merge)
170+ next <- atomicModifyIORef' stateRef (check merge)
171+ next
153 where
172 where
154 swapWithEmpty NoReaper = error "Control.Reaper.reaper: unexpected NoReaper (1)"
173 swapWithEmpty NoReaper = error "Control.Reaper.reaper: unexpected NoReaper (1)"
155 swapWithEmpty (Workload wl) = (Workload reaperEmpty, wl)
174 swapWithEmpty (Workload wl) = (Workload reaperEmpty, wl)
@@ -157,9 +176,9 @@ reaper settings@ReaperSettings{..} stateRef = do
157 check _ NoReaper = error "Control.Reaper.reaper: unexpected NoReaper (2)"
176 check _ NoReaper = error "Control.Reaper.reaper: unexpected NoReaper (2)"
158 check merge (Workload wl)
177 check merge (Workload wl)
159 -- If there is no job, reaper is terminated.
178 -- If there is no job, reaper is terminated.
160− | reaperNull wl' = (NoReaper, return ())
179+ | reaperNull wl' = (NoReaper, writeIORef tidRef Nothing)
161 -- If there are jobs, carry them out.
180 -- If there are jobs, carry them out.
162− | otherwise = (Workload wl', reaper settings stateRef)
181+ | otherwise = (Workload wl', reaper settings stateRef tidRef)
163 where
182 where
164 wl' = merge wl
183 wl' = merge wl
165
184
@@ -175,8 +194,8 @@ mkListAction :: (item -> IO (Maybe item'))
175mkListAction f =
194mkListAction f =
176 go id
195 go id
177 where
196 where
178− go front [] = return front
197+ go !front [] = return front
179− go front (x:xs) = do
198+ go !front (x:xs) = do
180 my <- f x
199 my <- f x
181 let front' =
200 let front' =
182 case my of
201 case my of
warp/Network/Wai/Handler/Warp/Timeout.hs+34−45
@@ -1,7 +1,4 @@
1−{-# LANGUAGE CPP #-}
2{-# LANGUAGE DeriveDataTypeable #-}
1{-# LANGUAGE DeriveDataTypeable #-}
3−{-# LANGUAGE MagicHash #-}
4−{-# LANGUAGE UnboxedTuples #-} -- for GHC 7.4 or earlier
5
2
6module Network.Wai.Handler.Warp.Timeout (
3module Network.Wai.Handler.Warp.Timeout (
7 -- ** Types
4 -- ** Types
@@ -11,6 +8,7 @@ module Network.Wai.Handler.Warp.Timeout (
11 -- ** Manager
8 -- ** Manager
12 , initialize
9 , initialize
13 , stopManager
10 , stopManager
11+ , killManager
14 , withManager
12 , withManager
15 -- ** Registration
13 -- ** Registration
16 , register
14 , register
@@ -24,25 +22,12 @@ module Network.Wai.Handler.Warp.Timeout (
24 , TimeoutThread (..)
22 , TimeoutThread (..)
25 ) where
23 ) where
26
24
27−#ifndef MIN_VERSION_base
28−#define MIN_VERSION_base(x,y,z) 1
29−#endif
30−
31−#if MIN_VERSION_base(4,6,0)
32−import Control.Concurrent (mkWeakThreadId, ThreadId)
33−#else
34−import GHC.Conc (ThreadId(..))
35−import GHC.Exts (mkWeak#)
36−import GHC.IO (IO (IO))
37−#endif
38import Control.Concurrent (myThreadId)
25import Control.Concurrent (myThreadId)
39import qualified Control.Exception as E
26import qualified Control.Exception as E
40−import GHC.Weak (Weak (..))
27+import Control.Reaper
28+import Data.Typeable (Typeable)
41import Network.Wai.Handler.Warp.IORef (IORef)
29import Network.Wai.Handler.Warp.IORef (IORef)
42import qualified Network.Wai.Handler.Warp.IORef as I
30import qualified Network.Wai.Handler.Warp.IORef as I
43−import System.Mem.Weak (deRefWeak)
44−import Data.Typeable (Typeable)
45−import Control.Reaper
46
31
47----------------------------------------------------------------
32----------------------------------------------------------------
48
33
@@ -53,7 +38,7 @@ type Manager = Reaper [Handle] Handle
53type TimeoutAction = IO ()
38type TimeoutAction = IO ()
54
39
55-- | A handle used by 'Manager'
40-- | A handle used by 'Manager'
56−data Handle = Handle TimeoutAction (IORef State)
41+data Handle = Handle !(IORef TimeoutAction) !(IORef State)
57
42
58data State = Active -- Manager turns it to Inactive.
43data State = Active -- Manager turns it to Inactive.
59 | Inactive -- Manager removes it with timeout action.
44 | Inactive -- Manager removes it with timeout action.
@@ -70,10 +55,11 @@ initialize timeout = mkReaper defaultReaperSettings
70 , reaperDelay = timeout
55 , reaperDelay = timeout
71 }
56 }
72 where
57 where
73− prune m@(Handle onTimeout iactive) = do
58+ prune m@(Handle actionRef stateRef) = do
74− state <- I.atomicModifyIORef' iactive (\x -> (inactivate x, x))
59+ state <- I.atomicModifyIORef' stateRef (\x -> (inactivate x, x))
75 case state of
60 case state of
76 Inactive -> do
61 Inactive -> do
62+ onTimeout <- I.readIORef actionRef
77 onTimeout `E.catch` ignoreAll
63 onTimeout `E.catch` ignoreAll
78 return Nothing
64 return Nothing
79 Canceled -> return Nothing
65 Canceled -> return Nothing
@@ -84,38 +70,46 @@ initialize timeout = mkReaper defaultReaperSettings
84
70
85----------------------------------------------------------------
71----------------------------------------------------------------
86
72
87−-- | Stopping timeout manager.
73+-- | Stopping timeout manager with onTimeout fired.
88stopManager :: Manager -> IO ()
74stopManager :: Manager -> IO ()
89stopManager mgr = E.mask_ (reaperStop mgr >>= mapM_ fire)
75stopManager mgr = E.mask_ (reaperStop mgr >>= mapM_ fire)
90 where
76 where
77+ fire (Handle actionRef _) = do
78+ onTimeout <- I.readIORef actionRef
91− fire (Handle onTimeout _) = onTimeout `E.catch` ignoreAll
79+ onTimeout `E.catch` ignoreAll
92
80
93ignoreAll :: E.SomeException -> IO ()
81ignoreAll :: E.SomeException -> IO ()
94ignoreAll _ = return ()
82ignoreAll _ = return ()
95
83
84+-- | Killing timeout manager immediately without firing onTimeout.
85+killManager :: Manager -> IO ()
86+killManager mgr = E.mask_ (reaperStop mgr >>= mapM_ fire)
87+ where
88+ fire (Handle actionRef _) = do
89+ onTimeout <- I.readIORef actionRef
90+ onTimeout `E.catch` ignoreAll
91+
96----------------------------------------------------------------
92----------------------------------------------------------------
97
93
98-- | Registering a timeout action.
94-- | Registering a timeout action.
99register :: Manager -> TimeoutAction -> IO Handle
95register :: Manager -> TimeoutAction -> IO Handle
100register mgr onTimeout = do
96register mgr onTimeout = do
97+ actionRef <- I.newIORef onTimeout
101− iactive <- I.newIORef Active
98+ stateRef <- I.newIORef Active
102− let h = Handle onTimeout iactive
99+ let h = Handle actionRef stateRef
103 reaperAdd mgr h
100 reaperAdd mgr h
104 return h
101 return h
105
102
106-- | Registering a timeout action of killing this thread.
103-- | Registering a timeout action of killing this thread.
107registerKillThread :: Manager -> IO Handle
104registerKillThread :: Manager -> IO Handle
108registerKillThread m = do
105registerKillThread m = do
106+ -- If we hold ThreadId, the stack and data of the thread is leaked.
107+ -- If we hold Weak ThreadId, the stack is released. However, its
108+ -- data is still leaked probably because of a bug of GHC.
109+ -- So, let's just use ThreadId and release ThreadId by
110+ -- overriding the timeout action by "cancel".
109− wtid <- myThreadId >>= mkWeakThreadId
111+ tid <- myThreadId
110− register m $ killIfExist wtid
111−
112−-- If ThreadId is hold referred by a strong reference,
113−-- it leaks even after the thread is killed.
114−-- So, let's use a weak reference so that CG can throw ThreadId away.
115−-- deRefWeak checks if ThreadId referenced by the weak reference
116−-- exists. If exists, it means that the thread is alive.
117−killIfExist :: Weak ThreadId -> TimeoutAction
118−killIfExist wtid = deRefWeak wtid >>= maybe (return ()) (`E.throwTo` TimeoutThread)
112+ register m $ E.throwTo tid TimeoutThread
119
113
120data TimeoutThread = TimeoutThread
114data TimeoutThread = TimeoutThread
121 deriving Typeable
115 deriving Typeable
@@ -123,29 +117,24 @@ instance E.Exception TimeoutThread
123instance Show TimeoutThread where
117instance Show TimeoutThread where
124 show TimeoutThread = "Thread killed by Warp's timeout reaper"
118 show TimeoutThread = "Thread killed by Warp's timeout reaper"
125
119
126−#if !MIN_VERSION_base(4,6,0)
127−mkWeakThreadId :: ThreadId -> IO (Weak ThreadId)
128−mkWeakThreadId t@(ThreadId t#) = IO $ \s ->
129− case mkWeak# t# t Nothing s of
130− (# s1, w #) -> (# s1, Weak w #)
131−#endif
132−
133----------------------------------------------------------------
120----------------------------------------------------------------
134
121
135-- | Setting the state to active.
122-- | Setting the state to active.
136-- 'Manager' turns active to inactive repeatedly.
123-- 'Manager' turns active to inactive repeatedly.
137tickle :: Handle -> IO ()
124tickle :: Handle -> IO ()
138−tickle (Handle _ iactive) = I.writeIORef iactive Active
125+tickle (Handle _ stateRef) = I.writeIORef stateRef Active
139
126
140-- | Setting the state to canceled.
127-- | Setting the state to canceled.
141-- 'Manager' eventually removes this without timeout action.
128-- 'Manager' eventually removes this without timeout action.
142cancel :: Handle -> IO ()
129cancel :: Handle -> IO ()
143−cancel (Handle _ iactive) = I.writeIORef iactive Canceled
130+cancel (Handle actionRef stateRef) = do
131+ I.writeIORef actionRef (return ()) -- ensuring to release ThreadId
132+ I.writeIORef stateRef Canceled
144
133
145-- | Setting the state to paused.
134-- | Setting the state to paused.
146-- 'Manager' does not change the value.
135-- 'Manager' does not change the value.
147pause :: Handle -> IO ()
136pause :: Handle -> IO ()
148−pause (Handle _ iactive) = I.writeIORef iactive Paused
137+pause (Handle _ stateRef) = I.writeIORef stateRef Paused
149
138
150-- | Setting the paused state to active.
139-- | Setting the paused state to active.
151-- This is an alias to 'tickle'.
140-- This is an alias to 'tickle'.