Skip to content

Commit c1dd857

Browse files
fix(shm): honest TL publish return, serialized-TL replay, fallback test
Addresses the review findings on the durable-shm TRANSIENT_LOCAL change: - Return code (finding #1): the TL publish path ignored send_to's result and always returned RMW_RET_OK, even when the live send of the current message was rejected by the kernel size cap (EMSGSIZE) — a lying success, and inconsistent with the non-TL path. Extract the TL cache+replay+send logic into transient_local_publish(), which now surfaces ConfigError as RMW_RET_ERROR. (ENOBUFS/oversize on a raised-buffer machine stays a soft drop in BOTH paths, unchanged — a shared, pre-existing limitation.) - Serialized TL gap (completeness): rmw_publish_serialized_message did not feed the replay cache at all, so a large latched payload published through it was sent inline once and dropped for late joiners. It now routes TL through the same transient_local_publish() helper (durable staging + replay) and gained the same known-subscriber prune-on-refresh as rmw_publish. - Fallback coverage (finding #6): add a cold-path test seam (env var RMW_UDS_TEST_FORCE_SHM_FAILURE, checked only on large latched staging) so a test can force the shm-unavailable path and assert the payload is cached inline (no durable segment, no flag) and still delivered. Tests: TransientLocalPublishReturnsErrorOnEMSGSIZE and TransientLocalSerializedLargeMessageLateJoiner both confirmed to FAIL on the pre-change publisher and pass here; TransientLocalLargeMessageInlineFallbackWhenShmUnavailable covers the fallback. Full suite: 95 tests, clean build.
1 parent 8c90cf2 commit c1dd857

3 files changed

Lines changed: 272 additions & 63 deletions

File tree

‎rmw_unix_socket_cpp/src/rmw_publisher.cpp‎

Lines changed: 115 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -188,6 +188,89 @@ rmw_ret_t rmw_destroy_publisher(rmw_node_t * node, rmw_publisher_t * publisher)
188188
return RMW_RET_OK;
189189
}
190190

