Prevent plasma store and manager from dying when a client dies. (#203)

* Prevent plasma store and manager from dying when a worker dies.

* Check errno inside of warn_if_sigpipe. Passing in errno doesn't work because the arguments to warn_if_sigpipe can be evaluated out of order.

src/global_scheduler/global_scheduler.c

@@ -237,7 +237,7 @@ void local_scheduler_table_handler(db_client_id client_id,
   UNUSED(id_string);
   LOG_DEBUG(
       "total workers = %d, task queue length = %d, available workers = %d",
-      info.num_total_workers, info.task_queue_length, info.available_workers);
+      info.total_num_workers, info.task_queue_length, info.available_workers);
   /* Update the local scheduler info struct. */
   local_scheduler *local_scheduler_ptr = get_local_scheduler(state, client_id);
   if (local_scheduler_ptr != NULL) {

src/photon/photon_scheduler.c

@@ -114,7 +114,19 @@ void assign_task_to_worker(local_scheduler_state *state,
                            bool from_global_scheduler) {
   CHECK(worker_index < utarray_len(state->workers));
   worker *w = (worker *) utarray_eltptr(state->workers, worker_index);
-  write_message(w->sock, EXECUTE_TASK, task_spec_size(spec), (uint8_t *) spec);
+  if (write_message(w->sock, EXECUTE_TASK, task_spec_size(spec),
+                    (uint8_t *) spec) < 0) {
+    if (errno == EPIPE || errno == EBADF) {
+      /* TODO(rkn): If this happens, the task should be added back to the task
+       * queue. */
+      LOG_WARN(
+          "Failed to give task to worker on fd %d. The client may have hung "
+          "up.",
+          w->sock);
+    } else {
+      LOG_FATAL("Failed to give task to client on fd %d.", w->sock);
+    }
+  }
   /* Update the global task table. */
   if (state->db != NULL) {
     task *task =
@@ -350,6 +362,9 @@ void start_server(const char *node_ip_address,
                   const char *plasma_manager_socket_name,
                   const char *plasma_manager_address,
                   bool global_scheduler_exists) {
+  /* Ignore SIGPIPE signals. If we don't do this, then when we attempt to write
+   * to a client that has already died, the local scheduler could die. */
+  signal(SIGPIPE, SIG_IGN);
   int fd = bind_ipc_sock(socket_name, true);
   event_loop *loop = event_loop_create();
   g_state = init_local_scheduler(

src/plasma/plasma.c

@@ -17,3 +17,17 @@ bool plasma_object_ids_distinct(int num_object_ids, object_id object_ids[]) {
   }
   return true;
 }
+
+void warn_if_sigpipe(int status, int client_sock) {
+  if (status >= 0) {
+    return;
+  }
+  if (errno == EPIPE || errno == EBADF) {
+    LOG_WARN(
+        "Received SIGPIPE or BAD FILE DESCRIPTOR when sending a message to "
+        "client on fd %d. The client on the other end may have hung up.",
+        client_sock);
+    return;
+  }
+  LOG_FATAL("Failed to write message to client on fd %d.", client_sock);
+}

src/plasma/plasma.h

@@ -135,4 +135,21 @@ typedef struct {
  */
 bool plasma_object_ids_distinct(int num_object_ids, object_id object_ids[]);
 
+/**
+ * Print a warning if the status is less than zero. This should be used to check
+ * the success of messages sent to plasma clients. We print a warning instead of
+ * failing because the plasma clients are allowed to die. This is used to handle
+ * situations where the store writes to a client file descriptor, and the client
+ * may already have disconnected. If we have processed the disconnection and
+ * closed the file descriptor, we should get a BAD FILE DESCRIPTOR error. If we
+ * have not, then we should get a SIGPIPE.
+ *
+ * @param status The status to check. If it is less less than zero, we will
+ *        print a warning.
+ * @param client_sock The client socket. This is just used to print some extra
+ *        information.
+ * @return Void.
+ */
+void warn_if_sigpipe(int status, int client_sock);
+
 #endif /* PLASMA_H */

src/plasma/plasma_manager.c

@@ -357,9 +357,10 @@ void remove_wait_request(plasma_manager_state *manager_state,
 void return_from_wait(plasma_manager_state *manager_state,
                       wait_request *wait_req) {
   /* Send the reply to the client. */
-  CHECK(plasma_send_WaitReply(wait_req->client_conn->fd, manager_state->builder,
-                              wait_req->object_requests,
-                              wait_req->num_object_requests) >= 0);
+  warn_if_sigpipe(plasma_send_WaitReply(
+                      wait_req->client_conn->fd, manager_state->builder,
+                      wait_req->object_requests, wait_req->num_object_requests),
+                  wait_req->client_conn->fd);
   /* Remove the wait request from each of the relevant object_wait_requests hash
    * tables if it is present there. */
   for (int i = 0; i < wait_req->num_object_requests; ++i) {
@@ -580,16 +581,20 @@ void send_queued_request(event_loop *loop,
   plasma_request_buffer *buf = conn->transfer_queue;
   switch (buf->type) {
   case MessageType_PlasmaDataRequest:
-    CHECK(plasma_send_DataRequest(conn->fd, state->builder, buf->object_id,
-                                  state->addr, state->port) >= 0);
+    warn_if_sigpipe(
+        plasma_send_DataRequest(conn->fd, state->builder, buf->object_id,
+                                state->addr, state->port),
+        conn->fd);
     break;
   case MessageType_PlasmaDataReply:
     LOG_DEBUG("Transferring object to manager");
     if (conn->cursor == 0) {
       /* If the cursor is zero, we haven't sent any requests for this object
        * yet, so send the initial data request. */
-      CHECK(plasma_send_DataReply(conn->fd, state->builder, buf->object_id,
-                                  buf->data_size, buf->metadata_size) >= 0);
+      warn_if_sigpipe(
+          plasma_send_DataReply(conn->fd, state->builder, buf->object_id,
+                                buf->data_size, buf->metadata_size),
+          conn->fd);
     }
     write_object_chunk(conn, buf);
     break;
@@ -1169,9 +1174,10 @@ void request_status_done(object_id object_id,
   client_connection *client_conn = (client_connection *) context;
   int status =
       request_status(object_id, manager_count, manager_vector, context);
-  CHECK(plasma_send_StatusReply(client_conn->fd,
-                                client_conn->manager_state->builder, &object_id,
-                                &status, 1) >= 0);
+  warn_if_sigpipe(plasma_send_StatusReply(client_conn->fd,
+                                          client_conn->manager_state->builder,
+                                          &object_id, &status, 1),
+                  client_conn->fd);
 }
 
 int request_status(object_id object_id,
@@ -1204,17 +1210,19 @@ void process_status_request(client_connection *client_conn,
   /* Return success immediately if we already have this object. */
   if (is_object_local(client_conn->manager_state, object_id)) {
     int status = ObjectStatus_Local;
-    CHECK(plasma_send_StatusReply(client_conn->fd,
-                                  client_conn->manager_state->builder,
-                                  &object_id, &status, 1) >= 0);
+    warn_if_sigpipe(plasma_send_StatusReply(client_conn->fd,
+                                            client_conn->manager_state->builder,
+                                            &object_id, &status, 1),
+                    client_conn->fd);
     return;
   }
 
   if (client_conn->manager_state->db == NULL) {
     int status = ObjectStatus_Nonexistent;
-    CHECK(plasma_send_StatusReply(client_conn->fd,
-                                  client_conn->manager_state->builder,
-                                  &object_id, &status, 1) >= 0);
+    warn_if_sigpipe(plasma_send_StatusReply(client_conn->fd,
+                                            client_conn->manager_state->builder,
+                                            &object_id, &status, 1),
+                    client_conn->fd);
     return;
   }
 
@@ -1427,6 +1435,9 @@ void start_server(const char *store_socket_name,
                   int port,
                   const char *db_addr,
                   int db_port) {
+  /* Ignore SIGPIPE signals. If we don't do this, then when we attempt to write
+   * to a client that has already died, the manager could die. */
+  signal(SIGPIPE, SIG_IGN);
   /* Bind the sockets before we try to connect to the plasma store.
    * In case the bind does not succeed, we want to be able to exit
    * without breaking the pipe to the store. */

src/plasma/plasma_store.c

@@ -353,9 +353,34 @@ void seal_object(client *client_context,
     /* Send notifications to the clients that were waiting for this object. */
     for (int i = 0; i < utarray_len(notify_entry->waiting_clients); ++i) {
       client **c = (client **) utarray_eltptr(notify_entry->waiting_clients, i);
-      CHECK(plasma_send_GetReply((*c)->sock, plasma_state->builder, &object_id,
-                                 &object, 1) >= 0);
-      CHECK(send_fd((*c)->sock, object.handle.store_fd) >= 0);
+      int status;
+      /* Send the get reply to the client. Handle errors on the write. If the
+       * client has hung up, that's ok. */
+      if (plasma_send_GetReply((*c)->sock, plasma_state->builder, &object_id,
+                               &object, 1) < 0) {
+        if (errno == EPIPE || errno == EBADF) {
+          LOG_WARN(
+              "Failed to send a message to client on fd %d. The client may "
+              "have hung up.",
+              (*c)->sock);
+          continue;
+        } else {
+          LOG_FATAL("Failed to send a message to client on fd %d.", (*c)->sock);
+        }
+      }
+      /* Send the object's file descriptor to the client. Handle errors on the
+       * write. If the client has hung up, that's ok. */
+      if (send_fd((*c)->sock, object.handle.store_fd) < 0) {
+        if (errno == EPIPE || errno == EBADF) {
+          LOG_WARN(
+              "Failed to send a message to client on fd %d. The client may "
+              "have hung up.",
+              (*c)->sock);
+          continue;
+        } else {
+          LOG_FATAL("Failed to send a message to client on fd %d.", (*c)->sock);
+        }
+      }
       /* Record that the client is using this object. */
       add_client_to_object_clients(entry, *c);
     }
@@ -458,7 +483,7 @@ void send_notifications(event_loop *loop,
                           send_notifications, plasma_state);
       break;
     } else {
-      CHECKM(0, "This code should be unreachable.");
+      LOG_WARN("Failed to send notification to client on fd %d", client_sock);
     }
     num_processed += 1;
   }
@@ -474,8 +499,14 @@ void send_notifications(event_loop *loop,
 void subscribe_to_updates(client *client_context, int conn) {
   LOG_DEBUG("subscribing to updates");
   plasma_store_state *plasma_state = client_context->plasma_state;
+  /* TODO(rkn): The store could block here if the client doesn't send a file
+   * descriptor. */
   int fd = recv_fd(conn);
-  CHECK(fd >= 0);
+  if (fd < 0) {
+    /* This may mean that the client died before sending the file descriptor. */
+    LOG_WARN("Failed to receive file descriptor from client on fd %d.", conn);
+    return;
+  }
   CHECKM(HASH_CNT(handle, plasma_state->plasma_store_info->objects) == 0,
          "plasma_subscribe should be called before any objects are created.");
   /* Create a new array to buffer notifications that can't be sent to the
@@ -515,19 +546,24 @@ void process_message(event_loop *loop,
                               &metadata_size);
     int error_code = create_object(client_context, object_ids[0], data_size,
                                    metadata_size, &objects[0]);
-    CHECK(plasma_send_CreateReply(client_sock, state->builder, object_ids[0],
-                                  &objects[0], error_code) >= 0);
+    warn_if_sigpipe(
+        plasma_send_CreateReply(client_sock, state->builder, object_ids[0],
+                                &objects[0], error_code),
+        client_sock);
     if (error_code == PlasmaError_OK) {
-      CHECK(send_fd(client_sock, objects[0].handle.store_fd) >= 0);
+      warn_if_sigpipe(send_fd(client_sock, objects[0].handle.store_fd),
+                      client_sock);
     }
   } break;
   case MessageType_PlasmaGetRequest: {
     plasma_read_GetRequest(input, object_ids, 1);
     if (get_object(client_context, client_sock, object_ids[0], &objects[0]) ==
         OBJECT_FOUND) {
-      CHECK(plasma_send_GetReply(client_sock, state->builder, object_ids,
-                                 objects, 1) >= 0);
-      CHECK(send_fd(client_sock, objects[0].handle.store_fd) >= 0);
+      warn_if_sigpipe(plasma_send_GetReply(client_sock, state->builder,
+                                           object_ids, objects, 1),
+                      client_sock);
+      warn_if_sigpipe(send_fd(client_sock, objects[0].handle.store_fd),
+                      client_sock);
     }
   } break;
   case MessageType_PlasmaGetLocalRequest: {
@@ -535,13 +571,19 @@ void process_message(event_loop *loop,
     if (get_object_local(client_context, client_sock, object_ids[0],
                          &objects[0]) == OBJECT_FOUND) {
       int has_object = 1;
-      CHECK(plasma_send_GetLocalReply(client_sock, state->builder, object_ids,
-                                      objects, &has_object, 1) >= 0);
-      CHECK(send_fd(client_sock, objects[0].handle.store_fd) >= 0);
+      warn_if_sigpipe(
+          plasma_send_GetLocalReply(client_sock, state->builder, object_ids,
+                                    objects, &has_object, 1),
+          client_sock);
+      warn_if_sigpipe(send_fd(client_sock, objects[0].handle.store_fd),
+                      client_sock);
     } else {
       int has_object = 0;
-      CHECK(plasma_send_GetLocalReply(client_sock, state->builder, object_ids,
-                                      objects, &has_object, 1) >= 0);
+
+      warn_if_sigpipe(
+          plasma_send_GetLocalReply(client_sock, state->builder, object_ids,
+                                    objects, &has_object, 1),
+          client_sock);
     }
   } break;
   case MessageType_PlasmaReleaseRequest:
@@ -551,11 +593,13 @@ void process_message(event_loop *loop,
   case MessageType_PlasmaContainsRequest:
     plasma_read_ContainsRequest(input, &object_ids[0]);
     if (contains_object(client_context, object_ids[0]) == OBJECT_FOUND) {
-      CHECK(plasma_send_ContainsReply(client_sock, state->builder,
-                                      object_ids[0], 1) >= 0);
+      warn_if_sigpipe(plasma_send_ContainsReply(client_sock, state->builder,
+                                                object_ids[0], 1),
+                      client_sock);
     } else {
-      CHECK(plasma_send_ContainsReply(client_sock, state->builder,
-                                      object_ids[0], 0) >= 0);
+      warn_if_sigpipe(plasma_send_ContainsReply(client_sock, state->builder,
+                                                object_ids[0], 0),
+                      client_sock);
     }
     break;
   case MessageType_PlasmaSealRequest: {
@@ -575,19 +619,21 @@ void process_message(event_loop *loop,
         &num_objects_to_evict, &objects_to_evict);
     remove_objects(client_context->plasma_state, num_objects_to_evict,
                    objects_to_evict);
-    CHECK(plasma_send_EvictReply(client_sock, state->builder,
-                                 num_bytes_evicted) >= 0);
+    warn_if_sigpipe(
+        plasma_send_EvictReply(client_sock, state->builder, num_bytes_evicted),
+        client_sock);
   } break;
   case MessageType_PlasmaSubscribeRequest:
     subscribe_to_updates(client_context, client_sock);
     break;
-  case MessageType_PlasmaConnectRequest:
-    CHECK(plasma_send_ConnectReply(client_sock, state->builder,
-                                   state->plasma_store_info->memory_capacity) >=
-          0);
-    break;
+  case MessageType_PlasmaConnectRequest: {
+    warn_if_sigpipe(
+        plasma_send_ConnectReply(client_sock, state->builder,
+                                 state->plasma_store_info->memory_capacity),
+        client_sock);
+  } break;
   case DISCONNECT_CLIENT: {
-    LOG_DEBUG("Disconnecting client on fd %d", client_sock);
+    LOG_INFO("Disconnecting client on fd %d", client_sock);
     event_loop_remove_file(loop, client_sock);
     /* If this client was using any objects, remove it from the appropriate
      * lists. */
@@ -597,6 +643,10 @@ void process_message(event_loop *loop,
               temp_entry) {
       remove_client_from_object_clients(entry, client_context);
     }
+    /* Note, the store may still attempt to send a message to the disconnected
+     * client (for example, when an object ID that the client was waiting for
+     * is ready). In these cases, the attempt to send the message will fail, but
+     * the store should just ignore the failure. */
   } break;
   default:
     /* This code should be unreachable. */
@@ -629,6 +679,10 @@ void signal_handler(int signal) {
 }
 
 void start_server(char *socket_name, int64_t system_memory) {
+  /* Ignore SIGPIPE signals. If we don't do this, then when we attempt to write
+   * to a client that has already died, the store could die. */
+  signal(SIGPIPE, SIG_IGN);
+  /* Create the event loop. */
   event_loop *loop = event_loop_create();
   plasma_store_state *state = init_plasma_store(loop, system_memory);
   int socket = bind_ipc_sock(socket_name, true);