From 96962cdee0c6803695505dfb3c3cdcc451ce7a5f Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Fri, 16 Jun 2017 12:11:01 -0700 Subject: [PATCH] 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 | 2 +- src/common/common.cc | 8 ++++++++ src/common/common.h | 7 +++++++ src/local_scheduler/local_scheduler.cc | 14 ++++++++++++++ src/local_scheduler/local_scheduler_shared.h | 3 +++ src/plasma/plasma_manager.cc | 16 ++++++++++++++++ src/plasma/plasma_manager.h | 6 ++---- 7 files changed, 51 insertions(+), 5 deletions(-) diff --git a/src/common/CMakeLists.txt b/src/common/CMakeLists.txt index 303ecf4ab..27cb1a969 100644 --- a/src/common/CMakeLists.txt +++ b/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 "") diff --git a/src/common/common.cc b/src/common/common.cc index 91b36797c..e032ae3ec 100644 --- a/src/common/common.cc +++ b/src/common/common.cc @@ -1,5 +1,6 @@ #include "common.h" +#include #include #include #include @@ -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::system_clock::now().time_since_epoch()); + return ms_since_epoch.count(); +} diff --git a/src/common/common.h b/src/common/common.h index 9560123d7..6c09913ac 100644 --- a/src/common/common.h +++ b/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 diff --git a/src/local_scheduler/local_scheduler.cc b/src/local_scheduler/local_scheduler.cc index f6500d3e0..1454f5e15 100644 --- a/src/local_scheduler/local_scheduler.cc +++ b/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); diff --git a/src/local_scheduler/local_scheduler_shared.h b/src/local_scheduler/local_scheduler_shared.h index 94627b7ab..8b35c5ec4 100644 --- a/src/local_scheduler/local_scheduler_shared.h +++ b/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 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. */ diff --git a/src/plasma/plasma_manager.cc b/src/plasma/plasma_manager.cc index e9308a295..b54cb8fdc 100644 --- a/src/plasma/plasma_manager.cc +++ b/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; } diff --git a/src/plasma/plasma_manager.h b/src/plasma/plasma_manager.h index f28224570..e9247a4f1 100644 --- a/src/plasma/plasma_manager.h +++ b/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