Init_table_callback now takes ownership of passed in data (#80)

* temp commit

* Stuff

* Ownership is now taken by init table callback

* Fixed lint errors

* Fixed travis warnings

* Fixed spacing

* add .gitkeep

* fix global scheduler

* Whitespace.

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;
 

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;
 

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);
 }
 

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,

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. */

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,

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;

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);

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);

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);
 

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);

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();

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();

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,

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,

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);
   }
 }