Release QUIC read flow control as the WT app reads

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<quic::QuicSocket> quicSocket,
           WtHandlerPtr{wtHandler, WtHandlerDeleter{.owning = false}}) {
 }
 
-QuicWtSession::QuicWtSession(std::shared_ptr<quic::QuicSocket> quicSocket,
-                             WtHandlerPtr wtHandler)
-    : QuicWtSessionBase(
-          std::move(quicSocket), std::move(wtHandler), createQuicConfig()) {
+QuicWtSession::QuicWtSession(
+    const std::shared_ptr<quic::QuicSocket>& quicSocket, WtHandlerPtr wtHandler)
+    : QuicWtSessionBase(quicSocket,
+                        std::move(wtHandler),
+                        createQuicConfig(quicSocket->getTransportSettings())) {
+  manualRead_ = true;
   quicSocket_->setConnectionCallback(&connCb_);
   quicSocket_->setDatagramCallback(&dgramCb_);
 }

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<quic::HTTPPriorityQueue> 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> quicSocket, WtHandlerPtr wtHandler);
+  QuicWtSession(const std::shared_ptr<QuicSocket>& quicSocket,
+                WtHandlerPtr wtHandler);
 
   struct QuicConnectionCallback : public QuicSocket::ConnectionCallback {
     QuicWtSession& sess;

proxygen/lib/http/webtransport/WtStreamManager.cpp

@@ -939,7 +939,7 @@ folly::SemiFuture<StreamData> 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<folly::Unit, ReadHandle::ErrCode> 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) {

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;
 

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.

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