Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add a callback for gcs table lookup failures. #1702

Merged
merged 6 commits into from
Mar 16, 2018
Merged

Add a callback for gcs table lookup failures. #1702

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
27cf34d
Add callback to gcs client for table lookup failures.
elibol Mar 12, 2018
5100de2
linting
elibol Mar 13, 2018
07f641c
linting
elibol Mar 13, 2018
c59a065
updates to plasma_manager reflecting changes to gcs callback.
elibol Mar 15, 2018
f1f7aa7
respond with blank string and remove is_nil.
elibol Mar 16, 2018
a099e6b
linting
elibol Mar 16, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/common/redis_module/ray_redis_module.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -524,6 +524,9 @@ int TableLookup_RedisCommand(RedisModuleCtx *ctx,
RedisModuleString *id = argv[2];

RedisModuleKey *key = OpenPrefixedKey(ctx, "T:", id, REDISMODULE_READ);
if (key == nullptr) {
return RedisModule_ReplyWithNull(ctx);
}
size_t len = 0;
const char *buf = RedisModule_StringDMA(key, &len, REDISMODULE_READ);

Expand Down
3 changes: 3 additions & 0 deletions src/plasma/plasma_manager.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1335,6 +1335,9 @@ void log_object_hash_mismatch_error_result_callback(ObjectID object_id,
DBClientID::from_binary(t->scheduler_id), std::vector<ObjectID>());
log_object_hash_mismatch_error_task_callback(task, user_context);
Task_free(task);
},
[user_context](gcs::AsyncGcsClient *, const TaskID &) {
// TODO(pcmoritz): Handle failure.
}));
#endif
}
Expand Down
24 changes: 20 additions & 4 deletions src/ray/gcs/client_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,13 +98,19 @@ void Lookup(gcs::AsyncGcsClient *client, const UniqueID &id,
test->Stop();
}

void LookupFailed(gcs::AsyncGcsClient *client, const UniqueID &id) {
// Object entry failed.
RAY_CHECK(false);
test->Stop();
}

