Revert "Unhandled exception handler based on local ref counting (#14049)" (#14099)

This reverts commit 9dc671ae02.

src/ray/common/ray_object.h

@@ -92,20 +92,12 @@ class RayObject {
   /// large to return directly as part of a gRPC response).
   bool IsInPlasmaError() const;
 
-  /// Mark this object as accessed before.
-  void SetAccessed() { accessed_ = true; };
-
-  /// Check if this object was accessed before.
-  bool WasAccessed() const { return accessed_; }
-
  private:
   std::shared_ptr<Buffer> data_;
   std::shared_ptr<Buffer> metadata_;
   const std::vector<ObjectID> nested_ids_;
   /// Whether this class holds a data copy.
   bool has_data_copy_;
-  /// Whether this object was accessed.
-  bool accessed_ = false;
 };
 
 }  // namespace ray

src/ray/core_worker/core_worker.cc

@@ -422,7 +422,7 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_
         return Status::OK();
       },
       options_.ref_counting_enabled ? reference_counter_ : nullptr, local_raylet_client_,
-      options_.check_signals, options_.unhandled_exception_handler));
+      options_.check_signals));
 
   auto check_node_alive_fn = [this](const NodeID &node_id) {
     auto node = gcs_client_->Nodes().Get(node_id);

src/ray/core_worker/core_worker.h

@@ -82,7 +82,6 @@ struct CoreWorkerOptions {
         spill_objects(nullptr),
         restore_spilled_objects(nullptr),
         delete_spilled_objects(nullptr),
-        unhandled_exception_handler(nullptr),
         get_lang_stack(nullptr),
         kill_main(nullptr),
         ref_counting_enabled(false),
@@ -147,8 +146,6 @@ struct CoreWorkerOptions {
   /// Application-language callback to delete objects from external storage.
   std::function<void(const std::vector<std::string> &, rpc::WorkerType)>
       delete_spilled_objects;
-  /// Function to call on error objects never retrieved.
-  std::function<void(const RayObject &error)> unhandled_exception_handler;
   /// Language worker callback to get the current call stack.
   std::function<void(std::string *)> get_lang_stack;
   // Function that tries to interrupt the currently running Python thread.

src/ray/core_worker/store_provider/memory_store/memory_store.cc

@@ -93,7 +93,6 @@ void GetRequest::Set(const ObjectID &object_id, std::shared_ptr<RayObject> objec
   if (is_ready_) {
     return;  // We have already hit the number of objects to return limit.
   }
-  object->SetAccessed();
   objects_.emplace(object_id, object);
   if (objects_.size() == num_objects_ ||
       (abort_if_any_object_is_exception_ && object->IsException() &&
@@ -107,7 +106,6 @@ std::shared_ptr<RayObject> GetRequest::Get(const ObjectID &object_id) const {
   std::unique_lock<std::mutex> lock(mutex_);
   auto iter = objects_.find(object_id);
   if (iter != objects_.end()) {
-    iter->second->SetAccessed();
     return iter->second;
   }
 
@@ -118,13 +116,11 @@ CoreWorkerMemoryStore::CoreWorkerMemoryStore(
     std::function<void(const RayObject &, const ObjectID &)> store_in_plasma,
     std::shared_ptr<ReferenceCounter> counter,
     std::shared_ptr<raylet::RayletClient> raylet_client,
-    std::function<Status()> check_signals,
-    std::function<void(const RayObject &)> unhandled_exception_handler)
+    std::function<Status()> check_signals)
     : store_in_plasma_(store_in_plasma),
       ref_counter_(counter),
       raylet_client_(raylet_client),
-      check_signals_(check_signals),
-      unhandled_exception_handler_(unhandled_exception_handler) {}
+      check_signals_(check_signals) {}
 
 void CoreWorkerMemoryStore::GetAsync(
     const ObjectID &object_id, std::function<void(std::shared_ptr<RayObject>)> callback) {
@@ -140,7 +136,6 @@ void CoreWorkerMemoryStore::GetAsync(
   }
   // It's important for performance to run the callback outside the lock.
   if (ptr != nullptr) {
-    ptr->SetAccessed();
     callback(ptr);
   }
 }
@@ -151,7 +146,6 @@ std::shared_ptr<RayObject> CoreWorkerMemoryStore::GetOrPromoteToPlasma(
   auto iter = objects_.find(object_id);
   if (iter != objects_.end()) {
     auto obj = iter->second;
-    obj->SetAccessed();
     if (obj->IsInPlasmaError()) {
       return nullptr;
     }
@@ -216,8 +210,6 @@ bool CoreWorkerMemoryStore::Put(const RayObject &object, const ObjectID &object_
     if (should_add_entry) {
       // If there is no existing get request, then add the `RayObject` to map.
       objects_.emplace(object_id, object_entry);
-    } else {
-      OnErase(object_entry);
     }
   }
 
@@ -231,7 +223,6 @@ bool CoreWorkerMemoryStore::Put(const RayObject &object, const ObjectID &object_
 
   // It's important for performance to run the callbacks outside the lock.
   for (const auto &cb : async_callbacks) {
-    object_entry->SetAccessed();
     cb(object_entry);
   }
 
@@ -266,7 +257,6 @@ Status CoreWorkerMemoryStore::GetImpl(const std::vector<ObjectID> &object_ids,
       const auto &object_id = object_ids[i];
       auto iter = objects_.find(object_id);
       if (iter != objects_.end()) {
-        iter->second->SetAccessed();
         (*results)[i] = iter->second;
         if (remove_after_get) {
           // Note that we cannot remove the object_id from `objects_` now,
@@ -436,7 +426,6 @@ void CoreWorkerMemoryStore::Delete(const absl::flat_hash_set<ObjectID> &object_i
       if (it->second->IsInPlasmaError()) {
         plasma_ids_to_delete->insert(object_id);
       } else {
-        OnErase(it->second);
         objects_.erase(it);
       }
     }
@@ -446,11 +435,7 @@ void CoreWorkerMemoryStore::Delete(const absl::flat_hash_set<ObjectID> &object_i
 void CoreWorkerMemoryStore::Delete(const std::vector<ObjectID> &object_ids) {
   absl::MutexLock lock(&mu_);
   for (const auto &object_id : object_ids) {
-    auto it = objects_.find(object_id);
-    if (it != objects_.end()) {
-      OnErase(it->second);
-      objects_.erase(it);
-    }
+    objects_.erase(object_id);
   }
 }
 
@@ -466,14 +451,6 @@ bool CoreWorkerMemoryStore::Contains(const ObjectID &object_id, bool *in_plasma)
   return false;
 }
 
-void CoreWorkerMemoryStore::OnErase(std::shared_ptr<RayObject> obj) {
-  // TODO(ekl) note that this doesn't warn on errors that are stored in plasma.
-  if (obj->IsException() && !obj->IsInPlasmaError() && !obj->WasAccessed() &&
-      unhandled_exception_handler_ != nullptr) {
-    unhandled_exception_handler_(*obj);
-  }
-}
-
 MemoryStoreStats CoreWorkerMemoryStore::GetMemoryStoreStatisticalData() {
   absl::MutexLock lock(&mu_);
   MemoryStoreStats item;

src/ray/core_worker/store_provider/memory_store/memory_store.h

@@ -35,8 +35,7 @@ class CoreWorkerMemoryStore {
       std::function<void(const RayObject &, const ObjectID &)> store_in_plasma = nullptr,
       std::shared_ptr<ReferenceCounter> counter = nullptr,
       std::shared_ptr<raylet::RayletClient> raylet_client = nullptr,
-      std::function<Status()> check_signals = nullptr,
-      std::function<void(const RayObject &)> unhandled_exception_handler = nullptr);
+      std::function<Status()> check_signals = nullptr);
   ~CoreWorkerMemoryStore(){};
 
   /// Put an object with specified ID into object store.
@@ -144,9 +143,6 @@ class CoreWorkerMemoryStore {
                  std::vector<std::shared_ptr<RayObject>> *results,
                  bool abort_if_any_object_is_exception);
 
-  /// Called when an object is erased from the store.
-  void OnErase(std::shared_ptr<RayObject> obj);
-
   /// Optional callback for putting objects into the plasma store.
   std::function<void(const RayObject &, const ObjectID &)> store_in_plasma_;
 
@@ -177,9 +173,6 @@ class CoreWorkerMemoryStore {
 
   /// Function passed in to be called to check for signals (e.g., Ctrl-C).
   std::function<Status()> check_signals_;
-
-  /// Function called to report unhandled exceptions.
-  std::function<void(const RayObject &)> unhandled_exception_handler_;
 };
 
 }  // namespace ray

src/ray/core_worker/test/memory_store_test.cc

@@ -1,66 +0,0 @@
-// Copyright 2017 The Ray Authors.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//  http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-#include "ray/core_worker/store_provider/memory_store/memory_store.h"
-
-#include "gtest/gtest.h"
-#include "ray/common/test_util.h"
-
-namespace ray {
-
-TEST(TestMemoryStore, TestReportUnhandledErrors) {
-  std::vector<std::shared_ptr<RayObject>> results;
-  WorkerContext context(WorkerType::WORKER, WorkerID::FromRandom(), JobID::FromInt(0));
-  int unhandled_count = 0;
-
-  std::shared_ptr<CoreWorkerMemoryStore> provider =
-      std::make_shared<CoreWorkerMemoryStore>(
-          nullptr, nullptr, nullptr, nullptr,
-          [&](const RayObject &obj) { unhandled_count++; });
-  RayObject obj1(rpc::ErrorType::TASK_EXECUTION_EXCEPTION);
-  RayObject obj2(rpc::ErrorType::TASK_EXECUTION_EXCEPTION);
-  auto id1 = ObjectID::FromRandom();
-  auto id2 = ObjectID::FromRandom();
-
-  // Check delete without get.
-  RAY_CHECK(provider->Put(obj1, id1));
-  RAY_CHECK(provider->Put(obj2, id2));
-  ASSERT_EQ(unhandled_count, 0);
-  provider->Delete({id1, id2});
-  ASSERT_EQ(unhandled_count, 2);
-  unhandled_count = 0;
-
-  // Check delete after get.
-  RAY_CHECK(provider->Put(obj1, id1));
-  RAY_CHECK(provider->Put(obj1, id2));
-  provider->Get({id1}, 1, 100, context, false, &results);
-  provider->GetOrPromoteToPlasma(id2);
-  provider->Delete({id1, id2});
-  ASSERT_EQ(unhandled_count, 0);
-
-  // Check delete after async get.
-  provider->GetAsync({id2}, [](std::shared_ptr<RayObject> obj) {});
-  RAY_CHECK(provider->Put(obj1, id1));
-  RAY_CHECK(provider->Put(obj2, id2));
-  provider->GetAsync({id1}, [](std::shared_ptr<RayObject> obj) {});
-  provider->Delete({id1, id2});
-  ASSERT_EQ(unhandled_count, 0);
-}
-
-}  // namespace ray
-
-int main(int argc, char **argv) {
-  ::testing::InitGoogleTest(&argc, argv);
-  return RUN_ALL_TESTS();
-}