diff --git a/src/common/state/db.h b/src/common/state/db.h index e50adbbd8..2f0032604 100644 --- a/src/common/state/db.h +++ b/src/common/state/db.h @@ -30,11 +30,13 @@ db_handle *db_connect(const char *db_address, * Attach global system store connection to an event loop. Callbacks from * queries to the global system store will trigger events in the event loop. * - * @param db The database in question. - * @param loop The event loop to attach to. + * @param db The handle to the database that is connected. + * @param loop The event loop the database gets connected to. + * @param reattach Can only be true in unit tests. If true, the database is + * reattached to the loop. * @return Void. */ -void db_attach(db_handle *db, event_loop *loop); +void db_attach(db_handle *db, event_loop *loop, bool reattach); /** * Disconnect from the global system store. diff --git a/src/common/state/object_table.c b/src/common/state/object_table.c index ec5eda9b1..6d00c04d0 100644 --- a/src/common/state/object_table.c +++ b/src/common/state/object_table.c @@ -27,7 +27,7 @@ void object_table_subscribe( object_table_object_available_callback object_available_callback, void *subscribe_context, retry_info *retry, - object_table_done_callback done_callback, + object_table_lookup_done_callback done_callback, void *user_context) { CHECK(db_handle != NULL); object_table_subscribe_data *sub_data = diff --git a/src/common/state/object_table.h b/src/common/state/object_table.h index 00abb964c..94f6b8c45 100644 --- a/src/common/state/object_table.h +++ b/src/common/state/object_table.h @@ -89,8 +89,8 @@ void object_table_remove(db_handle *db, */ /* Callback called when object object_id is available. */ -typedef void (*object_table_object_available_callback)(object_id object_id, - void *user_context); +typedef object_table_lookup_done_callback + object_table_object_available_callback; /** * Subcribing to new object available function. @@ -98,13 +98,13 @@ typedef void (*object_table_object_available_callback)(object_id object_id, * @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. + * 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. + * @param user_context User context to be passed into the done and fail + * callbacks. * @return Void. */ @@ -114,7 +114,7 @@ void object_table_subscribe( object_table_object_available_callback object_available_callback, void *subscribe_context, retry_info *retry, - object_table_done_callback done_callback, + object_table_lookup_done_callback done_callback, void *user_context); /* Data that is needed to register new object available callbacks with the state diff --git a/src/common/state/redis.c b/src/common/state/redis.c index a411466c7..ea4e705b8 100644 --- a/src/common/state/redis.c +++ b/src/common/state/redis.c @@ -26,7 +26,8 @@ LOG_FATAL("could not allocate redis context"); \ } \ if (_context->err) { \ - LOG_REDIS_ERROR(_context, M, ##__VA_ARGS__); \ + LOG_ERROR(M, ##__VA_ARGS__); \ + LOG_REDIS_ERROR(_context, ""); \ exit(-1); \ } \ } while (0) @@ -38,10 +39,11 @@ db_handle *DB = c->data; \ table_callback_data *CB_DATA = \ outstanding_callbacks_find((int64_t) privdata); \ - if (CB_DATA == NULL) \ + if (CB_DATA == NULL) { \ /* the callback data structure has been \ * already freed; just ignore this reply */ \ return; \ + } \ do { \ } while (0) @@ -120,10 +122,18 @@ void db_disconnect(db_handle *db) { free(db); } -void db_attach(db_handle *db, event_loop *loop) { +void db_attach(db_handle *db, event_loop *loop, bool reattach) { db->loop = loop; - redisAeAttach(loop, db->context); - redisAeAttach(loop, db->sub_context); + int err = redisAeAttach(loop, db->context); + /* If the database is reattached in the tests, redis normally gives + * an error which we can safely ignore. */ + if (!reattach) { + CHECKM(err == REDIS_OK, "failed to attach the event loop"); + } + err = redisAeAttach(loop, db->sub_context); + if (!reattach) { + CHECKM(err == REDIS_OK, "failed to attach the event loop"); + } } /** @@ -373,13 +383,33 @@ void redis_object_table_get_entry(redisAsyncContext *c, memcpy(managers[j].id, reply->element[j]->str, sizeof(managers[j].id)); redis_get_cached_db_client(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); + if (done_callback) { + done_callback(callback_data->id, manager_count, manager_vector, + callback_data->user_context); + } + + if (callback_data->data != NULL) { + /* This callback was called from a subscribe call. */ + object_table_subscribe_data *sub_data = callback_data->data; + object_table_object_available_callback sub_callback = + sub_data->object_available_callback; + if (manager_count > 0) { + if (sub_callback) { + sub_callback(callback_data->id, manager_count, manager_vector, + sub_data->subscribe_context); + } + } + /* For the subscribe, don't delete the callback, only the timer. */ + event_loop_remove_timer(callback_data->db_handle->loop, + callback_data->timer_id); + } else { + /* This callback was called from a publish call. */ + /* For the lookup, remove timer and callback handler. */ + destroy_timer_callback(callback_data->db_handle->loop, callback_data); + } + if (manager_count > 0) { free(manager_vector); } @@ -389,35 +419,6 @@ void redis_object_table_get_entry(redisAsyncContext *c, free(managers); } -void redis_object_table_subscribe_lookup(redisAsyncContext *c, - void *r, - void *privdata) { - REDIS_CALLBACK_HEADER(db, callback_data, r); - redisReply *reply = r; - - if (reply->type == REDIS_REPLY_ARRAY) { - if (reply->elements > 0) { - CHECK(reply->element[0]->len == UNIQUE_ID_SIZE); - /* Check that the reply corresponds to the right object ID. */ - CHECK(strncmp(reply->element[0]->str, (char *) callback_data->id.id, - UNIQUE_ID_SIZE)); - object_table_subscribe_data *data = callback_data->data; - if (data->object_available_callback) { - data->object_available_callback(callback_data->id, - data->subscribe_context); - } - } - } else { - LOG_FATAL("expected integer or string, received type %d", reply->type); - } - - 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); -} - void object_table_redis_subscribe_callback(redisAsyncContext *c, void *r, void *privdata) { @@ -427,32 +428,15 @@ void object_table_redis_subscribe_callback(redisAsyncContext *c, 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. */ - bool is_add = - reply->element[1]->str && strcmp(reply->element[1]->str, "sadd") == 0; - if (is_add) { - /* Do a lookup to see if the key has been in redis before we started the - * subscription. */ - int status = - redisAsyncCommand(db->context, redis_object_table_subscribe_lookup, - (void *) callback_data->timer_id, "SMEMBERS obj:%b", - callback_data->id.id, sizeof(callback_data->id.id)); - if ((status == REDIS_ERR) || db->context->err) { - LOG_REDIS_ERROR(db->context, - "error in redis_object_table_subscribe_callback"); - } - return; - } - /* If the subscription is issued, parse the task and call the callback. */ - if (strcmp(reply->element[0]->str, "message") == 0) { - object_table_subscribe_data *data = callback_data->data; - - if (data->object_available_callback) { - data->object_available_callback(callback_data->id, - data->subscribe_context); - } + /* Do a lookup for the actual data. */ + int status = + redisAsyncCommand(db->context, redis_object_table_get_entry, + (void *) callback_data->timer_id, "SMEMBERS obj:%b", + callback_data->id.id, sizeof(callback_data->id.id)); + if ((status == REDIS_ERR) || db->context->err) { + LOG_REDIS_ERROR(db->context, + "error in redis_object_table_subscribe_callback"); } } @@ -463,7 +447,7 @@ void redis_object_table_subscribe(table_callback_data *callback_data) { object_id id = callback_data->id; int status = redisAsyncCommand( db->sub_context, object_table_redis_subscribe_callback, - (void *) callback_data->timer_id, "SUBSCRIBE __keyspace@0__:obj:%b sadd", + (void *) callback_data->timer_id, "SUBSCRIBE __keyspace@0__:obj:%b", id.id, sizeof(id.id)); if ((status == REDIS_ERR) || db->sub_context->err) { LOG_REDIS_DEBUG(db->sub_context, diff --git a/src/common/test/db_tests.c b/src/common/test/db_tests.c index 8b4594e31..062a19e22 100644 --- a/src/common/test/db_tests.c +++ b/src/common/test/db_tests.c @@ -69,8 +69,8 @@ TEST object_table_lookup_test(void) { manager_port1); db_handle *db2 = db_connect("127.0.0.1", 6379, "plasma_manager", manager_addr, manager_port2); - db_attach(db1, loop); - db_attach(db2, loop); + db_attach(db1, loop, false); + db_attach(db2, loop, false); unique_id id = globally_unique_id(); retry_info retry = { .num_retries = NUM_RETRIES, @@ -137,7 +137,7 @@ TEST task_table_test(void) { task_table_test_callback_called = 0; event_loop *loop = event_loop_create(); db_handle *db = db_connect("127.0.0.1", 6379, "local_scheduler", "", -1); - db_attach(db, loop); + db_attach(db, loop, false); node_id node = globally_unique_id(); task_spec *spec = example_task_spec(); task_table_test_task = alloc_task(spec, TASK_STATUS_SCHEDULED, node); @@ -169,7 +169,7 @@ void task_table_all_test_callback(task *task, void *user_data) { TEST task_table_all_test(void) { event_loop *loop = event_loop_create(); db_handle *db = db_connect("127.0.0.1", 6379, "local_scheduler", "", -1); - db_attach(db, loop); + db_attach(db, loop, false); task_spec *spec = example_task_spec(); /* Schedule two tasks on different nodes. */ task *task1 = alloc_task(spec, TASK_STATUS_SCHEDULED, globally_unique_id()); diff --git a/src/common/test/object_table_tests.c b/src/common/test/object_table_tests.c index 365df0caa..4d54f2edd 100644 --- a/src/common/test/object_table_tests.c +++ b/src/common/test/object_table_tests.c @@ -73,7 +73,7 @@ TEST new_object_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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, @@ -120,7 +120,7 @@ TEST new_object_no_task_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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, @@ -162,7 +162,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, .fail_callback = lookup_fail_callback, }; @@ -198,7 +198,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, .fail_callback = add_fail_callback, }; @@ -219,7 +219,10 @@ TEST add_timeout_test(void) { const char *subscribe_timeout_context = "subscribe_timeout"; int subscribe_failed = 0; -void subscribe_done_callback(object_id object_id, void *user_context) { +void subscribe_done_callback(object_id object_id, + int manager_count, + const char *manager_vector[], + void *user_context) { /* The done callback should not be called. */ CHECK(0); } @@ -236,7 +239,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, @@ -268,7 +271,7 @@ int64_t reconnect_context_callback(event_loop *loop, 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); + db_attach(db, loop, true); return EVENT_LOOP_TIMER_DONE; } @@ -303,7 +306,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, @@ -349,7 +352,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, @@ -393,11 +396,14 @@ int64_t reconnect_sub_context_callback(event_loop *loop, 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); + db_attach(db, loop, true); return EVENT_LOOP_TIMER_DONE; } -void subscribe_retry_done_callback(object_id object_id, void *user_context) { +void subscribe_retry_done_callback(object_id object_id, + int manager_count, + const char *manager_vector[], + void *user_context) { CHECK(user_context == (void *) subscribe_retry_context); subscribe_retry_succeeded = 1; } @@ -413,7 +419,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, @@ -466,7 +472,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 0, .timeout = 0, @@ -508,7 +514,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 0, .timeout = 0, .fail_callback = add_late_fail_callback, }; @@ -541,7 +547,10 @@ void subscribe_late_fail_callback(unique_id id, subscribe_late_failed = 1; } -void subscribe_late_done_callback(object_id object_id, void *user_context) { +void subscribe_late_done_callback(object_id object_id, + int manager_count, + const char *manager_vector[], + void *user_context) { /* This function should never be called. */ CHECK(0); } @@ -550,7 +559,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 0, .timeout = 0, @@ -587,7 +596,10 @@ void subscribe_success_fail_callback(unique_id id, CHECK(0); } -void subscribe_success_done_callback(object_id object_id, void *user_context) { +void subscribe_success_done_callback(object_id object_id, + int manager_count, + const char *manager_vector[], + void *user_context) { retry_info retry = { .num_retries = 0, .timeout = 0, .fail_callback = NULL, }; @@ -596,6 +608,8 @@ void subscribe_success_done_callback(object_id object_id, void *user_context) { } void subscribe_success_object_available_callback(object_id object_id, + int manager_count, + const char *manager_vector[], void *user_context) { CHECK(user_context == (void *) subscribe_success_context); subscribe_success_succeeded = 1; @@ -605,7 +619,7 @@ 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); + db_attach(db, g_loop, false); unique_id id = globally_unique_id(); retry_info retry = { @@ -637,17 +651,21 @@ TEST subscribe_success_test(void) { const char *subscribe_object_present_context = "subscribe_object_present"; int subscribe_object_present_succeeded = 0; -void subscribe_object_present_object_available_callback(object_id object_id, - void *user_context) { +void subscribe_object_present_object_available_callback( + object_id object_id, + int manager_count, + const char *manager_vector[], + void *user_context) { CHECK(user_context == (void *) subscribe_object_present_context); subscribe_object_present_succeeded = 1; + CHECK(manager_count == 1); } TEST subscribe_object_present_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); + db_attach(db, g_loop, false); unique_id id = globally_unique_id(); retry_info retry = { .num_retries = 0, .timeout = 100, .fail_callback = NULL, @@ -678,6 +696,8 @@ int subscribe_object_not_present_succeeded = 0; void subscribe_object_not_present_object_available_callback( object_id object_id, + int manager_count, + const char *manager_vector[], void *user_context) { CHECK(user_context == (void *) subscribe_object_not_present_context); subscribe_object_not_present_succeeded = 1; @@ -687,7 +707,7 @@ TEST subscribe_object_not_present_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); + db_attach(db, g_loop, false); unique_id id = globally_unique_id(); retry_info retry = { .num_retries = 0, .timeout = 100, .fail_callback = NULL, @@ -717,10 +737,13 @@ int subscribe_object_available_later_succeeded = 0; void subscribe_object_available_later_object_available_callback( object_id object_id, + int manager_count, + const char *manager_vector[], void *user_context) { CHECK(user_context == (void *) subscribe_object_available_later_context); /* Make sure the callback is only called once. */ subscribe_object_available_later_succeeded += 1; + CHECK(manager_count == 1); } int64_t add_object_callback(event_loop *loop, int64_t timer_id, void *context) { @@ -737,7 +760,7 @@ TEST subscribe_object_available_later_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); + db_attach(db, g_loop, false); unique_id id = NIL_ID; retry_info retry = { .num_retries = 0, .timeout = 100, .fail_callback = NULL, @@ -759,7 +782,7 @@ TEST subscribe_object_available_later_test(void) { db_disconnect(db); destroy_outstanding_callbacks(g_loop); event_loop_destroy(g_loop); - ASSERT(subscribe_object_available_later_succeeded == 1); + ASSERT_EQ(subscribe_object_available_later_succeeded, 1); PASS(); } diff --git a/src/common/test/redis_tests.c b/src/common/test/redis_tests.c index 2e271b5bd..08bf0fc66 100644 --- a/src/common/test/redis_tests.c +++ b/src/common/test/redis_tests.c @@ -103,7 +103,7 @@ TEST async_redis_socket_test(void) { /* Start connection to Redis. */ db_handle *db = db_connect("127.0.0.1", 6379, "", "", 0); - db_attach(db, loop); + db_attach(db, loop, false); /* Send a command to the Redis process. */ int client_fd = connect_ipc_sock(socket_pathname); @@ -177,7 +177,7 @@ TEST logging_test(void) { /* Start connection to Redis. */ db_handle *conn = db_connect("127.0.0.1", 6379, "", "", 0); - db_attach(conn, loop); + db_attach(conn, loop, false); /* Send a command to the Redis process. */ int client_fd = connect_ipc_sock(socket_pathname); diff --git a/src/common/test/task_table_tests.c b/src/common/test/task_table_tests.c index e82168cb8..37e87267c 100644 --- a/src/common/test/task_table_tests.c +++ b/src/common/test/task_table_tests.c @@ -40,7 +40,7 @@ TEST lookup_nil_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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 1000, @@ -96,7 +96,7 @@ TEST add_lookup_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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 1000, @@ -138,7 +138,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, @@ -179,7 +179,7 @@ TEST publish_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); + db_attach(db, g_loop, false); task *task = example_task(); retry_info retry = { .num_retries = 5, .timeout = 100, .fail_callback = publish_fail_callback, @@ -208,7 +208,7 @@ int64_t reconnect_db_callback(event_loop *loop, 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); + db_attach(db, loop, true); return EVENT_LOOP_TIMER_DONE; } @@ -241,7 +241,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 5, .timeout = 100, @@ -288,7 +288,7 @@ TEST publish_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); + db_attach(db, g_loop, false); task *task = example_task(); retry_info retry = { .num_retries = 5, @@ -337,7 +337,7 @@ 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); + db_attach(db, g_loop, false); retry_info retry = { .num_retries = 0, .timeout = 0, @@ -382,7 +382,7 @@ TEST publish_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); + db_attach(db, g_loop, false); task *task = example_task(); retry_info retry = { .num_retries = 0, diff --git a/src/global_scheduler/global_scheduler.c b/src/global_scheduler/global_scheduler.c index 63834746f..40e3269ac 100644 --- a/src/global_scheduler/global_scheduler.c +++ b/src/global_scheduler/global_scheduler.c @@ -31,7 +31,7 @@ global_scheduler_state *init_global_scheduler(event_loop *loop, int redis_port) { global_scheduler_state *state = malloc(sizeof(global_scheduler_state)); state->db = db_connect(redis_addr, redis_port, "global_scheduler", "", -1); - db_attach(state->db, loop); + db_attach(state->db, loop, false); utarray_new(state->local_schedulers, &local_scheduler_icd); return state; } diff --git a/src/photon/photon_scheduler.c b/src/photon/photon_scheduler.c index 9d9838fbe..e9ba5f1e5 100644 --- a/src/photon/photon_scheduler.c +++ b/src/photon/photon_scheduler.c @@ -37,7 +37,7 @@ local_scheduler_state *init_local_scheduler( /* Connect to Redis if a Redis address is provided. */ if (redis_addr != NULL) { state->db = db_connect(redis_addr, redis_port, "photon", "", -1); - db_attach(state->db, loop); + db_attach(state->db, loop, false); } else { state->db = NULL; } diff --git a/src/plasma/plasma_manager.c b/src/plasma/plasma_manager.c index bdf64acaf..5d311a9de 100644 --- a/src/plasma/plasma_manager.c +++ b/src/plasma/plasma_manager.c @@ -408,7 +408,7 @@ plasma_manager_state *init_plasma_manager_state(const char *store_socket_name, if (db_addr) { state->db = db_connect(db_addr, db_port, "plasma_manager", manager_addr, manager_port); - db_attach(state->db, state->loop); + db_attach(state->db, state->loop, false); } else { state->db = NULL; LOG_DEBUG("No db connection specified"); @@ -1223,16 +1223,21 @@ void process_wait_request1(client_connection *client_conn, } } +/* TODO(pcm): unify with wait_object_available_callback. */ void wait_object_lookup_callback(object_id object_id, int manager_count, const char *manager_vector[], void *context) { if (manager_count > 0) { - wait_object_available_callback(object_id, context); + wait_object_available_callback(object_id, manager_count, manager_vector, + context); } } -void wait_object_available_callback(object_id object_id, void *user_context) { +void wait_object_available_callback(object_id object_id, + int manager_count, + const char *manager_vector[], + void *user_context) { client_connection *client_conn = (client_connection *) user_context; CHECK(client_conn != NULL); plasma_manager_state *manager_state = client_conn->manager_state; diff --git a/src/plasma/plasma_manager.h b/src/plasma/plasma_manager.h index d6ac7b716..a8b63c5ef 100644 --- a/src/plasma/plasma_manager.h +++ b/src/plasma/plasma_manager.h @@ -354,7 +354,10 @@ void process_wait_request1(client_connection *client_conn, * called. * @return Void. */ -void wait_object_available_callback(object_id object_id, void *user_context); +void wait_object_available_callback(object_id object_id, + int manager_count, + const char *manager_vector[], + void *user_context); /** * Object is available (sealed) in the local Object Store. This is part of