From 232601f90d618cd0aa85323856594effd0239517 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Fri, 24 Feb 2017 12:41:32 -0800 Subject: [PATCH] Change all table calls to use default retry behavior. (#312) * Change all table calls to use default retry behavior and change default retry behavior. * Add warning for table retries. --- src/common/state/table.c | 28 +++++++++---------- src/common/state/table.h | 8 ++++-- src/global_scheduler/global_scheduler.c | 15 +++-------- src/photon/photon.h | 7 ----- src/photon/photon_algorithm.c | 21 +++++---------- src/photon/photon_scheduler.c | 28 +++++++------------ src/photon/test/photon_tests.c | 9 +++---- src/plasma/plasma_manager.c | 36 ++++--------------------- 8 files changed, 48 insertions(+), 104 deletions(-) diff --git a/src/common/state/table.c b/src/common/state/table.c index 7624512ca..ace594bc5 100644 --- a/src/common/state/table.c +++ b/src/common/state/table.c @@ -3,16 +3,10 @@ #include #include "redis.h" -void default_table_failure_callback(object_id id, - void *user_context, - void *user_data) { - CHECKM(0, "default_table_failure_callback was called."); -} - -static const retry_info default_retry = { - .num_retries = 0, - .timeout = 1000, - .fail_callback = default_table_failure_callback}; +/* The default behavior is to retry every ten seconds forever. */ +static const retry_info default_retry = {.num_retries = -1, + .timeout = 10000, + .fail_callback = NULL}; table_callback_data *init_table_callback(db_handle *db_handle, unique_id id, @@ -84,9 +78,10 @@ int64_t table_timeout_handler(event_loop *loop, CHECK(user_context != NULL); table_callback_data *callback_data = (table_callback_data *) user_context; - CHECK(callback_data->retry.num_retries >= 0) - LOG_DEBUG("retrying operation, retry_count = %d", - callback_data->retry.num_retries); + CHECK(callback_data->retry.num_retries >= 0 || + callback_data->retry.num_retries == -1); + LOG_WARN("retrying operation, retry_count = %d", + callback_data->retry.num_retries); if (callback_data->retry.num_retries == 0) { /* We didn't get a response from the database after exhausting all retries; @@ -101,8 +96,11 @@ int64_t table_timeout_handler(event_loop *loop, return EVENT_LOOP_TIMER_DONE; } - /* Decrement retry count and try again. */ - callback_data->retry.num_retries--; + /* Decrement retry count and try again. We use -1 to indicate infinite + * retries. */ + if (callback_data->retry.num_retries != -1) { + callback_data->retry.num_retries--; + } callback_data->retry_callback(callback_data); return callback_data->retry.timeout; } diff --git a/src/common/state/table.h b/src/common/state/table.h index 8aee2d243..6b739b494 100644 --- a/src/common/state/table.h +++ b/src/common/state/table.h @@ -29,10 +29,14 @@ typedef void (*table_fail_callback)(unique_id id, typedef void (*table_retry_callback)(table_callback_data *callback_data); /** - * Data structure consolidating the retry related varaibles. + * Data structure consolidating the retry related variables. If a NULL + * retry_info struct is used, the default behavior will be to retry infinitely + * many times. */ typedef struct { - /** Number of retries left. */ + /** Number of retries. This field will be decremented every time a retry + * occurs (unless the value is -1). If this value is -1, then there will be + * infinitely many retries. */ int num_retries; /** Timeout, in milliseconds. */ uint64_t timeout; diff --git a/src/global_scheduler/global_scheduler.c b/src/global_scheduler/global_scheduler.c index 645a5b3e5..85f02619b 100644 --- a/src/global_scheduler/global_scheduler.c +++ b/src/global_scheduler/global_scheduler.c @@ -39,13 +39,10 @@ void assign_task_to_local_scheduler(global_scheduler_state *state, object_id_to_string(local_scheduler_id, id_string, ID_STRING_SIZE)); task_set_state(task, TASK_STATUS_SCHEDULED); task_set_local_scheduler(task, local_scheduler_id); - retry_info retry = { - .num_retries = 0, .timeout = 100, .fail_callback = NULL, - }; LOG_DEBUG("Issuing a task table update for task = %s", object_id_to_string(task_task_id(task), id_string, ID_STRING_SIZE)); UNUSED(id_string); - task_table_update(state->db, copy_task(task), &retry, NULL, NULL); + task_table_update(state->db, copy_task(task), NULL, NULL, NULL); /* TODO(rkn): We should probably pass around local_scheduler struct pointers * instead of db_client_id objects. */ @@ -336,25 +333,21 @@ int task_cleanup_handler(event_loop *loop, timer_id id, void *context) { void start_server(const char *redis_addr, int redis_port) { event_loop *loop = event_loop_create(); g_state = init_global_scheduler(loop, redis_addr, redis_port); - /* Generic retry information for notification subscriptions. */ - retry_info retry = { - .num_retries = 0, .timeout = 100, .fail_callback = NULL, - }; /* TODO(rkn): subscribe to notifications from the object table. */ /* Subscribe to notifications about new local schedulers. TODO(rkn): this * needs to also get all of the clients that registered with the database * before this call to subscribe. */ db_client_table_subscribe(g_state->db, process_new_db_client, - (void *) g_state, &retry, NULL, NULL); + (void *) g_state, NULL, NULL, NULL); /* Subscribe to notifications about waiting tasks. TODO(rkn): this may need to * get tasks that were submitted to the database before the subscribe. */ task_table_subscribe(g_state->db, NIL_ID, TASK_STATUS_WAITING, - process_task_waiting, (void *) g_state, &retry, NULL, + process_task_waiting, (void *) g_state, NULL, NULL, NULL); object_table_subscribe_to_notifications(g_state->db, true, object_table_subscribe_callback, - g_state, &retry, NULL, NULL); + g_state, NULL, NULL, NULL); /* Subscribe to notifications from local schedulers. These notifications serve * as heartbeats and contain informaion about the load on the local * schedulers. */ diff --git a/src/photon/photon.h b/src/photon/photon.h index 247e3bcc8..793619799 100644 --- a/src/photon/photon.h +++ b/src/photon/photon.h @@ -8,13 +8,6 @@ #include "utarray.h" #include "uthash.h" -/* Retry values for state table operations. For now, only try each command once - * and give it one second to succeed. */ -/* TODO(swang): We should set retry values in a config file somewhere. */ -static const retry_info photon_retry = {.num_retries = 0, - .timeout = 1000, - .fail_callback = NULL}; - enum photon_message_type { /** Notify the local scheduler that a task has finished. */ TASK_DONE = 64, diff --git a/src/photon/photon_algorithm.c b/src/photon/photon_algorithm.c index 1c42ea712..0b2db5e1a 100644 --- a/src/photon/photon_algorithm.c +++ b/src/photon/photon_algorithm.c @@ -365,13 +365,11 @@ void add_task_to_actor_queue(local_scheduler_state *state, if (from_global_scheduler) { /* If the task is from the global scheduler, it's already been added to * the task table, so just update the entry. */ - task_table_update(state->db, task, (retry_info *) &photon_retry, NULL, - NULL); + task_table_update(state->db, task, NULL, NULL, NULL); } else { /* Otherwise, this is the first time the task has been seen in the system * (unless it's a resubmission of a previous task), so add the entry. */ - task_table_add_task(state->db, task, (retry_info *) &photon_retry, NULL, - NULL); + task_table_add_task(state->db, task, NULL, NULL, NULL); } } } @@ -663,13 +661,11 @@ task_queue_entry *queue_task(local_scheduler_state *state, if (from_global_scheduler) { /* If the task is from the global scheduler, it's already been added to * the task table, so just update the entry. */ - task_table_update(state->db, task, (retry_info *) &photon_retry, NULL, - NULL); + task_table_update(state->db, task, NULL, NULL, NULL); } else { /* Otherwise, this is the first time the task has been seen in the system * (unless it's a resubmission of a previous task), so add the entry. */ - task_table_add_task(state->db, task, (retry_info *) &photon_retry, NULL, - NULL); + task_table_add_task(state->db, task, NULL, NULL, NULL); } } @@ -767,8 +763,7 @@ void give_task_to_local_scheduler(local_scheduler_state *state, /* Assign the task to the relevant local scheduler. */ DCHECK(state->config.global_scheduler_exists); task *task = alloc_task(spec, TASK_STATUS_SCHEDULED, local_scheduler_id); - task_table_add_task(state->db, task, (retry_info *) &photon_retry, NULL, - NULL); + task_table_add_task(state->db, task, NULL, NULL, NULL); } /** @@ -791,8 +786,7 @@ void give_task_to_global_scheduler(local_scheduler_state *state, DCHECK(state->config.global_scheduler_exists); task *task = alloc_task(spec, TASK_STATUS_WAITING, NIL_ID); DCHECK(state->db != NULL); - task_table_add_task(state->db, task, (retry_info *) &photon_retry, NULL, - NULL); + task_table_add_task(state->db, task, NULL, NULL, NULL); } bool resource_constraints_satisfied(local_scheduler_state *state, @@ -822,8 +816,7 @@ void update_result_table(local_scheduler_state *state, task_spec *spec) { task_id task_id = task_spec_id(spec); for (int64_t i = 0; i < task_num_returns(spec); ++i) { object_id return_id = task_return(spec, i); - result_table_add(state->db, return_id, task_id, - (retry_info *) &photon_retry, NULL, NULL); + result_table_add(state->db, return_id, task_id, NULL, NULL, NULL); } } } diff --git a/src/photon/photon_scheduler.c b/src/photon/photon_scheduler.c index eefcd248e..9c4d8bcba 100644 --- a/src/photon/photon_scheduler.c +++ b/src/photon/photon_scheduler.c @@ -423,8 +423,7 @@ void assign_task_to_worker(local_scheduler_state *state, worker->task_in_progress = copy_task(task); /* Update the global task table. */ if (state->db != NULL) { - task_table_update(state->db, task, (retry_info *) &photon_retry, NULL, - NULL); + task_table_update(state->db, task, NULL, NULL, NULL); } else { free_task(task); } @@ -495,9 +494,9 @@ void reconstruct_result_lookup_callback(object_id reconstruct_object_id, * already being taken care of. NOTE: This codepath is not responsible for * detecting failure of the other reconstruction, or updating the * scheduling_state accordingly. */ - task_table_test_and_update( - state->db, task_id, TASK_STATUS_DONE, TASK_STATUS_RECONSTRUCTING, - (retry_info *) &photon_retry, reconstruct_task_update_callback, state); + task_table_test_and_update(state->db, task_id, TASK_STATUS_DONE, + TASK_STATUS_RECONSTRUCTING, NULL, + reconstruct_task_update_callback, state); } void reconstruct_object_lookup_callback(object_id reconstruct_object_id, @@ -511,8 +510,7 @@ void reconstruct_object_lookup_callback(object_id reconstruct_object_id, local_scheduler_state *state = user_context; if (manager_count == 0) { /* Look up the task that created the object in the result table. */ - result_table_lookup(state->db, reconstruct_object_id, - (retry_info *) &photon_retry, + result_table_lookup(state->db, reconstruct_object_id, NULL, reconstruct_result_lookup_callback, (void *) state); } } @@ -524,8 +522,7 @@ void reconstruct_object(local_scheduler_state *state, CHECK(state->db != NULL); /* Determine if reconstruction is necessary by checking if the object exists * on a node. */ - object_table_lookup(state->db, reconstruct_object_id, - (retry_info *) &photon_retry, + object_table_lookup(state->db, reconstruct_object_id, NULL, reconstruct_object_lookup_callback, (void *) state); } @@ -631,8 +628,8 @@ void process_message(event_loop *loop, if (state->db != NULL) { /* Update control state tables. */ task_set_state(worker->task_in_progress, TASK_STATUS_DONE); - task_table_update(state->db, worker->task_in_progress, - (retry_info *) &photon_retry, NULL, NULL); + task_table_update(state->db, worker->task_in_progress, NULL, NULL, + NULL); /* The call to task_table_update takes ownership of the * task_in_progress, so we set the pointer to NULL so it is not used. */ } else { @@ -818,20 +815,15 @@ void start_server(const char *node_ip_address, * local scheduler by the global scheduler or by other local schedulers. * TODO(rkn): we also need to get any tasks that were assigned to this local * scheduler before the call to subscribe. */ - retry_info retry; - memset(&retry, 0, sizeof(retry)); - retry.num_retries = 0; - retry.timeout = 100; - retry.fail_callback = NULL; if (g_state->db != NULL) { task_table_subscribe(g_state->db, get_db_client_id(g_state->db), TASK_STATUS_SCHEDULED, handle_task_scheduled_callback, - NULL, &retry, NULL, NULL); + NULL, NULL, NULL, NULL); } /* Subscribe to notifications about newly created actors. */ if (g_state->db != NULL) { actor_notification_table_subscribe( - g_state->db, handle_actor_creation_callback, g_state, &retry); + g_state->db, handle_actor_creation_callback, g_state, NULL); } /* Create a timer for publishing information about the load on the local * scheduler to the local scheduler table. This message also serves as a diff --git a/src/photon/test/photon_tests.c b/src/photon/test/photon_tests.c index 3699c63c4..a90af4ad3 100644 --- a/src/photon/test/photon_tests.c +++ b/src/photon/test/photon_tests.c @@ -183,8 +183,7 @@ TEST object_reconstruction_test(void) { * that would suppress object reconstruction. */ task *task = alloc_task(spec, TASK_STATUS_DONE, get_db_client_id(photon->photon_state->db)); - task_table_add_task(photon->photon_state->db, task, - (retry_info *) &photon_retry, NULL, NULL); + task_table_add_task(photon->photon_state->db, task, NULL, NULL, NULL); /* Trigger reconstruction, and run the event loop again. */ object_id return_id = task_return(spec, 0); photon_reconstruct_object(worker, return_id); @@ -282,8 +281,7 @@ TEST object_reconstruction_recursive_test(void) { * condition that would suppress object reconstruction. */ task *last_task = alloc_task(specs[NUM_TASKS - 1], TASK_STATUS_DONE, get_db_client_id(photon->photon_state->db)); - task_table_add_task(photon->photon_state->db, last_task, - (retry_info *) &photon_retry, NULL, NULL); + task_table_add_task(photon->photon_state->db, last_task, NULL, NULL, NULL); /* Trigger reconstruction for the last object, and run the event loop * again. */ object_id return_id = task_return(specs[NUM_TASKS - 1], 0); @@ -346,8 +344,7 @@ TEST object_reconstruction_suppression_test(void) { 2, db_connect_args); db_attach(db, photon->loop, false); /* Add the object to the object table. */ - object_table_add(db, return_id, 1, (unsigned char *) NIL_DIGEST, - (retry_info *) &photon_retry, + object_table_add(db, return_id, 1, (unsigned char *) NIL_DIGEST, NULL, object_reconstruction_suppression_callback, (void *) worker); /* Run the event loop. NOTE: OSX appears to require the parent process to diff --git a/src/plasma/plasma_manager.c b/src/plasma/plasma_manager.c index ef3deeae5..0cc5cb556 100644 --- a/src/plasma/plasma_manager.c +++ b/src/plasma/plasma_manager.c @@ -1053,14 +1053,9 @@ void process_fetch_requests(client_connection *client_conn, * available. The notifications will call the callback that was passed to * object_table_subscribe_to_notifications, which will initiate a transfer * of the object to this plasma manager. */ - retry_info retry; - memset(&retry, 0, sizeof(retry)); - retry.num_retries = 0; - retry.timeout = MANAGER_TIMEOUT; - retry.fail_callback = fatal_table_callback; object_table_request_notifications(manager_state->db, num_object_ids_to_request, - object_ids_to_request, &retry); + object_ids_to_request, NULL); } free(object_ids_to_request); } @@ -1140,14 +1135,9 @@ void process_wait_request(client_connection *client_conn, * become available. The notifications will call the callback that was * passed to object_table_subscribe_to_notifications, which will update * the wait request. */ - retry_info retry; - memset(&retry, 0, sizeof(retry)); - retry.num_retries = 0; - retry.timeout = MANAGER_TIMEOUT; - retry.fail_callback = fatal_table_callback; object_table_request_notifications(manager_state->db, num_object_ids_to_request, - object_ids_to_request, &retry); + object_ids_to_request, NULL); } /* Set a timer that will cause the wait request to return to the client. */ @@ -1232,13 +1222,7 @@ void process_status_request(client_connection *client_conn, } /* The object is not local, so check whether it is stored remotely. */ - retry_info retry = { - .num_retries = NUM_RETRIES, - .timeout = MANAGER_TIMEOUT, - .fail_callback = object_table_lookup_fail_callback, - }; - - object_table_lookup(client_conn->manager_state->db, object_id, &retry, + object_table_lookup(client_conn->manager_state->db, object_id, NULL, request_status_done, client_conn); } @@ -1254,12 +1238,7 @@ void process_delete_object_notification(plasma_manager_state *state, /* Remove this object from the (redis) object table. */ if (state->db) { - retry_info retry = { - .num_retries = NUM_RETRIES, - .timeout = MANAGER_TIMEOUT, - .fail_callback = fatal_table_callback, - }; - object_table_remove(state->db, obj_id, NULL, &retry, NULL, NULL); + object_table_remove(state->db, obj_id, NULL, NULL, NULL, NULL); } /* NOTE: There could be pending wait requests for this object that will now @@ -1281,14 +1260,9 @@ void process_add_object_notification(plasma_manager_state *state, if (state->db) { /* TODO(swang): Log the error if we fail to add the object, and possibly * retry later? */ - retry_info retry = { - .num_retries = NUM_RETRIES, - .timeout = MANAGER_TIMEOUT, - .fail_callback = fatal_table_callback, - }; object_table_add(state->db, obj_id, object_info.data_size + object_info.metadata_size, - object_info.digest, &retry, NULL, NULL); + object_info.digest, NULL, NULL, NULL); } /* If we were trying to fetch this object, finish up the fetch request. */