Io uring lib

CMakeLists.txt

@@ -102,6 +102,7 @@ find_package(OpenSSL)
 # Check for IO faculties
 check_symbol_exists(epoll_create "sys/epoll.h" TS_USE_EPOLL)
 check_symbol_exists(kqueue "sys/event.h" TS_USE_KQUEUE)
+set(CMAKE_REQUIRED_LIBRARIES uring)
 check_symbol_exists(io_uring_queue_init "liburing.h" HAVE_IOURING)
 check_symbol_exists(getresuid unistd.h HAVE_GETRESUID)
 check_symbol_exists(getresgid unistd.h HAVE_GETRESGID)
@@ -111,9 +112,10 @@ check_symbol_exists(eventfd sys/eventfd.h HAVE_EVENTFD)
 check_symbol_exists(SSL_CTX_set_tlsext_ticket_key_cb openssl/ssl.h HAVE_SSL_CTX_SET_TLSEXT_TICKET_KEY_CB)
 
 option(USE_IOURING "Use experimental io_uring (linux only)" 0)
-if (HAVE_IOURING AND USE_IOUIRNG)
-    set(TS_USE_LINUX_IO_URING)
-endif(HAVE_IOURING AND USE_IOUIRNG)
+if (HAVE_IOURING AND USE_IOURING)
+    message(Using io_uring)
+    set(TS_USE_LINUX_IO_URING 1)
+endif(HAVE_IOURING AND USE_IOURING)
 
 # Check ssl functionality
 list(APPEND CMAKE_REQUIRED_INCLUDES ${OPENSSL_INCLUDE_DIR})

configure.ac

@@ -1686,6 +1686,7 @@ AC_ARG_ENABLE([experimental-linux-io-uring],
   [enable_linux_io_uring="${enableval}"],
   [enable_linux_io_uring=no]
 )
