diff --git a/src/common/state/actor_notification_table.cc b/src/common/state/actor_notification_table.cc index 1d416365a..fdbcba913 100644 --- a/src/common/state/actor_notification_table.cc +++ b/src/common/state/actor_notification_table.cc @@ -12,7 +12,8 @@ void actor_notification_table_subscribe( sub_data->subscribe_callback = subscribe_callback; sub_data->subscribe_context = subscribe_context; - init_table_callback(db_handle, NIL_ID, __func__, sub_data, retry, NULL, + init_table_callback(db_handle, NIL_ID, __func__, + new CommonCallbackData(sub_data), retry, NULL, redis_actor_notification_table_subscribe, NULL); } diff --git a/src/common/state/db_client_table.cc b/src/common/state/db_client_table.cc index 0a01c3b59..03c5042ea 100644 --- a/src/common/state/db_client_table.cc +++ b/src/common/state/db_client_table.cc @@ -6,7 +6,8 @@ void db_client_table_remove(DBHandle *db_handle, RetryInfo *retry, db_client_table_done_callback done_callback, void *user_context) { - init_table_callback(db_handle, db_client_id, __func__, NULL, retry, + init_table_callback(db_handle, db_client_id, __func__, + new CommonCallbackData(NULL), retry, (table_done_callback) done_callback, redis_db_client_table_remove, user_context); } @@ -23,7 +24,8 @@ void db_client_table_subscribe( sub_data->subscribe_callback = subscribe_callback; sub_data->subscribe_context = subscribe_context; - init_table_callback(db_handle, NIL_ID, __func__, sub_data, retry, + init_table_callback(db_handle, NIL_ID, __func__, + new CommonCallbackData(sub_data), retry, (table_done_callback) done_callback, redis_db_client_table_subscribe, user_context); } @@ -80,7 +82,7 @@ void plasma_manager_send_heartbeat(DBHandle *db_handle) { RayConfig::instance().heartbeat_timeout_milliseconds(); heartbeat_retry.fail_callback = NULL; - init_table_callback(db_handle, NIL_ID, __func__, NULL, + init_table_callback(db_handle, NIL_ID, __func__, new CommonCallbackData(NULL), (RetryInfo *) &heartbeat_retry, NULL, redis_plasma_manager_send_heartbeat, NULL); } diff --git a/src/common/state/driver_table.cc b/src/common/state/driver_table.cc index 9e951b805..d458c6d40 100644 --- a/src/common/state/driver_table.cc +++ b/src/common/state/driver_table.cc @@ -9,13 +9,15 @@ void driver_table_subscribe(DBHandle *db_handle, (DriverTableSubscribeData *) malloc(sizeof(DriverTableSubscribeData)); sub_data->subscribe_callback = subscribe_callback; sub_data->subscribe_context = subscribe_context; - init_table_callback(db_handle, NIL_ID, __func__, sub_data, retry, NULL, + init_table_callback(db_handle, NIL_ID, __func__, + new CommonCallbackData(sub_data), retry, NULL, redis_driver_table_subscribe, NULL); } void driver_table_send_driver_death(DBHandle *db_handle, WorkerID driver_id, RetryInfo *retry) { - init_table_callback(db_handle, driver_id, __func__, NULL, retry, NULL, + init_table_callback(db_handle, driver_id, __func__, + new CommonCallbackData(NULL), retry, NULL, redis_driver_table_send_driver_death, NULL); } diff --git a/src/common/state/error_table.cc b/src/common/state/error_table.cc index 516bd98b8..f7d30196e 100644 --- a/src/common/state/error_table.cc +++ b/src/common/state/error_table.cc @@ -26,6 +26,6 @@ void push_error(DBHandle *db_handle, UniqueID error_key = globally_unique_id(); memcpy(info->error_key, error_key.id, sizeof(info->error_key)); - init_table_callback(db_handle, NIL_ID, __func__, info, NULL, NULL, - redis_push_error, NULL); + init_table_callback(db_handle, NIL_ID, __func__, new CommonCallbackData(info), + NULL, NULL, redis_push_error, NULL); } diff --git a/src/common/state/local_scheduler_table.cc b/src/common/state/local_scheduler_table.cc index dfcc545a2..9ae058d36 100644 --- a/src/common/state/local_scheduler_table.cc +++ b/src/common/state/local_scheduler_table.cc @@ -13,7 +13,8 @@ void local_scheduler_table_subscribe( sub_data->subscribe_callback = subscribe_callback; sub_data->subscribe_context = subscribe_context; - init_table_callback(db_handle, NIL_ID, __func__, sub_data, retry, NULL, + init_table_callback(db_handle, NIL_ID, __func__, + new CommonCallbackData(sub_data), retry, NULL, redis_local_scheduler_table_subscribe, NULL); } @@ -36,8 +37,8 @@ void local_scheduler_table_send_info(DBHandle *db_handle, data->size = fbb.GetSize(); memcpy(&data->flatbuffer_data[0], fbb.GetBufferPointer(), fbb.GetSize()); - init_table_callback(db_handle, NIL_ID, __func__, data, retry, NULL, - redis_local_scheduler_table_send_info, NULL); + init_table_callback(db_handle, NIL_ID, __func__, new CommonCallbackData(data), + retry, NULL, redis_local_scheduler_table_send_info, NULL); } void local_scheduler_table_disconnect(DBHandle *db_handle) { diff --git a/src/common/state/object_table.cc b/src/common/state/object_table.cc index 7992f969f..0c6d55252 100644 --- a/src/common/state/object_table.cc +++ b/src/common/state/object_table.cc @@ -7,7 +7,8 @@ void object_table_lookup(DBHandle *db_handle, object_table_lookup_done_callback done_callback, void *user_context) { CHECK(db_handle != NULL); - init_table_callback(db_handle, object_id, __func__, NULL, retry, + init_table_callback(db_handle, object_id, __func__, + new CommonCallbackData(NULL), retry, (table_done_callback) done_callback, redis_object_table_lookup, user_context); } @@ -25,7 +26,8 @@ void object_table_add(DBHandle *db_handle, (ObjectTableAddData *) malloc(sizeof(ObjectTableAddData)); info->object_size = object_size; memcpy(&info->digest[0], digest, DIGEST_SIZE); - init_table_callback(db_handle, object_id, __func__, info, retry, + init_table_callback(db_handle, object_id, __func__, + new CommonCallbackData(info), retry, (table_done_callback) done_callback, redis_object_table_add, user_context); } @@ -43,7 +45,8 @@ void object_table_remove(DBHandle *db_handle, client_id_copy = (DBClientID *) malloc(sizeof(DBClientID)); *client_id_copy = *client_id; } - init_table_callback(db_handle, object_id, __func__, client_id_copy, retry, + init_table_callback(db_handle, object_id, __func__, + new CommonCallbackData(client_id_copy), retry, (table_done_callback) done_callback, redis_object_table_remove, user_context); } @@ -63,10 +66,10 @@ void object_table_subscribe_to_notifications( sub_data->subscribe_context = subscribe_context; sub_data->subscribe_all = subscribe_all; - init_table_callback(db_handle, NIL_OBJECT_ID, __func__, sub_data, retry, - (table_done_callback) done_callback, - redis_object_table_subscribe_to_notifications, - user_context); + init_table_callback( + db_handle, NIL_OBJECT_ID, __func__, new CommonCallbackData(sub_data), + retry, (table_done_callback) done_callback, + redis_object_table_subscribe_to_notifications, user_context); } void object_table_request_notifications(DBHandle *db_handle, @@ -82,7 +85,8 @@ void object_table_request_notifications(DBHandle *db_handle, data->num_object_ids = num_object_ids; memcpy(data->object_ids, object_ids, num_object_ids * sizeof(ObjectID)); - init_table_callback(db_handle, NIL_OBJECT_ID, __func__, data, retry, NULL, + init_table_callback(db_handle, NIL_OBJECT_ID, __func__, + new CommonCallbackData(data), retry, NULL, redis_object_table_request_notifications, NULL); } @@ -97,7 +101,8 @@ void object_info_subscribe(DBHandle *db_handle, sub_data->subscribe_callback = subscribe_callback; sub_data->subscribe_context = subscribe_context; - init_table_callback(db_handle, NIL_OBJECT_ID, __func__, sub_data, retry, + init_table_callback(db_handle, NIL_OBJECT_ID, __func__, + new CommonCallbackData(sub_data), retry, (table_done_callback) done_callback, redis_object_info_subscribe, user_context); } @@ -113,7 +118,8 @@ void result_table_add(DBHandle *db_handle, (ResultTableAddInfo *) malloc(sizeof(ResultTableAddInfo)); info->task_id = task_id; info->is_put = is_put; - init_table_callback(db_handle, object_id, __func__, info, retry, + init_table_callback(db_handle, object_id, __func__, + new CommonCallbackData(info), retry, (table_done_callback) done_callback, redis_result_table_add, user_context); } @@ -123,7 +129,8 @@ void result_table_lookup(DBHandle *db_handle, RetryInfo *retry, result_table_lookup_callback done_callback, void *user_context) { - init_table_callback(db_handle, object_id, __func__, NULL, retry, + init_table_callback(db_handle, object_id, __func__, + new CommonCallbackData(NULL), retry, (table_done_callback) done_callback, redis_result_table_lookup, user_context); } diff --git a/src/common/state/redis.cc b/src/common/state/redis.cc index 2bb909e08..ba4f06bf2 100644 --- a/src/common/state/redis.cc +++ b/src/common/state/redis.cc @@ -398,7 +398,7 @@ void redis_object_table_add_callback(redisAsyncContext *c, void redis_object_table_add(TableCallbackData *callback_data) { DBHandle *db = callback_data->db_handle; - ObjectTableAddData *info = (ObjectTableAddData *) callback_data->data; + ObjectTableAddData *info = (ObjectTableAddData *) callback_data->data->Get(); ObjectID obj_id = callback_data->id; int64_t object_size = info->object_size; unsigned char *digest = info->digest; @@ -446,7 +446,7 @@ void redis_object_table_remove(TableCallbackData *callback_data) { ObjectID obj_id = callback_data->id; /* If the caller provided a manager ID to delete, use it. Otherwise, use our * own client ID as the ID to delete. */ - DBClientID *client_id = (DBClientID *) callback_data->data; + DBClientID *client_id = (DBClientID *) callback_data->data->Get(); if (client_id == NULL) { client_id = &db->client; } @@ -502,7 +502,7 @@ void redis_result_table_add(TableCallbackData *callback_data) { CHECK(callback_data); DBHandle *db = callback_data->db_handle; ObjectID id = callback_data->id; - ResultTableAddInfo *info = (ResultTableAddInfo *) callback_data->data; + ResultTableAddInfo *info = (ResultTableAddInfo *) callback_data->data->Get(); int is_put = info->is_put ? 1 : 0; redisAsyncContext *context = get_redis_context(db, id); @@ -726,7 +726,7 @@ void object_table_redis_subscribe_to_notifications_callback( /* Call the subscribe callback. */ ObjectTableSubscribeData *data = - (ObjectTableSubscribeData *) callback_data->data; + (ObjectTableSubscribeData *) callback_data->data->Get(); if (data->object_available_callback) { data->object_available_callback(obj_id, data_size, manager_ids, data->subscribe_context); @@ -763,9 +763,10 @@ void redis_object_table_subscribe_to_notifications( * as the channel name so this channel is specific to this client. * TODO(rkn): * The channel name should probably be the client ID with some prefix. */ - CHECKM(callback_data->data != NULL, + CHECKM(callback_data->data->Get() != NULL, "Object table subscribe data passed as NULL."); - if (((ObjectTableSubscribeData *) (callback_data->data))->subscribe_all) { + if (((ObjectTableSubscribeData *) (callback_data->data->Get())) + ->subscribe_all) { /* Subscribe to the object broadcast channel. */ status = redisAsyncCommand( db->subscribe_contexts[i], @@ -806,7 +807,7 @@ void redis_object_table_request_notifications( DBHandle *db = callback_data->db_handle; ObjectTableRequestNotificationsData *request_data = - (ObjectTableRequestNotificationsData *) callback_data->data; + (ObjectTableRequestNotificationsData *) callback_data->data->Get(); int num_object_ids = request_data->num_object_ids; ObjectID *object_ids = request_data->object_ids; @@ -866,7 +867,7 @@ void redis_task_table_get_task_callback(redisAsyncContext *c, void redis_task_table_get_task(TableCallbackData *callback_data) { DBHandle *db = callback_data->db_handle; - CHECK(callback_data->data == NULL); + CHECK(callback_data->data->Get() == NULL); TaskID task_id = callback_data->id; redisAsyncContext *context = get_redis_context(db, task_id); @@ -894,8 +895,9 @@ void redis_task_table_add_task_callback(redisAsyncContext *c, strcmp(reply->str, "No subscribers received message.") == 0) { LOG_WARN("No subscribers received the task_table_add message."); if (callback_data->retry.fail_callback != NULL) { - callback_data->retry.fail_callback( - callback_data->id, callback_data->user_context, callback_data->data); + callback_data->retry.fail_callback(callback_data->id, + callback_data->user_context, + callback_data->data->Get()); } } else { CHECK(reply->type != REDIS_REPLY_ERROR); @@ -914,7 +916,7 @@ void redis_task_table_add_task_callback(redisAsyncContext *c, void redis_task_table_add_task(TableCallbackData *callback_data) { DBHandle *db = callback_data->db_handle; - Task *task = (Task *) callback_data->data; + Task *task = (Task *) callback_data->data->Get(); TaskID task_id = Task_task_id(task); DBClientID local_scheduler_id = Task_local_scheduler(task); redisAsyncContext *context = get_redis_context(db, task_id); @@ -947,8 +949,9 @@ void redis_task_table_update_callback(redisAsyncContext *c, strcmp(reply->str, "No subscribers received message.") == 0) { LOG_WARN("No subscribers received the task_table_update message."); if (callback_data->retry.fail_callback != NULL) { - callback_data->retry.fail_callback( - callback_data->id, callback_data->user_context, callback_data->data); + callback_data->retry.fail_callback(callback_data->id, + callback_data->user_context, + callback_data->data->Get()); } } else { CHECK(reply->type != REDIS_REPLY_ERROR); @@ -968,7 +971,7 @@ void redis_task_table_update_callback(redisAsyncContext *c, void redis_task_table_update(TableCallbackData *callback_data) { DBHandle *db = callback_data->db_handle; - Task *task = (Task *) callback_data->data; + Task *task = (Task *) callback_data->data->Get(); TaskID task_id = Task_task_id(task); redisAsyncContext *context = get_redis_context(db, task_id); DBClientID local_scheduler_id = Task_local_scheduler(task); @@ -1024,7 +1027,7 @@ void redis_task_table_test_and_update(TableCallbackData *callback_data) { TaskID task_id = callback_data->id; redisAsyncContext *context = get_redis_context(db, task_id); TaskTableTestAndUpdateData *update_data = - (TaskTableTestAndUpdateData *) callback_data->data; + (TaskTableTestAndUpdateData *) callback_data->data->Get(); int status; /* If the test local scheduler ID is NIL, then ignore it. */ @@ -1086,7 +1089,7 @@ void redis_task_table_subscribe_callback(redisAsyncContext *c, /* Call the subscribe callback if there is one. */ TaskTableSubscribeData *data = - (TaskTableSubscribeData *) callback_data->data; + (TaskTableSubscribeData *) callback_data->data->Get(); if (data->subscribe_callback != NULL) { data->subscribe_callback(task, data->subscribe_context); } @@ -1112,7 +1115,8 @@ void redis_task_table_subscribe_callback(redisAsyncContext *c, void redis_task_table_subscribe(TableCallbackData *callback_data) { DBHandle *db = callback_data->db_handle; - TaskTableSubscribeData *data = (TaskTableSubscribeData *) callback_data->data; + TaskTableSubscribeData *data = + (TaskTableSubscribeData *) callback_data->data->Get(); /* TASK_CHANNEL_PREFIX is defined in ray_redis_module.cc and must be kept in * sync with that file. */ const char *TASK_CHANNEL_PREFIX = "TT:"; @@ -1229,7 +1233,7 @@ void redis_db_client_table_subscribe_callback(redisAsyncContext *c, redis_db_client_table_scan(db, db_clients); /* Call the subscription callback for all entries that we missed. */ DBClientTableSubscribeData *data = - (DBClientTableSubscribeData *) callback_data->data; + (DBClientTableSubscribeData *) callback_data->data->Get(); for (auto db_client : db_clients) { data->subscribe_callback(&db_client, data->subscribe_context); } @@ -1249,7 +1253,7 @@ void redis_db_client_table_subscribe_callback(redisAsyncContext *c, /* Call the subscription callback. */ DBClientTableSubscribeData *data = - (DBClientTableSubscribeData *) callback_data->data; + (DBClientTableSubscribeData *) callback_data->data->Get(); if (data->subscribe_callback) { data->subscribe_callback(&db_client, data->subscribe_context); } @@ -1305,7 +1309,7 @@ void redis_local_scheduler_table_subscribe_callback(redisAsyncContext *c, /* Call the subscribe callback. */ LocalSchedulerTableSubscribeData *data = - (LocalSchedulerTableSubscribeData *) callback_data->data; + (LocalSchedulerTableSubscribeData *) callback_data->data->Get(); if (data->subscribe_callback) { data->subscribe_callback(client_id, info, data->subscribe_context); } @@ -1349,7 +1353,7 @@ void redis_local_scheduler_table_send_info_callback(redisAsyncContext *c, void redis_local_scheduler_table_send_info(TableCallbackData *callback_data) { DBHandle *db = callback_data->db_handle; LocalSchedulerTableSendInfoData *data = - (LocalSchedulerTableSendInfoData *) callback_data->data; + (LocalSchedulerTableSendInfoData *) callback_data->data->Get(); int64_t size = data->size; uint8_t *flatbuffer_data = data->flatbuffer_data; @@ -1406,7 +1410,7 @@ void redis_driver_table_subscribe_callback(redisAsyncContext *c, /* Call the subscribe callback. */ DriverTableSubscribeData *data = - (DriverTableSubscribeData *) callback_data->data; + (DriverTableSubscribeData *) callback_data->data->Get(); if (data->subscribe_callback) { data->subscribe_callback(driver_id, data->subscribe_context); } @@ -1504,7 +1508,7 @@ void redis_actor_notification_table_subscribe_callback(redisAsyncContext *c, * subscribe callback. */ redisReply *payload = reply->element[2]; ActorNotificationTableSubscribeData *data = - (ActorNotificationTableSubscribeData *) callback_data->data; + (ActorNotificationTableSubscribeData *) callback_data->data->Get(); /* The payload should be the concatenation of three IDs. */ ActorID actor_id; WorkerID driver_id; @@ -1597,7 +1601,7 @@ void redis_object_info_subscribe_callback(redisAsyncContext *c, } /* Otherwise, parse the payload and call the callback. */ ObjectInfoSubscribeData *data = - (ObjectInfoSubscribeData *) callback_data->data; + (ObjectInfoSubscribeData *) callback_data->data->Get(); ObjectID object_id; memcpy(object_id.id, payload->str, sizeof(object_id.id)); /* payload->str should have the format: "ObjectID:object_size_int" */ @@ -1641,7 +1645,7 @@ void redis_push_error_hmset_callback(redisAsyncContext *c, CHECKM(strcmp(reply->str, "OK") == 0, "reply->str is %s", reply->str); /* Add the error to this driver's list of errors. */ - ErrorInfo *info = (ErrorInfo *) callback_data->data; + ErrorInfo *info = (ErrorInfo *) callback_data->data->Get(); int status = redisAsyncCommand(db->context, redis_push_error_rpush_callback, (void *) callback_data->timer_id, "RPUSH ErrorKeys Error:%b:%b", @@ -1654,7 +1658,7 @@ void redis_push_error_hmset_callback(redisAsyncContext *c, void redis_push_error(TableCallbackData *callback_data) { DBHandle *db = callback_data->db_handle; - ErrorInfo *info = (ErrorInfo *) callback_data->data; + ErrorInfo *info = (ErrorInfo *) callback_data->data->Get(); CHECK(info->error_index < MAX_ERROR_INDEX && info->error_index >= 0); /* Look up the error type. */ const char *error_type = error_types[info->error_index]; diff --git a/src/common/state/table.cc b/src/common/state/table.cc index 6c678f051..a70eaa7bc 100644 --- a/src/common/state/table.cc +++ b/src/common/state/table.cc @@ -4,6 +4,30 @@ #include #include "redis.h" +BaseCallbackData::BaseCallbackData(void *data) { + data_ = data; +} + +BaseCallbackData::~BaseCallbackData(void) {} + +void *BaseCallbackData::Get(void) { + return data_; +} + +CommonCallbackData::CommonCallbackData(void *data) : BaseCallbackData(data) {} + +CommonCallbackData::~CommonCallbackData(void) { + free(data_); +} + +TaskCallbackData::TaskCallbackData(Task *task_data) + : BaseCallbackData(task_data) {} + +TaskCallbackData::~TaskCallbackData(void) { + Task *task = (Task *) data_; + Task_free(task); +} + /* The default behavior is to retry every ten seconds forever. */ static const RetryInfo default_retry = {.num_retries = -1, .timeout = 10000, @@ -14,13 +38,14 @@ static int64_t callback_data_id = 0; TableCallbackData *init_table_callback(DBHandle *db_handle, UniqueID id, const char *label, - OWNER void *data, + OWNER BaseCallbackData *data, RetryInfo *retry, table_done_callback done_callback, table_retry_callback retry_callback, void *user_context) { CHECK(db_handle); CHECK(db_handle->loop); + CHECK(data); /* If no retry info is provided, use the default retry info. */ if (retry == NULL) { retry = (RetryInfo *) &default_retry; @@ -72,10 +97,9 @@ void destroy_table_callback(TableCallbackData *callback_data) { if (callback_data->requests_info) free(callback_data->requests_info); - if (callback_data->data) { - free(callback_data->data); - callback_data->data = NULL; - } + CHECK(callback_data->data != NULL); + delete callback_data->data; + callback_data->data = NULL; outstanding_callbacks_remove(callback_data); @@ -101,8 +125,9 @@ int64_t table_timeout_handler(event_loop *loop, LOG_WARN("Table command %s with timer ID %" PRId64 " failed", callback_data->label, timer_id); if (callback_data->retry.fail_callback) { - callback_data->retry.fail_callback( - callback_data->id, callback_data->user_context, callback_data->data); + callback_data->retry.fail_callback(callback_data->id, + callback_data->user_context, + callback_data->data->Get()); } destroy_table_callback(callback_data); return EVENT_LOOP_TIMER_DONE; diff --git a/src/common/state/table.h b/src/common/state/table.h index 9674eb782..1fadcf339 100644 --- a/src/common/state/table.h +++ b/src/common/state/table.h @@ -2,10 +2,43 @@ #define TABLE_H #include "common.h" +#include "task.h" #include "db.h" typedef struct TableCallbackData TableCallbackData; +/* An abstract class for any data passed by the user into a table operation. + * This class wraps arbitrary pointers and allows the caller to define a custom + * destructor, for data that is not allocated with malloc. */ +class BaseCallbackData { + public: + BaseCallbackData(void *data); + virtual ~BaseCallbackData(void) = 0; + + /* Return the pointer to the data. */ + void *Get(void); + + protected: + /* The pointer to the data. */ + void *data_; +}; + +/* A common class for malloc'ed data passed by the user into a table operation. + * This should ONLY be used when only a free is necessary. */ +class CommonCallbackData : public BaseCallbackData { + public: + CommonCallbackData(void *data); + ~CommonCallbackData(void); +}; + +/* A class for Task data passed by the user into a table operation. This calls + * task cleanup in the destructor. */ +class TaskCallbackData : public BaseCallbackData { + public: + TaskCallbackData(Task *task_data); + ~TaskCallbackData(void); +}; + typedef void *table_done_callback; /* The callback called when the database operation hasn't completed after @@ -58,7 +91,7 @@ struct TableCallbackData { /** 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 callback takes ownership * over this data and will free it. */ - void *data; + BaseCallbackData *data; /** Pointer to the data used internally to handle multiple database requests. */ void *requests_info; @@ -90,7 +123,9 @@ int64_t table_timeout_handler(event_loop *loop, * @param id ID of the object that is looked up, added or removed. * @param label A string label to identify the type of table request for * logging purposes. - * @param data Data entered into the table. Shall be freed by the user. + * @param data Data entered into the table. Shall be freed by the user. Caller + * must specify a destructor by wrapping a void *pointer in a + * BaseCallbackData class. * @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. @@ -103,7 +138,7 @@ int64_t table_timeout_handler(event_loop *loop, TableCallbackData *init_table_callback(DBHandle *db_handle, UniqueID id, const char *label, - OWNER void *data, + OWNER BaseCallbackData *data, RetryInfo *retry, table_done_callback done_callback, table_retry_callback retry_callback, diff --git a/src/common/state/task_table.cc b/src/common/state/task_table.cc index 2c0d3faea..2a3d737d2 100644 --- a/src/common/state/task_table.cc +++ b/src/common/state/task_table.cc @@ -8,9 +8,9 @@ void task_table_get_task(DBHandle *db_handle, RetryInfo *retry, task_table_get_callback get_callback, void *user_context) { - init_table_callback(db_handle, task_id, __func__, NULL, retry, - (void *) get_callback, redis_task_table_get_task, - user_context); + init_table_callback( + db_handle, task_id, __func__, new CommonCallbackData(NULL), retry, + (void *) get_callback, redis_task_table_get_task, user_context); } void task_table_add_task(DBHandle *db_handle, @@ -18,7 +18,8 @@ void task_table_add_task(DBHandle *db_handle, RetryInfo *retry, task_table_done_callback done_callback, void *user_context) { - init_table_callback(db_handle, Task_task_id(task), __func__, task, retry, + init_table_callback(db_handle, Task_task_id(task), __func__, + new TaskCallbackData(task), retry, (table_done_callback) done_callback, redis_task_table_add_task, user_context); } @@ -28,7 +29,8 @@ void task_table_update(DBHandle *db_handle, RetryInfo *retry, task_table_done_callback done_callback, void *user_context) { - init_table_callback(db_handle, Task_task_id(task), __func__, task, retry, + init_table_callback(db_handle, Task_task_id(task), __func__, + new TaskCallbackData(task), retry, (table_done_callback) done_callback, redis_task_table_update, user_context); } @@ -49,7 +51,8 @@ void task_table_test_and_update( update_data->update_state = update_state; /* Update the task entry's local scheduler with this client's ID. */ update_data->local_scheduler_id = db_handle->client; - init_table_callback(db_handle, task_id, __func__, update_data, retry, + init_table_callback(db_handle, task_id, __func__, + new CommonCallbackData(update_data), retry, (table_done_callback) done_callback, redis_task_table_test_and_update, user_context); } @@ -70,7 +73,8 @@ void task_table_subscribe(DBHandle *db_handle, sub_data->subscribe_callback = subscribe_callback; sub_data->subscribe_context = subscribe_context; - init_table_callback(db_handle, local_scheduler_id, __func__, sub_data, retry, + init_table_callback(db_handle, local_scheduler_id, __func__, + new CommonCallbackData(sub_data), retry, (table_done_callback) done_callback, redis_task_table_subscribe, user_context); }