diff --git a/src/common/state/db_client_table.c b/src/common/state/db_client_table.c index 5a706c756..401cbc70e 100644 --- a/src/common/state/db_client_table.c +++ b/src/common/state/db_client_table.c @@ -10,7 +10,6 @@ void db_client_table_subscribe( void *user_context) { db_client_table_subscribe_data *sub_data = malloc(sizeof(db_client_table_subscribe_data)); - utarray_push_back(db_handle->callback_freelist, &sub_data); sub_data->subscribe_callback = subscribe_callback; sub_data->subscribe_context = subscribe_context; diff --git a/src/common/state/object_table.c b/src/common/state/object_table.c index 108011745..ec5eda9b1 100644 --- a/src/common/state/object_table.c +++ b/src/common/state/object_table.c @@ -32,7 +32,6 @@ void object_table_subscribe( CHECK(db_handle != NULL); 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; diff --git a/src/common/state/redis.c b/src/common/state/redis.c index 3e59ff26c..951b660d4 100644 --- a/src/common/state/redis.c +++ b/src/common/state/redis.c @@ -91,7 +91,6 @@ db_handle *db_connect(const char *address, db->client = client; db->db_client_cache = NULL; db->sync_context = context; - utarray_new(db->callback_freelist, &ut_ptr_icd); /* Establish async connection */ db->context = redisAsyncConnect(address, port); @@ -118,11 +117,6 @@ void db_disconnect(db_handle *db) { free(e); } free(db->client_type); - void **p = NULL; - while ((p = (void **) utarray_next(db->callback_freelist, p))) { - free(*p); - } - utarray_free(db->callback_freelist); free(db); } @@ -243,8 +237,6 @@ void redis_result_table_add_callback(redisAsyncContext *c, result_table_done_callback done_callback = callback_data->done_callback; done_callback(callback_data->id, callback_data->user_context); } - task_id *task_id = callback_data->data; - free(task_id); destroy_timer_callback(db->loop, callback_data); } @@ -283,7 +275,6 @@ void redis_result_table_lookup_task_callback(redisAsyncContext *c, done_callback(callback_data->id, task_reply, callback_data->user_context); free_task(task_reply); } - free(result_task_id); destroy_timer_callback(db->loop, callback_data); } diff --git a/src/common/state/redis.h b/src/common/state/redis.h index ee01ddc86..ff8c32e1e 100644 --- a/src/common/state/redis.h +++ b/src/common/state/redis.h @@ -45,8 +45,6 @@ struct db_handle { /** Redis context for synchronous connections. This should only be used very * rarely, it is not asynchronous. */ redisContext *sync_context; - /** Data structure for callbacks that needs to be freed. */ - UT_array *callback_freelist; }; void redis_object_table_get_entry(redisAsyncContext *c, diff --git a/src/common/state/table.c b/src/common/state/table.c index 973659af5..b3cf6f85c 100644 --- a/src/common/state/table.c +++ b/src/common/state/table.c @@ -6,7 +6,7 @@ table_callback_data *init_table_callback(db_handle *db_handle, unique_id id, const char *label, - void *data, + OWNER void *data, retry_info *retry, table_done_callback done_callback, table_retry_callback retry_callback, @@ -51,6 +51,11 @@ void destroy_table_callback(table_callback_data *callback_data) { if (callback_data->requests_info) free(callback_data->requests_info); + if (callback_data->data) { + free(callback_data->data); + callback_data->data = NULL; + } + outstanding_callbacks_remove(callback_data); /* Timer is removed via EVENT_LOOP_TIMER_DONE in the timeout callback. */ diff --git a/src/common/state/table.h b/src/common/state/table.h index 4cbe389db..8aee2d243 100644 --- a/src/common/state/table.h +++ b/src/common/state/table.h @@ -55,8 +55,8 @@ struct table_callback_data { */ retry_info retry; /** Pointer to the data that is entered into the table. This can be used to - * pass the result of the call to the callback. The user is responsible for - * freeing data in both the fail_callback and done_callback. */ + * pass the result of the call to the callback. The callback takes ownership + * over this data and will free it. */ void *data; /** Pointer to the data used internally to handle multiple database requests. */ @@ -103,7 +103,7 @@ int64_t table_timeout_handler(event_loop *loop, table_callback_data *init_table_callback(db_handle *db_handle, unique_id id, const char *label, - void *data, + OWNER void *data, retry_info *retry, table_done_callback done_callback, table_retry_callback retry_callback, diff --git a/src/common/state/task_table.c b/src/common/state/task_table.c index 614daf1d6..e139662d9 100644 --- a/src/common/state/task_table.c +++ b/src/common/state/task_table.c @@ -13,7 +13,7 @@ void task_table_get_task(db_handle *db_handle, } void task_table_add_task(db_handle *db_handle, - task *task, + OWNER task *task, retry_info *retry, task_table_done_callback done_callback, void *user_context) { @@ -22,7 +22,7 @@ void task_table_add_task(db_handle *db_handle, } void task_table_update(db_handle *db_handle, - task *task, + OWNER task *task, retry_info *retry, task_table_done_callback done_callback, void *user_context) { @@ -41,7 +41,6 @@ void task_table_subscribe(db_handle *db_handle, void *user_context) { task_table_subscribe_data *sub_data = malloc(sizeof(task_table_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; diff --git a/src/common/state/task_table.h b/src/common/state/task_table.h index 576481eb1..c7b788a57 100644 --- a/src/common/state/task_table.h +++ b/src/common/state/task_table.h @@ -54,7 +54,7 @@ void task_table_get_task(db_handle *db, * @return Void. */ void task_table_add_task(db_handle *db_handle, - task *task, + OWNER task *task, retry_info *retry, task_table_done_callback done_callback, void *user_context); @@ -77,7 +77,7 @@ void task_table_add_task(db_handle *db_handle, * @return Void. */ void task_table_update(db_handle *db_handle, - task *task, + OWNER task *task, retry_info *retry, task_table_done_callback done_callback, void *user_context); diff --git a/src/common/task.c b/src/common/task.c index cbc7aca1d..949cbf878 100644 --- a/src/common/task.c +++ b/src/common/task.c @@ -305,6 +305,14 @@ task *alloc_task(task_spec *spec, scheduling_state state, node_id node) { return result; } +task *copy_task(task *other) { + int64_t size = task_size(other); + task *copy = malloc(size); + CHECK(copy != NULL); + memcpy(copy, other, size); + return copy; +} + task *alloc_nil_task(task_id task_id) { task_spec *nil_spec = alloc_nil_task_spec(task_id); task *nil_task = alloc_task(nil_spec, 0, NIL_ID); diff --git a/src/common/task.h b/src/common/task.h index ef5aee2bc..757cb43a3 100644 --- a/src/common/task.h +++ b/src/common/task.h @@ -285,6 +285,14 @@ typedef struct task_impl task; */ task *alloc_task(task_spec *spec, scheduling_state state, node_id node); +/** + * Create a copy of the task. Must be freed with free_task after use. + * + * @param other The task that will be copied. + * @returns Pointer to the copy of the task. + */ +task *copy_task(task *other); + /** Size of task structure in bytes. */ int64_t task_size(task *task); diff --git a/src/common/test/db_tests.c b/src/common/test/db_tests.c index ed3cb9662..5b818f152 100644 --- a/src/common/test/db_tests.c +++ b/src/common/test/db_tests.c @@ -154,7 +154,6 @@ TEST task_table_test(void) { event_loop_add_timer( loop, 200, (event_loop_timer_handler) task_table_delayed_add_task, db); event_loop_run(loop); - free_task(task_table_test_task); db_disconnect(db); destroy_outstanding_callbacks(loop); event_loop_destroy(loop); @@ -190,8 +189,6 @@ TEST task_table_all_test(void) { event_loop_add_timer(loop, 200, (event_loop_timer_handler) timeout_handler, NULL); event_loop_run(loop); - free(task2); - free(task1); free_task_spec(spec); db_disconnect(db); destroy_outstanding_callbacks(loop); diff --git a/src/common/test/object_table_tests.c b/src/common/test/object_table_tests.c index dbf579dbd..9196e2ba6 100644 --- a/src/common/test/object_table_tests.c +++ b/src/common/test/object_table_tests.c @@ -79,13 +79,12 @@ TEST new_object_test(void) { .timeout = 100, .fail_callback = new_object_fail_callback, }; - task_table_add_task(db, new_object_task, &retry, new_object_task_callback, - db); + task_table_add_task(db, copy_task(new_object_task), &retry, + new_object_task_callback, db); event_loop_run(g_loop); db_disconnect(db); destroy_outstanding_callbacks(g_loop); event_loop_destroy(g_loop); - free_task(new_object_task); ASSERT(new_object_succeeded); ASSERT(!new_object_failed); PASS(); diff --git a/src/common/test/task_table_tests.c b/src/common/test/task_table_tests.c index aa6890376..e82168cb8 100644 --- a/src/common/test/task_table_tests.c +++ b/src/common/test/task_table_tests.c @@ -102,14 +102,13 @@ TEST add_lookup_test(void) { .timeout = 1000, .fail_callback = add_lookup_fail_callback, }; - task_table_add_task(db, add_lookup_task, &retry, add_success_callback, - (void *) db); + task_table_add_task(db, copy_task(add_lookup_task), &retry, + add_success_callback, (void *) db); /* Disconnect the database to see if the lookup times out. */ event_loop_run(g_loop); db_disconnect(db); destroy_outstanding_callbacks(g_loop); event_loop_destroy(g_loop); - free(add_lookup_task); ASSERT(add_success); ASSERT(lookup_success); PASS(); @@ -195,7 +194,6 @@ TEST publish_timeout_test(void) { destroy_outstanding_callbacks(g_loop); event_loop_destroy(g_loop); ASSERT(publish_failed); - free_task(task); PASS(); } @@ -313,7 +311,6 @@ TEST publish_retry_test(void) { destroy_outstanding_callbacks(g_loop); event_loop_destroy(g_loop); ASSERT(publish_retry_succeeded); - free_task(task); PASS(); } @@ -406,19 +403,18 @@ TEST publish_late_test(void) { destroy_outstanding_callbacks(g_loop); event_loop_destroy(g_loop); ASSERT(publish_late_failed); - free_task(task); PASS(); } SUITE(task_table_tests) { RUN_REDIS_TEST(lookup_nil_test); RUN_REDIS_TEST(add_lookup_test); - 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); + RUN_REDIS_TEST(subscribe_timeout_test); + RUN_REDIS_TEST(publish_timeout_test); + RUN_REDIS_TEST(subscribe_retry_test); + RUN_REDIS_TEST(publish_retry_test); + RUN_REDIS_TEST(subscribe_late_test); + RUN_REDIS_TEST(publish_late_test); } GREATEST_MAIN_DEFS(); diff --git a/src/global_scheduler/global_scheduler.c b/src/global_scheduler/global_scheduler.c index e99482161..63834746f 100644 --- a/src/global_scheduler/global_scheduler.c +++ b/src/global_scheduler/global_scheduler.c @@ -23,7 +23,7 @@ void assign_task_to_local_scheduler(global_scheduler_state *state, retry_info retry = { .num_retries = 0, .timeout = 100, .fail_callback = NULL, }; - task_table_update(state->db, task, &retry, NULL, NULL); + task_table_update(state->db, copy_task(task), &retry, NULL, NULL); } global_scheduler_state *init_global_scheduler(event_loop *loop, diff --git a/src/photon/photon_algorithm.c b/src/photon/photon_algorithm.c index 724bfe30f..e5eab90e1 100644 --- a/src/photon/photon_algorithm.c +++ b/src/photon/photon_algorithm.c @@ -160,7 +160,6 @@ void give_task_to_global_scheduler(scheduler_info *info, task *task = alloc_task(spec, TASK_STATUS_WAITING, NIL_ID); DCHECK(info->db != NULL); task_table_add_task(info->db, task, (retry_info *) &photon_retry, NULL, NULL); - free_task(task); } void handle_task_submitted(scheduler_info *info, diff --git a/src/photon/photon_scheduler.c b/src/photon/photon_scheduler.c index 60726ee36..c1f7c14a8 100644 --- a/src/photon/photon_scheduler.c +++ b/src/photon/photon_scheduler.c @@ -122,7 +122,6 @@ void assign_task_to_worker(scheduler_info *info, } else { task_table_add_task(info->db, task, (retry_info *) &retry, NULL, NULL); } - free_task(task); } }