feat(listener): lazily-started thread for callback-driven delivery

[benaliabderrahmane]

Description

rclcpp's EventsExecutor — the default in performance_test and
ros2-benchmark-container — never calls rmw_wait, and only calls rmw_take
once a listener callback has told it there is something to take. With delivery
happening exclusively inside rmw_wait, that is a deadlock by construction:
nothing drains the socket, so the callback never fires; the callback never fires,
so rmw_take is never called. Registering the callback is the entire
interaction, and the executor then waits forever. This is why the crash fixed
in #61 turned into a silent hang rather than working delivery.

Each context can now run one listener thread that drains a watched socket and
fires its callback with no application thread involved.

It starts lazily, and only on a callback registration. An executor that waits
registers nothing, so a SingleThreadedExecutor or MultiThreadedExecutor
process still starts no background thread at all — async delivery is opt-in by
what the application does, not by configuration. There is a test asserting that
against /proc/self/task.

Three things make it safe to share an endpoint with rmw_wait:

• listener_mutex is held across the drain, and listener_unwatch() takes
  it, so a thread destroying an endpoint blocks until the drain — and the callback
  it fires — has returned. rmw_destroy_subscription unwatches before it frees
  anything.

• A per-wait-set delivery_fd closes a lost-wakeup window. Both the listener
  and a wait set may drain the same socket, which is safe in itself, but if the
  listener wins the race between a wait's queue scan and its epoll_wait then the
  socket is empty, epoll has nothing left to report, and the wait would sleep
  with a full queue. The listener signals delivery_fd strictly after
  enqueueing and rmw_wait drains it strictly before scanning its queues, so
  a message either lands in that scan or leaves the level-triggered eventfd
  readable. Same ordering pair as ring_doorbells.

  The fd belongs to the wait set rather than to the listener. rmw_create_wait_set
  creates it and registers it in context->delivery_fds; the listener writes every
  registered fd after each enqueue. One shared per-context fd is one credit,
  because an eventfd read drains the whole counter — a wait set that gained no
  work of its own still consumed that credit and threw it away, leaving the wait
  set that needed it asleep on a socket the listener had already emptied. With a
  finite timeout the message still surfaces at the caller's deadline, so the cost
  is latency; an infinite wait stalls until the next drain in the context
  enqueues something, which for the last message of a burst or a one-shot client
  response is never. Creating it with the wait set also means it exists before any
  listener can, so a callback registered while a wait is already blocked still has
  an armed fd to signal; resolving it from a listener-running flag at the top of
  rmw_wait, which is what an earlier commit on this PR did, armed nothing at all
  for exactly that case. It is therefore armed unconditionally, and
  rmw_destroy_wait_set deregisters under listener_mutex strictly before the
  close, or the listener writes 8 bytes into a recycled fd number.

  Because the listener signals every wait set for any endpoint it drains, the wake
  itself is not progress for the caller: the queue scan is factored into
  caller_queue_has_work() and the new ARMED_DELIVERY dispatch re-runs it
  instead of treating the wake as work.

• A per-endpoint drain_mutex keeps the queue in publisher order.
  drain_endpoint recv()s outside queue_mutex and push_back()s inside it,
  so two drains of one endpoint — the listener thread and an rmw_wait or
  rmw_take on an application thread, which is the configuration this PR makes
  routine — can interleave as recv A / recv B / push B / push A and land
  same-publisher datagrams out of order. queue_mutex prevents corruption, not
  reordering, and ROS 2 guarantees per-publisher FIFO. The mutex is held for the
  whole drain, taken before queue_mutex and callback_mutex, and nothing takes
  listener_mutex while holding it, so the lock graph stays acyclic.

One more race is worth naming, because it is reachable through the public API with
no misuse. listener_stop has to release listener_mutex before it joins — the
loop takes that mutex per event, so joining under it would deadlock — and a
registration landing in that window found listener_running already false and
called listener_start, which replaces the epoll and wake fds the pending join
depends on and move-assigns over a still-joinable std::thread. Both halves are
fatal: std::terminate, or a join that never returns. is_shutdown does not
cover it on its own, since rmw_context_fini reaches listener_stop without ever
setting it, so a listener_stopping flag is now set across the join and read
together with is_shutdown under the mutex. listener_start also catches the
std::system_error std::thread throws when the process is out of threads,
rather than letting it escape an extern "C" entry point.

