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; }