Use QuicExecutor in QMUX sessions

Summary: There is no functionality change in this commit. We're just swapping out the `folly::EventBase` for the `QuicExecutor`.

Reviewed By: kvtsoy

Differential Revision: D114159297

fbshipit-source-id: 0c8eec053174b0d8bab0fb40baaa180f6d9d57ac

Author: Aman Sharma

proxygen/lib/http/webtransport/CMakeLists.txt

@@ -98,7 +98,6 @@ proxygen_add_library(proxygen_http_webtransport_wt_util
     proxygen_http_codec_webtransport_webtransport_capsule_codec
     proxygen_http_webtransport_wt_stream_manager
     Folly::folly_executor
-    Folly::folly_io_async_async_base
 )
 
 proxygen_add_library(proxygen_http_webtransport_webtransport_session

proxygen/lib/http/webtransport/WtUtils.h

@@ -9,7 +9,6 @@
 #pragma once
 
 #include <folly/Executor.h>
-#include <folly/io/async/EventBase.h>
 #include <proxygen/lib/http/codec/TransportDirection.h>
 #include <proxygen/lib/http/codec/webtransport/WebTransportCapsuleCodec.h>
 #include <proxygen/lib/http/webtransport/WtStreamManager.h>
@@ -211,10 +210,6 @@ class WtSessionBase : public WebTransport {
     return executor_;
   }
 
-  folly::EventBase* evb() {
-    return dynamic_cast<folly::EventBase*>(executor_);
-  }
-
  protected:
   folly::Promise<folly::Unit>& uniCreditPromise() noexcept {
     return awaitUniCredit_;

proxygen/lib/transport/qmux/CMakeLists.txt

@@ -63,16 +63,14 @@ proxygen_add_library(proxygen_transport_qmux_qmux_session
     proxygen_http_webtransport_wt_util
     proxygen_transport_qmux_qmux_framer
     proxygen_transport_qmux_qmux_transport
-    Folly::folly_io_async_async_base
+    mvfst::mvfst_common_events_quic_executor
 )
 
 proxygen_add_library(proxygen_transport_qmux_qmux_connector
   SRCS
     QmuxConnector.cpp
   DEPS
     mvfst::mvfst_folly_utils
-    Folly::folly_coro_timeout
-    Folly::folly_futures_core
     Folly::folly_io_iobuf
     Folly::folly_logging_logging
     fmt::fmt
@@ -81,6 +79,7 @@ proxygen_add_library(proxygen_transport_qmux_qmux_connector
     proxygen_transport_qmux_qmux_framer
     proxygen_transport_qmux_qmux_session
     proxygen_transport_qmux_qmux_transport
+    mvfst::mvfst_common_events_quic_executor
     Folly::folly_coro_task
 )
 

proxygen/lib/transport/qmux/QmuxConnector.cpp

@@ -8,10 +8,9 @@
 
 #include <proxygen/lib/transport/qmux/QmuxConnector.h>
 
+#include <chrono>
 #include <cstdint>
 #include <fmt/core.h>
-#include <folly/coro/Timeout.h>
-#include <folly/futures/ThreadWheelTimekeeper.h>
 #include <folly/io/Cursor.h>
 #include <folly/io/IOBufQueue.h>
 #include <folly/logging/xlog.h>
@@ -148,12 +147,21 @@ folly::coro::Task<QxTransportParams> readPeerTransportParams(
     QmuxTransport& transport,
     folly::IOBufQueue& ingressBuf,
     std::chrono::milliseconds timeout) {
+  const auto deadline = std::chrono::steady_clock::now() + timeout;
   while (true) {
+    const auto remaining =
+        std::chrono::duration_cast<std::chrono::milliseconds>(
+            deadline - std::chrono::steady_clock::now());
+    if (remaining <= std::chrono::milliseconds::zero()) {
+      co_yield co_error(
+          std::runtime_error("timed out waiting for QX_TRANSPORT_PARAMETERS"));
+    }
+
     auto readRes = co_await folly::coro::co_awaitTry(
         transport.read(ingressBuf,
                        /*minReadSize=*/1,
                        /*newAllocationSize=*/4096,
-                       timeout));
+                       remaining));
     if (readRes.hasException()) {
       co_yield co_error(readRes.exception());
     }
@@ -194,7 +202,7 @@ WtStreamManager::WtConfig makeWtConfig(
 }
 
 folly::coro::Task<QmuxSession::Ptr> QmuxConnector::connect(
-    folly::EventBase* evb,
+    std::shared_ptr<quic::QuicExecutor> executor,
     WtDir dir,
     QxTransportParams selfParams,
     std::unique_ptr<QmuxTransport> transport,
@@ -220,10 +228,9 @@ folly::coro::Task<QmuxSession::Ptr> QmuxConnector::connect(
   //    timeout. peelTransportParams trims the TP frame and rewraps any
   //    same-record trailing bytes so they parse cleanly downstream.
   folly::IOBufQueue ingressBuf{folly::IOBufQueue::cacheChainLength()};
-  folly::EventBaseThreadTimekeeper tk{*evb};
 
-  auto peerParams = co_await folly::coro::co_awaitTry(folly::coro::timeout(
-      readPeerTransportParams(*transport, ingressBuf, timeout), timeout, &tk));
+  auto peerParams = co_await folly::coro::co_awaitTry(
+      readPeerTransportParams(*transport, ingressBuf, timeout));
   if (peerParams.hasException()) {
     XLOG(ERR) << "QmuxConnector: TP read failed: " << peerParams.exception();
     co_yield co_error(peerParams.exception());
@@ -246,7 +253,7 @@ folly::coro::Task<QmuxSession::Ptr> QmuxConnector::connect(
     effectiveMaxIdleTimeoutMs = peerParams->maxIdleTimeout;
   }
 
-  co_return std::make_shared<QmuxSession>(evb,
+  co_return std::make_shared<QmuxSession>(std::move(executor),
                                           dir,
                                           std::move(selfParams),
                                           std::move(transport),

proxygen/lib/transport/qmux/QmuxConnector.h

@@ -14,17 +14,14 @@
 #include <proxygen/lib/transport/qmux/QmuxFramer.h>
 #include <proxygen/lib/transport/qmux/QmuxSession.h>
 #include <proxygen/lib/transport/qmux/QmuxTransport.h>
-
-namespace folly {
-class EventBase;
-} // namespace folly
+#include <quic/common/events/QuicExecutor.h>
 
 namespace proxygen::qmux {
 
 class QmuxConnector {
  public:
   static folly::coro::Task<QmuxSession::Ptr> connect(
-      folly::EventBase* evb,
+      std::shared_ptr<quic::QuicExecutor> executor,
       WtDir dir,
       QxTransportParams selfParams,
       std::unique_ptr<QmuxTransport> transport,

proxygen/lib/transport/qmux/QmuxSession.cpp

@@ -134,7 +134,7 @@ class QmuxCallback : public QmuxCodec::Callback {
 
 //////// QmuxSession ////////
 
-QmuxSession::QmuxSession(folly::EventBase* evb,
+QmuxSession::QmuxSession(std::shared_ptr<quic::QuicExecutor> executor,
                          WtDir dir,
                          QxTransportParams selfParams,
                          std::unique_ptr<QmuxTransport> transport,
@@ -144,7 +144,8 @@ QmuxSession::QmuxSession(folly::EventBase* evb,
                          std::unique_ptr<folly::IOBuf> initialIngress,
                          Config config)
     : CoroWtSessionBase(dir, wtConfig),
-      WtSessionBase(evb, sm),
+      WtSessionBase(executor.get(), sm),
+      executor_(std::move(executor)),
       localAddr_(transport->getLocalAddress()),
       peerAddr_(transport->getPeerAddress()),
       transport_(std::move(transport)),
@@ -233,7 +234,7 @@ folly::coro::Task<void> QmuxSession::readLoop(Ptr self) {
       break;
     }
   }
-  idleTimeout_.cancelTimeout();
+  idleTimeout_.cancelTimerCallback();
   XLOG(DBG4) << "QmuxSession::readLoop exiting";
   sm.shutdown(WtStreamManager::CloseSession{.err = 0x00,
                                             .msg = "stream ingress closed"});
@@ -246,8 +247,8 @@ void QmuxSession::resetIdleTimeout() {
     // connection has no idle deadline. Nothing to do.
     return;
   }
-  idleTimeout_.cancelTimeout();
-  evb()->timer().scheduleTimeout(
+  idleTimeout_.cancelTimerCallback();
+  executor_->scheduleTimeout(
       &idleTimeout_, std::chrono::milliseconds(effectiveMaxIdleTimeoutMs_));
 }
 

proxygen/lib/transport/qmux/QmuxSession.h

@@ -9,16 +9,16 @@
 #pragma once
 
 #include <deque>
-#include <folly/io/async/HHWheelTimer.h>
 #include <proxygen/lib/http/coro/util/CoroWtSession.h>
 #include <proxygen/lib/http/webtransport/WtStreamManager.h>
 #include <proxygen/lib/http/webtransport/WtUtils.h>
 #include <proxygen/lib/transport/qmux/QmuxFramer.h>
 #include <proxygen/lib/transport/qmux/QmuxTransport.h>
+#include <quic/common/events/QuicExecutor.h>
 
 namespace folly {
 class AsyncTransport;
-}
+} // namespace folly
 
 namespace proxygen::qmux {
 
@@ -37,7 +37,7 @@ class QmuxSession
  public:
   using Ptr = std::shared_ptr<QmuxSession>;
 
-  QmuxSession(folly::EventBase* evb,
+  QmuxSession(std::shared_ptr<quic::QuicExecutor> executor,
               WtDir dir,
               QxTransportParams selfParams,
               std::unique_ptr<QmuxTransport> transport,
@@ -46,6 +46,7 @@ class QmuxSession
               uint64_t effectiveMaxIdleTimeoutMs,
               std::unique_ptr<folly::IOBuf> initialIngress,
               Config config);
+
   ~QmuxSession() override;
 
   void setHandler(WebTransportHandler* handler) {
@@ -77,7 +78,7 @@ class QmuxSession
   [[nodiscard]] folly::AsyncTransport* getUnderlyingTransport() const noexcept;
 
   [[nodiscard]] bool isIdleTimeoutScheduled() const noexcept {
-    return idleTimeout_.isScheduled();
+    return idleTimeout_.isTimerCallbackScheduled();
   }
 
  private:
@@ -90,19 +91,22 @@ class QmuxSession
   void resetIdleTimeout();
   void onIdleTimeout();
 
-  class IdleTimeoutCallback : public folly::HHWheelTimer::Callback {
+  class IdleTimeoutCallback : public quic::QuicTimerCallback {
    public:
     explicit IdleTimeoutCallback(QmuxSession& session) noexcept
         : session_(session) {
     }
     void timeoutExpired() noexcept override {
       session_.onIdleTimeout();
     }
+    void callbackCanceled() noexcept override {
+    }
 
    private:
     QmuxSession& session_;
   };
 
+  std::shared_ptr<quic::QuicExecutor> executor_;
   WebTransportHandler* wtHandler_{nullptr};
   folly::SocketAddress localAddr_;
   folly::SocketAddress peerAddr_;

proxygen/lib/transport/qmux/test/CMakeLists.txt

@@ -45,6 +45,7 @@ proxygen_add_test(TARGET QmuxSessionTest
     proxygen_http_codec_webtransport_webtransport_framer
     proxygen_http_webtransport_wt_stream_manager
     proxygen_http_webtransport_wt_util
+    mvfst::mvfst_common_events_quic_folly_executor_impl
     TestCoroTransport
     testmain
 )
@@ -56,6 +57,7 @@ proxygen_add_test(TARGET QmuxConnectorTest
     proxygen_transport_qmux_qmux_connector
     proxygen_transport_qmux_folly_qmux_transport
     proxygen_transport_qmux_qmux_framer
+    mvfst::mvfst_common_events_quic_folly_executor_impl
     TestCoroTransport
     testmain
 )

proxygen/lib/transport/qmux/test/QmuxConnectorTest.cpp

@@ -15,6 +15,7 @@
 #include <proxygen/lib/http/coro/transport/test/TestCoroTransport.h>
 #include <proxygen/lib/transport/qmux/FollyQmuxTransport.h>
 #include <proxygen/lib/transport/qmux/QmuxFramer.h>
+#include <quic/common/events/QuicFollyExecutorImpl.h>
 
 using namespace proxygen;
 using namespace proxygen::qmux;
@@ -83,15 +84,19 @@ std::unique_ptr<folly::IOBuf> tpAndPingRecord(const QxTransportParams& params,
 // in a capturing lambda would dangle the captures once the lambda temporary
 // destructs at the call site.
 folly::coro::Task<void> connectAndCapture(
-    folly::EventBase* evb,
+    std::shared_ptr<quic::QuicExecutor> executor,
     WtDir dir,
     QxTransportParams selfParams,
     std::unique_ptr<QmuxTransport> transport,
     std::chrono::milliseconds timeout,
     folly::Try<QmuxSession::Ptr>* out,
     bool* done) {
-  *out = co_await folly::coro::co_awaitTry(QmuxConnector::connect(
-      evb, dir, std::move(selfParams), std::move(transport), timeout));
+  *out = co_await folly::coro::co_awaitTry(
+      QmuxConnector::connect(std::move(executor),
+                             dir,
+                             std::move(selfParams),
+                             std::move(transport),
+                             timeout));
   *done = true;
 }
 
@@ -113,8 +118,8 @@ class QmuxConnectorTest : public ::testing::Test {
       QxTransportParams selfParams = sampleSelfParams()) {
     folly::Try<QmuxSession::Ptr> result;
     bool done = false;
-    co_withExecutor(&evb_,
-                    connectAndCapture(&evb_,
+    co_withExecutor(executor_.get(),
+                    connectAndCapture(executor_,
                                       WtDir::Client,
                                       std::move(selfParams),
                                       std::make_unique<FollyQmuxTransport>(
@@ -130,6 +135,8 @@ class QmuxConnectorTest : public ::testing::Test {
   }
 
   folly::EventBase evb_;
+  std::shared_ptr<quic::QuicFollyExecutorImpl> executor_{
+      std::make_shared<quic::QuicFollyExecutorImpl>(&evb_)};
 };
 
 } // namespace

proxygen/lib/transport/qmux/test/QmuxSessionTest.cpp

@@ -17,6 +17,7 @@
 #include <proxygen/lib/transport/qmux/FollyQmuxTransport.h>
 #include <proxygen/lib/transport/qmux/QmuxCodec.h>
 #include <proxygen/lib/transport/qmux/QmuxFramer.h>
+#include <quic/common/events/QuicFollyExecutorImpl.h>
 
 using namespace proxygen;
 using namespace proxygen::qmux;
@@ -273,7 +274,7 @@ class QmuxSessionTest : public ::testing::Test {
         .peerMaxStreamDataUni = peerParams.initialMaxStreamDataUni};
 
     session_ = std::make_shared<QmuxSession>(
-        &evb_,
+        executor_,
         WtDir::Server,
         selfParams,
         std::make_unique<FollyQmuxTransport>(std::move(transport)),
@@ -338,6 +339,8 @@ class QmuxSessionTest : public ::testing::Test {
   static constexpr uint64_t kPeerBidiId = 0x00;
 
   folly::EventBase evb_;
+  std::shared_ptr<quic::QuicFollyExecutorImpl> executor_{
+      std::make_shared<quic::QuicFollyExecutorImpl>(&evb_)};
   std::unique_ptr<TestCoroTransport::State> state_;
   TestCoroTransport* transport_{nullptr};
   std::unique_ptr<TestWtHandler> handler_;
@@ -492,7 +495,7 @@ TEST_F(QmuxSessionTest, InitialIngress_IsDrainedOnStartup) {
   auto preroll = streamRecord(kPeerBidiId, "preroll", /*fin=*/false);
 
   auto session = std::make_shared<QmuxSession>(
-      &evb_,
+      executor_,
       WtDir::Server,
       selfParams_,
       std::make_unique<FollyQmuxTransport>(std::move(transport)),
@@ -598,7 +601,7 @@ TEST_F(QmuxSessionTest, IdleTimer_ArmedWhenEffectiveTimeoutNonzero) {
                                      .peerMaxStreamDataUni = 1 << 16};
 
   auto session = std::make_shared<QmuxSession>(
-      &evb_,
+      executor_,
       WtDir::Server,
       selfParams_,
       std::make_unique<FollyQmuxTransport>(std::move(transport)),