git.iko.soy/typeable/wai
on master help sign in
preventing thread leak.
commit468d99232e39236612db680e2268ab73374ca964 authorKazu Yamamoto <kazu@iij.ad.jp> dateMon 21 Dec 2015, 14:15 (11 years ago) parenta5386c8
5 files changed+39−21
compare with master
unifiedsplit
warp/Network/Wai/Handler/Warp/HTTP2.hs+1−1 warp/Network/Wai/Handler/Warp/HTTP2/Manager.hs+18−7 warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+5−5 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+13−6 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+2−2
warp/Network/Wai/Handler/Warp/HTTP2.hs+1−1
@@ -45,7 +45,7 @@ http2 conn ii addr transport settings readN app = do
4545 -- Sender
4646 -- frameSender is the main thread because it ensures to send
4747 -- a goway frame.
48− frameSender ctx conn ii settings `E.finally` do
48+ frameSender ctx conn ii settings mgr `E.finally` do
4949 clearContext ctx
5050 stop mgr
5151 killThread tid
warp/Network/Wai/Handler/Warp/HTTP2/Manager.hs+18−7
@@ -9,7 +9,8 @@ module Network.Wai.Handler.Warp.HTTP2.Manager (
99 , setAction
1010 , stop
1111 , spawnAction
12− , replaceWithAction
12+ , addMyId
13+ , deleteMyId
1314 ) where
1415
1516#if __GLASGOW_HASKELL__ < 709
@@ -29,7 +30,7 @@ import qualified Network.Wai.Handler.Warp.Timeout as T
2930
3031type Action = T.Manager -> IO ()
3132
32−data Command = Stop | Spawn | Replace ThreadId
33+data Command = Stop | Spawn | Add ThreadId | Delete ThreadId
3334
3435data Manager = Manager (TQueue Command) (IORef Action)
3536
@@ -48,9 +49,12 @@ start set = do
4849 go q !tset0 ref timmgr = do
4950 x <- atomically $ readTQueue q
5051 case x of
51− Stop -> kill tset0 >> T.killManager timmgr
52− Spawn -> next tset0
53− Replace oldtid -> next $ del oldtid tset0
52+ Stop -> kill tset0 >> T.killManager timmgr
53+ Spawn -> next tset0
54+ Add newtid -> let !tset = add newtid tset0
55+ in go q tset ref timmgr
56+ Delete oldtid -> let !tset = del oldtid tset0
57+ in go q tset ref timmgr
5458 where
5559 next tset = do
5660 action <- readIORef ref
@@ -67,8 +71,15 @@ stop (Manager q _) = atomically $ writeTQueue q Stop
6771spawnAction :: Manager -> IO ()
6872spawnAction (Manager q _) = atomically $ writeTQueue q Spawn
6973
70−replaceWithAction :: Manager -> ThreadId -> IO ()
71−replaceWithAction (Manager q _) tid = atomically $ writeTQueue q $ Replace tid
74+addMyId :: Manager -> IO ()
75+addMyId (Manager q _) = do
76+ tid <- myThreadId
77+ atomically $ writeTQueue q $ Add tid
78+
79+deleteMyId :: Manager -> IO ()
80+deleteMyId (Manager q _) = do
81+ tid <- myThreadId
82+ atomically $ writeTQueue q $ Delete tid
7283
7384----------------------------------------------------------------
7485
warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+5−5
@@ -7,7 +7,6 @@ module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
77#if __GLASGOW_HASKELL__ < 709
88import Control.Applicative
99#endif
10−import Control.Concurrent (forkIO)
1110import Control.Concurrent.STM
1211import qualified Control.Exception as E
1312import Control.Monad (unless, void, when)
@@ -22,6 +21,7 @@ import Network.Wai
2221import Network.Wai.Handler.Warp.Buffer
2322import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
2423import Network.Wai.Handler.Warp.HTTP2.HPACK
24+import Network.Wai.Handler.Warp.HTTP2.Manager (Manager)
2525import Network.Wai.Handler.Warp.HTTP2.Types
2626import Network.Wai.Handler.Warp.IORef
2727import qualified Network.Wai.Handler.Warp.Settings as S
@@ -61,10 +61,10 @@ waitStreaming tbq = do
6161 isEmpty <- isEmptyTBQueue tbq
6262 check (isEmpty == False)
6363
64−frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
64+frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> Manager -> IO ()
6565frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}
6666 conn@Connection{connWriteBuffer,connBufferSize,connSendAll}
67− ii settings = loop `E.catch` ignore
67+ ii settings mgr = loop `E.catch` ignore
6868 where
6969 dequeueControl = Left <$> readTQueue controlQ
7070 dequeueOutput = Right <$> do
@@ -155,13 +155,13 @@ frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}
155155 checkStreaming tbq = do
156156 isEmpty <- atomically $ isEmptyTBQueue tbq
157157 if isEmpty then
158− void . forkIO $ enqueueWhenReady (waitStreaming tbq) outputQ out
158+ forkAndEnqueueWhenReady (waitStreaming tbq) outputQ out mgr
159159 else
160160 checkStreamWindowSize
161161 checkStreamWindowSize = do
162162 sws <- getStreamWindowSize strm
163163 if sws == 0 then do
164− void . forkIO $ enqueueWhenReady (waitStreamWindowSize strm) outputQ out
164+ forkAndEnqueueWhenReady (waitStreamWindowSize strm) outputQ out mgr
165165 else
166166 body sws
167167 resetStream e = do
warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+13−6
@@ -8,14 +8,17 @@ import Data.ByteString.Builder (Builder)
88#if __GLASGOW_HASKELL__ < 709
99import Control.Applicative ((<$>),(<*>))
1010#endif
11+import Control.Concurrent (forkIO)
1112import Control.Concurrent.STM
12−import Control.Exception (SomeException)
13+import Control.Exception (SomeException, bracket)
14+import Control.Monad (void)
1315import Data.ByteString (ByteString)
1416import qualified Data.ByteString as BS
1517import Data.IntMap.Strict (IntMap, IntMap)
1618import qualified Data.IntMap.Strict as M
1719import qualified Network.HTTP.Types as H
1820import Network.Wai (Request, Response)
21+import Network.Wai.Handler.Warp.HTTP2.Manager
1922import Network.Wai.Handler.Warp.IORef
2023import Network.Wai.Handler.Warp.Types
2124
@@ -215,11 +218,15 @@ remove (StreamTable ref) k = atomicModifyIORef' ref $ \m ->
215218search :: StreamTable -> M.Key -> IO (Maybe Stream)
216219search (StreamTable ref) k = M.lookup k <$> readIORef ref
217220
218−{-# INLINE enqueueWhenReady #-}
219−enqueueWhenReady :: STM () -> PriorityTree Output -> Output -> IO ()
220−enqueueWhenReady wait outQ out = do
221− atomically wait
222− enqueueOutput outQ out
221+{-# INLINE forkAndEnqueueWhenReady #-}
222+forkAndEnqueueWhenReady :: STM () -> PriorityTree Output -> Output -> Manager -> IO ()
223+forkAndEnqueueWhenReady wait outQ out mgr = bracket setup teardown $ \_ ->
224+ void . forkIO $ do
225+ atomically wait
226+ enqueueOutput outQ out
227+ where
228+ setup = addMyId mgr
229+ teardown _ = deleteMyId mgr
223230
224231{-# INLINE enqueueOutput #-}
225232enqueueOutput :: PriorityTree Output -> Output -> IO ()
warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+2−2
@@ -15,7 +15,6 @@ module Network.Wai.Handler.Warp.HTTP2.Worker (
1515import Control.Applicative
1616import Data.Monoid (mempty)
1717#endif
18−import Control.Concurrent
1918import Control.Concurrent.STM
2019import Control.Exception (SomeException(..), AsyncException(..))
2120import qualified Control.Exception as E
@@ -111,7 +110,7 @@ response ii settings Context{outputQ} mgr tconf th strm req rsp
111110 -- So, this work occupies this thread.
112111 --
113112 -- We need to increase the number of workers.
114− myThreadId >>= replaceWithAction mgr
113+ spawnAction mgr
115114 -- After this work, this thread stops to decease
116115 -- the number of workers.
117116 setThreadContinue tconf False
@@ -126,6 +125,7 @@ response ii settings Context{outputQ} mgr tconf th strm req rsp
126125 flush = atomically $ writeTBQueue sq SFlush
127126 _ <- strmbdy push flush
128127 atomically $ writeTBQueue sq SFinish
128+ deleteMyId mgr
129129 return ResponseReceived
130130
131131worker :: Context -> S.Settings -> Application -> Responder -> T.Manager -> IO ()