[IO_URING] Change submit_and_wait to take ink_hrtime. Fix test_AIO for io_uring.

CMakeLists.txt

@@ -183,3 +183,4 @@ add_subdirectory(mgmt/utils)
 add_subdirectory(mgmt/config)
 add_subdirectory(mgmt/rpc)
 add_subdirectory(src/traffic_server)
+add_subdirectory(src/tests)

iocore/aio/CMakeLists.txt

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

iocore/aio/sample.cfg

@@ -14,4 +14,6 @@ write_skip 5
 chains 1
 delete_disks 1
 disk_path ./aio.tst
+io_uring_queue_entries 32
+num_processors 1
 

iocore/aio/test_AIO.cc

@@ -71,6 +71,14 @@ int threads_per_disk = 1;
 int delete_disks     = 0;
 int max_size         = 0;
 int use_lseek        = 0;
+int num_processors   = 0;
+#if AIO_MODE == AIO_MODE_IO_URING
+int io_uring_queue_entries = 32;
+int io_uring_sq_poll_ms    = 0;
+int io_uring_attach_wq     = 0;
+int io_uring_wq_bounded    = 0;
+int io_uring_wq_unbounded  = 0;
+#endif
 
 int chains                    = 1;
 double seq_read_percent       = 0.0;
@@ -210,6 +218,14 @@ dump_summary()
   printf("%0.2f total mbytes/sec\n", sr + sw + rr);
   printf("----------------------------------------------------------\n");
 
+#if AIO_MODE == AIO_MODE_IO_URING
+  printf("-----------------\n");
+  printf("IO_URING results\n");
+  printf("-----------------\n");
+  printf("submissions: %lu\n", io_uring_submissions.load());
+  printf("completions: %lu\n", io_uring_completions.load());
+#endif
+
   if (delete_disks) {
     for (int i = 0; i < n_disk_path; i++) {
       unlink(disk_path[i]);
@@ -375,6 +391,14 @@ read_config(const char *config_filename)
     PARAM(chains)
     PARAM(threads_per_disk)
     PARAM(delete_disks)
+    PARAM(num_processors)
+#if AIO_MODE == AIO_MODE_IO_URING
+    PARAM(io_uring_queue_entries)
+    PARAM(io_uring_sq_poll_ms)
+    PARAM(io_uring_attach_wq)
+    PARAM(io_uring_wq_bounded)
+    PARAM(io_uring_wq_unbounded)
+#endif
     else if (strcmp(field_name, "disk_path") == 0)
     {
       assert(n_disk_path < MAX_DISK_THREADS);
@@ -405,16 +429,64 @@ read_config(const char *config_filename)
   return (1);
 }
 
+#if AIO_MODE == AIO_MODE_IO_URING
+
+class IOUringLoopTailHandler : public EThread::LoopTailHandler
+{
+public:
+  int
+  waitForActivity(ink_hrtime timeout) override
+  {
+    IOUringContext::local_context()->submit_and_wait(timeout);
+
+    return 0;
+  }
+  /** Unblock.
+
+  This is required to unblock (wake up) the block created by calling @a cb.
+      */
+  void
+  signalActivity() override
+  {
+  }
+
+  ~IOUringLoopTailHandler() override {}
+} uring_handler;
+
+#endif
+
 int
 main(int /* argc ATS_UNUSED */, char *argv[])
 {
   int i;
+  printf("input file %s\n", argv[1]);
+  if (!read_config(argv[1])) {
+    exit(1);
+  }
+  if (num_processors == 0) {
+    num_processors = ink_number_of_processors();
+  }
+  printf("Using %d processor threads\n", num_processors);
+
+#if AIO_MODE == AIO_MODE_IO_URING
+  {
+    IOUringConfig cfg;
+
+    cfg.queue_entries = io_uring_queue_entries;
+    cfg.sq_poll_ms    = io_uring_sq_poll_ms;
+    cfg.attach_wq     = io_uring_attach_wq;
+    cfg.wq_bounded    = io_uring_wq_bounded;
+    cfg.wq_unbounded  = io_uring_wq_unbounded;
+
+    IOUringContext::set_config(cfg);
+  };
+#endif
 
   Layout::create();
   init_diags("", nullptr);
   RecProcessInit();
   ink_event_system_init(EVENT_SYSTEM_MODULE_PUBLIC_VERSION);
-  eventProcessor.start(ink_number_of_processors());
+  eventProcessor.start(num_processors);
 
   Thread *main_thread = new EThread;
   main_thread->set_specific();
@@ -425,14 +497,15 @@ main(int /* argc ATS_UNUSED */, char *argv[])
     et->schedule_imm(et->diskHandler);
   }
 #endif
+#if AIO_MODE == AIO_MODE_IO_URING
+  for (EThread *et : eventProcessor.active_group_threads(ET_NET)) {
+    et->set_tail_handler(&uring_handler);
+  }
+#endif
 
   RecProcessStart();
   ink_aio_init(AIO_MODULE_PUBLIC_VERSION);
   ts::Random::seed(time(nullptr));
-  printf("input file %s\n", argv[1]);
-  if (!read_config(argv[1])) {
-    exit(1);
-  }
 
   max_size = seq_read_size;
   if (seq_write_size > max_size) {
@@ -466,7 +539,12 @@ main(int /* argc ATS_UNUSED */, char *argv[])
   }
 
   while (!TSSystemState::is_event_system_shut_down()) {
+#if AIO_MODE == AIO_MODE_IO_URING
+    IOUringContext::local_context()->submit_and_wait(1 * HRTIME_SECOND);
+#else
     sleep(1);
+#endif
   }
+
   delete main_thread;
 }

iocore/io_uring/I_IO_URING.h

@@ -25,6 +25,7 @@ Linux io_uring helper library
 
 #include <liburing.h>
 #include <utility>
+#include "tscore/ink_hrtime.h"
 
 struct IOUringConfig {
   int queue_entries = 1024;
@@ -63,7 +64,7 @@ class IOUringContext
 
   void submit();
   void service();
-  void submit_and_wait(int ms);
+  void submit_and_wait(ink_hrtime ms);
 
   int register_eventfd();
 

iocore/io_uring/io_uring.cc

@@ -26,6 +26,8 @@ Linux io_uring helper library
 #include <cstring>
 #include <stdexcept>
 
+#include <unistd.h>
+
 #include "I_IO_URING.h"
 #include "tscore/ink_hrtime.h"
 
@@ -76,7 +78,7 @@ IOUringContext::IOUringContext()
 IOUringContext::~IOUringContext()
 {
   if (evfd != -1) {
-    close(evfd);
+    ::close(evfd);
     evfd = -1;
   }
   io_uring_queue_exit(&ring);
@@ -144,14 +146,14 @@ IOUringContext::service()
 }
 
 void
-IOUringContext::submit_and_wait(int ms)
+IOUringContext::submit_and_wait(ink_hrtime t)
 {
-  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);
+  int count = io_uring_submit_and_wait_timeout(&ring, &cqe, 1, &timeout, nullptr);
+  io_uring_submissions.fetch_add(count);
   while (cqe) {
     handle_cqe(cqe);
     io_uring_completions++;

iocore/io_uring/unit_tests/test_diskIO.cc

@@ -32,6 +32,7 @@
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
+#include "tscore/ink_hrtime.h"
 
 ts::file::path
 temp_prefix(const char *basename)
@@ -137,13 +138,13 @@ TEST_CASE("disk_io", "[io_uring]")
   REQUIRE(fd != -1);
 
   io_uring_write(ctx, fd, "hello", 5, [](int result) { REQUIRE(result == 5); });
-  ctx.submit_and_wait(100);
+  ctx.submit_and_wait(100 * HRTIME_MSECOND);
   io_uring_close(ctx, fd, [&fd](int result) {
     REQUIRE(result == 0);
     fd = -1;
   });
 
-  ctx.submit_and_wait(100);
+  ctx.submit_and_wait(100 * HRTIME_MSECOND);
 
   REQUIRE(fd == -1);
 
@@ -156,7 +157,7 @@ TEST_CASE("disk_io", "[io_uring]")
     REQUIRE("hello"sv == std::string_view(buffer, result));
   });
 
-  ctx.submit_and_wait(100);
+  ctx.submit_and_wait(100 * HRTIME_MSECOND);
 }
 
 void
@@ -263,7 +264,7 @@ TEST_CASE("net_io", "[io_uring]")
   uint64_t completions_before = io_uring_completions;
   uint64_t needed             = 2;
   while ((io_uring_completions - completions_before) < needed) {
-    ctx.submit_and_wait(1000);
+    ctx.submit_and_wait(1 * HRTIME_SECOND);
   }
 
   REQUIRE(server.clients == 1);

src/tests/CMakeLists.txt

@@ -0,0 +1,29 @@
+#######################
+#
+#  Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+#  agreements.  See the NOTICE file distributed with this work for additional information regarding
+#  copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
+#  (the "License"); you may not use this file except in compliance with the License.  You may obtain
+#  a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software distributed under the License
+#  is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+#  or implied. See the License for the specific language governing permissions and limitations under
+#  the License.
+#
+#######################
+
+# This is for tests that depend on multiple libraries where dependency order matters.
+
+add_executable(test_AIO ${CMAKE_SOURCE_DIR}/iocore/aio/test_AIO.cc)
+target_include_directories(test_AIO PRIVATE
+        ${IOCORE_INCLUDE_DIRS}
+        ${PROXY_INCLUDE_DIRS}
+        ${CMAKE_SOURCE_DIR}/mgmt
+        )
+target_link_libraries(test_AIO aio inkevent proxy tscore tscpputil)
+if (TS_USE_LINUX_IO_URING)
+    target_link_libraries(test_AIO inkuring uring)
+endif (TS_USE_LINUX_IO_URING)

src/traffic_server/CMakeLists.txt

@@ -25,7 +25,8 @@ add_executable(traffic_server
         SocksProxy.cc
         traffic_server.cc
         RpcAdminPubHandlers.cc
-        ${CMAKE_SOURCE_DIR}/src/shared/overridable_txn_vars.cc)
+        ${CMAKE_SOURCE_DIR}/src/shared/overridable_txn_vars.cc
+        )
 target_include_directories(traffic_server PRIVATE
         ${IOCORE_INCLUDE_DIRS}
         ${PROXY_INCLUDE_DIRS}

src/traffic_server/traffic_server.cc

@@ -2247,7 +2247,7 @@ main(int /* argc ATS_UNUSED */, const char **argv)
 
   while (!TSSystemState::is_event_system_shut_down()) {
 #if TS_USE_LINUX_IO_URING == 1
-    ur->submit_and_wait(1000);
+    ur->submit_and_wait(1 * HRTIME_SECOND);
 #else
     sleep(1);
 #endif

src/traffic_via/test_traffic_via

@@ -26,12 +26,12 @@ else
     srcdir=$(pwd)
 fi
 
-find $srcdir/tests -type f |
+find $srcdir/traffic_via/tests -type f |
 while read f ; do
     name=$(basename "$f")
     echo "testing $name"
-    ./traffic_via "$name" > "$tmpfile" 2>&1 || true
-    diff -u "$tmpfile" "$srcdir/tests/$name"
+    ./traffic_via/traffic_via "$name" > "$tmpfile" 2>&1 || true
+    diff -u "$tmpfile" "$srcdir/traffic_via/tests/$name"
 done
 
 rm -f $tmpfile