diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,8 +1,12 @@ # Revision history for http3 +## 0.0.22 + +* Using `ThreadManager` of `time-manager`. + ## 0.0.21 -* Using `withHandle` of time-manager. +* Using `withHandle` of `time-manager`. ## 0.0.20 diff --git a/Network/HQ/Server.hs b/Network/HQ/Server.hs --- a/Network/HQ/Server.hs +++ b/Network/HQ/Server.hs @@ -72,15 +72,15 @@ processRequest :: Config -> SockAddr -> SockAddr -> Server -> Stream -> IO () processRequest conf mysa peersa server strm - | QUIC.isClientInitiatedBidirectional sid = do - th <- T.register (confTimeoutManager conf) (return ()) - vt <- recvHeader strm mysa - src <- newSource strm - refH <- newIORef Nothing - let readB = readSource' src - req = Request $ InpObj vt Nothing readB refH - aux = Aux th mysa peersa - server req aux $ sendResponse conf strm + | QUIC.isClientInitiatedBidirectional sid = + void $ T.withHandle (confTimeoutManager conf) (return ()) $ \th -> do + vt <- recvHeader strm mysa + src <- newSource strm + refH <- newIORef Nothing + let readB = readSource' src + req = Request $ InpObj vt Nothing readB refH + aux = Aux th mysa peersa + server req aux $ sendResponse conf strm | otherwise = return () -- fixme: should consume the data? where sid = QUIC.streamId strm diff --git a/Network/HTTP3/Client.hs b/Network/HTTP3/Client.hs --- a/Network/HTTP3/Client.hs +++ b/Network/HTTP3/Client.hs @@ -45,17 +45,12 @@ -- | Running an HTTP\/3 client. run :: Connection -> ClientConfig -> Config -> Client a -> IO a -run conn ClientConfig{..} conf client = E.bracket open close $ \ctx -> do - tid0 <- forkIO $ setupUnidirectional conn conf - addThreadId ctx tid0 - tid1 <- forkIO $ readerClient ctx - addThreadId ctx tid1 +run conn ClientConfig{..} conf client = withContext conn conf $ \ctx -> do + forkManaged ctx "H3 client: unidirectional setter" $ + setupUnidirectional conn conf + forkManaged ctx "H3 client: readerClient" $ readerClient ctx client (sendRequest ctx scheme authority) aux where - open = do - ref <- newIORef IInit - newContext conn conf (controlStream conn ref) - close = clearContext aux = Aux { auxPossibleClientStreams = possibleMyStreams conn @@ -70,9 +65,9 @@ process strm | QUIC.isClientInitiatedUnidirectional sid = return () -- error | QUIC.isClientInitiatedBidirectional sid = return () - | QUIC.isServerInitiatedUnidirectional sid = do - tid <- forkIO $ unidirectional ctx strm - addThreadId ctx tid + | QUIC.isServerInitiatedUnidirectional sid = + forkManaged ctx "H3 client: unidirectional handler" $ + unidirectional ctx strm | otherwise = return () -- push? where sid = QUIC.streamId strm @@ -80,29 +75,25 @@ sendRequest :: Context -> Scheme -> Authority -> Request -> (Response -> IO a) -> IO a sendRequest ctx scm auth (Request outobj) processResponse = - withHandle ctx $ \th -> do - let hdr = outObjHeaders outobj - hdr' = - (":scheme", scm) - : (":authority", C8.pack auth) - : hdr - E.bracket (newStream ctx) closeStream $ \strm -> do + E.bracket (newStream ctx) closeStream $ \strm -> do + forkManagedTimeout ctx "H3 client: sendRequest" $ \th -> do sendHeader ctx strm th hdr' - tid <- forkIO $ do - sendBody ctx strm th outobj - QUIC.shutdownStream strm - addThreadId ctx tid - src <- newSource strm - mvt <- recvHeader ctx src - case mvt of - Nothing -> do - QUIC.resetStream strm H3MessageError - threadDelay 100000 - -- just for type inference - E.throwIO $ QUIC.ApplicationProtocolErrorIsSent H3MessageError "" - Just vt -> do - refI <- newIORef IInit - refH <- newIORef Nothing - let readB = recvBody ctx src refI refH - rsp = Response $ InpObj vt Nothing readB refH - processResponse rsp + sendBody ctx strm th outobj + QUIC.shutdownStream strm + src <- newSource strm + mvt <- recvHeader ctx src + case mvt of + Nothing -> do + QUIC.resetStream strm H3MessageError + threadDelay 100000 + -- just for type inference + E.throwIO $ QUIC.ApplicationProtocolErrorIsSent H3MessageError "" + Just vt -> do + refI <- newIORef IInit + refH <- newIORef Nothing + let readB = recvBody ctx src refI refH + rsp = Response $ InpObj vt Nothing readB refH + processResponse rsp + where + hdr = outObjHeaders outobj + hdr' = (":scheme", scm) : (":authority", C8.pack auth) : hdr diff --git a/Network/HTTP3/Context.hs b/Network/HTTP3/Context.hs --- a/Network/HTTP3/Context.hs +++ b/Network/HTTP3/Context.hs @@ -3,8 +3,7 @@ module Network.HTTP3.Context ( Context, - newContext, - clearContext, + withContext, unidirectional, isH3Server, isH3Client, @@ -15,15 +14,17 @@ newStream, closeStream, pReadMaker, - addThreadId, abort, getHooks, Hooks (..), -- re-export getMySockAddr, getPeerSockAddr, + forkManaged, + forkManagedTimeout, + forkManagedTimeoutFinally, + isAsyncException, ) where -import Control.Concurrent import qualified Control.Exception as E import qualified Data.ByteString as BS import Data.IORef @@ -31,10 +32,11 @@ import Network.QUIC import Network.QUIC.Internal (connDebugLog, isClient, isServer) import Network.Socket (SockAddr) -import System.Mem.Weak -import qualified System.TimeManager as T +import qualified System.ThreadManager as T import Network.HTTP3.Config +import Network.HTTP3.Control +import Network.HTTP3.Frame import Network.HTTP3.Stream import Network.QPACK import Network.QPACK.Internal @@ -45,15 +47,20 @@ , ctxQDecoder :: QDecoder , ctxUniSwitch :: H3StreamType -> InstructionHandler , ctxPReadMaker :: PositionReadMaker - , ctxManager :: T.Manager + , ctxThreadManager :: T.ThreadManager , ctxHooks :: Hooks , ctxMySockAddr :: SockAddr , ctxPeerSockAddr :: SockAddr - , ctxThreads :: IORef [Weak ThreadId] } -newContext :: Connection -> Config -> InstructionHandler -> IO Context -newContext conn conf ctl = do +withContext :: Connection -> Config -> (Context -> IO a) -> IO a +withContext conn conf action = do + ctx <- newContext conn conf + T.stopAfter (ctxThreadManager ctx) (action ctx) (\_ -> return ()) + +newContext :: Connection -> Config -> IO Context +newContext conn conf = do + ctl <- controlStream conn <$> newIORef IInit (enc, handleDI) <- newQEncoder defaultQEncoderConfig (dec, handleEI) <- newQDecoder defaultQDecoderConfig info <- getConnectionInfo conn @@ -61,18 +68,35 @@ handleEI' recv = handleEI recv `E.catch` abortWith QpackEncoderStreamError sw = switch conn ctl handleEI' handleDI' preadM = confPositionReadMaker conf - timmgr = confTimeoutManager conf hooks = confHooks conf mysa = localSockAddr info peersa = remoteSockAddr info - Context conn enc dec sw preadM timmgr hooks mysa peersa <$> newIORef [] + mgr <- T.newThreadManager $ confTimeoutManager conf + return $ + Context + { ctxConnection = conn + , ctxQEncoder = enc + , ctxQDecoder = dec + , ctxUniSwitch = sw + , ctxPReadMaker = preadM + , ctxThreadManager = mgr + , ctxHooks = hooks + , ctxMySockAddr = mysa + , ctxPeerSockAddr = peersa + } where + abortWith :: ApplicationProtocolError -> E.SomeException -> IO () abortWith aerr se - | Just E.ThreadKilled <- E.fromException se = return () - | otherwise = abortConnection conn aerr "" + | isAsyncException se = E.throwIO se + | otherwise = do + print se + abortConnection conn aerr "" -clearContext :: Context -> IO () -clearContext ctx = clearThreads ctx +isAsyncException :: E.Exception e => e -> Bool +isAsyncException e = + case E.fromException (E.toException e) of + Just (E.SomeAsyncException _) -> True + Nothing -> False switch :: Connection @@ -108,8 +132,8 @@ let typ = toH3StreamType $ fromIntegral w8 ctxUniSwitch typ (recvStream strm) -withHandle :: Context -> (T.Handle -> IO a) -> IO a -withHandle Context{..} = T.withHandle ctxManager (return ()) +withHandle :: Context -> (T.Handle -> IO a) -> IO (Maybe a) +withHandle Context{..} = T.withHandle ctxThreadManager (return ()) newStream :: Context -> IO Stream newStream Context{..} = stream ctxConnection @@ -117,22 +141,17 @@ pReadMaker :: Context -> PositionReadMaker pReadMaker = ctxPReadMaker -addThreadId :: Context -> ThreadId -> IO () -addThreadId Context{..} tid = do - wtid <- mkWeakThreadId tid - atomicModifyIORef' ctxThreads $ \ts -> (wtid : ts, ()) +forkManaged :: Context -> String -> IO () -> IO () +forkManaged Context{..} = T.forkManaged ctxThreadManager -clearThreads :: Context -> IO () -clearThreads Context{..} = do - wtids <- readIORef ctxThreads - mapM_ kill wtids - writeIORef ctxThreads [] - where - kill wtid = do - mtid <- deRefWeak wtid - case mtid of - Nothing -> return () - Just tid -> killThread tid +forkManagedTimeout :: Context -> String -> (T.Handle -> IO ()) -> IO () +forkManagedTimeout Context{..} = + T.forkManagedTimeout ctxThreadManager + +forkManagedTimeoutFinally + :: Context -> String -> (T.Handle -> IO ()) -> IO () -> IO () +forkManagedTimeoutFinally Context{..} = + T.forkManagedTimeoutFinally ctxThreadManager abort :: Context -> ApplicationProtocolError -> IO () abort ctx aerr = abortConnection (ctxConnection ctx) aerr "" diff --git a/Network/HTTP3/Server.hs b/Network/HTTP3/Server.hs --- a/Network/HTTP3/Server.hs +++ b/Network/HTTP3/Server.hs @@ -45,28 +45,22 @@ -- | Running an HTTP\/3 server. run :: Connection -> Config -> Server -> IO () -run conn conf server = E.bracket open close $ \ctx -> do - myThreadId >>= \t -> labelThread t "H3 run" - tid <- forkIO $ setupUnidirectional conn conf - labelThread tid "H3 unidirectional setter" - addThreadId ctx tid +run conn conf server = withContext conn conf $ \ctx -> do + myThreadId >>= \t -> labelThread t "H3 server: run" + forkManaged ctx "H3 server: unidirectional setter" $ + setupUnidirectional conn conf readerServer ctx $ \strm -> - void $ - forkFinally - (processRequest ctx server strm) - (\_ -> closeStream strm) - where - open = do - ref <- newIORef IInit - newContext conn conf (controlStream conn ref) - close = clearContext + forkManagedTimeoutFinally + ctx + "H3 server: processRequest" + (processRequest ctx server strm) + (closeStream strm) runIO :: Connection -> Config -> (ServerIO Stream -> IO (IO ())) -> IO () -runIO conn conf action = E.bracket open close $ \ctx -> do - tid <- forkIO $ setupUnidirectional conn conf - labelThread tid "H3 unidirectional setter" +runIO conn conf action = withContext conn conf $ \ctx -> do + forkManaged ctx "H3 server: unidirectional setter" $ + setupUnidirectional conn conf info <- getConnectionInfo conn - addThreadId ctx tid reqq <- newTQueueIO let sio = ServerIO @@ -78,11 +72,6 @@ put strmreq = atomically $ writeTQueue reqq strmreq io <- action sio concurrently_ io (readerServer ctx $ processRequestIO ctx put) - where - open = do - ref <- newIORef IInit - newContext conn conf (controlStream conn ref) - close = clearContext readerServer :: Context -> (Stream -> IO ()) -> IO () readerServer ctx action = loop @@ -91,34 +80,33 @@ accept ctx >>= process loop process strm - | QUIC.isClientInitiatedUnidirectional sid = do - tid <- forkIO $ do - tid <- myThreadId - labelThread tid "H3 unidirectional handler" + | QUIC.isClientInitiatedUnidirectional sid = + forkManaged ctx "H3 server: unidirectional handler" $ unidirectional ctx strm - addThreadId ctx tid | QUIC.isClientInitiatedBidirectional sid = action strm | QUIC.isServerInitiatedUnidirectional sid = return () -- error | otherwise = return () where sid = QUIC.streamId strm -processRequest :: Context -> Server -> Stream -> IO () -processRequest ctx server strm = E.handle reset $ do - tid <- myThreadId - labelThread tid "H3 processRequest" - withHandle ctx $ \th -> do - src <- newSource strm - mvt <- recvHeader ctx src - case mvt of - Nothing -> QUIC.resetStream strm H3MessageError - Just ht -> do - req <- mkRequest ctx strm src ht - let aux = Aux th (getMySockAddr ctx) (getPeerSockAddr ctx) - server req aux $ sendResponse ctx strm th +processRequest + :: Context + -> Server + -> Stream + -> T.Handle + -> IO () +processRequest ctx server strm th = E.handle reset $ do + src <- newSource strm + mvt <- recvHeader ctx src + case mvt of + Nothing -> QUIC.resetStream strm H3MessageError + Just ht -> do + req <- mkRequest ctx strm src ht + let aux = Aux th (getMySockAddr ctx) (getPeerSockAddr ctx) + server req aux $ sendResponse ctx strm th where reset se - | Just E.ThreadKilled <- E.fromException se = return () + | isAsyncException se = E.throwIO se | Just (_ :: DecodeError) <- E.fromException se = abort ctx QpackDecompressionFailed | otherwise = QUIC.resetStream strm H3MessageError @@ -134,7 +122,7 @@ put (strm, req) where reset se - | Just E.ThreadKilled <- E.fromException se = return () + | isAsyncException se = E.throwIO se | Just (_ :: DecodeError) <- E.fromException se = abort ctx QpackDecompressionFailed | otherwise = QUIC.resetStream strm H3MessageError @@ -171,7 +159,7 @@ sendResponseIO :: Context -> Stream -> Response -> IO () sendResponseIO ctx strm (Response outobj) = - withHandle ctx $ \th -> do + void $ withHandle ctx $ \th -> do sendHeader ctx strm th $ outObjHeaders outobj sendBody ctx strm th outobj QUIC.closeStream strm diff --git a/http3.cabal b/http3.cabal --- a/http3.cabal +++ b/http3.cabal @@ -1,6 +1,6 @@ cabal-version: 2.4 name: http3 -version: 0.0.21 +version: 0.0.22 license: BSD-3-Clause license-file: LICENSE maintainer: Kazu Yamamoto @@ -90,7 +90,7 @@ quic >= 0.2 && < 0.3, sockaddr, stm, - time-manager >= 0.1.3 && < 0.2 + time-manager >= 0.2 && < 0.3 executable h3-server main-is: h3-server.hs diff --git a/test/HTTP3/Server.hs b/test/HTTP3/Server.hs --- a/test/HTTP3/Server.hs +++ b/test/HTTP3/Server.hs @@ -10,6 +10,7 @@ ) where import Control.Concurrent + -- cryptonite import qualified Control.Exception as E