Log fatal error if plasma manager or local scheduler heartbeats take too long. (#676)

* Log fatal error if plasma manager or local scheduler take too long to send heartbeat.

* Fix linting.

* Use int64_t for milliseconds since unix epoch.

src/common/CMakeLists.txt

@@ -79,7 +79,7 @@ function(define_test test_name library)
   add_executable(${test_name} test/${test_name}.cc ${ARGN})
   add_dependencies(${test_name} hiredis flatbuffers_ep)
   target_link_libraries(${test_name} common ${FLATBUFFERS_STATIC_LIB} ${library})
-  target_compile_options(${test_name} PUBLIC "-DPLASMA_TEST -DLOCAL_SCHEDULER_TEST -DCOMMON_TEST -DRAY_COMMON_LOG_LEVEL=4 -DRAY_TIMEOUT=50")
+  target_compile_options(${test_name} PUBLIC "-DPLASMA_TEST -DLOCAL_SCHEDULER_TEST -DCOMMON_TEST -DRAY_COMMON_LOG_LEVEL=4")
 endfunction()
 
 define_test(common_tests "")

src/common/common.cc

@@ -1,5 +1,6 @@
 #include "common.h"
 
+#include <chrono>
 #include <stdio.h>
 #include <unistd.h>
 #include <sys/types.h>
@@ -65,3 +66,10 @@ char *ObjectID_to_string(ObjectID obj_id, char *id_string, int id_length) {
 
   return id_string;
 }
+
+int64_t current_time_ms() {
+  std::chrono::milliseconds ms_since_epoch =
+      std::chrono::duration_cast<std::chrono::milliseconds>(
+          std::chrono::system_clock::now().time_since_epoch());
+  return ms_since_epoch.count();
+}

src/common/common.h

@@ -232,4 +232,11 @@ bool DBClientID_equal(DBClientID first_id, DBClientID second_id);
 
 extern const unsigned char NIL_DIGEST[DIGEST_SIZE];
 
+/**
+ * Return the current time in milliseconds since the Unix epoch.
+ *
+ * @return The number of milliseconds since the Unix epoch.
+ */
+int64_t current_time_ms();
+
 #endif

src/local_scheduler/local_scheduler.cc

@@ -405,6 +405,9 @@ LocalSchedulerState *LocalSchedulerState_init(
     start_worker(state, NIL_ACTOR_ID);
   }
 
+  /* Initialize the time at which the previous heartbeat was sent. */
+  state->previous_heartbeat_time = current_time_ms();
+
   return state;
 }
 
@@ -1122,6 +1125,17 @@ void handle_actor_creation_callback(ActorID actor_id,
 int heartbeat_handler(event_loop *loop, timer_id id, void *context) {
   LocalSchedulerState *state = (LocalSchedulerState *) context;
   SchedulingAlgorithmState *algorithm_state = state->algorithm_state;
+
+  /* Check that the last heartbeat was not sent too long ago. */
+  int64_t current_time = current_time_ms();
+  CHECK(current_time >= state->previous_heartbeat_time);
+  if (current_time - state->previous_heartbeat_time >
+      NUM_HEARTBEATS_TIMEOUT * HEARTBEAT_TIMEOUT_MILLISECONDS) {
+    LOG_FATAL("The last heartbeat was sent %" PRId64 " milliseconds ago.",
+              current_time - state->previous_heartbeat_time);
+  }
+  state->previous_heartbeat_time = current_time;
+
   LocalSchedulerInfo info;
   /* Ask the scheduling algorithm to fill out the scheduler info struct. */
   provide_scheduler_info(state, algorithm_state, &info);

src/local_scheduler/local_scheduler_shared.h

@@ -78,6 +78,9 @@ struct LocalSchedulerState {
    *  available_gpus.size() == dynamic_resources[ResourceIndex_GPU] should
    *  always be true. */
   std::vector<int> available_gpus;
+  /** The time (in milliseconds since the Unix epoch) when the most recent
+   *  heartbeat was sent. */
+  int64_t previous_heartbeat_time;
 };
 
 /** Contains all information associated with a local scheduler client. */

src/plasma/plasma_manager.cc

@@ -236,6 +236,9 @@ struct PlasmaManagerState {
   ObjectWaitRequests *object_wait_requests_remote;
   /** Initialize an empty hash map for the cache of local available object. */
   AvailableObject *local_available_objects;
+  /** The time (in milliseconds since the Unix epoch) when the most recent
+   *  heartbeat was sent. */
+  int64_t previous_heartbeat_time;
 };
 
 PlasmaManagerState *g_manager_state = NULL;
@@ -553,6 +556,8 @@ PlasmaManagerState *PlasmaManagerState_init(const char *store_socket_name,
   /* Add the callback that processes the notification to the event loop. */
   event_loop_add_file(state->loop, plasma_fd, EVENT_LOOP_READ,
                       process_object_notification, state);
+  /* Initialize the time at which the previous heartbeat was sent. */
+  state->previous_heartbeat_time = current_time_ms();
   return state;
 }
 
@@ -1590,6 +1595,17 @@ void process_message(event_loop *loop,
 
 int heartbeat_handler(event_loop *loop, timer_id id, void *context) {
   PlasmaManagerState *state = (PlasmaManagerState *) context;
+
+  /* Check that the last heartbeat was not sent too long ago. */
+  int64_t current_time = current_time_ms();
+  CHECK(current_time >= state->previous_heartbeat_time);
+  if (current_time - state->previous_heartbeat_time >
+      NUM_HEARTBEATS_TIMEOUT * HEARTBEAT_TIMEOUT_MILLISECONDS) {
+    LOG_FATAL("The last heartbeat was sent %" PRId64 " milliseconds ago.",
+              current_time - state->previous_heartbeat_time);
+  }
+  state->previous_heartbeat_time = current_time;
+
   plasma_manager_send_heartbeat(state->db);
   return HEARTBEAT_TIMEOUT_MILLISECONDS;
 }

src/plasma/plasma_manager.h

@@ -12,11 +12,9 @@
 #endif
 
 /* Timeouts are in milliseconds. */
-#ifndef RAY_TIMEOUT
 #define MANAGER_TIMEOUT 1000
-#else
-#define MANAGER_TIMEOUT RAY_TIMEOUT
-#endif
+
+#define NUM_HEARTBEATS_TIMEOUT 100
 
 /* The buffer size in bytes. Data will get transfered in multiples of this */
 #define BUFSIZE 4096