quic 0.2.5 → 0.2.6
raw patch · 8 files changed
+104/−71 lines, 8 filesdep ~network
Dependency ranges changed: network
Files
- ChangeLog.md +5/−0
- Network/QUIC/Client/Run.hs +1/−0
- Network/QUIC/Closer.hs +1/−1
- Network/QUIC/Connection/Role.hs +7/−8
- Network/QUIC/Connection/Types.hs +2/−2
- Network/QUIC/Server/Reader.hs +53/−35
- Network/QUIC/Server/Run.hs +32/−22
- quic.cabal +3/−3
ChangeLog.md view
@@ -1,5 +1,10 @@ # ChangeLog +## 0.2.6++* Using `ServerState` instead of `killThread`.+* Don't catch asynchronous exceptions.+ ## 0.2.5 * Re-throwing asynchronous exceptions.
Network/QUIC/Client/Run.hs view
@@ -47,6 +47,7 @@ vers = ccVersions conf in VersionInfo ver vers _ -> let ver = resumptionVersion resInfo in VersionInfo ver [ver]+ -- Exceptions except NextVersion are passed through. ex <- E.try $ runClient conf client False verInfo case ex of Right v -> return v
Network/QUIC/Closer.hs view
@@ -35,7 +35,7 @@ E.throwIO $ ApplicationProtocolErrorIsSent err desc | Just (VerNego vers) <- E.fromException se = do E.throwIO $ NextVersion vers- | otherwise = E.throwIO se+ | otherwise = E.throwIO se -- including asynchronous exceptions closure' :: Connection -> LDCC -> Frame -> IO () closure' conn ldcc frame = do
Network/QUIC/Connection/Role.hs view
@@ -16,13 +16,12 @@ getUnregister, setTokenManager, getTokenManager,- setBaseThreadId,- getBaseThreadId,+ setStopServer,+ getStopServer, setCertificateChain, getCertificateChain, ) where -import Control.Concurrent import qualified Crypto.Token as CT import Data.X509 (CertificateChain) @@ -133,12 +132,12 @@ ---------------------------------------------------------------- -setBaseThreadId :: Connection -> ThreadId -> IO ()-setBaseThreadId Connection{..} tid = atomicModifyIORef'' roleInfo $- \si -> si{baseThreadId = tid}+setStopServer :: Connection -> IO () -> IO ()+setStopServer Connection{..} action = atomicModifyIORef'' roleInfo $+ \si -> si{stopServer = action} -getBaseThreadId :: Connection -> IO ThreadId-getBaseThreadId Connection{..} = baseThreadId <$> readIORef roleInfo+getStopServer :: Connection -> IO (IO ())+getStopServer Connection{..} = stopServer <$> readIORef roleInfo ----------------------------------------------------------------
Network/QUIC/Connection/Types.hs view
@@ -50,7 +50,7 @@ , registerCID :: CID -> Connection -> IO () , unregisterCID :: CID -> IO () , askRetry :: Bool- , baseThreadId :: ~ThreadId+ , stopServer :: IO () , certChain :: Maybe CertificateChain } @@ -69,7 +69,7 @@ , registerCID = \_ _ -> return () , unregisterCID = \_ -> return () , askRetry = False- , baseThreadId = undefined+ , stopServer = return () , certChain = Nothing }
Network/QUIC/Server/Reader.hs view
@@ -9,12 +9,12 @@ tokenMgr, -- * Accepting- accept, Accept (..), -- * Receiving and reading RecvQ, recvServer,+ ServerState (..), ) where import Control.Concurrent@@ -28,7 +28,7 @@ import Network.ByteOrder import Network.Control (LRUCache) import qualified Network.Control as LRUCache-import Network.Socket (Socket)+import Network.Socket (Socket, waitReadSocketSTM) import qualified Network.Socket.ByteString as NSB import qualified System.IO.Error as E @@ -49,7 +49,6 @@ { tokenMgr :: CT.TokenManager , dstTable :: IORef ConnectionDict , srcTable :: IORef RecvQDict- , acceptQ :: AcceptQ } newDispatch :: ServerConfig -> IO Dispatch@@ -58,7 +57,6 @@ <$> CT.spawnTokenManager conf <*> newIORef emptyConnectionDict <*> newIORef emptyRecvQDict- <*> newAcceptQ where conf = CT.defaultConfig@@ -128,40 +126,56 @@ , accTime :: TimeMicrosecond } -newtype AcceptQ = AcceptQ (TQueue Accept)--newAcceptQ :: IO AcceptQ-newAcceptQ = AcceptQ <$> newTQueueIO--readAcceptQ :: AcceptQ -> IO Accept-readAcceptQ (AcceptQ q) = atomically $ readTQueue q--writeAcceptQ :: AcceptQ -> Accept -> IO ()-writeAcceptQ (AcceptQ q) x = atomically $ writeTQueue q x+---------------------------------------------------------------- -accept :: Dispatch -> IO Accept-accept = readAcceptQ . acceptQ+runDispatcher+ :: Dispatch+ -> ServerConfig+ -> TVar ServerState+ -> (Accept -> IO ())+ -> Socket+ -> IO ThreadId+runDispatcher d conf stvar forkConn mysock = forkIO $ dispatcher d conf stvar forkConn mysock -----------------------------------------------------------------+data ServerState = Running | Stopped deriving (Eq, Show) -runDispatcher :: Dispatch -> ServerConfig -> Socket -> IO ThreadId-runDispatcher d conf mysock = forkIO $ dispatcher d conf mysock+checkLoop :: TVar ServerState -> STM () -> IO Bool+checkLoop stvar waitsock = atomically $ do+ st <- readTVar stvar+ if st == Stopped+ then+ return False+ else do+ waitsock -- blocking is retry+ return True -dispatcher :: Dispatch -> ServerConfig -> Socket -> IO ()-dispatcher d conf mysock = handleLogUnit logAction $ do+dispatcher+ :: Dispatch+ -> ServerConfig+ -> TVar ServerState+ -> (Accept -> IO ())+ -> Socket+ -> IO ()+dispatcher d conf stvar forkConnection mysock = do labelMe "QUIC dispatcher"- forever $ do- (peersa, bs, cmsgs, _) <- safeRecv $ NSB.recvMsg mysock 2048 2048 0- now <- getTimeMicrosecond- let send' b = void $ NSB.sendMsg mysock peersa [b] cmsgs 0- -- cf: greaseQuicBit $ getMyParameters conn- quicBit = greaseQuicBit $ scParameters conf- cpckts <- decodeCryptPackets bs (not quicBit)- let bytes = BS.length bs- peerInfo = PeerInfo peersa cmsgs- switch = dispatch d conf logAction mysock peerInfo send' bytes now- mapM_ switch cpckts+ wait <- waitReadSocketSTM mysock+ handleLogUnit logAction $ loop wait where+ loop wait = do+ cont <- checkLoop stvar wait+ when cont $ do+ (peersa, bs, cmsgs, _) <- safeRecv $ NSB.recvMsg mysock 2048 2048 0+ now <- getTimeMicrosecond+ let send' b = void $ NSB.sendMsg mysock peersa [b] cmsgs 0+ -- cf: greaseQuicBit $ getMyParameters conn+ quicBit = greaseQuicBit $ scParameters conf+ cpckts <- decodeCryptPackets bs (not quicBit)+ let bytes = BS.length bs+ peerInfo = PeerInfo peersa cmsgs+ switch = dispatch d conf forkConnection logAction mysock peerInfo send' bytes now+ mapM_ switch cpckts+ loop wait+ doDebug = isJust $ scDebugLog conf logAction msg | doDebug = stdoutLogger ("dispatch(er): " <> msg)@@ -171,6 +185,7 @@ ex <- E.try $ windowsThreadBlockHack rcv case ex of Right x -> return x+ Left se | isAsyncException se -> E.throwIO (se :: E.SomeException) Left se -> case E.fromException se of Just e | E.ioeGetErrorType e == E.InvalidArgument -> E.throwIO se _ -> do@@ -189,6 +204,7 @@ dispatch :: Dispatch -> ServerConfig+ -> (Accept -> IO ()) -> DebugLogger -> Socket -> PeerInfo@@ -200,6 +216,7 @@ dispatch Dispatch{..} ServerConfig{..}+ forkConnection logAction mysock peerInfo@@ -248,7 +265,7 @@ writeRecvQ q $ mkReceivedPacket cpkt tim siz lvl let reg = registerConnectionDict dstTable unreg = unregisterConnectionDict dstTable- ent =+ acc = Accept { accVersionInfo = VersionInfo peerVer myVersions , accMyAuthCIDs = myAuthCIDs@@ -262,8 +279,7 @@ , accAddressValidated = addrValid , accTime = tim }- -- fixme: check acceptQ length- writeAcceptQ acceptQ ent+ forkConnection acc -- Initial: DCID=S1, SCID=C1 -> -- <- Initial: DCID=C1, SCID=S2 -- ...@@ -334,6 +350,7 @@ Dispatch{..} _ _+ _ _mysock _peerInfo _@@ -347,6 +364,7 @@ ---------------------------------------------------------------- dispatch Dispatch{..}+ _ _ logAction mysock
Network/QUIC/Server/Run.hs view
@@ -10,9 +10,10 @@ import Control.Concurrent import Control.Concurrent.Async+import Control.Concurrent.STM import qualified Control.Exception as E import qualified Network.Socket as NS-import System.Log.FastLogger+import System.Log.FastLogger hiding (check) import Network.QUIC.Closer import Network.QUIC.Common@@ -42,22 +43,23 @@ run :: ServerConfig -> (Connection -> IO ()) -> IO () run conf server = NS.withSocketsDo $ handleLogUnit debugLog $ do labelMe "QUIC run"- baseThreadId <- myThreadId- E.bracket setup teardown $ \(dispatch, _, _) -> do+ stvar <- newTVarIO Running+ E.bracket (setup stvar) teardown $ \(_, _, _) -> do onServerReady $ scHooks conf- forever $ do- acc <- accept dispatch- void $ forkIO (runServer conf server dispatch baseThreadId acc)+ atomically $ do+ st <- readTVar stvar+ check $ st == Stopped where doDebug = isJust $ scDebugLog conf debugLog msg | doDebug = stdoutLogger ("run: " <> msg) | otherwise = return ()- setup = do+ setup stvar = do dispatch <- newDispatch conf+ let forkConn acc = void $ forkIO (runServer conf server dispatch stvar acc) -- fixme: the case where sockets cannot be created. ssas <- mapM serverSocket $ scAddresses conf- tids <- mapM (runDispatcher dispatch conf) ssas+ tids <- mapM (runDispatcher dispatch conf stvar forkConn) ssas return (dispatch, tids, ssas) teardown (dispatch, tids, ssas) = do clearDispatch dispatch@@ -70,21 +72,22 @@ runWithSockets :: [NS.Socket] -> ServerConfig -> (Connection -> IO ()) -> IO () runWithSockets ssas conf server = NS.withSocketsDo $ handleLogUnit debugLog $ do labelMe "QUIC runWithSockets"- baseThreadId <- myThreadId- E.bracket setup teardown $ \(dispatch, _) -> do+ stvar <- newTVarIO Running+ E.bracket (setup stvar) teardown $ \(_, _) -> do onServerReady $ scHooks conf- forever $ do- acc <- accept dispatch- void $ forkIO (runServer conf server dispatch baseThreadId acc)+ atomically $ do+ st <- readTVar stvar+ check $ st == Stopped where doDebug = isJust $ scDebugLog conf debugLog msg | doDebug = stdoutLogger ("run: " <> msg) | otherwise = return ()- setup = do+ setup stvar = do dispatch <- newDispatch conf+ let forkConn acc = void $ forkIO (runServer conf server dispatch stvar acc) -- fixme: the case where sockets cannot be created.- tids <- mapM (runDispatcher dispatch conf) ssas+ tids <- mapM (runDispatcher dispatch conf stvar forkConn) ssas return (dispatch, tids) teardown (dispatch, tids) = do clearDispatch dispatch@@ -93,8 +96,13 @@ -- Typically, ConnectionIsClosed breaks acceptStream. -- And the exception should be ignored. runServer- :: ServerConfig -> (Connection -> IO ()) -> Dispatch -> ThreadId -> Accept -> IO ()-runServer conf server0 dispatch baseThreadId acc = do+ :: ServerConfig+ -> (Connection -> IO ())+ -> Dispatch+ -> TVar ServerState+ -> Accept+ -> IO ()+runServer conf server0 dispatch stvar acc = do labelMe "QUIC runServer" E.bracket open clse $ \(ConnRes conn myAuthCIDs _reader) -> handleLogUnit (debugLog conn) $ do@@ -130,7 +138,7 @@ sendFinal conn closure conn ldcc ex where- open = createServerConnection conf dispatch acc baseThreadId+ open = createServerConnection conf dispatch acc stvar clse connRes = do let conn = connResConnection connRes setDead conn@@ -143,9 +151,9 @@ :: ServerConfig -> Dispatch -> Accept- -> ThreadId+ -> TVar ServerState -> IO ConnRes-createServerConnection conf@ServerConfig{..} dispatch Accept{..} baseThreadId = do+createServerConnection conf@ServerConfig{..} dispatch Accept{..} stvar = do sref <- newIORef accMySocket piref <- newIORef accPeerInfo let send buf siz = void $ do@@ -195,7 +203,7 @@ let mgr = tokenMgr dispatch setTokenManager conn mgr --- setBaseThreadId conn baseThreadId+ setStopServer conn $ atomically $ writeTVar stvar Stopped -- setRegister conn accRegister accUnregister accRegister myCID conn@@ -224,4 +232,6 @@ -- | Stopping the base thread of the server. stop :: Connection -> IO ()-stop conn = getBaseThreadId conn >>= killThread+stop conn = do+ action <- getStopServer conn+ action
quic.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: quic-version: 0.2.5+version: 0.2.6 license: BSD3 license-file: LICENSE maintainer: kazu@iij.ad.jp@@ -144,7 +144,7 @@ fast-logger >= 3.2.2 && < 3.3, unix-time >= 0.4.12 && < 0.5, iproute >= 1.7.12 && < 1.8,- network >= 3.1.4,+ network >= 3.2.2, network-byte-order >= 0.1.7 && < 0.2, network-control >= 0.1 && < 0.2, random >= 1.2.1 && < 1.3,@@ -251,7 +251,7 @@ containers, crypton, hspec,- network >=3.1.2,+ network >=3.2.2, quic, tls, unix-time