191+
// Cache a TRANSIENT_LOCAL message for late-joiner replay and send it now.
192+
// Shared by rmw_publish and rmw_publish_serialized_message. Large payloads
193+
// (>= SHM_PAYLOAD_THRESHOLD) are staged into a dedicated durable shm segment
194+
// (shm_stage_durable) so the cached entry and its replays are a small
195+
// descriptor rather than a datagram the kernel would reject; small payloads and
196+
// the shm-unavailable fallback are cached inline. `payload` is taken by value:
197+
// callers std::move a serialized vector in, so the common path is one move, and
198+
// the inline branch keeps the bytes with no copy. Returns RMW_RET_ERROR if the
199+
// live send of the current message hits the kernel size cap (EMSGSIZE),
200+
// mirroring the non-TL path; replay of previously-cached messages to a new
201+
// subscriber is best-effort and not reflected in the return code.
202+
static rmw_ret_t transient_local_publish(
203+
rmw_uds::UdsPublisher * pub_data,
204+
rmw_uds::WireHeader hdr,
205+
std::vector<uint8_t> payload,
206+
const std::shared_ptr<const std::vector<std::string>> & sub_paths)
207+
{
208+
std::lock_guard<std::mutex> lock(pub_data->cache_mutex);
209+
210+
rmw_uds::CachedMessage cached;
211+
cached.header = hdr;
212+
if (payload.size() >= rmw_uds::SHM_PAYLOAD_THRESHOLD) {
213+
rmw_uds::ShmPayloadDescriptor desc;
214+
auto seg = rmw_uds::shm_stage_durable(
215+
pub_data->context->domain_id, payload.data(), payload.size(), desc);
216+
if (seg) {
217+
cached.header.msg_type |= rmw_uds::SHM_PAYLOAD_FLAG;
218+
cached.header.payload_size = static_cast<uint32_t>(sizeof(desc));
219+
cached.payload.assign(
220+
reinterpret_cast<const uint8_t *>(&desc),
221+
reinterpret_cast<const uint8_t *>(&desc) + sizeof(desc));
222+
cached.shm_seg = std::move(seg);
223+
} else {
224+
cached.payload = std::move(payload); // inline fallback
225+
}
226+
} else {
227+
cached.payload = std::move(payload);
228+
}
229+
pub_data->message_cache.push_back(std::move(cached));
230+
while (pub_data->message_cache.size() > pub_data->qos.depth) {
231+
pub_data->message_cache.pop_front();
232+
}
233+
234+
if (sub_paths) {
235+
for (const auto & path : *sub_paths) {
236+
if (pub_data->known_subscriber_paths.count(path) == 0) {
237+
pub_data->known_subscriber_paths.insert(path);
238+
for (size_t i = 0; i + 1 < pub_data->message_cache.size(); ++i) {
239+
const auto & cm = pub_data->message_cache[i];
240+
rmw_uds::send_to(
241+
pub_data->context->send_socket_fd,
242+
path, cm.header, cm.payload.data(), cm.payload.size());
243+
}
244+
}
245+
}
246+
}
247+
248+
// Send the current message, read from the cached copy since payload was
249+
// moved. Trimming never disturbs back(), so it is the message just pushed.
250+
// Surface EMSGSIZE the way the non-TL path does so a latched payload the
251+
// kernel rejects is not reported as a successful publish.
252+
const auto & current = pub_data->message_cache.back();
253+
bool config_error = false;
254+
if (sub_paths) {
255+
for (const auto & path : *sub_paths) {
256+
if (rmw_uds::send_to(
257+
pub_data->context->send_socket_fd,
258+
path, current.header, current.payload.data(), current.payload.size())
259+
== rmw_uds::SendResult::ConfigError)
260+
{
261+
config_error = true;
262+
}
263+
}
264+
}
265+
if (config_error) {
266+
RMW_SET_ERROR_MSG(
267+
"UDS send: TRANSIENT_LOCAL message too large for kernel send buffer "
268+
"(EMSGSIZE). Raise net.core.wmem_max or reduce message size.");
269+
return RMW_RET_ERROR;
270+
}
271+
return RMW_RET_OK;
272+
}
273+
191274
rmw_ret_t rmw_publish(
192275
const rmw_publisher_t * publisher,
193276
const void * ros_message,
@@ -266,64 +349,9 @@ rmw_ret_t rmw_publish(
266349
sub_paths = pub_data->cached_subscriber_paths; // refcount copy under lock
267350
}
268351

269-
// TRANSIENT_LOCAL: cache message and replay to late-joining subscribers
352+
// TRANSIENT_LOCAL: cache message and replay to late-joining subscribers.
270353
if (pub_data->qos.durability == RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL) {
271-
std::lock_guard<std::mutex> lock(pub_data->cache_mutex);
272-
273-
rmw_uds::CachedMessage cached;
274-
cached.header = hdr;
275-
// Large payloads are staged into a dedicated durable shm segment so the
276-
// cache entry — and every replay of it to a late joiner — is a small
277-
// descriptor rather than a multi-megabyte datagram the kernel would drop.
278-
// The segment lives as long as this cache entry (unlinked on eviction).
279-
// Falls back to caching the payload inline when shm is unavailable.
280-
if (payload.size() >= rmw_uds::SHM_PAYLOAD_THRESHOLD) {
281-
rmw_uds::ShmPayloadDescriptor desc;
282-
auto seg = rmw_uds::shm_stage_durable(
283-
pub_data->context->domain_id, payload.data(), payload.size(), desc);
284-
if (seg) {
285-
cached.header.msg_type |= rmw_uds::SHM_PAYLOAD_FLAG;
286-
cached.header.payload_size = static_cast<uint32_t>(sizeof(desc));
287-
cached.payload.assign(
288-
reinterpret_cast<const uint8_t *>(&desc),
289-
reinterpret_cast<const uint8_t *>(&desc) + sizeof(desc));
290-
cached.shm_seg = std::move(seg);
291-
} else {
292-
cached.payload = std::move(payload); // inline fallback
293-
}
294-
} else {
295-
cached.payload = std::move(payload); // last use of payload
296-
}
297-
pub_data->message_cache.push_back(std::move(cached));
298-
while (pub_data->message_cache.size() > pub_data->qos.depth) {
299-
pub_data->message_cache.pop_front();
300-
}
301-
302-
if (sub_paths) {
303-
for (const auto & path : *sub_paths) {
304-
if (pub_data->known_subscriber_paths.count(path) == 0) {
305-
pub_data->known_subscriber_paths.insert(path);
306-
for (size_t i = 0; i + 1 < pub_data->message_cache.size(); ++i) {
307-
const auto & cm = pub_data->message_cache[i];
308-
rmw_uds::send_to(
309-
pub_data->context->send_socket_fd,
310-
path, cm.header, cm.payload.data(), cm.payload.size());
311-
}
312-
}
313-
}
314-
}
315-
316-
// Send the current message, read from the cached copy since payload was
317-
// moved. Trimming never disturbs back(), so it is the message just pushed.
318-
const auto & current = pub_data->message_cache.back();
319-
if (sub_paths) {
320-
for (const auto & path : *sub_paths) {
321-
rmw_uds::send_to(
322-
pub_data->context->send_socket_fd,
323-
path, current.header, current.payload.data(), current.payload.size());
324-
}
325-
}
326-
return RMW_RET_OK;
354+
return transient_local_publish(pub_data, hdr, std::move(payload), sub_paths);
327355
}
328356

329357
// Large payloads travel through the publisher's shm ring: stage the bytes
@@ -413,18 +441,42 @@ rmw_ret_t rmw_publish_serialized_message(
413441
}
414442
pub_data->cached_generation = current_gen;
415443
pub_data->cached_subscriber_paths = std::move(fresh);
444+
445+
// TRANSIENT_LOCAL: prune known subscribers no longer present, mirroring
446+
// rmw_publish (nested lock order sub_cache_mutex -> cache_mutex) so a
447+
// concurrent refresh cannot make the pruning set stale.
448+
if (pub_data->qos.durability == RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL) {
449+
std::lock_guard<std::mutex> prune_lock(pub_data->cache_mutex);
450+
std::set<std::string> current(
451+
pub_data->cached_subscriber_paths->begin(),
452+
pub_data->cached_subscriber_paths->end());
453+
for (auto it = pub_data->known_subscriber_paths.begin();
454+
it != pub_data->known_subscriber_paths.end(); )
455+
{
456+
it = (current.count(*it) == 0) ?
457+
pub_data->known_subscriber_paths.erase(it) : std::next(it);
458+
}
459+
}
416460
}
417461
sub_paths = pub_data->cached_subscriber_paths;
418462
}
419463

