http3 0.0.21 → 0.0.22
raw patch · 7 files changed
+129/−126 lines, 7 filesdep ~time-managerPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: time-manager
API changes (from Hackage documentation)
Files
- ChangeLog.md +5/−1
- Network/HQ/Server.hs +9/−9
- Network/HTTP3/Client.hs +28/−37
- Network/HTTP3/Context.hs +52/−33
- Network/HTTP3/Server.hs +32/−44
- http3.cabal +2/−2
- test/HTTP3/Server.hs +1/−0
ChangeLog.md view
@@ -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
Network/HQ/Server.hs view
@@ -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
Network/HTTP3/Client.hs view
@@ -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
Network/HTTP3/Context.hs view
@@ -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 ""
Network/HTTP3/Server.hs view
@@ -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
http3.cabal view
@@ -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 <kazu@iij.ad.jp>@@ -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
test/HTTP3/Server.hs view
@@ -10,6 +10,7 @@ ) where import Control.Concurrent+ -- cryptonite import qualified Control.Exception as E