Is this user-facing behavior change?

Yes — this is the significant one in the series.

• A process that registers a listener callback now has one extra thread. The
  zero-background-thread property still holds for every executor that waits, but
  it is no longer unconditional. Documented in the docs PR in this series.
• Every wait set now costs one eventfd, and every rmw_wait one extra
  read(). The delivery fd is created with the wait set and armed
  unconditionally, so a process that never registers a callback pays one fd per
  wait set and one EAGAIN read per wait where it previously paid nothing.
  Gating it on a listener-running flag is what produced the lost wakeup above,
  so the cost is the price of the fix rather than an oversight.
• EventsExecutor works. Subscriptions deliver with no thread in rmw_wait.
• A listener callback must not block, and must not touch any endpoint in its
  context — listener_mutex is one per context and is held across the drain
  that fires the callback, so destroying, registering on, or clearing the callback
  of an endpoint, and creating or destroying a wait set, deadlock against the
  drain the callback was called from. listener.hpp states that rule for all
  three endpoint kinds; the paths that actually take the mutex at this point in
  the stack are subscription registration and destroy, plus wait set
  create/destroy, and services and clients join them in the next PR. It must not
  rmw_take its own endpoint either: the callback fires inside drain_endpoint,
  which holds that endpoint's drain_mutex. rclcpp's callbacks only push to a
  queue, which is what rmw/event_callback_type.h expects of them, so this holds
  in practice.
• Registration failure is returned rather than swallowed, and the callback is
  rolled back with it. An endpoint whose socket cannot be watched delivers only
  from rmw_wait, which is exactly the silent hang above, so rclcpp should
  raise it at construction instead — and un-installing the callback is what makes
  the error true, rather than leaving the caller with a failure and a callback
  that still fires. Worth a second opinion during review — the alternative is
  logging and returning OK. The one deliberate exception is a context already past
  rmw_shutdown or inside listener_stop: listener_watch returns OK without
  watching, because restarting the thread for an endpoint on its way out would
  leak it.

How was this tested?

test_rmw_listener_callbacks.cpp grows to 17 tests. Two fail on the parent
commit, and one of them fails by reproducing the hang:

• CallbackFiresWithNoThreadInWaitOrTake — times out after the full 2 s budget on
  the parent, passes in 5 ms here. This is the bug itself.
• NoListenerThreadUntilACallbackIsRegistered — no thread is started on the
  parent.

Also added: destroy-while-messages-arrive churn (20 rounds) for the lifetime
rule, and a 50-round publish-into-a-blocked-rmw_wait stress for the
delivery_fd window. That last one drives the race rather than interleaving it
exactly — a deterministic test of that window would need fault injection.
ASecondWaitSetDoesNotStealTheDeliveryWakeup runs those same 50 rounds with a
second wait set spinning rmw_wait against them, and
DestroyingWaitSetsWhileTheListenerDeliversIsSafe churns 200 wait-set
create/destroy rounds against a listener that is delivering the whole time, for
the recycled-fd path. Neither lost wakeup is directly unit-testable — both need
the victim to be between its queue scan and its epoll_wait, a window a test
cannot open from outside — so these drive the paths repeatedly instead, and the
recycled-fd one pays off under the sanitizer jobs.

Two more regression tests came out of reviewing this PR, both failing on its own
earlier commits:

• RegisteringACallbackWhileTheContextShutsDownDoesNotAbort — 250 rounds, four
  concurrent registrar threads per round. It hung forever before (main in
  futex_wait on the join, the listener in ep_poll, zero CPU); its 250 rounds
  now finish in 81 ms.
• ConcurrentDrainsPreservePublisherOrder — 1500 samples at depth 5000, asserting
  zero inversions. Before the drain_mutex it reported "first inversion: 531 then
  530" with all 1500 samples collected, so the gap is reordering rather than
  overflow.

