diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,19 @@ and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). +## 0.1.2.0 - 2026-09-03 + +- Servers and clients are datagram-enabled by default. +- Added server input queue to runServerStateful setup callback. + * Can be used to inject datagrams as server queries. +- startClientAsync no longer inherits the caller's exception mask, so stopping a client started inside a bracket cannot deadlock on the quic watchdog thread. + +- `Network.QUIC.Simple.Stream` is now meant for qualified import: `streamSerialise` and `streamCodec` became `Stream.serialise` and `Stream.codec`. +- Added `Stream.open` and `Stream.acceptLoop` for running multiple protocols over one connection, each stream announced by a header message. +- Added `Stream.sendMessage` and `Stream.recvMessage` for one-shot CBOR exchanges on bare streams, e.g. handshakes and stream headers. +- Added `Stream.serialiseFrom` and `Stream.codecFrom` to pick up a stream after a header, and `Stream.consume` for raw byte payloads. +- Stream workers now finish when the peer closes the stream instead of spinning on the end of stream. + ## 0.1.1.0 - 2025-12-02 - Stream wrappers now use async and will close their streams on exit. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -4,8 +4,15 @@ - `QUIC.runServer [("127.0.0.1", 14443)] \conn stream ->` -- start a generic server with random TLS credentials and auto-accept the first stream. - `QUIC.runClient "127.0.0.1" "14443" \conn stream ->` -- start a generic client and request an initial stream. -- `(writeQ, readQ) <- QUIC.streamCodec encode decodeIncremental stream` -- convert a stream to a pair of queues. -- `(writeQ, readQ) <- QUIC.streamSerialise stream` -- run a CBOR codec over a stream. +- `(writeQ, readQ) <- Stream.codec encode decodeIncremental stream` -- convert a stream to a pair of queues. +- `(writeQ, readQ) <- Stream.serialise stream` -- run a CBOR codec over a stream. + +Handshakes and multiple streams per connection, with `import Network.QUIC.Simple.Stream qualified as Stream`: + +- `Stream.sendMessage stream hello` / `(reply, leftovers) <- Stream.recvMessage stream` -- exchange one-shot CBOR messages on a bare stream. +- `stream <- Stream.open conn header` -- request a stream and announce its protocol with a header message. +- `Stream.acceptLoop conn \header leftovers stream ->` -- accept the peer's streams and dispatch them by header, each in its own thread. +- `Stream.serialiseFrom leftovers stream` / `Stream.consume leftovers stream sink` -- continue a stream after its header with CBOR messages or raw bytes. An extra-simple pair of wrappers for QUIC and dirty RPC: diff --git a/quic-simple.cabal b/quic-simple.cabal --- a/quic-simple.cabal +++ b/quic-simple.cabal @@ -1,11 +1,11 @@ cabal-version: 2.2 --- This file has been generated from package.yaml by hpack version 0.38.1. +-- This file has been generated from package.yaml by hpack version 0.39.6. -- -- see: https://github.com/sol/hpack name: quic-simple -version: 0.1.1.0 +version: 0.1.2.0 synopsis: Quick-start wrappers for QUIC description: A few layers over QUIC, to get the first bytes out faster. The top level is RPC-like, using Serialise as a codec. @@ -50,13 +50,13 @@ , bytestring , crypton , crypton-x509 - , hourglass , iproute - , memory , network - , quic >=0.2.7 + , quic >=0.3.2 + , ram , serialise , stm + , time-hourglass , tls default-language: GHC2021 diff --git a/src/Network/QUIC/Simple.hs b/src/Network/QUIC/Simple.hs --- a/src/Network/QUIC/Simple.hs +++ b/src/Network/QUIC/Simple.hs @@ -19,11 +19,10 @@ import Control.Concurrent.STM import Network.QUIC -import Network.QUIC.Simple.Stream import Codec.Serialise (Serialise) import Control.Concurrent (threadDelay) -import Control.Concurrent.Async (Async, async, cancel, link, link2) +import Control.Concurrent.Async (Async, asyncWithUnmask, cancel, link, link2) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) import Control.Exception (SomeException, handle, onException) import Control.Monad (forever) @@ -33,6 +32,7 @@ import Network.QUIC.Server (ServerConfig(..), defaultServerConfig) import Network.QUIC.Server qualified as Server import Network.QUIC.Simple.Credentials (genCredentials) +import Network.QUIC.Simple.Stream qualified as Stream import Network.Socket (HostName, PortNumber, ServiceName) {- $intro @@ -60,6 +60,9 @@ sc = defaultServerConfig { scCredentials , scAddresses +#if MIN_VERSION_quic(0,3,2) + , scMaxDatagramFrameSize = 0xFFFF -- advertise support +#endif } Server.run sc \conn -> do defaultStream <- acceptStream conn @@ -79,7 +82,7 @@ runServerSimple host port action = runServerStateful host port setup teardown handler where - setup _conn _wq = pure () + setup _conn _rq _wq = pure () teardown _conn _s = pure () handler s q = do r <- action q @@ -95,13 +98,13 @@ :: (Serialise q, Serialise r) => IP -> PortNumber - -> (Connection -> TBQueue r -> IO s) + -> (Connection -> TBQueue q -> TBQueue r -> IO s) -> (Connection -> s -> IO ()) -> (s -> q -> IO (s, Maybe r)) -> IO () runServerStateful host port setup teardown action = runServer [(host, port)] \conn stream0 -> do - (codec, (writeQ, readQ)) <- streamSerialise stream0 + (codec, (writeQ, readQ)) <- Stream.serialise stream0 link codec let loop !s = handle (\(_ :: SomeException) -> teardown conn s) do @@ -109,7 +112,7 @@ (s', reply_) <- action s query mapM_ (atomically . writeTBQueue writeQ) reply_ loop s' - setup conn writeQ >>= loop + setup conn readQ writeQ >>= loop {- | Run a client connecting to the provided host/port and auto-request a stream. @@ -132,6 +135,9 @@ , ccSockConnected = True , ccWatchDog = True #endif +#if MIN_VERSION_quic(0,3,2) + , ccMaxDatagramFrameSize = 1200 -- a conservative request as datagrams must not be fragmented +#endif } {- | Start a client wrapper that will wait for a connection. @@ -166,11 +172,11 @@ :: (Serialise q, Serialise r) => HostName -> ServiceName - -> IO (Async (), Connection, MessageQueues q r) + -> IO (Async (), Connection, Stream.MessageQueues q r) startClientAsync host port = do client <- newEmptyMVar - tid <- async $ runClient host port \conn stream0 -> do - queues <- streamSerialise stream0 + tid <- asyncWithUnmask \unmask -> unmask $ runClient host port \conn stream0 -> do + queues <- Stream.serialise stream0 putMVar client (conn, queues) forever (threadDelay maxBound) (conn, (codec, queues)) <- takeMVar client `onException` cancel tid @@ -180,3 +186,4 @@ , conn , queues ) + diff --git a/src/Network/QUIC/Simple/Stream.hs b/src/Network/QUIC/Simple/Stream.hs --- a/src/Network/QUIC/Simple/Stream.hs +++ b/src/Network/QUIC/Simple/Stream.hs @@ -1,20 +1,45 @@ module Network.QUIC.Simple.Stream - ( MessageQueues - , streamSerialise - , streamCodec + ( -- $intro + + -- * Message queues + MessageQueues + , serialise + , serialiseFrom + , codec + , codecFrom + -- * One-shot messages + , sendMessage + , recvMessage + -- * Raw bytes + , consume + -- * Multiple streams + , open + , acceptLoop ) where -import Codec.Serialise (Serialise, serialise, deserialiseIncremental) +import Codec.Serialise (Serialise, deserialiseIncremental) +import Codec.Serialise qualified as CBOR import Codec.Serialise qualified as IDecode (IDecode(..)) -import Control.Concurrent.Async (Async, async, race_) +import Control.Concurrent.Async (Async, async, cancel, link, poll, race_) import Control.Concurrent.STM import Control.Exception (finally, throwIO) +import Control.Monad (filterM, forever, unless) import Control.Monad.ST (stToIO) import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as BSL import Data.IORef +import Data.Maybe (isNothing) import Network.QUIC qualified as QUIC +{- $intro +This module is meant to be imported qualified: + +> import Network.QUIC.Simple.Stream qualified as Stream + +The connection wrappers in "Network.QUIC.Simple" hand out the first stream. +The functions here run protocols over it, or open and accept more streams. +-} + {- | A pair of bounded queues wrapping a stream. -} type MessageQueues sendMsg recvMsg = (TBQueue sendMsg, TBQueue recvMsg) @@ -25,12 +50,23 @@ No extra framing is required since CBOR is self-delimiting. -} -streamSerialise +serialise + :: (Serialise sendMsg, Serialise recvMsg) + => QUIC.Stream + -> IO (Async (), MessageQueues sendMsg recvMsg) +serialise = serialiseFrom "" + +{- | Same as 'serialise', but starts decoding from the bytes already received. + +Use with the leftovers from 'recvMessage' when a stream starts with a header. +-} +serialiseFrom :: forall sendMsg recvMsg . (Serialise sendMsg, Serialise recvMsg) - => QUIC.Stream + => BS.ByteString + -> QUIC.Stream -> IO (Async (), MessageQueues sendMsg recvMsg) -streamSerialise stream = do +serialiseFrom leftovers stream = do initial <- stToIO $ deserialiseIncremental @recvMsg state <- newIORef initial let @@ -39,49 +75,146 @@ case decoder of IDecode.Fail _leftovers _offset err -> throwIO err -- crash writer (thus the stream, and the reader/writer etc) - IDecode.Done leftovers _consumed msg -> do + IDecode.Done leftovers' _consumed msg -> do stToIO deserialiseIncremental >>= writeIORef state -- restart decoder - pure (leftovers, Just msg) - IDecode.Partial consume -> do + pure (leftovers', Just msg) + IDecode.Partial consume_ -> do -- want more data (initial state?) - stToIO (consume $ Just chunk) >>= writeIORef state -- step decoder + stToIO (consume_ $ Just chunk) >>= writeIORef state -- step decoder if starting then -- re-check if done decode False "" else -- suspend and wait for next chunk pure ("", Nothing) - streamCodec serialise (decode True) stream + codecFrom CBOR.serialise (decode True) leftovers stream {- | Wrap the stream with a codec to provide a TBQueue interface to it. The decoder loop is stateless. But it runs in IO so you can use external state and terminate the stream by erroring out. + +The worker finishes when the peer closes the stream, or when either the codec or the stream fails. +Cancel it to close the stream from this side. +Messages still queued for sending at that moment are dropped. -} -streamCodec +codec :: (sendMsg -> BSL.ByteString) -- ^ Encoder for outgoing messages -> (BS.ByteString -> IO (BS.ByteString, Maybe recvMsg)) -- ^ Decoder for incomming chunks -> QUIC.Stream -> IO (Async (), MessageQueues sendMsg recvMsg) -streamCodec encode decode stream = do +codec encode decode = codecFrom encode decode "" + +{- | Same as 'codec', but starts decoding from the bytes already received. +-} +codecFrom + :: (sendMsg -> BSL.ByteString) -- ^ Encoder for outgoing messages + -> (BS.ByteString -> IO (BS.ByteString, Maybe recvMsg)) -- ^ Decoder for incomming chunks + -> BS.ByteString -- ^ Bytes received before the codec took over + -> QUIC.Stream + -> IO (Async (), MessageQueues sendMsg recvMsg) +codecFrom encode decode leftovers stream = do readQ <- newTBQueueIO 1024 writeQ <- newTBQueueIO 1024 worker <- async $ - race_ (reader "" readQ) (writer writeQ) `finally` QUIC.closeStream stream + race_ (reader leftovers readQ) (writer writeQ) `finally` QUIC.closeStream stream pure (worker, (writeQ, readQ)) where - reader leftovers readQ = do + reader buffered readQ = do chunk <- - if BS.null leftovers then + if BS.null buffered then QUIC.recvStream stream 4096 else - pure leftovers - (leftovers', message_) <- decode chunk - mapM_ (atomically . writeTBQueue readQ) message_ - reader leftovers' readQ + pure buffered + unless (BS.null chunk) do + (buffered', message_) <- decode chunk + mapM_ (atomically . writeTBQueue readQ) message_ + reader buffered' readQ writer writeQ = do message <- atomically $ readTBQueue writeQ let chunks = BSL.toChunks $ encode message QUIC.sendStreamMany stream chunks writer writeQ + +{- | Send a single CBOR message to a stream. +-} +sendMessage :: Serialise msg => QUIC.Stream -> msg -> IO () +sendMessage stream = QUIC.sendStreamMany stream . BSL.toChunks . CBOR.serialise + +{- | Receive a single CBOR message from a stream. + +Bytes received past the end of the message are returned along with it. +They belong to whatever comes next on the stream, +so feed them to 'serialiseFrom' or 'consume'. +In a lockstep exchange there will be none. + +Throws a decoding error if the stream ends or gets garbled before the message is complete. +-} +recvMessage :: forall msg. Serialise msg => QUIC.Stream -> IO (msg, BS.ByteString) +recvMessage stream = stToIO (deserialiseIncremental @msg) >>= step + where + step = \case + IDecode.Done leftovers _offset msg -> + pure (msg, leftovers) + IDecode.Fail _leftovers _offset err -> + throwIO err + IDecode.Partial consume_ -> do + chunk <- QUIC.recvStream stream 4096 + stToIO (consume_ $ if BS.null chunk then Nothing else Just chunk) >>= step + +{- | Feed the incoming bytes to a sink until the peer closes the stream. + +The bytes received before, e.g. the leftovers from 'recvMessage', go first. +-} +consume :: BS.ByteString -> QUIC.Stream -> (BS.ByteString -> IO ()) -> IO () +consume leftovers stream sink = do + unless (BS.null leftovers) $ sink leftovers + drain + where + drain = do + chunk <- QUIC.recvStream stream 65536 + unless (BS.null chunk) do + sink chunk + drain + +{- | Request a new bidirectional stream and announce its purpose with a header message. + +The peer is expected to run 'acceptLoop' with a matching header type. +Anything sent right after the header will arrive as leftovers on the other side. +-} +open :: Serialise header => QUIC.Connection -> header -> IO QUIC.Stream +open conn header = do + new <- QUIC.stream conn + sendMessage new header + pure new + +{- | Accept streams opened by the peer and dispatch them by their header message. + +Each stream gets a thread of its own that closes the stream when the handler returns. +The handler receives the bytes that arrived along with the header, +to be passed along to 'serialiseFrom' or 'consume'. + +A crashing handler takes down the accept loop, and thus the connection. +Leaving the loop, by cancelling it or by connection failure, cancels the handlers still running. +-} +acceptLoop + :: Serialise header + => QUIC.Connection + -> (header -> BS.ByteString -> QUIC.Stream -> IO ()) + -> IO () +acceptLoop conn handler = do + running <- newTVarIO [] + let + cancelAll = readTVarIO running >>= mapM_ cancel + prune = readTVarIO running >>= filterM (fmap isNothing . poll) >>= atomically . writeTVar running + handleStream new = do + (header, leftovers) <- recvMessage new + handler header leftovers new + accept = forever do + new <- QUIC.acceptStream conn + worker <- async $ handleStream new `finally` QUIC.closeStream new + link worker + prune + atomically $ modifyTVar' running (worker :) + accept `finally` cancelAll diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -4,17 +4,17 @@ import Control.Concurrent.STM import Control.Monad -import Control.Concurrent.Async (async, cancel, link, race_, replicateConcurrently_) -import Control.Exception (bracket) +import Control.Concurrent.Async (async, cancel, concurrently_, link, race_, replicateConcurrently_, wait) +import Control.Exception (bracket, throwIO) import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as BSL -import Data.IORef (newIORef, atomicModifyIORef') +import Data.IORef (newIORef, atomicModifyIORef', readIORef) import Data.IntMap.Strict qualified as IntMap import Data.Text (Text) import Data.Text.Encoding (encodeUtf8, decodeUtf8) import GHC.Generics (Generic) import Network.QUIC.Simple qualified as QUIC -import Network.QUIC.Simple.Stream (MessageQueues, streamCodec, streamSerialise) +import Network.QUIC.Simple.Stream qualified as Stream import System.Timeout (timeout) main :: IO () @@ -27,6 +27,10 @@ race_ serverBox clientBox putStrLn "" + putStrLn "Datagram" + race_ serverDatagram clientDatagram + putStrLn "" + putStrLn "Serialise" race_ serverSerialise clientSerialise putStrLn "" @@ -39,6 +43,10 @@ race_ serverStateful clientAsync putStrLn "" + putStrLn "Multistream" + race_ serverMulti clientMulti + putStrLn "" + -- * Raw clientRaw :: IO () @@ -78,10 +86,10 @@ This allows sending empty messages, but may break if the Text has a NUL in there. What kind of "text" is that anyway?! -} -cstringCodec :: QUIC.Stream -> IO (MessageQueues Text Text) +cstringCodec :: QUIC.Stream -> IO (Stream.MessageQueues Text Text) cstringCodec stream = do previous <- newIORef BSL.empty -- gotta store partial messages somewhere - snd <$> streamCodec encode (decode previous) stream + snd <$> Stream.codec encode (decode previous) stream where -- add framing encode msg = BSL.fromChunks [encodeUtf8 msg, "\NUL"] @@ -133,6 +141,47 @@ putStrLn $ "Server got query: " <> show query atomically $ writeTBQueue writeQ $ "got yer bytes: " <> query +-- * Unreliable datagrams + +clientDatagram :: IO () +clientDatagram = QUIC.runClient "127.0.0.1" "14443" \conn stream -> do + -- the initial stream comes pre-requested and ready to go + putStrLn "Client connected:" + QUIC.getConnectionInfo conn >>= print + + -- streams have no framing on their own + -- but that's fine, for now... + QUIC.sendStream stream "hi there" + QUIC.sendDatagram conn "yo" + yo <- QUIC.recvDatagram conn + putStrLn $ "Client got datagram: " <> show yo + reply <- QUIC.recvStream stream 4096 + putStrLn $ "Client got reply: " <> show reply + + QUIC.closeStream stream + putStrLn "Client quits" + +serverDatagram :: IO () +serverDatagram = QUIC.runServer [("127.0.0.1", 14443)] \conn stream -> do + -- the initial stream comes pre-accepted and ready to go + putStrLn "Server accepted connection:" + QUIC.getConnectionInfo conn >>= print + + -- wait until *something* arrives and take it all in + query <- QUIC.recvStream stream 4096 + putStrLn $ "Server got query: " <> show query + + yo <- QUIC.recvDatagram conn + putStrLn $ "Server got datagram: " <> show yo + QUIC.sendDatagram conn yo + + QUIC.sendStream stream $ "got yer bytes: " <> query + + -- the final bytes will be "", signalling the connection getting closed + finalBytes <- QUIC.recvStream stream 4096 + putStrLn $ "Server quits after " <> show finalBytes + + -- * Serialised messages data ClientMessage @@ -159,7 +208,7 @@ clientSerialise :: IO () clientSerialise = do QUIC.runClient "127.0.0.1" "14443" \_conn stream -> do - (writeQ, readQ) <- snd <$> streamSerialise stream + (writeQ, readQ) <- snd <$> Stream.serialise stream replicateM_ 5 do -- send messages one by one atomically $ writeTBQueue writeQ Hello @@ -174,7 +223,7 @@ serverSerialise :: IO () serverSerialise = QUIC.runServer [("127.0.0.1", 14443)] \_conn stream -> do putStrLn "Server accepted connection:" - (writeQ, readQ) <- snd <$> streamSerialise stream + (writeQ, readQ) <- snd <$> Stream.serialise stream let -- simple state-passing loop loop counter = do @@ -308,7 +357,7 @@ QUIC.runServerStateful "127.0.0.1" 14443 (setup conns connIds) (teardown conns) handler where -- every connection has a local state too - setup conns counter _conn writeQ = do + setup conns counter _conn _readQ writeQ = do -- generate an explicit key connId <- atomicModifyIORef' counter \old -> (old + 1, old) -- don't store ThreadIDs directly! @@ -338,3 +387,117 @@ _ -> -- ignore casted Hellos and any form of Bye pure (connState, Nothing) + +-- * Handshake and multiple streams + +{- | The first stream carries only the handshake. + +Nothing else can be said before the session is established +and nothing about the handshake can be said after. +-} +data ClientHello = ClientHello { version :: Int, token :: Text } + deriving (Show, Generic, QUIC.Serialise) + +data ServerHello = Welcome Int | Rejected Text + deriving (Show, Generic, QUIC.Serialise) + +-- | Proof of a completed handshake, carrying whatever it produced. +newtype Session = Session { sessionId :: Int } + +{- | Every stream opened afterwards starts with a header naming its protocol. + +The rest of the stream is whatever that protocol says: CBOR messages, raw bytes, anything. +-} +data StreamHeader + = Rpc + | Upload Text Int + deriving (Show, Generic, QUIC.Serialise) + +data Received = Received Int + deriving (Show, Generic, QUIC.Serialise) + +clientMulti :: IO () +clientMulti = QUIC.runClient "127.0.0.1" "14443" \conn hello -> do + Stream.sendMessage hello ClientHello{version = 1, token = "letmein"} + -- the handshake is lockstep, so there are no leftovers to worry about + (reply, _) <- Stream.recvMessage hello + session <- case reply of + Welcome sid -> pure (Session sid) + Rejected why -> throwIO (userError (show why)) + QUIC.closeStream hello + putStrLn $ "Client got session " <> show (sessionId session) + + -- the streams run their protocols concurrently over one connection + concurrently_ (rpc conn) (upload conn) + putStrLn "Client quits" + where + rpc conn = do + stream <- Stream.open conn Rpc + -- the RPC stream is CBOR messages after the header + (codec, (writeQ, readQ)) <- Stream.serialise stream + link codec + replicateM_ 3 do + atomically $ writeTBQueue writeQ Hello + Ok n <- atomically $ readTBQueue readQ + putStrLn $ "Client got RPC reply " <> show n + -- the replies are in, so the codec has nothing left to send + cancel codec + + upload conn = do + let + pattern = BS.pack [0 .. 255] + block = BS.concat (replicate 256 pattern) + blocks = replicate 64 block + size = sum (map BS.length blocks) + -- the upload stream is raw bytes after the header, no framing needed + stream <- Stream.open conn (Upload "blob" size) + QUIC.sendStreamMany stream blocks + -- finish sending, but keep the stream open for the reply + QUIC.shutdownStream stream + (Received got, _) <- Stream.recvMessage stream + QUIC.closeStream stream + unless (got == size) $ + throwIO (userError $ "Upload size mismatch: " <> show (got, size)) + putStrLn $ "Client uploaded " <> show got <> " bytes" + +serverMulti :: IO () +serverMulti = QUIC.runServer [("127.0.0.1", 14443)] \conn hello -> do + (ClientHello{version, token}, _) <- Stream.recvMessage hello + session <- + if version == 1 && token == "letmein" then do + Stream.sendMessage hello (Welcome 42) + pure (Session 42) + else do + Stream.sendMessage hello (Rejected "bad credentials") + throwIO (userError "handshake failed") + QUIC.closeStream hello + putStrLn $ "Server established session " <> show (sessionId session) + + -- from now on the peer announces what each stream is for + Stream.acceptLoop conn \header leftovers stream -> case header of + Rpc -> + rpc session leftovers stream + Upload name size -> + upload name size leftovers stream + where + rpc Session{sessionId} leftovers stream = do + -- the first message may have arrived together with the header + (codec, (writeQ, readQ)) <- Stream.serialiseFrom leftovers stream + counter <- newIORef sessionId + let + serve = forever do + Hello <- atomically $ readTBQueue readQ + n <- atomicModifyIORef' counter \old -> (old + 1, old) + atomically $ writeTBQueue writeQ (Ok n) + -- the codec finishes when the client closes the stream + race_ (wait codec) serve + putStrLn "Server RPC stream closed" + + upload name size leftovers stream = do + received <- newIORef 0 + -- the bytes after the header are the payload + Stream.consume leftovers stream \chunk -> + atomicModifyIORef' received \old -> (old + BS.length chunk, ()) + total <- readIORef received + putStrLn $ "Server received " <> show name <> ": " <> show total <> " of " <> show size + Stream.sendMessage stream (Received total)