diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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. diff --git a/Network/QUIC/Client/Run.hs b/Network/QUIC/Client/Run.hs --- a/Network/QUIC/Client/Run.hs +++ b/Network/QUIC/Client/Run.hs @@ -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 diff --git a/Network/QUIC/Closer.hs b/Network/QUIC/Closer.hs --- a/Network/QUIC/Closer.hs +++ b/Network/QUIC/Closer.hs @@ -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 diff --git a/Network/QUIC/Connection/Role.hs b/Network/QUIC/Connection/Role.hs --- a/Network/QUIC/Connection/Role.hs +++ b/Network/QUIC/Connection/Role.hs @@ -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 ---------------------------------------------------------------- diff --git a/Network/QUIC/Connection/Types.hs b/Network/QUIC/Connection/Types.hs --- a/Network/QUIC/Connection/Types.hs +++ b/Network/QUIC/Connection/Types.hs @@ -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 } diff --git a/Network/QUIC/Server/Reader.hs b/Network/QUIC/Server/Reader.hs --- a/Network/QUIC/Server/Reader.hs +++ b/Network/QUIC/Server/Reader.hs @@ -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 diff --git a/Network/QUIC/Server/Run.hs b/Network/QUIC/Server/Run.hs --- a/Network/QUIC/Server/Run.hs +++ b/Network/QUIC/Server/Run.hs @@ -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 diff --git a/quic.cabal b/quic.cabal --- a/quic.cabal +++ b/quic.cabal @@ -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