folly async io_uring: set EVB for IoSqeBase derived classes

Summary:
There are two categories of IoSqeBase derived classes:
1. IoSqe in IoUringBackend
2. Various derivatives in AsyncIoUringSocket

Call `IoSqeBase::setEventBase()` from all derived classes to set/unset the EVB.

It's possible for `AsyncIoUringSocket` to detach and attach to different EVBs. Ensure that `IoSqeBase`s are synced.

Reviewed By: yfeldblum

Differential Revision: D68540579

fbshipit-source-id: 83ad739aec37378e08e4d7fd4855166af6bef290

Author: spikeh

folly/io/async/AsyncIoUringSocket.cpp

@@ -141,6 +141,7 @@ AsyncIoUringSocket::ReadSqe::ReadSqe(AsyncIoUringSocket* parent)
     : IoSqeBase(IoSqeBase::Type::Read), parent_(parent) {
   supportsMultishotRecv_ = parent->options_.multishotRecv &&
       parent->backend_->kernelSupportsRecvmsgMultishot();
+  setEventBase(parent->evb_);
 }
 
 AsyncIoUringSocket::~AsyncIoUringSocket() {
@@ -868,6 +869,8 @@ AsyncIoUringSocket::WriteSqe::WriteSqe(
   msg_.msg_control = nullptr;
   msg_.msg_controllen = 0;
   msg_.msg_flags = 0;
+
+  setEventBase(parent->evb_);
 }
 
 int AsyncIoUringSocket::WriteSqe::sendMsgFlags() const {
@@ -1019,8 +1022,8 @@ void AsyncIoUringSocket::attachEventBase(EventBase* evb) {
     std::move(*detachedWriteResult_)
         .via(evb)
         .thenValue(
-            [w = writeSqeActive_,
-             a = std::weak_ptr<folly::Unit>(alive_)](auto&& resFlagsPairs) {
+            [w = writeSqeActive_, a = std::weak_ptr<folly::Unit>(alive_), evb](
+                auto&& resFlagsPairs) {
               VLOG(5) << "attached write done, " << resFlagsPairs.size();
               if (!a.lock()) {
                 return;
@@ -1031,6 +1034,7 @@ void AsyncIoUringSocket::attachEventBase(EventBase* evb) {
                 cqe.res = res;
                 cqe.flags = flags;
 
+                evb->bumpHandlingTime();
                 if (w->cancelled()) {
                   w->callbackCancelled(&cqe);
                 } else {
@@ -1138,6 +1142,7 @@ void AsyncIoUringSocket::detachEventBase() {
   }
   readSqe_ = ReadSqe::UniquePtr(new ReadSqe(this));
   readSqe_->setReadCallback(oldReadCallback, false);
+  readSqe_->setEventBase(nullptr);
 
   unregisterFd();
   if (!drc) {
@@ -1178,6 +1183,7 @@ folly::Optional<folly::SemiFuture<std::unique_ptr<IOBuf>>>
 AsyncIoUringSocket::ReadSqe::detachEventBase() {
   alive_ = nullptr;
   parent_ = nullptr;
+  setEventBase(nullptr);
   return std::move(oldEventBaseRead_);
 }
 
@@ -1193,6 +1199,7 @@ void AsyncIoUringSocket::ReadSqe::attachEventBase() {
     return;
   }
   auto* evb = parent_->evb_;
+  setEventBase(evb);
   alive_ = std::make_shared<folly::Unit>();
   folly::Func deferred =
       [p = parent_, a = std::weak_ptr<folly::Unit>(alive_)]() {
@@ -1219,6 +1226,7 @@ AsyncIoUringSocket::FastOpenSqe::FastOpenSqe(
       parent_(parent),
       initialWrite(std::move(i)) {
   addrLen_ = addr.getAddress(&addrStorage);
+  setEventBase(parent->evb_);
 }
 
 void AsyncIoUringSocket::FastOpenSqe::cleanupMsg() noexcept {
@@ -1311,6 +1319,7 @@ AsyncIoUringSocket::WriteSqe::detachEventBase() {
   newSqe->refs_ = refs_;
 
   parent_ = nullptr;
+  setEventBase(nullptr);
   detachedSignal_ =
       [prom = std::move(promise),
        ret = std::vector<std::pair<int, uint32_t>>{},

folly/io/async/AsyncIoUringSocket.h

@@ -351,7 +351,10 @@ class AsyncIoUringSocket : public AsyncSocketTransport {
 
   struct CloseSqe : IoSqeBase {
     explicit CloseSqe(AsyncIoUringSocket* parent)
-        : IoSqeBase(IoSqeBase::Type::Close), parent_(parent) {}
+        : IoSqeBase(IoSqeBase::Type::Close), parent_(parent) {
+      setEventBase(parent->evb_);
+    }
+
     void processSubmit(struct io_uring_sqe* sqe) noexcept override {
       parent_->closeProcessSubmit(sqe);
     }
@@ -417,7 +420,10 @@ class AsyncIoUringSocket : public AsyncSocketTransport {
     explicit ConnectSqe(AsyncIoUringSocket* parent)
         : IoSqeBase(IoSqeBase::Type::Connect),
           AsyncTimeout(parent->evb_),
-          parent_(parent) {}
+          parent_(parent) {
+      setEventBase(parent->evb_);
+    }
+
     void processSubmit(struct io_uring_sqe* sqe) noexcept override {
       parent_->processConnectSubmit(sqe, addrStorage);
     }

folly/io/async/IoUringBackend.cpp

@@ -1258,6 +1258,7 @@ int IoUringBackend::eb_event_add(Event& event, const struct timeval* timeout) {
     auto* ioSqe = allocIoSqe(event.getCallback());
     CHECK(ioSqe);
     ioSqe->event_ = &event;
+    ioSqe->setEventBase(event.eb_ev_base());
 
     // just append it
     submitList_.push_back(*ioSqe);

folly/io/async/IoUringBackend.h

@@ -522,6 +522,7 @@ class IoUringBackend : public EventBaseBackendBase {
     FOLLY_ALWAYS_INLINE void resetEvent() {
       // remove it from the list
       unlink();
+      setEventBase(nullptr);
       if (event_) {
         event_->setUserData(nullptr);
         event_ = nullptr;