From 3052ce25a62bb2fa0c54e83c31011c8c87a96482 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Mon, 19 Jun 2017 15:57:51 -0700 Subject: [PATCH] =?UTF-8?q?Divide=20up=20large=20fetch=20requests=20from?= =?UTF-8?q?=20local=20scheduler,=20also=20print=20warni=E2=80=A6=20(#683)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Divide up large fetch requests from local scheduler, also print warning if fetch handler is slow. * Fix linting. * Fix typo. --- .../local_scheduler_algorithm.cc | 25 ++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/src/local_scheduler/local_scheduler_algorithm.cc b/src/local_scheduler/local_scheduler_algorithm.cc index e10755cd0..01b82e780 100644 --- a/src/local_scheduler/local_scheduler_algorithm.cc +++ b/src/local_scheduler/local_scheduler_algorithm.cc @@ -511,6 +511,8 @@ bool can_run(SchedulingAlgorithmState *algorithm_state, TaskSpec *task) { /* TODO(swang): This method is not covered by any valgrind tests. */ int fetch_object_timeout_handler(event_loop *loop, timer_id id, void *context) { + int64_t start_time = current_time_ms(); + LocalSchedulerState *state = (LocalSchedulerState *) context; /* Only try the fetches if we are connected to the object store manager. */ if (!plasma_manager_is_connected(state->plasma_conn)) { @@ -528,11 +530,28 @@ int fetch_object_timeout_handler(event_loop *loop, timer_id id, void *context) { object_ids[i] = entry.second.object_id; i++; } - ARROW_CHECK_OK(state->plasma_conn->Fetch(num_object_ids, object_ids)); - for (int i = 0; i < num_object_ids; ++i) { - reconstruct_object(state, object_ids[i]); + /* Divide very large fetch requests into smaller fetch requests so that a + * single fetch request doesn't block the plasma manager for a long time. */ + int fetch_request_size = 10000; + for (int j = 0; j < num_object_ids; j += fetch_request_size) { + int num_objects_in_request = + std::min(num_object_ids, j + fetch_request_size) - j; + ARROW_CHECK_OK( + state->plasma_conn->Fetch(num_objects_in_request, &object_ids[j])); + } + for (int k = 0; k < num_object_ids; ++k) { + reconstruct_object(state, object_ids[k]); } free(object_ids); + + /* Print a warning if this method took too long. */ + int64_t end_time = current_time_ms(); + int64_t max_time_for_handler = 1000; + if (end_time - start_time > max_time_for_handler) { + LOG_WARN("fetch_object_timeout_handler took %" PRId64 " milliseconds.", + end_time - start_time); + } + return LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS; }