From ca4774f0252a9406fb2d82c1dc2af0d9808d312f Mon Sep 17 00:00:00 2001 From: Abderahmane BENALI Date: Mon, 1 Jun 2026 20:27:09 +0200 Subject: [PATCH] fix: surface EMSGSIZE as RMW_RET_ERROR on publish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rmw_publish previously discarded send_to()'s return value and always returned RMW_RET_OK. When sendmsg() failed with EMSGSIZE (datagram larger than the kernel-clamped SO_SNDBUF, e.g. net.core.wmem_max not tuned), the application saw "success" while no subscriber received anything. This was a silent data-loss bug at ~416 KB and above on a default-configured host. Replace send_to()'s bool return with an enum SendResult that distinguishes ConfigError (EMSGSIZE — hard, non-transient) from SoftDrop (EAGAIN/ENOBUFS/ENOENT/ECONNREFUSED — routine for a non-blocking transport). rmw_publish and rmw_publish_serialized_message now propagate ConfigError as RMW_RET_ERROR with a descriptive RMW_SET_ERROR_MSG; SoftDrop semantics are unchanged (still logged inside send_to, still RMW_RET_OK to the caller). Also add a one-shot per-process check at rmw_init that warns if /proc/sys/net/core/{wmem,rmem}_max is below the SO_{SND,RCV}BUF we request, so the misconfiguration surfaces at startup instead of on the first big publish. Tests: - PublishReturnsErrorOnEMSGSIZE: shrinks SO_SNDBUF and asserts that rmw_publish_serialized_message returns RMW_RET_ERROR. - PublishStillReturnsOkOnSoftDropPeerGone: unlinks a sub's socket and asserts the next publish still returns RMW_RET_OK (ENOENT must not be propagated). Both fail on HEAD and pass with this change. --- rmw_unix_socket_cpp/src/rmw_init.cpp | 2 + rmw_unix_socket_cpp/src/rmw_publisher.cpp | 35 +++++++-- rmw_unix_socket_cpp/src/transport.cpp | 46 +++++++++++- rmw_unix_socket_cpp/src/transport.hpp | 9 ++- rmw_unix_socket_cpp/test/test_rmw_qos.cpp | 78 +++++++++++++++++++++ rmw_unix_socket_cpp/test/test_transport.cpp | 12 ++-- 6 files changed, 166 insertions(+), 16 deletions(-) diff --git a/rmw_unix_socket_cpp/src/rmw_init.cpp b/rmw_unix_socket_cpp/src/rmw_init.cpp index ddfaf1b..cae8fd3 100644 --- a/rmw_unix_socket_cpp/src/rmw_init.cpp +++ b/rmw_unix_socket_cpp/src/rmw_init.cpp @@ -108,6 +108,8 @@ rmw_ret_t rmw_init(const rmw_init_options_t * options, rmw_context_t * context) rmw_uds::cleanup_orphan_socket_files(domain_id); } + rmw_uds::warn_if_sysctl_buffers_undersized(); + // Create send socket ctx->send_socket_fd = rmw_uds::create_send_socket(); if (ctx->send_socket_fd < 0) { diff --git a/rmw_unix_socket_cpp/src/rmw_publisher.cpp b/rmw_unix_socket_cpp/src/rmw_publisher.cpp index 63e1698..5577eb1 100644 --- a/rmw_unix_socket_cpp/src/rmw_publisher.cpp +++ b/rmw_unix_socket_cpp/src/rmw_publisher.cpp @@ -243,11 +243,21 @@ rmw_ret_t rmw_publish( } } - // Send current message to each subscriber (no registry lock held) + // Surface EMSGSIZE; soft drops (EAGAIN/ENOENT) are logged in send_to. + bool config_error = false; for (const auto & path : sub_paths) { - rmw_uds::send_to( - pub_data->context->send_socket_fd, - path, hdr, payload.data(), payload.size()); + if (rmw_uds::send_to( + pub_data->context->send_socket_fd, + path, hdr, payload.data(), payload.size()) == rmw_uds::SendResult::ConfigError) + { + config_error = true; + } + } + if (config_error) { + RMW_SET_ERROR_MSG( + "UDS send: message too large for kernel send buffer (EMSGSIZE). " + "Raise net.core.wmem_max or reduce message size."); + return RMW_RET_ERROR; } return RMW_RET_OK; @@ -297,10 +307,21 @@ rmw_ret_t rmw_publish_serialized_message( sub_paths = pub_data->cached_subscriber_paths; } + bool config_error = false; for (const auto & path : sub_paths) { - rmw_uds::send_to( - pub_data->context->send_socket_fd, - path, hdr, serialized_message->buffer, serialized_message->buffer_length); + if (rmw_uds::send_to( + pub_data->context->send_socket_fd, + path, hdr, serialized_message->buffer, + serialized_message->buffer_length) == rmw_uds::SendResult::ConfigError) + { + config_error = true; + } + } + if (config_error) { + RMW_SET_ERROR_MSG( + "UDS send: serialized message too large for kernel send buffer (EMSGSIZE). " + "Raise net.core.wmem_max or reduce message size."); + return RMW_RET_ERROR; } return RMW_RET_OK; diff --git a/rmw_unix_socket_cpp/src/transport.cpp b/rmw_unix_socket_cpp/src/transport.cpp index ad5aa32..9737a6b 100644 --- a/rmw_unix_socket_cpp/src/transport.cpp +++ b/rmw_unix_socket_cpp/src/transport.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include @@ -91,7 +92,7 @@ int create_send_socket() return fd; } -bool send_to( +SendResult send_to( int send_fd, const std::string & dest_path, const WireHeader & header, @@ -118,7 +119,7 @@ bool send_to( ssize_t sent = sendmsg(send_fd, &msg, MSG_DONTWAIT | MSG_NOSIGNAL); if (sent >= 0) { - return true; + return SendResult::Ok; } // Classify the failure. Hot path — every category is throttled so a @@ -127,6 +128,7 @@ bool send_to( // to identify the offending subscriber from the message alone. const int err = errno; const size_t total = sizeof(WireHeader) + payload_size; + SendResult result = SendResult::SoftDrop; switch (err) { case EMSGSIZE: // Message exceeds the kernel's per-datagram cap (typically the send @@ -137,6 +139,7 @@ bool send_to( "UDS send to '%s' failed: message too big (%zu bytes incl. header). " "Raise net.core.{wmem_max,rmem_max} or split the message.", dest_path.c_str(), total); + result = SendResult::ConfigError; break; case EAGAIN: #if defined(EWOULDBLOCK) && EWOULDBLOCK != EAGAIN @@ -169,7 +172,7 @@ bool send_to( dest_path.c_str(), std::strerror(err), err, total); break; } - return false; + return result; } bool recv_from( @@ -318,4 +321,41 @@ void cleanup_orphan_socket_files(size_t domain_id) closedir(dir); } +void warn_if_sysctl_buffers_undersized() +{ + static std::once_flag once; + std::call_once(once, []() { + auto read_proc_int = [](const char * path) -> long { + FILE * f = std::fopen(path, "r"); + if (!f) { + return -1; + } + long v = -1; + if (std::fscanf(f, "%ld", &v) != 1) { + v = -1; + } + std::fclose(f); + return v; + }; + + const long wmax = read_proc_int("/proc/sys/net/core/wmem_max"); + const long rmax = read_proc_int("/proc/sys/net/core/rmem_max"); + + if (wmax > 0 && wmax < SEND_BUF_SIZE) { + RMW_UDS_LOG_WARN( + "net.core.wmem_max=%ld < requested SO_SNDBUF=%d. Large outbound " + "datagrams will fail with EMSGSIZE. Run: sudo sysctl -w " + "net.core.wmem_max=%d (or larger) and persist in /etc/sysctl.d/.", + wmax, SEND_BUF_SIZE, SEND_BUF_SIZE); + } + if (rmax > 0 && rmax < RECV_BUF_SIZE) { + RMW_UDS_LOG_WARN( + "net.core.rmem_max=%ld < requested SO_RCVBUF=%d. Subscribers may " + "drop large datagrams. Run: sudo sysctl -w net.core.rmem_max=%d " + "(or larger) and persist in /etc/sysctl.d/.", + rmax, RECV_BUF_SIZE, RECV_BUF_SIZE); + } + }); +} + } // namespace rmw_uds diff --git a/rmw_unix_socket_cpp/src/transport.hpp b/rmw_unix_socket_cpp/src/transport.hpp index 99997c6..6bb342c 100644 --- a/rmw_unix_socket_cpp/src/transport.hpp +++ b/rmw_unix_socket_cpp/src/transport.hpp @@ -22,9 +22,11 @@ int create_bound_socket(const std::string & path); // Returns fd >= 0 on success, -1 on failure. int create_send_socket(); +// ConfigError = EMSGSIZE (hard, surface to caller); SoftDrop = transient. +enum class SendResult { Ok, SoftDrop, ConfigError }; + // Send a datagram (header + payload) to the given socket path. -// Returns true on success. -bool send_to( +SendResult send_to( int send_fd, const std::string & dest_path, const WireHeader & header, @@ -49,6 +51,9 @@ void close_socket(int fd, const std::string & path); // Safe to call at startup to clean up after ungraceful shutdowns. void cleanup_orphan_socket_files(size_t domain_id); +// One-shot per-process: warn if net.core.{w,r}mem_max < SO_{SND,RCV}BUF. +void warn_if_sysctl_buffers_undersized(); + } // namespace rmw_uds #endif // RMW_UNIX_SOCKET_CPP__TRANSPORT_HPP_ diff --git a/rmw_unix_socket_cpp/test/test_rmw_qos.cpp b/rmw_unix_socket_cpp/test/test_rmw_qos.cpp index f30d9a1..bd54b20 100644 --- a/rmw_unix_socket_cpp/test/test_rmw_qos.cpp +++ b/rmw_unix_socket_cpp/test/test_rmw_qos.cpp @@ -1,9 +1,13 @@ #include "test_base.hpp" +#include #include #include #include +#include +#include + #include "test_msgs/msg/basic_types.hpp" #include "test_msgs/srv/basic_types.hpp" @@ -11,6 +15,8 @@ #include "rosidl_typesupport_cpp/message_type_support.hpp" #include "rosidl_typesupport_cpp/service_type_support.hpp" +#include "types.hpp" + class QosTest : public RmwUdsNodeTest { protected: @@ -134,6 +140,78 @@ TEST_F(QosTest, TransientLocalCacheDepthEnforced) auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); } +// --- Transport error propagation --- + +TEST_F(QosTest, PublishReturnsErrorOnEMSGSIZE) +{ + // Shrink SO_SNDBUF so any send hits EMSGSIZE — publish must return ERROR. + auto * ctx_impl = reinterpret_cast(context.impl); + int small_buf = 2048; + ASSERT_EQ( + 0, + setsockopt( + ctx_impl->send_socket_fd, SOL_SOCKET, SO_SNDBUF, + &small_buf, sizeof(small_buf))); + + auto qos = make_qos( + RMW_QOS_POLICY_RELIABILITY_RELIABLE, + RMW_QOS_POLICY_DURABILITY_VOLATILE); + + auto pub_opts = rmw_get_default_publisher_options(); + auto * pub = rmw_create_publisher(node, ts, "/emsgsize", &qos, &pub_opts); + ASSERT_NE(nullptr, pub); + + auto sub_opts = rmw_get_default_subscription_options(); + auto * sub = rmw_create_subscription(node, ts, "/emsgsize", &qos, &sub_opts); + ASSERT_NE(nullptr, sub); + + // 64 KB — well above SOCK_MIN_SNDBUF the kernel will clamp us to. + constexpr size_t big_size = 64 * 1024; + rmw_serialized_message_t serialized; + serialized.buffer_capacity = big_size; + serialized.buffer_length = big_size; + serialized.buffer = static_cast(std::malloc(big_size)); + ASSERT_NE(nullptr, serialized.buffer); + std::memset(serialized.buffer, 0x42, big_size); + serialized.allocator = rcutils_get_default_allocator(); + + EXPECT_EQ(RMW_RET_ERROR, rmw_publish_serialized_message(pub, &serialized, nullptr)); + + std::free(serialized.buffer); + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); +} + +TEST_F(QosTest, PublishStillReturnsOkOnSoftDropPeerGone) +{ + // ENOENT on a vanished peer must stay RET_OK — only EMSGSIZE escalates. + auto qos = make_qos( + RMW_QOS_POLICY_RELIABILITY_RELIABLE, + RMW_QOS_POLICY_DURABILITY_VOLATILE); + + auto pub_opts = rmw_get_default_publisher_options(); + auto * pub = rmw_create_publisher(node, ts, "/peer_gone", &qos, &pub_opts); + ASSERT_NE(nullptr, pub); + + auto sub_opts = rmw_get_default_subscription_options(); + auto * sub = rmw_create_subscription(node, ts, "/peer_gone", &qos, &sub_opts); + ASSERT_NE(nullptr, sub); + + // Warm pub's path cache, then unlink the sub's socket → next sendmsg = ENOENT. + test_msgs::msg::BasicTypes m; + m.int32_value = 1; + ASSERT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); + + auto * sub_impl = static_cast(sub->data); + unlink(sub_impl->socket_path.c_str()); + + m.int32_value = 2; + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); + + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub); +} + TEST_F(QosTest, VolatileNoCache) { auto qos = make_qos( diff --git a/rmw_unix_socket_cpp/test/test_transport.cpp b/rmw_unix_socket_cpp/test/test_transport.cpp index b3f6ffc..7fa081d 100644 --- a/rmw_unix_socket_cpp/test/test_transport.cpp +++ b/rmw_unix_socket_cpp/test/test_transport.cpp @@ -54,8 +54,10 @@ TEST_F(TransportTest, SendAndReceive) std::vector payload = {1, 2, 3, 4, 5}; send_hdr.payload_size = static_cast(payload.size()); - ASSERT_TRUE(rmw_uds::send_to( - send_fd, recv_path, send_hdr, payload.data(), payload.size())); + ASSERT_EQ( + rmw_uds::SendResult::Ok, + rmw_uds::send_to( + send_fd, recv_path, send_hdr, payload.data(), payload.size())); // Receive rmw_uds::WireHeader recv_hdr; @@ -99,8 +101,10 @@ TEST_F(TransportTest, MultipleMessages) hdr.sequence_number = i; hdr.payload_size = sizeof(int); hdr.msg_type = 0; - ASSERT_TRUE(rmw_uds::send_to( - send_fd, path, hdr, reinterpret_cast(&i), sizeof(i))); + ASSERT_EQ( + rmw_uds::SendResult::Ok, + rmw_uds::send_to( + send_fd, path, hdr, reinterpret_cast(&i), sizeof(i))); } for (int i = 0; i < N; ++i) {