diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,11 @@ # ChangeLog for http2 +## 5.3.8 + +* `forkManagedTimeout` ensures that only one asynchronous exception is + thrown. Fixing the thread leak via `Weak ThreadId` and `modifyTVar'`. + [#156](https://github.com/kazu-yamamoto/http2/pull/156) + ## 5.3.7 * Using `withHandle` of time-manager. diff --git a/Network/HTTP2/H2/Manager.hs b/Network/HTTP2/H2/Manager.hs --- a/Network/HTTP2/H2/Manager.hs +++ b/Network/HTTP2/H2/Manager.hs @@ -9,7 +9,7 @@ stopAfter, forkManaged, forkManagedUnmask, - withTimeout, + forkManagedTimeout, KilledByHttp2ThreadManager (..), waitCounter0, ) where @@ -19,8 +19,10 @@ import Control.Exception import qualified Control.Exception as E import Data.Foldable -import Data.Map (Map) -import qualified Data.Map.Strict as Map +import Data.IORef +import Data.IntMap (IntMap) +import qualified Data.IntMap.Strict as Map +import System.Mem.Weak (Weak, deRefWeak) import qualified System.TimeManager as T import Imports @@ -30,17 +32,15 @@ -- | Manager to manage the thread and the timer. data Manager = Manager T.Manager (TVar ManagedThreads) -type ManagedThreads = Map ThreadId TimeoutHandle +type ManagedThreads = IntMap ManagedThread ---------------------------------------------------------------- -data TimeoutHandle - = ThreadWithTimeout T.Handle - | ThreadWithoutTimeout - -cancelTimeout :: TimeoutHandle -> IO () -cancelTimeout (ThreadWithTimeout th) = T.cancel th -cancelTimeout ThreadWithoutTimeout = return () +-- 'IORef' prevents race between WAI TimeManager (TimeoutThread) +-- and stopAfter (KilledByHttp2ThreadManager). +-- It is initialized with 'False' and turned into 'True' when locked. +-- The winner can throw an asynchronous exception. +data ManagedThread = ManagedThread (Weak ThreadId) (IORef Bool) ---------------------------------------------------------------- @@ -74,10 +74,10 @@ m0 <- readTVar var writeTVar var Map.empty return m0 - forM_ (Map.elems m) cancelTimeout - let er = either Just (const Nothing) ma - forM_ (Map.keys m) $ \tid -> - E.throwTo tid $ KilledByHttp2ThreadManager er + let ths = Map.elems m + er = either Just (const Nothing) ma + ex = KilledByHttp2ThreadManager er + forM_ ths $ \(ManagedThread wtid ref) -> lockAndKill wtid ref ex case ma of Left err -> cleanup (Just err) >> throwIO err Right a -> cleanup Nothing >> return a @@ -97,19 +97,45 @@ forkManagedUnmask :: Manager -> String -> ((forall x. IO x -> IO x) -> IO ()) -> IO () forkManagedUnmask (Manager _timmgr var) label io = - -- This is the top level of thread. - -- So, SomeException should be reasonable. void $ mask_ $ forkIOWithUnmask $ \unmask -> E.handle ignore $ do labelMe label - tid <- myThreadId - atomically $ modifyTVar var $ Map.insert tid ThreadWithoutTimeout - -- We catch the exception and do not rethrow it: we don't want the - -- exception printed to stderr. - io unmask `catch` ignore - atomically $ modifyTVar var $ Map.delete tid - where - ignore (E.SomeException _) = return () + E.bracket (setup var) (clear var) $ \_ -> io unmask +forkManagedTimeout :: Manager -> String -> (T.Handle -> IO ()) -> IO () +forkManagedTimeout (Manager timmgr var) label io = + void $ forkIO $ E.handle ignore $ do + labelMe label + E.bracket (setup var) (clear var) $ \(_n, wtid, ref) -> + -- 'TimeoutThread' is ignored by 'withHandle'. + T.withHandle timmgr (lockAndKill wtid ref T.TimeoutThread) io + +setup :: TVar (IntMap ManagedThread) -> IO (Int, Weak ThreadId, IORef Bool) +setup var = do + (wtid, n) <- myWeakThradId + ref <- newIORef False + let ent = ManagedThread wtid ref + -- asking to throw KilledByHttp2ThreadManager to me + atomically $ modifyTVar' var $ Map.insert n ent + return (n, wtid, ref) + +lockAndKill :: Exception e => Weak ThreadId -> IORef Bool -> e -> IO () +lockAndKill wtid ref e = do + alreadyLocked <- atomicModifyIORef' ref (\b -> (True, b)) -- try to lock + unless alreadyLocked $ do + mtid <- deRefWeak wtid + case mtid of + Nothing -> return () + Just tid -> E.throwTo tid e + +clear + :: TVar (IntMap ManagedThread) + -> (Map.Key, Weak ThreadId, IORef Bool) + -> IO () +clear var (n, _, _) = atomically $ modifyTVar' var $ Map.delete n + +ignore :: KilledByHttp2ThreadManager -> IO () +ignore (KilledByHttp2ThreadManager _) = return () + waitCounter0 :: Manager -> IO () waitCounter0 (Manager _timmgr var) = atomically $ do m <- readTVar var @@ -117,10 +143,9 @@ ---------------------------------------------------------------- -withTimeout :: Manager -> (T.Handle -> IO ()) -> IO () -withTimeout (Manager timmgr var) action = - T.withHandleKillThread timmgr (return ()) $ \th -> do - tid <- myThreadId - -- overriding ThreadWithoutTimeout - atomically $ modifyTVar var $ Map.insert tid $ ThreadWithTimeout th - action th +myWeakThradId :: IO (Weak ThreadId, Int) +myWeakThradId = do + tid <- myThreadId + wtid <- mkWeakThreadId tid + let n = read (drop 9 $ show tid) -- drop "ThreadId " + return (wtid, n) diff --git a/Network/HTTP2/Server/Worker.hs b/Network/HTTP2/Server/Worker.hs --- a/Network/HTTP2/Server/Worker.hs +++ b/Network/HTTP2/Server/Worker.hs @@ -23,15 +23,13 @@ runServer :: Config -> Server -> Launch runServer conf server ctx@Context{..} strm req = - forkManaged threadManager label $ - withTimeout threadManager $ \th -> do - -- FIXME: exception - let req' = pauseRequestBody th - aux = Aux th mySockAddr peerSockAddr - request = Request req' - lc <- newLoopCheck strm Nothing - server request aux $ sendResponse conf ctx lc strm request - adjustRxWindow ctx strm + forkManagedTimeout threadManager label $ \th -> do + let req' = pauseRequestBody th + aux = Aux th mySockAddr peerSockAddr + request = Request req' + lc <- newLoopCheck strm Nothing + server request aux $ sendResponse conf ctx lc strm request + adjustRxWindow ctx strm where label = "H2 response sender for stream " ++ show (streamNumber strm) pauseRequestBody th = req{inpObjBody = readBody'} @@ -169,21 +167,20 @@ -> IO (TBQueue StreamingChunk) sendStreaming Context{..} strm strmbdy = do tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 - forkManaged threadManager label $ - withTimeout threadManager $ \th -> - withOutBodyIface tbq id $ \iface -> do - let iface' = - iface - { outBodyPush = \b -> do - T.pause th - outBodyPush iface b - T.resume th - , outBodyPushFinal = \b -> do - T.pause th - outBodyPushFinal iface b - T.resume th - } - strmbdy iface' + forkManagedTimeout threadManager label $ \th -> + withOutBodyIface tbq id $ \iface -> do + let iface' = + iface + { outBodyPush = \b -> do + T.pause th + outBodyPush iface b + T.resume th + , outBodyPushFinal = \b -> do + T.pause th + outBodyPushFinal iface b + T.resume th + } + strmbdy iface' return tbq where label = "H2 response streaming sender for " ++ show (streamNumber strm) diff --git a/http2.cabal b/http2.cabal --- a/http2.cabal +++ b/http2.cabal @@ -1,6 +1,6 @@ cabal-version: >=1.10 name: http2 -version: 5.3.7 +version: 5.3.8 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto