diff --git a/iocore/io_uring/I_IO_URING.h b/iocore/io_uring/I_IO_URING.h index a4068a117d3..fe8c8ca7a0b 100644 --- a/iocore/io_uring/I_IO_URING.h +++ b/iocore/io_uring/I_IO_URING.h @@ -46,6 +46,8 @@ class IOUringContext IOUringContext(); ~IOUringContext(); + IOUringContext(const IOUringContext &) = delete; + io_uring_sqe * next_sqe(IOUringCompletionHandler *handler) { @@ -72,7 +74,8 @@ class IOUringContext static int get_main_queue_fd(); private: - io_uring ring; + io_uring ring = {}; + int evfd = -1; void handle_cqe(io_uring_cqe *); static IOUringConfig config; diff --git a/iocore/io_uring/Makefile.am b/iocore/io_uring/Makefile.am index ba6474fc245..1434fd6ee45 100644 --- a/iocore/io_uring/Makefile.am +++ b/iocore/io_uring/Makefile.am @@ -28,3 +28,26 @@ libinkuring_a_SOURCES = \ io_uring.cc \ I_IO_URING.h \ P_IO_URING.h + + +check_PROGRAMS = test_diskIO + +test_LD_FLAGS = \ + @AM_LDFLAGS@ \ + @OPENSSL_LDFLAGS@ + +test_CPP_FLAGS = \ + $(AM_CPPFLAGS) \ + $(iocore_include_dirs) \ + -I$(abs_top_srcdir)/tests/include + +test_LD_ADD = \ + $(top_builddir)/src/tscore/libtscore.la \ + $(top_builddir)/iocore/io_uring/libinkuring.a \ + @HWLOC_LIBS@ + + +test_diskIO_SOURCES = unit_tests/test_diskIO.cc +test_diskIO_CPPFLAGS = $(test_CPP_FLAGS) +test_diskIO_LDFLAGS = $(test_LD_FLAGS) +test_diskIO_LDADD = $(test_LD_ADD) diff --git a/iocore/io_uring/io_uring.cc b/iocore/io_uring/io_uring.cc index bf9c075414f..717351df14b 100644 --- a/iocore/io_uring/io_uring.cc +++ b/iocore/io_uring/io_uring.cc @@ -35,6 +35,12 @@ std::atomic io_uring_completions = 0; IOUringConfig IOUringContext::config; +void +IOUringContext::set_config(const IOUringConfig &cfg) +{ + config = cfg; +} + IOUringContext::IOUringContext() { io_uring_params p{}; @@ -69,6 +75,10 @@ IOUringContext::IOUringContext() IOUringContext::~IOUringContext() { + if (evfd != -1) { + close(evfd); + evfd = -1; + } io_uring_queue_exit(&ring); } @@ -155,11 +165,12 @@ IOUringContext::submit_and_wait(int ms) int IOUringContext::register_eventfd() { - int fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); - - io_uring_register_eventfd(&ring, fd); + if (evfd == -1) { + evfd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); - return fd; + io_uring_register_eventfd(&ring, evfd); + } + return evfd; } IOUringContext * diff --git a/iocore/io_uring/unit_tests/test_diskIO.cc b/iocore/io_uring/unit_tests/test_diskIO.cc new file mode 100644 index 00000000000..9047460f79c --- /dev/null +++ b/iocore/io_uring/unit_tests/test_diskIO.cc @@ -0,0 +1,266 @@ +/** @file + + Catch based unit tests for EventSystem + + @section license License + + 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. + */ +#define CATCH_CONFIG_MAIN +#include "catch.hpp" + +#include "I_IO_URING.h" +#include "tscore/ts_file.h" + +#include + +#include +#include +#include + +ts::file::path +temp_prefix(const char *basename) +{ + char buffer[PATH_MAX]; + std::error_code err; + auto tmpdir = ts::file::temp_directory_path(); + snprintf(buffer, sizeof(buffer), "%s/%s.XXXXXX", tmpdir.c_str(), basename); + auto prefix = ts::file::path(mkdtemp(buffer)); + bool result = ts::file::create_directories(prefix, err, 0755); + if (!result) { + throw std::runtime_error("Failed to create directory"); + } + ink_assert(result); + + return prefix; +} + +int +open_path(const ts::file::path &path, int oflags = O_CREAT | O_RDWR, int mode = 0644) +{ + return open(path.c_str(), oflags, mode); +} + +template class FunctionHolderHandler : public IOUringCompletionHandler +{ +public: + FunctionHolderHandler(F &&f) : f(std::move(f)) {} + + void + handle_complete(io_uring_cqe *c) override + { + f(c->res); + } + +private: + F f; +}; + +IOUringCompletionHandler * +handle(const std::function &f) +{ + return new FunctionHolderHandler(std::move(f)); +} + +void +io_uring_write(IOUringContext &ur, int fd, const char *data, size_t len, const std::function &f) +{ + io_uring_sqe *s = ur.next_sqe(handle(f)); + + io_uring_prep_write(s, fd, data, len, 0); +} +void +io_uring_read(IOUringContext &ur, int fd, char *data, size_t len, const std::function &f) +{ + io_uring_sqe *s = ur.next_sqe(handle(f)); + + io_uring_prep_read(s, fd, data, len, 0); +} + +void +io_uring_close(IOUringContext &ur, int fd, const std::function &f) +{ + io_uring_sqe *s = ur.next_sqe(handle(f)); + + io_uring_prep_close(s, fd); +} + +void +io_uring_accept(IOUringContext &ur, int sock, sockaddr *addr, socklen_t *addrlen, const std::function &f) +{ + io_uring_sqe *s = ur.next_sqe(handle(f)); + + io_uring_prep_accept(s, sock, addr, addrlen, 0); +} + +void +io_uring_connect(IOUringContext &ur, int sock, sockaddr *addr, socklen_t addrlen, const std::function &f) +{ + io_uring_sqe *s = ur.next_sqe(handle(f)); + + io_uring_prep_connect(s, sock, addr, addrlen); +} + +TEST_CASE("disk_io", "[io_uring]") +{ + IOUringConfig cfg = { + .queue_entries = 32, + }; + IOUringContext::set_config(cfg); + IOUringContext ctx; + + auto tmp = temp_prefix("disk_io"); + + REQUIRE(ts::file::exists(tmp)); + + auto apath = tmp / "a"; + + int fd = open_path(apath); + + std::printf("%s\n", apath.c_str()); + + REQUIRE(fd != -1); + + io_uring_write(ctx, fd, "hello", 5, [](int result) { REQUIRE(result == 5); }); + ctx.submit_and_wait(100); + io_uring_close(ctx, fd, [&fd](int result) { + REQUIRE(result == 0); + fd = -1; + }); + + ctx.submit_and_wait(100); + + REQUIRE(fd == -1); + + fd = open_path(apath, O_RDONLY); + char buffer[6] = {0}; + io_uring_read(ctx, fd, buffer, sizeof(buffer), [&](int result) { + using namespace std::literals; + + REQUIRE(result == 5); + REQUIRE("hello"sv == std::string_view(buffer, result)); + }); + + ctx.submit_and_wait(100); +} + +void +set_reuseport(int s) +{ + int optval = 1; + setsockopt(s, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval)); +} + +void +set_reuseaddr(int s) +{ + int optval = 1; + setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)); +} + +sockaddr_in +any_addr(int port) +{ + return {.sin_family = AF_INET, .sin_port = htons(port), .sin_addr = {.s_addr = htonl(INADDR_ANY)}}; +} + +sockaddr_in +make_addr(const std::string &ip, int port) +{ + sockaddr_in addr = {}; + + ::inet_aton(ip.c_str(), &addr.sin_addr); + addr.sin_port = htons(port); + addr.sin_family = AF_INET; + + return addr; +} + +int +make_listen_socket(int port) +{ + int s = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + set_reuseaddr(s); + set_reuseport(s); + + const sockaddr_in addr = any_addr(port); + int rc = ::bind(s, reinterpret_cast(&addr), sizeof(addr)); + if (rc == -1) { + throw std::runtime_error("failed to bind"); + } + rc = ::listen(s, 10000); + if (rc == -1) { + throw std::runtime_error("failed to listen"); + } + + return s; +} + +int +make_client_socket() +{ + int s = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + return s; +} + +class SimpleTestServer +{ +public: + SimpleTestServer(int port) : s(make_listen_socket(port)), port(port), clients(0) {} + + void + start(IOUringContext &ctx) + { + client_len = sizeof(client); + io_uring_accept(ctx, s, reinterpret_cast(&client), &client_len, [&](int result) { + REQUIRE(result > 0); + clients++; + }); + } + + int s; + int port; + int clients; + sockaddr_in client; + socklen_t client_len; +}; + +TEST_CASE("net_io", "[io_uring]") +{ + IOUringConfig cfg = { + .queue_entries = 32, + }; + IOUringContext::set_config(cfg); + IOUringContext ctx; + + SimpleTestServer server(4321); + + server.start(ctx); + + auto client_addr = make_addr("127.0.0.1", 4321); + int s = make_client_socket(); + bool connected = false; + io_uring_connect(ctx, s, reinterpret_cast(&client_addr), sizeof(client_addr), [&](int result) { + REQUIRE(result == 0); + connected = true; + }); + + ctx.submit_and_wait(1000); + + REQUIRE(server.clients == 1); + REQUIRE(connected); +} \ No newline at end of file diff --git a/iocore/net/IOUringNet.cc b/iocore/net/IOUringNet.cc index 529a6bb85f1..c5598d142ee 100644 --- a/iocore/net/IOUringNet.cc +++ b/iocore/net/IOUringNet.cc @@ -24,6 +24,7 @@ #include "P_IOUringNetVConnection.h" #include "tscore/Diags.h" #include "tscore/ink_assert.h" +#include "I_IO_URING.h" #include #include @@ -60,10 +61,12 @@ void IOUringNetHandler::signalActivity() { static uint64_t counter = 1; - auto ret = write(thread->evfd, &counter, sizeof counter); - Debug(TAG, "signalActivity on fd=%d returned %zu", thread->evfd, ret); - if (ret < 0) { - Warning("io_uring: failed to write to event fd"); + if (thread != nullptr && thread->evfd != ts::NO_FD) { + auto ret = write(thread->evfd, &counter, sizeof counter); + Debug(TAG, "signalActivity on fd=%d returned %zu", thread->evfd, ret); + if (ret < 0) { + Warning("io_uring: failed to write to event fd"); + } } } @@ -84,36 +87,9 @@ prep_eventfd_read(struct io_uring *ring, EThread *thread) int IOUringNetHandler::waitForActivity(ink_hrtime timeout) { - struct io_uring_cqe *cqe; - ink_release_assert(this_thread() == thread); - - // In the case when the work queue is empty, an incoming event should - // send a notification to the ring via eventfd to wake the thread up - // from this sleep. - Debug(TAG, "waiting for cqe"); - auto ret = io_uring_submit(&ring); - if (ret < 0) { - Warning("Failed to io_uring_submit: %s", strerror(-ret)); - } - ret = io_uring_wait_cqe(&ring, &cqe); - if (ret < 0) { - Warning("io_uring_wait_cqe failed: %s", strerror(-ret)); - } else { - // pending--; - // TODO: handle completed I/O - // TODO: break here if new events are scheduled by I/O completion handlers - Debug(TAG, "processing cqes"); - while (io_uring_peek_cqe(&ring, &cqe) == 0) { - Debug(TAG, "cqe->res = %d", cqe->res); - if (io_uring_cqe_get_data64(cqe) == 0) { - // eventfd - Debug(TAG, "Read from eventfd:"); - prep_eventfd_read(&ring, thread); - } - io_uring_cqe_seen(&ring, cqe); - } - } + IOUringContext *ur = IOUringContext::local_context(); + ur->submit_and_wait(timeout); return 0; } @@ -124,19 +100,5 @@ initialize_thread_for_iouring(EThread *thread) inh.emplace(); ink_release_assert(inh); - // TODO: replace placeholder values - const int queue_depth = 1024; - io_uring_params p{}; - - Debug(TAG, "Started io_uring thread"); - - auto ret = io_uring_queue_init_params(queue_depth, &inh->ring, &p); - if (ret < 0) { - Fatal("Failed to initialize io_uring: %s", strerror(-ret)); - return; - } - inh->thread = thread; - prep_eventfd_read(&inh->ring, thread); - thread->set_tail_handler(&inh.value()); } diff --git a/iocore/net/IOUringNetAccept.cc b/iocore/net/IOUringNetAccept.cc index 816268dbfca..ae290ff3222 100644 --- a/iocore/net/IOUringNetAccept.cc +++ b/iocore/net/IOUringNetAccept.cc @@ -28,17 +28,14 @@ #include #include #include +#include "I_IO_URING.h" static constexpr auto TAG = "io_uring_accept"; // Number of sq entries for accept threads static constexpr int queue_depth = 1; -// Ring for accept threads -static thread_local std::optional ring; -// Connection buffers for accept thread -static thread_local std::vector cons{queue_depth}; -IOUringNetAccept::IOUringNetAccept(NetProcessor::AcceptOptions const &opt) : NetAccept(opt) {} +IOUringNetAccept::IOUringNetAccept(NetProcessor::AcceptOptions const &opt) : NetAccept(opt), connections(queue_depth) {} void IOUringNetAccept::init_accept_loop() @@ -74,22 +71,6 @@ IOUringNetAccept::init_accept_loop() } } -static void -queue_accept(int fd, int idx) -{ - io_uring_sqe *accept_sqe = nullptr; - Debug(TAG, "queueing accept: fd = %d, idx = %d", fd, idx); - - ink_release_assert(idx < queue_depth); - auto &con = cons[idx]; - con.addrlen = sizeof(con.addr); - - accept_sqe = io_uring_get_sqe(&ring.value()); - ink_release_assert(accept_sqe != nullptr); - io_uring_prep_accept(accept_sqe, fd, &con.addr.sa, &con.addrlen, SOCK_CLOEXEC); - io_uring_sqe_set_data64(accept_sqe, idx); -} - void IOUringNetAccept::safe_delay(int msec) { @@ -99,66 +80,55 @@ IOUringNetAccept::safe_delay(int msec) int IOUringNetAccept::acceptLoopEvent(int event, void *ep) { - int ret = 0; - EThread *t = this_ethread(); - bool stop = false; - - ink_release_assert(!ring); - // Create a new iouring - ring.emplace(); - io_uring_params p{}; - - ink_release_assert(ring); - p.cq_entries = queue_depth; - ret = io_uring_queue_init_params(queue_depth, &ring.value(), &p); - if (ret < 0) { - Fatal("Failed to initialize io_uring for accept: %s", strerror(-ret)); - return EVENT_ERROR; - } + IOUringContext *ctx = IOUringContext::local_context(); - // TODO: use multishot when available - // TODO: use direct for more efficiency - for (int i = 0; i < queue_depth; i++) { - queue_accept(server.fd, i); + for (auto &con : connections) { + con.prep_accept(ctx, this); } + // setup eventfd for activity? + do { - ret = io_uring_submit(&ring.value()); - if (ret < 0) { - Fatal("accept: io_uring_submit failed: %s", strerror(-ret)); - return EVENT_ERROR; - } - io_uring_cqe *pcqe = nullptr; - io_uring_cqe cqe; - ret = io_uring_wait_cqe(&ring.value(), &pcqe); - if (ret == -EINTR) { - Debug(TAG, "accept: iouring_wait_cqe interrupted by signal."); - continue; - } else if (ret < 0) { - Fatal("accept: io_uring_wait_cqe failed: %s", strerror(-ret)); - return EVENT_ERROR; - } - ink_assert(pcqe); - cqe = *pcqe; - io_uring_cqe_seen(&ring.value(), pcqe); - auto idx = io_uring_cqe_get_data64(pcqe); - Connection &con = cons[idx]; - - char buf[INET6_ADDRSTRLEN]; - ats_ip_ntop(con.addr, buf, sizeof buf); - Debug(TAG, "Accepted a connection %s:%u.", buf, con.addr.host_order_port()); - stop = process_accept(cqe.res, t, con); - queue_accept(server.fd, idx); - } while (!stop); + ctx->submit_and_wait(1000); + } while (!TSSystemState::is_event_system_shut_down()); Warning("Accept loop stopped!"); - ink_release_assert(ring); return EVENT_DONE; } +int +IOUringNetAccept::accept_startup(int, void *) +{ + IOUringContext *ctx = IOUringContext::local_context(); + if (do_listen(BLOCKING)) { + return 1; + } + + for (auto &con : connections) { + con.prep_accept(ctx, this); + } + return 0; +} + void IOUringNetAccept::init_accept_per_thread() { + int i, n; + + SET_HANDLER(&IOUringNetAccept::accept_startup); + n = eventProcessor.thread_group[opt.etype]._count; + + for (i = 0; i < n; i++) { + IOUringNetAccept *na; + na = new IOUringNetAccept(opt); + *na = *this; + SET_CONTINUATION_HANDLER(na, &IOUringNetAccept::accept_startup); + + EThread *t = eventProcessor.thread_group[opt.etype]._thread[i]; + // shouldnt need a mutex for + // a->mutex = get_NetHandler(t)->mutex; + t->schedule_imm(na); + } } int @@ -195,3 +165,39 @@ IOUringNetAccept::initialize_vc(NetVConnection *_vc, Connection &con, EThread *l // TODO: Does vc need the mutex from its NetProcessor? } +void +IOUringNetAccept::handle_complete(io_uring_cqe *sqe) +{ +} + +void +IOUringAcceptConnection::handle_complete(io_uring_cqe *cqe) +{ + IOUringContext *ctx = IOUringContext::local_context(); + + char buf[INET6_ADDRSTRLEN]; + ats_ip_ntop(conn.addr, buf, sizeof buf); + Debug(TAG, "Accepted a connection %s:%u with fd %d.", buf, conn.addr.host_order_port(), conn.fd); + + auto stop = na->process_accept(cqe->res, this_ethread(), conn); + + if (!stop) { + auto *sqe = ctx->next_sqe(this); + ink_release_assert(sqe != nullptr); + conn.addrlen = sizeof(conn.addr); + io_uring_prep_accept(sqe, na->server.fd, &conn.addr.sa, &conn.addrlen, SOCK_CLOEXEC); + } +} + +void +IOUringAcceptConnection::prep_accept(IOUringContext *ctx, NetAccept *pna) +{ + conn.addrlen = sizeof(conn.addr); + na = pna; + auto *sqe = ctx->next_sqe(this); + + ink_release_assert(sqe != nullptr); + // TODO: use multishot when available + // TODO: use direct for more efficiency + io_uring_prep_accept(sqe, na->server.fd, &conn.addr.sa, &conn.addrlen, SOCK_CLOEXEC); +} diff --git a/iocore/net/IOUringNetAccept.h b/iocore/net/IOUringNetAccept.h index 3c3b851182b..e68b1a4b41c 100644 --- a/iocore/net/IOUringNetAccept.h +++ b/iocore/net/IOUringNetAccept.h @@ -27,8 +27,36 @@ #include "P_IOUringNetProcessor.h" #include "I_EThread.h" #include "liburing.h" +#include "I_IO_URING.h" -class IOUringNetAccept : public NetAccept +class IOUringAcceptConnection : public IOUringCompletionHandler +{ +public: + IOUringAcceptConnection() = default; + IOUringAcceptConnection(const IOUringAcceptConnection &other) + { + conn.addr = other.conn.addr; + conn.addrlen = other.conn.addrlen; + } + virtual ~IOUringAcceptConnection() = default; + + void handle_complete(io_uring_cqe *) override; + + IOUringAcceptConnection & + operator=(const IOUringAcceptConnection &other) + { + conn.addr = other.conn.addr; + conn.addrlen = other.conn.addrlen; + return *this; + } + + void prep_accept(IOUringContext *ctx, NetAccept *pna); + + Connection conn; + NetAccept *na; +}; + +class IOUringNetAccept : public NetAccept, public IOUringCompletionHandler { public: IOUringNetAccept(NetProcessor::AcceptOptions const &opt); @@ -41,6 +69,10 @@ class IOUringNetAccept : public NetAccept return &ioUringNetProcessor; } + int accept_startup(int, void *); + + void handle_complete(io_uring_cqe *) override; + protected: void safe_delay(int msec) override; void initialize_vc(NetVConnection *_vc, Connection &con, EThread *localt) override; @@ -51,4 +83,6 @@ class IOUringNetAccept : public NetAccept // Handler for per-thread accepts int acceptEvent(int event, void *ep) override; + + std::vector connections; }; diff --git a/iocore/net/P_EventIO.h b/iocore/net/P_EventIO.h index 79eb984cb97..fdc58acaa42 100644 --- a/iocore/net/P_EventIO.h +++ b/iocore/net/P_EventIO.h @@ -86,7 +86,7 @@ struct EventIO { EVENTIO_DNS_CONNECTION = 3, EVENTIO_UDP_CONNECTION = 4, EVENTIO_ASYNC_SIGNAL = 5, - EVENTIO_DISK = 6 + EVENTIO_IO_URING = 6 }; /** The start methods all logically Setup a class to be called diff --git a/iocore/net/P_NetAccept.h b/iocore/net/P_NetAccept.h index c3256e4e5d9..53f15619751 100644 --- a/iocore/net/P_NetAccept.h +++ b/iocore/net/P_NetAccept.h @@ -132,9 +132,11 @@ struct NetAccept : public Continuation, EventIOUser { return server.close(); } + // TODO(cmcfarlen): I moved this from protected + bool process_accept(int res, EThread *, Connection &con); + protected: virtual int do_listen(bool non_blocking); - bool process_accept(int res, EThread *, Connection &con); virtual void safe_delay(int msec); virtual void initialize_vc(NetVConnection *_vc, Connection &con, EThread *localt); }; diff --git a/iocore/net/P_UnixNet.h b/iocore/net/P_UnixNet.h index d3db50704b8..0395e76fc71 100644 --- a/iocore/net/P_UnixNet.h +++ b/iocore/net/P_UnixNet.h @@ -169,6 +169,10 @@ class NetHandler : public Continuation, public EThread::LoopTailHandler Que(NetEvent, active_queue_link) active_queue; uint32_t active_queue_size = 0; +#ifdef TS_USE_LINUX_IO_URING + EventIO uring_evio; +#endif + /// configuration settings for managing the active and keep-alive queues struct Config { uint32_t max_connections_in = 0; diff --git a/iocore/net/UnixNet.cc b/iocore/net/UnixNet.cc index 565c8f27d42..6da55b7e40b 100644 --- a/iocore/net/UnixNet.cc +++ b/iocore/net/UnixNet.cc @@ -259,13 +259,14 @@ initialize_thread_for_net(EThread *thread) thread->ep->type = EventIO::EVENTIO_ASYNC_SIGNAL; #if HAVE_EVENTFD thread->ep->start(pd, thread->evfd, nullptr, EVENTIO_READ); -#if TS_USE_LINUX_IO_URING - // TODO: - // thread->ep->start(pd, DiskHandler::local_context()); -#endif #else thread->ep->start(pd, thread->evpipe[0], nullptr, EVENTIO_READ); #endif + +#ifdef TS_USE_LINUX_IO_URING + nh->uring_evio.type = EventIO::EVENTIO_IO_URING; + nh->uring_evio.start(pd, IOUringContext::local_context()->register_eventfd(), nullptr, EVENTIO_READ); +#endif } // NetHandler method definitions @@ -557,7 +558,7 @@ NetHandler::waitForActivity(ink_hrtime timeout) } else if (epd->type == EventIO::EVENTIO_NETACCEPT) { this->thread->schedule_imm(reinterpret_cast(epd->_user)); #if AIO_MODE == AIO_MODE_IO_URING - } else if (epd->type == EventIO::EVENTIO_DISK) { + } else if (epd->type == EventIO::EVENTIO_IO_URING) { servicedh = true; #endif }