wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-19 11:27:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Change all table calls to use default retry behavior. (#312)

Browse Source 
* Change all table calls to use default retry behavior and change default retry behavior.

* Add warning for table retries.
This commit is contained in:
Robert Nishihara
2017-02-24 12:41:32 -08:00
committed by Stephanie Wang
parent aa174e6311
commit 232601f90d
8 changed files with 48 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/state/table.c
+13 -15
View File
@@ -3,16 +3,10 @@
#include <inttypes.h>
#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;
}
src/common/state/table.h
+6 -2
View File
@@ -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;
src/global_scheduler/global_scheduler.c
+4 -11
View File
@@ -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. */
src/photon/photon.h
-7
View File
@@ -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,
src/photon/photon_algorithm.c
+7 -14
View File
@@ -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);
}
}
}
src/photon/photon_scheduler.c
+10 -18
View File
@@ -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
src/photon/test/photon_tests.c
+3 -6
View File
@@ -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
src/plasma/plasma_manager.c
+5 -31
View File
@@ -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. */