diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,9 @@ # ChangeLog +## 0.2.21 + +* Terminating threads properly with event-poll model. + ## 0.2.20 * Fix recvStream overflow on pending data. 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 @@ -77,23 +77,24 @@ if ccUse0RTT conf then wait0RTTReady conn else wait1RTTReady conn - client0 conn + r <- client0 conn + mainDone conn + return r ldcc = connLDCC conn let s1 = labelMe "handshaker" >> handshaker s2 = labelMe "sender" >> sender conn s3 = labelMe "receiver" >> receiver conn s4 = labelMe "resender" >> resender ldcc s5 = labelMe "ldccTimer" >> ldccTimer ldcc + s6 = labelMe "QUIC client" >> client c1 = labelMe "concurrently1" >> concurrently_ s1 s2 c2 = labelMe "concurrently2" >> concurrently_ c1 s3 c3 = labelMe "concurrently3" >> concurrently_ c2 s4 c4 = labelMe "concurrently4" >> concurrently_ c3 s5 - supporters = c4 - runThreads = do - er <- race supporters (labelMe "QUIC client" >> client) - case er of - Left () -> E.throwIO MustNotReached - Right r -> return r + c5 = + labelMe "concurrently5" + >> concurrently (c4 `E.catch` \(_ :: InternalControl) -> return ()) s6 + runThreads = snd <$> c5 when (ccWatchDog conf) $ forkManaged conn $ watchDog conn ex <- E.try runThreads sendFinal conn 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 @@ -280,6 +280,7 @@ , minIdleTimeout :: IORef Microseconds , bytesTx :: IORef Int , bytesRx :: IORef Int + , connDone :: TVar Bool , -- TLS pendingQ :: Array EncryptionLevel (TVar [ReceivedPacket]) , ciphers :: IOArray EncryptionLevel Cipher @@ -376,6 +377,7 @@ minIdleTimeout <- newIORef (milliToMicro $ maxIdleTimeout myParameters) bytesTx <- newIORef 0 bytesRx <- newIORef 0 + connDone <- newTVarIO False -- TLS pendingQ <- makePendingQ ciphers <- newArray (InitialLevel, RTT1Level) defaultCipher @@ -484,3 +486,8 @@ shared1RTTReady <- newIORef False sharedSendStreamQ <- newTQueueIO return Shared{..} + +---------------------------------------------------------------- + +mainDone :: Connection -> IO () +mainDone conn = atomically $ writeTVar (connDone conn) True diff --git a/Network/QUIC/IO.hs b/Network/QUIC/IO.hs --- a/Network/QUIC/IO.hs +++ b/Network/QUIC/IO.hs @@ -182,8 +182,9 @@ -- an empty bytestring is returned. recvStream :: Stream - -> Int -- ^ Number of bytes to receive. In certain cases, `recvStream` can return - -- fewer bytes than requested, but never more bytes than requested.. + -> Int + -- ^ Number of bytes to receive. In certain cases, `recvStream` can return + -- fewer bytes than requested, but never more bytes than requested.. -> IO ByteString recvStream s n = do bs <- takeRecvStreamQwithSize s n diff --git a/Network/QUIC/Receiver.hs b/Network/QUIC/Receiver.hs --- a/Network/QUIC/Receiver.hs +++ b/Network/QUIC/Receiver.hs @@ -219,7 +219,9 @@ streamNotCreatedYet conn sid - "a locally-initiated stream that has not yet been created" + ( "a locally-initiated stream that has not yet been created, sid = " + <> shortpack (show sid) + ) guardStream _ _ _ = return () -- fixme: what about unidirection stream? diff --git a/Network/QUIC/Recovery.hs b/Network/QUIC/Recovery.hs --- a/Network/QUIC/Recovery.hs +++ b/Network/QUIC/Recovery.hs @@ -1,6 +1,7 @@ module Network.QUIC.Recovery ( -- Interface checkWindowOpenSTM, + isEmptyPingSTM, takePingSTM, speedup, resender, diff --git a/Network/QUIC/Recovery/Interface.hs b/Network/QUIC/Recovery/Interface.hs --- a/Network/QUIC/Recovery/Interface.hs +++ b/Network/QUIC/Recovery/Interface.hs @@ -3,6 +3,7 @@ module Network.QUIC.Recovery.Interface ( checkWindowOpenSTM, takePingSTM, + isEmptyPingSTM, speedup, resender, ) where @@ -24,6 +25,9 @@ checkWindowOpenSTM LDCC{..} siz = do CC{..} <- readTVar recoveryCC check (siz <= congestionWindow - bytesInFlight) + +isEmptyPingSTM :: LDCC -> STM Bool +isEmptyPingSTM LDCC{..} = isNothing <$> readTVar ptoPing takePingSTM :: LDCC -> STM EncryptionLevel takePingSTM LDCC{..} = do diff --git a/Network/QUIC/Sender.hs b/Network/QUIC/Sender.hs --- a/Network/QUIC/Sender.hs +++ b/Network/QUIC/Sender.hs @@ -228,8 +228,23 @@ | SwStrm TxStreamData sender :: Connection -> IO () -sender conn = handleLogT logAction $ forever $ sendP conn +sender conn = handleLogT logAction loop where + loop = do + exit <- atomically $ do + done <- readTVar $ connDone conn + a <- isEmptyPingSTM (connLDCC conn) + b <- isEmptyOutputSTM conn + c <- isEmptyStreamSTM conn + if done + then return (a && b && c) + else if (not a || not b || not c) then return False else retry + if exit + then + E.throwIO ExitConnection + else do + sendP conn + loop logAction msg = connDebugLog conn ("debug: sender: " <> msg) sendP :: Connection -> IO () 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 @@ -270,7 +270,7 @@ unless exist $ do let reg = registerConnectionDict dstTable unreg cid = - fire' (Microseconds 10000000) $ unregisterConnectionDict dstTable cid + fire' (Microseconds 1000000) $ unregisterConnectionDict dstTable cid acc = Accept { accVersionInfo = VersionInfo peerVer myVersions 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 @@ -112,22 +112,22 @@ wait1RTTReady conn afterHandshakeServer conf conn server0 conn + mainDone conn ldcc = connLDCC conn let s1 = labelMe "handshaker" >> handshaker s2 = labelMe "sender" >> sender conn s3 = labelMe "receiver" >> receiver conn s4 = labelMe "resender" >> resender ldcc s5 = labelMe "ldccTimer" >> ldccTimer ldcc + s6 = labelMe "QUIC server" >> server c1 = labelMe "concurrently1" >> concurrently_ s1 s2 c2 = labelMe "concurrently2" >> concurrently_ c1 s3 c3 = labelMe "concurrently3" >> concurrently_ c2 s4 c4 = labelMe "concurrently4" >> concurrently_ c3 s5 - supporters = c4 - runThreads = do - er <- race supporters (labelMe "QUIC server" >> server) - case er of - Left () -> E.throwIO MustNotReached - Right r -> return r + c5 = + labelMe "concurrently5" + >> concurrently_ (c4 `E.catch` \(_ :: InternalControl) -> return ()) s6 + runThreads = c5 ex <- E.try runThreads sendFinal conn setConnectionClosed conn diff --git a/Network/QUIC/Stream/Reass.hs b/Network/QUIC/Stream/Reass.hs --- a/Network/QUIC/Stream/Reass.hs +++ b/Network/QUIC/Stream/Reass.hs @@ -41,7 +41,8 @@ takeRecvStreamQwithSize :: Stream - -> Int -- ^ Number of bytes to receive. + -> Int + -- ^ Number of bytes to receive. -> IO ByteString takeRecvStreamQwithSize strm siz0 = do eos <- getEndOfStream strm 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.20 +version: 0.2.21 license: BSD3 license-file: LICENSE maintainer: kazu@iij.ad.jp diff --git a/util/ClientX.hs b/util/ClientX.hs --- a/util/ClientX.hs +++ b/util/ClientX.hs @@ -2,7 +2,7 @@ {-# LANGUAGE RecordWildCards #-} module ClientX ( - Aux (..), + Misc (..), Cli, clientHQ, clientH3, @@ -11,35 +11,43 @@ import Control.Concurrent import Control.Concurrent.Async +import qualified Control.Exception as E +import Control.Monad import qualified Data.ByteString as BS +import qualified Data.ByteString.Char8 as C8 +import Data.IORef import Network.ByteOrder +import System.IO import H3 import Network.QUIC +import qualified Network.QUIC.Internal as QUIC -data Aux = Aux - { auxAuthority :: String - , auxDebug :: String -> IO () - , auxShow :: ByteString -> IO () - , auxCheckClose :: IO Bool +data Misc = Misc + { miscAuthority :: String + , miscDebug :: String -> IO () + , miscShow :: ByteString -> IO () + , miscCheckClose :: IO Bool + , miscH3NegoDone :: IORef Bool + , miscInteractive :: Bool } -type Cli = Aux -> [ByteString] -> Connection -> IO () +type Cli = Misc -> [ByteString] -> Connection -> IO () clientHQ :: Int -> Cli -clientHQ n0 aux paths conn = - foldr1 concurrently_ $ map (clientHQ' n0 aux conn) paths +clientHQ n0 misc paths conn = + foldr1 concurrently_ $ map (clientHQ' n0 misc conn) paths -clientHQ' :: Int -> Aux -> Connection -> ByteString -> IO () -clientHQ' n0 aux@Aux{..} conn path = loop n0 +clientHQ' :: Int -> Misc -> Connection -> ByteString -> IO () +clientHQ' n0 misc@Misc{..} conn path = loop n0 where cmd = "GET " <> path <> "\r\n" - loop 0 = auxDebug "Connection finished" + loop 0 = miscDebug "Connection finished" loop 1 = do - auxDebug "GET" + miscDebug "GET" get loop n = do - auxDebug "GET" + miscDebug "GET" get threadDelay 100000 loop (n - 1) @@ -47,32 +55,37 @@ s <- stream conn sendStream s cmd shutdownStream s - consume aux s + consume misc s clientH3 :: Int -> Cli -clientH3 n0 aux paths conn = do - s2 <- unidirectionalStream conn - s6 <- unidirectionalStream conn - s10 <- unidirectionalStream conn - -- 0: control, 4 settings - sendStream s2 (BS.pack [0, 4, 8, 1, 80, 0, 6, 128, 0, 128, 0]) - -- 2: from encoder to decoder - sendStream s6 (BS.pack [2]) - -- 3: from decoder to encoder - sendStream s10 (BS.pack [3]) - foldr1 concurrently_ $ map (clientH3' n0 aux conn) paths +clientH3 n0 misc paths conn = do + done <- readIORef $ miscH3NegoDone misc + unless done $ do + s2 <- unidirectionalStream conn + s6 <- unidirectionalStream conn + s10 <- unidirectionalStream conn + -- 0: control, 4 settings + sendStream s2 (BS.pack [0, 4, 8, 1, 80, 0, 6, 128, 0, 128, 0]) + -- 2: from encoder to decoder + sendStream s6 (BS.pack [2]) + -- 3: from decoder to encoder + sendStream s10 (BS.pack [3]) + writeIORef (miscH3NegoDone misc) True + if miscInteractive misc then console paths go conn else go + where + go = foldr1 concurrently_ $ map (clientH3' n0 misc conn) paths -clientH3' :: Int -> Aux -> Connection -> ByteString -> IO () -clientH3' n0 aux@Aux{..} conn path = do - hdrblk <- taglen 1 <$> qpackClient path auxAuthority +clientH3' :: Int -> Misc -> Connection -> ByteString -> IO () +clientH3' n0 misc@Misc{..} conn path = do + hdrblk <- taglen 1 <$> qpackClient path miscAuthority loop n0 hdrblk where - loop 0 _ = auxDebug "Connection finished" + loop 0 _ = miscDebug "Connection finished" loop 1 hdrblk = do - auxDebug "GET" + miscDebug "GET" get hdrblk loop n hdrblk = do - auxDebug "GET" + miscDebug "GET" get hdrblk threadDelay 100000 loop (n - 1) hdrblk @@ -80,22 +93,22 @@ s <- stream conn sendStream s hdrblk shutdownStream s - consume aux s + consume misc s -consume :: Aux -> Stream -> IO () -consume aux@Aux{..} s = do +consume :: Misc -> Stream -> IO () +consume misc@Misc{..} s = do bs <- recvStream s 1024 if bs == "" then do - auxDebug "Fin received" + miscDebug "Fin received" closeStream s else do - auxShow bs - auxDebug $ show (BS.length bs) ++ " bytes received" - consume aux s + miscShow bs + miscDebug $ show (BS.length bs) ++ " bytes received" + consume misc s clientPF :: Word64 -> Cli -clientPF n Aux{..} _paths conn = do +clientPF n Misc{..} _paths conn = do cmd <- withWriteBuffer 8 $ \wbuf -> write64 wbuf n s <- stream conn sendStream s cmd @@ -106,8 +119,53 @@ bs <- recvStream s 1024 if bs == "" then do - auxDebug "Connection finished" + miscDebug "Connection finished" closeStream s else do - auxShow bs + miscShow bs loop s + +console :: [ByteString] -> IO () -> Connection -> IO () +console paths client conn = do + waitEstablished conn + putStrLn "q -- quit" + putStrLn "g -- get" + putStrLn "p -- ping" + putStrLn "M -- change server CID" + putStrLn "N -- change client CID" + putStrLn "B -- NAT rebinding" + putStrLn "A -- address mobility" + mvar <- newEmptyMVar + loop mvar `E.catch` \(E.SomeException _) -> return () + where + loop mvar = do + hSetBuffering stdout NoBuffering + putStr "> " + hSetBuffering stdout LineBuffering + l <- getLine + case l of + "q" -> putStrLn "bye" + "g" -> do + mapM_ (\p -> putStrLn $ "GET " ++ C8.unpack p) paths + _ <- client >> putMVar mvar () + takeMVar mvar + loop mvar + "p" -> do + putStrLn "Ping" + QUIC.sendFrames conn QUIC.RTT1Level [QUIC.Ping] + loop mvar + "M" -> do + QUIC.controlConnection conn QUIC.ChangeServerCID >>= print + loop mvar + "N" -> do + QUIC.controlConnection conn QUIC.ChangeClientCID >>= print + loop mvar + "B" -> do + QUIC.controlConnection conn QUIC.NATRebinding >>= print + loop mvar + "A" -> do + QUIC.controlConnection conn QUIC.ActiveMigration >>= print + loop mvar + _ -> do + putStrLn "No such command" + loop mvar diff --git a/util/quic-client.hs b/util/quic-client.hs --- a/util/quic-client.hs +++ b/util/quic-client.hs @@ -9,6 +9,7 @@ import Data.ByteString (ByteString) import qualified Data.ByteString as BS import qualified Data.ByteString.Char8 as C8 +import Data.IORef import Data.UnixTime import Data.Word import Foreign.C.Types @@ -16,7 +17,6 @@ import System.Console.GetOpt import System.Environment import System.Exit -import System.IO import qualified System.Timeout as T import Text.Printf @@ -248,22 +248,25 @@ showContent | optShow = C8.putStrLn | otherwise = \_ -> return () - aux = - Aux - { auxAuthority = host - , auxDebug = debug - , auxShow = showContent - , auxCheckClose = do + h3NegoDone <- newIORef False + let misc = + Misc + { miscAuthority = host + , miscDebug = debug + , miscShow = showContent + , miscCheckClose = do mx <- T.timeout 1000000 $ takeMVar cmvar case mx of Nothing -> return False _ -> return True + , miscH3NegoDone = h3NegoDone + , miscInteractive = optInteractive } - runClient cc opts aux paths + runClient cc opts misc paths -runClient :: ClientConfig -> Options -> Aux -> [ByteString] -> IO () -runClient cc opts@Options{..} aux@Aux{..} paths = do - auxDebug "------------------------" +runClient :: ClientConfig -> Options -> Misc -> [ByteString] -> IO () +runClient cc opts@Options{..} misc@Misc{..} paths = do + miscDebug "------------------------" (info1, info2, res, mig, client') <- run cc $ \conn -> do i1 <- getConnectionInfo conn let client = case alpn i1 of @@ -275,14 +278,10 @@ Nothing -> return False Just mtyp -> do x <- controlConnection conn mtyp - auxDebug $ "Migration by " ++ show mtyp + miscDebug $ "Migration by " ++ show mtyp return x t1 <- getUnixTime - if optInteractive - then do - console aux paths client conn - else do - client aux paths conn + client misc paths conn stats <- getConnectionStats conn print stats t2 <- getUnixTime @@ -300,7 +299,7 @@ | optResumption -> do if isResumptionPossible res then do - info3 <- runClient2 cc opts aux paths res client' + info3 <- runClient2 cc opts misc paths res client' if handshakeMode info3 == PreSharedKey then do putStrLn "Result: (R) TLS resumption ... OK" @@ -314,7 +313,7 @@ | opt0RTT -> do if is0RTTPossible res then do - info3 <- runClient2 cc opts aux paths res client' + info3 <- runClient2 cc opts misc paths res client' if handshakeMode info3 == RTT0 then do putStrLn "Result: (Z) 0-RTT ... OK" @@ -367,7 +366,7 @@ Nothing -> do putStrLn "Result: (H) handshake ... OK" putStrLn "Result: (D) stream data ... OK" - closeCompleted <- auxCheckClose + closeCompleted <- miscCheckClose when closeCompleted $ putStrLn "Result: (C) close completed ... OK" case alpn info1 of Nothing -> return () @@ -379,17 +378,17 @@ runClient2 :: ClientConfig -> Options - -> Aux + -> Misc -> [ByteString] -> ResumptionInfo -> Cli -> IO ConnectionInfo -runClient2 cc Options{..} aux@Aux{..} paths res client = do +runClient2 cc Options{..} misc@Misc{..} paths res client = do threadDelay 100000 - auxDebug "<<<< next connection >>>>" - auxDebug "------------------------" + miscDebug "<<<< next connection >>>>" + miscDebug "------------------------" run cc' $ \conn -> do - void $ client aux paths conn + void $ client misc paths conn getConnectionInfo conn where cc' = @@ -417,46 +416,3 @@ / fromIntegral millisecs / 1024 / 1024 - -console :: Aux -> [ByteString] -> Cli -> Connection -> IO () -console aux paths client conn = do - waitEstablished conn - putStrLn "q -- quit" - putStrLn "g -- get" - putStrLn "p -- ping" - putStrLn "M -- change server CID" - putStrLn "N -- change client CID" - putStrLn "B -- NAT rebinding" - putStrLn "A -- address mobility" - loop - where - loop = do - hSetBuffering stdout NoBuffering - putStr "> " - hSetBuffering stdout LineBuffering - l <- getLine - case l of - "q" -> putStrLn "bye" - "g" -> do - mapM_ (\p -> putStrLn $ "GET " ++ C8.unpack p) paths - _ <- forkIO $ client aux paths conn - loop - "p" -> do - putStrLn "Ping" - sendFrames conn RTT1Level [Ping] - loop - "M" -> do - controlConnection conn ChangeServerCID >>= print - loop - "N" -> do - controlConnection conn ChangeClientCID >>= print - loop - "B" -> do - controlConnection conn NATRebinding >>= print - loop - "A" -> do - controlConnection conn ActiveMigration >>= print - loop - _ -> do - putStrLn "No such command" - loop