quic 0.2.20 → 0.2.21

raw patch · 14 files changed

+180/−130 lines, 14 files

Files

• ChangeLog.md +4/−0
• Network/QUIC/Client/Run.hs +8/−7
• Network/QUIC/Connection/Types.hs +7/−0
• Network/QUIC/IO.hs +3/−2
• Network/QUIC/Receiver.hs +3/−1
• Network/QUIC/Recovery.hs +1/−0
• Network/QUIC/Recovery/Interface.hs +4/−0
• Network/QUIC/Sender.hs +16/−1
• Network/QUIC/Server/Reader.hs +1/−1
• Network/QUIC/Server/Run.hs +6/−6
• Network/QUIC/Stream/Reass.hs +2/−1
• quic.cabal +1/−1
• util/ClientX.hs +100/−42
• util/quic-client.hs +24/−68

@@ -1,5 +1,9 @@ # ChangeLog +## 0.2.21++* Terminating threads properly with event-poll model.+ ## 0.2.20  * Fix recvStream overflow on pending data.

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

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

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

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

@@ -1,6 +1,7 @@ module Network.QUIC.Recovery (     -- Interface     checkWindowOpenSTM,+    isEmptyPingSTM,     takePingSTM,     speedup,     resender,

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

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

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

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

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

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

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

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