wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-17 11:32:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

Define a wrapper class for callback_data.data (#1301)

Browse Source
This commit is contained in:
Stephanie Wang
2017-12-08 11:48:21 -08:00
committed by GitHub
parent 5adbdfecd0
commit bac39a134e
10 changed files with 146 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/state/actor_notification_table.cc
+2 -1
View File
@@ -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);
}
src/common/state/db_client_table.cc
+5 -3
View File
@@ -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);
}
src/common/state/driver_table.cc
+4 -2
View File
@@ -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);
}
src/common/state/error_table.cc
+2 -2
View File
@@ -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);
}
src/common/state/local_scheduler_table.cc
+4 -3
View File
@@ -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) {
src/common/state/object_table.cc
+18 -11
View File
@@ -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);
}
src/common/state/redis.cc
+30 -26
View File
@@ -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];
src/common/state/table.cc
+32 -7
View File
@@ -4,6 +4,30 @@
#include <inttypes.h>
#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;
src/common/state/table.h
+38 -3
View File
@@ -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,
src/common/state/task_table.cc
+11 -7
View File
@@ -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);
}