Make core worker exit with system error when receiving IOError from w…

…riting to local object store

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

src/ray/raylet_client/raylet_client.cc

@@ -85,9 +85,10 @@ Status raylet::RayletConnection::AtomicRequestReply(MessageType request_type,
 }
 
 void raylet::RayletConnection::ShutdownIfLocalRayletDisconnected(const Status &status) {
-  if (!status.ok() && IsRayletFailed(RayConfig::instance().RAYLET_PID())) {
+  if ((!status.ok() && IsRayletFailed(RayConfig::instance().RAYLET_PID())) ||
+      status.IsIOError()) {
     RAY_LOG(WARNING) << "The connection is failed because the local raylet has been "
-                        "dead. Terminate the process. Status: "
+                        "dead or is unreachable. Terminate the process. Status: "
                      << status;
     QuickExit();
     RAY_LOG(FATAL) << "Unreachable.";

src/ray/raylet_client/raylet_client.h

@@ -269,7 +269,8 @@ class RayletConnection {
                                  flatbuffers::FlatBufferBuilder *fbb = nullptr);
 
  private:
-  /// Shutdown the raylet if the local connection is disconnected.
+  /// Shutdown the raylet if the local connection is disconnected (either terminated or
+  // unreachable).
   void ShutdownIfLocalRayletDisconnected(const Status &status);
   /// The connection to raylet.
   std::shared_ptr<ServerConnection> conn_;