Skip to content

[core] Keep RaySyncer reactors alive for every handler dispatched to io_context #67399

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lonexreb wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[core] Keep RaySyncer reactors alive for every handler dispatched to io_context #67399

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2011efd
[core] Keep RaySyncer reactors alive for every handler dispatched to …
lonexreb Oct 25, 2026
ca7f6d6
Apply clang-format
lonexreb Oct 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions src/ray/ray_syncer/ray_syncer_bidi_reactor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,8 +120,11 @@ class RaySyncerBidiReactor {
/// See MessagesReceivedCallback in common.h.
MessagesReceivedCallback on_messages_received_;

/// Self-owned shared_ptr for lifetime guarding in async callbacks (timer/dispatch). Set
/// by caller/factory, released in OnDone.
/// Self-owned shared_ptr for lifetime guarding in async callbacks. Set by the
/// caller/factory before any gRPC callback can fire and released in OnDone. Every
/// handler dispatched to io_context_ captures a copy so that it can never outlive
/// the reactor; the batch timer holds a weak_ptr instead so that a pending timer
/// does not keep a finished reactor alive.
std::shared_ptr<RaySyncerBidiReactor> self_ref_;

private:
Expand Down
11 changes: 8 additions & 3 deletions src/ray/ray_syncer/ray_syncer_bidi_reactor_base.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,9 +233,14 @@ class RaySyncerBidiReactorBase : public RaySyncerBidiReactor, public T {
using T::StartRead;
using T::StartWrite;

// NOTE: Every handler dispatched to io_context_ below captures a strong
// reference to the reactor (self_ref_). OnDone() releases self_ref_ from a
// handler on the same io_context_, and gRPC may invoke OnReadDone/OnWriteDone
// before that handler runs; without the capture, a handler queued earlier
// could run after the reactor has been destroyed (use-after-free).
void OnWriteDone(bool ok) override {
io_context_.dispatch(
[this, disconnected = IsDisconnected(), ok]() {
[this, self = self_ref_, disconnected = IsDisconnected(), ok]() {
if (*disconnected) {
Comment on lines 241 to 244

@gemini-code-assist gemini-code-assist Bot Oct 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To prevent potential compiler warnings about unused lambda captures (such as -Wunused-lambda-capture in Clang), it is recommended to explicitly cast the captured self to void inside the lambda body. This also clearly documents that the capture is intended solely for lifetime management.

Suggested change
void OnWriteDone(bool ok) override {
io_context_.dispatch(
[this, disconnected = IsDisconnected(), ok]() {
[this, self = self_ref_, disconnected = IsDisconnected(), ok]() {
if (*disconnected) {
void OnWriteDone(bool ok) override {
io_context_.dispatch(
[this, self = self_ref_, disconnected = IsDisconnected(), ok]() {
(void)self;
if (*disconnected) {

@lonexreb lonexreb Oct 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not adding (void)self;: clang's -Wunused-lambda-capture doesn't fire for captures whose type has a non-trivial destructor like std::shared_ptr, since holding the object alive is a legitimate use. The macOS build compiles with -Werror (.bazelrc) and bazel test //src/ray/ray_syncer/tests:ray_syncer_test built and passed cleanly with clang. The comment in ray_syncer_bidi_reactor.h documents that these captures exist for lifetime.

return;
}
Expand All @@ -253,7 +258,7 @@ class RaySyncerBidiReactorBase : public RaySyncerBidiReactor, public T {
void OnReadDone(bool ok) override {
if (!ok) {
io_context_.dispatch(
[this]() {
[this, self = self_ref_]() {
RAY_LOG_EVERY_MS(INFO, 1000) << "Failed to read a message from node: "
<< NodeID::FromBinary(GetRemoteNodeID());
Disconnect();
Expand Down Expand Up @@ -309,7 +314,7 @@ class RaySyncerBidiReactorBase : public RaySyncerBidiReactor, public T {
// task is queued and this read must post one. If not, a task is already queued and
// will pick up all messages in the buffer later.
if (buffer_was_empty) {
io_context_.dispatch([this]() { ProcessBufferedSyncMessages(); },
io_context_.dispatch([this, self = self_ref_]() { ProcessBufferedSyncMessages(); },
"RaySyncer.ProcessBufferedSyncMessages");
}
}
Expand Down
7 changes: 5 additions & 2 deletions src/ray/ray_syncer/ray_syncer_client.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,11 @@ RayClientBidiReactor::RayClientBidiReactor(
}

void RayClientBidiReactor::OnDone(const grpc::Status &status) {
// `self` keeps the reactor alive until this handler has finished, including
// through cleanup_cb_; resetting self_ref_ then drops the last owning
// reference once the handler returns.
io_context_.dispatch(
[this, status]() {
[this, self = self_ref_, status]() {
cleanup_cb_(this, !status.ok());
self_ref_.reset();
Comment on lines 61 to 64

@gemini-code-assist gemini-code-assist Bot Oct 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To prevent potential compiler warnings about unused lambda captures (such as -Wunused-lambda-capture in Clang), it is recommended to explicitly cast the captured self to void inside the lambda body. This also clearly documents that the capture is intended solely for lifetime management.

  io_context_.dispatch(
      [this, self = self_ref_, status]() {
        (void)self;
        cleanup_cb_(this, !status.ok());
        self_ref_.reset();

@lonexreb lonexreb Oct 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not adding (void)self;: clang's -Wunused-lambda-capture doesn't fire for captures whose type has a non-trivial destructor like std::shared_ptr, since holding the object alive is a legitimate use. The macOS build compiles with -Werror (.bazelrc) and bazel test //src/ray/ray_syncer/tests:ray_syncer_test built and passed cleanly with clang. The comment in ray_syncer_bidi_reactor.h documents that these captures exist for lifetime.

},
Expand All @@ -65,7 +68,7 @@ void RayClientBidiReactor::OnDone(const grpc::Status &status) {

void RayClientBidiReactor::DoDisconnect() {
io_context_.dispatch(
[this]() {
[this, self = self_ref_]() {
StartWritesDone();
// Free the hold to allow OnDone being called.
RemoveHold();
Expand Down
10 changes: 7 additions & 3 deletions src/ray/ray_syncer/ray_syncer_server.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,16 +109,20 @@ RayServerBidiReactor::RayServerBidiReactor(
}

void RayServerBidiReactor::DoDisconnect() {
io_context_.dispatch([this]() { Finish(grpc::Status::OK); }, "");
io_context_.dispatch([this, self = self_ref_]() { Finish(grpc::Status::OK); }, "");
}

void RayServerBidiReactor::OnCancel() {
io_context_.dispatch([this]() { Disconnect(); }, "");
io_context_.dispatch([this, self = self_ref_]() { Disconnect(); }, "");
}

void RayServerBidiReactor::OnDone() {
// See RayClientBidiReactor::OnDone for why `self` is captured here.
io_context_.dispatch(
[this, cleanup_cb = cleanup_cb_, remote_node_id = GetRemoteNodeID()]() {
[this,
self = self_ref_,
cleanup_cb = cleanup_cb_,
remote_node_id = GetRemoteNodeID()]() {
cleanup_cb(this, false);
self_ref_.reset();
Comment on lines 121 to 127

@gemini-code-assist gemini-code-assist Bot Oct 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To prevent potential compiler warnings about unused lambda captures (such as -Wunused-lambda-capture in Clang), it is recommended to explicitly cast the captured self to void inside the lambda body. This also clearly documents that the capture is intended solely for lifetime management.

  io_context_.dispatch(
      [this,
       self = self_ref_,
       cleanup_cb = cleanup_cb_,
       remote_node_id = GetRemoteNodeID()]() {
        (void)self;
        cleanup_cb(this, false);
        self_ref_.reset();

@lonexreb lonexreb Oct 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not adding (void)self;: clang's -Wunused-lambda-capture doesn't fire for captures whose type has a non-trivial destructor like std::shared_ptr, since holding the object alive is a legitimate use. The macOS build compiles with -Werror (.bazelrc) and bazel test //src/ray/ray_syncer/tests:ray_syncer_test built and passed cleanly with clang. The comment in ray_syncer_bidi_reactor.h documents that these captures exist for lifetime.

},
Expand Down
Loading