From e9dab4be71198ec7b6619856c5091311d0d056f5 Mon Sep 17 00:00:00 2001 From: Alan Frindell Date: Sat, 31 Oct 2026 09:07:35 -0700 Subject: [PATCH] Release QUIC read flow control as the WT app reads Summary: QuicWtSession drained QUIC streams eagerly into its own buffer, so QUIC released flow control credit for data the app had not read yet. The WT buffer size was also fixed at 64KB, unrelated to the QUIC window. Now each read side stream uses mvfst manual read mode, the WT buffer matches the QUIC stream window from the transport settings, and credit is released as the app consumes data. When the app stops reading, any buffered unread data counts as consumed, so its connection credit is not lost if QUIC already removed the stream. Differential Revision: D128780069 --- .../lib/http/webtransport/QuicWtSession.cpp | 42 +++++++++++++------ .../lib/http/webtransport/QuicWtSession.h | 9 +++- .../lib/http/webtransport/WtStreamManager.cpp | 12 +++++- .../lib/http/webtransport/WtStreamManager.h | 4 +- .../webtransport/test/QuicWtSessionTest.cpp | 36 +++++++++++++++- .../webtransport/test/WtStreamManagerTest.cpp | 2 +- 6 files changed, 85 insertions(+), 20 deletions(-) diff --git a/proxygen/lib/http/webtransport/QuicWtSession.cpp b/proxygen/lib/http/webtransport/QuicWtSession.cpp index 077a987c0b..b791dbaf5c 100644 --- a/proxygen/lib/http/webtransport/QuicWtSession.cpp +++ b/proxygen/lib/http/webtransport/QuicWtSession.cpp @@ -15,7 +15,6 @@ using namespace proxygen::detail; using FCState = WebTransport::FCState; namespace { -static constexpr uint64_t kMaxWtIngressBuf = 65'535; static constexpr uint64_t kDefaultWtWrite = 65'535; using WtConfig = WtStreamManager::WtConfig; @@ -26,14 +25,17 @@ WtConfig& overwritePeerMaxDataWtConfig(WtConfig& config) { return config; } -WtConfig createQuicConfig() { +WtConfig createQuicConfig(const quic::TransportSettings& settings) { WtConfig config; config.selfMaxStreamsBidi = kMaxVarint; config.selfMaxStreamsUni = kMaxVarint; config.selfMaxConnData = kMaxVarint; - config.selfMaxStreamDataBidiLocal = kMaxWtIngressBuf; - config.selfMaxStreamDataBidiRemote = kMaxWtIngressBuf; - config.selfMaxStreamDataUni = kMaxWtIngressBuf; + config.selfMaxStreamDataBidiLocal = + settings.advertisedInitialBidiLocalStreamFlowControlWindow; + config.selfMaxStreamDataBidiRemote = + settings.advertisedInitialBidiRemoteStreamFlowControlWindow; + config.selfMaxStreamDataUni = + settings.advertisedInitialUniStreamFlowControlWindow; config.peerMaxStreamsBidi = kMaxVarint; config.peerMaxStreamsUni = kMaxVarint; config.peerMaxConnData = kMaxVarint; @@ -187,11 +189,19 @@ auto QuicWtSessionBase::createWtEgressHandle(StreamId id) noexcept quicSocket_->setStopSendingCallback(id, &stopSendingCb_); if (bidi) { sm_.setReadCb(*res.readHandle, &smCb_); - quicSocket_->setReadCallback(id, &readCb_); + installReadCallback(id); } return res; } +void QuicWtSessionBase::installReadCallback(StreamId id) noexcept { + quicSocket_->setReadCallback(id, &readCb_); + if (manualRead_) { + auto res = quicSocket_->setStreamManualRead(id); + XLOG_IF(ERR, res.hasError()) << __func__ << "; err id=" << id; + } +} + bool QuicWtSessionBase::hasEgressUniCredit() const noexcept { return quicSocket_->getNumOpenableUnidirectionalStreams() && sm_.canCreateUni(); @@ -330,7 +340,13 @@ void QuicWtSessionBase::QuicStopSendingCallback::onStopSending( // -- StreamManagerCallback overrides -- void QuicWtSessionBase::StreamManagerCallback::readReady( - WtStreamManager::WtReadHandle& rh) noexcept { + WtStreamManager::WtReadHandle& rh, uint64_t bytesRead) noexcept { + if (sess.manualRead_) { + // fails harmlessly after stopSending, which already released everything + auto res = + sess.quicSocket_->releaseManualReadFlowControl(rh.getID(), bytesRead); + XLOG_IF(DBG4, res.hasError()) << __func__ << "; err id=" << rh.getID(); + } sess.maybeResumeIngress(rh); } @@ -434,7 +450,7 @@ bool QuicWtSessionBase::acquireIngressStream(uint64_t id) noexcept { XLOG(DBG6) << __func__ << "; id=" << id << "; ok=" << success; if (success) { sm_.setReadCb(*handle.readHandle, &smCb_); - quicSocket_->setReadCallback(id, &readCb_); + installReadCallback(id); if (bidi) { quicSocket_->setStopSendingCallback(id, &stopSendingCb_); grantMaxData(*quicSocket_, sm_, *handle.writeHandle); @@ -467,10 +483,12 @@ QuicWtSession::QuicWtSession(std::shared_ptr quicSocket, WtHandlerPtr{wtHandler, WtHandlerDeleter{.owning = false}}) { } -QuicWtSession::QuicWtSession(std::shared_ptr quicSocket, - WtHandlerPtr wtHandler) - : QuicWtSessionBase( - std::move(quicSocket), std::move(wtHandler), createQuicConfig()) { +QuicWtSession::QuicWtSession( + const std::shared_ptr& quicSocket, WtHandlerPtr wtHandler) + : QuicWtSessionBase(quicSocket, + std::move(wtHandler), + createQuicConfig(quicSocket->getTransportSettings())) { + manualRead_ = true; quicSocket_->setConnectionCallback(&connCb_); quicSocket_->setDatagramCallback(&dgramCb_); } diff --git a/proxygen/lib/http/webtransport/QuicWtSession.h b/proxygen/lib/http/webtransport/QuicWtSession.h index df31203c0e..952f8fc482 100644 --- a/proxygen/lib/http/webtransport/QuicWtSession.h +++ b/proxygen/lib/http/webtransport/QuicWtSession.h @@ -110,7 +110,8 @@ class QuicWtSessionBase QuicWtSessionBase& sess; explicit StreamManagerCallback(QuicWtSessionBase& session) : sess(session) { } - void readReady(detail::WtStreamManager::WtReadHandle& rh) noexcept override; + void readReady(detail::WtStreamManager::WtReadHandle& rh, + uint64_t bytesRead) noexcept override; void eventsAvailable() noexcept override; void onNewPeerStream(uint64_t streamId) noexcept override; @@ -140,9 +141,12 @@ class QuicWtSessionBase std::unique_ptr priorityQueue_; detail::WtStreamManager sm_; H3ConnectStreamCallback* observer_{nullptr}; + // quic releases receive credit only as the app reads from sm_ + bool manualRead_{false}; private: BidiStreamHandle createWtEgressHandle(StreamId id) noexcept; + void installReadCallback(StreamId id) noexcept; // -- StreamWriteCallback overrides -- void onStreamWriteReady(StreamId id, uint64_t maxToSend) noexcept override; @@ -174,7 +178,8 @@ class QuicWtSession final : public QuicWtSessionBase { } private: - QuicWtSession(std::shared_ptr quicSocket, WtHandlerPtr wtHandler); + QuicWtSession(const std::shared_ptr& quicSocket, + WtHandlerPtr wtHandler); struct QuicConnectionCallback : public QuicSocket::ConnectionCallback { QuicWtSession& sess; diff --git a/proxygen/lib/http/webtransport/WtStreamManager.cpp b/proxygen/lib/http/webtransport/WtStreamManager.cpp index d517e5b159..c27d18527f 100644 --- a/proxygen/lib/http/webtransport/WtStreamManager.cpp +++ b/proxygen/lib/http/webtransport/WtStreamManager.cpp @@ -939,7 +939,7 @@ folly::SemiFuture ReadHandle::readStreamData() { auto res = StreamData{ingress_.chain.pop(), ingress_.fin}; XLOG(DBG6) << __func__ << "; len=" << len << "; fin=" << res.fin; if (rcb_) { - rcb_->readReady(*this); + rcb_->readReady(*this, len); } finish(ingress_.fin); return folly::makeSemiFuture(std::move(res)); @@ -959,6 +959,14 @@ folly::Expected ReadHandle::stopSending( return folly::makeUnexpected(ReadHandle::ErrCode::SEND_ERROR); } state_ = ReadHandleState::HalfClosed; + // the app will not read the buffered data, so it counts as consumed + if (auto len = ingress_.chain.computeChainDataLength(); len > 0) { + ingress_.chain = folly::IOBuf(); + smAccessor_.maybeGrantFc(/*rh=*/nullptr, len); + if (rcb_) { + rcb_->readReady(*this, len); + } + } smAccessor_.stopSending(*this, error); if (auto p = resetPromise(); p.valid()) { // resolve pending read with ex p.setException(makeWtException(kInternalWtErr, "tx stop_sending")); @@ -984,7 +992,7 @@ Result ReadHandle::enqueue(StreamData&& data) noexcept { smAccessor_.maybeGrantFc(ingress_.fin ? nullptr : this, len); p.setValue(StreamData{ingress_.chain.pop(), ingress_.fin}); if (rcb_) { - rcb_->readReady(*this); + rcb_->readReady(*this, len); } finish(ingress_.fin); } else if (state_ == ReadHandleState::HalfClosed && ingress_.fin) { diff --git a/proxygen/lib/http/webtransport/WtStreamManager.h b/proxygen/lib/http/webtransport/WtStreamManager.h index 01da045242..89cc013047 100644 --- a/proxygen/lib/http/webtransport/WtStreamManager.h +++ b/proxygen/lib/http/webtransport/WtStreamManager.h @@ -207,8 +207,8 @@ struct WtStreamManager { */ struct ReadCallback { virtual ~ReadCallback() noexcept = default; - // invoked when the read handle has available buffer space - virtual void readReady(WtReadHandle&) noexcept = 0; + // invoked when the app consumed bytesRead, freeing buffer space + virtual void readReady(WtReadHandle&, uint64_t bytesRead) noexcept = 0; }; void setReadCb(WtReadHandle& rh, ReadCallback* rcb) noexcept; diff --git a/proxygen/lib/http/webtransport/test/QuicWtSessionTest.cpp b/proxygen/lib/http/webtransport/test/QuicWtSessionTest.cpp index 3a0a09dc26..f4925d35aa 100644 --- a/proxygen/lib/http/webtransport/test/QuicWtSessionTest.cpp +++ b/proxygen/lib/http/webtransport/test/QuicWtSessionTest.cpp @@ -21,7 +21,7 @@ using quic::MockQuicSocketDriver; namespace { constexpr uint32_t WT_ERROR_1 = 19; constexpr uint32_t WT_ERROR_2 = 77; -constexpr uint64_t kMaxWtIngressBuf = 65'535; +constexpr uint64_t kMaxWtIngressBuf = quic::kDefaultStreamFlowControlWindow; class MockDeliveryCallback : public WebTransport::ByteEventCallback { public: @@ -953,6 +953,40 @@ TEST_F(QuicWtSessionTest, IngressBackpressure) { MockQuicSocketDriver::PAUSED); } +TEST_F(QuicWtSessionTest, ReleaseFlowControlOnAppRead) { + // peer opens a bidi stream; quic credit is released only once the app reads + auto sock = socketDriver_.getSocket(); + EXPECT_CALL(*sock, setStreamManualRead(0)); + WebTransport::StreamReadHandle* readHandle = nullptr; + EXPECT_CALL(*handler_, onNewBidiStream(_)) + .WillOnce([&](WebTransport::BidiStreamHandle handle) { + readHandle = handle.readHandle; + }); + EXPECT_CALL(*sock, releaseManualReadFlowControl(_, _)).Times(0); + socketDriver_.addReadEvent(0, folly::IOBuf::copyBuffer("hello"), false); + eventBase_.loopOnce(); + ASSERT_NE(readHandle, nullptr); + + EXPECT_CALL(*sock, releaseManualReadFlowControl(0, 5)); + auto fut = readHandle->readStreamData(); + EXPECT_TRUE(fut.isReady()); +} + +TEST_F(QuicWtSessionTest, ReleaseFlowControlOnStopSendingUnread) { + // peer uni stream data+fin is buffered; app stops sending without reading + auto sock = socketDriver_.getSocket(); + WebTransport::StreamReadHandle* readHandle = nullptr; + EXPECT_CALL(*handler_, onNewUniStream(_)) + .WillOnce( + [&](WebTransport::StreamReadHandle* handle) { readHandle = handle; }); + socketDriver_.addReadEvent(2, folly::IOBuf::copyBuffer("hello"), true); + eventBase_.loopOnce(); + ASSERT_NE(readHandle, nullptr); + + EXPECT_CALL(*sock, releaseManualReadFlowControl(2, 5)); + readHandle->stopSending(WT_ERROR_1); +} + /* * H3WtSession shares most code w/ QuicWtSessionBase, so we test just the * overridden methods and validate the expected side-effects (e.g. diff --git a/proxygen/lib/http/webtransport/test/WtStreamManagerTest.cpp b/proxygen/lib/http/webtransport/test/WtStreamManagerTest.cpp index 58633e762c..cd80ba1624 100644 --- a/proxygen/lib/http/webtransport/test/WtStreamManagerTest.cpp +++ b/proxygen/lib/http/webtransport/test/WtStreamManagerTest.cpp @@ -1167,7 +1167,7 @@ using WtWriteHandle = WebTransport::StreamWriteHandle; struct WtReadCallback : WtStreamManager::ReadCallback { ~WtReadCallback() override = default; - void readReady(WtReadHandle&) noexcept override { + void readReady(WtReadHandle&, uint64_t) noexcept override { signalled = true; }