From 1f68df76fed9c42be553c9ca8ad2efb72cbc1b98 Mon Sep 17 00:00:00 2001 From: Dominik Thalhammer Date: Tue, 10 Feb 2026 22:20:00 +0100 Subject: [PATCH 1/2] :sparkles: Allow setting tcp nodely socket option --- include/asyncpp/io/detail/io_engine.h | 1 + include/asyncpp/io/socket.h | 7 +++++-- src/io_engine_generic_unix.cpp | 8 ++++++++ src/io_engine_generic_unix.h | 1 + src/io_engine_iocp.cpp | 14 ++++++++++++++ src/io_engine_select.cpp | 12 ++++++++++++ src/io_engine_uring.cpp | 11 ++++++++++- src/socket.cpp | 6 ++++++ 8 files changed, 57 insertions(+), 3 deletions(-) diff --git a/include/asyncpp/io/detail/io_engine.h b/include/asyncpp/io/detail/io_engine.h index be52138..0749c59 100644 --- a/include/asyncpp/io/detail/io_engine.h +++ b/include/asyncpp/io/detail/io_engine.h @@ -85,6 +85,7 @@ namespace asyncpp::io::detail { virtual void socket_multicast_set_ttl(socket_handle_t socket, size_t ttl) = 0; virtual void socket_multicast_set_loopback(socket_handle_t socket, bool enabled) = 0; virtual void socket_allow_reuse_address(socket_handle_t socket, bool enabled) = 0; + virtual void socket_enable_nagles_algorithm(socket_handle_t socket, bool enabled) = 0; virtual void socket_shutdown(socket_handle_t socket, bool receive, bool send) = 0; virtual bool enqueue_connect(socket_handle_t socket, endpoint ep, completion_data* cd) = 0; virtual bool enqueue_accept(socket_handle_t socket, completion_data* cd) = 0; diff --git a/include/asyncpp/io/socket.h b/include/asyncpp/io/socket.h index 853c5df..cff0446 100644 --- a/include/asyncpp/io/socket.h +++ b/include/asyncpp/io/socket.h @@ -108,6 +108,7 @@ namespace asyncpp::io { void multicast_set_loopback(bool enabled); void allow_reuse_address(bool enabled); + void enable_nagles_algorithm(bool enabled); [[nodiscard]] detail::io_engine::socket_handle_t native_handle() const noexcept { return m_fd; } [[nodiscard]] detail::io_engine::socket_handle_t release() noexcept { @@ -177,7 +178,8 @@ namespace asyncpp::io { requires(std::is_invocable_v) void connect(const endpoint& ep, FN&& cb, asyncpp::stop_token st = {}); template - requires(std::is_invocable_v>) + requires(std::is_invocable_v> || + (std::is_invocable_v && std::is_invocable_v)) void accept(FN&& cb, asyncpp::stop_token st = {}); template requires(std::is_invocable_v) @@ -700,7 +702,8 @@ namespace asyncpp::io { } template - requires(std::is_invocable_v>) + requires(std::is_invocable_v> || + (std::is_invocable_v && std::is_invocable_v)) inline void socket::accept(FN&& cb, asyncpp::stop_token st) { struct data : detail::io_engine::completion_data { FN real_cb; diff --git a/src/io_engine_generic_unix.cpp b/src/io_engine_generic_unix.cpp index 987c17b..6714172 100644 --- a/src/io_engine_generic_unix.cpp +++ b/src/io_engine_generic_unix.cpp @@ -9,6 +9,7 @@ #include #include +#include #include #include #include @@ -240,6 +241,13 @@ namespace asyncpp::io::detail { if (res < 0) throw std::system_error(errno, std::system_category(), "setsockopt failed"); } + void io_engine_generic_unix::socket_enable_nagles_algorithm(socket_handle_t socket, bool enabled) { + // Note: The convention of asyncpp-io is inverted to the default socket one (because honestly TCP_NODELAY should be the default). + int val = enabled ? 0 : 1; + auto res = setsockopt(socket, SOL_TCP, TCP_NODELAY, reinterpret_cast(&val), sizeof(val)); + if (res < 0) throw std::system_error(errno, std::system_category(), "setsockopt failed"); + } + void io_engine_generic_unix::socket_shutdown(socket_handle_t socket, bool receive, bool send) { int mode = 0; if (receive && send) diff --git a/src/io_engine_generic_unix.h b/src/io_engine_generic_unix.h index c755245..d8ae9fe 100644 --- a/src/io_engine_generic_unix.h +++ b/src/io_engine_generic_unix.h @@ -20,6 +20,7 @@ namespace asyncpp::io::detail { void socket_multicast_set_ttl(socket_handle_t socket, size_t ttl) override; void socket_multicast_set_loopback(socket_handle_t socket, bool enabled) override; void socket_allow_reuse_address(socket_handle_t socket, bool enabled) override; + void socket_enable_nagles_algorithm(socket_handle_t socket, bool enabled) override; void socket_shutdown(socket_handle_t socket, bool receive, bool send) override; file_handle_t file_open(const char* filename, std::ios_base::openmode mode) override; diff --git a/src/io_engine_iocp.cpp b/src/io_engine_iocp.cpp index 74c700d..fb34bf5 100644 --- a/src/io_engine_iocp.cpp +++ b/src/io_engine_iocp.cpp @@ -90,6 +90,7 @@ namespace asyncpp::io::detail { void socket_multicast_set_ttl(socket_handle_t socket, size_t ttl) override; void socket_multicast_set_loopback(socket_handle_t socket, bool enabled) override; void socket_allow_reuse_address(socket_handle_t socket, bool enabled) override; + void socket_enable_nagles_algorithm(socket_handle_t socket, bool enabled) override; void socket_shutdown(socket_handle_t socket, bool receive, bool send) override; bool enqueue_connect(socket_handle_t socket, endpoint ep, completion_data* cd) override; bool enqueue_accept(socket_handle_t socket, completion_data* cd) override; @@ -166,6 +167,12 @@ namespace asyncpp::io::detail { cd->result = std::error_code(GetLastError(), std::system_category()); return true; } + if (int opt = 1; setsockopt(state->accept_sock, SOL_TCP, TCP_NODELAY, + reinterpret_cast(&opt), sizeof(opt)) == SOCKET_ERROR) { + closesocket(state->accept_sock); + cd->result = std::error_code(GetLastError(), std::system_category()); + return true; + } cd->result_handle = state->accept_sock; } else { cd->result_size = num_bytes; @@ -459,6 +466,13 @@ namespace asyncpp::io::detail { if (res < 0) throw std::system_error(WSAGetLastError(), std::system_category(), "setsockopt failed"); } + void io_engine_iocp::socket_enable_nagles_algorithm(socket_handle_t socket, bool enabled) { + // Note: The convention of asyncpp-io is inverted to the default socket one (because honestly TCP_NODELAY should be the default). + int val = enabled ? 0 : 1; + auto res = setsockopt(socket, SOL_TCP, TCP_NODELAY, reinterpret_cast(&val), sizeof(val)); + if (res < 0) throw std::system_error(WSAGetLastError(), std::system_category(), "setsockopt failed"); + } + void io_engine_iocp::socket_shutdown(socket_handle_t socket, bool receive, bool send) { int mode = 0; if (receive && send) diff --git a/src/io_engine_select.cpp b/src/io_engine_select.cpp index a6505a0..2adb829 100644 --- a/src/io_engine_select.cpp +++ b/src/io_engine_select.cpp @@ -7,12 +7,14 @@ namespace asyncpp::io::detail { #else #include "io_engine_generic_unix.h" +#include #include #include #include #include #include +#include #include #include #include @@ -105,6 +107,7 @@ namespace asyncpp::io::detail { std::unique_ptr create_io_engine_select() { return std::make_unique(); } io_engine_select::io_engine_select() { + signal(SIGPIPE, SIG_IGN); #ifdef USE_EVENTFD m_wake_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); if (m_wake_fd < 0) throw std::system_error(errno, std::system_category(), "eventfd failed"); @@ -204,6 +207,7 @@ namespace asyncpp::io::detail { if (res >= 0) { e.state.send.len -= res; e.state.send.buf = static_cast(e.state.send.buf) + res; + e.done->result_size += res; if (e.state.send.len == 0) { e.done->result.clear(); m_done_callbacks.push_back(e.done); @@ -222,6 +226,13 @@ namespace asyncpp::io::detail { if (res >= 0) { e.done->result.clear(); e.done->result_handle = res; + + if (int opt = 1; + setsockopt(res, SOL_TCP, TCP_NODELAY, reinterpret_cast(&opt), sizeof(opt)) < 0) { + e.done->result = std::error_code(errno, std::system_category()); + close(e.done->result_handle); + e.done->result_handle = -1; + } } else if (errno != EAGAIN) { e.done->result = std::error_code(errno, std::system_category()); } else @@ -364,6 +375,7 @@ namespace asyncpp::io::detail { if (res >= 0) { len -= res; buf = static_cast(buf) + res; + cd->result_size += res; } else if (errno != EAGAIN) { cd->result = std::error_code(errno, std::system_category()); return true; diff --git a/src/io_engine_uring.cpp b/src/io_engine_uring.cpp index b76f25e..34ceea5 100644 --- a/src/io_engine_uring.cpp +++ b/src/io_engine_uring.cpp @@ -13,6 +13,7 @@ namespace asyncpp::io::detail { #include #include #include +#include #include #include #include @@ -134,7 +135,15 @@ namespace asyncpp::io::detail { auto state = info->es_get(); info->result = std::error_code(opres < 0 ? -opres : 0, std::system_category()); switch (state->op) { - case uring_op::accept: info->result_handle = opres; break; + case uring_op::accept: + info->result_handle = opres; + if (int opt = 1; !info->result && + setsockopt(opres, SOL_TCP, TCP_NODELAY, reinterpret_cast(&opt), sizeof(opt)) < 0) { + info->result = std::error_code(errno, std::system_category()); + close(info->result_handle); + info->result_handle = -1; + } + break; default: info->result_size = static_cast(opres); break; } diff --git a/src/socket.cpp b/src/socket.cpp index f3cbfec..cf55f23 100644 --- a/src/socket.cpp +++ b/src/socket.cpp @@ -4,6 +4,7 @@ namespace asyncpp::io { socket socket::create_tcp(io_service& io, address_type addrtype) { auto fd = io.engine()->socket_create(addrtype, detail::io_engine::socket_type::stream); + io.engine()->socket_enable_nagles_algorithm(fd, false); return socket(&io, fd); } @@ -152,6 +153,11 @@ namespace asyncpp::io { m_io->engine()->socket_allow_reuse_address(m_fd, enabled); } + void socket::enable_nagles_algorithm(bool enabled) { + if (m_fd == detail::io_engine::invalid_socket_handle) throw std::logic_error("invalid socket"); + m_io->engine()->socket_enable_nagles_algorithm(m_fd, enabled); + } + void socket::close_send() { if (m_fd == detail::io_engine::invalid_socket_handle) throw std::logic_error("invalid socket"); m_io->engine()->socket_shutdown(m_fd, false, true); From b1cc6ff37b3c91783bc9dc97f30735c7537f4ce6 Mon Sep 17 00:00:00 2001 From: Dominik Thalhammer Date: Thu, 12 Feb 2026 09:16:00 +0100 Subject: [PATCH 2/2] :bug: Replace linux specific SOL_TCP with IPPROTO_TCP --- src/io_engine_generic_unix.cpp | 2 +- src/io_engine_iocp.cpp | 4 ++-- src/io_engine_select.cpp | 2 +- src/io_engine_uring.cpp | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/io_engine_generic_unix.cpp b/src/io_engine_generic_unix.cpp index 6714172..61a355a 100644 --- a/src/io_engine_generic_unix.cpp +++ b/src/io_engine_generic_unix.cpp @@ -244,7 +244,7 @@ namespace asyncpp::io::detail { void io_engine_generic_unix::socket_enable_nagles_algorithm(socket_handle_t socket, bool enabled) { // Note: The convention of asyncpp-io is inverted to the default socket one (because honestly TCP_NODELAY should be the default). int val = enabled ? 0 : 1; - auto res = setsockopt(socket, SOL_TCP, TCP_NODELAY, reinterpret_cast(&val), sizeof(val)); + auto res = setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast(&val), sizeof(val)); if (res < 0) throw std::system_error(errno, std::system_category(), "setsockopt failed"); } diff --git a/src/io_engine_iocp.cpp b/src/io_engine_iocp.cpp index fb34bf5..692e457 100644 --- a/src/io_engine_iocp.cpp +++ b/src/io_engine_iocp.cpp @@ -167,7 +167,7 @@ namespace asyncpp::io::detail { cd->result = std::error_code(GetLastError(), std::system_category()); return true; } - if (int opt = 1; setsockopt(state->accept_sock, SOL_TCP, TCP_NODELAY, + if (int opt = 1; setsockopt(state->accept_sock, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast(&opt), sizeof(opt)) == SOCKET_ERROR) { closesocket(state->accept_sock); cd->result = std::error_code(GetLastError(), std::system_category()); @@ -469,7 +469,7 @@ namespace asyncpp::io::detail { void io_engine_iocp::socket_enable_nagles_algorithm(socket_handle_t socket, bool enabled) { // Note: The convention of asyncpp-io is inverted to the default socket one (because honestly TCP_NODELAY should be the default). int val = enabled ? 0 : 1; - auto res = setsockopt(socket, SOL_TCP, TCP_NODELAY, reinterpret_cast(&val), sizeof(val)); + auto res = setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast(&val), sizeof(val)); if (res < 0) throw std::system_error(WSAGetLastError(), std::system_category(), "setsockopt failed"); } diff --git a/src/io_engine_select.cpp b/src/io_engine_select.cpp index 2adb829..1b93e39 100644 --- a/src/io_engine_select.cpp +++ b/src/io_engine_select.cpp @@ -228,7 +228,7 @@ namespace asyncpp::io::detail { e.done->result_handle = res; if (int opt = 1; - setsockopt(res, SOL_TCP, TCP_NODELAY, reinterpret_cast(&opt), sizeof(opt)) < 0) { + setsockopt(res, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast(&opt), sizeof(opt)) < 0) { e.done->result = std::error_code(errno, std::system_category()); close(e.done->result_handle); e.done->result_handle = -1; diff --git a/src/io_engine_uring.cpp b/src/io_engine_uring.cpp index 34ceea5..9a81c5e 100644 --- a/src/io_engine_uring.cpp +++ b/src/io_engine_uring.cpp @@ -138,7 +138,7 @@ namespace asyncpp::io::detail { case uring_op::accept: info->result_handle = opres; if (int opt = 1; !info->result && - setsockopt(opres, SOL_TCP, TCP_NODELAY, reinterpret_cast(&opt), sizeof(opt)) < 0) { + setsockopt(opres, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast(&opt), sizeof(opt)) < 0) { info->result = std::error_code(errno, std::system_category()); close(info->result_handle); info->result_handle = -1;