diff --git a/demo/Makefile b/demo/Makefile index 543588e..2ab089a 100644 --- a/demo/Makefile +++ b/demo/Makefile @@ -15,7 +15,7 @@ endif override CXXFLAGS += -g -Wall -std=c++17 -I.. -lfmt -luring -pthread -all_targets = file_server http_client link_cp threading test bench echo_server +all_targets = file_server http_client link_cp threading test bench echo_server server_with_read_timeout all: $(all_targets) @@ -45,3 +45,6 @@ bench: bench.cpp ../include/task.hpp ../include/io_service.hpp echo_server: echo_server.cpp ../include/task.hpp ../include/io_service.hpp $(CXX_COMPILER) ./echo_server.cpp -I../include -o echo_server $(CXXFLAGS) + +server_with_read_timeout: server_with_read_timeout.cpp ../include/task.hpp ../include/io_service.hpp + $(CXX_COMPILER) ./server_with_read_timeout.cpp -I../include -o server_with_read_timeout $(CXXFLAGS) diff --git a/demo/server_with_read_timeout.cpp b/demo/server_with_read_timeout.cpp new file mode 100644 index 0000000..a647c01 --- /dev/null +++ b/demo/server_with_read_timeout.cpp @@ -0,0 +1,83 @@ +#include +#include +#include +#include +#include // https://github.com/fmtlib/fmt +#include +#include + +#include + +#define USE_SPLICE 0 +#define USE_LINK 0 +#define USE_POLL 0 + +enum { + BUF_SIZE = 512, + MAX_CONN_SIZE = 512, +}; + +int runningCoroutines = 0; + +uio::task<> accept_connection(uio::io_service &service, int serverfd) { + while (int clientfd = co_await service.accept(serverfd, nullptr, nullptr)) { + [](uio::io_service &service, int clientfd) -> uio::task<> { + __kernel_timespec timeout = { .tv_sec = 1, .tv_nsec = 0 }; + fmt::print("sockfd {} is accepted; number of running coroutines: {}\n", + clientfd, ++runningCoroutines); + std::vector buf(BUF_SIZE); + while (true) { + auto recvTask = service.read(clientfd, buf.data(), BUF_SIZE, 0, IOSQE_IO_LINK); + service.link_timeout(&timeout); + int r = co_await recvTask; + if (r <= 0) { + if (r == -ETIMEDOUT) { + fmt::println("Read timed out! Closing!"); + break; + } + fmt::println("Read failed! Error: {}", strerror(-r)); + break; + }; + co_await service.send(clientfd, buf.data(), r, MSG_NOSIGNAL); + } + service.shutdown(clientfd, SHUT_RDWR, IOSQE_IO_LINK); + co_await service.close(clientfd); + fmt::print("sockfd {} is closed; number of running coroutines: {}\n", + clientfd, --runningCoroutines); + }(service, clientfd); + } +} + +int main(int argc, char *argv[]) { + using uio::io_service; + using uio::panic_on_err; + using uio::on_scope_exit; + using uio::panic; + + uint16_t server_port = 0; + if (argc == 2) { + server_port = (uint16_t) std::strtoul(argv[1], nullptr, 10); + } + if (server_port == 0) { + fmt::print("Usage: {} \n", argv[0]); + return 1; + } + + io_service service(MAX_CONN_SIZE); + + int sockfd = socket(AF_INET, SOCK_STREAM, 0) | panic_on_err("socket creation", true); + on_scope_exit closesock([=]() { shutdown(sockfd, SHUT_RDWR); }); + + if (sockaddr_in addr = { + .sin_family = AF_INET, + .sin_port = htons(server_port), + .sin_addr = {INADDR_ANY}, + .sin_zero = {}, + }; bind(sockfd, reinterpret_cast(&addr), sizeof(sockaddr_in))) + panic("socket binding", errno); + + if (listen(sockfd, MAX_CONN_SIZE * 2)) panic("listen", errno); + fmt::print("Listening: {}\n", server_port); + + service.run(accept_connection(service, sockfd)); +} diff --git a/include/liburing/io_service.hpp b/include/liburing/io_service.hpp index 416cacf..6ff8dfc 100644 --- a/include/liburing/io_service.hpp +++ b/include/liburing/io_service.hpp @@ -201,6 +201,12 @@ class io_service { return await_work(sqe, iflags); } + void link_timeout(__kernel_timespec* timeout) { + auto* sqe = io_uring_get_sqe_safe(); + io_uring_prep_link_timeout(sqe, timeout, 0); + io_uring_sqe_set_data(sqe, new callback_resolver(+[](int) {})); + } + /** Read from a file descriptor at a given offset asynchronously * @see pread(2) * @see io_uring_enter(2) IORING_OP_READ