[Data] [Core] [2.5/n] Remove BlockRefCounter.clear() and expand callback throughput benchmark

[rayhhome]

Summary

• Remove BlockRefCounter.clear() and its shutdown wiring
• Expand callback throughput release benchmark

Description

Removes BlockRefCounter.clear() and the _block_ref_counter.clear() call in StreamingExecutor.shutdown(). The clear() method was added for deterministic cleanup on executor shutdown, but profiling and benchmarking showed:

1. Blocks are freed naturally via out-of-scope callbacks when their ObjectRefs go out of scope. Explicit clearing is unnecessary.
2. Removing clear() had no measurable impact on release test benchmarks.

Also expands test_callback_throughput.py (previously only had an incremental pipeline test) with additional benchmark tests at multiple scales (100 to 10,000 blocks):

• Registration cost: on_block_produced overhead via the Data-layer path
• Incremental pipeline: produce-consume-release, one block at a time
• Burst drop: all refs dropped at once, measures total drain time
• Per-callback latency: refs dropped one at a time with per-block timestamps
• BlockRefCounter burst drop: same as burst drop but through the full BRC path

Related issues

Depends on #64157 (BlockRefCounter implementation).
Related to #63601 (prototype), #63074 (previous manual BlockRefCounter).

[gemini-code-assist]

Warning

Gemini encountered an error creating the review. You can try again by commenting /gemini review.

[Copilot]

Pull request overview

Adds a symmetric “remove” API for Ray Core’s object out-of-scope/freed callback mechanism so callers (notably Python) can explicitly deregister callbacks and trigger their cleanup logic deterministically, rather than relying on callbacks becoming lazy no-ops.

Changes:

• Introduces ReferenceCounterInterface::RemoveObjectOutOfScopeOrFreedCallbacks and implements it in ReferenceCounter.
• Plumbs the API through CoreWorker and exposes it to Python via libcoreworker.pxd and _raylet.pyx.
• Adds a C++ unit test covering “remove fires once; natural out-of-scope does not fire again”.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
src/ray/core_worker/tests/reference_counter_test.cc	Adds a unit test validating removal fires callbacks exactly once.
src/ray/core_worker/reference_counter.h	Declares the new removal API on the concrete reference counter.
src/ray/core_worker/reference_counter.cc	Implements callback firing + clearing for the new removal API.
src/ray/core_worker/reference_counter_interface.h	Adds the new API to the core reference counter interface and documents behavior.
src/ray/core_worker/core_worker.h	Declares a CoreWorker wrapper for the new removal API.
src/ray/core_worker/core_worker.cc	Implements the CoreWorker wrapper by delegating to the reference counter.
python/ray/includes/libcoreworker.pxd	Exposes the CoreWorker removal method to Cython.
python/ray/_raylet.pyx	Adds a Python-facing internal method to trigger callback removal/cleanup.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Kunchd]

I don't have any big concerns with this PR, but I'm struggling to determine the use case for this. We currently clean up the callback when the object goes out of scope and the callback is invoked. Why is explicit clean up necessary? Could we include the motivation in the PR description?

[Kunchd]

Why is invoking all the callbacks neecessary before removing the callbacks? Doesn't removing the callback mean we don't care about them anymore?

[rayhhome]

BlockRefCounter's clear() is called when StreamingExecutor calls shutdown(), which happens when a dataset execution ends. In multi-epoch training (or any jobs with many dataset executions), this would be triggered many times. Since the callbacks have Py_DECREF to preserve the _on_object_freed closures, not triggering them would result in minor leaks. What's your take on this?

[Kunchd]

Do we expect object produced within a specific epoch to stay around after the epoch finishes?

[rayhhome]

This PR is mainly for aligning the behavior of clear() with what the function name suggests. In the current implementation, only the Data side dictionary is cleared when clear() is called. The Core callbacks will still be fired when ObjectRefs go out of scope, which seems unintuitive based on what the function is intended to do.

However, I think this does raise the question on whether we need the clear() function at all, considering that no code reads the accounting after the shutdown. Will check this offline and come back with the decision.

[rayhhome]

BlockRefCounter's clear() is called when StreamingExecutor calls shutdown(), which happens when a dataset execution ends. In multi-epoch training (or any jobs with many dataset executions), this would be triggered many times. Since the callbacks have Py_DECREF to preserve the _on_object_freed closures, not triggering them would result in minor leaks. What's your take on this?

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.

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