wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-08-03 13:10:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add a callback for gcs table lookup failures. (#1702)

Browse Source 
* Add callback to gcs client for table lookup failures.

* update plasma_manager reflecting changes to gcs callback.
This commit is contained in:
Melih Elibol
2018-03-15 22:25:01 -07:00
committed by GitHub
parent 1f027344f1
commit 3c080f4baa
5 changed files with 46 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/redis_module/ray_redis_module.cc
+3
View File
@@ -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);
src/plasma/plasma_manager.cc
+3
View File
@@ -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
}
src/ray/gcs/client_test.cc
+20 -4
View File
@@ -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();
@@ -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) {
@@ -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;
src/ray/gcs/redis_context.cc
+2
View File
@@ -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) {
@@ -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];
}
src/ray/gcs/tables.h
+18 -9
View File
@@ -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;
@@ -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) {
@@ -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;
@@ -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()) {