mirror of
https://github.com/wassname/ray.git
synced 2026-07-21 12:50:45 +08:00
remove store client template (#8160)
Co-authored-by: 灵洵 <fengbin.ffb@antfin.com>
This commit is contained in:
fangfengbin
@@ -23,24 +23,46 @@ namespace gcs {
|
||||
template <typename Key, typename Data>
|
||||
Status GcsTable<Key, Data>::Put(const Key &key, const Data &value,
|
||||
const StatusCallback &callback) {
|
||||
return store_client_->AsyncPut(table_name_, key, value, callback);
|
||||
return store_client_->AsyncPut(table_name_, key.Binary(), value.SerializeAsString(),
|
||||
callback);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data>
|
||||
Status GcsTable<Key, Data>::Get(const Key &key,
|
||||
const OptionalItemCallback<Data> &callback) {
|
||||
return store_client_->AsyncGet(table_name_, key, callback);
|
||||
auto on_done = [callback](const Status &status,
|
||||
const boost::optional<std::string> &result) {
|
||||
boost::optional<Data> value;
|
||||
if (result) {
|
||||
Data data;
|
||||
data.ParseFromString(*result);
|
||||
value = std::move(data);
|
||||
}
|
||||
callback(status, value);
|
||||
};
|
||||
return store_client_->AsyncGet(table_name_, key.Binary(), on_done);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data>
|
||||
Status GcsTable<Key, Data>::GetAll(
|
||||
const SegmentedCallback<std::pair<Key, Data>> &callback) {
|
||||
return store_client_->AsyncGetAll(table_name_, callback);
|
||||
auto on_done = [callback](
|
||||
const Status &status, bool has_more,
|
||||
const std::vector<std::pair<std::string, std::string>> &result) {
|
||||
std::vector<std::pair<Key, Data>> values;
|
||||
for (auto &item : result) {
|
||||
Data data;
|
||||
data.ParseFromString(item.second);
|
||||
values.emplace_back(std::move(std::make_pair(Key::FromBinary(item.first), data)));
|
||||
}
|
||||
callback(status, has_more, values);
|
||||
};
|
||||
return store_client_->AsyncGetAll(table_name_, on_done);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data>
|
||||
Status GcsTable<Key, Data>::Delete(const Key &key, const StatusCallback &callback) {
|
||||
return store_client_->AsyncDelete(table_name_, key, callback);
|
||||
return store_client_->AsyncDelete(table_name_, key.Binary(), callback);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data>
|
||||
@@ -50,13 +72,13 @@ Status GcsTable<Key, Data>::BatchDelete(const std::vector<Key> &keys,
|
||||
auto finished_count = std::make_shared<int>(0);
|
||||
int size = keys.size();
|
||||
for (Key key : keys) {
|
||||
auto done = [finished_count, size, callback](Status status) {
|
||||
auto done = [finished_count, size, callback](const Status &status) {
|
||||
++(*finished_count);
|
||||
if (*finished_count == size) {
|
||||
callback(Status::OK());
|
||||
}
|
||||
};
|
||||
RAY_CHECK_OK(store_client_->AsyncDelete(table_name_, key, done));
|
||||
RAY_CHECK_OK(store_client_->AsyncDelete(table_name_, key.Binary(), done));
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
@@ -64,8 +86,9 @@ Status GcsTable<Key, Data>::BatchDelete(const std::vector<Key> &keys,
|
||||
template <typename Key, typename Data>
|
||||
Status GcsTableWithJobId<Key, Data>::Put(const Key &key, const Data &value,
|
||||
const StatusCallback &callback) {
|
||||
return this->store_client_->AsyncPutWithIndex(this->table_name_, key,
|
||||
GetJobIdFromKey(key), value, callback);
|
||||
return this->store_client_->AsyncPutWithIndex(this->table_name_, key.Binary(),
|
||||
GetJobIdFromKey(key).Binary(),
|
||||
value.SerializeAsString(), callback);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data>
|
||||
@@ -79,7 +102,8 @@ Status GcsTableWithJobId<Key, Data>::GetByJobId(
|
||||
template <typename Key, typename Data>
|
||||
Status GcsTableWithJobId<Key, Data>::DeleteByJobId(const JobID &job_id,
|
||||
const StatusCallback &callback) {
|
||||
return this->store_client_->AsyncDeleteByIndex(this->table_name_, job_id, callback);
|
||||
return this->store_client_->AsyncDeleteByIndex(this->table_name_, job_id.Binary(),
|
||||
callback);
|
||||
}
|
||||
|
||||
template class GcsTable<JobID, JobTableData>;
|
||||
|
||||
@@ -50,7 +50,7 @@ using rpc::WorkerFailureData;
|
||||
template <typename Key, typename Data>
|
||||
class GcsTable {
|
||||
public:
|
||||
explicit GcsTable(std::shared_ptr<StoreClient<Key, Data, JobID>> store_client)
|
||||
explicit GcsTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: store_client_(store_client) {}
|
||||
|
||||
virtual ~GcsTable() = default;
|
||||
@@ -93,7 +93,7 @@ class GcsTable {
|
||||
|
||||
protected:
|
||||
std::string table_name_;
|
||||
std::shared_ptr<StoreClient<Key, Data, JobID>> store_client_;
|
||||
std::shared_ptr<StoreClient> store_client_;
|
||||
};
|
||||
|
||||
/// \class GcsTableWithJobId
|
||||
@@ -105,7 +105,7 @@ class GcsTable {
|
||||
template <typename Key, typename Data>
|
||||
class GcsTableWithJobId : public GcsTable<Key, Data> {
|
||||
public:
|
||||
explicit GcsTableWithJobId(std::shared_ptr<StoreClient<Key, Data, JobID>> store_client)
|
||||
explicit GcsTableWithJobId(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable<Key, Data>(store_client) {}
|
||||
|
||||
/// Write data to the table asynchronously.
|
||||
@@ -137,31 +137,27 @@ class GcsTableWithJobId : public GcsTable<Key, Data> {
|
||||
|
||||
class GcsJobTable : public GcsTable<JobID, JobTableData> {
|
||||
public:
|
||||
explicit GcsJobTable(
|
||||
std::shared_ptr<StoreClient<JobID, JobTableData, JobID>> store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsJobTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::JOB);
|
||||
}
|
||||
};
|
||||
|
||||
class GcsActorTable : public GcsTableWithJobId<ActorID, ActorTableData> {
|
||||
public:
|
||||
explicit GcsActorTable(
|
||||
std::shared_ptr<StoreClient<ActorID, ActorTableData, JobID>> store_client)
|
||||
: GcsTableWithJobId(std::move(store_client)) {
|
||||
explicit GcsActorTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTableWithJobId(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::ACTOR);
|
||||
}
|
||||
|
||||
private:
|
||||
JobID GetJobIdFromKey(const ActorID &key) { return key.JobId(); }
|
||||
JobID GetJobIdFromKey(const ActorID &key) override { return key.JobId(); }
|
||||
};
|
||||
|
||||
class GcsActorCheckpointTable : public GcsTable<ActorCheckpointID, ActorCheckpointData> {
|
||||
public:
|
||||
explicit GcsActorCheckpointTable(
|
||||
std::shared_ptr<StoreClient<ActorCheckpointID, ActorCheckpointData, JobID>>
|
||||
store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsActorCheckpointTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::ACTOR_CHECKPOINT);
|
||||
}
|
||||
};
|
||||
@@ -169,124 +165,112 @@ class GcsActorCheckpointTable : public GcsTable<ActorCheckpointID, ActorCheckpoi
|
||||
class GcsActorCheckpointIdTable
|
||||
: public GcsTableWithJobId<ActorID, ActorCheckpointIdData> {
|
||||
public:
|
||||
explicit GcsActorCheckpointIdTable(
|
||||
std::shared_ptr<StoreClient<ActorID, ActorCheckpointIdData, JobID>> store_client)
|
||||
: GcsTableWithJobId(std::move(store_client)) {
|
||||
explicit GcsActorCheckpointIdTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTableWithJobId(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::ACTOR_CHECKPOINT_ID);
|
||||
}
|
||||
|
||||
private:
|
||||
JobID GetJobIdFromKey(const ActorID &key) { return key.JobId(); }
|
||||
JobID GetJobIdFromKey(const ActorID &key) override { return key.JobId(); }
|
||||
};
|
||||
|
||||
class GcsTaskTable : public GcsTableWithJobId<TaskID, TaskTableData> {
|
||||
public:
|
||||
explicit GcsTaskTable(
|
||||
std::shared_ptr<StoreClient<TaskID, TaskTableData, JobID>> store_client)
|
||||
: GcsTableWithJobId(std::move(store_client)) {
|
||||
explicit GcsTaskTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTableWithJobId(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::TASK);
|
||||
}
|
||||
|
||||
private:
|
||||
JobID GetJobIdFromKey(const TaskID &key) { return key.ActorId().JobId(); }
|
||||
JobID GetJobIdFromKey(const TaskID &key) override { return key.ActorId().JobId(); }
|
||||
};
|
||||
|
||||
class GcsTaskLeaseTable : public GcsTableWithJobId<TaskID, TaskLeaseData> {
|
||||
public:
|
||||
explicit GcsTaskLeaseTable(
|
||||
std::shared_ptr<StoreClient<TaskID, TaskLeaseData, JobID>> store_client)
|
||||
: GcsTableWithJobId(std::move(store_client)) {
|
||||
explicit GcsTaskLeaseTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTableWithJobId(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::TASK_LEASE);
|
||||
}
|
||||
|
||||
private:
|
||||
JobID GetJobIdFromKey(const TaskID &key) { return key.ActorId().JobId(); }
|
||||
JobID GetJobIdFromKey(const TaskID &key) override { return key.ActorId().JobId(); }
|
||||
};
|
||||
|
||||
class GcsTaskReconstructionTable
|
||||
: public GcsTableWithJobId<TaskID, TaskReconstructionData> {
|
||||
public:
|
||||
explicit GcsTaskReconstructionTable(
|
||||
std::shared_ptr<StoreClient<TaskID, TaskReconstructionData, JobID>> store_client)
|
||||
: GcsTableWithJobId(std::move(store_client)) {
|
||||
explicit GcsTaskReconstructionTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTableWithJobId(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::TASK_RECONSTRUCTION);
|
||||
}
|
||||
|
||||
private:
|
||||
JobID GetJobIdFromKey(const TaskID &key) { return key.ActorId().JobId(); }
|
||||
JobID GetJobIdFromKey(const TaskID &key) override { return key.ActorId().JobId(); }
|
||||
};
|
||||
|
||||
class GcsObjectTable : public GcsTableWithJobId<ObjectID, ObjectTableDataList> {
|
||||
public:
|
||||
explicit GcsObjectTable(
|
||||
std::shared_ptr<StoreClient<ObjectID, ObjectTableDataList, JobID>> store_client)
|
||||
: GcsTableWithJobId(std::move(store_client)) {
|
||||
explicit GcsObjectTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTableWithJobId(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::OBJECT);
|
||||
}
|
||||
|
||||
private:
|
||||
JobID GetJobIdFromKey(const ObjectID &key) { return key.TaskId().JobId(); }
|
||||
JobID GetJobIdFromKey(const ObjectID &key) override { return key.TaskId().JobId(); }
|
||||
};
|
||||
|
||||
class GcsNodeTable : public GcsTable<ClientID, GcsNodeInfo> {
|
||||
public:
|
||||
explicit GcsNodeTable(
|
||||
std::shared_ptr<StoreClient<ClientID, GcsNodeInfo, JobID>> store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsNodeTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::CLIENT);
|
||||
}
|
||||
};
|
||||
|
||||
class GcsNodeResourceTable : public GcsTable<ClientID, ResourceMap> {
|
||||
public:
|
||||
explicit GcsNodeResourceTable(
|
||||
std::shared_ptr<StoreClient<ClientID, ResourceMap, JobID>> store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsNodeResourceTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::NODE_RESOURCE);
|
||||
}
|
||||
};
|
||||
|
||||
class GcsHeartbeatTable : public GcsTable<ClientID, HeartbeatTableData> {
|
||||
public:
|
||||
explicit GcsHeartbeatTable(
|
||||
std::shared_ptr<StoreClient<ClientID, HeartbeatTableData, JobID>> store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsHeartbeatTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::HEARTBEAT);
|
||||
}
|
||||
};
|
||||
|
||||
class GcsHeartbeatBatchTable : public GcsTable<ClientID, HeartbeatBatchTableData> {
|
||||
public:
|
||||
explicit GcsHeartbeatBatchTable(
|
||||
std::shared_ptr<StoreClient<ClientID, HeartbeatBatchTableData, JobID>> store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsHeartbeatBatchTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::HEARTBEAT_BATCH);
|
||||
}
|
||||
};
|
||||
|
||||
class GcsErrorInfoTable : public GcsTable<JobID, ErrorTableData> {
|
||||
public:
|
||||
explicit GcsErrorInfoTable(
|
||||
std::shared_ptr<StoreClient<JobID, ErrorTableData, JobID>> store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsErrorInfoTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::ERROR_INFO);
|
||||
}
|
||||
};
|
||||
|
||||
class GcsProfileTable : public GcsTable<UniqueID, ProfileTableData> {
|
||||
public:
|
||||
explicit GcsProfileTable(
|
||||
std::shared_ptr<StoreClient<UniqueID, ProfileTableData, JobID>> store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsProfileTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::PROFILE);
|
||||
}
|
||||
};
|
||||
|
||||
class GcsWorkerFailureTable : public GcsTable<WorkerID, WorkerFailureData> {
|
||||
public:
|
||||
explicit GcsWorkerFailureTable(
|
||||
std::shared_ptr<StoreClient<WorkerID, WorkerFailureData, JobID>> store_client)
|
||||
: GcsTable(std::move(store_client)) {
|
||||
explicit GcsWorkerFailureTable(std::shared_ptr<StoreClient> &store_client)
|
||||
: GcsTable(store_client) {
|
||||
table_name_ = TablePrefix_Name(TablePrefix::WORKER_FAILURE);
|
||||
}
|
||||
};
|
||||
@@ -373,6 +357,7 @@ class GcsTableStorage {
|
||||
}
|
||||
|
||||
protected:
|
||||
std::shared_ptr<StoreClient> store_client_;
|
||||
std::unique_ptr<GcsJobTable> job_table_;
|
||||
std::unique_ptr<GcsActorTable> actor_table_;
|
||||
std::unique_ptr<GcsActorCheckpointTable> actor_checkpoint_table_;
|
||||
@@ -396,45 +381,22 @@ class GcsTableStorage {
|
||||
class RedisGcsTableStorage : public GcsTableStorage {
|
||||
public:
|
||||
explicit RedisGcsTableStorage(std::shared_ptr<RedisClient> redis_client) {
|
||||
job_table_.reset(new GcsJobTable(
|
||||
std::make_shared<RedisStoreClient<JobID, JobTableData, JobID>>(redis_client)));
|
||||
actor_table_.reset(new GcsActorTable(
|
||||
std::make_shared<RedisStoreClient<ActorID, ActorTableData, JobID>>(
|
||||
redis_client)));
|
||||
actor_checkpoint_table_.reset(new GcsActorCheckpointTable(
|
||||
std::make_shared<RedisStoreClient<ActorCheckpointID, ActorCheckpointData, JobID>>(
|
||||
redis_client)));
|
||||
actor_checkpoint_id_table_.reset(new GcsActorCheckpointIdTable(
|
||||
std::make_shared<RedisStoreClient<ActorID, ActorCheckpointIdData, JobID>>(
|
||||
redis_client)));
|
||||
task_table_.reset(new GcsTaskTable(
|
||||
std::make_shared<RedisStoreClient<TaskID, TaskTableData, JobID>>(redis_client)));
|
||||
task_lease_table_.reset(new GcsTaskLeaseTable(
|
||||
std::make_shared<RedisStoreClient<TaskID, TaskLeaseData, JobID>>(redis_client)));
|
||||
task_reconstruction_table_.reset(new GcsTaskReconstructionTable(
|
||||
std::make_shared<RedisStoreClient<TaskID, TaskReconstructionData, JobID>>(
|
||||
redis_client)));
|
||||
object_table_.reset(new GcsObjectTable(
|
||||
std::make_shared<RedisStoreClient<ObjectID, ObjectTableDataList, JobID>>(
|
||||
redis_client)));
|
||||
node_table_.reset(new GcsNodeTable(
|
||||
std::make_shared<RedisStoreClient<ClientID, GcsNodeInfo, JobID>>(redis_client)));
|
||||
node_resource_table_.reset(new GcsNodeResourceTable(
|
||||
std::make_shared<RedisStoreClient<ClientID, ResourceMap, JobID>>(redis_client)));
|
||||
heartbeat_table_.reset(new GcsHeartbeatTable(
|
||||
std::make_shared<RedisStoreClient<ClientID, HeartbeatTableData, JobID>>(
|
||||
redis_client)));
|
||||
heartbeat_batch_table_.reset(new GcsHeartbeatBatchTable(
|
||||
std::make_shared<RedisStoreClient<ClientID, HeartbeatBatchTableData, JobID>>(
|
||||
redis_client)));
|
||||
error_info_table_.reset(new GcsErrorInfoTable(
|
||||
std::make_shared<RedisStoreClient<JobID, ErrorTableData, JobID>>(redis_client)));
|
||||
profile_table_.reset(new GcsProfileTable(
|
||||
std::make_shared<RedisStoreClient<UniqueID, ProfileTableData, JobID>>(
|
||||
redis_client)));
|
||||
worker_failure_table_.reset(new GcsWorkerFailureTable(
|
||||
std::make_shared<RedisStoreClient<WorkerID, WorkerFailureData, JobID>>(
|
||||
redis_client)));
|
||||
store_client_ = std::make_shared<RedisStoreClient>(redis_client);
|
||||
job_table_.reset(new GcsJobTable(store_client_));
|
||||
actor_table_.reset(new GcsActorTable(store_client_));
|
||||
actor_checkpoint_table_.reset(new GcsActorCheckpointTable(store_client_));
|
||||
actor_checkpoint_id_table_.reset(new GcsActorCheckpointIdTable(store_client_));
|
||||
task_table_.reset(new GcsTaskTable(store_client_));
|
||||
task_lease_table_.reset(new GcsTaskLeaseTable(store_client_));
|
||||
task_reconstruction_table_.reset(new GcsTaskReconstructionTable(store_client_));
|
||||
object_table_.reset(new GcsObjectTable(store_client_));
|
||||
node_table_.reset(new GcsNodeTable(store_client_));
|
||||
node_resource_table_.reset(new GcsNodeResourceTable(store_client_));
|
||||
heartbeat_table_.reset(new GcsHeartbeatTable(store_client_));
|
||||
heartbeat_batch_table_.reset(new GcsHeartbeatBatchTable(store_client_));
|
||||
error_info_table_.reset(new GcsErrorInfoTable(store_client_));
|
||||
profile_table_.reset(new GcsProfileTable(store_client_));
|
||||
worker_failure_table_.reset(new GcsWorkerFailureTable(store_client_));
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -23,24 +23,19 @@ namespace ray {
|
||||
|
||||
namespace gcs {
|
||||
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
Status RedisStoreClient<Key, Data, IndexKey>::AsyncPut(const std::string &table_name,
|
||||
const Key &key, const Data &data,
|
||||
const StatusCallback &callback) {
|
||||
std::string full_key = table_name + key.Binary();
|
||||
std::string data_str = data.SerializeAsString();
|
||||
RAY_CHECK(!data_str.empty());
|
||||
return DoPut(full_key, data_str, callback);
|
||||
Status RedisStoreClient::AsyncPut(const std::string &table_name, const std::string &key,
|
||||
const std::string &data,
|
||||
const StatusCallback &callback) {
|
||||
std::string full_key = table_name + key;
|
||||
return DoPut(full_key, data, callback);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
Status RedisStoreClient<Key, Data, IndexKey>::AsyncPutWithIndex(
|
||||
const std::string &table_name, const Key &key, const IndexKey &index_key,
|
||||
const Data &data, const StatusCallback &callback) {
|
||||
std::string data_str = data.SerializeAsString();
|
||||
RAY_CHECK(!data_str.empty());
|
||||
|
||||
auto write_callback = [this, table_name, key, data_str, callback](Status status) {
|
||||
Status RedisStoreClient::AsyncPutWithIndex(const std::string &table_name,
|
||||
const std::string &key,
|
||||
const std::string &index_key,
|
||||
const std::string &data,
|
||||
const StatusCallback &callback) {
|
||||
auto write_callback = [this, table_name, key, data, callback](Status status) {
|
||||
if (!status.ok()) {
|
||||
// Run callback if failed.
|
||||
if (callback != nullptr) {
|
||||
@@ -50,8 +45,8 @@ Status RedisStoreClient<Key, Data, IndexKey>::AsyncPutWithIndex(
|
||||
}
|
||||
|
||||
// Write data to Redis.
|
||||
std::string full_key = table_name + key.Binary();
|
||||
status = DoPut(full_key, data_str, callback);
|
||||
std::string full_key = table_name + key;
|
||||
status = DoPut(full_key, data, callback);
|
||||
|
||||
if (!status.ok()) {
|
||||
// Run callback if failed.
|
||||
@@ -62,18 +57,16 @@ Status RedisStoreClient<Key, Data, IndexKey>::AsyncPutWithIndex(
|
||||
};
|
||||
|
||||
// Write index to Redis.
|
||||
std::string index_table_key = index_key.Binary() + table_name + key.Binary();
|
||||
return DoPut(index_table_key, key.Binary(), write_callback);
|
||||
std::string index_table_key = index_key + table_name + key;
|
||||
return DoPut(index_table_key, key, write_callback);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
Status RedisStoreClient<Key, Data, IndexKey>::DoPut(const std::string &key,
|
||||
const std::string &data,
|
||||
const StatusCallback &callback) {
|
||||
Status RedisStoreClient::DoPut(const std::string &key, const std::string &data,
|
||||
const StatusCallback &callback) {
|
||||
std::vector<std::string> args = {"SET", key, data};
|
||||
RedisCallback write_callback = nullptr;
|
||||
if (callback) {
|
||||
write_callback = [callback](std::shared_ptr<CallbackReply> reply) {
|
||||
write_callback = [callback](const std::shared_ptr<CallbackReply> &reply) {
|
||||
auto status = reply->ReadAsStatus();
|
||||
callback(status);
|
||||
};
|
||||
@@ -83,43 +76,38 @@ Status RedisStoreClient<Key, Data, IndexKey>::DoPut(const std::string &key,
|
||||
return shard_context->RunArgvAsync(args, write_callback);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
Status RedisStoreClient<Key, Data, IndexKey>::AsyncGet(
|
||||
const std::string &table_name, const Key &key,
|
||||
const OptionalItemCallback<Data> &callback) {
|
||||
Status RedisStoreClient::AsyncGet(const std::string &table_name, const std::string &key,
|
||||
const OptionalItemCallback<std::string> &callback) {
|
||||
RAY_CHECK(callback != nullptr);
|
||||
|
||||
auto redis_callback = [callback](std::shared_ptr<CallbackReply> reply) {
|
||||
boost::optional<Data> result;
|
||||
auto redis_callback = [callback](const std::shared_ptr<CallbackReply> &reply) {
|
||||
boost::optional<std::string> result;
|
||||
if (!reply->IsNil()) {
|
||||
std::string data_str = reply->ReadAsString();
|
||||
if (!data_str.empty()) {
|
||||
Data data;
|
||||
RAY_CHECK(data.ParseFromString(data_str));
|
||||
std::string data = reply->ReadAsString();
|
||||
if (!data.empty()) {
|
||||
result = std::move(data);
|
||||
}
|
||||
}
|
||||
callback(Status::OK(), result);
|
||||
};
|
||||
|
||||
std::string full_key = table_name + key.Binary();
|
||||
std::string full_key = table_name + key;
|
||||
std::vector<std::string> args = {"GET", full_key};
|
||||
|
||||
auto shard_context = redis_client_->GetShardContext(full_key);
|
||||
return shard_context->RunArgvAsync(args, redis_callback);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
Status RedisStoreClient<Key, Data, IndexKey>::AsyncGetAll(
|
||||
Status RedisStoreClient::AsyncGetAll(
|
||||
const std::string &table_name,
|
||||
const SegmentedCallback<std::pair<Key, Data>> &callback) {
|
||||
const SegmentedCallback<std::pair<std::string, std::string>> &callback) {
|
||||
RAY_CHECK(0) << "Not implemented! Will implement this function in next PR.";
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
Status RedisStoreClient<Key, Data, IndexKey>::AsyncDelete(
|
||||
const std::string &table_name, const Key &key, const StatusCallback &callback) {
|
||||
Status RedisStoreClient::AsyncDelete(const std::string &table_name,
|
||||
const std::string &key,
|
||||
const StatusCallback &callback) {
|
||||
RedisCallback delete_callback = nullptr;
|
||||
if (callback) {
|
||||
delete_callback = [callback](std::shared_ptr<CallbackReply> reply) {
|
||||
@@ -129,37 +117,20 @@ Status RedisStoreClient<Key, Data, IndexKey>::AsyncDelete(
|
||||
};
|
||||
}
|
||||
|
||||
std::string full_key = table_name + key.Binary();
|
||||
std::string full_key = table_name + key;
|
||||
std::vector<std::string> args = {"DEL", full_key};
|
||||
|
||||
auto shard_context = redis_client_->GetShardContext(full_key);
|
||||
return shard_context->RunArgvAsync(args, delete_callback);
|
||||
}
|
||||
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
Status RedisStoreClient<Key, Data, IndexKey>::AsyncDeleteByIndex(
|
||||
const std::string &table_name, const IndexKey &index_key,
|
||||
const StatusCallback &callback) {
|
||||
Status RedisStoreClient::AsyncDeleteByIndex(const std::string &table_name,
|
||||
const std::string &index_key,
|
||||
const StatusCallback &callback) {
|
||||
RAY_CHECK(0) << "Not implemented! Will implement this function in next PR.";
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
template class RedisStoreClient<ActorID, rpc::ActorTableData, JobID>;
|
||||
template class RedisStoreClient<JobID, rpc::JobTableData, JobID>;
|
||||
template class RedisStoreClient<ActorCheckpointID, rpc::ActorCheckpointData, JobID>;
|
||||
template class RedisStoreClient<ActorID, rpc::ActorCheckpointIdData, JobID>;
|
||||
template class RedisStoreClient<TaskID, rpc::TaskTableData, JobID>;
|
||||
template class RedisStoreClient<TaskID, rpc::TaskLeaseData, JobID>;
|
||||
template class RedisStoreClient<TaskID, rpc::TaskReconstructionData, JobID>;
|
||||
template class RedisStoreClient<ObjectID, rpc::ObjectTableDataList, JobID>;
|
||||
template class RedisStoreClient<ClientID, rpc::GcsNodeInfo, JobID>;
|
||||
template class RedisStoreClient<ClientID, rpc::ResourceMap, JobID>;
|
||||
template class RedisStoreClient<ClientID, rpc::HeartbeatTableData, JobID>;
|
||||
template class RedisStoreClient<ClientID, rpc::HeartbeatBatchTableData, JobID>;
|
||||
template class RedisStoreClient<JobID, rpc::ErrorTableData, JobID>;
|
||||
template class RedisStoreClient<UniqueID, rpc::ProfileTableData, JobID>;
|
||||
template class RedisStoreClient<WorkerID, rpc::WorkerFailureData, JobID>;
|
||||
|
||||
} // namespace gcs
|
||||
|
||||
} // namespace ray
|
||||
|
||||
@@ -26,31 +26,29 @@ namespace ray {
|
||||
|
||||
namespace gcs {
|
||||
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
class RedisStoreClient : public StoreClient<Key, Data, IndexKey> {
|
||||
class RedisStoreClient : public StoreClient {
|
||||
public:
|
||||
RedisStoreClient(std::shared_ptr<RedisClient> redis_client)
|
||||
: redis_client_(std::move(redis_client)) {}
|
||||
|
||||
virtual ~RedisStoreClient() {}
|
||||
Status AsyncPut(const std::string &table_name, const std::string &key,
|
||||
const std::string &data, const StatusCallback &callback) override;
|
||||
|
||||
Status AsyncPut(const std::string &table_name, const Key &key, const Data &data,
|
||||
const StatusCallback &callback) override;
|
||||
|
||||
Status AsyncPutWithIndex(const std::string &table_name, const Key &key,
|
||||
const IndexKey &index_key, const Data &data,
|
||||
Status AsyncPutWithIndex(const std::string &table_name, const std::string &key,
|
||||
const std::string &index_key, const std::string &data,
|
||||
const StatusCallback &callback) override;
|
||||
|
||||
Status AsyncGet(const std::string &table_name, const Key &key,
|
||||
const OptionalItemCallback<Data> &callback) override;
|
||||
Status AsyncGet(const std::string &table_name, const std::string &key,
|
||||
const OptionalItemCallback<std::string> &callback) override;
|
||||
|
||||
Status AsyncGetAll(const std::string &table_name,
|
||||
const SegmentedCallback<std::pair<Key, Data>> &callback) override;
|
||||
Status AsyncGetAll(
|
||||
const std::string &table_name,
|
||||
const SegmentedCallback<std::pair<std::string, std::string>> &callback) override;
|
||||
|
||||
Status AsyncDelete(const std::string &table_name, const Key &key,
|
||||
Status AsyncDelete(const std::string &table_name, const std::string &key,
|
||||
const StatusCallback &callback) override;
|
||||
|
||||
Status AsyncDeleteByIndex(const std::string &table_name, const IndexKey &index_key,
|
||||
Status AsyncDeleteByIndex(const std::string &table_name, const std::string &index_key,
|
||||
const StatusCallback &callback) override;
|
||||
|
||||
private:
|
||||
@@ -60,8 +58,6 @@ class RedisStoreClient : public StoreClient<Key, Data, IndexKey> {
|
||||
std::shared_ptr<RedisClient> redis_client_;
|
||||
};
|
||||
|
||||
typedef RedisStoreClient<ActorID, rpc::ActorTableData, JobID> RedisActorStoreTable;
|
||||
|
||||
} // namespace gcs
|
||||
|
||||
} // namespace ray
|
||||
|
||||
@@ -30,10 +30,9 @@ namespace gcs {
|
||||
|
||||
/// \class StoreClient
|
||||
/// Abstract interface of the storage client.
|
||||
template <typename Key, typename Data, typename IndexKey>
|
||||
class StoreClient {
|
||||
public:
|
||||
virtual ~StoreClient() {}
|
||||
virtual ~StoreClient() = default;
|
||||
|
||||
/// Write data to the given table asynchronously.
|
||||
///
|
||||
@@ -42,8 +41,8 @@ class StoreClient {
|
||||
/// \param data The value of the key that will be written to the table.
|
||||
/// \param callback Callback that will be called after write finishes.
|
||||
/// \return Status
|
||||
virtual Status AsyncPut(const std::string &table_name, const Key &key, const Data &data,
|
||||
const StatusCallback &callback) = 0;
|
||||
virtual Status AsyncPut(const std::string &table_name, const std::string &key,
|
||||
const std::string &data, const StatusCallback &callback) = 0;
|
||||
|
||||
/// Write data to the given table asynchronously.
|
||||
///
|
||||
@@ -53,8 +52,8 @@ class StoreClient {
|
||||
/// \param data The value of the key that will be written to the table.
|
||||
/// \param callback Callback that will be called after write finishes.
|
||||
/// \return Status
|
||||
virtual Status AsyncPutWithIndex(const std::string &table_name, const Key &key,
|
||||
const IndexKey &index_key, const Data &data,
|
||||
virtual Status AsyncPutWithIndex(const std::string &table_name, const std::string &key,
|
||||
const std::string &index_key, const std::string &data,
|
||||
const StatusCallback &callback) = 0;
|
||||
|
||||
/// Get data from the given table asynchronously.
|
||||
@@ -63,8 +62,8 @@ class StoreClient {
|
||||
/// \param key The key to lookup from the table.
|
||||
/// \param callback Callback that will be called after read finishes.
|
||||
/// \return Status
|
||||
virtual Status AsyncGet(const std::string &table_name, const Key &key,
|
||||
const OptionalItemCallback<Data> &callback) = 0;
|
||||
virtual Status AsyncGet(const std::string &table_name, const std::string &key,
|
||||
const OptionalItemCallback<std::string> &callback) = 0;
|
||||
|
||||
/// Get all data from the given table asynchronously.
|
||||
///
|
||||
@@ -72,8 +71,9 @@ class StoreClient {
|
||||
/// \param callback Callback that will be called after data has been received.
|
||||
/// If the callback return `has_more == true` mean there's more data to be received.
|
||||
/// \return Status
|
||||
virtual Status AsyncGetAll(const std::string &table_name,
|
||||
const SegmentedCallback<std::pair<Key, Data>> &callback) = 0;
|
||||
virtual Status AsyncGetAll(
|
||||
const std::string &table_name,
|
||||
const SegmentedCallback<std::pair<std::string, std::string>> &callback) = 0;
|
||||
|
||||
/// Delete data from the given table asynchronously.
|
||||
///
|
||||
@@ -81,7 +81,7 @@ class StoreClient {
|
||||
/// \param key The key that will be deleted from the table.
|
||||
/// \param callback Callback that will be called after delete finishes.
|
||||
/// \return Status
|
||||
virtual Status AsyncDelete(const std::string &table_name, const Key &key,
|
||||
virtual Status AsyncDelete(const std::string &table_name, const std::string &key,
|
||||
const StatusCallback &callback) = 0;
|
||||
|
||||
/// Delete by index from the given table asynchronously.
|
||||
@@ -92,15 +92,13 @@ class StoreClient {
|
||||
/// \param callback Callback that will be called after delete finishes.
|
||||
/// \return Status
|
||||
virtual Status AsyncDeleteByIndex(const std::string &table_name,
|
||||
const IndexKey &index_key,
|
||||
const std::string &index_key,
|
||||
const StatusCallback &callback) = 0;
|
||||
|
||||
protected:
|
||||
StoreClient() = default;
|
||||
};
|
||||
|
||||
typedef StoreClient<ActorID, rpc::ActorTableData, JobID> ActorStoreTable;
|
||||
|
||||
} // namespace gcs
|
||||
|
||||
} // namespace ray
|
||||
|
||||
@@ -31,9 +31,7 @@ class RedisStoreClientTest : public StoreClientTestBase {
|
||||
redis_client_ = std::make_shared<RedisClient>(options);
|
||||
RAY_CHECK_OK(redis_client_->Connect(io_service_pool_->GetAll()));
|
||||
|
||||
store_client_ =
|
||||
std::make_shared<RedisStoreClient<ActorID, rpc::ActorTableData, JobID>>(
|
||||
redis_client_);
|
||||
store_client_ = std::make_shared<RedisStoreClient>(redis_client_);
|
||||
}
|
||||
|
||||
void DisconnectStoreClient() override { redis_client_->Disconnect(); }
|
||||
@@ -44,96 +42,94 @@ class RedisStoreClientTest : public StoreClientTestBase {
|
||||
|
||||
TEST_F(RedisStoreClientTest, AsyncPutAndAsyncGetTest) {
|
||||
// AsyncPut without index.
|
||||
auto put_calllback = [this](Status status) {
|
||||
auto put_calllback = [this](const Status &status) {
|
||||
RAY_CHECK_OK(status);
|
||||
--pending_count_;
|
||||
};
|
||||
for (const auto &elem : key_to_value_) {
|
||||
++pending_count_;
|
||||
Status status =
|
||||
store_client_->AsyncPut(table_name_, elem.first, elem.second, put_calllback);
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK_OK(store_client_->AsyncPut(table_name_, elem.first.Binary(),
|
||||
elem.second.SerializeAsString(), put_calllback));
|
||||
}
|
||||
WaitPendingDone();
|
||||
|
||||
// AsyncGet
|
||||
auto get_callback = [this](Status status,
|
||||
const boost::optional<rpc::ActorTableData> &result) {
|
||||
auto get_callback = [this](const Status &status,
|
||||
const boost::optional<std::string> &result) {
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK(result);
|
||||
ActorID actor_id = ActorID::FromBinary(result->actor_id());
|
||||
rpc::ActorTableData data;
|
||||
RAY_CHECK(data.ParseFromString(*result));
|
||||
ActorID actor_id = ActorID::FromBinary(data.actor_id());
|
||||
auto it = key_to_value_.find(actor_id);
|
||||
RAY_CHECK(it != key_to_value_.end());
|
||||
--pending_count_;
|
||||
};
|
||||
for (const auto &elem : key_to_value_) {
|
||||
++pending_count_;
|
||||
Status status = store_client_->AsyncGet(table_name_, elem.first, get_callback);
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK_OK(store_client_->AsyncGet(table_name_, elem.first.Binary(), get_callback));
|
||||
}
|
||||
WaitPendingDone();
|
||||
}
|
||||
|
||||
TEST_F(RedisStoreClientTest, AsyncDeleteTest) {
|
||||
// AsyncPut
|
||||
auto put_calllback = [this](Status status) { --pending_count_; };
|
||||
auto put_calllback = [this](const Status &status) { --pending_count_; };
|
||||
for (const auto &elem : key_to_value_) {
|
||||
++pending_count_;
|
||||
Status status =
|
||||
store_client_->AsyncPut(table_name_, elem.first, elem.second, put_calllback);
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK_OK(store_client_->AsyncPut(table_name_, elem.first.Binary(),
|
||||
elem.second.SerializeAsString(), put_calllback));
|
||||
}
|
||||
WaitPendingDone();
|
||||
|
||||
// AsyncDelete
|
||||
auto delete_calllback = [this](Status status) {
|
||||
auto delete_calllback = [this](const Status &status) {
|
||||
RAY_CHECK_OK(status);
|
||||
--pending_count_;
|
||||
};
|
||||
for (const auto &elem : key_to_value_) {
|
||||
++pending_count_;
|
||||
Status status = store_client_->AsyncDelete(table_name_, elem.first, delete_calllback);
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK_OK(
|
||||
store_client_->AsyncDelete(table_name_, elem.first.Binary(), delete_calllback));
|
||||
}
|
||||
WaitPendingDone();
|
||||
|
||||
// AsyncGet
|
||||
auto get_callback = [this](Status status,
|
||||
const boost::optional<rpc::ActorTableData> &result) {
|
||||
auto get_callback = [this](const Status &status,
|
||||
const boost::optional<std::string> &result) {
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK(!result);
|
||||
--pending_count_;
|
||||
};
|
||||
for (const auto &elem : key_to_value_) {
|
||||
++pending_count_;
|
||||
Status status = store_client_->AsyncGet(table_name_, elem.first, get_callback);
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK_OK(store_client_->AsyncGet(table_name_, elem.first.Binary(), get_callback));
|
||||
}
|
||||
WaitPendingDone();
|
||||
}
|
||||
|
||||
TEST_F(RedisStoreClientTest, DISABLED_AsyncGetAllTest) {
|
||||
// AsyncPut
|
||||
auto put_calllback = [this](Status status) { --pending_count_; };
|
||||
auto put_calllback = [this](const Status &status) { --pending_count_; };
|
||||
for (const auto &elem : key_to_value_) {
|
||||
++pending_count_;
|
||||
// Get index
|
||||
auto it = key_to_index_.find(elem.first);
|
||||
const JobID &index = it->second;
|
||||
Status status = store_client_->AsyncPutWithIndex(table_name_, elem.first, index,
|
||||
elem.second, put_calllback);
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK_OK(
|
||||
store_client_->AsyncPutWithIndex(table_name_, elem.first.Binary(), index.Binary(),
|
||||
elem.second.SerializeAsString(), put_calllback));
|
||||
}
|
||||
WaitPendingDone();
|
||||
|
||||
// AsyncGetAll
|
||||
auto get_all_callback =
|
||||
[this](Status status, bool has_more,
|
||||
const std::vector<std::pair<ActorID, rpc::ActorTableData>> &result) {
|
||||
[this](const Status &status, bool has_more,
|
||||
const std::vector<std::pair<std::string, std::string>> &result) {
|
||||
RAY_CHECK_OK(status);
|
||||
static std::unordered_set<ActorID> received_keys;
|
||||
for (const auto &item : result) {
|
||||
const ActorID &actor_id = item.first;
|
||||
const ActorID &actor_id = ActorID::FromBinary(item.first);
|
||||
auto it = received_keys.find(actor_id);
|
||||
RAY_CHECK(it == received_keys.end());
|
||||
received_keys.emplace(actor_id);
|
||||
@@ -148,8 +144,7 @@ TEST_F(RedisStoreClientTest, DISABLED_AsyncGetAllTest) {
|
||||
};
|
||||
|
||||
pending_count_ += key_to_value_.size();
|
||||
Status status = store_client_->AsyncGetAll(table_name_, get_all_callback);
|
||||
RAY_CHECK_OK(status);
|
||||
RAY_CHECK_OK(store_client_->AsyncGetAll(table_name_, get_all_callback));
|
||||
WaitPendingDone();
|
||||
}
|
||||
|
||||
|
||||
@@ -94,7 +94,7 @@ class StoreClientTestBase : public RedisServiceManagerForTest {
|
||||
size_t io_service_num_{2};
|
||||
std::shared_ptr<IOServicePool> io_service_pool_;
|
||||
|
||||
std::shared_ptr<StoreClient<ActorID, rpc::ActorTableData, JobID>> store_client_;
|
||||
std::shared_ptr<StoreClient> store_client_;
|
||||
|
||||
std::string table_name_{"test_table"};
|
||||
size_t key_count_{5000};
|
||||
|
||||
Reference in New Issue
Block a user