Skip to content

Commit 3ca379b

Browse files
fix(rmw_wait): keep the wait timeout's sub-millisecond resolution (#83)
* fix(rmw_wait): keep the wait timeout's sub-millisecond resolution rmw_wait converted the caller's timeout to whole milliseconds for epoll_wait: a timeout under 1 ms became 1 ms, the deadline was computed from the truncated value, and every re-block rounded the remaining time up. rcl passes the time to the next timer in nanoseconds, so a 1 kHz timer woke up to a whole period late and rclcpp skipped about 6 % of its periods, and a 1.7 ms wait returned RMW_RET_TIMEOUT after 1 ms. Keep the timeout and the deadline in nanoseconds and block with epoll_pwait2. Where it is unavailable (glibc < 2.35 at build time, ENOSYS or EPERM at run time) fall back to epoll_wait with the remaining time rounded up, so the fallback never returns before the deadline either. Test: WaitKeepsSubMillisecondTimeoutResolution and OneKilohertzWaitLoopStaysOnSchedule fail on devel (1.7 ms waits time out after ~1.06 ms; 264 of 500 periods wake more than 0.5 ms late) and pass with the fix. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jh5PFqFmFoZnaSX5FfVv4G * Address review: drop a redundant comment Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jh5PFqFmFoZnaSX5FfVv4G --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 6d9793c commit 3ca379b

4 files changed

Lines changed: 131 additions & 18 deletions

File tree

‎rmw_unix_socket_cpp/DESIGN.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -534,7 +534,7 @@ Each `rmw_wait()` runs the same sequence on the calling thread:
534534
1. **Drain first.** Every subscription, service, and client socket is drained into a per-entity message queue before anything blocks. The receive sockets are `SOCK_DGRAM | SOCK_NONBLOCK`, so the drain loop calls `recv_from` repeatedly and stops cleanly on `EAGAIN` (nothing left to read). This step exists because data may have arrived between the previous `rmw_wait()` and this one; draining up front means such data is not missed.
535535
2. **Check the graph.** The wait set carries its context (stored at `rmw_create_wait_set`), so this step runs for every wait — including a wait set that holds only guard conditions, which is exactly the shape rclcpp's GraphListener uses. The check first drains the context's doorbell socket, then reads the registry's `generation` counter and compares it against the value cached on the context. If it moved, the graph changed: every node's graph guard condition is triggered (see the next subsection). (`TRANSIENT_LOCAL` replay no longer runs here — it is pull-based, served inside `rmw_create_subscription` of the joining process.) The drain-before-read order is half of the lost-wakeup proof; the other half is on the registry's writer side.
536536
3. **Arm the fds.** Every entity fd, guard-condition eventfd, and the context's doorbell fd is added to the epoll instance with `EPOLL_CTL_ADD`. This is idempotent across calls: a still-live fd returns `EEXIST` and is treated as already armed, and a fd number reused after its previous owner closed gets freshly armed. The kernel removes closed fds from an epoll set automatically, so there is no matching `EPOLL_CTL_DEL` and no per-call teardown.
537-
4. **Check ready, then block only if needed.** If any queue already holds a message, or any guard-condition eventfd reads as triggered, the call skips blocking entirely and reports what is ready. Only when nothing is ready does it block in `epoll_wait()`. The thread is asleep in the kernel here, not spinning; there is no internal poll interval. A wake caused only by the doorbell is internal: the wait re-runs the graph check of step 2 and goes back to sleep for the caller's remaining time. `epoll_wait()` is retried on `EINTR` against the same deadline so a signal neither returns a false timeout nor busy-loops.
537+
4. **Check ready, then block only if needed.** If any queue already holds a message, or any guard-condition eventfd reads as triggered, the call skips blocking entirely and reports what is ready. Only when nothing is ready does it block in `epoll_pwait2()`, which takes the caller's timeout with nanosecond resolution so a sub-millisecond wait (the remainder of a 1 kHz timer period) is neither stretched nor cut to a whole millisecond. Where `epoll_pwait2()` is unavailable (glibc < 2.35, kernel < 5.11, or a seccomp profile that blocks it) it falls back to `epoll_wait()` with the timeout rounded up to the next millisecond. The thread is asleep in the kernel here, not spinning; there is no internal poll interval. A wake caused only by the doorbell is internal: the wait re-runs the graph check of step 2 and goes back to sleep for the caller's remaining time. The wait is retried on `EINTR` against the same deadline so a signal neither returns a false timeout nor busy-loops.
538538
5. **Drain again and report.** After waking, the sockets are drained once more, then the output arrays are pruned: entities with no pending data are set to `NULL`, and the ones that are ready are left in place for the executor to service. `RMW_RET_TIMEOUT` is returned only once the caller's own deadline has passed. A wake that drained nothing to take returns `RMW_RET_OK` with every entry `NULL` instead: a drain can legitimately yield nothing — a subscription with `ignore_local_publications` set drops its own context's messages, and a descriptor whose sender is gone is dropped — and that is a spurious wake, not a timeout. An infinite wait therefore never reports `RMW_RET_TIMEOUT`.
539539

