[core] Fix RAY_CHECK failure during shutdown due to plasma store race condition

Signed-off-by: Sagar Sumit <[email protected]>

Author: codope

src/ray/core_worker/core_worker.cc

@@ -538,6 +538,10 @@ void CoreWorker::Shutdown() {
   RAY_LOG(INFO) << "Core worker ready to be deallocated.";
 }
 
+bool CoreWorker::IsShuttingDown() const {
+  return is_shutdown_.load();
+}
+
 void CoreWorker::ConnectToRayletInternal() {
   // Tell the raylet the port that we are listening on.
   // NOTE: This also marks the worker as available in Raylet. We do this at the

src/ray/core_worker/core_worker.h

@@ -246,6 +246,12 @@ class CoreWorker {
   ///
   void Shutdown();
 
+  /// Check if the core worker is currently shutting down.
+  /// This can be used to avoid operations that might fail during shutdown.
+  ///
+  /// \return true if shutdown has been initiated, false otherwise.
+  bool IsShuttingDown() const;
+
   /// Start receiving and executing tasks.
   void RunTaskExecutionLoop();
 

src/ray/core_worker/core_worker_process.cc

@@ -423,8 +423,35 @@ std::shared_ptr<CoreWorker> CoreWorkerProcessImpl::CreateCoreWorker(
       /*put_in_local_plasma_callback=*/
       [this](const RayObject &object, const ObjectID &object_id) {
         auto core_worker = GetCoreWorker();
-        RAY_CHECK_OK(
-            core_worker->PutInLocalPlasmaStore(object, object_id, /*pin_object=*/true));
+        
+        // Check if the core worker is shutting down before attempting plasma operations.
+        // During shutdown, the plasma store connection may already be broken, so we
+        // should avoid plasma operations entirely.
+        if (core_worker->IsShuttingDown()) {
+          RAY_LOG(INFO) << "Skipping plasma store operation for error object " << object_id 
+                       << " because core worker is shutting down.";
+          return;
+        }
+        
+        auto status = core_worker->PutInLocalPlasmaStore(object, object_id, /*pin_object=*/true);
+        if (!status.ok()) {
+          if (status.IsIOError() && 
+              (status.message().find("Broken pipe") != std::string::npos ||
+               status.message().find("Connection reset") != std::string::npos ||
+               status.message().find("Bad file descriptor") != std::string::npos)) {
+            // This is likely a shutdown race where the plasma store
+            // connection was closed before we could complete the operation.
+            // Log as warning since this is expected during shutdown scenarios.
+            RAY_LOG(WARNING) << "Failed to put error object " << object_id 
+                            << " in plasma store due to connection error (likely shutdown): " 
+                            << status.ToString();
+          } else {
+            // For other types of errors, maintain the original
+            // behavior with RAY_CHECK_OK to catch real issues.
+            RAY_CHECK_OK(status) << "Failed to put error object " << object_id 
+                                << " in plasma store: " << status.ToString();
+          }
+        }
       },
       /* retry_task_callback= */
       [this](TaskSpecification &spec, bool object_recovery, uint32_t delay_ms) {