fix(drain): notify services and clients, and report the batch count - #66
Open
benaliabderrahmane wants to merge 5 commits into
Open
benaliabderrahmane wants to merge 5 commits into
benaliabderrahmane wants to merge 5 commits into
Conversation
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
benaliabderrahmane
force-pushed
the
fix/drain-notifies-services-and-clients
branch
from
September 18, 2026 12:44
8cd872e to
c3f0e94
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
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
on_new_request_cbandon_new_response_cbwere stored by their setters andflushed 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: every request and response after that arrived without notification.
Now that one drain fills all three queues, populating the callback fields in the
two
drain_target()overloads is all that bug needed. Two corrections to thenotification protocol ride along with it, one in
drain_endpoint()and one inall three
set_on_new_*_callback()setters.The notification also moves out of the receive loop. It fired once per datagram
with a count of
1; it now fires once per drain that grew the queue, carryingthe number of entries it gained.
rmw/event_callback_type.hdefinesnumber_of_eventsas the count since the callback was last called, explicitlyallows
> 1, and says it should never be0.The count is how much the queue grew —
enqueued - dropped— not how manydatagrams the socket handed over.
number_of_eventsis a take credit: theexecutor calls
take()once per event it is told about, so an overflow thatpushed 100 and popped 90 must report 10, or the executor spends 90 takes finding
nothing. There is at most one pop per push, so the subtraction never underflows,
and a drain whose net gain is zero now stays silent for the same reason an empty
drain does.
The second correction is in the setters: all three flush the queued backlog only
when a callback takes over from none. rclcpp installs the callback twice in a
row on purpose — once against a stack temporary, then again against its
permanent storage, to close the window where the
std::functionis replaced butthe middleware still holds the old pointer (
subscription_base.hpp) — and anunconditional flush pays the same backlog out for both calls. A TRANSIENT_LOCAL
subscription with ten replayed samples was handing the executor twenty events.
Is this user-facing behavior change?
Yes:
queue, not just on the setter's one-shot flush.
number_of_eventsis now a batch count rather than always1. A drain thatgains ten entries reports ten once instead of one ten times, and a drain whose
arrivals were all popped again by the overflow trim reports nothing at all.
reports that backlog once instead of twice. Because rclcpp registers twice in
a row, a TRANSIENT_LOCAL subscription with ten replayed samples reported
twenty events and now reports ten.
One judgment call for review: the flush fires only on the none→callback
transition, so replacing a non-null callback with a different non-null one
while a backlog is queued hands the new callback nothing — it hears about those
entries only on the next drain that grows the queue. That is the cost of closing
the double payout, and the re-registration rclcpp actually performs is the case
the guard was written for, but if we would rather flush on every transition into
a non-null callback, this is the line to say so on.
How was this tested?
New
test/test_rmw_listener_callbacks.cpp, 8 tests: all three endpoint typesnotified from an
rmw_wait-driven drain, the batch count, that the countexcludes datagrams the overflow trim popped, that a second registration does not
re-flush the backlog, the never-zero rule, and that clearing a callback stops
notification.
Five of the eight fail on the parent commit: both service and client tests
see no callback at all;
SubscriptionCallbackReportsTheBatchCountsees threecalls of
1instead of one call of3;BatchCountExcludesDatagramsDroppedByOverflowsees 15 events for a depth-10queue instead of 10; and
ReRegisteringTheCallbackDoesNotReflushTheBacklogsees6 events in 2 calls for 3 queued messages instead of 3 in 1.
Full suite on jazzy: 165 tests, 0 failures. kilted, rolling and lyrical are
on CI.
Did you use Generative AI?
Additional Information
Builds on the drain unification PR; merge that first.