540540
### Which guard condition the graph change wakes

‎rmw_unix_socket_cpp/src/rmw_wait.cpp‎

Lines changed: 45 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,11 @@
2020

2121
#include <algorithm>
2222
#include <array>
23+
#include <atomic>
2324
#include <cerrno>
2425
#include <chrono>
2526
#include <cstring>
27+
#include <ctime>
2628
#include <limits>
2729
#include <map>
2830
#include <unordered_map>
@@ -50,6 +52,36 @@ static int64_t wall_now_ns()
5052
std::chrono::system_clock::now().time_since_epoch()).count();
5153
}
5254

55+
// Block in epoll for up to `timeout_ns` nanoseconds; -1 blocks forever.
56+
// epoll_pwait2 (Linux 5.11, glibc 2.35) takes the timeout with nanosecond
57+
// resolution. epoll_wait only takes milliseconds, so it is the fallback where
58+
// epoll_pwait2 is missing: an older glibc at build time, or an older kernel
59+
// (ENOSYS) or a container seccomp profile that blocks it (EPERM) at run time.
60+
static int epoll_wait_ns(
61+
int epfd, struct epoll_event * events, int max_events, int64_t timeout_ns)
62+
{
63+
#if defined(__GLIBC__) && (__GLIBC__ > 2 || (__GLIBC__ == 2 && __GLIBC_MINOR__ >= 35))
64+
static std::atomic<bool> pwait2_unavailable{false};
65+
if (!pwait2_unavailable.load(std::memory_order_relaxed)) {
66+
struct timespec ts;
67+
ts.tv_sec = static_cast<time_t>(timeout_ns / 1000000000);
68+
ts.tv_nsec = static_cast<decltype(ts.tv_nsec)>(timeout_ns % 1000000000);
69+
int n = epoll_pwait2(epfd, events, max_events, timeout_ns < 0 ? nullptr : &ts, nullptr);
70+
if (n >= 0 || (errno != ENOSYS && errno != EPERM)) {
71+
return n;
72+
}
73+
pwait2_unavailable.store(true, std::memory_order_relaxed);
74+
}
75+
#endif
76+
// Round up so the fallback never returns before the caller's deadline.
77+
int timeout_ms = -1;
78+
if (timeout_ns >= 0) {
79+
timeout_ms = static_cast<int>(std::min<int64_t>(
80+
(timeout_ns + 999999) / 1000000, std::numeric_limits<int>::max()));
81+
}
82+
return epoll_wait(epfd, events, max_events, timeout_ms);
83+
}
84+
5385
// Drain a socket into a message queue (subscription, service, or client).
5486
// `shm_cache`/`domain_id` resolve large-payload descriptors: topic, request,
5587
// and response messages can all carry SHM_PAYLOAD_FLAG, so every caller passes
@@ -470,19 +502,17 @@ rmw_ret_t rmw_wait(
470502
}
471503
}
472504

473-
// Compute timeout. -1 means block forever (epoll_wait sentinel).
474-
int timeout_ms = -1;
505+
// Compute timeout in nanoseconds. -1 means block forever.
506+
int64_t timeout_ns = -1;
475507
if (wait_timeout) {
476508
// Accumulate in int64_t; RMW_DURATION_INFINITE (~9.2e12 ms) overflows int.
477509
int64_t ms = static_cast<int64_t>(wait_timeout->sec) * 1000 +
478510
static_cast<int64_t>(wait_timeout->nsec) / 1000000;
479511
if (ms > std::numeric_limits<int>::max()) {
480-
timeout_ms = -1; // Infinite (or beyond epoll's range) -> block forever
512+
timeout_ns = -1; // Infinite (or beyond epoll's range) -> block forever
481513
} else {
482-
timeout_ms = static_cast<int>(ms);
483-
if (timeout_ms == 0 && wait_timeout->nsec > 0) {
484-
timeout_ms = 1; // At least 1ms
485-
}
514+
timeout_ns = static_cast<int64_t>(wait_timeout->sec) * 1000000000 +
515+
static_cast<int64_t>(wait_timeout->nsec);
486516
}
487517
}
488518

