Propagate WT stream priority to the QUIC socket

Summary: WebTransport `setPriority` never made it to quic: the stream-id version was a stub that did nothing, and the write handle version only reordered our own egress queue. So the QUIC stream sat at default priority and was never ranked against the datagram flows, which do set priority on the socket. Now the stream-id version delegates to the write handle, and the write handle tells the transport, which is where `QuicWtSessionBase` calls `QuicSocket::setStreamPriority`. The capsule transports need nothing -- their queue is the schedule. `setPriorityQueue` is still a stub, separate change.

Reviewed By: sharmafb

Differential Revision: D115277068

fbshipit-source-id: e01c28b58c6d15278ec5e35a8b3b4d37b364e5ae

Author: afrind

third-party/proxygen/src/proxygen/lib/http/webtransport/QuicWtSession.cpp

@@ -314,6 +314,21 @@ void QuicWtSessionBase::StreamManagerCallback::eventsAvailable() noexcept {
   }
 }
 
+// Mirror the priority into the QuicSocket write queue, so wt streams and wt
+// datagram flows are scheduled against each other in the connection rather than
+// the stream sitting at default priority.
+//
+// Best effort. The transport can legitimately no longer have the stream (eg. it
+// just reset it), and priority is only a scheduling hint -- anything that
+// actually depends on the stream, like the next write, will fail louder.
+void QuicWtSessionBase::StreamManagerCallback::onStreamPriority(
+    uint64_t streamId, quic::PriorityQueue::Priority priority) noexcept {
+  XCHECK(sess.quicSocket_);
+  auto res = sess.quicSocket_->setStreamPriority(streamId, priority);
+  XLOG_IF(DBG4, res.hasError())
+      << __func__ << "; id=" << streamId << "; err=" << toString(res.error());
+}
+
 void QuicWtSessionBase::StreamManagerCallback::onNewPeerStream(
     uint64_t /*streamId*/) noexcept {
 }

third-party/proxygen/src/proxygen/lib/http/webtransport/QuicWtSession.h

@@ -111,6 +111,9 @@ class QuicWtSessionBase
     }
     void readReady(detail::WtStreamManager::WtReadHandle& rh) noexcept override;
     void eventsAvailable() noexcept override;
+    void onStreamPriority(
+        uint64_t streamId,
+        quic::PriorityQueue::Priority priority) noexcept override;
     void onNewPeerStream(uint64_t streamId) noexcept override;
   } smCb_{*this};
 

third-party/proxygen/src/proxygen/lib/http/webtransport/WtStreamManager.cpp

@@ -93,6 +93,7 @@ struct WtStreamManager::Accessor {
                    uint32_t err,
                    uint64_t reliableSize) noexcept;
   void onStreamWritable(WriteHandle& wh) noexcept;
+  void onStreamPriority(WriteHandle& wh) noexcept;
   // invoked when write handle or read handle are done
   void done(WriteHandle& wh) noexcept;
   void done(ReadHandle& rh) noexcept;
@@ -292,6 +293,10 @@ void Accessor::onStreamWritable(WriteHandle& wh) noexcept {
   sm_.onStreamWritable(wh);
 }
 
