http2 5.3.8 → 5.3.9
raw patch · 10 files changed
+23/−179 lines, 10 filesdep ~time-manager
Dependency ranges changed: time-manager
Files
- ChangeLog.md +4/−0
- Network/HTTP2/Client/Run.hs +5/−4
- Network/HTTP2/H2.hs +0/−2
- Network/HTTP2/H2/Context.hs +3/−4
- Network/HTTP2/H2/Manager.hs +0/−151
- Network/HTTP2/H2/Receiver.hs +3/−3
- Network/HTTP2/Internal.hs +0/−6
- Network/HTTP2/Server/Run.hs +2/−1
- Network/HTTP2/Server/Worker.hs +4/−4
- http2.cabal +2/−4
ChangeLog.md view
@@ -1,5 +1,9 @@ # ChangeLog for http2 +## 5.3.9++* Using `ThreadManager` of `time-manager`.+ ## 5.3.8 * `forkManagedTimeout` ensures that only one asynchronous exception is
Network/HTTP2/Client/Run.hs view
@@ -17,6 +17,7 @@ import Network.HTTP.Semantics.Client.Internal import Network.HTTP.Semantics.IO import Network.Socket (SockAddr)+import qualified System.ThreadManager as T import Text.Read (readMaybe) import Imports@@ -97,7 +98,7 @@ wrapClient :: Context -> IO a -> IO a wrapClient ctx client = do x <- client- waitCounter0 $ threadManager ctx+ T.waitUntilAllGone $ threadManager ctx let frame = goawayFrame 0 NoError "graceful closing" enqueueControl (controlQ ctx) $ CFrames Nothing [frame] enqueueControl (controlQ ctx) $ CFinish GoAwayIsSent@@ -147,7 +148,7 @@ runH2 :: Config -> Context -> IO a -> IO a runH2 conf ctx runClient = do- stopAfter mgr runAll $ \res ->+ T.stopAfter mgr runAll $ \res -> closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) res where mgr = threadManager ctx@@ -238,7 +239,7 @@ (pop, out) <- makeOutput strm ot pushOutput sid out lc <- newLoopCheck strm mtbq- forkManaged threadManager label $ syncWithSender' ctx pop lc+ T.forkManaged threadManager label $ syncWithSender' ctx pop lc where label = "H2 request sender for stream " ++ show (streamNumber strm) pushOutput sid out = atomically $ do@@ -254,7 +255,7 @@ -> IO (TBQueue StreamingChunk) sendStreaming Context{..} strm strmbdy = do tbq <- newTBQueueIO 10 -- fixme: hard coding: 10- forkManagedUnmask threadManager label $ \unmask ->+ T.forkManagedUnmask threadManager label $ \unmask -> withOutBodyIface tbq unmask strmbdy return tbq where
Network/HTTP2/H2.hs view
@@ -3,7 +3,6 @@ module Network.HTTP2.H2.Context, module Network.HTTP2.H2.EncodeFrame, module Network.HTTP2.H2.HPACK,- module Network.HTTP2.H2.Manager, module Network.HTTP2.H2.Queue, module Network.HTTP2.H2.Receiver, module Network.HTTP2.H2.Sender,@@ -19,7 +18,6 @@ import Network.HTTP2.H2.Context import Network.HTTP2.H2.EncodeFrame import Network.HTTP2.H2.HPACK-import Network.HTTP2.H2.Manager import Network.HTTP2.H2.Queue import Network.HTTP2.H2.Receiver import Network.HTTP2.H2.Sender
Network/HTTP2/H2/Context.hs view
@@ -10,12 +10,11 @@ import Data.IORef import Network.Control import Network.Socket (SockAddr)-import qualified System.TimeManager as T+import qualified System.ThreadManager as T import Imports hiding (insert) import Network.HPACK import Network.HTTP2.Frame-import Network.HTTP2.H2.Manager import Network.HTTP2.H2.Settings import Network.HTTP2.H2.Stream import Network.HTTP2.H2.StreamTable@@ -88,7 +87,7 @@ , rstRate :: Rate , mySockAddr :: SockAddr , peerSockAddr :: SockAddr- , threadManager :: Manager+ , threadManager :: T.ThreadManager , senderDone :: TVar Bool } @@ -130,7 +129,7 @@ <*> newRate <*> return confMySockAddr <*> return confPeerSockAddr- <*> start timmgr+ <*> T.newThreadManager timmgr <*> newTVarIO False where rl = case rinfo of
− Network/HTTP2/H2/Manager.hs
@@ -1,151 +0,0 @@-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE ScopedTypeVariables #-}---- | A thread manager.--- The manager has responsibility to kill worker threads.-module Network.HTTP2.H2.Manager (- Manager,- start,- stopAfter,- forkManaged,- forkManagedUnmask,- forkManagedTimeout,- KilledByHttp2ThreadManager (..),- waitCounter0,-) where--import Control.Concurrent-import Control.Concurrent.STM-import Control.Exception-import qualified Control.Exception as E-import Data.Foldable-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---------------------------------------------------------------------- | Manager to manage the thread and the timer.-data Manager = Manager T.Manager (TVar ManagedThreads)--type ManagedThreads = IntMap ManagedThread---------------------------------------------------------------------- '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)---------------------------------------------------------------------- | Starting a thread manager.--- Its action is initially set to 'return ()' and should be set--- by 'setAction'. This allows that the action can include--- the manager itself.-start :: T.Manager -> IO Manager-start timmgr = Manager timmgr <$> newTVarIO Map.empty--------------------------------------------------------------------data KilledByHttp2ThreadManager = KilledByHttp2ThreadManager (Maybe SomeException)- deriving (Show)--instance Exception KilledByHttp2ThreadManager where- toException = asyncExceptionToException- fromException = asyncExceptionFromException---- | Stopping the manager.------ The action is run in the scope of an exception handler that catches all--- exceptions (including asynchronous ones); this allows the cleanup handler--- to cleanup in all circumstances. If an exception is caught, it is rethrown--- after the cleanup is complete.-stopAfter :: Manager -> IO a -> (Maybe SomeException -> IO ()) -> IO a-stopAfter (Manager _timmgr var) action cleanup = do- mask $ \unmask -> do- ma <- try $ unmask action- m <- atomically $ do- m0 <- readTVar var- writeTVar var Map.empty- return m0- 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---------------------------------------------------------------------- | Fork managed thread------ This guarantees that the thread ID is added to the manager's queue before--- the thread starts, and is removed again when the thread terminates--- (normally or abnormally).-forkManaged :: Manager -> String -> IO () -> IO ()-forkManaged mgr label io =- forkManagedUnmask mgr label $ \unmask -> unmask io---- | Like 'forkManaged', but run action with exceptions masked-forkManagedUnmask- :: Manager -> String -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()-forkManagedUnmask (Manager _timmgr var) label io =- void $ mask_ $ forkIOWithUnmask $ \unmask -> E.handle ignore $ do- labelMe label- 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- check (Map.size m == 0)--------------------------------------------------------------------myWeakThradId :: IO (Weak ThreadId, Int)-myWeakThradId = do- tid <- myThreadId- wtid <- mkWeakThreadId tid- let n = read (drop 9 $ show tid) -- drop "ThreadId "- return (wtid, n)
Network/HTTP2/H2/Receiver.hs view
@@ -18,13 +18,13 @@ import Data.IORef import Network.Control import Network.HTTP.Semantics+import qualified System.ThreadManager as T import Imports hiding (delete, insert) import Network.HTTP2.Frame import Network.HTTP2.H2.Context import Network.HTTP2.H2.EncodeFrame import Network.HTTP2.H2.HPACK-import Network.HTTP2.H2.Manager import Network.HTTP2.H2.Queue import Network.HTTP2.H2.Settings import Network.HTTP2.H2.Stream@@ -60,10 +60,10 @@ sendGoaway se | isAsyncException se = E.throwIO se | Just GoAwayIsSent <- E.fromException se = do- waitCounter0 threadManager+ T.waitUntilAllGone threadManager enqueueControl controlQ $ CFinish GoAwayIsSent | Just ConnectionIsClosed <- E.fromException se = do- waitCounter0 threadManager+ T.waitUntilAllGone threadManager enqueueControl controlQ $ CFinish ConnectionIsClosed | Just e@(ConnectionErrorIsReceived _ _ _) <- E.fromException se = enqueueControl controlQ $ CFinish e
− Network/HTTP2/Internal.hs
@@ -1,6 +0,0 @@-module Network.HTTP2.Internal (- -- * Thread Manager- module Network.HTTP2.H2.Manager,-) where--import Network.HTTP2.H2.Manager
Network/HTTP2/Server/Run.hs view
@@ -12,6 +12,7 @@ import Network.HTTP.Semantics.Server import Network.HTTP.Semantics.Server.Internal import Network.Socket (SockAddr)+import qualified System.ThreadManager as T import Network.HTTP2.Frame import Network.HTTP2.H2@@ -123,7 +124,7 @@ runReceiver = frameReceiver ctx conf runSender = frameSender ctx conf runBackgroundThreads = concurrently_ runReceiver runSender- stopAfter mgr runBackgroundThreads $ \res ->+ T.stopAfter mgr runBackgroundThreads $ \res -> closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) res -- connClose must not be called here since Run:fork calls it
Network/HTTP2/Server/Worker.hs view
@@ -13,7 +13,7 @@ import Network.HTTP.Semantics.Server import Network.HTTP.Semantics.Server.Internal import Network.HTTP.Types-import qualified System.TimeManager as T+import qualified System.ThreadManager as T import Imports hiding (insert) import Network.HTTP2.Frame@@ -23,7 +23,7 @@ runServer :: Config -> Server -> Launch runServer conf server ctx@Context{..} strm req =- forkManagedTimeout threadManager label $ \th -> do+ T.forkManagedTimeout threadManager label $ \th -> do let req' = pauseRequestBody th aux = Aux th mySockAddr peerSockAddr request = Request req'@@ -95,7 +95,7 @@ check (n >= lim) push _ [] n = return (n :: Int) push tvar (pp : pps) n = do- forkManaged threadManager "H2 server push" $ do+ T.forkManaged threadManager "H2 server push" $ do (pid, newstrm) <- makePushStream ctx pstrm let scheme = fromJust $ getFieldValue tokenScheme reqvt -- fixme: this value can be Nothing@@ -167,7 +167,7 @@ -> IO (TBQueue StreamingChunk) sendStreaming Context{..} strm strmbdy = do tbq <- newTBQueueIO 10 -- fixme: hard coding: 10- forkManagedTimeout threadManager label $ \th ->+ T.forkManagedTimeout threadManager label $ \th -> withOutBodyIface tbq id $ \iface -> do let iface' = iface
http2.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: http2-version: 5.3.8+version: 5.3.9 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto <kazu@iij.ad.jp>@@ -60,7 +60,6 @@ Network.HTTP2.Client Network.HTTP2.Client.Internal Network.HTTP2.Frame- Network.HTTP2.Internal Network.HTTP2.Server Network.HTTP2.Server.Internal @@ -90,7 +89,6 @@ Network.HTTP2.H2.Context Network.HTTP2.H2.EncodeFrame Network.HTTP2.H2.HPACK- Network.HTTP2.H2.Manager Network.HTTP2.H2.Queue Network.HTTP2.H2.Receiver Network.HTTP2.H2.Sender@@ -123,7 +121,7 @@ network-byte-order >=0.1.7 && <0.2, network-control >=0.1 && <0.2, stm >=2.5 && <2.6,- time-manager >=0.1.3 && <0.2,+ time-manager >=0.2 && <0.3, unix-time >=0.4.11 && <0.5, utf8-string >=1.0 && <1.1