+AM_CONDITIONAL([ENABLE_IO_URING], [ test "x${enable_linux_io_uring}" = "xyes" ])
 
 AS_IF([test "x$enable_linux_io_uring" = "xyes"], [
   URING_LIBS="-luring"
@@ -2316,6 +2317,7 @@ AC_SUBST([default_stack_size], [$with_default_stack_size])
 #
 iocore_include_dirs="\
 -I\$(abs_top_srcdir)/iocore/eventsystem \
+-I\$(abs_top_srcdir)/iocore/io_uring \
 -I\$(abs_top_srcdir)/iocore/net \
 -I\$(abs_top_srcdir)/iocore/net/quic \
 -I\$(abs_top_srcdir)/iocore/aio \
@@ -2367,6 +2369,7 @@ AC_CONFIG_FILES([
   include/tscore/ink_config.h
   iocore/Makefile
   iocore/aio/Makefile
+  iocore/io_uring/Makefile
   iocore/cache/Makefile
   iocore/dns/Makefile
   iocore/eventsystem/Makefile

iocore/CMakeLists.txt

@@ -17,6 +17,7 @@
 
 
 add_subdirectory(eventsystem)
+add_subdirectory(io_uring)
 add_subdirectory(net)
 add_subdirectory(aio)
 add_subdirectory(dns)
@@ -27,6 +28,7 @@ add_subdirectory(cache)
 set(IOCORE_INCLUDE_DIRS
         ${CMAKE_SOURCE_DIR}/iocore/eventsystem
         ${CMAKE_SOURCE_DIR}/iocore/dns
+        ${CMAKE_SOURCE_DIR}/iocore/io_uring
         ${CMAKE_SOURCE_DIR}/iocore/aio
         ${CMAKE_SOURCE_DIR}/iocore/net
         ${CMAKE_SOURCE_DIR}/iocore/cache

iocore/Makefile.am

@@ -17,3 +17,6 @@
 #  limitations under the License.
 
 SUBDIRS = eventsystem net aio dns hostdb utils cache
+if ENABLE_IO_URING
+SUBDIRS += io_uring
+endif

iocore/aio/AIO.cc

@@ -73,12 +73,10 @@ RecInt aio_io_uring_wq_unbounded  = 0;
 RecRawStatBlock *aio_rsb      = nullptr;
 Continuation *aio_err_callbck = nullptr;
 // AIO Stats
-std::atomic<uint64_t> aio_num_read         = 0;
-std::atomic<uint64_t> aio_bytes_read       = 0;
-std::atomic<uint64_t> aio_num_write        = 0;
-std::atomic<uint64_t> aio_bytes_written    = 0;
-std::atomic<uint64_t> io_uring_submissions = 0;
-std::atomic<uint64_t> io_uring_completions = 0;
+std::atomic<uint64_t> aio_num_read      = 0;
+std::atomic<uint64_t> aio_bytes_read    = 0;
+std::atomic<uint64_t> aio_num_write     = 0;
+std::atomic<uint64_t> aio_bytes_written = 0;
 
 /*
  * Stats
@@ -534,87 +532,12 @@ AIOThreadInfo::aio_thread_main(AIOThreadInfo *thr_info)
 
 #elif AIO_MODE == AIO_MODE_IO_URING
 
-std::atomic<int> aio_main_wq_fd;
-
-DiskHandler::DiskHandler()
-{
-  io_uring_params p{};
-
-  if (aio_io_uring_attach_wq > 0) {
-    int wq_fd = get_main_queue_fd();
-    if (wq_fd > 0) {
-      p.flags = IORING_SETUP_ATTACH_WQ;
-      p.wq_fd = wq_fd;
-    }
-  }
-
-  if (aio_io_uring_sq_poll_ms > 0) {
-    p.flags          |= IORING_SETUP_SQPOLL;
-    p.sq_thread_idle = aio_io_uring_sq_poll_ms;
-  }
-
-  int ret = io_uring_queue_init_params(aio_io_uring_queue_entries, &ring, &p);
-  if (ret < 0) {
-    throw std::runtime_error(strerror(-ret));
-  }
-
-  /* no sharing for non-fixed either */
-  if (aio_io_uring_sq_poll_ms && !(p.features & IORING_FEAT_SQPOLL_NONFIXED)) {
-    throw std::runtime_error("No SQPOLL sharing with nonfixed");
-  }
-
-  // assign this handler to the thread
-  // TODO(cmcfarlen): maybe a bad place for this!
-  this_ethread()->diskHandler = this;
-}
-
-DiskHandler::~DiskHandler()
-{
-  io_uring_queue_exit(&ring);
-}
-
-void
-DiskHandler::set_main_queue(DiskHandler *dh)
-{
-  dh->set_wq_max_workers(aio_io_uring_wq_bounded, aio_io_uring_wq_unbounded);
-  aio_main_wq_fd.store(dh->ring.ring_fd);
-}
-
-int
-DiskHandler::get_main_queue_fd()
-{
-  return aio_main_wq_fd.load();
-}
-
-int
-DiskHandler::set_wq_max_workers(unsigned int bounded, unsigned int unbounded)
-{
-  if (bounded == 0 && unbounded == 0) {
-    return 0;
-  }
-  unsigned int args[2] = {bounded, unbounded};
-  int result           = io_uring_register_iowq_max_workers(&ring, args);
-  return result;
-}
-
-std::pair<int, int>
-DiskHandler::get_wq_max_workers()
-{
-  unsigned int args[2] = {0, 0};
-  io_uring_register_iowq_max_workers(&ring, args);
-  return std::make_pair(args[0], args[1]);
-}
-
-void
-DiskHandler::submit()
-{
-  io_uring_submissions.fetch_add(io_uring_submit(&ring));
-}
+#include "I_IO_URING.h"
 
 void
-DiskHandler::handle_cqe(io_uring_cqe *cqe)
+ink_aiocb::handle_complete(io_uring_cqe *cqe)
 {
-  AIOCallback *op = static_cast<AIOCallback *>(io_uring_cqe_get_data(cqe));
+  AIOCallback *op = this_op;
 
   op->aio_result = static_cast<int64_t>(cqe->res);
   op->link.prev  = nullptr;
@@ -643,68 +566,15 @@ DiskHandler::handle_cqe(io_uring_cqe *cqe)
   }
 }
 
-void
-DiskHandler::service()
-{
-  io_uring_cqe *cqe = nullptr;
-  io_uring_peek_cqe(&ring, &cqe);
-  while (cqe) {
-    handle_cqe(cqe);
-    io_uring_completions.fetch_add(1);
-    io_uring_cqe_seen(&ring, cqe);
-
-    cqe = nullptr;
-    io_uring_peek_cqe(&ring, &cqe);
-  }
-}
-
-void
-DiskHandler::submit_and_wait(int ms)
-{
-  ink_hrtime t              = ink_hrtime_from_msec(ms);
-  timespec ts               = ink_hrtime_to_timespec(t);
-  __kernel_timespec timeout = {ts.tv_sec, ts.tv_nsec};
-  io_uring_cqe *cqe         = nullptr;
-
-  io_uring_submit_and_wait_timeout(&ring, &cqe, 1, &timeout, nullptr);
-  while (cqe) {
-    handle_cqe(cqe);
-    io_uring_completions.fetch_add(1);
-    io_uring_cqe_seen(&ring, cqe);
-
-    cqe = nullptr;
-    io_uring_peek_cqe(&ring, &cqe);
-  }
-}
-
-int
-DiskHandler::register_eventfd()
-{
-  int fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
-
-  io_uring_register_eventfd(&ring, fd);
-
-  return fd;
-}
-
-DiskHandler *
-DiskHandler::local_context()
-{
-  // TODO(cmcfarlen): load config
-  thread_local DiskHandler threadContext;
-
-  return &threadContext;
-}
-
 int
 ink_aio_read(AIOCallback *op_in, int /* fromAPI ATS_UNUSED */)
 {
-  EThread *t      = this_ethread();
-  AIOCallback *op = op_in;
+  IOUringContext *ur = IOUringContext::local_context();
+  AIOCallback *op    = op_in;
   while (op) {
-    io_uring_sqe *sqe = t->diskHandler->next_sqe();
+    op->aiocb.this_op = op;
+    io_uring_sqe *sqe = ur->next_sqe(&op->aiocb);
     io_uring_prep_read(sqe, op->aiocb.aio_fildes, op->aiocb.aio_buf, op->aiocb.aio_nbytes, op->aiocb.aio_offset);
-    io_uring_sqe_set_data(sqe, op);
     op->aiocb.aio_lio_opcode = LIO_READ;
     if (op->then) {
       sqe->flags |= IOSQE_IO_LINK;
@@ -720,12 +590,12 @@ ink_aio_read(AIOCallback *op_in, int /* fromAPI ATS_UNUSED */)
 int
 ink_aio_write(AIOCallback *op_in, int /* fromAPI ATS_UNUSED */)
 {
-  EThread *t      = this_ethread();
-  AIOCallback *op = op_in;
+  IOUringContext *ur = IOUringContext::local_context();
+  AIOCallback *op    = op_in;
   while (op) {
-    io_uring_sqe *sqe = t->diskHandler->next_sqe();
+    op->aiocb.this_op = op;
+    io_uring_sqe *sqe = ur->next_sqe(&op->aiocb);
     io_uring_prep_write(sqe, op->aiocb.aio_fildes, op->aiocb.aio_buf, op->aiocb.aio_nbytes, op->aiocb.aio_offset);
-    io_uring_sqe_set_data(sqe, op);
     op->aiocb.aio_lio_opcode = LIO_WRITE;
     if (op->then) {
       sqe->flags |= IOSQE_IO_LINK;

iocore/aio/CMakeLists.txt

@@ -18,4 +18,4 @@
 
 add_library(aio)
 target_sources(aio PRIVATE AIO.cc Inline.cc)
-target_include_directories(aio PRIVATE ${CMAKE_SOURCE_DIR}/iocore/eventsystem)
+target_include_directories(aio PRIVATE ${CMAKE_SOURCE_DIR}/iocore/eventsystem ${CMAKE_SOURCE_DIR}/iocore/io_uring)

iocore/aio/I_AIO.h

@@ -68,18 +68,21 @@ typedef struct io_event ink_io_event_t;
 #define aio_buf    u.c.buf
 
 #elif AIO_MODE == AIO_MODE_IO_URING
-#include <liburing.h>
+#include "I_IO_URING.h"
 
 struct AIOCallback;
-struct ink_aiocb {
+struct ink_aiocb : public IOUringCompletionHandler {
   int aio_fildes    = -1;      /* file descriptor or status: AIO_NOT_IN_PROGRESS */
   void *aio_buf     = nullptr; /* buffer location */
   size_t aio_nbytes = 0;       /* length of transfer */
   off_t aio_offset  = 0;       /* file offset */
 
-  int aio_lio_opcode  = 0; /* listio operation */
-  int aio_state       = 0; /* state flag for List I/O */
-  AIOCallback *aio_op = nullptr;
+  int aio_lio_opcode   = 0; /* listio operation */
+  int aio_state        = 0; /* state flag for List I/O */
+  AIOCallback *this_op = nullptr;
+  AIOCallback *aio_op  = nullptr;
+
+  void handle_complete(io_uring_cqe *) override;
 };
 
 #else
@@ -153,41 +156,6 @@ struct DiskHandler : public Continuation {
 };
 #endif
 
-#if AIO_MODE == AIO_MODE_IO_URING
-
-class DiskHandler
-{
-public:
-  DiskHandler();
-  ~DiskHandler();
-
-  io_uring_sqe *
-  next_sqe()
-  {
-    return io_uring_get_sqe(&ring);
-  }
-
-  int set_wq_max_workers(unsigned int bounded, unsigned int unbounded);
-  std::pair<int, int> get_wq_max_workers();
-
-  void submit();
-  void service();
-  void submit_and_wait(int ms);
-
-  int register_eventfd();
-
-  static DiskHandler *local_context();
-  static void set_main_queue(DiskHandler *);
-  static int get_main_queue_fd();
-
-private:
-  io_uring ring;
-
-  void handle_cqe(io_uring_cqe *);
-};
-
-#endif
-
 void ink_aio_init(ts::ModuleVersion version);
 int ink_aio_start();
 void ink_aio_set_callback(Continuation *error_callback);

iocore/aio/Makefile.am

@@ -18,6 +18,7 @@
 
 AM_CPPFLAGS += \
 	-I$(abs_top_srcdir)/iocore/eventsystem \
+	-I$(abs_top_srcdir)/iocore/io_uring \
 	-I$(abs_top_srcdir)/include \
 	-I$(abs_top_srcdir)/lib \
 	@SWOC_INCLUDES@

iocore/cache/CMakeLists.txt

@@ -34,6 +34,7 @@ add_library(inkcache STATIC
 )
 target_include_directories(inkcache PRIVATE
         ${CMAKE_SOURCE_DIR}/iocore/eventsystem
+        ${CMAKE_SOURCE_DIR}/iocore/io_uring
         ${CMAKE_SOURCE_DIR}/iocore/dns
         ${CMAKE_SOURCE_DIR}/iocore/aio
         ${CMAKE_SOURCE_DIR}/iocore/net

iocore/dns/CMakeLists.txt

@@ -25,6 +25,7 @@ add_library(inkdns STATIC
 target_include_directories(inkdns PRIVATE
         ${CMAKE_SOURCE_DIR}/iocore/eventsystem
         ${CMAKE_SOURCE_DIR}/iocore/dns
+        ${CMAKE_SOURCE_DIR}/iocore/io_uring
         ${CMAKE_SOURCE_DIR}/iocore/aio
         ${CMAKE_SOURCE_DIR}/iocore/net
         ${CMAKE_SOURCE_DIR}/iocore/cache

iocore/hostdb/CMakeLists.txt

@@ -25,6 +25,7 @@ add_library(inkhostdb STATIC
 )
 target_include_directories(inkhostdb PRIVATE
         ${CMAKE_SOURCE_DIR}/iocore/eventsystem
+        ${CMAKE_SOURCE_DIR}/iocore/io_uring
         ${CMAKE_SOURCE_DIR}/iocore/dns
         ${CMAKE_SOURCE_DIR}/iocore/aio
         ${CMAKE_SOURCE_DIR}/iocore/net