Plasma and worker node failure. (#373)

* Failing test case

* Local scheduler exits cleanly after plasma store dies

* Tolerate one plasma store failure

* Tolerate plasma store failures on all nodes except head node

* Plasma manager heartbeats

* Component failure tests

* Don't run the helper for Python testing

* Fix C test

* Fix hanging plasma transfer test

* Fix python3

* Consolidate ClientConnection code

* Fix valgrind test

* fix c test

* We can restart worker nodes!

* Fix flatbuffers bug

* Address comments

* Only register actual workers with the local scheduler

* Fix bug

* Fix segfaults

* Add test case that tests for driver liveness, fix local scheduler bug

* Clean up after tests

* Allocate retry info on the stack

* Send SIGKILL before waiting

* Relax unit test conditions

* Driver liveness test case and documentation

src/common/common.h

@@ -20,6 +20,14 @@ extern "C" {
 }
 #endif
 
+/** The duration between heartbeats. These are sent by the plasma manager and
+ *  local scheduler. */
+#define HEARTBEAT_TIMEOUT_MILLISECONDS 100
+/** If a component has not sent a heartbeat in the last NUM_HEARTBEATS_TIMEOUT
+ *  heartbeat intervals, the global scheduler or monitor process will report it
+ *  as dead to the db_client table. */
+#define NUM_HEARTBEATS_TIMEOUT 100
+
 /** Definitions for Ray logging levels. */
 #define RAY_COMMON_DEBUG 0
 #define RAY_COMMON_INFO 1

src/common/io.cc

@@ -193,6 +193,7 @@ int connect_inet_sock(const char *ip_addr, int port) {
   struct hostent *manager = gethostbyname(ip_addr); /* TODO(pcm): cache this */
   if (!manager) {
     LOG_ERROR("Failed to get hostname from address %s:%d.", ip_addr, port);
+    close(fd);
     return -1;
   }
 
@@ -203,6 +204,7 @@ int connect_inet_sock(const char *ip_addr, int port) {
 
   if (connect(fd, (struct sockaddr *) &addr, sizeof(addr)) != 0) {
     LOG_ERROR("Connection to socket failed for address %s:%d.", ip_addr, port);
+    close(fd);
     return -1;
   }
   return fd;

src/common/redis_module/ray_redis_module.cc

@@ -73,6 +73,12 @@ flatbuffers::Offset<flatbuffers::String> RedisStringToFlatbuf(
  * Publish a notification to a client's notification channel about an insertion
  * or deletion to the db client table.
  *
+ * TODO(swang): Use flatbuffers for the notification message.
+ * The format for the published notification is:
+ *  <ray_client_id>:<client type> <aux_address> <is_insertion>
+ * If no auxiliary address is provided, aux_address will be set to ":". If
+ * is_insertion is true, then the last field will be "1", else "0".
+ *
  * @param ctx The Redis context.
  * @param ray_client_id The ID of the database client that was inserted or
  *        deleted.
@@ -159,14 +165,20 @@ int Connect_RedisCommand(RedisModuleCtx *ctx,
   RedisModuleKey *db_client_table_key =
       OpenPrefixedKey(ctx, DB_CLIENT_PREFIX, ray_client_id, REDISMODULE_WRITE);
 
+  if (RedisModule_KeyType(db_client_table_key) != REDISMODULE_KEYTYPE_EMPTY) {
+    return RedisModule_ReplyWithError(ctx, "Client already exists");
+  }
+
   /* This will be used to construct a publish message. */
   RedisModuleString *aux_address = NULL;
   RedisModuleString *aux_address_key =
       RedisModule_CreateString(ctx, "aux_address", strlen("aux_address"));
+  RedisModuleString *deleted = RedisModule_CreateString(ctx, "0", strlen("0"));
 
   RedisModule_HashSet(db_client_table_key, REDISMODULE_HASH_CFIELDS,
                       "ray_client_id", ray_client_id, "node_ip_address",
-                      node_ip_address, "client_type", client_type, NULL);
+                      node_ip_address, "client_type", client_type, "deleted",
+                      deleted, NULL);
 
   for (int i = 4; i < argc; i += 2) {
     RedisModuleString *key = argv[i];
@@ -178,6 +190,7 @@ int Connect_RedisCommand(RedisModuleCtx *ctx,
     }
   }
   /* Clean up. */
+  RedisModule_FreeString(ctx, deleted);
   RedisModule_FreeString(ctx, aux_address_key);
   RedisModule_CloseKey(db_client_table_key);
   if (!PublishDBClientNotification(ctx, ray_client_id, client_type, aux_address,
@@ -213,32 +226,47 @@ int Disconnect_RedisCommand(RedisModuleCtx *ctx,
   /* Get the client type. */
   RedisModuleKey *db_client_table_key =
       OpenPrefixedKey(ctx, DB_CLIENT_PREFIX, ray_client_id, REDISMODULE_WRITE);
-  if (RedisModule_KeyType(db_client_table_key) == REDISMODULE_KEYTYPE_EMPTY) {
-    /* Someone else already deleted this client. */
+
+  RedisModuleString *deleted_string;
+  RedisModule_HashGet(db_client_table_key, REDISMODULE_HASH_CFIELDS, "deleted",
+                      &deleted_string, NULL);
+  long long deleted;
+  int parsed = RedisModule_StringToLongLong(deleted_string, &deleted);
+  RedisModule_FreeString(ctx, deleted_string);
+  if (parsed != REDISMODULE_OK) {
     RedisModule_CloseKey(db_client_table_key);
-    RedisModule_ReplyWithSimpleString(ctx, "OK");
-    return REDISMODULE_OK;
+    return RedisModule_ReplyWithError(ctx, "Unable to parse deleted field");
   }
 
-  RedisModuleString *client_type;
-  RedisModuleString *aux_address;
-  RedisModule_HashGet(db_client_table_key, REDISMODULE_HASH_CFIELDS,
-                      "client_type", &client_type, "aux_address", &aux_address,
-                      NULL);
+  bool published = true;
+  if (deleted == 0) {
+    /* Remove the client from the client table. */
+    RedisModuleString *deleted =
+        RedisModule_CreateString(ctx, "1", strlen("1"));
+    RedisModule_HashSet(db_client_table_key, REDISMODULE_HASH_CFIELDS,
+                        "deleted", deleted, NULL);
+    RedisModule_FreeString(ctx, deleted);
+
+    RedisModuleString *client_type;
+    RedisModuleString *aux_address;
+    RedisModule_HashGet(db_client_table_key, REDISMODULE_HASH_CFIELDS,
+                        "client_type", &client_type, "aux_address",
+                        &aux_address, NULL);
+
+    /* Publish the deletion notification on the db client channel. */
+    published = PublishDBClientNotification(ctx, ray_client_id, client_type,
+                                            aux_address, false);
+    if (aux_address != NULL) {
+      RedisModule_FreeString(ctx, aux_address);
+    }
+    RedisModule_FreeString(ctx, client_type);
+  }
 
-  /* Remove the client from the client table. */
-  CHECK_ERROR(RedisModule_DeleteKey(db_client_table_key),
-              "Unable to delete db client key.");
   RedisModule_CloseKey(db_client_table_key);
 
-  /* Publish the deletion notification on the db client channel. */
-  bool published = PublishDBClientNotification(ctx, ray_client_id, client_type,
-                                               aux_address, false);
-
-  RedisModule_FreeString(ctx, aux_address);
-  RedisModule_FreeString(ctx, client_type);
-
   if (!published) {
+    /* Return an error message if we weren't able to publish the deletion
+     * notification. */
     return RedisModule_ReplyWithError(ctx, "PUBLISH unsuccessful");
   }
 

src/common/state/db_client_table.cc

@@ -27,3 +27,14 @@ void db_client_table_subscribe(
                       (table_done_callback) done_callback,
                       redis_db_client_table_subscribe, user_context);
 }
+
+void plasma_manager_send_heartbeat(DBHandle *db_handle) {
+  RetryInfo heartbeat_retry;
+  heartbeat_retry.num_retries = 0;
+  heartbeat_retry.timeout = HEARTBEAT_TIMEOUT_MILLISECONDS;
+  heartbeat_retry.fail_callback = NULL;
+
+  init_table_callback(db_handle, NIL_ID, __func__, NULL,
+                      (RetryInfo *) &heartbeat_retry, NULL,
+                      redis_plasma_manager_send_heartbeat, NULL);
+}

src/common/state/db_client_table.h

@@ -65,4 +65,20 @@ typedef struct {
   void *subscribe_context;
 } DBClientTableSubscribeData;
 
+/*
+ * ==== Plasma manager heartbeats ====
+ */
+
+/**
+ * Start sending heartbeats to the plasma_managers channel. Each
+ * heartbeat contains this database client's ID. Heartbeats can be subscribed
+ * to through the plasma_managers channel. Once called, this "retries" the
+ * heartbeat operation forever, every HEARTBEAT_TIMEOUT_MILLISECONDS
+ * milliseconds.
+ *
+ * @param db_handle Database handle.
+ * @return Void.
+ */
+void plasma_manager_send_heartbeat(DBHandle *db_handle);
+
 #endif /* DB_CLIENT_TABLE_H */

src/common/state/redis.cc

@@ -1068,6 +1068,23 @@ void redis_local_scheduler_table_send_info(TableCallbackData *callback_data) {
   }
 }
 
+void redis_plasma_manager_send_heartbeat(TableCallbackData *callback_data) {
+  DBHandle *db = callback_data->db_handle;
+  /* NOTE(swang): We purposefully do not provide a callback, leaving the table
+   * operation and timer active. This allows us to send a new heartbeat every
+   * HEARTBEAT_TIMEOUT_MILLISECONDS without having to allocate and deallocate
+   * memory for callback data each time. */
+  int status = redisAsyncCommand(
+      db->context, NULL, (void *) callback_data->timer_id,
+      "PUBLISH plasma_managers %b", db->client.id, sizeof(db->client.id));
+  if ((status == REDIS_ERR) || db->context->err) {
+    LOG_REDIS_DEBUG(db->context,
+                    "error in redis_plasma_manager_send_heartbeat");
+  }
+  /* Clean up the timer and callback. */
+  destroy_timer_callback(db->loop, callback_data);
+}
+
 void redis_actor_notification_table_subscribe_callback(redisAsyncContext *c,
                                                        void *r,
                                                        void *privdata) {

src/common/state/redis.h

@@ -253,6 +253,8 @@ void redis_local_scheduler_table_subscribe(TableCallbackData *callback_data);
  */
 void redis_local_scheduler_table_send_info(TableCallbackData *callback_data);
 
+void redis_plasma_manager_send_heartbeat(TableCallbackData *callback_data);
+
 /**
  * Subscribe to updates about newly created actors.
  *

src/global_scheduler/global_scheduler.cc

@@ -386,6 +386,11 @@ int task_cleanup_handler(event_loop *loop, timer_id id, void *context) {
     }
   }
 
+  return GLOBAL_SCHEDULER_TASK_CLEANUP_MILLISECONDS;
+}
+
+int heartbeat_timeout_handler(event_loop *loop, timer_id id, void *context) {
+  GlobalSchedulerState *state = (GlobalSchedulerState *) context;
   /* Check for local schedulers that have missed a number of heartbeats. If any
    * local schedulers have died, notify others so that the state can be cleaned
    * up. */
@@ -395,8 +400,7 @@ int task_cleanup_handler(event_loop *loop, timer_id id, void *context) {
   for (int i = utarray_len(state->local_schedulers) - 1; i >= 0; --i) {
     local_scheduler_ptr =
         (LocalScheduler *) utarray_eltptr(state->local_schedulers, i);
-    if (local_scheduler_ptr->num_heartbeats_missed >=
-        GLOBAL_SCHEDULER_HEARTBEAT_TIMEOUT) {
+    if (local_scheduler_ptr->num_heartbeats_missed >= NUM_HEARTBEATS_TIMEOUT) {
       LOG_WARN(
           "Missed too many heartbeats from local scheduler, marking as dead.");
       /* Notify others by updating the global state. */
@@ -409,7 +413,7 @@ int task_cleanup_handler(event_loop *loop, timer_id id, void *context) {
   }
 
   /* Reset the timer. */
-  return GLOBAL_SCHEDULER_TASK_CLEANUP_MILLISECONDS;
+  return HEARTBEAT_TIMEOUT_MILLISECONDS;
 }
 
 void start_server(const char *redis_addr, int redis_port) {
@@ -442,6 +446,8 @@ void start_server(const char *redis_addr, int redis_port) {
    * timer should notice and schedule the task. */
   event_loop_add_timer(loop, GLOBAL_SCHEDULER_TASK_CLEANUP_MILLISECONDS,
                        task_cleanup_handler, g_state);
+  event_loop_add_timer(loop, HEARTBEAT_TIMEOUT_MILLISECONDS,
+                       heartbeat_timeout_handler, g_state);
   /* Start the event loop. */
   event_loop_run(loop);
 }

src/global_scheduler/global_scheduler.h

@@ -11,10 +11,6 @@
 /* The frequency with which the global scheduler checks if there are any tasks
  * that haven't been scheduled yet. */
 #define GLOBAL_SCHEDULER_TASK_CLEANUP_MILLISECONDS 100
-/* If a local scheduler has not sent a heartbeat in the last
- * GLOBAL_SCHEDULER_HEARTBEAT_TIMEOUT heartbeat intervals, we will report it
- * dead to the db_client table. */
-#define GLOBAL_SCHEDULER_HEARTBEAT_TIMEOUT 100
 
 /** Contains all information that is associated with a local scheduler. */
 typedef struct {

src/local_scheduler/local_scheduler.cc

@@ -146,6 +146,11 @@ void kill_worker(LocalSchedulerClient *worker, bool cleanup) {
 }
 
 void LocalSchedulerState_free(LocalSchedulerState *state) {
+  /* Reset the SIGTERM handler to default behavior, so we try to clean up the
+   * local scheduler at most once. If a SIGTERM is caught afterwards, there is
+   * the possibility of orphan worker processes. */
+  signal(SIGTERM, SIG_DFL);
+
   /* Free the command for starting new workers. */
   if (state->config.start_worker_command != NULL) {
     int i = 0;
@@ -471,7 +476,10 @@ void process_plasma_notification(event_loop *loop,
   /* Read the notification from Plasma. */
   uint8_t *notification = read_message_async(loop, client_sock);
   if (!notification) {
-    return;
+    /* The store has closed the socket. */
+    LocalSchedulerState_free(state);
+    LOG_FATAL(
+        "Lost connection to the plasma store, local scheduler is exiting!");
   }
   auto object_info = flatbuffers::GetRoot<ObjectInfo>(notification);
   ObjectID object_id = from_flatbuf(object_info->object_id());
@@ -773,6 +781,10 @@ LocalSchedulerState *g_state;
 void signal_handler(int signal) {
   LOG_DEBUG("Signal was %d", signal);
   if (signal == SIGTERM) {
+    /* NOTE(swang): This call removes the SIGTERM handler to ensure that we
+     * free the local scheduler state at most once. If another SIGTERM is
+     * caught during this call, there is the possibility of orphan worker
+     * processes. */
     LocalSchedulerState_free(g_state);
     exit(0);
   }
@@ -842,7 +854,7 @@ int heartbeat_handler(event_loop *loop, timer_id id, void *context) {
   /* Publish the heartbeat to all subscribers of the local scheduler table. */
   local_scheduler_table_send_info(state->db, &info, NULL);
   /* Reset the timer. */
-  return LOCAL_SCHEDULER_HEARTBEAT_TIMEOUT_MILLISECONDS;
+  return HEARTBEAT_TIMEOUT_MILLISECONDS;
 }
 
 void start_server(const char *node_ip_address,
@@ -887,7 +899,7 @@ void start_server(const char *node_ip_address,
    * scheduler to the local scheduler table. This message also serves as a
    * heartbeat. */
   if (g_state->db != NULL) {
-    event_loop_add_timer(loop, LOCAL_SCHEDULER_HEARTBEAT_TIMEOUT_MILLISECONDS,
+    event_loop_add_timer(loop, HEARTBEAT_TIMEOUT_MILLISECONDS,
                          heartbeat_handler, g_state);
   }
   /* Create a timer for fetching queued tasks' missing object dependencies. */

src/local_scheduler/local_scheduler.h

@@ -4,9 +4,6 @@
 #include "task.h"
 #include "event_loop.h"
 
-/* The duration between local scheduler heartbeats. */
-#define LOCAL_SCHEDULER_HEARTBEAT_TIMEOUT_MILLISECONDS 100
-
 /* The duration that we wait after sending a worker SIGTERM before sending the
  * worker SIGKILL. */
 #define KILL_WORKER_TIMEOUT_MILLISECONDS 100

src/local_scheduler/local_scheduler_client.cc

@@ -9,18 +9,26 @@
 
 LocalSchedulerConnection *LocalSchedulerConnection_init(
     const char *local_scheduler_socket,
-    ActorID actor_id) {
+    ActorID actor_id,
+    bool is_worker) {
   LocalSchedulerConnection *result =
       (LocalSchedulerConnection *) malloc(sizeof(LocalSchedulerConnection));
   result->conn = connect_ipc_sock_retry(local_scheduler_socket, -1, -1);
-  flatbuffers::FlatBufferBuilder fbb;
-  auto message =
-      CreateRegisterWorkerInfo(fbb, to_flatbuf(fbb, actor_id), getpid());
-  fbb.Finish(message);
-  /* Register the process ID with the local scheduler. */
-  int success = write_message(result->conn, MessageType_RegisterWorkerInfo,
-                              fbb.GetSize(), fbb.GetBufferPointer());
-  CHECKM(success == 0, "Unable to register worker with local scheduler");
+
+  if (is_worker) {
+    /* If we are a worker, register with the local scheduler.
+     * NOTE(swang): If the local scheduler exits and we are registered as a
+     * worker, we will get killed. */
+    flatbuffers::FlatBufferBuilder fbb;
+    auto message =
+        CreateRegisterWorkerInfo(fbb, to_flatbuf(fbb, actor_id), getpid());
+    fbb.Finish(message);
+    /* Register the process ID with the local scheduler. */
+    int success = write_message(result->conn, MessageType_RegisterWorkerInfo,
+                                fbb.GetSize(), fbb.GetBufferPointer());
+    CHECKM(success == 0, "Unable to register worker with local scheduler");
+  }
+
   return result;
 }
 

src/local_scheduler/local_scheduler_client.h

@@ -17,11 +17,14 @@ typedef struct {
  *        local scheduler.
  * @param actor_id The ID of the actor running on this worker. If no actor is
  *        running on this actor, this should be NIL_ACTOR_ID.
+ * @param is_worker Whether this client is a worker. If it is a worker, an
+ *        additional message will be sent to register as one.
  * @return The connection information.
  */
 LocalSchedulerConnection *LocalSchedulerConnection_init(
     const char *local_scheduler_socket,
-    ActorID actor_id);
+    ActorID actor_id,
+    bool is_worker);
 
 /**
  * Disconnect from the local scheduler.

src/local_scheduler/local_scheduler_extension.cc

@@ -18,13 +18,14 @@ static int PyLocalSchedulerClient_init(PyLocalSchedulerClient *self,
                                        PyObject *kwds) {
   char *socket_name;
   ActorID actor_id;
-  if (!PyArg_ParseTuple(args, "sO&", &socket_name, PyStringToUniqueID,
-                        &actor_id)) {
+  PyObject *is_worker;
+  if (!PyArg_ParseTuple(args, "sO&O", &socket_name, PyStringToUniqueID,
+                        &actor_id, &is_worker)) {
     return -1;
   }
   /* Connect to the local scheduler. */
-  self->local_scheduler_connection =
-      LocalSchedulerConnection_init(socket_name, actor_id);
+  self->local_scheduler_connection = LocalSchedulerConnection_init(
+      socket_name, actor_id, (bool) PyObject_IsTrue(is_worker));
   return 0;
 }
 

src/local_scheduler/test/local_scheduler_tests.cc

@@ -103,7 +103,7 @@ LocalSchedulerMock *LocalSchedulerMock_init(int num_workers,
       sizeof(LocalSchedulerConnection *) * num_mock_workers);
   for (int i = 0; i < num_mock_workers; ++i) {
     mock->conns[i] = LocalSchedulerConnection_init(
-        utstring_body(local_scheduler_socket_name), NIL_ACTOR_ID);
+        utstring_body(local_scheduler_socket_name), NIL_ACTOR_ID, true);
     new_client_connection(mock->loop, mock->local_scheduler_fd,
                           (void *) mock->local_scheduler_state, 0);
   }

src/plasma/plasma.cc

@@ -7,16 +7,16 @@
 
 #include "plasma_protocol.h"
 
-void warn_if_sigpipe(int status, int client_sock) {
+bool warn_if_sigpipe(int status, int client_sock) {
   if (status >= 0) {
-    return;
+    return false;
   }
-  if (errno == EPIPE || errno == EBADF) {
+  if (errno == EPIPE || errno == EBADF || errno == ECONNRESET) {
     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;
+    return true;
   }
   LOG_FATAL("Failed to write message to client on fd %d.", client_sock);
 }

src/plasma/plasma.h

@@ -132,7 +132,7 @@ typedef struct {
  *        information.
  * @return Void.
  */
-void warn_if_sigpipe(int status, int client_sock);
+bool warn_if_sigpipe(int status, int client_sock);
 
 uint8_t *create_object_info_buffer(ObjectInfoT *object_info);
 

src/plasma/plasma_manager.cc

@@ -38,6 +38,7 @@
 #include "state/object_table.h"
 #include "state/error_table.h"
 #include "state/task_table.h"
+#include "state/db_client_table.h"
 
 /**
  * Process either the fetch or the status request.
@@ -266,9 +267,6 @@ struct ClientConnection {
   int fd;
   /** Timer id for timing out wait (or fetch). */
   int64_t timer_id;
-  /** The objects that we are waiting for and their callback
-   *  contexts, for either a fetch or a wait operation. */
-  ClientObjectRequest *active_objects;
   /** The number of objects that we have left to return for
    *  this fetch or wait operation. */
   int num_return_objects;
@@ -280,6 +278,34 @@ struct ClientConnection {
   UT_hash_handle manager_hh;
 };
 
+/**
+ * Initializes the state for a plasma client connection.
+ *
+ * @param state The plasma manager state.
+ * @param client_sock The socket that we use to communicate with the client.
+ * @param client_key A string uniquely identifying the client. If the client is
+ *        another plasma manager, this is the manager's IP address and port.
+ *        Else, the client is the string of the client's socket.
+ * @return A pointer to the initialized client state.
+ */
+ClientConnection *ClientConnection_init(PlasmaManagerState *state,
+                                        int client_sock,
+                                        char *client_key);
+
+/**
+ * Destroys a plasma client and its connection.
+ *
+ * @param client_conn The client's state.
+ * @return Void.
+ */
+void ClientConnection_free(ClientConnection *client_conn);
+
+void object_table_subscribe_callback(ObjectID object_id,
+                                     int64_t data_size,
+                                     int manager_count,
+                                     const char *manager_vector[],
+                                     void *context);
+
 ObjectWaitRequests **object_wait_requests_table_ptr_from_type(
     PlasmaManagerState *manager_state,
     int type) {
@@ -505,30 +531,10 @@ PlasmaManagerState *PlasmaManagerState_init(const char *store_socket_name,
 }
 
 void PlasmaManagerState_free(PlasmaManagerState *state) {
-  ClientConnection *manager_conn, *tmp;
-  HASH_ITER(manager_hh, state->manager_connections, manager_conn, tmp) {
-    HASH_DELETE(manager_hh, state->manager_connections, manager_conn);
-
-    /* Free the hash table of object IDs that are waiting to be transferred. */
-    PlasmaRequestBuffer *request_buffer, *tmp_buffer;
-    HASH_ITER(hh, manager_conn->pending_object_transfers, request_buffer,
-              tmp_buffer) {
-      /* We do not free the PlasmaRequestBuffer here because it is also in the
-       * transfer queue and will be freed below. */
-      HASH_DELETE(hh, manager_conn->pending_object_transfers, request_buffer);
-    }
-
-    /* Free the transfer queue. */
-    PlasmaRequestBuffer *head = manager_conn->transfer_queue;
-    while (head) {
-      DL_DELETE(manager_conn->transfer_queue, head);
-      free(head);
-      head = manager_conn->transfer_queue;
-    }
-    /* Close the manager connection and free the remaining state. */
-    close(manager_conn->fd);
-    free(manager_conn->ip_addr_port);
-    free(manager_conn);
+  ClientConnection *manager_conn, *tmp_manager_conn;
+  HASH_ITER(manager_hh, state->manager_connections, manager_conn,
+            tmp_manager_conn) {
+    ClientConnection_free(manager_conn);
   }
 
   if (state->fetch_requests != NULL) {
@@ -538,6 +544,12 @@ void PlasmaManagerState_free(PlasmaManagerState *state) {
     }
   }
 
+  AvailableObject *entry, *tmp_object_entry;
+  HASH_ITER(hh, state->local_available_objects, entry, tmp_object_entry) {
+    HASH_DELETE(hh, state->local_available_objects, entry);
+    free(entry);
+  }
+
   plasma_disconnect(state->plasma_conn);
   event_loop_destroy(state->loop);
   free_protocol_builder(state->builder);
@@ -601,9 +613,10 @@ void send_queued_request(event_loop *loop,
   }
 
   PlasmaRequestBuffer *buf = conn->transfer_queue;
+  bool sigpipe = false;
   switch (buf->type) {
   case MessageType_PlasmaDataRequest:
-    warn_if_sigpipe(
+    sigpipe = warn_if_sigpipe(
         plasma_send_DataRequest(conn->fd, state->builder, buf->object_id,
                                 state->addr, state->port),
         conn->fd);
@@ -613,7 +626,7 @@ void send_queued_request(event_loop *loop,
     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. */
-      warn_if_sigpipe(
+      sigpipe = warn_if_sigpipe(
           plasma_send_DataReply(conn->fd, state->builder, buf->object_id,
                                 buf->data_size, buf->metadata_size),
           conn->fd);
@@ -624,6 +637,12 @@ void send_queued_request(event_loop *loop,
     LOG_FATAL("Buffered request has unknown type.");
   }
 
+  /* If there was a SIGPIPE, stop sending to this manager. */
+  if (sigpipe) {
+    ClientConnection_free(conn);
+    return;
+  }
+
   /* If we are done sending this request, remove it from the transfer queue. */
   if (conn->cursor == 0) {
     if (buf->type == MessageType_PlasmaDataReply) {
@@ -728,21 +747,13 @@ ClientConnection *get_manager_connection(PlasmaManagerState *state,
             utstring_len(ip_addr_port), manager_conn);
   if (!manager_conn) {
     /* If we don't already have a connection to this manager, start one. */
-    int fd = connect_inet_sock_retry(ip_addr, port, -1, -1);
-    /* TODO(swang): Handle the case when connection to this manager was
-     * unsuccessful. */
-    CHECK(fd >= 0);
-    manager_conn = (ClientConnection *) malloc(sizeof(ClientConnection));
-    manager_conn->fd = fd;
-    manager_conn->manager_state = state;
-    manager_conn->transfer_queue = NULL;
-    manager_conn->pending_object_transfers = NULL;
-    manager_conn->cursor = 0;
-    manager_conn->ip_addr_port = strdup(utstring_body(ip_addr_port));
-    HASH_ADD_KEYPTR(manager_hh,
-                    manager_conn->manager_state->manager_connections,
-                    manager_conn->ip_addr_port,
-                    strlen(manager_conn->ip_addr_port), manager_conn);
+    int fd = connect_inet_sock(ip_addr, port);
+    if (fd < 0) {
+      return NULL;
+    }
+
+    manager_conn =
+        ClientConnection_init(state, fd, utstring_body(ip_addr_port));
   }
   utstring_free(ip_addr_port);
   return manager_conn;
@@ -755,6 +766,9 @@ void process_transfer_request(event_loop *loop,
                               ClientConnection *conn) {
   ClientConnection *manager_conn =
       get_manager_connection(conn->manager_state, addr, port);
+  if (manager_conn == NULL) {
+    return;
+  }
 
   /* If there is already a request in the transfer queue with the same object
    * ID, do not add the transfer request. */
@@ -765,6 +779,20 @@ void process_transfer_request(event_loop *loop,
     return;
   }
 
+  /* Allocate and append the request to the transfer queue. */
+  ObjectBuffer object_buffer;
+  /* We pass in 0 to indicate that the command should return immediately. */
+  plasma_get(conn->manager_state->plasma_conn, &obj_id, 1, 0, &object_buffer);
+  if (object_buffer.data_size == -1) {
+    /* If the object wasn't locally available, exit immediately. If the object
+     * later appears locally, the requesting plasma manager should request the
+     * transfer again. */
+    LOG_WARN(
+        "Unable to transfer object to requesting plasma manager, object not "
+        "local.");
+    return;
+  }
+
   /* If we already have a connection to this manager and its inactive,
    * (re)register it with the event loop again. */
   if (manager_conn->transfer_queue == NULL) {
@@ -772,38 +800,17 @@ void process_transfer_request(event_loop *loop,
                         send_queued_request, manager_conn);
   }
 
-  /* Allocate and append the request to the transfer queue. */
-  /* TODO(swang): A non-blocking plasma_get, or else we could block here
-   * forever if we don't end up sealing this object. */
-  /* The corresponding call to plasma_release will happen in
-   * write_object_chunk. */
-  /* TODO(rkn): The manager currently will block here if the object is not
-   * present in the store. This is completely unacceptable. The manager should
-   * do a non-blocking get call on the store, and if the object isn't there then
-   * perhaps the manager should initiate the transfer when it receives a
-   * notification from the store that the object is present. */
-  ObjectBuffer obj_buffer;
-  int counter = 0;
-  do {
-    /* We pass in 0 to indicate that the command should return immediately. */
-    ObjectID obj_id_array[1] = {obj_id};
-    plasma_get(conn->manager_state->plasma_conn, obj_id_array, 1, 0,
-               &obj_buffer);
-    if (counter > 0) {
-      LOG_WARN("Blocking in the plasma manager.");
-    }
-    counter += 1;
-  } while (obj_buffer.data_size == -1);
-  DCHECK(obj_buffer.metadata == obj_buffer.data + obj_buffer.data_size);
+  DCHECK(object_buffer.metadata ==
+         object_buffer.data + object_buffer.data_size);
   PlasmaRequestBuffer *buf =
       (PlasmaRequestBuffer *) malloc(sizeof(PlasmaRequestBuffer));
   buf->type = MessageType_PlasmaDataReply;
   buf->object_id = obj_id;
   /* We treat buf->data as a pointer to the concatenated data and metadata, so
    * we don't actually use buf->metadata. */
-  buf->data = obj_buffer.data;
-  buf->data_size = obj_buffer.data_size;
-  buf->metadata_size = obj_buffer.metadata_size;
+  buf->data = object_buffer.data;
+  buf->data_size = object_buffer.data_size;
+  buf->metadata_size = object_buffer.metadata_size;
 
   DL_APPEND(manager_conn->transfer_queue, buf);
   HASH_ADD(hh, manager_conn->pending_object_transfers, object_id,
@@ -868,14 +875,7 @@ void process_data_request(event_loop *loop,
 }
 
 void request_transfer_from(PlasmaManagerState *manager_state,
-                           ObjectID object_id) {
-  FetchRequest *fetch_req;
-  HASH_FIND(hh, manager_state->fetch_requests, &object_id, sizeof(object_id),
-            fetch_req);
-  /* TODO(rkn): This probably can be NULL so we should remove this check, and
-   * instead return in the case where there is no fetch request. */
-  CHECK(fetch_req != NULL);
-
+                           FetchRequest *fetch_req) {
   CHECK(fetch_req->manager_count > 0);
   CHECK(fetch_req->next_manager >= 0 &&
         fetch_req->next_manager < fetch_req->manager_count);
@@ -886,30 +886,33 @@ void request_transfer_from(PlasmaManagerState *manager_state,
 
   ClientConnection *manager_conn =
       get_manager_connection(manager_state, addr, port);
+  if (manager_conn != NULL) {
+    /* Check that this manager isn't trying to request an object from itself.
+     * TODO(rkn): Later this should not be fatal. */
+    uint8_t temp_addr[4];
+    sscanf(addr, "%hhu.%hhu.%hhu.%hhu", &temp_addr[0], &temp_addr[1],
+           &temp_addr[2], &temp_addr[3]);
+    if (memcmp(temp_addr, manager_state->addr, 4) == 0 &&
+        port == manager_state->port) {
+      LOG_FATAL(
+          "This manager is attempting to request a transfer from itself.");
+    }
 
-  /* Check that this manager isn't trying to request an object from itself.
-   * TODO(rkn): Later this should not be fatal. */
-  uint8_t temp_addr[4];
-  sscanf(addr, "%hhu.%hhu.%hhu.%hhu", &temp_addr[0], &temp_addr[1],
-         &temp_addr[2], &temp_addr[3]);
-  if (memcmp(temp_addr, manager_state->addr, 4) == 0 &&
-      port == manager_state->port) {
-    LOG_FATAL("This manager is attempting to request a transfer from itself.");
+    PlasmaRequestBuffer *transfer_request =
+        (PlasmaRequestBuffer *) malloc(sizeof(PlasmaRequestBuffer));
+    transfer_request->type = MessageType_PlasmaDataRequest;
+    transfer_request->object_id = fetch_req->object_id;
+
+    if (manager_conn->transfer_queue == NULL) {
+      /* If we already have a connection to this manager and it's inactive,
+       * (re)register it with the event loop. */
+      event_loop_add_file(manager_state->loop, manager_conn->fd,
+                          EVENT_LOOP_WRITE, send_queued_request, manager_conn);
+    }
+    /* Add this transfer request to this connection's transfer queue. */
+    DL_APPEND(manager_conn->transfer_queue, transfer_request);
   }
 
-  PlasmaRequestBuffer *transfer_request =
-      (PlasmaRequestBuffer *) malloc(sizeof(PlasmaRequestBuffer));
-  transfer_request->type = MessageType_PlasmaDataRequest;
-  transfer_request->object_id = fetch_req->object_id;
-
-  if (manager_conn->transfer_queue == NULL) {
-    /* If we already have a connection to this manager and its inactive,
-     * (re)register it with the event loop. */
-    event_loop_add_file(manager_state->loop, manager_conn->fd, EVENT_LOOP_WRITE,
-                        send_queued_request, manager_conn);
-  }
-  /* Add this transfer request to this connection's transfer queue. */
-  DL_APPEND(manager_conn->transfer_queue, transfer_request);
   /* On the next attempt, try the next manager in manager_vector. */
   fetch_req->next_manager += 1;
   fetch_req->next_manager %= fetch_req->manager_count;
@@ -917,13 +920,39 @@ void request_transfer_from(PlasmaManagerState *manager_state,
 
 int fetch_timeout_handler(event_loop *loop, timer_id id, void *context) {
   PlasmaManagerState *manager_state = (PlasmaManagerState *) context;
-  /* Loop over the fetch requests and reissue the requests. */
+
+  /* Allocate a vector of object IDs to resend requests for location
+   * notifications. */
+  int num_object_ids_to_request = 0;
+  int num_object_ids = HASH_COUNT(manager_state->fetch_requests);
+  /* This is allocating more space than necessary, but we do not know the exact
+   * number of object IDs to request notifications for yet. */
+  ObjectID *object_ids_to_request =
+      (ObjectID *) malloc(num_object_ids * sizeof(ObjectID));
+
+  /* Loop over the fetch requests and reissue requests for objects whose
+   * locations we know. */
   FetchRequest *fetch_req, *tmp;
   HASH_ITER(hh, manager_state->fetch_requests, fetch_req, tmp) {
     if (fetch_req->manager_count > 0) {
-      request_transfer_from(manager_state, fetch_req->object_id);
+      request_transfer_from(manager_state, fetch_req);
+      /* If we've tried all of the managers that we know about for this object,
+       * add this object to the list to resend requests for. */
+      if (fetch_req->next_manager == 0) {
+        object_ids_to_request[num_object_ids_to_request] = fetch_req->object_id;
+        ++num_object_ids_to_request;
+      }
     }
   }
+
+  /* Resend requests for notifications on these objects' locations. */
+  if (num_object_ids_to_request > 0 && manager_state->db != NULL) {
+    object_table_request_notifications(manager_state->db,
+                                       num_object_ids_to_request,
+                                       object_ids_to_request, NULL);
+  }
+  free(object_ids_to_request);
+
   return MANAGER_TIMEOUT;
 }
 
@@ -980,7 +1009,7 @@ void request_transfer(ObjectID object_id,
   }
   /* Wait for the object data for the default number of retries, which timeout
    * after a default interval. */
-  request_transfer_from(manager_state, object_id);
+  request_transfer_from(manager_state, fetch_req);
 }
 
 /* This method is only called from the tests. */
@@ -1351,7 +1380,9 @@ void process_object_notification(event_loop *loop,
   PlasmaManagerState *state = (PlasmaManagerState *) context;
   uint8_t *notification = read_message_async(loop, client_sock);
   if (notification == NULL) {
-    return;
+    PlasmaManagerState_free(state);
+    LOG_FATAL(
+        "Lost connection to the plasma store, plasma manager is exiting!");
   }
   auto object_info = flatbuffers::GetRoot<ObjectInfo>(notification);
   /* Add object to locally available object. */
@@ -1367,6 +1398,79 @@ void process_object_notification(event_loop *loop,
   free(notification);
 }
 
+/* TODO(pcm): Split this into two methods: new_worker_connection
+ * and new_manager_connection and also split ClientConnection
+ * into two structs, one for workers and one for other plasma managers. */
+ClientConnection *ClientConnection_init(PlasmaManagerState *state,
+                                        int client_sock,
+                                        char *client_key) {
+  /* Create a new data connection context per client. */
+  ClientConnection *conn =
+      (ClientConnection *) malloc(sizeof(ClientConnection));
+  conn->manager_state = state;
+  conn->cursor = 0;
+  conn->transfer_queue = NULL;
+  conn->pending_object_transfers = NULL;
+  conn->fd = client_sock;
+  conn->num_return_objects = 0;
+
+  conn->ip_addr_port = strdup(client_key);
+  HASH_ADD_KEYPTR(manager_hh, conn->manager_state->manager_connections,
+                  conn->ip_addr_port, strlen(conn->ip_addr_port), conn);
+  return conn;
+}
+
+ClientConnection *ClientConnection_listen(event_loop *loop,
+                                          int listener_sock,
+                                          void *context,
+                                          int events) {
+  PlasmaManagerState *state = (PlasmaManagerState *) context;
+  int new_socket = accept_client(listener_sock);
+  char client_key[8];
+  snprintf(client_key, sizeof(client_key), "%d", new_socket);
+  ClientConnection *conn = ClientConnection_init(state, new_socket, client_key);
+
+  event_loop_add_file(loop, new_socket, EVENT_LOOP_READ, process_message, conn);
+  LOG_DEBUG("New client connection with fd %d", new_socket);
+  return conn;
+}
+
+void ClientConnection_free(ClientConnection *client_conn) {
+  PlasmaManagerState *state = client_conn->manager_state;
+  HASH_DELETE(manager_hh, state->manager_connections, client_conn);
+  /* Free the hash table of object IDs that are waiting to be transferred. */
+  PlasmaRequestBuffer *request_buffer, *tmp_buffer;
+  HASH_ITER(hh, client_conn->pending_object_transfers, request_buffer,
+            tmp_buffer) {
+    /* We do not free the PlasmaRequestBuffer here because it is also in the
+     * transfer queue and will be freed below. */
+    HASH_DELETE(hh, client_conn->pending_object_transfers, request_buffer);
+  }
+
+  /* Free the transfer queue. */
+  PlasmaRequestBuffer *head = client_conn->transfer_queue;
+  while (head) {
+    DL_DELETE(client_conn->transfer_queue, head);
+    free(head);
+    head = client_conn->transfer_queue;
+  }
+  /* Close the manager connection and free the remaining state. */
+  close(client_conn->fd);
+  free(client_conn->ip_addr_port);
+  free(client_conn);
+}
+
+void handle_new_client(event_loop *loop,
+                       int listener_sock,
+                       void *context,
+                       int events) {
+  (void) ClientConnection_listen(loop, listener_sock, context, events);
+}
+
+int get_client_sock(ClientConnection *conn) {
+  return conn->fd;
+}
+
 void process_message(event_loop *loop,
                      int client_sock,
                      void *context,
@@ -1433,11 +1537,8 @@ void process_message(event_loop *loop,
   } break;
   case DISCONNECT_CLIENT: {
     LOG_INFO("Disconnecting client on fd %d", client_sock);
-    /* TODO(swang): Check if this connection was to a plasma manager. If so,
-     * delete it. */
     event_loop_remove_file(loop, client_sock);
-    close(client_sock);
-    free(conn);
+    ClientConnection_free(conn);
   } break;
   default:
     LOG_FATAL("invalid request %" PRId64, type);
@@ -1445,39 +1546,10 @@ void process_message(event_loop *loop,
   free(data);
 }
 
-/* TODO(pcm): Split this into two methods: new_worker_connection
- * and new_manager_connection and also split ClientConnection
- * into two structs, one for workers and one for other plasma managers. */
-ClientConnection *ClientConnection_init(event_loop *loop,
-                                        int listener_sock,
-                                        void *context,
-                                        int events) {
-  int new_socket = accept_client(listener_sock);
-  /* Create a new data connection context per client. */
-  ClientConnection *conn =
-      (ClientConnection *) malloc(sizeof(ClientConnection));
-  conn->manager_state = (PlasmaManagerState *) context;
-  conn->cursor = 0;
-  conn->transfer_queue = NULL;
-  /* TODO(rkn): Is this pending_object_transfers hash table ever used? */
-  conn->pending_object_transfers = NULL;
-  conn->fd = new_socket;
-  conn->active_objects = NULL;
-  conn->num_return_objects = 0;
-  event_loop_add_file(loop, new_socket, EVENT_LOOP_READ, process_message, conn);
-  LOG_DEBUG("New client connection with fd %d", new_socket);
-  return conn;
-}
-
-void handle_new_client(event_loop *loop,
-                       int listener_sock,
-                       void *context,
-                       int events) {
-  (void) ClientConnection_init(loop, listener_sock, context, events);
-}
-
-int get_client_sock(ClientConnection *conn) {
-  return conn->fd;
+int heartbeat_handler(event_loop *loop, timer_id id, void *context) {
+  PlasmaManagerState *state = (PlasmaManagerState *) context;
+  plasma_manager_send_heartbeat(state->db);
+  return HEARTBEAT_TIMEOUT_MILLISECONDS;
 }
 
 void start_server(const char *store_socket_name,
@@ -1523,6 +1595,9 @@ void start_server(const char *store_socket_name,
    * requests and reissue requests for transfers of those objects. */
   event_loop_add_timer(g_manager_state->loop, MANAGER_TIMEOUT,
                        fetch_timeout_handler, g_manager_state);
+  /* Publish the heartbeats to all subscribers of the plasma manager table. */
+  event_loop_add_timer(g_manager_state->loop, HEARTBEAT_TIMEOUT_MILLISECONDS,
+                       heartbeat_handler, g_manager_state);
   /* Run the event loop. */
   event_loop_run(g_manager_state->loop);
 }

src/plasma/plasma_manager.h

@@ -154,10 +154,10 @@ void send_queued_request(event_loop *loop,
  * @param context The plasma manager state.
  * @return Void.
  */
-ClientConnection *ClientConnection_init(event_loop *loop,
-                                        int listener_sock,
-                                        void *context,
-                                        int events);
+ClientConnection *ClientConnection_listen(event_loop *loop,
+                                          int listener_sock,
+                                          void *context,
+                                          int events);
 
 /**
  * The following definitions are internal to the plasma manager code but are

src/plasma/test/manager_tests.cc

@@ -77,8 +77,8 @@ plasma_mock *init_plasma_mock(plasma_mock *remote_mock) {
         get_manager_connection(remote_mock->state, manager_addr, mock->port);
     wait_for_pollin(mock->manager_remote_fd);
     mock->read_conn =
-        ClientConnection_init(mock->loop, mock->manager_remote_fd, mock->state,
-                              PLASMA_DEFAULT_RELEASE_DELAY);
+        ClientConnection_listen(mock->loop, mock->manager_remote_fd,
+                                mock->state, PLASMA_DEFAULT_RELEASE_DELAY);
   } else {
     mock->write_conn = NULL;
     mock->read_conn = NULL;
@@ -88,19 +88,14 @@ plasma_mock *init_plasma_mock(plasma_mock *remote_mock) {
   mock->plasma_conn = plasma_connect(plasma_store_socket_name,
                                      utstring_body(manager_socket_name), 0);
   wait_for_pollin(mock->manager_local_fd);
-  mock->client_conn =
-      ClientConnection_init(mock->loop, mock->manager_local_fd, mock->state, 0);
+  mock->client_conn = ClientConnection_listen(
+      mock->loop, mock->manager_local_fd, mock->state, 0);
   utstring_free(manager_socket_name);
   return mock;
 }
 
 void destroy_plasma_mock(plasma_mock *mock) {
-  if (mock->read_conn != NULL) {
-    close(get_client_sock(mock->read_conn));
-    free(mock->read_conn);
-  }
   PlasmaManagerState_free(mock->state);
-  free(mock->client_conn);
   plasma_disconnect(mock->plasma_conn);
   close(mock->local_store);
   close(mock->manager_local_fd);