Repository navigation
[core][__on_object_freed__][4/n] Run the hook when an object's last copy on the node is deleted, or when an actor task on its node moves it - #67443
Conversation
There was a problem hiding this comment.
Code Review
This pull request introduces changes to ensure that the __on_object_freed__ hook is reliably executed when the last copy of an object (either in-memory or spilled to disk) is deleted. It updates LocalObjectManager, ObjectManager, and NodeManager to track disk copies and coordinate firing the hook, and adds comprehensive Python and C++ tests. The review feedback highlights three important issues: a correctness edge case where the hook is leaked if an object is freed during a failing spill, a potential dangling reference in local_object_manager.cc from holding a reference to a map element during callback execution, and a data race in the test mock FakeCoreWorkerClient due to non-thread-safe access to on_object_freed_requests.
| ASSERT_EQ(spilled_copies_deleted, expected); | ||
| } | ||
|
|
||
| TEST_F(LocalObjectManagerTest, TestFailedSpillDeletesNoSpilledCopy) { |
There was a problem hiding this comment.
There is a correctness bug/edge case here where the __on_object_freed__ hook will be leaked (never run) if an object is freed while it is being spilled, and then the spill subsequently fails.
Step-by-step flow:
- An object is currently being spilled (it is in
objects_pending_spill_, soHasDiskCopyreturnstrue). - The object is freed by the application, calling
ReleaseFreedLocalObjectwhich deletes the object from plasma. - This triggers
HandleObjectMissinginNodeManager. HandleObjectMissingchecksHasDiskCopy(object_id). Since it istrue(pending spill), it skips running the hook, expecting the spilled copy's deletion to run it later.- The spill subsequently fails.
- Since the spill failed, no spilled copy is created on disk, and
on_spilled_copy_deleted_is never called. - The hook is never run, resulting in a leak of the hook.
Recommendation:
On spill failure, if the object has already been freed (i.e., it is no longer in plasma and was released), we should trigger the hook (e.g., by invoking on_spilled_copy_deleted_ or handling it in the spill failure callback).
| const WorkerID &producer_worker_id = | ||
| local_objects_.at(object_id).producer_worker_id_; |
There was a problem hiding this comment.
To prevent potential dangling reference issues, copy producer_worker_id by value instead of taking a reference. The callback on_spilled_copy_deleted_ executes arbitrary logic (including RPCs and manager state updates) which could potentially modify local_objects_ and invalidate the reference.
| const WorkerID &producer_worker_id = | |
| local_objects_.at(object_id).producer_worker_id_; | |
| const WorkerID producer_worker_id = | |
| local_objects_.at(object_id).producer_worker_id_; |
| void OnObjectFreed(const OnObjectFreedRequest &request, | ||
| const ClientCallback<OnObjectFreedReply> &callback) override {} | ||
| const ClientCallback<OnObjectFreedReply> &callback) override { | ||
| on_object_freed_requests.push_back(request); | ||
| } |
There was a problem hiding this comment.
The on_object_freed_requests vector is not thread-safe. Since OnObjectFreed can be called from multiple threads (e.g., RPC threads or main thread in tests), writing to on_object_freed_requests without a lock is a data race. Consider guarding it with mutex_ as done with other members like callbacks_.
| void OnObjectFreed(const OnObjectFreedRequest &request, | |
| const ClientCallback<OnObjectFreedReply> &callback) override {} | |
| const ClientCallback<OnObjectFreedReply> &callback) override { | |
| on_object_freed_requests.push_back(request); | |
| } | |
| void OnObjectFreed(const OnObjectFreedRequest &request, | |
| const ClientCallback<OnObjectFreedReply> &callback) override { | |
| absl::MutexLock lock(&mutex_); | |
| on_object_freed_requests.push_back(request); | |
| } |
009ba71 to
a14175d
Compare
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
There are 2 total unresolved issues (including 1 from previous review).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit a14175d. Configure here.
| /// spilling, eviction of the in-memory copy, and a free while spilled. | ||
| WorkerID GetProducerWorkerId(const ObjectID &object_id) const override; | ||
|
|
||
| bool HasDiskCopy(const ObjectID &object_id) const override; |
There was a problem hiding this comment.
Missing Doxygen on new APIs
Low Severity
/** ... */ block comments using @ tags (@brief, and @param/@return where applicable). New declarations HasDiskCopy and SetLocalObjectProducerWorkerId omit that documentation.
Additional Locations (2)
Triggered by project rule: Bugbot Rules
Reviewed by Cursor Bugbot for commit a14175d. Configure here.
4a921ec to
e024ab3
Compare
e024ab3 to
7d29351
Compare
…de is deleted Signed-off-by: kartica.modi <kartica.modi@anyscale.com>
… node moves the object Signed-off-by: kartica.modi <kartica.modi@anyscale.com>
The 9 tests start their own clusters and take about 70 seconds in total, past the 60 second limit of the small test group. Signed-off-by: kartica.modi <kartica.modi@anyscale.com>
7d29351 to
5fb1055
Compare


This is PR 4 of a series of PRs that add the
__on_object_freed__hook: an actor that defines the hook will be told when an object it created is freed. This PR runs the hook. The raylet on the producer's node calls it once: when an actor task on that same node consumes the object, or otherwise when the object's last copy on the node is deleted, using the producer recorded by #67377 and theOnObjectFreedRPC from #67308.The complete design doc is here: https://docs.google.com/document/d/1ZJhgRmwVyKfNCR21CWf6xMcc0B6HJwSPQsfebTRvnnE/edit?tab=t.0#heading=h.7wcod1f9afpg
Changes
NodeManager::HandleObjectMissingruns the hook unless the object is spilled or being spilled (LocalObjectManager::HasDiskCopy). The raylet reads the copy's producer beforeObjectManager::HandleObjectDeletederases it, and passes it in.LocalObjectManagercalls the newon_spilled_copy_deletedcallback with the producer recorded at pin, when it deletes a freed object's spilled copy.NodeManager::HandleSpilledCopyDeletedruns the hook unless an in-memory copy is still there.NodeManager::HandleSpilledCopyDeletedrecords the producer from the pin on that copy (ObjectManager::SetLocalObjectProducerWorkerId), and deleting the copy runs the hook.ray.get, andray.waitwith the defaultfetch_local=True, restore an evicted spilled object on the node.NodeManager::FireOnObjectFreedsendsOnObjectFreedto the producer worker. If that worker has exited, so the raylet no longer has it registered (WorkerPool::GetRegisteredWorkerreturns null), it does nothing. The request isn't retried.consume_onceobject moves to the actor task that consumes it, and that task can run on the producer's node or on another node.NodeManager::HandleMovedActorCallArgsruns it for each moved arg created on this node, before any pull or restore. It first reads the producer from the object manager's entry for the in-memory copy (ObjectManager::GetLocalObjectProducerWorkerId), which is set when the copy is sealed, so it has the producer even if the producer's pin arrives after the consumption. If that has none (which could happen when the object is spilled and evicted and thus no copy in ObjectManager), it reads the pinned object'sLocalObjectInfo(LocalObjectManager::GetProducerWorkerId), which still has the producer after the in-memory copy of a spilled object is evicted.objects_freed_at_consume_remembers the objects whose hook ran at a same-node consumption, and the last copy's deletion removes them without running the hook again.