@@ -495,9 +525,9 @@ rmw_ret_t rmw_wait(
495525
// re-blocks. RMW_RET_TIMEOUT surfaces only at the caller's own deadline; an
496526
// infinite wait never surfaces a synthetic timeout. EINTR re-enters the
497527
// loop, so a signal neither returns TIMEOUT early nor busy-loops.
498-
const bool infinite = (timeout_ms < 0);
528+
const bool infinite = (timeout_ns < 0);
499529
const int64_t caller_deadline_ns =
500-
infinite ? 0 : steady_now_ns() + static_cast<int64_t>(timeout_ms) * 1000000;
530+
infinite ? 0 : steady_now_ns() + timeout_ns;
501531
// 4. Block on epoll, then drain only the fds it reported ready. This is what
502532
// keeps the wait O(ready) instead of O(entities in the wait set). When step 3
503533
// already found queued work we still make one non-blocking pass, so an entity
@@ -508,26 +538,24 @@ rmw_ret_t rmw_wait(
508538
{
509539
struct epoll_event ready_events[64];
510540
while (true) {
511-
int block_ms = 0;
541+
int64_t block_ns = 0;
512542
if (!poll_only) {
513-
block_ms = -1;
543+
block_ns = -1;
514544
if (!infinite) {
515545
const int64_t rem_ns = caller_deadline_ns - steady_now_ns();
516-
const int64_t rem_ms = (rem_ns > 0) ? (rem_ns + 999999) / 1000000 : 0; // ceil
517-
block_ms = static_cast<int>(
518-
std::min<int64_t>(rem_ms, std::numeric_limits<int>::max()));
546+
block_ns = (rem_ns > 0) ? rem_ns : 0;
519547
}
520548
if ((graph_gc_unwired || !unarmed.empty()) &&
521-
(block_ms < 0 || block_ms > 200))
549+
(block_ns < 0 || block_ns > 200000000))
522550
{
523551
// No doorbell wiring (registry full), or an entity epoll cannot
524552
// watch: never block unbounded, or its event is silently lost
525553
// forever. The early return is a spurious wake (OK, nothing ready)
526554
// that lets the next rmw_wait retry registration / re-drain.
527-
block_ms = 200;
555+
block_ns = 200000000; // 200 ms
528556
}
529557
}
530-
int n = epoll_wait(ws_data->epoll_fd, ready_events, 64, block_ms);
558+
int n = epoll_wait_ns(ws_data->epoll_fd, ready_events, 64, block_ns);
531559
if (n < 0) {
532560
if (errno == EINTR) {
533561
continue;

‎rmw_unix_socket_cpp/test/README.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -130,6 +130,8 @@ Tests the epoll-based wait set.
130130
| `WaitWithGuardCondition` | Pre-triggered guard condition causes immediate return with `RMW_RET_OK` | Validates the "already ready" fast path — avoids unnecessary epoll_wait when data is already available. |
131131
| `WaitTimeoutWhenNoData` | Returns `RMW_RET_TIMEOUT` after specified duration, guard condition nulled out | Validates the timeout path. Without this, the executor would hang indefinitely when no messages arrive. |
132132
| `WaitWithSubscription` | Publish → wait on subscription → returns ready → take succeeds | End-to-end: message arrives on socket → epoll detects it → drain into queue → subscription marked ready → take succeeds. Tests the full executor wakeup path. |
133+
| `WaitKeepsSubMillisecondTimeoutResolution` | 0.3 ms and 1.7 ms waits return `RMW_RET_TIMEOUT` no earlier than the timeout and less than 0.5 ms after it | rcl passes the time to the next timer in nanoseconds. Rounding it to whole milliseconds made a 1.7 ms wait end after 1 ms (a false timeout) and a 0.3 ms wait take 1 ms. |
134+
| `OneKilohertzWaitLoopStaysOnSchedule` | A 1 kHz loop of deadline-based `rmw_wait` calls wakes within 0.5 ms of each period in at least 95 % of 500 periods | This is what an executor does for a 1 kHz timer. With millisecond rounding the loop slid behind schedule and rclcpp skipped periods. |
133135

134136
### test_rmw_graph.cpp
135137

‎rmw_unix_socket_cpp/test/test_rmw_wait.cpp‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,11 @@
1414

1515
#include "test_base.hpp"
1616

17+
#include <algorithm>
1718
#include <atomic>
1819
#include <chrono>
1920
#include <cstring>
21+
#include <limits>
2022
#include <thread>
2123
#include <vector>
2224

@@ -299,6 +301,87 @@ TEST_F(RmwUdsNodeTest, WaitBlocksForFullCallerTimeout)
299301
auto _r2 [[maybe_unused]] = rmw_destroy_guard_condition(gc);
300302
}
301303

304+
TEST_F(RmwUdsNodeTest, WaitKeepsSubMillisecondTimeoutResolution)
305+
{
306+
// Scenario: rcl hands rmw_wait the time left until the next timer with
307+
// nanosecond resolution. Rounding it to whole milliseconds makes a short
308+
// wait return early (1.7 ms -> 1 ms, a false timeout) or late (0.3 ms ->
309+
// 1 ms, a late timer). The wait must last at least the caller's timeout and
310+
// must not be stretched to the next millisecond.
311+
auto * gc = rmw_create_guard_condition(&context);
312+
ASSERT_NE(nullptr, gc);
313+
auto * ws = rmw_create_wait_set(&context, 1);
314+
ASSERT_NE(nullptr, ws);
315+
316+
for (const int64_t timeout_ns : {300000, 1700000}) { // 0.3 ms and 1.7 ms
317+
int64_t shortest_ns = std::numeric_limits<int64_t>::max();
318+
for (int i = 0; i < 20; ++i) {
319+
void * gc_array[1] = {gc->data};
320+
rmw_guard_conditions_t gcs;
321+
gcs.guard_conditions = gc_array;
322+
gcs.guard_condition_count = 1;
323+
324+
rmw_time_t timeout{0, static_cast<uint64_t>(timeout_ns)}; // never triggered
325+
auto t0 = std::chrono::steady_clock::now();
326+
rmw_ret_t ret = rmw_wait(nullptr, &gcs, nullptr, nullptr, nullptr, ws, &timeout);
327+
int64_t elapsed_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
328+
std::chrono::steady_clock::now() - t0).count();
329+
330+
EXPECT_EQ(RMW_RET_TIMEOUT, ret);
331+
EXPECT_GE(elapsed_ns, timeout_ns) <<
332+
"rmw_wait returned TIMEOUT before the caller's " << timeout_ns << " ns deadline";
333+
shortest_ns = std::min(shortest_ns, elapsed_ns);
334+
}
335+
// The shortest of 20 waits is immune to an occasional preemption.
336+
EXPECT_LT(shortest_ns, timeout_ns + 500000) <<
337+
"a " << timeout_ns << " ns wait took at least " << shortest_ns << " ns";
338+
}
339+
340+
auto _r1 [[maybe_unused]] = rmw_destroy_wait_set(ws);
341+
auto _r2 [[maybe_unused]] = rmw_destroy_guard_condition(gc);
342+
}
343+
344+
TEST_F(RmwUdsNodeTest, OneKilohertzWaitLoopStaysOnSchedule)
345+
{
346+
// Scenario: an executor running a 1 kHz timer calls rmw_wait with whatever
347+
// is left of the current 1 ms period. If that remainder is rounded up to a
348+
// whole millisecond, each wake is late by the part of the period already
349+
// spent, so the loop slides behind schedule and rclcpp skips periods.
350+
auto * gc = rmw_create_guard_condition(&context);
351+
ASSERT_NE(nullptr, gc);
352+
auto * ws = rmw_create_wait_set(&context, 1);
353+
ASSERT_NE(nullptr, ws);
354+
355+
constexpr int kPeriods = 500;
356+
constexpr int64_t kPeriodNs = 1000000; // 1 kHz
357+
auto now_ns = [] {
358+
return std::chrono::duration_cast<std::chrono::nanoseconds>(
359+
std::chrono::steady_clock::now().time_since_epoch()).count();
360+
};
361+
362+
int late = 0;
363+
int64_t next_ns = now_ns();
364+
for (int i = 0; i < kPeriods; ++i) {
365+
next_ns += kPeriodNs;
366+
void * gc_array[1] = {gc->data};
367+
rmw_guard_conditions_t gcs;
368+
gcs.guard_conditions = gc_array;
369+
gcs.guard_condition_count = 1;
370+
371+
const int64_t left_ns = std::max<int64_t>(0, next_ns - now_ns());
372+
rmw_time_t timeout{0, static_cast<uint64_t>(left_ns)}; // never triggered
373+
auto _r [[maybe_unused]] = rmw_wait(nullptr, &gcs, nullptr, nullptr, nullptr, ws, &timeout);
374+
if (now_ns() - next_ns > kPeriodNs / 2) {
375+
++late;
376+
}
377+
}
378+
EXPECT_LT(late, kPeriods / 20) <<
379+
late << " of " << kPeriods << " periods woke more than 0.5 ms late";
380+
381+
auto _r1 [[maybe_unused]] = rmw_destroy_wait_set(ws);
382+
auto _r2 [[maybe_unused]] = rmw_destroy_guard_condition(gc);
383+
}
384+
302385
TEST_F(RmwUdsNodeTest, LatchedTopicSurvivesAnUnresponsiveParticipant)
303386
{
304387
// Scenario: one participant on the domain initializes but never services

0 commit comments

Comments
 (0)