quic 0.3.18 → 0.3.19

raw patch · 14 files changed

+310/−240 lines, 14 filesdep ~networkPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: network

API changes (from Hackage documentation)

Files

• ChangeLog.md +26/−0
• Network/QUIC/Client/Reader.hs +49/−23
• Network/QUIC/Client/Run.hs +6/−4
• Network/QUIC/Closer.hs +46/−33
• Network/QUIC/Connection/Misc.hs +29/−4
• Network/QUIC/Connection/Types.hs +7/−4
• Network/QUIC/Internal.hs +6/−3
• Network/QUIC/Server/Reader.hs +40/−74
• Network/QUIC/Server/Run.hs +17/−23
• Network/QUIC/Windows.hs +0/−46
• quic.cabal +3/−4
• test/Config.hs +34/−21
• test/MigrationSpec.hs +45/−0
• test/PacketSpec.hs +2/−1

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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 ()  ---------------------------------------------------------------- 

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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,

@@ -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 =

@@ -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

@@ -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