feat(listener): watch service and client sockets too - #68
Open
benaliabderrahmane wants to merge 8 commits into
Open
benaliabderrahmane wants to merge 8 commits into
benaliabderrahmane wants to merge 8 commits into
Conversation
Owner
Author
|
Correction to the sanitizer evidence in this PR description. This PR says every test binary is clean under The new TSan CI job (#71) found a pre-existing write-write data race in the seqlock registry — Two things worth being clear about:
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. |
The receive loop existed in four near-identical copies (drain_subscription, drain_socket in rmw_wait.cpp, and inline in rmw_take_request and rmw_take_response) that had drifted: each decided for itself whether to cap the queue, warn about an overflow, or notify a listener. drain_endpoint() is now the only receive loop. DrainTarget carries the per-endpoint policy (queue and its mutex, depth cap, accepted msg_type, shm reader cache, ignore_local, the TRANSIENT_LOCAL watermark map, the listener callback, the name to log) and the drain_target() overloads build it from each impl struct, so a call site can no longer pick its own policy by accident. Where the copies disagreed the stricter one wins: the rmw_wait drain now fires on_new_message and reports overflow, and the request/response queues are capped at SERVICE_QUEUE_DEPTH on the take path too - recorded in drain.hpp as a deliberate tradeoff. Squashed from: - refactor(drain): collapse the four socket drains into one - docs(drain): record the take-path queue cap as a deliberate tradeoff - fix(drain): restore the <cstring> include, retarget the names this PR renamed
drain_endpoint() resolved the shm descriptor and then asked whether the
datagram should be ignored. For a subscription with
ignore_local_publications set, a large same-context publication was
therefore mapped out of the sender's ring and copied into the payload
buffer before being thrown away - up to SHM ring-record size of pointless
copying per message, on a path whose whole purpose is to avoid the copy.
is_same_context() reads only WireHeader::gid, which is on the wire before
anything is resolved, so the check moves ahead of the resolve. That is
the same reasoning the TRANSIENT_LOCAL dedup above it already gives for
its own position ("Checked before the descriptor resolve so a duplicate
never maps a segment").
Behaviour is identical either way: a resolve failure and an ignore both
end in `continue`. So the new test does not fail beforehand - it covers a
branch nothing reached. The two existing ignore_local tests publish small
inline payloads, which carry no descriptor, so the shm path combined with
ignore_local had no coverage at all.
Full suite green on Jazzy: 157 tests, 0 failures.
The headline behaviour change of the unified drain - rmw_wait now fires on_new_message, where only the rmw_take path did - had no test on this branch. WaitDrainFiresOnNewMessageCallback publishes three messages, waits without ever taking, and expects the callback to have been credited three events. It fails on devel (0 events) and passes here. The total is summed and awaited so the test stays valid once a listener thread delivers it. Also corrects the SERVICE_QUEUE_DEPTH comment: services and clients do carry a QoS depth, it is just not enforced on these queues. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NUNQNo26cKRPrVXcaHZnje
on_new_request_cb and on_new_response_cb were stored by their setters and flushed once against an already-queued backlog, but no drain ever fired them again - a service or client was callback-dead the moment its registration returned. With one drain filling all three queues, the two drain_target() overloads now carry the callback. The notification also moves out of the receive loop: one call per drain that grew the queue, carrying enqueued - dropped. number_of_events is a take credit, so an overflow that pushed 100 and popped 90 reports 10, and a drain whose net gain is zero stays silent. The setters flush a queued backlog only when a callback takes over from none, because rclcpp registers twice in a row on purpose and paid the same backlog out twice. Squashed from: - fix(drain): notify services and clients, and report the batch count - fix(drain): report the queue's growth, and flush a backlog only once
No user code runs under queue_mutex any more: the setters copy the backlog size, release the queue lock and notify holding callback_mutex alone, and drain_endpoint() notifies the same way. That was only half of it. rmw_take drains before it pops, so a take from inside the callback re-enters drain_endpoint() on the thread already inside the callback, and a nested drain that gains a datagram notifies again - locking callback_mutex a second time on the same thread. With a std::mutex that is a self-deadlock, seen only when a message arrives during the callback. callback_mutex is now a std::recursive_mutex; it still keeps the callback pointer and user_data stable for the duration of a call, which is all it was ever for. ACallbackMayTakeWhileItsOwnEndpointGainsMessages has the callback publish before it takes, so the nested drain gains one every time. It hangs on the previous code and passes in ~130 ms after. Squashed from: - fix(callbacks): never run a listener callback under queue_mutex - fix(callbacks): make callback_mutex recursive so a callback can really take - test(callbacks): name the re-entrant fixture PublishThenTakeCallback Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NUNQNo26cKRPrVXcaHZnje
rclcpp's EventsExecutor never calls rmw_wait and only calls rmw_take once a listener callback has told it there is something to take. With delivery happening only inside rmw_wait that is a hang by construction: nothing drains the socket, so the callback never fires. 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, on the first callback registration, so an executor that waits still starts no background thread (pinned 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 destroying an endpoint blocks until an in-flight drain and the callback it fires have returned; the destroy paths unwatch first. - A per-endpoint drain_mutex serialises the listener against an rmw_wait/rmw_take draining the same socket, so one publisher's samples cannot be reordered by interleaved recv/push pairs. It covers the receive loop only and is released before the notification, so a callback that takes from its own endpoint re-enters drain_endpoint() without wedging on it; a callback runs holding callback_mutex alone. - Each wait set owns a delivery eventfd. The listener signals every registered one strictly after enqueueing and rmw_wait drains its own strictly before scanning its queues, which closes the lost-wakeup window without one wait set consuming another's credit. The thread wakes on socket readiness, never on a clock; it does not make deadline or liveliness enforceable. Squashed from: - feat(listener): lazily-started thread for callback-driven delivery - fix(listener): give each wait set its own delivery eventfd - fix(listener): close the shutdown/registration race and serialise drains - test(listener): wait for the event total instead of racing the listener - fix(drain): release drain_mutex before notifying, and allow a re-entrant take - docs(drain): drain_mutex covers the receive loop, not the whole drain Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NUNQNo26cKRPrVXcaHZnje
Only rmw_subscription_set_on_new_message_callback handed its socket to the listener, so a service or client under an EventsExecutor still had nobody moving its datagrams: the callback fired for the backlog its setter flushed, then never again. The two setters now watch on registration and unwatch when the callback is cleared, and rmw_destroy_service / rmw_destroy_client unwatch before tearing anything down - listener_unwatch blocks until an in-flight drain has returned, so the delete behind it cannot race one. Squashed from: - feat(listener): watch service and client sockets too - test(listener): the service and client totals race the listener too
Adding the service and client halves would have made a second and third copy of the same protocol - flush the backlog only when a callback takes over from none, call listener_watch outside callback_mutex, roll the callback back if the watch fails - and the copies had already drifted once. listener_set_callback() in listener.cpp owns it; DrainTarget already carries the fd, queue, mutexes and callback slots, so the setters keep their argument checks and hand the rest over in three lines each. The callback runs holding callback_mutex alone, a std::recursive_mutex here as in the layers below. Squashed from: - refactor(listener): one set_on_new_*_callback implementation, not three - fix(callbacks): run a listener callback under callback_mutex alone - fix(listener): follow callback_mutex to std::recursive_mutex in the shared setter Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NUNQNo26cKRPrVXcaHZnje
benaliabderrahmane
force-pushed
the
feat/listener-services-clients
branch
from
September 18, 2026 12:44
1be48f0 to
2a2c88d
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
The listener thread already drained and dispatched all three endpoint kinds; only
rmw_subscription_set_on_new_message_callbackhanded it a socket. A service orclient under an
EventsExecutortherefore still had nobody to move itsdatagrams: the callback fired for the backlog its setter flushed, and then never
again, because the executor that registered it never calls
rmw_wait.The two setters now watch on registration and unwatch when the callback is
cleared, and
rmw_destroy_service/rmw_destroy_clientunwatch before tearinganything down —
listener_unwatchblocks until an in-flight drain has returned,so the delete behind it cannot race one.
Registration itself no longer lives in the setters. Adding the service and client
halves would have made a second and third copy of the same protocol — flush the
backlog only when a callback takes over from none, call
listener_watchoutsidecallback_mutex, roll the callback back if the watch fails — and the copies hadalready drifted once. So it is one exported
listener_set_callback(UdsContext *, const DrainTarget &, kind, entity, uid, callback, user_data)in
listener.cpp.DrainTargetalready carries the endpoint's fd, queue,queue_mutex,callback_mutexand callback slots, anddrain_target()alreadybuilds one for each of the three endpoint kinds, so the helper needs no new
struct and no new overload set: the setters keep their argument checks and hand
the rest over in three lines each. That makes this PR rewrite
rmw_subscription_set_on_new_message_callbackas well — it loses 43 lines for 3,the service and client setters 14 for 3 each, against 45 lines of helper in
listener.cppand 13 of declaration inlistener.hpp(which now includesdrain.hppfor theDrainTarget). The mutexes and their order are the ones eachsetter already had.
The watch and the unwatch sit outside
callback_mutex: the listener holdslistener_mutexacross a drain and takescallback_mutexinside it, soacquiring them the other way round would deadlock against a drain in flight.
That rule now lives with the only code that has to obey it — the comment inside
listener_set_callback, and the contract inlistener.hpp. Writing the serviceand client halves out in place would have meant a copy in each of the three
setters to keep in sync.
Is this user-facing behavior change?
Yes — a service or client with a listener callback registered now delivers
without any thread in
rmw_wait, and contributes to starting the listenerthread if it is the first registration in the context.
Both setters can also fail now, which they could not before: they used to end in
an unconditional
return RMW_RET_OK;, and they now returnlistener_watch'scode —
RMW_RET_INVALID_ARGUMENTfor a bad fd,RMW_RET_ERRORfrom any of itsepoll, eventfd or listener-thread failures. On that failure the helper clears the
callback back out under
callback_mutex, so a failure means the callback is notinstalled.
rmw_subscription_set_on_new_message_callbackalready behaved thatway; the service and client entry points did not. A context already past
rmw_shutdownis not one of those failures:listener_watchreturnsRMW_RET_OKwithout watching, because the thread it would have to restart hasalready been joined.
How was this tested?
Two new tests,
ServiceCallbackFiresWithNoThreadInWaitOrTakeandClientCallbackFiresWithNoThreadInWaitOrTake. Both fail on the parent commit,timing out after the full 2 s budget; both pass here.
Full suite on jazzy: 176 tests, 0 failures, 10 repeat runs of
test_rmw_listener_callbackswith no flakes, and clean under-fsanitize=threadwith deadlock detection on — the only TSan reports left inthe suite are the pre-existing
registry.cppseqlock ones, which this stack doesnot touch.
-fsanitize=address,undefinedwith leak detection is also cleanacross all 16 test binaries at this branch's tip (
969168b), 0 sanitizererrors and 0 failures. kilted, rolling and lyrical are on CI.
Did you use Generative AI?
Additional Information
Builds on the listener thread PR; merge that first.