Skip to content

fix(drain): notify services and clients, and report the batch count - #66

Open
benaliabderrahmane wants to merge 5 commits into
develfrom
fix/drain-notifies-services-and-clients
Open

benaliabderrahmane wants to merge 5 commits into
develfrom
fix/drain-notifies-services-and-clients

Conversation

@benaliabderrahmane

@benaliabderrahmane benaliabderrahmane commented Sep 8, 2026

Copy link
Copy Markdown
Owner

Description

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: 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 the
notification protocol ride along with it, one in drain_endpoint() and one in
all 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, carrying
the number of entries it gained. rmw/event_callback_type.h defines
number_of_events as the count since the callback was last called, explicitly
allows > 1, and says it should never be 0.

The count is how much the queue grewenqueued - dropped — not how many
datagrams the socket handed over. number_of_events is a take credit: the
executor calls take() once per event it is told about, so an overflow that
pushed 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::function is replaced but
the middleware still holds the old pointer (subscription_base.hpp) — and an
unconditional 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:

  • Service and client listener callbacks now fire on every drain that grows the
    queue, not just on the setter's one-shot flush.
  • number_of_events is now a batch count rather than always 1. A drain that
    gains 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.
  • An endpoint that already has messages queued when its callback is installed
    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 types
notified from an rmw_wait-driven drain, the batch count, that the count
excludes 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; SubscriptionCallbackReportsTheBatchCount sees three
calls of 1 instead of one call of 3;
BatchCountExcludesDatagramsDroppedByOverflow sees 15 events for a depth-10
queue instead of 10; and ReRegisteringTheCallbackDoesNotReflushTheBacklog sees
6 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.

Abderahmane BENALI and others added 5 commits September 18, 2026 14:36
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
benaliabderrahmane force-pushed the fix/drain-notifies-services-and-clients branch from 8cd872e to c3f0e94 Compare September 18, 2026 12:44
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant