diff --git a/.travis.yml b/.travis.yml index 0ca105468..3b9c179c9 100644 --- a/.travis.yml +++ b/.travis.yml @@ -37,6 +37,10 @@ matrix: - sudo apt-get update -qq - sudo apt-get install -qq valgrind script: + - cd src/common + - make valgrind + - cd ../.. + - cd src/plasma - make valgrind - cd ../.. diff --git a/src/common/Makefile b/src/common/Makefile index 6982b7945..689c4c901 100644 --- a/src/common/Makefile +++ b/src/common/Makefile @@ -1,10 +1,10 @@ CC = gcc -CFLAGS = -g -Wall --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -fPIC -I. -Ithirdparty -Ithirdparty/ae -Wno-typedef-redefinition -Werror +CFLAGS = -g -Wall -Wno-typedef-redefinition --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -fPIC -I. -Ithirdparty -Ithirdparty/ae BUILD = build all: hiredis $(BUILD)/libcommon.a -$(BUILD)/libcommon.a: event_loop.o common.o task.o io.o state/redis.o thirdparty/ae/ae.o +$(BUILD)/libcommon.a: event_loop.o common.o task.o io.o state/redis.o state/table.o state/object_table.o state/task_log.o thirdparty/ae/ae.o ar rcs $@ $^ $(BUILD)/common_tests: test/common_tests.c $(BUILD)/libcommon.a @@ -13,6 +13,12 @@ $(BUILD)/common_tests: test/common_tests.c $(BUILD)/libcommon.a $(BUILD)/db_tests: hiredis test/db_tests.c $(BUILD)/libcommon.a $(CC) -o $@ test/db_tests.c $(BUILD)/libcommon.a thirdparty/hiredis/libhiredis.a $(CFLAGS) +$(BUILD)/object_table_tests: hiredis test/object_table_tests.c $(BUILD)/libcommon.a + $(CC) -o $@ test/object_table_tests.c $(BUILD)/libcommon.a thirdparty/hiredis/libhiredis.a $(CFLAGS) + +$(BUILD)/task_log_tests: hiredis test/task_log_tests.c $(BUILD)/libcommon.a + $(CC) -o $@ test/task_log_tests.c $(BUILD)/libcommon.a thirdparty/hiredis/libhiredis.a $(CFLAGS) + $(BUILD)/io_tests: test/io_tests.c $(BUILD)/libcommon.a $(CC) -o $@ $^ $(CFLAGS) @@ -32,9 +38,9 @@ redis: hiredis: git submodule update --init --recursive -- "thirdparty/hiredis" ; cd thirdparty/hiredis ; make -test: hiredis redis $(BUILD)/common_tests $(BUILD)/db_tests $(BUILD)/io_tests $(BUILD)/task_tests $(BUILD)/redis_tests FORCE +test: hiredis redis $(BUILD)/common_tests $(BUILD)/task_log_tests $(BUILD)/object_table_tests $(BUILD)/db_tests $(BUILD)/io_tests $(BUILD)/task_tests $(BUILD)/redis_tests FORCE ./thirdparty/redis-3.2.3/src/redis-server & - sleep 1s ; ./build/common_tests ; ./build/db_tests ; ./build/io_tests ; ./build/task_tests ; ./build/redis_tests + sleep 1s ; ./build/common_tests ; ./build/db_tests ; ./build/task_log_tests ; ./build/object_table_tests ; ./build/io_tests ; ./build/task_tests ; ./build/redis_tests valgrind: test valgrind --leak-check=full --error-exitcode=1 ./build/common_tests @@ -42,5 +48,7 @@ valgrind: test valgrind --leak-check=full --error-exitcode=1 ./build/io_tests valgrind --leak-check=full --error-exitcode=1 ./build/task_tests valgrind --leak-check=full --error-exitcode=1 ./build/redis_tests + valgrind --leak-check=full --error-exitcode=1 ./build/task_log_tests + valgrind --leak-check=full --error-exitcode=1 ./build/object_table_tests FORCE: diff --git a/src/common/common.h b/src/common/common.h index f09e91cd5..1a05f8924 100644 --- a/src/common/common.h +++ b/src/common/common.h @@ -5,6 +5,7 @@ #include #include #include +#include #ifndef RAY_COMMON_DEBUG #define LOG_DEBUG(M, ...) @@ -36,6 +37,10 @@ } \ } while (0); +/** This macro indicates that this pointer owns the data it is pointing to + * and is responsible for freeing it. */ +#define OWNER + #define UNIQUE_ID_SIZE 20 /* Cleanup method for running tests with the greatest library. diff --git a/src/common/event_loop.c b/src/common/event_loop.c index 5c56b44de..fa5b2030f 100644 --- a/src/common/event_loop.c +++ b/src/common/event_loop.c @@ -42,15 +42,14 @@ void event_loop_remove_file(event_loop *loop, int fd) { } int64_t event_loop_add_timer(event_loop *loop, - int64_t milliseconds, + int64_t timeout, event_loop_timer_handler handler, void *context) { - return aeCreateTimeEvent(loop, milliseconds, handler, context, NULL); + return aeCreateTimeEvent(loop, timeout, handler, context, NULL); } -void event_loop_remove_timer(event_loop *loop, timer_id timer_id) { - int err = aeDeleteTimeEvent(loop, timer_id); - CHECK(err == AE_OK); /* timer id found? */ +int event_loop_remove_timer(event_loop *loop, int64_t id) { + return aeDeleteTimeEvent(loop, id); } void event_loop_run(event_loop *loop) { diff --git a/src/common/event_loop.h b/src/common/event_loop.h index b0b584e6f..defb06465 100644 --- a/src/common/event_loop.h +++ b/src/common/event_loop.h @@ -4,6 +4,9 @@ #include #include "ae/ae.h" +/* Unique timer ID that will be generated when the timer is added to the + * event loop. Will not be reused later on in another call + * to event_loop_add_timer. */ typedef long long timer_id; typedef aeEventLoop event_loop; @@ -57,16 +60,30 @@ void event_loop_add_file(event_loop *loop, /* Remove a registered file event handler from the event loop. */ void event_loop_remove_file(event_loop *loop, int fd); -/* Register a handler that will be called after a time slice of - * "milliseconds" milliseconds. Can specify a context that will be passed - * as an argument to the handler. Return the id of the time event. */ +/** Register a handler that will be called after a time slice of + * "timeout" milliseconds. + * + * @param loop The event loop. + * @param timeout The timeout in milliseconds. + * @param handler The handler for the timeout. + * @param context User context that can be passed in and will be passed in + * as an argument for the timer handler. + * @return The ID of the timer. + */ int64_t event_loop_add_timer(event_loop *loop, - int64_t milliseconds, + int64_t timeout, event_loop_timer_handler handler, void *context); -/* Remove a registered time event handler from the event loop. */ -void event_loop_remove_timer(event_loop *loop, timer_id timer_id); +/** + * Remove a registered time event handler from the event loop. Can be called + * multiple times on the same timer. + * + * @param loop The event loop. + * @param timer_id The ID of the timer to be removed. + * @return Returns 0 if the removal was successful. + */ +int event_loop_remove_timer(event_loop *loop, int64_t timer_id); /* Run the event loop. */ void event_loop_run(event_loop *loop); diff --git a/src/common/state/db.h b/src/common/state/db.h index 50536e271..96ecf1e79 100644 --- a/src/common/state/db.h +++ b/src/common/state/db.h @@ -3,7 +3,7 @@ #include "event_loop.h" -typedef struct db_handle_impl db_handle; +typedef struct db_handle db_handle; /* Connect to the global system store at address and port. Returns * a handle to the database, which must be freed with db_disconnect diff --git a/src/common/state/object_table.c b/src/common/state/object_table.c new file mode 100644 index 000000000..58dde9abb --- /dev/null +++ b/src/common/state/object_table.c @@ -0,0 +1,38 @@ +#include "object_table.h" +#include "redis.h" + +void object_table_lookup(db_handle *db_handle, + object_id object_id, + retry_info *retry, + object_table_lookup_done_callback done_callback, + void *user_context) { + init_table_callback(db_handle, object_id, NULL, retry, done_callback, + redis_object_table_lookup, user_context); +} + +void object_table_add(db_handle *db_handle, + object_id object_id, + retry_info *retry, + object_table_done_callback done_callback, + void *user_context) { + init_table_callback(db_handle, object_id, NULL, retry, done_callback, + redis_object_table_add, user_context); +} + +void object_table_subscribe( + db_handle *db_handle, + object_id object_id, + object_table_object_available_callback object_available_callback, + void *subscribe_context, + retry_info *retry, + object_table_done_callback done_callback, + void *user_context) { + object_table_subscribe_data *sub_data = + malloc(sizeof(object_table_subscribe_data)); + utarray_push_back(db_handle->callback_freelist, &sub_data); + sub_data->object_available_callback = object_available_callback; + sub_data->subscribe_context = subscribe_context; + + init_table_callback(db_handle, object_id, sub_data, retry, done_callback, + redis_object_table_subscribe, user_context); +} diff --git a/src/common/state/object_table.h b/src/common/state/object_table.h index bab54bc7b..112d174e2 100644 --- a/src/common/state/object_table.h +++ b/src/common/state/object_table.h @@ -1,25 +1,126 @@ +#ifndef OBJECT_TABLE_H +#define OBJECT_TABLE_H + #include "common.h" +#include "table.h" #include "db.h" -/* The callback that is called when the result of a lookup - * in the object table comes back. The callback should free - * the manager_vector array, but NOT the strings they are pointing to. */ -typedef void (*lookup_callback)(object_id object_id, - int manager_count, - const char *manager_vector[], - void *context); +/* + * ==== Lookup call and callback ==== + */ -/* Register a new object with the directory. */ -/* TODO(pcm): Retry, print for each attempt. */ -void object_table_add(db_handle *db, object_id object_id); +/* Callback called when the lookup completes. The callback should free + * the manager_vector array, but NOT the strings they are pointing to. + */ +typedef void (*object_table_lookup_done_callback)( + object_id object_id, + int manager_count, + OWNER const char *manager_vector[], + void *user_context); -/* Remove object from the directory. */ -void object_table_remove(db_handle *db, +/** + * Return the list of nodes storing object_id in their plasma stores. + * + * @param db_handle Handle to object_table database. + * @param object_id ID of the object being looked up. + * @param retry Information about retrying the request to the database. + * @param done_callback Function to be called when database returns result. + * @param user_context Context passed by the caller. + * @return Void. + */ +void object_table_lookup(db_handle *db_handle, object_id object_id, - const char *manager); + retry_info *retry, + object_table_lookup_done_callback done_callback, + void *user_context); -/* Look up entry from the directory */ -void object_table_lookup(db_handle *db, +/* + * ==== Add object call and callback ==== + */ + +/* Callback called when the object add/remove operation completes. */ +typedef void (*object_table_done_callback)(object_id object_id, + void *user_context); + +/** + * Add the plasma manager that created the db_handle to the + * list of plasma managers that have the object_id. + * + * @param db_handle Handle to db. + * @param object_id Object unique identifier. + * @param retry Information about retrying the request to the database. + * @param done_callback Callback to be called when lookup completes. + * @param user_context User context to be passed in the callbacks. + * @return Void. + */ +void object_table_add(db_handle *db_handle, + object_id object_id, + retry_info *retry, + object_table_done_callback done_callback, + void *user_context); + +/* + * ==== Remove object call and callback ==== + */ + +/** + * Object remove function. + * + * @param db_handle Handle to db. + * @param object_id Object unique identifier. + * @param retry Information about retrying the request to the database. + * @param done_callback Callback to be called when lookup completes. + * @param user_context User context to be passed in the callbacks. + * @return Void. + */ +/* +void object_table_remove(db_handle *db, object_id object_id, lookup_callback callback, void *context); + retry_info *retry, + object_table_done_callback done_callback, + void *user_context); +*/ + +/* + * ==== Subscribe to be announced when new object available ==== + */ + +/* Callback called when object object_id is available. */ +typedef void (*object_table_object_available_callback)(object_id object_id, + void *user_context); + +/** + * Subcribing to new object available function. + * + * @param db_handle Handle to db. + * @param object_id Object unique identifier. + * @param object_available_callback callback to be called when new object + * becomes + * available. + * @param subscribe_context caller context which will be passed back in the + * object_available_callback. + * @param retry Information about retrying the request to the database. + * @param done_callback Callback to be called when subscription is installed. + * @param user_context User context to be passed in the callbacks. + * @return Void. + */ + +void object_table_subscribe( + db_handle *db, + object_id object_id, + object_table_object_available_callback object_available_callback, + void *subscribe_context, + retry_info *retry, + object_table_done_callback done_callback, + void *user_context); + +/* Data that is needed to register new object available callbacks with the state + * database. */ +typedef struct { + object_table_object_available_callback object_available_callback; + void *subscribe_context; +} object_table_subscribe_data; + +#endif /* OBJECT_TABLE_H */ diff --git a/src/common/state/redis.c b/src/common/state/redis.c index d53b0bf15..27c0d1327 100644 --- a/src/common/state/redis.c +++ b/src/common/state/redis.c @@ -1,6 +1,7 @@ /* Redis implementation of the global state store */ #include +#include #include #include "hiredis/adapters/ae.h" @@ -9,6 +10,7 @@ #include "common.h" #include "db.h" #include "object_table.h" +#include "task.h" #include "task_log.h" #include "event_loop.h" #include "redis.h" @@ -31,6 +33,18 @@ } \ } while (0); +#define REDIS_CALLBACK_HEADER(DB, CB_DATA, REPLY) \ + if ((REPLY) == NULL) { \ + return; \ + } \ + db_handle *DB = c->data; \ + table_callback_data *CB_DATA = \ + outstanding_callbacks_find((int64_t) privdata); \ + if (CB_DATA == NULL) \ + /* the callback data structure has been \ + * already freed; just ignore this reply */ \ + return; + db_handle *db_connect(const char *address, int port, const char *client_type, @@ -104,131 +118,291 @@ void db_disconnect(db_handle *db) { } void db_attach(db_handle *db, event_loop *loop) { + db->loop = loop; redisAeAttach(loop, db->context); redisAeAttach(loop, db->sub_context); } -void object_table_add(db_handle *db, unique_id object_id) { - redisAsyncCommand(db->context, NULL, NULL, "SADD obj:%b %d", &object_id.id[0], - UNIQUE_ID_SIZE, db->client_id); +/* + * ==== object_table callbacks ==== + */ + +void redis_object_table_add_callback(redisAsyncContext *c, + void *r, + void *privdata) { + REDIS_CALLBACK_HEADER(db, callback_data, r) + + if (callback_data->done_callback) { + task_log_done_callback done_callback = callback_data->done_callback; + done_callback(callback_data->id, callback_data->user_context); + } + destroy_timer_callback(db->loop, callback_data); +} + +void redis_object_table_add(table_callback_data *callback_data) { + CHECK(callback_data); + db_handle *db = callback_data->db_handle; + redisAsyncCommand(db->context, redis_object_table_add_callback, + (void *) callback_data->timer_id, "SADD obj:%b %d", + &callback_data->id.id[0], UNIQUE_ID_SIZE, db->client_id); if (db->context->err) { LOG_REDIS_ERR(db->context, "could not add object_table entry"); } } -void object_table_get_entry(redisAsyncContext *c, void *r, void *privdata) { - db_handle *db = c->data; - lookup_callback_data *cb_data = privdata; - redisReply *reply = r; - if (reply == NULL) - return; - int *result = malloc(reply->elements * sizeof(int)); - int64_t manager_count = reply->elements; - if (reply->type == REDIS_REPLY_ARRAY) { - for (int j = 0; j < reply->elements; j++) { - CHECK(reply->element[j]->type == REDIS_REPLY_STRING); - result[j] = atoi(reply->element[j]->str); - service_cache_entry *entry; - HASH_FIND_INT(db->service_cache, &result[j], entry); - if (!entry) { - redisReply *reply = redisCommand(db->sync_context, "HGET %s %lld", - db->client_type, result[j]); - CHECK(reply->type == REDIS_REPLY_STRING); - entry = malloc(sizeof(service_cache_entry)); - entry->service_id = result[j]; - entry->addr = strdup(reply->str); - HASH_ADD_INT(db->service_cache, service_id, entry); - freeReplyObject(reply); - } - } - } else { - LOG_ERR("expected integer or string, received type %d", reply->type); - exit(-1); - } - const char **manager_vector = malloc(manager_count * sizeof(char *)); - for (int j = 0; j < manager_count; ++j) { - service_cache_entry *entry; - HASH_FIND_INT(db->service_cache, &result[j], entry); - manager_vector[j] = entry->addr; - } - cb_data->callback(cb_data->object_id, manager_count, manager_vector, - cb_data->context); - free(privdata); - free(result); -} +void redis_object_table_lookup(table_callback_data *callback_data) { + CHECK(callback_data); + db_handle *db = callback_data->db_handle; -void object_table_lookup(db_handle *db, - object_id object_id, - lookup_callback callback, - void *context) { - lookup_callback_data *cb_data = malloc(sizeof(lookup_callback_data)); - cb_data->callback = callback; - cb_data->object_id = object_id; - cb_data->context = context; - redisAsyncCommand(db->context, object_table_get_entry, cb_data, - "SMEMBERS obj:%b", &object_id.id[0], UNIQUE_ID_SIZE); + /* Call redis asynchronously */ + redisAsyncCommand(db->context, redis_object_table_get_entry, + (void *) callback_data->timer_id, "SMEMBERS obj:%b", + &callback_data->id.id[0], UNIQUE_ID_SIZE); if (db->context->err) { LOG_REDIS_ERR(db->context, "error in object_table lookup"); } } -void task_log_add_task(db_handle *db, task_instance *task_instance) { - task_iid task_iid = *task_instance_id(task_instance); - redisAsyncCommand(db->context, NULL, NULL, "HMSET tasklog:%b 0 %b", - (char *) &task_iid.id[0], UNIQUE_ID_SIZE, - (char *) task_instance, task_instance_size(task_instance)); - if (db->context->err) { - LOG_REDIS_ERR(db->context, "error setting task in task_log_add_task"); +/** + * Get an entry from the plasma manager table in redis. + * + * @param db The database handle. + * @param index The index of the plasma manager. + * @param *manager The pointer where the IP address of the manager gets written. + * @return Void. + */ +void redis_get_cached_service(db_handle *db, int index, const char **manager) { + service_cache_entry *entry; + HASH_FIND_INT(db->service_cache, &index, entry); + if (!entry) { + /* This is a very rare case. */ + redisReply *reply = + redisCommand(db->sync_context, "HGET %s %lld", db->client_type, index); + CHECK(reply->type == REDIS_REPLY_STRING); + entry = malloc(sizeof(service_cache_entry)); + entry->service_id = index; + entry->addr = strdup(reply->str); + HASH_ADD_INT(db->service_cache, service_id, entry); + freeReplyObject(reply); } - node_id node = *task_instance_node(task_instance); - int32_t state = *task_instance_state(task_instance); - redisAsyncCommand(db->context, NULL, NULL, "PUBLISH task_log:%b:%d %b", - (char *) &node.id[0], UNIQUE_ID_SIZE, state, - (char *) task_instance, task_instance_size(task_instance)); - if (db->context->err) { - LOG_REDIS_ERR(db->context, "error publishing task in task_log_add_task"); + *manager = entry->addr; +} + +void redis_object_table_get_entry(redisAsyncContext *c, + void *r, + void *privdata) { + REDIS_CALLBACK_HEADER(db, callback_data, r) + redisReply *reply = r; + + int *managers = malloc(reply->elements * sizeof(int)); + int64_t manager_count = reply->elements; + + if (reply->type == REDIS_REPLY_ARRAY) { + const char **manager_vector = malloc(manager_count * sizeof(char *)); + for (int j = 0; j < reply->elements; j++) { + CHECK(reply->element[j]->type == REDIS_REPLY_STRING); + managers[j] = atoi(reply->element[j]->str); + redis_get_cached_service(db, managers[j], manager_vector + j); + } + + object_table_lookup_done_callback done_callback = + callback_data->done_callback; + done_callback(callback_data->id, manager_count, manager_vector, + callback_data->user_context); + /* remove timer */ + destroy_timer_callback(callback_data->db_handle->loop, callback_data); + free(managers); + } else { + LOG_ERR("expected integer or string, received type %d", reply->type); + exit(-1); } } -void task_log_redis_callback(redisAsyncContext *c, - void *reply, - void *privdata) { - redisReply *r = reply; - if (reply == NULL) - return; - CHECK(r->type == REDIS_REPLY_ARRAY); +void object_table_redis_callback(redisAsyncContext *c, + void *r, + void *privdata) { + REDIS_CALLBACK_HEADER(db, callback_data, r) + redisReply *reply = r; + + CHECK(reply->type == REDIS_REPLY_ARRAY); /* First entry is message type, second is topic, third is payload. */ - CHECK(r->elements > 2); + CHECK(reply->elements > 2); /* If this condition is true, we got the initial message that acknowledged the * subscription. */ - if (r->element[2]->str == NULL) { + if (strncmp(reply->element[1]->str, "add", 3) != 0) { + if (callback_data->done_callback) { + object_table_done_callback done_callback = callback_data->done_callback; + done_callback(callback_data->id, callback_data->user_context); + } + event_loop_remove_timer(db->loop, callback_data->timer_id); return; } /* Otherwise, parse the task and call the callback. */ - CHECK(privdata); - task_log_callback_data *callback_data = privdata; - task_instance *instance = malloc(r->element[2]->len); - memcpy(instance, r->element[2]->str, r->element[2]->len); - callback_data->callback(instance, callback_data->userdata); + object_table_subscribe_data *data = callback_data->data; + + if (data->object_available_callback) { + data->object_available_callback(callback_data->id, data->subscribe_context); + } +} + +void redis_object_table_subscribe(table_callback_data *callback_data) { + db_handle *db = callback_data->db_handle; + + /* subscribe to key notification associated to object id */ + + redisAsyncCommand(db->sub_context, object_table_redis_callback, + (void *) callback_data->timer_id, + "SUBSCRIBE __keyspace@0__:%b add", + (char *) &callback_data->id.id[0], UNIQUE_ID_SIZE); + + if (db->sub_context->err) { + LOG_REDIS_ERR(db->sub_context, + "error in redis_object_table_subscribe_callback"); + } +} + +/* + * ==== task_log callbacks ==== + */ + +void redis_task_log_publish(table_callback_data *callback_data) { + db_handle *db = callback_data->db_handle; + task_instance *task_instance = callback_data->data; + task_iid task_iid = *task_instance_id(task_instance); + node_id node = *task_instance_node(task_instance); + int32_t state = *task_instance_state(task_instance); + + LOG_DEBUG("Called log_publish callback"); + +/* Check whether the vector (requests_info) indicating the status of the + * requests has been allocated. + * If was not allocate it, allocate it and initialize it. + * This vector has an entry for each redis command, and it stores true if a + * reply for that command + * has been received, and false otherwise. + * The first entry in the callback corresponds to RPUSH, and the second entry to + * PUBLISH. + */ +#define NUM_DB_REQUESTS 2 +#define PUSH_INDEX 0 +#define PUBLISH_INDEX 1 + if (callback_data->requests_info == NULL) { + callback_data->requests_info = malloc(NUM_DB_REQUESTS * sizeof(bool)); + for (int i = 0; i < NUM_DB_REQUESTS; i++) { + ((bool *) callback_data->requests_info)[i] = false; + } + } + + if (((bool *) callback_data->requests_info)[PUSH_INDEX] == false) { + if (*task_instance_state(task_instance) == TASK_STATUS_WAITING) { + redisAsyncCommand(db->context, redis_task_log_publish_push_callback, + (void *) callback_data->timer_id, "RPUSH tasklog:%b %b", + (char *) &task_iid.id[0], UNIQUE_ID_SIZE, + (char *) task_instance, + task_instance_size(task_instance)); + } else { + task_update update = {.state = state, .node = node}; + redisAsyncCommand(db->context, redis_task_log_publish_push_callback, + (void *) callback_data->timer_id, "RPUSH tasklog:%b %b", + (char *) &task_iid.id[0], UNIQUE_ID_SIZE, + (char *) &update, sizeof(update)); + } + + if (db->context->err) { + LOG_REDIS_ERR(db->context, "error setting task in task_log_add_task"); + } + } + + if (((bool *) callback_data->requests_info)[PUBLISH_INDEX] == false) { + redisAsyncCommand(db->context, redis_task_log_publish_publish_callback, + (void *) callback_data->timer_id, + "PUBLISH task_log:%b:%d %b", (char *) &node.id[0], + UNIQUE_ID_SIZE, state, (char *) task_instance, + task_instance_size(task_instance)); + + if (db->context->err) { + LOG_REDIS_ERR(db->context, "error publishing task in task_log_add_task"); + } + } +} + +void redis_task_log_publish_push_callback(redisAsyncContext *c, + void *r, + void *privdata) { + REDIS_CALLBACK_HEADER(db, callback_data, r) + + CHECK(callback_data->requests_info != NULL); + ((bool *) callback_data->requests_info)[PUSH_INDEX] = true; + + if (((bool *) callback_data->requests_info)[PUBLISH_INDEX] == true) { + if (callback_data->done_callback) { + task_log_done_callback done_callback = callback_data->done_callback; + done_callback(callback_data->id, callback_data->user_context); + } + destroy_timer_callback(db->loop, callback_data); + } +} + +void redis_task_log_publish_publish_callback(redisAsyncContext *c, + void *r, + void *privdata) { + REDIS_CALLBACK_HEADER(db, callback_data, r) + + CHECK(callback_data->requests_info != NULL); + ((bool *) callback_data->requests_info)[PUBLISH_INDEX] = true; + + if (((bool *) callback_data->requests_info)[PUSH_INDEX] == true) { + if (callback_data->done_callback) { + task_log_done_callback done_callback = callback_data->done_callback; + done_callback(callback_data->id, callback_data->user_context); + } + destroy_timer_callback(db->loop, callback_data); + } +} + +void task_log_redis_callback(redisAsyncContext *c, void *r, void *privdata) { + REDIS_CALLBACK_HEADER(db, callback_data, r) + redisReply *reply = r; + + CHECK(reply->type == REDIS_REPLY_ARRAY); + /* First entry is message type, second is topic, third is payload. */ + CHECK(reply->elements > 2); + /* If this condition is true, we got the initial message that acknowledged the + * subscription. */ + if (reply->element[2]->str == NULL) { + if (callback_data->done_callback) { + task_log_done_callback done_callback = callback_data->done_callback; + done_callback(callback_data->id, callback_data->user_context); + } + /* Note that we do not destroy the callback data yet because the + * subscription callback needs this data. */ + event_loop_remove_timer(db->loop, callback_data->timer_id); + return; + } + /* Otherwise, parse the task and call the callback. */ + task_log_subscribe_data *data = callback_data->data; + + task_instance *instance = malloc(reply->element[2]->len); + memcpy(instance, reply->element[2]->str, reply->element[2]->len); + if (data->subscribe_callback) { + data->subscribe_callback(instance, data->subscribe_context); + } task_instance_free(instance); } -void task_log_register_callback(db_handle *db, - task_log_callback callback, - node_id node, - int32_t state, - void *userdata) { - task_log_callback_data *callback_data = - malloc(sizeof(task_log_callback_data)); - utarray_push_back(db->callback_freelist, &callback_data); - callback_data->callback = callback; - callback_data->userdata = userdata; - if (memcmp(&node.id[0], &NIL_ID.id[0], UNIQUE_ID_SIZE) == 0) { - redisAsyncCommand(db->sub_context, task_log_redis_callback, callback_data, - "PSUBSCRIBE task_log:*:%d", state); + +void redis_task_log_subscribe(table_callback_data *callback_data) { + db_handle *db = callback_data->db_handle; + task_log_subscribe_data *data = callback_data->data; + + if (memcmp(&data->node.id[0], &NIL_ID.id[0], UNIQUE_ID_SIZE) == 0) { + redisAsyncCommand(db->sub_context, task_log_redis_callback, + (void *) callback_data->timer_id, + "PSUBSCRIBE task_log:*:%d", data->state_filter); } else { - redisAsyncCommand(db->sub_context, task_log_redis_callback, callback_data, - "SUBSCRIBE task_log:%b:%d", (char *) &node.id[0], - UNIQUE_ID_SIZE, state); + redisAsyncCommand(db->sub_context, task_log_redis_callback, + (void *) callback_data->timer_id, + "SUBSCRIBE task_log:%b:%d", (char *) &data->node.id[0], + UNIQUE_ID_SIZE, data->state_filter); } if (db->sub_context->err) { LOG_REDIS_ERR(db->sub_context, "error in task_log_register_callback"); diff --git a/src/common/state/redis.h b/src/common/state/redis.h index cf368b992..8d29616fe 100644 --- a/src/common/state/redis.h +++ b/src/common/state/redis.h @@ -11,58 +11,121 @@ #include "utarray.h" typedef struct { - /* Unique ID for this service. */ + /** Unique ID for this service. */ int service_id; - /* IP address and port of this service. */ + /** IP address and port of this service. */ char *addr; - /* Handle for the uthash table. */ + /** Handle for the uthash table. */ UT_hash_handle hh; } service_cache_entry; -typedef struct { - /* The callback that will be called. */ - task_log_callback callback; - /* Userdata associated with the callback. */ - void *userdata; -} task_log_callback_data; - -struct db_handle_impl { - /* String that identifies this client type. */ +struct db_handle { + /** String that identifies this client type. */ char *client_type; - /* Unique ID for this client within the type. */ + /** Unique ID for this client within the type. */ int64_t client_id; - /* Redis context for this global state store connection. */ + /** Redis context for this global state store connection. */ redisAsyncContext *context; - /* Redis context for "subscribe" communication. + /** Redis context for "subscribe" communication. * Yes, we need a separate one for that, see * https://github.com/redis/hiredis/issues/55 */ redisAsyncContext *sub_context; - /* The event loop this global state store connection is part of. */ + /** The event loop this global state store connection is part of. */ event_loop *loop; - /* Index of the database connection in the event loop */ + /** Index of the database connection in the event loop */ int64_t db_index; - /* Cache for the IP addresses of services. */ + /** Cache for the IP addresses of services. */ service_cache_entry *service_cache; - /* Redis context for synchronous connections. + /** Redis context for synchronous connections. * Should only be used very rarely, it is not asynchronous. */ redisContext *sync_context; - /* Data structure for callbacks that needs to be freed. */ + /** Data structure for callbacks that needs to be freed. */ UT_array *callback_freelist; }; -typedef struct { - /* The callback that will be called. */ - lookup_callback callback; - /* Object ID that is looked up. */ - object_id object_id; - /* Data context for the callback. */ - void *context; -} lookup_callback_data; - -void object_table_get_entry(redisAsyncContext *c, void *r, void *privdata); +void redis_object_table_get_entry(redisAsyncContext *c, + void *r, + void *privdata); void object_table_lookup_callback(redisAsyncContext *c, void *r, void *privdata); -#endif +/* + * ==== Redis object table functions ==== + */ + +/** + * Lookup object table entry in redis. + * + * @param callback_data Data structure containing redis connection and timeout + * information. + * @return Void. + */ +void redis_object_table_lookup(table_callback_data *callback_data); + +/** + * Add an entry to the object table in redis. + * + * @param callback_data Data structure containing redis connection and timeout + * information. + * @return Void. + */ +void redis_object_table_add(table_callback_data *callback_data); + +/** + * Subscribe to learn when a new object becomes available. + * + * @param callback_data Data structure containing redis connection and timeout + * information. + * @return Void. + */ +void redis_object_table_subscribe(table_callback_data *callback_data); + +/* + * ==== Redis task table function ===== + */ + +/** + * Add or update task log entry with new scheduling information. + * + * @param callback_data Data structure containing redis connection and timeout + * information. + * @return Void. + */ +void redis_task_log_publish(table_callback_data *callback_data); + +/** + * Callback invoked when the replya from the task push command is received. + * + * @param c Redis context. + * @param r Reply (not used). + * @param privdata Data associated to the callback. + * @return Void. + */ +void redis_task_log_publish_push_callback(redisAsyncContext *c, + void *r, + void *privdata); + +/** + * Callback invoked when the replya from the task publish command is received. + * + * @param c Redis context. + * @param r Reply (not used). + * @param privdata Data associated to the callback. + * @return Void. + */ +void redis_task_log_publish_publish_callback(redisAsyncContext *c, + void *r, + void *privdata); + +/** + * Subscribe to updates of the task log. + * + * @param callback_data Data structure containing redis connection and timeout + * information. + * @return Void. + */ +void redis_task_log_subscribe(table_callback_data *callback_data); + +#endif /* REDIS_H */ diff --git a/src/common/state/table.c b/src/common/state/table.c new file mode 100644 index 000000000..f7337e1e5 --- /dev/null +++ b/src/common/state/table.c @@ -0,0 +1,133 @@ +#include "table.h" + +#include +#include "redis.h" + +table_callback_data *init_table_callback(db_handle *db_handle, + unique_id id, + void *data, + retry_info *retry, + table_done_callback done_callback, + table_retry_callback retry_callback, + void *user_context) { + CHECK(db_handle); + CHECK(db_handle->loop); + CHECK(retry); + /* Allocate and initialize callback data structure for object table */ + table_callback_data *callback_data = malloc(sizeof(table_callback_data)); + CHECKM(callback_data != NULL, "Memory allocation error!") + callback_data->id = id; + callback_data->retry = *retry; + callback_data->done_callback = done_callback; + callback_data->retry_callback = retry_callback; + callback_data->data = data; + callback_data->requests_info = NULL; + callback_data->user_context = user_context; + callback_data->db_handle = db_handle; + /* Add timer and initialize it. */ + callback_data->timer_id = event_loop_add_timer( + db_handle->loop, retry->timeout, + (event_loop_timer_handler) table_timeout_handler, callback_data); + outstanding_callbacks_add(callback_data); + + callback_data->retry_callback(callback_data); + + return callback_data; +} + +void destroy_timer_callback(event_loop *loop, + table_callback_data *callback_data) { + event_loop_remove_timer(loop, callback_data->timer_id); + destroy_table_callback(callback_data); +} + +void destroy_table_callback(table_callback_data *callback_data) { + CHECK(callback_data != NULL); + + if (callback_data->requests_info) + free(callback_data->requests_info); + + outstanding_callbacks_remove(callback_data); + + /* Timer is removed via EVENT_LOOP_TIMER_DONE in the timeout callback. */ + free(callback_data); +} + +int64_t table_timeout_handler(event_loop *loop, + int64_t timer_id, + void *user_context) { + CHECK(loop != NULL); + 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); + + if (callback_data->retry.num_retries == 0) { + /* We didn't get a response from the database after exhausting all retries; + * let user know, cleanup the state, and remove the timer. */ + if (callback_data->retry.fail_callback) { + callback_data->retry.fail_callback(callback_data->id, + callback_data->user_context); + } + destroy_table_callback(callback_data); + return EVENT_LOOP_TIMER_DONE; + } + + /* Decrement retry count and try again. */ + callback_data->retry.num_retries--; + callback_data->retry_callback(callback_data); + return callback_data->retry.timeout; +} + +/** + * Hash table maintaining the outstanding callbacks. + * + * This hash table is used to handle the following case: + * - a table command is issued with an associated callback and a callback data + * structure; + * - the last timeout associated to this command expires, as a result the + * callback data structure is freed; + * - a reply arrives, but now the callback data structure is gone, so we have + * to ignore this reply; + * + * This hash table enables us to ignore such replies. The operations on the + * hash table are as follows. + * + * When we issue a table command and a timeout event to wait for the reply, we + * add a new entry to the hash table that is keyed by the ID of the timer. Note + * that table commands must have unique timer IDs, which are assigned by the + * Redis ae event loop. + * + * When we receive the reply, we check whether the callback still exists in + * this hash table, and if not we just ignore the reply. If the callback does + * exist, the reply receiver is responsible for removing the timer and the + * entry associated to the callback, or else the timeout handler will continue + * firing. + * + * When the last timeout associated to the command expires we remove the entry + * associated to the callback. + */ +static table_callback_data *outstanding_callbacks = NULL; + +void outstanding_callbacks_add(table_callback_data *callback_data) { + HASH_ADD_INT(outstanding_callbacks, timer_id, callback_data); +} + +table_callback_data *outstanding_callbacks_find(int64_t key) { + table_callback_data *callback_data = NULL; + HASH_FIND_INT(outstanding_callbacks, &key, callback_data); + return callback_data; +} + +void outstanding_callbacks_remove(table_callback_data *callback_data) { + HASH_DEL(outstanding_callbacks, callback_data); +} + +void destroy_outstanding_callbacks(event_loop *loop) { + table_callback_data *callback_data, *tmp; + HASH_ITER(hh, outstanding_callbacks, callback_data, tmp) { + destroy_timer_callback(loop, callback_data); + } +} diff --git a/src/common/state/table.h b/src/common/state/table.h new file mode 100644 index 000000000..f82084403 --- /dev/null +++ b/src/common/state/table.h @@ -0,0 +1,154 @@ +#ifndef TABLE_H +#define TABLE_H + +#include "uthash.h" +#include "stdbool.h" + +#include "common.h" +#include "db.h" + +typedef struct table_callback_data table_callback_data; + +typedef void *table_done_callback; + +/* The callback called when the database operation hasn't completed after + * the number of retries specified for the operation. */ +typedef void (*table_fail_callback)(unique_id id, void *user_context); + +typedef void (*table_retry_callback)(table_callback_data *callback_data); + +/** + * Data structure consolidating the retry related varaibles. + */ +typedef struct { + /** Number of retries left. */ + int num_retries; + /** Timeout, in milliseconds. */ + uint64_t timeout; + /** The callback that will be called if there are no more retries left. */ + table_fail_callback fail_callback; +} retry_info; + +struct table_callback_data { + /** ID of the entry in the table that we are going to look up, remove or add. + */ + unique_id id; + /** The callback that will be called when results is returned. */ + table_done_callback done_callback; + /** The callback that will be called to initiate the next try. */ + table_retry_callback retry_callback; + /** Retry information containing the remaining number of retries, the timeout + * before the next retry, and a pointer to the failure callback. + */ + retry_info retry; + /** Pointer to the data that is entered into the table. */ + void *data; + /** Pointer to the data used internally to handle multiple database requests. + */ + void *requests_info; + /** User context. */ + void *user_context; + /** Handle to db. */ + db_handle *db_handle; + /** Handle to timer. */ + int64_t timer_id; + UT_hash_handle hh; /* makes this structure hashable */ +}; + +/** + * Function to handle the timeout event. + * + * @param loop Event loop. + * @param timer_id Timer identifier. + * @param context Pointer to the callback data for the object table + * @return Timeout to reset the timer if we need to try again, or + * EVENT_LOOP_TIMER_DONE if retry_count == 0. + */ +int64_t table_timeout_handler(event_loop *loop, + int64_t timer_id, + void *context); + +/** + * Initialize the table callback and call the retry_callback for the first time. + * + * @param db_handle Database handle. + * @param id ID of the object that is looked up, added or removed. + * @param data Data entered into the table. Shall be freed by the user. + * @param retry Retry relevant information: retry timeout, number of remaining + * retries, and retry callback. + * @param done_callback Function to be called when database returns result. + * @param fail_callback Function to be called when number of retries is + * exhausted. + * @param user_context Context that can be provided by the user and will be + * passed on to the various callbacks. + * @return New table callback data struct. + */ +table_callback_data *init_table_callback(db_handle *db_handle, + unique_id id, + void *data, + retry_info *retry, + table_done_callback done_callback, + table_retry_callback retry_callback, + void *user_context); + +/** + * Destroy any state associated with the callback data. This removes all + * associated state from the outstanding callbacks hash table and frees any + * associated memory. This does not remove any associated timer events. + * + * @param callback_data The pointer to the data structure of the callback we + * want to remove. + * @return Void. + */ +void destroy_table_callback(table_callback_data *callback_data); + +/** + * Destroy all state events associated with the callback data, including memory + * and timer events. + * + * @param callback_data The pointer to the data structure of the callback we + * want to remove. + * @return Void. + */ +void destroy_timer_callback(event_loop *loop, + table_callback_data *callback_data); + +/** + * Add an outstanding callback entry. + * + * @param callback_data The pointer to the data structure of the callback we + * want to insert. + * @return None. + */ +void outstanding_callbacks_add(table_callback_data *callback_data); + +/** + * Find an outstanding callback entry. + * + * @param key The key for the outstanding callbacks hash table. We use the + * timer ID assigned by the Redis ae event loop. + * @return Returns the callback data if found, NULL otherwise. + */ +table_callback_data *outstanding_callbacks_find(int64_t key); + +/** + * Remove an outstanding callback entry. This only removes the callback entry + * from the hash table. It does not free the entry or remove any associated + * timer events. + * + * @param callback_data The pointer to the data structure of the callback we + * want to remove. + * @return Void. + */ +void outstanding_callbacks_remove(table_callback_data *callback_data); + +/** + * Destroy all outstanding callbacks and remove their associated timer events + * from the event loop. + * + * @param loop The event loop from which we want to remove the timer events. + * @return Void. + */ +void destroy_outstanding_callbacks(event_loop *loop); + +#endif /* TABLE_H */ diff --git a/src/common/state/task_log.c b/src/common/state/task_log.c new file mode 100644 index 000000000..d7887da02 --- /dev/null +++ b/src/common/state/task_log.c @@ -0,0 +1,34 @@ +#include "task_log.h" +#include "redis.h" + +#define NUM_DB_REQUESTS 2 + +void task_log_publish(db_handle *db_handle, + task_instance *task_instance, + retry_info *retry, + task_log_done_callback done_callback, + void *user_context) { + init_table_callback(db_handle, *task_instance_id(task_instance), + task_instance, retry, done_callback, + redis_task_log_publish, user_context); +} + +/* TODO(swang): A corresponding task_log_unsubscribe. */ +void task_log_subscribe(db_handle *db_handle, + node_id node, + int32_t state_filter, + task_log_subscribe_callback subscribe_callback, + void *subscribe_context, + retry_info *retry, + task_log_done_callback done_callback, + void *user_context) { + task_log_subscribe_data *sub_data = malloc(sizeof(task_log_subscribe_data)); + utarray_push_back(db_handle->callback_freelist, &sub_data); + sub_data->node = node; + sub_data->state_filter = state_filter; + sub_data->subscribe_callback = subscribe_callback; + sub_data->subscribe_context = subscribe_context; + + init_table_callback(db_handle, node, sub_data, retry, done_callback, + redis_task_log_subscribe, user_context); +} diff --git a/src/common/state/task_log.h b/src/common/state/task_log.h index acf5dbcd0..1176976f9 100644 --- a/src/common/state/task_log.h +++ b/src/common/state/task_log.h @@ -2,9 +2,11 @@ #define TASK_LOG_H #include "db.h" +#include "table.h" #include "task.h" -/* The task log is a message bus that is used for all communication between +/** + * The task log is a message bus that is used for all communication between * local and global schedulers (and also persisted to the state database). * Here are examples of events that are recorded by the task log: * @@ -15,27 +17,72 @@ * 5) local scheduler writes it when a task finishes execution; * 6) global scheduler reads it to get the tasks that have finished; */ +/* Callback called when the task log operation completes. */ +typedef void (*task_log_done_callback)(task_iid task_iid, void *user_context); + +/* + * ==== Publish the task log ==== + */ + +/** + * Add or update a task instance to the task log. + * + * @param db_handle Database handle. + * @param retry Information about retrying the request to the database. + * @param done_callback Function to be called when database returns result. + * @param user_context Data that will be passed to done_callback and + * fail_callback. + * @return Void. + */ +void task_log_publish(db_handle *db_handle, + task_instance *task_instance, + retry_info *retry, + task_log_done_callback done_callback, + void *user_context); + +/* + * ==== Subscribing to the task log ==== + */ + /* Callback for subscribing to the task log. */ -typedef void (*task_log_callback)(task_instance *task_instance, void *userdata); +typedef void (*task_log_subscribe_callback)(task_instance *task_instance, + void *user_context); -/* Initially add a task instance to the task log. */ -void task_log_add_task(db_handle *db, task_instance *task_instance); +/** + * Register callback for a certain event. + * + * @param db_handle Database handle. + * @param subscribe_callback Callback that will be called when the task log is + * updated. + * @param subscribe_context Context that will be passed into the + * subscribe_callback. + * @param node Node whose events we want to listen to. If you want to register + * to updates from all nodes, set node = NIL_ID. + * @param state_filter Flags for events we want to listen to. If you want + * to listen to all events, use state_filter = TASK_WAITING | + * TASK_SCHEDULED | TASK_RUNNING | TASK_DONE. + * @param retry Information about retrying the request to the database. + * @param done_callback Function to be called when database returns result. + * @param user_context Data that will be passed to done_callback and + * fail_callback. + * @return Void. + */ +void task_log_subscribe(db_handle *db_handle, + node_id node, + int32_t state_filter, + task_log_subscribe_callback subscribe_callback, + void *subscribe_context, + retry_info *retry, + task_log_done_callback done_callback, + void *user_context); -/* Update task instance in the task log. */ -void task_log_update_task(db_handle *db, - task_iid task_iid, - int32_t state, - node_id node); - -/* Register callback for a certain event. The node specifies the node whose - * events we want to listen to. If you want to listen to all events for this - * node, use state_filter = - * TASK_WAITING | TASK_SCHEDULED | TASK_RUNNING | TASK_DONE. - * If you want to register to updates from all nodes, set node = NIL_ID. */ -void task_log_register_callback(db_handle *db, - task_log_callback callback, - node_id node, - int32_t state_filter, - void *userdata); +/* Data that is needed to register task log subscribe callbacks with the state + * database. */ +typedef struct { + node_id node; + int32_t state_filter; + task_log_subscribe_callback subscribe_callback; + void *subscribe_context; +} task_log_subscribe_data; #endif /* TASK_LOG_H */ diff --git a/src/common/task.h b/src/common/task.h index 782bf685b..8d818e3a3 100644 --- a/src/common/task.h +++ b/src/common/task.h @@ -27,9 +27,9 @@ typedef unique_id task_iid; typedef unique_id node_id; /* - * TASK SPECIFICATIONS: Contain all the information neccessary - * to execute the task (function id, arguments, return object ids). - * + * ==== Task specifications ==== + * Contain all the information neccessary to execute the + * task (function id, arguments, return object ids). */ typedef struct task_spec_impl task_spec; @@ -83,11 +83,10 @@ task_spec *read_task(int fd); void print_task(task_spec *spec, UT_string *output); /* - * SCHEDULED TASK: Contains information about a scheduled task: - * the task iid, the task specification and the task status - * (WAITING, SCHEDULED, RUNNING, DONE) and which node the - * task is scheduled on. - * + * ==== Task instance ==== + * Contains information about a scheduled task: The task iid, + * the task specification and the task status (WAITING, SCHEDULED, + * RUNNING, DONE) and which node the task is scheduled on. */ /* The scheduling_state can be used as a flag when we are listening @@ -129,4 +128,14 @@ task_spec *task_instance_task_spec(task_instance *instance); /* Free this task instance datastructure. */ void task_instance_free(task_instance *instance); +/* + * ==== Task update ==== + * Contains the information necessary to update a task in the task log. + */ + +typedef struct { + int32_t state; + node_id node; +} task_update; + #endif diff --git a/src/common/test/db_tests.c b/src/common/test/db_tests.c index 74dd9a850..bc0bfa6ae 100644 --- a/src/common/test/db_tests.c +++ b/src/common/test/db_tests.c @@ -14,6 +14,10 @@ SUITE(db_tests); +/* Retry 10 times with an 100ms timeout. */ +const int NUM_RETRIES = 10; +const uint64_t TIMEOUT = 50; + const char *manager_addr = "127.0.0.1"; int manager_port1 = 12345; int manager_port2 = 12346; @@ -22,11 +26,16 @@ char received_port1[6] = {0}; char received_addr2[16] = {0}; char received_port2[6] = {0}; +typedef struct { int test_number; } user_context; + +const int TEST_NUMBER = 10; + /* Test if entries have been written to the database. */ -void test_callback(object_id object_id, - int manager_count, - const char *manager_vector[], - void *context) { + +void lookup_done_callback(object_id object_id, + int manager_count, + const char *manager_vector[], + void *user_context) { CHECK(manager_count == 2); if (!manager_vector[0] || sscanf(manager_vector[0], "%15[0-9.]:%5[0-9]", received_addr1, @@ -41,7 +50,16 @@ void test_callback(object_id object_id, free(manager_vector); } -int timeout_handler(event_loop *loop, timer_id timer_id, void *context) { +/* Entry added to database successfully. */ +void add_done_callback(object_id object_id, void *user_context) {} + +/* Test if we got a timeout callback if we couldn't connect database. */ +void timeout_callback(object_id object_id, void *context) { + user_context *uc = (user_context *) context; + CHECK(uc->test_number == TEST_NUMBER) +} + +int64_t timeout_handler(event_loop *loop, int64_t id, void *context) { event_loop_stop(loop); return EVENT_LOOP_TIMER_DONE; } @@ -55,12 +73,19 @@ TEST object_table_lookup_test(void) { db_attach(db1, loop); db_attach(db2, loop); unique_id id = globally_unique_id(); - object_table_add(db1, id); - object_table_add(db2, id); - event_loop_add_timer(loop, 100, timeout_handler, NULL); + retry_info retry = { + .num_retries = NUM_RETRIES, + .timeout = TIMEOUT, + .fail_callback = timeout_callback, + }; + object_table_add(db1, id, &retry, add_done_callback, NULL); + object_table_add(db2, id, &retry, add_done_callback, NULL); + event_loop_add_timer(loop, 200, (event_loop_timer_handler) timeout_handler, + NULL); event_loop_run(loop); - object_table_lookup(db1, id, test_callback, NULL); - event_loop_add_timer(loop, 100, timeout_handler, NULL); + object_table_lookup(db1, id, &retry, lookup_done_callback, NULL); + event_loop_add_timer(loop, 200, (event_loop_timer_handler) timeout_handler, + NULL); event_loop_run(loop); int port1 = atoi(received_port1); int port2 = atoi(received_port2); @@ -71,6 +96,7 @@ TEST object_table_lookup_test(void) { db_disconnect(db1); db_disconnect(db2); + destroy_outstanding_callbacks(loop); event_loop_destroy(loop); PASS(); } @@ -90,14 +116,19 @@ TEST task_log_test(void) { task_spec *task = example_task(); task_instance *instance = make_task_instance(globally_unique_id(), task, TASK_STATUS_SCHEDULED, node); - task_log_register_callback(db, task_log_test_callback, node, - TASK_STATUS_SCHEDULED, instance); - task_log_add_task(db, instance); - event_loop_add_timer(loop, 100, timeout_handler, NULL); + retry_info retry = { + .num_retries = NUM_RETRIES, .timeout = TIMEOUT, .fail_callback = NULL, + }; + task_log_subscribe(db, node, TASK_STATUS_SCHEDULED, task_log_test_callback, + instance, &retry, NULL, NULL); + task_log_publish(db, instance, &retry, NULL, NULL); + event_loop_add_timer(loop, 200, (event_loop_timer_handler) timeout_handler, + NULL); event_loop_run(loop); task_instance_free(instance); free_task_spec(task); db_disconnect(db); + destroy_outstanding_callbacks(loop); event_loop_destroy(loop); PASS(); } @@ -118,16 +149,25 @@ TEST task_log_all_test(void) { globally_unique_id(), task, TASK_STATUS_SCHEDULED, globally_unique_id()); task_instance *instance2 = make_task_instance( globally_unique_id(), task, TASK_STATUS_SCHEDULED, globally_unique_id()); - task_log_register_callback(db, task_log_all_test_callback, NIL_ID, - TASK_STATUS_SCHEDULED, NULL); - task_log_add_task(db, instance1); - task_log_add_task(db, instance2); - event_loop_add_timer(loop, 100, timeout_handler, NULL); + retry_info retry = { + .num_retries = NUM_RETRIES, .timeout = TIMEOUT, .fail_callback = NULL, + }; + task_log_subscribe(db, NIL_ID, TASK_STATUS_SCHEDULED, + task_log_all_test_callback, NULL, &retry, NULL, NULL); + event_loop_add_timer(loop, 50, (event_loop_timer_handler) timeout_handler, + NULL); + event_loop_run(loop); + /* TODO(pcm): Get rid of this sleep once the robust pubsub is implemented. */ + task_log_publish(db, instance1, &retry, NULL, NULL); + task_log_publish(db, instance2, &retry, NULL, NULL); + event_loop_add_timer(loop, 200, (event_loop_timer_handler) timeout_handler, + NULL); event_loop_run(loop); task_instance_free(instance2); task_instance_free(instance1); free_task_spec(task); db_disconnect(db); + destroy_outstanding_callbacks(loop); event_loop_destroy(loop); ASSERT(num_test_callback_called == 2); PASS(); diff --git a/src/common/test/example_task.h b/src/common/test/example_task.h index 0dddc4dc1..f752b824d 100644 --- a/src/common/test/example_task.h +++ b/src/common/test/example_task.h @@ -11,4 +11,13 @@ task_spec *example_task(void) { return task; } +task_instance *example_task_instance(void) { + task_iid iid = globally_unique_id(); + task_spec *spec = example_task(); + task_instance *instance = + make_task_instance(iid, spec, TASK_STATUS_WAITING, NIL_ID); + free_task_spec(spec); + return instance; +} + #endif diff --git a/src/common/test/object_table_tests.c b/src/common/test/object_table_tests.c new file mode 100644 index 000000000..3345ac50f --- /dev/null +++ b/src/common/test/object_table_tests.c @@ -0,0 +1,513 @@ +#include "greatest.h" + +#include "event_loop.h" +#include "example_task.h" +#include "common.h" +#include "state/object_table.h" +#include "state/redis.h" + +#include + +SUITE(object_table_tests); + +static event_loop *g_loop; + +/* ==== Test if operations time out correctly ==== */ + +/* === Test lookup timeout === */ + +const char *lookup_timeout_context = "lookup_timeout"; +int lookup_failed = 0; + +void lookup_done_callback(object_id object_id, + int manager_count, + OWNER const char *manager_vector[], + void *context) { + /* The done callback should not be called. */ + CHECK(0); +} + +void lookup_fail_callback(unique_id id, void *user_data) { + lookup_failed = 1; + CHECK(user_data == (void *) lookup_timeout_context); + event_loop_stop(g_loop); +} + +TEST lookup_timeout_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 1234); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 5, .timeout = 100, .fail_callback = lookup_fail_callback, + }; + object_table_lookup(db, NIL_ID, &retry, lookup_done_callback, + (void *) lookup_timeout_context); + /* Disconnect the database to see if the lookup times out. */ + close(db->context->c.fd); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(lookup_failed); + PASS(); +} + +/* === Test add timeout === */ + +const char *add_timeout_context = "add_timeout"; +int add_failed = 0; + +void add_done_callback(object_id object_id, void *user_context) { + /* The done callback should not be called. */ + CHECK(0); +} + +void add_fail_callback(unique_id id, void *user_data) { + add_failed = 1; + CHECK(user_data == (void *) add_timeout_context); + event_loop_stop(g_loop); +} + +TEST add_timeout_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 1234); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 5, .timeout = 100, .fail_callback = add_fail_callback, + }; + object_table_add(db, NIL_ID, &retry, add_done_callback, + (void *) add_timeout_context); + /* Disconnect the database to see if the lookup times out. */ + close(db->context->c.fd); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(add_failed); + PASS(); +} + +/* === Test subscribe timeout === */ + +const char *subscribe_timeout_context = "subscribe_timeout"; +int subscribe_failed = 0; + +void subscribe_done_callback(object_id object_id, void *user_context) { + /* The done callback should not be called. */ + CHECK(0); +} + +void subscribe_fail_callback(unique_id id, void *user_data) { + subscribe_failed = 1; + CHECK(user_data == (void *) subscribe_timeout_context); + event_loop_stop(g_loop); +} + +TEST subscribe_timeout_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 1234); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 5, + .timeout = 100, + .fail_callback = subscribe_fail_callback, + }; + object_table_subscribe(db, NIL_ID, NULL, NULL, &retry, + subscribe_done_callback, + (void *) subscribe_timeout_context); + /* Disconnect the database to see if the lookup times out. */ + close(db->sub_context->c.fd); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(subscribe_failed); + PASS(); +} + +/* ==== Test if the retry is working correctly ==== */ + +int64_t reconnect_context_callback(event_loop *loop, + int64_t timer_id, + void *context) { + db_handle *db = context; + /* Reconnect to redis. This is not reconnecting the pub/sub channel. */ + redisAsyncFree(db->context); + redisFree(db->sync_context); + db->context = redisAsyncConnect("127.0.0.1", 6379); + db->context->data = (void *) db; + db->sync_context = redisConnect("127.0.0.1", 6379); + /* Re-attach the database to the event loop (the file descriptor changed). */ + db_attach(db, loop); + return EVENT_LOOP_TIMER_DONE; +} + +int64_t terminate_event_loop_callback(event_loop *loop, + int64_t timer_id, + void *context) { + event_loop_stop(loop); + return EVENT_LOOP_TIMER_DONE; +} + +/* === Test lookup retry === */ + +const char *lookup_retry_context = "lookup_retry"; +int lookup_retry_succeeded = 0; + +void lookup_retry_done_callback(object_id object_id, + int manager_count, + OWNER const char *manager_vector[], + void *context) { + CHECK(context == (void *) lookup_retry_context); + lookup_retry_succeeded = 1; + free(manager_vector); +} + +void lookup_retry_fail_callback(unique_id id, void *user_data) { + /* The fail callback should not be called. */ + CHECK(0); +} + +TEST lookup_retry_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11235); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 5, + .timeout = 100, + .fail_callback = lookup_retry_fail_callback, + }; + object_table_lookup(db, NIL_ID, &retry, lookup_retry_done_callback, + (void *) lookup_retry_context); + /* Disconnect the database to let the lookup time out the first time. */ + close(db->context->c.fd); + /* Install handler for reconnecting the database. */ + event_loop_add_timer( + g_loop, 150, (event_loop_timer_handler) reconnect_context_callback, db); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(g_loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(lookup_retry_succeeded); + PASS(); +} + +/* === Test add retry === */ + +const char *add_retry_context = "add_retry"; +int add_retry_succeeded = 0; + +void add_retry_done_callback(object_id object_id, void *user_context) { + CHECK(user_context == (void *) add_retry_context); + add_retry_succeeded = 1; +} + +void add_retry_fail_callback(unique_id id, void *user_data) { + /* The fail callback should not be called. */ + CHECK(0); +} + +TEST add_retry_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11235); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 5, + .timeout = 100, + .fail_callback = add_retry_fail_callback, + }; + object_table_add(db, NIL_ID, &retry, add_retry_done_callback, + (void *) add_retry_context); + /* Disconnect the database to let the add time out the first time. */ + close(db->context->c.fd); + /* Install handler for reconnecting the database. */ + event_loop_add_timer( + g_loop, 150, (event_loop_timer_handler) reconnect_context_callback, db); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(g_loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(add_retry_succeeded); + PASS(); +} + +/* === Test subscribe retry === */ + +const char *subscribe_retry_context = "subscribe_retry"; +int subscribe_retry_succeeded = 0; + +int64_t reconnect_sub_context_callback(event_loop *loop, + int64_t timer_id, + void *context) { + db_handle *db = context; + /* Reconnect to redis. This is not reconnecting the pub/sub channel. */ + redisAsyncFree(db->sub_context); + redisFree(db->sync_context); + db->sub_context = redisAsyncConnect("127.0.0.1", 6379); + db->sub_context->data = (void *) db; + db->sync_context = redisConnect("127.0.0.1", 6379); + /* Re-attach the database to the event loop (the file descriptor changed). */ + db_attach(db, loop); + return EVENT_LOOP_TIMER_DONE; +} + +void subscribe_retry_done_callback(object_id object_id, void *user_context) { + CHECK(user_context == (void *) subscribe_retry_context); + subscribe_retry_succeeded = 1; +} + +void subscribe_retry_fail_callback(unique_id id, void *user_data) { + /* The fail callback should not be called. */ + CHECK(0); +} + +TEST subscribe_retry_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11235); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 5, + .timeout = 100, + .fail_callback = subscribe_retry_fail_callback, + }; + object_table_subscribe(db, NIL_ID, NULL, NULL, &retry, + subscribe_retry_done_callback, + (void *) subscribe_retry_context); + /* Disconnect the database to let the subscribe times out the first time. */ + close(db->sub_context->c.fd); + /* Install handler for reconnecting the database. */ + event_loop_add_timer( + g_loop, 150, (event_loop_timer_handler) reconnect_sub_context_callback, + db); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(g_loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(subscribe_retry_succeeded); + PASS(); +} + +/* ==== Test if late succeed is working correctly ==== */ + +/* === Test lookup late succeed === */ + +const char *lookup_late_context = "lookup_late"; +int lookup_late_failed = 0; + +void lookup_late_fail_callback(unique_id id, void *user_context) { + CHECK(user_context == (void *) lookup_late_context); + lookup_late_failed = 1; +} + +void lookup_late_done_callback(object_id object_id, + int manager_count, + OWNER const char *manager_vector[], + void *context) { + /* This function should never be called. */ + CHECK(0); +} + +TEST lookup_late_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11236); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 0, + .timeout = 0, + .fail_callback = lookup_late_fail_callback, + }; + object_table_lookup(db, NIL_ID, &retry, lookup_late_done_callback, + (void *) lookup_late_context); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(g_loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + /* First process timer events to make sure the timeout is processed before + * anything else. */ + aeProcessEvents(g_loop, AE_TIME_EVENTS); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(lookup_late_failed); + PASS(); +} + +/* === Test add late succeed === */ + +const char *add_late_context = "add_late"; +int add_late_failed = 0; + +void add_late_fail_callback(unique_id id, void *user_context) { + CHECK(user_context == (void *) add_late_context); + add_late_failed = 1; +} + +void add_late_done_callback(object_id object_id, void *user_context) { + /* This function should never be called. */ + CHECK(0); +} + +TEST add_late_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11236); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 0, .timeout = 0, .fail_callback = add_late_fail_callback, + }; + object_table_add(db, NIL_ID, &retry, add_late_done_callback, + (void *) add_late_context); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(g_loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + /* First process timer events to make sure the timeout is processed before + * anything else. */ + aeProcessEvents(g_loop, AE_TIME_EVENTS); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(add_late_failed); + PASS(); +} + +/* === Test subscribe late succeed === */ + +const char *subscribe_late_context = "subscribe_late"; +int subscribe_late_failed = 0; + +void subscribe_late_fail_callback(unique_id id, void *user_context) { + CHECK(user_context == (void *) subscribe_late_context); + subscribe_late_failed = 1; +} + +void subscribe_late_done_callback(object_id object_id, void *user_context) { + /* This function should never be called. */ + CHECK(0); +} + +TEST subscribe_late_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11236); + db_attach(db, g_loop); + retry_info retry = { + .num_retries = 0, + .timeout = 0, + .fail_callback = subscribe_late_fail_callback, + }; + object_table_subscribe(db, NIL_ID, NULL, NULL, &retry, + subscribe_late_done_callback, + (void *) subscribe_late_context); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(g_loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + /* First process timer events to make sure the timeout is processed before + * anything else. */ + aeProcessEvents(g_loop, AE_TIME_EVENTS); + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + ASSERT(subscribe_late_failed); + PASS(); +} + +/* === Test subscribe object available succeed === */ + +const char *subscribe_success_context = "subscribe_success"; +int subscribe_success_done = 0; +int subscribe_success_succeeded = 0; + +void subscribe_success_fail_callback(unique_id id, void *user_context) { + /* This function should never be called. */ + CHECK(0); +} + +void subscribe_success_done_callback(object_id object_id, void *user_context) { + retry_info retry = { + .num_retries = 0, .timeout = 0, .fail_callback = NULL, + }; + object_table_add((db_handle *) user_context, object_id, &retry, NULL, NULL); + subscribe_success_done = 1; +} + +void subscribe_success_object_available_callback(object_id object_id, + void *user_context) { + CHECK(user_context == (void *) subscribe_success_context); + subscribe_success_succeeded = 1; +} + +TEST subscribe_success_test(void) { + g_loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11236); + db_attach(db, g_loop); + unique_id id = globally_unique_id(); + + retry_info retry = { + .num_retries = 0, + .timeout = 100, + .fail_callback = subscribe_success_fail_callback, + }; + object_table_subscribe(db, id, subscribe_success_object_available_callback, + (void *) subscribe_success_context, &retry, + subscribe_success_done_callback, (void *) db); + + /* Install handler for terminating the event loop. */ + event_loop_add_timer(g_loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + + event_loop_run(g_loop); + db_disconnect(db); + destroy_outstanding_callbacks(g_loop); + event_loop_destroy(g_loop); + + ASSERT(subscribe_success_done); + ASSERT(subscribe_success_succeeded); + PASS(); +} + +SUITE(object_table_tests) { + RUN_TEST(lookup_timeout_test); + RUN_TEST(add_timeout_test); + RUN_TEST(subscribe_timeout_test); + RUN_TEST(lookup_retry_test); + RUN_TEST(add_retry_test); + RUN_TEST(subscribe_retry_test); + RUN_TEST(lookup_late_test); + RUN_TEST(add_late_test); + RUN_TEST(subscribe_late_test); + RUN_TEST(subscribe_success_test); +} + +GREATEST_MAIN_DEFS(); + +int main(int argc, char **argv) { + GREATEST_MAIN_BEGIN(); + RUN_SUITE(object_table_tests); + GREATEST_MAIN_END(); +} diff --git a/src/common/test/redis_tests.c b/src/common/test/redis_tests.c index 0a50c7d1b..e756f871c 100644 --- a/src/common/test/redis_tests.c +++ b/src/common/test/redis_tests.c @@ -118,7 +118,7 @@ TEST async_redis_socket_test(void) { event_loop_add_timer(loop, 100, timeout_handler, NULL); event_loop_run(loop); - CHECK(async_redis_socket_test_callback_called); + ASSERT(async_redis_socket_test_callback_called); db_disconnect(db); event_loop_destroy(loop); @@ -193,7 +193,7 @@ TEST logging_test(void) { event_loop_add_timer(loop, 100, timeout_handler, NULL); event_loop_run(loop); - CHECK(logging_test_callback_called); + ASSERT(logging_test_callback_called); free_ray_logger(logger); db_disconnect(conn); diff --git a/src/common/test/task_log_tests.c b/src/common/test/task_log_tests.c new file mode 100644 index 000000000..570cecacf --- /dev/null +++ b/src/common/test/task_log_tests.c @@ -0,0 +1,316 @@ +#include "greatest.h" + +#include "event_loop.h" +#include "example_task.h" +#include "common.h" +#include "state/object_table.h" +#include "state/redis.h" + +#include +#include + +SUITE(task_log_tests); + +event_loop *loop; + +/* ==== Test if operations time out correctly ==== */ + +/* === Test subscribe timeout === */ + +const char *subscribe_timeout_context = "subscribe_timeout"; +int subscribe_failed = 0; + +void subscribe_done_callback(task_iid task_iid, void *user_context) { + /* The done callback should not be called. */ + CHECK(0); +} + +void subscribe_fail_callback(unique_id id, void *user_data) { + subscribe_failed = 1; + CHECK(user_data == (void *) subscribe_timeout_context); + event_loop_stop(loop); +} + +TEST subscribe_timeout_test(void) { + loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 1234); + db_attach(db, loop); + retry_info retry = { + .num_retries = 5, + .timeout = 100, + .fail_callback = subscribe_fail_callback, + }; + task_log_subscribe(db, NIL_ID, TASK_STATUS_WAITING, NULL, NULL, &retry, + subscribe_done_callback, + (void *) subscribe_timeout_context); + /* Disconnect the database to see if the subscribe times out. */ + close(db->sub_context->c.fd); + aeProcessEvents(loop, AE_TIME_EVENTS); + event_loop_run(loop); + db_disconnect(db); + destroy_outstanding_callbacks(loop); + event_loop_destroy(loop); + ASSERT(subscribe_failed); + PASS(); +} + +/* === Test publish timeout === */ + +const char *publish_timeout_context = "publish_timeout"; +const int publish_test_number = 272; +int publish_failed = 0; + +void publish_done_callback(task_iid task_iid, void *user_context) { + /* The done callback should not be called. */ + CHECK(0); +} + +void publish_fail_callback(unique_id id, void *user_data) { + publish_failed = 1; + CHECK(user_data == (void *) publish_timeout_context); + event_loop_stop(loop); +} + +TEST publish_timeout_test(void) { + loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 1234); + db_attach(db, loop); + task_instance *task = example_task_instance(); + retry_info retry = { + .num_retries = 5, .timeout = 100, .fail_callback = publish_fail_callback, + }; + task_log_publish(db, task, &retry, publish_done_callback, + (void *) publish_timeout_context); + /* Disconnect the database to see if the publish times out. */ + close(db->context->c.fd); + aeProcessEvents(loop, AE_TIME_EVENTS); + event_loop_run(loop); + db_disconnect(db); + destroy_outstanding_callbacks(loop); + event_loop_destroy(loop); + ASSERT(publish_failed); + task_instance_free(task); + PASS(); +} + +/* ==== Test if the retry is working correctly ==== */ + +int64_t reconnect_db_callback(event_loop *loop, + int64_t timer_id, + void *context) { + db_handle *db = context; + /* Reconnect to redis. */ + redisAsyncFree(db->sub_context); + db->sub_context = redisAsyncConnect("127.0.0.1", 6379); + db->sub_context->data = (void *) db; + /* Re-attach the database to the event loop (the file descriptor changed). */ + db_attach(db, loop); + return EVENT_LOOP_TIMER_DONE; +} + +int64_t terminate_event_loop_callback(event_loop *loop, + int64_t timer_id, + void *context) { + event_loop_stop(loop); + return EVENT_LOOP_TIMER_DONE; +} + +/* === Test subscribe retry === */ + +const char *subscribe_retry_context = "subscribe_retry"; +const int subscribe_retry_test_number = 273; +int subscribe_retry_succeeded = 0; + +void subscribe_retry_done_callback(object_id object_id, void *user_context) { + CHECK(user_context == (void *) subscribe_retry_context); + subscribe_retry_succeeded = 1; +} + +void subscribe_retry_fail_callback(unique_id id, void *user_data) { + /* The fail callback should not be called. */ + CHECK(0); +} + +TEST subscribe_retry_test(void) { + loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11235); + db_attach(db, loop); + retry_info retry = { + .num_retries = 5, + .timeout = 100, + .fail_callback = subscribe_retry_fail_callback, + }; + task_log_subscribe(db, NIL_ID, TASK_STATUS_WAITING, NULL, NULL, &retry, + subscribe_retry_done_callback, + (void *) subscribe_retry_context); + /* Disconnect the database to see if the subscribe times out. */ + close(db->sub_context->c.fd); + /* Install handler for reconnecting the database. */ + event_loop_add_timer(loop, 150, + (event_loop_timer_handler) reconnect_db_callback, db); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + event_loop_run(loop); + db_disconnect(db); + destroy_outstanding_callbacks(loop); + event_loop_destroy(loop); + ASSERT(subscribe_retry_succeeded); + PASS(); +} + +/* === Test publish retry === */ + +const char *publish_retry_context = "publish_retry"; +int publish_retry_succeeded = 0; + +void publish_retry_done_callback(object_id object_id, void *user_context) { + CHECK(user_context == (void *) publish_retry_context); + publish_retry_succeeded = 1; +} + +void publish_retry_fail_callback(unique_id id, void *user_data) { + /* The fail callback should not be called. */ + CHECK(0); +} + +TEST publish_retry_test(void) { + loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11235); + db_attach(db, loop); + task_instance *task = example_task_instance(); + retry_info retry = { + .num_retries = 5, + .timeout = 100, + .fail_callback = publish_retry_fail_callback, + }; + task_log_publish(db, task, &retry, publish_retry_done_callback, + (void *) publish_retry_context); + /* Disconnect the database to see if the publish times out. */ + close(db->sub_context->c.fd); + /* Install handler for reconnecting the database. */ + event_loop_add_timer(loop, 150, + (event_loop_timer_handler) reconnect_db_callback, db); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + event_loop_run(loop); + db_disconnect(db); + destroy_outstanding_callbacks(loop); + event_loop_destroy(loop); + ASSERT(publish_retry_succeeded); + task_instance_free(task); + PASS(); +} + +/* ==== Test if late succeed is working correctly ==== */ + +/* === Test subscribe late succeed === */ + +const char *subscribe_late_context = "subscribe_late"; +int subscribe_late_failed = 0; + +void subscribe_late_fail_callback(unique_id id, void *user_context) { + CHECK(user_context == (void *) subscribe_late_context); + subscribe_late_failed = 1; +} + +void subscribe_late_done_callback(task_iid task_iid, void *user_context) { + /* This function should never be called. */ + CHECK(0); +} + +TEST subscribe_late_test(void) { + loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11236); + db_attach(db, loop); + retry_info retry = { + .num_retries = 0, + .timeout = 0, + .fail_callback = subscribe_late_fail_callback, + }; + task_log_subscribe(db, NIL_ID, TASK_STATUS_WAITING, NULL, NULL, &retry, + subscribe_late_done_callback, + (void *) subscribe_late_context); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + /* First process timer events to make sure the timeout is processed before + * anything else. */ + aeProcessEvents(loop, AE_TIME_EVENTS); + event_loop_run(loop); + db_disconnect(db); + destroy_outstanding_callbacks(loop); + event_loop_destroy(loop); + ASSERT(subscribe_late_failed); + PASS(); +} + +/* === Test publish late succeed === */ + +const char *publish_late_context = "publish_late"; +int publish_late_failed = 0; + +void publish_late_fail_callback(unique_id id, void *user_context) { + CHECK(user_context == (void *) publish_late_context); + publish_late_failed = 1; +} + +void publish_late_done_callback(task_iid task_iik, void *user_context) { + /* This function should never be called. */ + CHECK(0); +} + +TEST publish_late_test(void) { + loop = event_loop_create(); + db_handle *db = + db_connect("127.0.0.1", 6379, "plasma_manager", "127.0.0.1", 11236); + db_attach(db, loop); + task_instance *task = example_task_instance(); + retry_info retry = { + .num_retries = 0, + .timeout = 0, + .fail_callback = publish_late_fail_callback, + }; + task_log_publish(db, task, &retry, publish_late_done_callback, + (void *) publish_late_context); + /* Install handler for terminating the event loop. */ + event_loop_add_timer(loop, 750, + (event_loop_timer_handler) terminate_event_loop_callback, + NULL); + /* First process timer events to make sure the timeout is processed before + * anything else. */ + aeProcessEvents(loop, AE_TIME_EVENTS); + event_loop_run(loop); + db_disconnect(db); + destroy_outstanding_callbacks(loop); + event_loop_destroy(loop); + ASSERT(publish_late_failed); + task_instance_free(task); + PASS(); +} + +SUITE(task_log_tests) { + RUN_TEST(subscribe_timeout_test); + RUN_TEST(publish_timeout_test); + RUN_TEST(subscribe_retry_test); + RUN_TEST(publish_retry_test); + RUN_TEST(subscribe_late_test); + RUN_TEST(publish_late_test); +} + +GREATEST_MAIN_DEFS(); + +int main(int argc, char **argv) { + GREATEST_MAIN_BEGIN(); + RUN_SUITE(task_log_tests); + GREATEST_MAIN_END(); +} diff --git a/src/photon/photon_algorithm.c b/src/photon/photon_algorithm.c index cfb92b5b2..8f2d6c1b5 100644 --- a/src/photon/photon_algorithm.c +++ b/src/photon/photon_algorithm.c @@ -141,7 +141,11 @@ void handle_task_submitted(scheduler_info *info, utarray_push_back(s->task_queue, &instance); } /* Submit the task to redis. */ - task_log_add_task(info->db, instance); + /* TODO(swang): We should set these values in a config file somewhere. */ + retry_info retry = { + .num_retries = 0, .timeout = 0, .fail_callback = NULL, + }; + task_log_publish(info->db, instance, &retry, NULL, NULL); if (schedule_locally) { /* If the task was scheduled locally, we need to free it. Otherwise, * ownership of the task is passed to the task_queue, and it will be freed diff --git a/src/plasma/plasma_manager.c b/src/plasma/plasma_manager.c index 5234d4dbd..63d6f4cea 100644 --- a/src/plasma/plasma_manager.c +++ b/src/plasma/plasma_manager.c @@ -136,6 +136,11 @@ int send_client_reply(client_connection *conn, plasma_reply *reply) { return (n != sizeof(plasma_reply)); } +int send_client_failure_reply(object_id object_id, client_connection *conn) { + plasma_reply reply = {.object_id = object_id, .has_object = 0}; + return send_client_reply(conn, &reply); +} + /** * Get the context for the given object ID for the given client * connection, if there is one active. @@ -259,9 +264,14 @@ void destroy_plasma_manager_state(plasma_manager_state *state) { free(manager_conn->ip_addr_port); free(manager_conn); } - /* There should not be any outstanding client connections if - * we're shutting down. */ - CHECK(state->fetch_connections == NULL); + + if (state->fetch_connections != NULL) { + LOG_DEBUG("There were outstanding fetch requests."); + client_object_connection *object_conn, *tmp; + HASH_ITER(fetch_hh, state->fetch_connections, object_conn, tmp) { + remove_object_connection(object_conn->client_conn, object_conn); + } + } free(state->plasma_conn); event_loop_destroy(state->loop); @@ -592,17 +602,26 @@ void request_transfer(object_id object_id, client_connection *client_conn = (client_connection *) context; client_object_connection *object_conn = get_object_connection(client_conn, object_id); - CHECK(object_conn); + /* If there's already an outstanding fetch for this object for this client, + * let the outstanding request finish the work. */ + if (object_conn) { + return; + } + /* If the object isn't on any managers, report a failure to the client. */ LOG_DEBUG("Object is on %d managers", manager_count); if (manager_count == 0) { /* TODO(swang): Instead of immediately counting this as a failure, maybe * register a Redis callback for changes to this object table entry. */ free(manager_vector); - plasma_reply reply = {.object_id = object_conn->object_id, .has_object = 0}; - send_client_reply(client_conn, &reply); - remove_object_connection(client_conn, object_conn); + send_client_failure_reply(object_id, client_conn); return; } + /* Register the new outstanding fetch with the current client connection. */ + object_conn = add_object_connection(client_conn, object_id); + if (!object_conn) { + LOG_DEBUG("Unable to allocate memory for object context."); + send_client_failure_reply(object_id, client_conn); + } /* Pick a different manager to request a transfer from on every attempt. */ object_conn->manager_count = manager_count; object_conn->manager_vector = malloc(manager_count * sizeof(char *)); @@ -640,18 +659,13 @@ void process_fetch_request(client_connection *client_conn, send_client_reply(client_conn, &reply); return; } - /* Register the new context with the current client connection. */ - /* TODO(swang): If there is already an outstanding fetch request for this - * object, exit now. */ - client_object_connection *object_conn = - add_object_connection(client_conn, object_id); - if (!object_conn) { - LOG_DEBUG("Unable to allocate memory for object context."); - reply.has_object = 0; - send_client_reply(client_conn, &reply); - } + retry_info retry = { + .num_retries = NUM_RETRIES, + .timeout = MANAGER_TIMEOUT, + .fail_callback = (table_fail_callback) send_client_failure_reply, + }; /* Request a transfer from a plasma manager that has this object. */ - object_table_lookup(client_conn->manager_state->db, object_id, + object_table_lookup(client_conn->manager_state->db, object_id, &retry, request_transfer, client_conn); } @@ -692,7 +706,15 @@ void process_message(event_loop *loop, case PLASMA_SEAL: LOG_DEBUG("Publishing to object table from DB client %d.", get_client_id(conn->manager_state->db)); - object_table_add(conn->manager_state->db, req->object_ids[0]); + /* 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 = NULL, + }; + object_table_add(conn->manager_state->db, req->object_ids[0], &retry, NULL, + NULL); break; case DISCONNECT_CLIENT: { LOG_INFO("Disconnecting client on fd %d", client_sock); diff --git a/src/plasma/test/manager_tests.c b/src/plasma/test/manager_tests.c index 90cc8e6ca..0babf7e9b 100644 --- a/src/plasma/test/manager_tests.c +++ b/src/plasma/test/manager_tests.c @@ -43,7 +43,6 @@ typedef struct { * object. */ plasma_connection *plasma_conn; client_connection *client_conn; - client_object_connection *object_conn; } plasma_mock; plasma_mock *init_plasma_mock(int port, plasma_mock *remote_mock) { @@ -66,32 +65,22 @@ plasma_mock *init_plasma_mock(int port, plasma_mock *remote_mock) { mock->write_conn = NULL; mock->read_conn = NULL; } - - mock->plasma_conn = NULL; - mock->client_conn = NULL; - mock->object_conn = NULL; - return mock; -} - -void add_mock_object_conn(plasma_mock *mock, object_id oid) { /* Connect a new client to the local plasma manager and mock a request to an * object. */ mock->plasma_conn = plasma_connect(store_socket_name, manager_socket_name); mock->client_conn = new_client_connection(mock->loop, mock->manager_local_fd, mock->state, 0); - mock->object_conn = add_object_connection(mock->client_conn, oid); + return mock; } void destroy_plasma_mock(plasma_mock *mock) { - if (mock->object_conn != NULL) { - free(mock->client_conn); - free(mock->plasma_conn); - } if (mock->read_conn != NULL) { close(get_client_sock(mock->read_conn)); free(mock->read_conn); } destroy_plasma_manager_state(mock->state); + free(mock->client_conn); + free(mock->plasma_conn); close(mock->local_store); close(mock->manager_local_fd); close(mock->manager_remote_fd); @@ -110,7 +99,6 @@ void destroy_plasma_mock(plasma_mock *mock) { */ TEST request_transfer_test(void) { plasma_mock *local_mock = init_plasma_mock(manager_port, NULL); - add_mock_object_conn(local_mock, oid); plasma_mock *remote_mock = init_plasma_mock(12346, local_mock); const char **manager_vector = malloc(sizeof(char *)); manager_vector[0] = "127.0.0.1:12346"; @@ -129,7 +117,6 @@ TEST request_transfer_test(void) { /* Clean up. */ free(req); destroy_plasma_mock(remote_mock); - remove_object_connection(local_mock->client_conn, local_mock->object_conn); destroy_plasma_mock(local_mock); PASS(); } @@ -148,7 +135,6 @@ TEST request_transfer_test(void) { */ TEST request_transfer_retry_test(void) { plasma_mock *local_mock = init_plasma_mock(manager_port, NULL); - add_mock_object_conn(local_mock, oid); plasma_mock *remote_mock1 = init_plasma_mock(12346, local_mock); plasma_mock *remote_mock2 = init_plasma_mock(12347, local_mock); const char **manager_vector = malloc(sizeof(char *) * 2); @@ -171,7 +157,6 @@ TEST request_transfer_retry_test(void) { free(req); destroy_plasma_mock(remote_mock2); destroy_plasma_mock(remote_mock1); - remove_object_connection(local_mock->client_conn, local_mock->object_conn); destroy_plasma_mock(local_mock); PASS(); } @@ -189,7 +174,6 @@ TEST request_transfer_retry_test(void) { */ TEST request_transfer_timeout_test(void) { plasma_mock *local_mock = init_plasma_mock(manager_port, NULL); - add_mock_object_conn(local_mock, oid); plasma_mock *remote_mock = init_plasma_mock(12346, local_mock); const char **manager_vector = malloc(sizeof(char *)); manager_vector[0] = "127.0.0.1:12346";