[core][taskEvents out of GCS][2.5/n] Optimize ray task event recorder by karticam · Pull Request #65288 · ray-project/ray

karticam · 2026-08-07T16:07:38Z

When running benchmarks on #64835 with flags on that trigger the ray task event recorder flow - namely:

enable_ray_event: true
enable_ray_task_event_recorder: true
enable_core_worker_task_event_to_gcs: false (GCS fed only via recorder->aggregator)

I found some regressions. This PR fixes those regressions.

There were 3 optimizations made to ray_task_event_recorder.
The 3 optimizations are applied as different commits in the PR (added below with the explanations). For easier review, go through the commits separately.

Optimizations:

c6efc55 [opt1] : Earlier all of ExportEvents function operated under the mutex. The function would first process the status events and profile events along with skipped events to make form the object to send to aggregator agent. Then it would do group, merge and serialize those event and then send it to the aggregator agent. Given the max buffer size which could be 100k potentially, this could take a lot of time and holding the mutex for long could lead to other AddEvents request to be stalled. The fix is to release the mutex as soon as the shared queues are drained out in local queues and then do the rest of the processing.
005e5d7 [opt2]: Earlier, the ray event protos for task status events (RayTaskDefinitionEvent, RayActorTaskDefinitionEvent and RayTaskLifecycleEvent) were made in the hot task execution path. This commit moves it to the export part. Since export runs on on a different thread, it doesn't hamper task execution.

For RayTaskDefinitionEvent and RayActorTaskDefinitionEvent, task event recorder's MergeData is a no-op and therefore, SerializeData can populate the event proto trivially (just call the populate function from SerializeData).
For 'RayTaskLifecycleEvent, we collate all the state updates in 'MergeData and then in SerializeData, apply each state update one by one.

Note that this thing has one caveat that now the proto formation happens in ExportEvents under the task event recorder mutex, while earlier they were out of the mutex (it used to happen before calling AddEvent on the proto). So this increases mutex contention.

To try to fix this, I tried to export less than the entire buffer at a time (say 10k out of the 100k buffer) so that there is less work under the mutex. But firstly, this led to poorer perf as compared to flushing the entire buffer and secondly to fully get it to work around all cases, some other changes were required in the recorder as well.