420-
// Same large-payload fork as rmw_publish: stage once, fan out descriptors.
421-
// TRANSIENT_LOCAL keeps the inline path for consistency with rmw_publish
422-
// (this entry point does not feed the replay cache).
464+
// TRANSIENT_LOCAL: cache + replay through the same durable-shm path as
465+
// rmw_publish, so a serialized latched payload (including one above the
466+
// datagram cap) reaches late joiners too.
467+
if (pub_data->qos.durability == RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL) {
468+
return transient_local_publish(
469+
pub_data, hdr,
470+
std::vector<uint8_t>(
471+
serialized_message->buffer,
472+
serialized_message->buffer + serialized_message->buffer_length),
473+
sub_paths);
474+
}
475+
476+
// Large non-TL payloads: stage once into the cycling ring, fan out descriptors.
423477
rmw_uds::ShmPayloadDescriptor desc;
424478
bool staged = false;
425-
if (serialized_message->buffer_length >= rmw_uds::SHM_PAYLOAD_THRESHOLD &&
426-
pub_data->qos.durability != RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL)
427-
{
479+
if (serialized_message->buffer_length >= rmw_uds::SHM_PAYLOAD_THRESHOLD) {
428480
std::lock_guard<std::mutex> lock(pub_data->shm_mutex);
429481
staged = rmw_uds::shm_stage_payload(
430482
pub_data->shm_ring, pub_data->context->domain_id,

‎rmw_unix_socket_cpp/src/shm_transport.cpp‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
#include <cerrno>
1818
#include <chrono>
1919
#include <cstdio>
20+
#include <cstdlib>
2021
#include <cstring>
2122

2223
#include <dirent.h>
@@ -205,6 +206,12 @@ std::unique_ptr<DurableShmSegment> shm_stage_durable(
205206
if (payload_size > UINT32_MAX) {
206207
return nullptr;
207208
}
209+
// Test seam (cold path, large latched publishes only): when this env var is
210+
// set, behave as if shared memory were unavailable so tests can exercise the
211+
// inline fallback deterministically. Never set in production.
212+
if (std::getenv("RMW_UDS_TEST_FORCE_SHM_FAILURE") != nullptr) {
213+
return nullptr;
214+
}
208215

209216
// One record at offset 0; the segment holds nothing else and never grows.
210217
const size_t record_bytes =

‎rmw_unix_socket_cpp/test/test_rmw_qos.cpp‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -228,6 +228,156 @@ TEST_F(QosTest, TransientLocalHugeMessageLateJoiner)
228228
auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
229229
}
230230

231+
TEST_F(QosTest, TransientLocalPublishReturnsErrorOnEMSGSIZE)
232+
{
233+
// A latched publish whose live send is rejected by the kernel size cap must
234+
// return RMW_RET_ERROR, not a lying RMW_RET_OK. Before the fix the TL path
235+
// ignored send_to's result entirely. Shrink SO_SNDBUF so a sub-threshold
236+
// (inline) latched message hits EMSGSIZE deterministically, independent of
237+
// the machine's net.core.wmem_max.
238+
auto * ctx_impl = reinterpret_cast<rmw_uds::UdsContext *>(context.impl);
239+
int small_buf = 2048;
240+
ASSERT_EQ(
241+
0,
242+
setsockopt(
243+
ctx_impl->send_socket_fd, SOL_SOCKET, SO_SNDBUF,
244+
&small_buf, sizeof(small_buf)));
245+
246+
auto seq_ts = rosidl_typesupport_cpp::get_message_type_support_handle<
247+
test_msgs::msg::UnboundedSequences>();
248+
auto qos = make_qos(
249+
RMW_QOS_POLICY_RELIABILITY_RELIABLE,
250+
RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5);
251+
auto pub_opts = rmw_get_default_publisher_options();
252+
auto * pub = rmw_create_publisher(node, seq_ts, "/tl_emsgsize", &qos, &pub_opts);
253+
ASSERT_NE(nullptr, pub);
254+
auto sub_opts = rmw_get_default_subscription_options();
255+
auto * sub = rmw_create_subscription(node, seq_ts, "/tl_emsgsize", &qos, &sub_opts);
256+
ASSERT_NE(nullptr, sub);
257+
258+
// 32 KiB: above the shrunken buffer, below SHM_PAYLOAD_THRESHOLD so it stays
259+
// on the inline datagram path (a durable-staged payload would bypass the cap).
260+
test_msgs::msg::UnboundedSequences msg;
261+
msg.uint8_values.resize(32 * 1024);
262+
EXPECT_EQ(RMW_RET_ERROR, rmw_publish(pub, &msg, nullptr));
263+
264+
auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub);
265+
auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
266+
}
267+
268+
TEST_F(QosTest, TransientLocalLargeMessageInlineFallbackWhenShmUnavailable)
269+
{
270+
// When shm staging is unavailable, a large latched payload falls back to
271+
// caching inline (no durable segment, no SHM_PAYLOAD_FLAG) and is still
272+
// delivered byte-equal to a late joiner. The test seam forces the failure.
273+
auto seq_ts = rosidl_typesupport_cpp::get_message_type_support_handle<
274+
test_msgs::msg::UnboundedSequences>();
275+
auto qos = make_qos(
276+
RMW_QOS_POLICY_RELIABILITY_RELIABLE,
277+
RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5);
278+
auto pub_opts = rmw_get_default_publisher_options();
279+
auto * pub = rmw_create_publisher(node, seq_ts, "/tl_fallback", &qos, &pub_opts);
280+
ASSERT_NE(nullptr, pub);
281+
282+
setenv("RMW_UDS_TEST_FORCE_SHM_FAILURE", "1", 1);
283+
test_msgs::msg::UnboundedSequences msg;
284+
msg.uint8_values.resize(100 * 1024);
285+
for (size_t i = 0; i < msg.uint8_values.size(); ++i) {
286+
msg.uint8_values[i] = static_cast<uint8_t>((i * 5 + 2) & 0xFF);
287+
}
288+
EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &msg, nullptr));
289+
unsetenv("RMW_UDS_TEST_FORCE_SHM_FAILURE"); // reset before it leaks to other tests
290+
291+
auto * pub_data = static_cast<rmw_uds::UdsPublisher *>(pub->data);
292+
{
293+
std::lock_guard<std::mutex> lock(pub_data->cache_mutex);
294+
ASSERT_EQ(1u, pub_data->message_cache.size());
295+
const auto & cm = pub_data->message_cache.back();
296+
EXPECT_EQ(nullptr, cm.shm_seg)
297+
<< "shm forced unavailable — the large payload must be cached inline";
298+
EXPECT_FALSE(cm.header.msg_type & rmw_uds::SHM_PAYLOAD_FLAG);
299+
}
300+
301+
auto sub_opts = rmw_get_default_subscription_options();
302+
auto * sub = rmw_create_subscription(node, seq_ts, "/tl_fallback", &qos, &sub_opts);
303+
ASSERT_NE(nullptr, sub);
304+
test_msgs::msg::UnboundedSequences trigger;
305+
trigger.uint8_values = {5, 6, 7};
306+
EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &trigger, nullptr));
307+
308+
test_msgs::msg::UnboundedSequences recv;
309+
bool taken = false;
310+
EXPECT_EQ(RMW_RET_OK, rmw_take(sub, &recv, &taken, nullptr));
311+
ASSERT_TRUE(taken) << "late joiner must receive the inline-fallback message";
312+
EXPECT_EQ(msg.uint8_values, recv.uint8_values);
313+
314+
auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub);
315+
auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
316+
}
317+
318+
TEST_F(QosTest, TransientLocalSerializedLargeMessageLateJoiner)
319+
{
320+
// rmw_publish_serialized_message now feeds the TL replay cache and stages
321+
// large latched payloads durably, so a >cap serialized latched message
322+
// reaches a late joiner. Previously that entry point never cached, so the
323+
// message was sent inline once and dropped for anyone who joined later.
324+
auto qos = make_qos(
325+
RMW_QOS_POLICY_RELIABILITY_RELIABLE,
326+
RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5);
327+
auto pub_opts = rmw_get_default_publisher_options();
328+
auto * pub = rmw_create_publisher(node, ts, "/tl_serialized_huge", &qos, &pub_opts);
329+
ASSERT_NE(nullptr, pub);
330+
331+
constexpr size_t big_size = 5 * 1024 * 1024;
332+
rmw_serialized_message_t serialized;
333+
serialized.buffer_capacity = big_size;
334+
serialized.buffer_length = big_size;
335+
serialized.buffer = static_cast<uint8_t *>(std::malloc(big_size));
336+
ASSERT_NE(nullptr, serialized.buffer);
337+
for (size_t i = 0; i < big_size; ++i) {
338+
serialized.buffer[i] = static_cast<uint8_t>((i * 131 + 7) & 0xFF);
339+
}
340+
serialized.allocator = rcutils_get_default_allocator();
341+
342+
// Publish BEFORE any subscriber — must be cached in a durable segment.
343+
EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &serialized, nullptr));
344+
auto * pub_data = static_cast<rmw_uds::UdsPublisher *>(pub->data);
345+
{
346+
std::lock_guard<std::mutex> lock(pub_data->cache_mutex);
347+
ASSERT_EQ(1u, pub_data->message_cache.size());
348+
EXPECT_NE(nullptr, pub_data->message_cache.back().shm_seg);
349+
}
350+
351+
auto sub_opts = rmw_get_default_subscription_options();
352+
auto * sub = rmw_create_subscription(node, ts, "/tl_serialized_huge", &qos, &sub_opts);
353+
ASSERT_NE(nullptr, sub);
354+
355+
uint8_t trig_bytes[] = {1, 2, 3, 4};
356+
rmw_serialized_message_t trig;
357+
trig.buffer = trig_bytes;
358+
trig.buffer_length = sizeof(trig_bytes);
359+
trig.buffer_capacity = sizeof(trig_bytes);
360+
trig.allocator = rcutils_get_default_allocator();
361+
EXPECT_EQ(RMW_RET_OK, rmw_publish_serialized_message(pub, &trig, nullptr));
362+
363+
rmw_serialized_message_t received = rmw_get_zero_initialized_serialized_message();
364+
ASSERT_EQ(RMW_RET_OK, rmw_serialized_message_init(&received, 0, &serialized.allocator));
365+
bool taken = false;
366+
EXPECT_EQ(RMW_RET_OK, rmw_take_serialized_message(sub, &received, &taken, nullptr));
367+
ASSERT_TRUE(taken) << "late joiner must receive the cached 5 MB serialized message";
368+
ASSERT_EQ(big_size, received.buffer_length);
369+
EXPECT_EQ(0, std::memcmp(serialized.buffer, received.buffer, big_size));
370+
371+
EXPECT_EQ(RMW_RET_OK, rmw_take_serialized_message(sub, &received, &taken, nullptr));
372+
ASSERT_TRUE(taken);
373+
ASSERT_EQ(sizeof(trig_bytes), received.buffer_length);
374+
375+
auto _f [[maybe_unused]] = rmw_serialized_message_fini(&received);
376+
std::free(serialized.buffer);
377+
auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub);
378+
auto _r2 [[maybe_unused]] = rmw_destroy_publisher(node, pub);
379+
}
380+
231381
TEST_F(QosTest, KnownSubscriberPathsPrunedOnChurn)
232382
{
233383
// The publisher's known_subscriber_paths must not accumulate dead entries as

0 commit comments

Comments
 (0)