io_uring backend error handling

Summary: io_uring backend error handling

Reviewed By: kevin-vigor

Differential Revision: D20737416

fbshipit-source-id: 23eebc98326bcd07ee308e8f71f972c7374f6108

folly/experimental/io/IoUringBackend.cpp

@@ -207,7 +207,7 @@ void* IoUringBackend::allocSubmissionEntry() {
 }
 
 int IoUringBackend::submitOne(IoCb* /*unused*/) {
-  return submitBusyCheck();
+  return submitBusyCheck(1, WaitForEventsMode::DONT_WAIT);
 }
 
 int IoUringBackend::cancelOne(IoCb* ioCb) {
@@ -221,7 +221,7 @@ int IoUringBackend::cancelOne(IoCb* ioCb) {
 
   rentry->prepPollRemove(sqe, ioCb); // prev entry
 
-  int ret = submitBusyCheck();
+  int ret = submitBusyCheck(1, WaitForEventsMode::DONT_WAIT);
 
   if (ret < 0) {
     // release the sqe
@@ -253,21 +253,33 @@ int IoUringBackend::getActiveEvents(WaitForEventsMode waitForEvents) {
   return static_cast<int>(i);
 }
 
-int IoUringBackend::submitBusyCheck() {
-  int num;
-  while ((num = ::io_uring_submit(&ioRing_)) == -EBUSY) {
-    // if we get EBUSY, try to consume some CQ entries
-    getActiveEvents(WaitForEventsMode::DONT_WAIT);
-  };
-  return num;
-}
+int IoUringBackend::submitBusyCheck(int num, WaitForEventsMode waitForEvents) {
+  int i = 0;
+  int res;
+  while (i < num) {
+    res = ::io_uring_submit(&ioRing_);
+    if (res == -EBUSY) {
+      // if we get EBUSY, try to consume some CQ entries
+      getActiveEvents(waitForEvents);
+      continue;
+    }
+    if (res < 0) {
+      // continue if interrupted
+      if (errno == EINTR) {
+        continue;
+      }
+
+      return res;
+    }
+
+    // we do not have any other entries to submit
+    if (res == 0) {
+      break;
+    }
+
+    i += res;
+  }
 
-int IoUringBackend::submitBusyCheckAndWait() {
-  int num;
-  while ((num = ::io_uring_submit_and_wait(&ioRing_, 1)) == -EBUSY) {
-    // if we get EBUSY, try to consume some CQ entries
-    getActiveEvents(WaitForEventsMode::DONT_WAIT);
-  };
   return num;
 }
 
@@ -291,14 +303,12 @@ size_t IoUringBackend::submitList(
         (ev->ev_events & EV_PERSIST) != 0);
     i++;
     if (ioCbs.empty()) {
-      int num = (waitForEvents == WaitForEventsMode::WAIT)
-          ? submitBusyCheckAndWait()
-          : submitBusyCheck();
+      int num = submitBusyCheck(i, waitForEvents);
       CHECK_EQ(num, i);
       ret += i;
     } else {
       if (static_cast<size_t>(i) == maxSubmit_) {
-        int num = submitBusyCheck();
+        int num = submitBusyCheck(i, waitForEvents);
         CHECK_EQ(num, i);
         ret += i;
         i = 0;

folly/experimental/io/IoUringBackend.h

@@ -79,8 +79,7 @@ class IoUringBackend : public PollIoBackend {
   int submitOne(IoCb* ioCb) override;
   int cancelOne(IoCb* ioCb) override;
 
-  int submitBusyCheck();
-  int submitBusyCheckAndWait();
+  int submitBusyCheck(int num, WaitForEventsMode waitForEvents);
 
   struct IoSqe : public PollIoBackend::IoCb {
     explicit IoSqe(PollIoBackend* backend = nullptr, bool poolAlloc = true)

folly/experimental/io/test/IoUringBackendTest.cpp

@@ -28,15 +28,16 @@ namespace {
 class EventFD : public folly::EventHandler {
  public:
   EventFD(
+      bool valid,
       uint64_t num,
       uint64_t& total,
       bool persist,
       folly::EventBase* eventBase)
-      : EventFD(total, createFd(num), persist, eventBase) {}
+      : EventFD(total, valid ? createFd(num) : -1, persist, eventBase) {}
   ~EventFD() override {
     unregisterHandler();
 
-    if (fd_ > 0) {
+    if (fd_ >= 0) {
       ::close(fd_);
       fd_ = -1;
     }
@@ -114,7 +115,7 @@ void testOverflow(bool overflow, bool persist) {
   eventsVec.reserve(kNumEventFds);
   for (size_t i = 0; i < kNumEventFds; i++) {
     eventsVec.emplace_back(
-        std::make_unique<EventFD>(kEventFdCount, total, persist, &evb));
+        std::make_unique<EventFD>(true, kEventFdCount, total, persist, &evb));
   }
 
   evb.loopForever();
@@ -123,6 +124,40 @@ void testOverflow(bool overflow, bool persist) {
     CHECK_GE(eventsVec[i]->getNum(), kEventFdCount);
   }
 }
+
+void testInvalidFd(size_t numTotal, size_t numValid, size_t numInvalid) {
+  static constexpr size_t kBackendCapacity = 128;
+  static constexpr size_t kBackendMaxSubmit = 64;
+
+  auto total = numTotal;
+
+  std::unique_ptr<folly::EventBaseBackendBase> backend;
+
+  try {
+    backend = std::make_unique<folly::IoUringBackend>(
+        kBackendCapacity, kBackendMaxSubmit);
+  } catch (const folly::IoUringBackend::NotAvailable&) {
+  }
+
+  SKIP_IF(!backend) << "Backend not available";
+
+  folly::EventBase evb(std::move(backend));
+
+  std::vector<std::unique_ptr<EventFD>> eventsVec;
+  eventsVec.reserve(numTotal);
+
+  for (size_t i = 0; i < numTotal; i++) {
+    bool valid = (i % (numValid + numInvalid)) < numValid;
+    eventsVec.emplace_back(
+        std::make_unique<EventFD>(valid, 1, total, false /*persist*/, &evb));
+  }
+
+  evb.loopForever();
+
+  for (size_t i = 0; i < numTotal; i++) {
+    CHECK_GE(eventsVec[i]->getNum(), 1);
+  }
+}
 } // namespace
 
 TEST(IoUringBackend, NoOverflowNoPersist) {
@@ -141,6 +176,21 @@ TEST(IoUringBackend, OverflowPersist) {
   testOverflow(true, true);
 }
 
+// 9 valid fds followed by an invalid one
+TEST(IoUringBackend, Invalid_fd_9_1) {
+  testInvalidFd(32, 10, 1);
+}
+
+// only invalid fds
+TEST(IoUringBackend, Invalid_fd_0_10) {
+  testInvalidFd(32, 0, 10);
+}
+
+// equal distribution
+TEST(IoUringBackend, Invalid_fd_5_5) {
+  testInvalidFd(32, 10, 10);
+}
+
 TEST(IoUringBackend, RegisteredFds) {
   static constexpr size_t kBackendCapacity = 64;
   static constexpr size_t kBackendMaxSubmit = 32;