7a2b482 [opt3]: Earlier, even when task_event_buffer to GCS or aggregator flow was turned off, the current code did form the TaskStatusEvent and put it in the ring buffer under mutex. The fix made is that if ``task_event_buffer` is not publishing anywhere, short circuit that path very early.

Results:
Given below are the regression % after before any optimizations and after each optimization.

Benchmark	recorder on (before opt)	+opt1	+opt1,2	+opt1,2,3 (this PR)	+opt1,2,3,4 ¹
`1_1_actor_calls_async`	-39.3%	-25.4%	-17.1%	-13.6%	-9.5%
`1_n_actor_calls_async`	-33.7%	-20.3%	-10.5%	-4.5%	+2.0%
`1_n_async_actor_calls_async`	-30.8%	-18.6%	-14.3%	-2.7%	+1.8%
`1_1_actor_calls_concurrent`	-29.1%	-14.8%	-13.9%	-12.7%	-10.3%
`n_n_actor_calls_async`	-28.8%	-18.1%	-13.2%	-11.2%	-8.1%
`n_n_async_actor_calls_async`	-27.6%	-14.6%	-8.6%	-9.1%	-7.0%
`single_client_tasks_and_get_batch`	-26.8%	-13.3%	-13.9%	-11.8%	-4.6%
`single_client_tasks_async`	-26.7%	-14.1%	-12.6%	-9.2%	-6.4%
`1_1_async_actor_calls_async`	-25.1%	-13.0%	-16.6%	-10.8%	-9.1%
`single_client_wait_1k_refs`	-19.9%	-12.0%	-9.7%	-9.2%	-7.8%
`multi_client_tasks_async`	-17.9%	-11.0%	-7.8%	-5.7%	-4.0%
`1_1_async_actor_calls_with_args_async`	-15.4%	-9.5%	-9.2%	-6.3%	-6.0%
`1_1_actor_calls_sync`	-14.8%	-6.9%	-10.0%	-4.5%	-4.2%
`1_1_async_actor_calls_sync`	-14.5%	-6.8%	-9.1%	-3.9%	-5.0%
`single_client_tasks_sync`	-6.3%	-3.2%	-3.4%	-3.2%	-1.7%
Mean	-23.8%	-13.4%	-11.3%	-7.9%	-5.3%
Worst	-39.3%	-25.4%	-17.1%	-13.6%	-10.3%

There are two other optimizations that I tried which are not a part of this PR:

[opt4]: cut per event overhead on the record path using some micro-optimizations like:

instead of keying by (task-id-binary-string, attempt) ,key using (TaskID, attempt) to prevent string allocations.
share the session name as a shared_ptr instead of copying it into each event.
hands events to the recorder one at a time instead of allocating a vector per record call.

But it changes several widely-used signatures (TaskAttemptId, GetSessionName, the record entry point). Since it gives ~2.5% gain and we are already below 8% I didn't include it in this PR.

[opt5]: Right now whenever we AddEvents, if the ring is full, we would drop events. While dropping events, we record it using MetricInterface which does some allocations etc. holding a mutex. This could increase contention over the mutex. I tried to batch the drops and then export them to the metric only during ExportEvents. This gave some perf gains (when I was trying flushing only 10k events per batch), but then it hampers metric fidelity since they are now stale between flushes.

gemini-code-assist

Code Review

This pull request refactors the task event recording and serialization logic. It extracts task event population helpers and the TaskStateUpdate struct into a separate observability library, and defers protobuf serialization for definition and lifecycle events to the export path, keeping it off the task's critical path. Additionally, it optimizes the task event buffer to only record events when at least one destination is active. The review feedback suggests several improvements: replacing std::make_move_iterator on a const reference with standard iterators, adding defensive null checks for task_spec in event constructors, replacing non-standard std::optional<const T> types with const std::optional<T>&, and simplifying implicit std::optional construction in unit tests.

gemini-code-assist · 2026-08-07T16:10:02Z

+  definition_event_data.mutable_required_resources()->insert(
+      std::make_move_iterator(required_resources.begin()),
+      std::make_move_iterator(required_resources.end()));


Since required_resources is a const reference, std::make_move_iterator cannot actually move the elements and will fall back to copying them. Using std::make_move_iterator on const iterators is misleading and can cause compiler warnings. We should use standard iterators instead.

Suggested change

definition_event_data.mutable_required_resources()->insert(

std::make_move_iterator(required_resources.begin()),

std::make_move_iterator(required_resources.end()));

definition_event_data.mutable_required_resources()->insert(

required_resources.begin(),

required_resources.end());

Seems to be a valid comment

gemini-code-assist · 2026-08-07T16:10:02Z

+      task_spec_(std::move(task_spec)),
+      task_id_(task_id),
+      job_id_(job_id),
+      task_attempt_(task_attempt) {}


For defensive programming, we should verify that task_spec is not null before storing and dereferencing it, as dereferencing a null shared_ptr leads to undefined behavior.

task_attempt_(task_attempt) { RAY_CHECK(task_spec_ != nullptr); }

gemini-code-assist · 2026-08-07T16:10:02Z

+      task_spec_(std::move(task_spec)),
+      task_id_(task_id),
+      job_id_(job_id),
+      task_attempt_(task_attempt) {}


For defensive programming, we should verify that task_spec is not null before storing and dereferencing it, as dereferencing a null shared_ptr leads to undefined behavior.

task_attempt_(task_attempt) { RAY_CHECK(task_spec_ != nullptr); }

gemini-code-assist · 2026-08-07T16:10:03Z

+                        const JobID &job_id,
+                        int32_t task_attempt,
+                        rpc::TaskStatus task_status,
+                        const std::optional<const TaskStateUpdate> &state_update,


Using std::optional<const TaskStateUpdate> is non-standard and can lead to compilation issues on some compilers or standard library implementations, as std::optional is typically expected to hold non-const types. It also introduces inconsistency with task_event_populators.h which uses const std::optional<TaskStateUpdate>&. We should change this to const std::optional<TaskStateUpdate>&.

const std::optional<TaskStateUpdate> &state_update,

gemini-code-assist · 2026-08-07T16:10:03Z

+    const JobID &job_id,
+    int32_t task_attempt,
+    rpc::TaskStatus task_status,
+    const std::optional<const TaskStateUpdate> &state_update,


Using std::optional<const TaskStateUpdate> is non-standard and can lead to compilation issues on some compilers or standard library implementations, as std::optional is typically expected to hold non-const types. It also introduces inconsistency with task_event_populators.h which uses const std::optional<TaskStateUpdate>&. We should change this to const std::optional<TaskStateUpdate>&.

Suggested change

const std::optional<const TaskStateUpdate> &state_update,

const std::optional<TaskStateUpdate> &state_update,

gemini-code-assist · 2026-08-07T16:10:03Z

+struct TaskStateUpdate {
+  TaskStateUpdate() = default;
+
+  explicit TaskStateUpdate(const std::optional<const rpc::RayErrorInfo> &error_info)


Using std::optional<const rpc::RayErrorInfo> is non-standard and can lead to compilation issues on some compilers or standard library implementations, as std::optional is typically expected to hold non-const types. We should change this to const std::optional<rpc::RayErrorInfo>&.

Suggested change

explicit TaskStateUpdate(const std::optional<const rpc::RayErrorInfo> &error_info)

explicit TaskStateUpdate(const std::optional<rpc::RayErrorInfo> &error_info)

gemini-code-assist · 2026-08-07T16:10:03Z

+      job_id,
+      /*task_attempt=*/0,
+      rpc::TaskStatus::SUBMITTED_TO_WORKER,
+      std::optional<const TaskStateUpdate>(TaskStateUpdate(node_id, worker_id)),


Since std::optional can be implicitly constructed from its underlying type, we can simplify this by passing TaskStateUpdate directly without the explicit std::optional wrapper.

Suggested change

std::optional<const TaskStateUpdate>(TaskStateUpdate(node_id, worker_id)),

TaskStateUpdate(node_id, worker_id),

gemini-code-assist · 2026-08-07T16:10:03Z

+      job_id,
+      /*task_attempt=*/0,
+      rpc::TaskStatus::RUNNING,
+      std::optional<const TaskStateUpdate>(TaskStateUpdate(static_cast<uint32_t>(4321))),


Since std::optional can be implicitly constructed from its underlying type, we can simplify this by passing TaskStateUpdate directly without the explicit std::optional wrapper.

Suggested change

std::optional<const TaskStateUpdate>(TaskStateUpdate(static_cast<uint32_t>(4321))),

TaskStateUpdate(static_cast<uint32_t>(4321)),

gemini-code-assist · 2026-08-07T16:10:03Z

+      job_id,
+      /*task_attempt=*/0,
+      rpc::TaskStatus::NIL,
+      std::optional<const TaskStateUpdate>(TaskStateUpdate(start_log)),


Since std::optional can be implicitly constructed from its underlying type, we can simplify this by passing TaskStateUpdate directly without the explicit std::optional wrapper.

Suggested change

std::optional<const TaskStateUpdate>(TaskStateUpdate(start_log)),

TaskStateUpdate(start_log),

gemini-code-assist · 2026-08-07T16:10:03Z

+      job_id,
+      /*task_attempt=*/0,
+      rpc::TaskStatus::NIL,
+      std::optional<const TaskStateUpdate>(TaskStateUpdate(end_log)),


Since std::optional can be implicitly constructed from its underlying type, we can simplify this by passing TaskStateUpdate directly without the explicit std::optional wrapper.

Suggested change

std::optional<const TaskStateUpdate>(TaskStateUpdate(end_log)),

TaskStateUpdate(end_log),

cursor · 2026-08-07T16:10:30Z

  }

  if (data->events_size() == 0 && metadata->dropped_task_attempts_size() == 0) {
+    MarkGrpcDone();


PR needs clearer description

Low Severity

⚠️ This PR needs a clearer title and/or description.

To help reviewers, please ensure your PR includes:

Title: A concise summary of the change

Description:

What problem does this solve?

How does this PR solve it?

Any relevant context for reviewers such as:

Why is the problem important to solve?

Why was this approach chosen over others?

See this list of PRs as examples for PRs that have gone above and beyond:

[Core] Introduce local port service discovery #59613

[Core] Improve Large-Scale Resource View Synchronization Through Sync Message Batching #57641

Remove node observability information from hot path of core components #56474

[core][rdt] Support out-of-order actors by extracting metadata when creating #59610

[core] fix open leak for plasma store memory (shm/fallback) by workers #52622

^{Triggered by project rule: Bugbot Rules}

^{Reviewed by Cursor Bugbot for commit 414d343. Configure here.}

MengjinYan

Done with opt1. Keeping looking at other opts

MengjinYan · 2026-08-07T22:11:54Z

                       const TaskAttemptId &attempt)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);

+  // Move the buffered events, and the dropped task attempts to report, out of the buffers


Nit: Use doxygen format for function comment.

MengjinYan · 2026-08-07T22:12:38Z

+  // so that the caller can group, serialize and send them without holding mutex_. Events
+  // belonging to a dropped attempt are discarded so that a task attempt is either sent in
+  // full or not at all.
+  void TakeEventsToSend(std::list<std::unique_ptr<RayEventInterface>> *events,


Minor: Prefer a return structure than output parameters.

MengjinYan · 2026-08-07T22:12:50Z

+      return;
+    }
+    // Skip if there's already an in-flight gRPC call to avoid overlapping requests.
+    if (grpc_in_progress_.exchange(true)) {


It seems that with the change, the behavior of when the grpc_in_progress_ is set is changed as well. Wondering is it the expected behavior? The same logic is in ray_event_recorder as well. So we might want to consolidate the logic for both places.

MengjinYan

Done with opt 2 and its fix. Moving on to opt 3

MengjinYan · 2026-08-07T22:50:09Z

+      : is_debugger_paused_(is_debugger_paused) {}
+
+  /// Node id if it's a SUBMITTED_TO_WORKER status change.
+  std::optional<NodeID> node_id_ = std::nullopt;


style: struct data members shouldn't have trailing underscore, per google c++ style guide.

MengjinYan · 2026-08-07T22:59:36Z

+  definition_event_data.mutable_required_resources()->insert(
+      std::make_move_iterator(required_resources.begin()),
+      std::make_move_iterator(required_resources.end()));


Seems to be a valid comment

MengjinYan · 2026-08-07T23:40:33Z


 std::string RayActorTaskDefinitionEvent::GetEntityId() const {
-  return data_.task_id() + std::to_string(data_.task_attempt());
+  return task_id_.Binary() + std::to_string(task_attempt_);


Prefer absl::StrCat() for string concatenation

MengjinYan · 2026-08-07T23:44:22Z

should we add tests for actor definition event as well?

MengjinYan

Opt3 looks good. Since the new task event recorder will replace the task event buffer in the near future, it's okay for now to add an additional boolean field in the class.

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

cursor

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

There are 3 total unresolved issues (including 2 from previous reviews).

^{Reviewed by Cursor Bugbot for commit 9edc5ae. Configure here.}

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

MengjinYan

All resolved comments looks good. One minor followup comment.

MengjinYan · 2026-08-09T00:06:43Z


 std::string RayTaskLifecycleEvent::GetEntityId() const {
-  return data_.task_id() + std::to_string(data_.task_attempt());
+  return task_id_.Binary() + std::to_string(task_attempt_);


minor: Prefer absl::StrCat here.

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

karticam requested review from MengjinYan and sampan-s-nayak August 7, 2026 16:07

karticam requested a review from a team as a code owner August 7, 2026 16:07

karticam added core Issues that should be addressed in Ray Core go add ONLY when ready to merge, run all tests labels Aug 7, 2026

gemini-code-assist Bot reviewed Aug 7, 2026

View reviewed changes

cursor Bot reviewed Aug 7, 2026

View reviewed changes

karticam force-pushed the karticam/optimize-task-event-recorder branch from 414d343 to 7a2b482 Compare August 7, 2026 17:56

MengjinYan reviewed Aug 7, 2026

View reviewed changes

karticam added 4 commits August 7, 2026 18:42

opt1: Get heavy operations out of mutex

6f42e9f

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

opt2: Add late proto formation - remove it from the hot task exec path

f78f2bc

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

fix: task_log_info shoudl be merged rather than overwritten

4ef87d0

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

opt3: early short circuit when not using task event buffer

9edc5ae

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

karticam force-pushed the karticam/optimize-task-event-recorder branch from 7a2b482 to 9edc5ae Compare August 8, 2026 01:42

cursor Bot reviewed Aug 8, 2026

View reviewed changes

Comment thread src/ray/observability/ray_task_event_recorder.h Outdated

karticam added 3 commits August 8, 2026 10:46

Resolvign cmt (p1)

08119f7

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

Resolving cmt (p2)

88e348f

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

fix bug due to rename members

af9dde3

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

MengjinYan reviewed Aug 9, 2026

View reviewed changes

Resolving cmt

cab8b41

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

	const std::optional<const TaskStateUpdate> &state_update,
	const std::optional<TaskStateUpdate> &state_update,

	explicit TaskStateUpdate(const std::optional<const rpc::RayErrorInfo> &error_info)
	explicit TaskStateUpdate(const std::optional<rpc::RayErrorInfo> &error_info)

	std::optional<const TaskStateUpdate>(TaskStateUpdate(node_id, worker_id)),
	TaskStateUpdate(node_id, worker_id),

	std::optional<const TaskStateUpdate>(TaskStateUpdate(static_cast<uint32_t>(4321))),
	TaskStateUpdate(static_cast<uint32_t>(4321)),

	std::optional<const TaskStateUpdate>(TaskStateUpdate(start_log)),
	TaskStateUpdate(start_log),

	std::optional<const TaskStateUpdate>(TaskStateUpdate(end_log)),
	TaskStateUpdate(end_log),

Uh oh!

Conversation

karticam commented Aug 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

cursor Bot Aug 7, 2026

Choose a reason for hiding this comment

PR needs clearer description

Uh oh!

Uh oh!

MengjinYan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

MengjinYan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

karticam commented Aug 7, 2026 •

edited

Loading