io: add io_uring wrapper

bazel/foreign_cc/BUILD

@@ -6,6 +6,13 @@ licenses(["notice"])  # Apache 2
 
 envoy_package()
 
+configure_make(
+    name = "liburing",
+    configure_in_place = True,
+    lib_source = "@com_github_axboe_liburing//:all",
+    tags = ["skip_on_windows"],
+)
+
 # autotools packages are unusable on Windows as-is
 # TODO: Consider our own gperftools.BUILD file as we do with many other packages
 configure_make(

bazel/repositories.bzl

@@ -156,6 +156,7 @@ def envoy_dependencies(skip_targets = []):
     # The long repo names (`com_github_fmtlib_fmt` instead of `fmtlib`) are
     # semi-standard in the Bazel community, intended to avoid both duplicate
     # dependencies and name conflicts.
+    _com_github_axboe_liburing()
     _com_github_c_ares_c_ares()
     _com_github_circonus_labs_libcircllhist()
     _com_github_cyan4973_xxhash()
@@ -269,6 +270,16 @@ def _com_github_circonus_labs_libcircllhist():
         actual = "@com_github_circonus_labs_libcircllhist//:libcircllhist",
     )
 
+def _com_github_axboe_liburing():
+    external_http_archive(
+        name = "com_github_axboe_liburing",
+        build_file_content = BUILD_ALL_CONTENT,
+    )
+    native.bind(
+        name = "uring",
+        actual = "@envoy//bazel/foreign_cc:liburing",
+    )
+
 def _com_github_c_ares_c_ares():
     external_http_archive(
         name = "com_github_c_ares_c_ares",

bazel/repository_locations.bzl

@@ -112,6 +112,18 @@ REPOSITORY_LOCATIONS_SPEC = dict(
         release_date = "2021-06-03",
         cpe = "N/A",
     ),
+    com_github_axboe_liburing = dict(
+        project_name = "liburing",
+        project_desc = "C helpers to set up and tear down io_uring instances",
+        project_url = "https://github.com/axboe/liburing",
+        version = "2.1",
+        sha256 = "f1e0500cb3934b0b61c5020c3999a973c9c93b618faff1eba75aadc95bb03e07",
+        strip_prefix = "liburing-liburing-{version}",
+        urls = ["https://github.com/axboe/liburing/archive/liburing-{version}.tar.gz"],
+        use_category = ["dataplane_core", "controlplane"],
+        release_date = "2021-09-09",
+        cpe = "N/A",
+    ),
     # This dependency is built only when performance tracing is enabled with the
     # option --define=perf_tracing=enabled. It's never built for releases.
     com_github_google_perfetto = dict(

source/common/io/BUILD

@@ -0,0 +1,33 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_package()
+
+envoy_cc_library(
+    name = "io_uring_interface",
+    hdrs = [
+        "io_uring.h",
+    ],
+    deps = [
+        "//source/common/network:address_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "io_uring_impl_lib",
+    srcs = [
+        "io_uring_impl.cc",
+    ],
+    hdrs = [
+        "io_uring_impl.h",
+    ],
+    external_deps = ["uring"],
+    deps = [
+        ":io_uring_interface",
+    ],
+)

source/common/io/io_uring.h

@@ -0,0 +1,121 @@
+#pragma once
+
+#include "envoy/common/pure.h"
+
+#include "source/common/network/address_impl.h"
+
+namespace Envoy {
+namespace Io {
+
+/**
+ * Callback invoked when iterating over entries in the completion queue.
+ * @param user_data is any data attached to an entry submitted to the submission
+ * queue.
+ * @param result is a return code of submitted system call.
+ */
+using CompletionCb = std::function<void(void* user_data, int32_t result)>;
+
+enum class IoUringResult { Ok, Busy, Failed };
+
+/**
+ * Abstract wrapper around `io_uring`.
+ */
+class IoUring {
+public:
+  virtual ~IoUring() = default;
+
+  /**
+   * Registers an eventfd file descriptor for the ring and returns it.
+   * It can be used for integration with event loops.
+   */
+  virtual os_fd_t registerEventfd() PURE;
+
+  /**
+   * Resets the eventfd file descriptor for the ring.
+   */
+  virtual void unregisterEventfd() PURE;
+
+  /**
+   * Returns true if an eventfd file descriptor is registered with the ring.
+   */
+  virtual bool isEventfdRegistered() const PURE;
+
+  /**
+   * Iterates over entries in the completion queue, calls the given callback for
+   * every entry and marks them consumed.
+   */
+  virtual void forEveryCompletion(CompletionCb completion_cb) PURE;
+
+  /**
+   * Prepares an accept system call and puts it into the submission queue.
+   * Returns IoUringResult::Failed in case the submission queue is full already
+   * and IoUringResult::Ok otherwise.
+   */
+  virtual IoUringResult prepareAccept(os_fd_t fd, struct sockaddr* remote_addr,
+                                      socklen_t* remote_addr_len, void* user_data) PURE;
+
+  /**
+   * Prepares a connect system call and puts it into the submission queue.
+   * Returns IoUringResult::Failed in case the submission queue is full already
+   * and IoUringResult::Ok otherwise.
+   */
+  virtual IoUringResult prepareConnect(os_fd_t fd,
+                                       const Network::Address::InstanceConstSharedPtr& address,
+                                       void* user_data) PURE;
+
+  /**
+   * Prepares a readv system call and puts it into the submission queue.
+   * Returns IoUringResult::Failed in case the submission queue is full already
+   * and IoUringResult::Ok otherwise.
+   */
+  virtual IoUringResult prepareReadv(os_fd_t fd, const struct iovec* iovecs, unsigned nr_vecs,
+                                     off_t offset, void* user_data) PURE;
+
+  /**
+   * Prepares a writev system call and puts it into the submission queue.
+   * Returns IoUringResult::Failed in case the submission queue is full already
+   * and IoUringResult::Ok otherwise.
+   */
+  virtual IoUringResult prepareWritev(os_fd_t fd, const struct iovec* iovecs, unsigned nr_vecs,
+                                      off_t offset, void* user_data) PURE;
+
+  /**
+   * Prepares a close system call and puts it into the submission queue.
+   * Returns IoUringResult::Failed in case the submission queue is full already
+   * and IoUringResult::Ok otherwise.
+   */
+  virtual IoUringResult prepareClose(os_fd_t fd, void* user_data) PURE;
+
+  /**
+   * Submits the entries in the submission queue to the kernel using the
+   * `io_uring_enter()` system call.
+   * Returns IoUringResult::Ok in case of success and may return
+   * IoUringResult::Busy if we over commit the number of requests. In the latter
+   * case the application should drain the completion queue by handling some completions
+   * with the forEveryCompletion() method and try again.
+   */
+  virtual IoUringResult submit() PURE;
+};
+
+/**
+ * Abstract factory for IoUring wrappers.
+ */
+class IoUringFactory {
+public:
+  virtual ~IoUringFactory() = default;
+
+  /**
+   * Returns an instance of IoUring and creates it if needed for the current
+   * thread.
+   */
+  virtual IoUring& getOrCreate() const PURE;
+
+  /**
+   * Initializes a factory upon server readiness. For example this method can be
+   * used to set TLS.
+   */
+  virtual void onServerInitialized() PURE;
+};
+
+} // namespace Io
+} // namespace Envoy

source/common/io/io_uring_impl.cc

@@ -0,0 +1,135 @@
+#include "source/common/io/io_uring_impl.h"
+
+#include <sys/eventfd.h>
+
+namespace Envoy {
+namespace Io {
+
+IoUringFactoryImpl::IoUringFactoryImpl(uint32_t io_uring_size, bool use_submission_queue_polling,
+                                       ThreadLocal::SlotAllocator& tls)
+    : io_uring_size_(io_uring_size), use_submission_queue_polling_(use_submission_queue_polling),
+      tls_(tls) {}
+
+IoUring& IoUringFactoryImpl::getOrCreate() const {
+  return const_cast<IoUringImpl&>(tls_.get().ref());
+}
+
+void IoUringFactoryImpl::onServerInitialized() {
+  tls_.set([io_uring_size = io_uring_size_,
+            use_submission_queue_polling = use_submission_queue_polling_](Event::Dispatcher&) {
+    return std::make_shared<IoUringImpl>(io_uring_size, use_submission_queue_polling);
+  });
+}
+
+IoUringImpl::IoUringImpl(uint32_t io_uring_size, bool use_submission_queue_polling)
+    : io_uring_size_(io_uring_size), cqes_(io_uring_size_, nullptr) {
+  struct io_uring_params p {};
+  if (use_submission_queue_polling) {
+    p.flags |= IORING_SETUP_SQPOLL;
+  }
+  int ret = io_uring_queue_init_params(io_uring_size_, &ring_, &p);
+  RELEASE_ASSERT(ret == 0, fmt::format("unable to initialize io_uring: {}", errorDetails(-ret)));
+}
+
+IoUringImpl::~IoUringImpl() { io_uring_queue_exit(&ring_); }
+
+os_fd_t IoUringImpl::registerEventfd() {
+  ASSERT(!isEventfdRegistered());
+  event_fd_ = eventfd(0, 0);
+  int res = io_uring_register_eventfd(&ring_, event_fd_);
+  RELEASE_ASSERT(res == 0, fmt::format("unable to register eventfd: {}", errorDetails(-res)));
+  return event_fd_;
+}
+
+void IoUringImpl::unregisterEventfd() {
+  int res = io_uring_unregister_eventfd(&ring_);
+  RELEASE_ASSERT(res == 0, fmt::format("unable to unregister eventfd: {}", errorDetails(-res)));
+  SET_SOCKET_INVALID(event_fd_);
+}
+
+bool IoUringImpl::isEventfdRegistered() const { return SOCKET_VALID(event_fd_); }
+
+void IoUringImpl::forEveryCompletion(CompletionCb completion_cb) {
+  ASSERT(SOCKET_VALID(event_fd_));
+
+  eventfd_t v;
+  int ret = eventfd_read(event_fd_, &v);
+  RELEASE_ASSERT(ret == 0, "unable to drain eventfd");
+
+  unsigned count = io_uring_peek_batch_cqe(&ring_, cqes_.data(), io_uring_size_);
+
+  for (unsigned i = 0; i < count; ++i) {
+    struct io_uring_cqe* cqe = cqes_[i];
+    completion_cb(reinterpret_cast<void*>(cqe->user_data), cqe->res);
+  }
+  io_uring_cq_advance(&ring_, count);
+}
+
+IoUringResult IoUringImpl::prepareAccept(os_fd_t fd, struct sockaddr* remote_addr,
+                                         socklen_t* remote_addr_len, void* user_data) {
+  struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
+  if (sqe == nullptr) {
+    return IoUringResult::Failed;
+  }
+
+  io_uring_prep_accept(sqe, fd, remote_addr, remote_addr_len, 0);
+  io_uring_sqe_set_data(sqe, user_data);
+  return IoUringResult::Ok;
+}
+
+IoUringResult IoUringImpl::prepareConnect(os_fd_t fd,
+                                          const Network::Address::InstanceConstSharedPtr& address,
+                                          void* user_data) {
+  struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
+  if (sqe == nullptr) {
+    return IoUringResult::Failed;
+  }
+
+  io_uring_prep_connect(sqe, fd, address->sockAddr(), address->sockAddrLen());
+  io_uring_sqe_set_data(sqe, user_data);
+  return IoUringResult::Ok;
+}
+
+IoUringResult IoUringImpl::prepareReadv(os_fd_t fd, const struct iovec* iovecs, unsigned nr_vecs,
+                                        off_t offset, void* user_data) {
+  struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
+  if (sqe == nullptr) {
+    return IoUringResult::Failed;
+  }
+
+  io_uring_prep_readv(sqe, fd, iovecs, nr_vecs, offset);
+  io_uring_sqe_set_data(sqe, user_data);
+  return IoUringResult::Ok;
+}
+
+IoUringResult IoUringImpl::prepareWritev(os_fd_t fd, const struct iovec* iovecs, unsigned nr_vecs,
+                                         off_t offset, void* user_data) {
+  struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
+  if (sqe == nullptr) {
+    return IoUringResult::Failed;
+  }
+
+  io_uring_prep_writev(sqe, fd, iovecs, nr_vecs, offset);
+  io_uring_sqe_set_data(sqe, user_data);
+  return IoUringResult::Ok;
+}
+
+IoUringResult IoUringImpl::prepareClose(os_fd_t fd, void* user_data) {
+  struct io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
+  if (sqe == nullptr) {
+    return IoUringResult::Failed;
+  }
+
+  io_uring_prep_close(sqe, fd);
+  io_uring_sqe_set_data(sqe, user_data);
+  return IoUringResult::Ok;
+}
+
+IoUringResult IoUringImpl::submit() {
+  int res = io_uring_submit(&ring_);
+  RELEASE_ASSERT(res >= 0 || res == -EBUSY, "unable to submit io_uring queue entries");
+  return res == -EBUSY ? IoUringResult::Busy : IoUringResult::Ok;
+}
+
+} // namespace Io
+} // namespace Envoy