+void Accessor::onStreamPriority(WriteHandle& wh) noexcept {
+  sm_.egressCb_.onStreamPriority(wh.getID(), wh.getPriority());
+}
+
 void Accessor::done(WriteHandle& wh) noexcept {
   XCHECK_EQ(wh.state_, WriteHandleState::Closed);
   if (wh.bidi_->rh.state_ == ReadHandleState::Closed) { // bidi done
@@ -1085,6 +1090,7 @@ folly::Expected<folly::Unit, WriteHandle::ErrCode> WriteHandle::setPriority(
   }
   StreamWriteHandle::setPriority(priority);
   smAccessor_.writableStreams().update(getID(), getPriority());
+  smAccessor_.onStreamPriority(*this);
   return folly::unit;
 }
 

third-party/proxygen/src/proxygen/lib/http/webtransport/WtStreamManager.h

@@ -92,6 +92,18 @@ struct WtStreamManager {
   struct EgressCallback {
     virtual ~EgressCallback() = default;
     virtual void eventsAvailable() noexcept = 0;
+    /**
+     * Invoked when the application reprioritizes an egress stream. The
+     * manager's own queue has already been updated; this is only for transports
+     * that also schedule wt streams somewhere else -- quic ranks them against
+     * wt datagram flows in the connection write queue. For the capsule
+     * transports the manager's queue is the schedule, so the default no-op is
+     * correct.
+     */
+    virtual void onStreamPriority(
+        uint64_t /*streamId*/,
+        quic::PriorityQueue::Priority /*priority*/) noexcept {
+    }
   };
   /**
    * IngressCallback::onNewPeerStream is invoked whenever a new peer stream has

third-party/proxygen/src/proxygen/lib/http/webtransport/WtUtils.cpp

@@ -475,7 +475,10 @@ WtExpected<folly::Unit>::Type WtSessionBase::resetStream(
 
 WtExpected<folly::Unit>::Type WtSessionBase::setPriority(
     uint64_t streamId, quic::PriorityQueue::Priority priority) noexcept {
-  return folly::unit;
+  if (auto* wh = sm_.getBidiHandle(streamId).writeHandle) {
+    return wh->setPriority(priority);
+  }
+  return folly::makeUnexpected(WtErrCode::INVALID_STREAM_ID);
 }
 
 WtExpected<folly::Unit>::Type WtSessionBase::setPriorityQueue(

third-party/proxygen/src/proxygen/lib/http/webtransport/test/QuicWtSessionTest.cpp

@@ -420,6 +420,72 @@ TEST_F(QuicWtSessionTest, SetPriority) {
   EXPECT_EQ(httpPriority->incremental, true);
 }
 
+// The priority must reach the QuicSocket, otherwise the stream sits at default
+// priority in the connection's write queue and is never compared against the
+// other streams (or the wt datagram flows) sharing it.
+TEST_F(QuicWtSessionTest, SetPriorityPropagatesToQuicSocket) {
+  auto handle = session_->createUniStream();
+  ASSERT_TRUE(handle.hasValue());
+  auto id = handle.value()->getID();
+
+  // deliberately every field differs from HTTPPriorityQueue's default
+  // (u=3, i=true, o=0), so this cannot pass on an unpropagated priority
+  quic::HTTPPriorityQueue::Priority priority(/*u=*/5,
+                                             /*i=*/false,
+                                             /*o=*/42);
+  socketDriver_.expectSetPriority(id, priority);
+
+  EXPECT_TRUE(session_->setPriority(id, priority).hasValue());
+  // the local egress schedule is updated too
+  EXPECT_EQ(quic::HTTPPriorityQueue::Priority(handle.value()->getPriority()),
+            priority);
+}
+
+// Applications reprioritize through the write handle they already hold rather
+// than by stream id, so this is the path that has to reach the socket.
+TEST_F(QuicWtSessionTest, WriteHandleSetPriorityPropagatesToQuicSocket) {
+  auto handle = session_->createUniStream();
+  ASSERT_TRUE(handle.hasValue());
+  auto id = handle.value()->getID();
+
+  quic::HTTPPriorityQueue::Priority priority(/*u=*/5,
+                                             /*i=*/false,
+                                             /*o=*/42);
+  socketDriver_.expectSetPriority(id, priority);
+
+  EXPECT_TRUE(handle.value()->setPriority(priority).hasValue());
+  EXPECT_EQ(quic::HTTPPriorityQueue::Priority(handle.value()->getPriority()),
+            priority);
+}
+
+TEST_F(QuicWtSessionTest, SetPriorityUnknownStreamId) {
+  // no stream has been created, so there is nothing to prioritize
+  constexpr uint64_t kUnknownStreamId = 4;
+  EXPECT_CALL(*socketDriver_.getSocket(), setStreamPriority(_, _)).Times(0);
+
+  auto res = session_->setPriority(kUnknownStreamId,
+                                   quic::HTTPPriorityQueue::Priority(5, false));
+  ASSERT_TRUE(res.hasError());
+  EXPECT_EQ(res.error(), WebTransport::ErrorCode::INVALID_STREAM_ID);
+}
+
+// The socket half is best effort -- a transport that no longer has the stream
+// does not make the call a failure, and the local schedule still applies.
+TEST_F(QuicWtSessionTest, SetPriorityQuicSocketErrorIgnored) {
+  auto handle = session_->createUniStream();
+  ASSERT_TRUE(handle.hasValue());
+  auto id = handle.value()->getID();
+
+  quic::HTTPPriorityQueue::Priority priority(/*u=*/5, /*i=*/false, /*o=*/42);
+  EXPECT_CALL(*socketDriver_.getSocket(), setStreamPriority(id, _))
+      .WillOnce(Return(
+          quic::make_unexpected(quic::LocalErrorCode::STREAM_NOT_EXISTS)));
+
+  EXPECT_TRUE(session_->setPriority(id, priority).hasValue());
+  EXPECT_EQ(quic::HTTPPriorityQueue::Priority(handle.value()->getPriority()),
+            priority);
+}
+
 TEST_F(QuicWtSessionTest, AwaitWritable) {
   // Note: This test verifies basic functionality but doesn't test the blocking
   // scenario where awaitWritable returns a "not ready" future.
@@ -986,6 +1052,22 @@ TEST_F(H3WtSessionTest, CapsuleParseErrorClosesSession) {
       connectStreamCb_.events.back()));
 }
 
+// H3WtSession shares the QuicSocket with the http/3 session, so its wt streams
+// have to be ranked in that shared write queue like any other quic stream.
+TEST_F(H3WtSessionTest, SetPriorityPropagatesToQuicSocket) {
+  socketDriver_.setMaxUniStreams(1);
+  auto handle = session_->createUniStream();
+  ASSERT_TRUE(handle.hasValue());
+  auto id = handle.value()->getID();
+
+  quic::HTTPPriorityQueue::Priority priority(/*u=*/5,
+                                             /*i=*/false,
+                                             /*o=*/42);
+  socketDriver_.expectSetPriority(id, priority);
+
+  EXPECT_TRUE(handle.value()->setPriority(priority).hasValue());
+}
+
 TEST_F(H3WtSessionTest, AcquireIngressStream) {
   // client-initiated stream ids (server is the local endpoint)
   constexpr uint64_t kClientBidiId = 0;