diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,31 @@ # ChangeLog + +## 0.3.19 + +* The dispatcher is stopped by the stop variable alone: it waits for that + and for its receive at once, through network's `recvFromWith`. The + empty datagram it used to be woken with, and the machinery which sent + it, are gone. +* `rebind` retires the reader of the socket it replaces rather than + closing that socket under it, and the reader's idle timeout is waited + for rather than thrown. A close ends a thread blocked on a descriptor + only because the event manager says so, and that is being removed. + (#173) +* The dispatcher's receive tells a closed socket from a datagram's + failure with network's `isClosedSocketError` rather than by comparing + the error type here. +* Requires network 3.2.10 and network-run 0.7. +* The closer's drain gives up on its receive at the deadline rather than + having `timeout` abandon it, so the six calls it could leave behind, and + the close which ended them, are gone. +* `killReaders` retires the reader of the current socket before + `stopAfter` throws into the managed threads, which does not reach one + blocked in a socket call on Windows. +* `Network.QUIC.Windows` is removed. It ran a blocking call on a thread + of its own and abandoned it, which held only because the socket was + closed soon after; network's `withReadable` cancels the call instead. + ## 0.3.18 * The threads of a connection are owned by `time-manager`'s diff --git a/Network/QUIC/Client/Reader.hs b/Network/QUIC/Client/Reader.hs --- a/Network/QUIC/Client/Reader.hs +++ b/Network/QUIC/Client/Reader.hs @@ -3,6 +3,7 @@ module Network.QUIC.Client.Reader ( readerClient, + rebind, recvClient, ConnectionControl (..), controlConnection, @@ -10,6 +11,9 @@ ) where import Control.Concurrent +import Control.Concurrent.STM +import Data.Functor (($>)) +import qualified Control.Monad.STM as STM import qualified Control.Exception as E import Data.List (intersect) import Network.Socket (Socket, close, connect, getSocketName) @@ -27,26 +31,35 @@ import Network.QUIC.Recovery import Network.QUIC.Socket import Network.QUIC.Types -import Network.QUIC.Windows --- | readerClient dies when the socket is closed. -readerClient :: Socket -> Connection -> IO () -readerClient s0 conn = handleLogUnit logAction $ do +-- What ended a wait in 'readerClient'. +data ReaderEnd = Retired | TimedOut + +-- | The reader of a client's socket. It returns when the socket is +-- retired, which is how 'rebind' ends the reader of the socket it +-- replaced, and when the connection has been idle for too long. +-- +-- Nothing is closed and nothing is thrown to make it return: the receive +-- and both of those are waited for at once. The socket it was reading is +-- the caller's to close afterwards -- __not at once, though__: nothing +-- here says when this has returned, so a caller which closes before it +-- has is back to closing a socket under a reader. 'rebind' leaves the +-- delay it is given before closing, for that. +readerClient :: TVar Bool -> Socket -> Connection -> IO () +readerClient retired s0 conn = handleLogUnit logAction $ do labelMe "readerClient" wait connected <- getSockConnected conn peersa0 <- peerSockAddr <$> getPathInfo conn - -- The idle timeout below throws into this thread to end the wait, and - -- 'killReaders' does the same when the connection closes. Neither - -- reaches a thread blocked in a socket call on Windows, so the call -- - -- the call alone, not the loop around it -- goes to a thread of its own - -- there. Nothing is left racing for a datagram: the timeout closes the - -- socket and stops, and closing it is also what ends an abandoned call. - let recv - | connected = windowsThreadBlockHack $ NSB.recv s0 2048 + let recv giveUp + | connected = NSB.recvWith s0 2048 giveUp | otherwise = do - (bs, peersa) <- windowsThreadBlockHack $ NSB.recvFrom s0 2048 - if peersa /= peersa0 then recv else return bs + r <- NSB.recvFromWith s0 2048 giveUp + case r of + Left a -> return $ Left a + Right (bs, peersa) + | peersa /= peersa0 -> recv giveUp + | otherwise -> return $ Right bs loop recv where wait = do @@ -57,11 +70,17 @@ yield wait loop recv = do - ito <- readMinIdleTimeout conn - mbs <- timeout ito "readeClient" recv - case mbs of - Nothing -> close s0 - Just bs -> do + Microseconds ito <- readMinIdleTimeout conn + expired <- registerDelay ito + ebs <- + recv $ + ((readTVar retired >>= STM.check) $> Retired) + `orElse` ((readTVar expired >>= STM.check) $> TimedOut) + case ebs of + -- The socket belongs to whoever retired it; it closes it. + Left Retired -> return () + Left TimedOut -> close s0 + Right bs -> do now <- getTimeMicrosecond let quicBit = greaseQuicBit $ getMyParameters conn pkts <- decodePackets bs (not quicBit) @@ -242,7 +261,14 @@ newSock <- natRebinding peersa connected <- getSockConnected conn when connected $ connect newSock peersa - oldSock <- setSocket conn newSock - let reader = readerClient newSock conn - forkManaged conn "QUIC reader" reader - fire conn microseconds $ close oldSock + newRetired <- newTVarIO False + (oldSock, oldRetired) <- swapSocket conn newSock newRetired + (forkManaged conn "QUIC reader" $ readerClient newRetired newSock conn) + `E.finally` do + -- Telling the old reader to go rather than closing the socket + -- under it: a close ends a thread blocked on a descriptor only + -- because the event manager says so, and a descriptor number is + -- reused. Said even if the fork above failed, since the old + -- socket is not this connection's any more either way. + atomically $ writeTVar oldRetired True + fire conn microseconds $ close oldSock 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 @@ -134,15 +134,16 @@ flip E.onException (NS.close sock) $ do when ccSockConnected $ NS.connect sock peersa q <- newRecvQ - sref <- newIORef sock + retire0 <- newTVarIO False + sref <- newIORef (sock, retire0) pathInfo <- newPathInfo peersa piref <- newIORef $ PeerInfo pathInfo Nothing let send buf siz | ccSockConnected = do - s <- readIORef sref + (s, _) <- readIORef sref void $ NS.sendBuf s buf siz | otherwise = do - s <- readIORef sref + (s, _) <- readIORef sref PeerInfo pinfo _ <- readIORef piref void $ NS.sendBufTo s buf siz $ peerSockAddr pinfo recv = recvClient q @@ -195,7 +196,8 @@ setMaxPacketSize conn pktSiz setInitialCongestionWindow (connLDCC conn) pktSiz setAddressValidated pathInfo - let reader = readerClient sock conn -- dies when s0 is closed. + retired <- getSocketRetire conn + let reader = readerClient retired sock conn -- Handing the qlog over. Nothing below can throw, so from here -- 'freeResources' is the only releaser. addResource conn qclean diff --git a/Network/QUIC/Closer.hs b/Network/QUIC/Closer.hs --- a/Network/QUIC/Closer.hs +++ b/Network/QUIC/Closer.hs @@ -3,7 +3,9 @@ module Network.QUIC.Closer (closure) where import Control.Concurrent +import Control.Concurrent.STM import qualified Control.Exception as E +import Data.Either (isRight) import Foreign.Marshal.Alloc import Foreign.Ptr import qualified Network.Socket as NS @@ -18,7 +20,6 @@ import Network.QUIC.Recovery import Network.QUIC.Sender import Network.QUIC.Types -import Network.QUIC.Windows closure :: Connection -> LDCC -> Either E.SomeException a -> IO a closure conn ldcc (Right x) = do @@ -98,25 +99,33 @@ | otherwise = void $ NS.sendBufTo sock sendbuf siz peersa -- recv and clos killReaders conn -- client only - let (recv, clos) = case mrecvbuf of - Nothing -> (void $ connRecv conn, return ()) + let (recvWithin, clos) = case mrecvbuf of + Nothing -> + -- The server reads its datagrams from a queue, not from + -- a socket, and a timeout does reach a thread waiting + -- on one. + ( \tmo -> isJust <$> timeout tmo "closer" (void $ connRecv conn) + , return () + ) Just recvbuf -> - -- 'closer' puts a timeout around this, and a timeout - -- does not reach a thread blocked in a socket call on - -- Windows, so the call goes to a thread of its own - -- there. Each timeout abandons one; there are six at - -- most, they share a buffer nothing reads, and 'clos' - -- below closes the socket and ends them all. - let recv' - | connected = - windowsThreadBlockHack $ - void $ - NS.recvBuf sock recvbuf bufsiz - | otherwise = do - (_, sa) <- - windowsThreadBlockHack $ - NS.recvBufFrom sock recvbuf bufsiz - when (sa /= peersa) recv' + -- The deadline is waited for beside the receive rather + -- than thrown at it: an asynchronous exception does not + -- reach a thread blocked in a socket call on Windows. + -- Nothing is abandoned, so nothing has to be closed to + -- end it either. + let recv' (Microseconds us) = do + expired <- registerDelay us + let giveUp = check =<< readTVar expired + again = recv' (Microseconds us) + if connected + then isRight <$> NS.recvBufWith sock recvbuf bufsiz giveUp + else do + r <- NS.recvBufFromWith sock recvbuf bufsiz giveUp + case r of + Left () -> return False + Right (_, sa) + | sa /= peersa -> again + | otherwise -> return True clos' = do NS.close sock -- This is just in case. @@ -132,7 +141,7 @@ -- of its own misses them. The closer repeats it for a peer that -- did not hear it, which is what it is for. send - void $ forkFinally (closer conn pto send recv hook) $ \e -> do + void $ forkFinally (closer conn pto send recvWithin hook) $ \e -> do case e of Left e' -> connDebugLog conn $ "closure' " <> bhow e' Right _ -> return () @@ -170,8 +179,16 @@ else return 0 -closer :: Connection -> Microseconds -> IO () -> IO () -> IO () -> IO () -closer _conn (Microseconds pto) send recv hook = do +-- The fourth argument receives a datagram within the given time, +-- answering 'False' if none came. +closer + :: Connection + -> Microseconds + -> IO () + -> (Microseconds -> IO Bool) + -> IO () + -> IO () +closer _conn (Microseconds pto) send recvWithin hook = do labelMe "QUIC closer" loop (3 :: Int) where @@ -179,15 +196,11 @@ loop n = do send getTimeMicrosecond >>= skip (Microseconds pto) - mx <- timeout (Microseconds (pto !>>. 1)) "closer 1" recv - case mx of - Nothing -> hook - Just () -> loop (n - 1) + got <- recvWithin $ Microseconds (pto !>>. 1) + if got then loop (n - 1) else hook skip tmo@(Microseconds duration) base = do - mx <- timeout tmo "closer 2" recv - case mx of - Nothing -> return () - Just () -> do - Microseconds elapsed <- getElapsedTimeMicrosecond base - let duration' = duration - elapsed - when (duration' >= 5000) $ skip (Microseconds duration') base + got <- recvWithin tmo + when got $ do + Microseconds elapsed <- getElapsedTimeMicrosecond base + let duration' = duration - elapsed + when (duration' >= 5000) $ skip (Microseconds duration') base diff --git a/Network/QUIC/Connection/Misc.hs b/Network/QUIC/Connection/Misc.hs --- a/Network/QUIC/Connection/Misc.hs +++ b/Network/QUIC/Connection/Misc.hs @@ -9,6 +9,8 @@ getOriginalVersion, getSocket, setSocket, + getSocketRetire, + swapSocket, clearSocket, getPathInfo, addPathInfo, @@ -35,6 +37,7 @@ abortConnection, ) where +import Control.Concurrent.STM (TVar, atomically, writeTVar) import qualified Control.Exception as E import Network.Socket (SockAddr, Socket) import qualified System.ThreadManager as T @@ -68,14 +71,29 @@ ---------------------------------------------------------------- getSocket :: Connection -> IO Socket -getSocket Connection{..} = readIORef connSocket +getSocket Connection{..} = fst <$> readIORef connSocket +-- | Swapping the socket alone, leaving the reader of the old one to be +-- retired through the variable they share. setSocket :: Connection -> Socket -> IO Socket -setSocket Connection{..} sock = atomicModifyIORef' connSocket (sock,) +setSocket Connection{..} sock = + atomicModifyIORef' connSocket $ \(old, retire) -> ((sock, retire), old) +-- | The variable which retires the reader of the current socket. +getSocketRetire :: Connection -> IO (TVar Bool) +getSocketRetire Connection{..} = snd <$> readIORef connSocket + +-- | Swapping the socket and the variable which retires its reader at +-- once, returning the pair they replace. The two go together, so +-- nothing can see one of them swapped and not the other. +swapSocket :: Connection -> Socket -> TVar Bool -> IO (Socket, TVar Bool) +swapSocket Connection{..} sock retire = + atomicModifyIORef' connSocket ((sock, retire),) + -- fixme clearSocket :: Connection -> IO Socket -clearSocket Connection{..} = atomicModifyIORef' connSocket (undefined,) +clearSocket Connection{..} = + atomicModifyIORef' connSocket $ \(old, retire) -> ((undefined, retire), old) ---------------------------------------------------------------- @@ -207,7 +225,14 @@ -- | Killing the threads of 'forkManaged'. One forked after this is -- refused rather than run. killReaders :: Connection -> IO () -killReaders Connection{..} = T.stopAfter readers (return ()) $ \_ -> return () +killReaders conn@Connection{..} = do + -- Telling the reader of the current socket to go. 'stopAfter' below + -- throws into the threads it manages, and an asynchronous exception + -- does not reach one blocked in a socket call on Windows; the reader + -- waits on this variable beside its receive instead. + retired <- getSocketRetire conn + atomically $ writeTVar retired True + T.stopAfter readers (return ()) $ \_ -> return () ---------------------------------------------------------------- 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 @@ -262,7 +262,10 @@ -- Manage , connRecvQ :: RecvQ , connRecvDatagramQ :: DatagramQ - , connSocket :: IORef Socket + , connSocket :: IORef (Socket, TVar Bool) + -- ^ The socket and the variable which retires its reader: setting it + -- tells the reader that the socket is not its to read any more. + -- One reference, so that the two cannot be swapped apart. , genStatelessResetToken :: CID -> StatelessResetToken , readers :: T.ThreadManager , mainThreadId :: ThreadId @@ -369,7 +372,7 @@ -> DebugLogger -> QLogger -> Hooks - -> IORef Socket + -> IORef (Socket, TVar Bool) -> IORef PeerInfo -> RecvQ -> DatagramQ @@ -469,7 +472,7 @@ -> DebugLogger -> QLogger -> Hooks - -> IORef Socket + -> IORef (Socket, TVar Bool) -> IORef PeerInfo -> RecvQ -> DatagramQ @@ -493,7 +496,7 @@ -> DebugLogger -> QLogger -> Hooks - -> IORef Socket + -> IORef (Socket, TVar Bool) -> IORef PeerInfo -> RecvQ -> DatagramQ diff --git a/Network/QUIC/Internal.hs b/Network/QUIC/Internal.hs --- a/Network/QUIC/Internal.hs +++ b/Network/QUIC/Internal.hs @@ -14,11 +14,15 @@ module Network.QUIC.Utils, module Network.QUIC.Recovery, module Network.QUIC.Client.Reader, - module Network.QUIC.Windows, module Network.QUIC.Socket, ) where -import Network.QUIC.Client.Reader (ConnectionControl (..), controlConnection) +import Network.QUIC.Client.Reader ( + ConnectionControl (..), + controlConnection, + readerClient, + rebind, + ) import Network.QUIC.Config import Network.QUIC.Connection import Network.QUIC.Connector @@ -34,4 +38,3 @@ import Network.QUIC.TLS import Network.QUIC.Types import Network.QUIC.Utils -import Network.QUIC.Windows 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 @@ -12,7 +12,6 @@ tokenMgr, genStatelessReset, stopServerNow, - wakeDispatcher, -- * Accepting Accept (..), @@ -25,24 +24,21 @@ import Control.Concurrent import Control.Concurrent.STM +import qualified Control.Monad.STM as STM import qualified Control.Exception as E import qualified Crypto.Token as CT import qualified Data.ByteString as BS import Data.Map.Strict (Map) import qualified Data.Map.Strict as M -import qualified GHC.IO.Exception as E import Network.ByteOrder import Network.Control (LRUCacheRef, Rate, getRate, newRate) import qualified Network.Control as LRUCache import Network.Socket ( + isClosedSocketError, SockAddr (..), Socket, - getSocketName, - tupleToHostAddress, - tupleToHostAddress6, ) import qualified Network.Socket.ByteString as NSB -import qualified System.IO.Error as E import System.Log.FastLogger import qualified System.ThreadManager as T @@ -182,37 +178,6 @@ data ServerState = Running | Stopped deriving (Eq, Show) --- | Waking a dispatcher that is waiting for a datagram, by sending it one. --- --- The dispatchers used to wait for the socket to become readable and for the --- stop variable at once, in a single 'atomically', and so saw a stop where --- they waited. Readiness is not something every I/O manager has to report: --- a completion-based one has nothing to say about a socket being readable, --- and on Windows the wait reaches 'waitRead#', which the threaded RTS does --- not merely refuse but aborts the process over. So the wait is a plain --- receive now, and stopping has to put something into it. --- --- An empty datagram to the socket's own address does that, and keeps what --- the shutdown handler promises: no socket is closed and nothing is raised. --- The loop reads the stop variable again when the receive returns and ends --- there. An empty datagram carries no packet, so one arriving from anywhere --- else is just as harmless. -wakeDispatcher :: Socket -> IO () -wakeDispatcher s = E.handle ignore $ do - sa <- getSocketName s - void $ NSB.sendTo s "" $ reachable sa - --- | The address a socket can be reached on from the host it is bound on. --- A wildcard bind is not an address to send to, so the loopback stands in --- for it. -reachable :: SockAddr -> SockAddr -reachable (SockAddrInet p a) - | a == 0 = SockAddrInet p $ tupleToHostAddress (127, 0, 0, 1) -reachable (SockAddrInet6 p f a sc) - | a == (0, 0, 0, 0) = - SockAddrInet6 p f (tupleToHostAddress6 (0, 0, 0, 0, 0, 0, 0, 1)) sc -reachable sa = sa - dispatcher :: Dispatch -> ServerConfig @@ -224,53 +189,54 @@ labelMe "QUIC dispatcher" handleLogUnit logAction loop where - -- The loop used to wait for the socket to become readable, watching the - -- stop variable in the same 'atomically' so that 'stop' broke it without - -- an exception. Readiness is not a thing a completion-based I/O manager - -- reports, so on Windows that wait is not available at all; the loop - -- simply blocks in the receive instead, and 'stop' is answered by the - -- 'killThread' and the 'close' that 'run's own teardown does next. The - -- stop variable is still read, for the datagram that arrives between the - -- two. + -- The receive and the stop variable are waited for at once, so stopping + -- the dispatcher needs nothing put into the socket and nothing thrown + -- into this thread: 'recvFromWith' gives up on the receive itself. It + -- used to be woken with an empty datagram to its own address, which was + -- the only way that held on Windows. loop = do - st <- readTVarIO stvar - when (st == Running) $ do - (bs, peersa) <- safeRecv $ NSB.recvFrom mysock 2048 - now <- getTimeMicrosecond - let send' b = void $ NSB.sendTo mysock b peersa - -- cf: greaseQuicBit $ getMyParameters conn - quicBit = greaseQuicBit $ scParameters conf - let bytes = BS.length bs - switch = dispatch d conf forkConnection logAction mysock peersa send' bytes now - -- One datagram's failure is one datagram's problem. - -- - -- Letting it out of the loop ends the dispatcher, and nothing - -- restarts it. The socket stays bound, because 'run' holds it - -- for as long as the server lives, so the server goes on looking - -- like a server and answers nothing for the rest of its life -- - -- with the handler above logging it to a logger that discards - -- what it is given. Whatever a peer could find that raises in - -- here would be one datagram against the whole server, from any - -- address, before any handshake. - handleLogUnit logAction $ do - cpckts <- decodeCryptPackets bs (not quicBit) - mapM_ switch cpckts - loop + er <- safeRecv $ NSB.recvFromWith mysock 2048 stopped + case er of + Left () -> return () + Right (bs, peersa) -> do + now <- getTimeMicrosecond + let send' b = void $ NSB.sendTo mysock b peersa + -- cf: greaseQuicBit $ getMyParameters conn + quicBit = greaseQuicBit $ scParameters conf + let bytes = BS.length bs + switch = dispatch d conf forkConnection logAction mysock peersa send' bytes now + -- One datagram's failure is one datagram's problem. + -- + -- Letting it out of the loop ends the dispatcher, and nothing + -- restarts it. The socket stays bound, because 'run' holds it + -- for as long as the server lives, so the server goes on + -- looking like a server and answers nothing for the rest of + -- its life -- with the handler above logging it to a logger + -- that discards what it is given. Whatever a peer could find + -- that raises in here would be one datagram against the whole + -- server, from any address, before any handshake. + handleLogUnit logAction $ do + cpckts <- decodeCryptPackets bs (not quicBit) + mapM_ switch cpckts + loop + stopped = do + st <- readTVar stvar + STM.check $ st /= Running + logAction _msg = return () - -- No thread of its own for the receive. The dispatcher is not ended by - -- an exception thrown into it -- it is told, and sees it where it waits - -- -- so it does not need to be interruptible there, and a forked thread - -- for every datagram a server receives is 4.5 microseconds of each on - -- Windows. + -- One datagram's failure is one datagram's problem here too: a receive + -- which fails is tried again rather than ending the dispatcher. safeRecv rcv = do ex <- E.try 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 + -- The socket was closed under the receive, which is not a + -- datagram's failure and not something to try again. + Just e | isClosedSocketError e -> E.throwIO se _ -> do logAction $ "recv again: " <> bhow se rcv 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 @@ -44,11 +44,7 @@ run conf server = do labelMe "QUIC run" stvar <- newTVarIO Running - -- The sockets do not exist yet and the stop has to reach them, so what - -- stops this server is built here and what it wakes is filled in by - -- 'setup'. See 'wakeDispatcher'. - wakeRef <- newIORef $ return () - let stopNow = stopping stvar wakeRef + let stopNow = stopping stvar installShutdownHandler conf stopNow -- Outside handleLogUnit on purpose. If the addresses cannot be bound -- there is no server, and that is the caller's business: swallowing it @@ -56,7 +52,7 @@ -- onServerReady, and left anyone waiting on that hook waiting for good. -- An IOSpec run wedged for two and a half days that way, on a port the -- previous test had not finished releasing. - E.bracket (setup stvar stopNow wakeRef) (teardown stopNow) $ \(_, _, _) -> handleLogUnit debugLog $ do + E.bracket (setup stvar stopNow) (teardown stopNow) $ \(_, _, _) -> handleLogUnit debugLog $ do onServerReady $ scHooks conf atomically $ do st <- readTVar stvar @@ -68,24 +64,22 @@ -- port is taken leaves the first socket bound and the token manager -- thread running, for good. Each step frees what the steps before it -- took, which for a list means each element frees itself. - setup stvar stopNow wakeRef = do + setup stvar stopNow = do dispatch <- newDispatch conf stopNow let forkConn acc = spawnConnection dispatch $ runServer conf server dispatch acc flip E.onException (clearDispatch dispatch) $ do ssas <- openAll $ scAddresses conf - writeIORef wakeRef $ mapM_ wakeDispatcher ssas tids <- runAll dispatch conf stvar forkConn ssas `E.onException` mapM_ NS.close ssas return (dispatch, tids, ssas) - -- Telling the dispatchers first. They end where they wait, on the - -- datagram 'stopping' sends them, and the 'killThread' below then has - -- nothing to reach -- which matters because on Windows it could not - -- have reached them anyway: a thread blocked in a socket call there is - -- not interruptible. Running them on a thread of their own so that it - -- could be was a forked thread for every datagram the server received. + -- Telling the dispatchers first. They end where they wait, on the stop + -- variable 'stopping' sets, and the 'killThread' below then has nothing + -- to reach -- which matters because on Windows it could not have + -- reached them anyway: a thread blocked in a socket call there is not + -- interruptible. -- -- The usual way here is through the shutdown handler, which has already -- done this; doing it again is free. The way that had not, until now, @@ -108,9 +102,7 @@ runWithSockets ssas conf server = do labelMe "QUIC runWithSockets" stvar <- newTVarIO Running - -- As in 'run', except that the sockets are in hand already. - wakeRef <- newIORef $ mapM_ wakeDispatcher ssas - let stopNow = stopping stvar wakeRef + let stopNow = stopping stvar installShutdownHandler conf stopNow -- As in 'run'. E.bracket (setup stvar stopNow) (teardown stopNow) $ \(_, _) -> handleLogUnit debugLog $ do @@ -156,10 +148,11 @@ -- | Telling this server to stop: the state the dispatchers read, and then -- the datagram that gets them to read it. -stopping :: TVar ServerState -> IORef (IO ()) -> IO () -stopping stvar wakeRef = do - atomically $ writeTVar stvar Stopped - join $ readIORef wakeRef +-- The dispatchers wait on this variable and on their receive at once, so +-- setting it is all there is to stopping them: nothing has to be sent to +-- the sockets and nothing has to be thrown into the threads. +stopping :: TVar ServerState -> IO () +stopping stvar = atomically $ writeTVar stvar Stopped -- | Ending the connections the server still has, while the sockets are -- still open. @@ -350,11 +343,12 @@ -> Accept -> IO ConnRes createServerConnection conf@ServerConfig{..} dispatch Accept{..} = do - sref <- newIORef accMySocket + retire0 <- newTVarIO False + sref <- newIORef (accMySocket, retire0) pathInfo <- newPathInfo accPeerSockAddr piref <- newIORef $ PeerInfo pathInfo Nothing let send buf siz = void $ do - sock <- readIORef sref + (sock, _) <- readIORef sref PeerInfo pinfo _ <- readIORef piref NS.sendBufTo sock buf siz $ peerSockAddr pinfo recv = recvServer accRecvQ diff --git a/Network/QUIC/Windows.hs b/Network/QUIC/Windows.hs deleted file mode 100644 --- a/Network/QUIC/Windows.hs +++ /dev/null @@ -1,46 +0,0 @@ -{-# LANGUAGE CPP #-} - --- | Making a blocking socket call interruptible on Windows. -module Network.QUIC.Windows ( - windowsThreadBlockHack, -) where - -#if defined(mingw32_HOST_OS) -import Control.Concurrent -import qualified Control.Exception as E -import GHC.Conc.Sync (labelThread) - --- | Running a blocking call on a thread of its own, and waiting for it --- here. --- --- A thread blocked in a socket call on Windows cannot be reached by an --- asynchronous exception: neither 'killThread' nor the timeouts quic builds --- on the event manager get at it, and the readiness this package used to --- wait for instead -- 'threadWaitReadSTM' and so the socket's STM wrappers --- -- is not available under the native I/O manager at all, because --- completion ports have nothing to say about a socket being readable. --- --- So the call goes to a thread of its own and this one waits on an MVar, --- which is interruptible. It is the classic answer; warp has used it since --- 3.2.17. --- --- The call is abandoned rather than cancelled. When this thread is --- interrupted the forked one stays in the call until the socket is closed, --- so this belongs only where the socket is closed soon after -- and where --- another reader starting on the same socket in the meantime would do no --- harm, since the two would then race for the next datagram. --- --- On every other platform the thread can be reached and this is 'id'. -windowsThreadBlockHack :: IO a -> IO a -windowsThreadBlockHack act = do - var <- newEmptyMVar :: IO (MVar (Either E.SomeException a)) - tid <- forkIO $ E.try act >>= putMVar var - labelThread tid "QUIC blocking call" - res <- takeMVar var - case res of - Left e -> E.throwIO e - Right r -> return r -#else -windowsThreadBlockHack :: IO a -> IO a -windowsThreadBlockHack = id -#endif diff --git a/quic.cabal b/quic.cabal --- a/quic.cabal +++ b/quic.cabal @@ -1,6 +1,6 @@ cabal-version: 2.0 name: quic -version: 0.3.18 +version: 0.3.19 license: BSD3 license-file: LICENSE maintainer: kazu@iij.ad.jp @@ -119,7 +119,6 @@ Network.QUIC.Types.Resumption Network.QUIC.Types.Time Network.QUIC.Utils - Network.QUIC.Windows default-language: Haskell2010 default-extensions: Strict StrictData @@ -140,7 +139,7 @@ fast-logger >=3.2.8 && <3.3, filepath, iproute >=1.7.12 && <1.8, - network >=3.2.3, + network >=3.2.10, network-byte-order >=0.1.7 && <0.2, network-control >=0.1.5 && <0.2, ram, @@ -268,7 +267,7 @@ fast-logger >=3.2.8 && <3.3, filepath, hspec, - network >=3.2.2, + network >=3.2.10, quic, ram, stm, diff --git a/test/Config.hs b/test/Config.hs --- a/test/Config.hs +++ b/test/Config.hs @@ -18,6 +18,7 @@ ) where import Control.Concurrent +import Control.Concurrent.STM import qualified Control.Exception as E import Control.Monad import Data.Bits ((.&.)) @@ -228,6 +229,7 @@ E.bracket (startRelay saS sockC sockS irefC irefS) stopRelay $ \_ -> body where startRelay saS sockC sockS irefC irefS = do + stop <- newTVarIO False -- Where to send what comes back from the server. Set once the client -- has introduced itself, which is before the server can have anything -- to say about it. @@ -262,13 +264,13 @@ -- of a datagram can say which test it belongs to. A test that -- leaves a client behind breaks the next one however this -- chooses, so a test must not leave one behind. - (bs, saO) <- waitForClientHello sockC - writeIORef peerRef $ Just saO - n0 <- atomicModifyIORef' irefC $ \x -> (x + 1, x) - dropPacket0 <- shouldDrop scenario True n0 - unless dropPacket0 $ void $ sendTo sockS bs saS - forever $ do - (bs1, sa) <- relayRecv sockC + mhello <- waitForClientHello stop sockC + forM_ mhello $ \(bs, saO) -> do + writeIORef peerRef $ Just saO + n0 <- atomicModifyIORef' irefC $ \x -> (x + 1, x) + dropPacket0 <- shouldDrop scenario True n0 + unless dropPacket0 $ void $ sendTo sockS bs saS + relayLoop stop sockC $ \(bs1, sa) -> do -- Only from the client we latched onto. The connect this -- replaces did that in the kernel; doing it here keeps -- leftovers from the connection that just closed from being @@ -282,8 +284,7 @@ void $ sendTo sockS bs1 saS -- from server - tid1 <- forkIO $ forever $ do - (bs, _) <- relayRecv sockS + tid1 <- forkIO $ relayLoop stop sockS $ \(bs, _) -> do n <- atomicModifyIORef' irefS $ \x -> (x + 1, x) dropPacket <- shouldDrop scenario False n let isCC = BS.length bs < 200 @@ -291,18 +292,30 @@ mpeer <- readIORef peerRef forM_ mpeer $ \sa -> delayIf (shouldDelay scenario False n) $ void $ sendTo sockC bs sa - return (tid0, tid1) - stopRelay (tid0, tid1) = killThread tid0 >> killThread tid1 - -- 'stopRelay' kills these threads and the brackets close the sockets - -- straight after, and a thread blocked in a socket call on Windows is - -- not reachable by 'killThread' -- it would be left in the call and - -- would die of the close instead, from a thread nobody is watching. - relayRecv sock = windowsThreadBlockHack $ recvFrom sock 2048 - waitForClientHello sockC = do - (bs, saO) <- relayRecv sockC - if isClientHello bs - then return (bs, saO) - else waitForClientHello sockC + return (stop, tid0, tid1) + -- Telling these threads to stop rather than killing them: a thread + -- blocked in a socket call on Windows is not reachable by 'killThread', + -- and the brackets close the sockets straight after. The kill stays + -- for a thread which is somewhere else. + stopRelay (stop, tid0, tid1) = do + atomically $ writeTVar stop True + killThread tid0 + killThread tid1 + relayRecv stop sock = recvFromWith sock 2048 (check =<< readTVar stop) + relayLoop stop sock act = loop + where + loop = do + r <- relayRecv stop sock + case r of + Left () -> return () + Right x -> act x >> loop + waitForClientHello stop sockC = do + r <- relayRecv stop sockC + case r of + Left () -> return Nothing + Right (bs, saO) + | isClientHello bs -> return $ Just (bs, saO) + | otherwise -> waitForClientHello stop sockC isClientHello bs = BS.length bs >= 1200 && BS.head bs .&. 0x80 /= 0 hints = diff --git a/test/MigrationSpec.hs b/test/MigrationSpec.hs --- a/test/MigrationSpec.hs +++ b/test/MigrationSpec.hs @@ -1,6 +1,10 @@ module MigrationSpec where +import Control.Concurrent import Control.Concurrent.Async (mapConcurrently) +import Control.Concurrent.STM +import Control.Monad (void) +import Network.Socket (connect, getSocketName) import Test.Hspec import Network.QUIC.Internal @@ -27,3 +31,44 @@ (_, serverConn) <- makeConnections serverConf Version1 claimMigration serverConn `shouldReturn` True claimMigration serverConn `shouldReturn` False + + -- 'rebind' swaps a new socket in and has to end the reader of the old + -- one. It used to do that by closing the old socket under it, which + -- works only because the event manager tells a blocked thread that the + -- descriptor closed; the reader is told now. Retiring the wrong one of + -- the two leaves a connection whose socket nobody reads, and both + -- readers are alive either way, so this watches which variable is set. + -- See #173. + describe "rebind" $ do + it "retires the reader of the socket it replaced, not the new one" $ do + (clientConn, _) <- makeConnections serverConf Version1 + oldSock <- getSocket clientConn + oldRetired <- getSocketRetire clientConn + rebind clientConn $ Microseconds 1000000 + newSock <- getSocket clientConn + newRetired <- getSocketRetire clientConn + readTVarIO oldRetired `shouldReturn` True + readTVarIO newRetired `shouldReturn` False + (newSock == oldSock) `shouldBe` False + -- The reader 'rebind' forked is on the new socket; let it go. + atomically $ writeTVar newRetired True + + it "lets a retired reader return, leaving its socket open" $ do + (clientConn, _) <- makeConnections serverConf Version1 + sock <- getSocket clientConn + -- The reader waits for the socket to be bound, and on Windows + -- getSocketName says so only once it is. A client binds it by + -- sending; nothing is sent here, so connect does it instead. + peersa <- peerSockAddr <$> getPathInfo clientConn + connect sock peersa + retired <- getSocketRetire clientConn + done <- newEmptyMVar + void $ forkIO $ readerClient retired sock clientConn >> putMVar done () + threadDelay 100000 + atomically $ writeTVar retired True + returned <- timeout (Microseconds 2000000) "retired reader" $ takeMVar done + returned `shouldBe` Just () + -- The socket belongs to whoever retired the reader: it is still + -- open for them to close, which getSocketName would not be on a + -- closed one. + void $ getSocketName sock diff --git a/test/PacketSpec.hs b/test/PacketSpec.hs --- a/test/PacketSpec.hs +++ b/test/PacketSpec.hs @@ -130,7 +130,8 @@ let clientConf = testClientConfig (sock, peersa) <- clientSocket "127.0.0.1" "2000" q <- newRecvQ - sref <- newIORef sock + retire <- newTVarIO False + sref <- newIORef (sock, retire) pathInfo <- newPathInfo peersa piref <- newIORef $ PeerInfo pathInfo Nothing let ver = v