void TestObjectTable(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> client) {
auto data = std::make_shared<ObjectTableDataT>();
data->managers.push_back("A");
data->managers.push_back("B");
ObjectID object_id = ObjectID::from_random();
RAY_CHECK_OK(client->object_table().Add(job_id, object_id, data, &ObjectAdded));
RAY_CHECK_OK(client->object_table().Lookup(job_id, object_id, &Lookup));
RAY_CHECK_OK(client->object_table().Lookup(job_id, object_id, &Lookup, &LookupFailed));
// Run the event loop. The loop will only stop if the Lookup callback is
// called (or an assertion failure).
test->Start();
Expand All @@ -130,16 +136,25 @@ void TaskLookup(gcs::AsyncGcsClient *client, const TaskID &id,
ASSERT_EQ(data->scheduling_state, SchedulingState_SCHEDULED);
}

void TaskLookupFailure(gcs::AsyncGcsClient *client, const TaskID &id) {
RAY_CHECK(false);
}

void TaskLookupAfterUpdate(gcs::AsyncGcsClient *client, const TaskID &id,
std::shared_ptr<TaskTableDataT> data) {
ASSERT_EQ(data->scheduling_state, SchedulingState_LOST);
test->Stop();
}

void TaskLookupAfterUpdateFailure(gcs::AsyncGcsClient *client, const TaskID &id) {
RAY_CHECK(false);
test->Stop();
}

void TaskUpdateCallback(gcs::AsyncGcsClient *client, const TaskID &task_id,
const TaskTableDataT &task, bool updated) {
RAY_CHECK_OK(
client->task_table().Lookup(DriverID::nil(), task_id, &TaskLookupAfterUpdate));
RAY_CHECK_OK(client->task_table().Lookup(
DriverID::nil(), task_id, &TaskLookupAfterUpdate, &TaskLookupAfterUpdateFailure));
}

void TestTaskTable(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> client) {
Expand All @@ -149,7 +164,8 @@ void TestTaskTable(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> cli
data->scheduler_id = local_scheduler_id.binary();
TaskID task_id = TaskID::from_random();
RAY_CHECK_OK(client->task_table().Add(job_id, task_id, data, &TaskAdded));
RAY_CHECK_OK(client->task_table().Lookup(job_id, task_id, &TaskLookup));
RAY_CHECK_OK(
client->task_table().Lookup(job_id, task_id, &TaskLookup, &TaskLookupFailure));
auto update = std::make_shared<TaskTableTestAndUpdateT>();
update->test_scheduler_id = local_scheduler_id.binary();
update->test_state_bitmask = SchedulingState_SCHEDULED;
Expand Down
2 changes: 2 additions & 0 deletions src/ray/gcs/redis_context.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ void GlobalRedisCallback(void *c, void *r, void *privdata) {
redisReply *reply = reinterpret_cast<redisReply *>(r);
std::string data = "";
if (reply->type == REDIS_REPLY_NIL) {
// Respond with blank string, which triggers a failure callback for lookups.
} else if (reply->type == REDIS_REPLY_STRING) {
data = std::string(reply->str, reply->len);
} else if (reply->type == REDIS_REPLY_ARRAY) {
Expand Down Expand Up @@ -84,6 +85,7 @@ int64_t RedisCallbackManager::add(const RedisCallback &function) {

RedisCallbackManager::RedisCallback &RedisCallbackManager::get(
int64_t callback_index) {
RAY_CHECK(callbacks_.find(callback_index) != callbacks_.end());
return *callbacks_[callback_index];
}

Expand Down
27 changes: 18 additions & 9 deletions src/ray/gcs/tables.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,13 @@ class Table {
using DataT = typename Data::NativeTableType;
using Callback = std::function<void(AsyncGcsClient *client, const ID &id,
std::shared_ptr<DataT> data)>;
using FailureCallback = std::function<void(AsyncGcsClient *client, const ID &id)>;

struct CallbackData {
ID id;
std::shared_ptr<DataT> data;
Callback callback;
FailureCallback failure;
// An optional callback to call for subscription operations, where the
// first message is a notification of subscription success.
Callback subscription_callback;
Expand All @@ -58,7 +60,7 @@ class Table {
Status Add(const JobID &job_id, const ID &id, std::shared_ptr<DataT> data,
const Callback &done) {
auto d = std::shared_ptr<CallbackData>(
new CallbackData({id, data, done, nullptr, this, client_}));
new CallbackData({id, data, done, nullptr, nullptr, this, client_}));
int64_t callback_index =
RedisCallbackManager::instance().add([d](const std::string &data) {
if (d->callback != nullptr) {
Expand All @@ -79,16 +81,23 @@ class Table {
/// \param id The ID of the data that is looked up in the GCS.
/// \param lookup Callback that is called after lookup.
/// \return Status
Status Lookup(const JobID &job_id, const ID &id, const Callback &lookup) {
Status Lookup(const JobID &job_id, const ID &id, const Callback &lookup,
const FailureCallback &failure) {
auto d = std::shared_ptr<CallbackData>(
new CallbackData({id, nullptr, lookup, nullptr, this, client_}));
new CallbackData({id, nullptr, lookup, failure, nullptr, this, client_}));
int64_t callback_index =
RedisCallbackManager::instance().add([d](const std::string &data) {
auto result = std::make_shared<DataT>();
auto root = flatbuffers::GetRoot<Data>(data.data());
root->UnPackTo(result.get());
if (d->callback != nullptr) {
(d->callback)(d->client, d->id, result);
if (data.empty()) {
if (d->failure != nullptr) {
(d->failure)(d->client, d->id);
}
} else {
auto result = std::make_shared<DataT>();
auto root = flatbuffers::GetRoot<Data>(data.data());
root->UnPackTo(result.get());
if (d->callback != nullptr) {
(d->callback)(d->client, d->id, result);
}
}
});
std::vector<uint8_t> nil;
Expand All @@ -110,7 +119,7 @@ class Table {
Status Subscribe(const JobID &job_id, const ClientID &client_id,
const Callback &subscribe, const Callback &done) {
auto d = std::shared_ptr<CallbackData>(
new CallbackData({client_id, nullptr, subscribe, done, this, client_}));
new CallbackData({client_id, nullptr, subscribe, nullptr, done, this, client_}));
int64_t callback_index =
RedisCallbackManager::instance().add([d](const std::string &data) {
if (data.empty()) {
Expand Down