Three of the parent's tests change with the listener, since registering a callback
now also hands the socket to a second drainer.
SubscriptionCallbackFiresFromWait becomes
SubscriptionCallbackFiresExactlyOnceOnDelivery and
SubscriptionCallbackReportsTheBatchCount drops its calls == 1 assertion:
which path delivers, and how many batches a burst is split into, is no longer the
middleware's to promise — the event total is.
BatchCountExcludesDatagramsDroppedByOverflow becomes
BatchCountMatchesWhatCanBeTaken and compares credits against takes rather than
against qos.depth, because how much of a 15-message burst survives the socket
buffer is timing and asserting it flaked 7 runs in 15 here. It fails 6 of 6 on the
old count and passes 20 of 20 now.

Full suite on jazzy: 174 tests, 0 failures, and 12 repeat runs of
test_rmw_listener_callbacks with no flakes. All 16 test binaries are clean under
-fsanitize=thread — deadlock detection confirmed live against a
lock-order-inversion control, so the new mutex is covered. The only TSan reports
left in the suite are the pre-existing registry.cpp seqlock ones, which this
stack does not touch. -fsanitize=address,undefined with leak detection was
clean on this PR's first commit and has not been re-run since the per-wait-set
delivery fd and the drain_mutex landed; it would not be the evidence for the
join in any case, since LeakSanitizer tracks heap allocations only, and a process
that leaks an epoll fd and an eventfd and detaches an unjoined thread still exits
0 under it. What covers the join is the shutdown-race test above, which reproduces
by hanging or aborting.

End-to-end, with a real rclcpp::experimental::executors::EventsExecutor node
(measured with the whole series applied): devel receives 0 of 5 messages,
this receives 5 of 5.

kilted, rolling and lyrical are on CI.

Did you use Generative AI?

Additional Information

Scope is subscriptions. The listener itself is endpoint-agnostic and already
dispatches all three kinds, but only
rmw_subscription_set_on_new_message_callback registers; services and clients
are the next PR in the series. The drain_mutex is added to all three endpoint
types anyway, since rmw_wait and rmw_take already race each other there.

drain_endpoint returns a count again: the number of datagrams it pushed into the
queue, before the QoS trim, which is what the listener tests before signalling the
delivery fds. That is deliberately not the batch count the callback is handed —
since the parent PR that number is the queue's net growth, enqueued - dropped —
because the signal is about whether the drain moved anything into a queue, not
about take credits.

The thread is joined in rmw_shutdown, not rmw_context_fini, so an endpoint
destroyed between those two calls is not drained by a live thread;
rmw_context_fini repeats the call for a context finalized without a shutdown.

[benaliabderrahmane]

Correction to the sanitizer evidence in this PR description.

This PR says every test binary is clean under -fsanitize=thread. That was one run per binary, and it is weaker evidence than the wording implies.

The new TSan CI job (#71) found a pre-existing write-write data race in the seqlock registry — write_slot_payload (registry.cpp:235) against teardown_slot (registry.cpp:319), because registry_remove publishes ENTRY_EMPTY before zeroing the payload, letting a concurrent try_add_once claim the slot and write it at the same time. Reproduced locally at 4 of 40 runs of test_registry_concurrent, so a single clean run has roughly a 90% chance of missing it.

Two things worth being clear about:

• It is not caused by this PR. It reproduces on plain devel, it is in the registry rather than the listener, and it is already diagnosed and fixed on the unmerged fix/registry-teardown-before-slot-release (a05ba3f).
• The listener-specific claims still hold, but on the same caveat: the lifetime and lock-order checks here were single runs. DestroySubscriptionWhileMessagesArrive does 20 rounds and WaitAndListenerOnTheSameSubscriptionDoNotHang does 50, so those exercise their own interleavings repeatedly, but the sanitizer sweep behind them was not repeated.

No change requested here — the four-distro CI on this PR is green and this is a separate, older bug. Flagging it so the sanitizer line in the description is not read as stronger than it is.