Skip to content

P
Paddle

BaiXuePrincess / Paddle
与 Fork 源项目一致

Fork自 PaddlePaddle / Paddle

通知 1
Star 1
Fork 0
P
Paddle
未验证 提交 6f20a383 编写于 作者: W wangxiaoning 提交者： GitHub
浏览文件
操作

[code style fix] cpplint code style fix (#49554) 

* fix graph_brpc_server.h code style
* fix ps_local_client.h
* fix ps_local_client.cc
上级 ac9c7e4d
显示空白变更内容
内联 并排
Showing with 93 addition and 89 deletion
paddle/fluid/distributed/ps/service/graph_brpc_server.h
浏览文件 @ 6f20a383
...@@ -63,14 +63,14 @@ class GraphBrpcService; ...@@ -63,14 +63,14 @@ class GraphBrpcService;
typedef int32_t (GraphBrpcService::*serviceFunc)( typedef int32_t (GraphBrpcService::*serviceFunc)(
Table *table, Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
class GraphBrpcService : public PsBaseService { class GraphBrpcService : public PsBaseService {
public: public:
virtual int32_t Initialize() override; int32_t Initialize() override;
virtual void service(::google::protobuf::RpcController *controller, void service(::google::protobuf::RpcController *controller,
const PsRequestMessage *request, const PsRequestMessage *request,
PsResponseMessage *response, PsResponseMessage *response,
::google::protobuf::Closure *done) override; ::google::protobuf::Closure *done) override;
...@@ -80,80 +80,81 @@ class GraphBrpcService : public PsBaseService { ...@@ -80,80 +80,81 @@ class GraphBrpcService : public PsBaseService {
int32_t InitializeShardInfo(); int32_t InitializeShardInfo();
int32_t pull_graph_list(Table *table, int32_t pull_graph_list(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t graph_random_sample_neighbors(Table *table, int32_t graph_random_sample_neighbors(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t graph_random_sample_nodes(Table *table, int32_t graph_random_sample_nodes(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t graph_get_node_feat(Table *table, int32_t graph_get_node_feat(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t graph_set_node_feat(Table *table, int32_t graph_set_node_feat(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t clear_nodes(Table *table, int32_t clear_nodes(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t add_graph_node(Table *table, int32_t add_graph_node(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t remove_graph_node(Table *table, int32_t remove_graph_node(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t Barrier(Table *table, int32_t Barrier(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t LoadOneTable(Table *table, int32_t LoadOneTable(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t LoadAllTable(Table *table, int32_t LoadAllTable(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t StopServer(Table *table, int32_t StopServer(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t StartProfiler(Table *table, int32_t StartProfiler(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t StopProfiler(Table *table, int32_t StopProfiler(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t PrintTableStat(Table *table, int32_t PrintTableStat(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t sample_neighbors_across_multi_servers(Table *table, int32_t sample_neighbors_across_multi_servers(
Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t use_neighbors_sample_cache(Table *table, int32_t use_neighbors_sample_cache(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
int32_t load_graph_split_config(Table *table, int32_t load_graph_split_config(Table *table,
const PsRequestMessage &request, const PsRequestMessage &request,
PsResponseMessage &response, PsResponseMessage &response, // NOLINT
brpc::Controller *cntl); brpc::Controller *cntl);
private: private:
......
paddle/fluid/distributed/ps/service/ps_local_client.cc
浏览文件 @ 6f20a383
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#include "paddle/fluid/distributed/ps/table/table.h" #include "paddle/fluid/distributed/ps/table/table.h"
//#define pslib_debug_dense_compress // #define pslib_debug_dense_compress
namespace paddle { namespace paddle {
namespace distributed { namespace distributed {
...@@ -36,13 +36,13 @@ int32_t PsLocalClient::Initialize() { ...@@ -36,13 +36,13 @@ int32_t PsLocalClient::Initialize() {
::std::future<int32_t> PsLocalClient::Shrink(uint32_t table_id, ::std::future<int32_t> PsLocalClient::Shrink(uint32_t table_id,
const std::string threshold) { const std::string threshold) {
// TODO // TODO // NOLINT
return done(); return done();
} }
::std::future<int32_t> PsLocalClient::Load(const std::string& epoch, ::std::future<int32_t> PsLocalClient::Load(const std::string& epoch,
const std::string& mode) { const std::string& mode) {
// TODO // TODO // NOLINT
for (auto& it : _table_map) { for (auto& it : _table_map) {
Load(it.first, epoch, mode); Load(it.first, epoch, mode);
} }
...@@ -51,7 +51,7 @@ int32_t PsLocalClient::Initialize() { ...@@ -51,7 +51,7 @@ int32_t PsLocalClient::Initialize() {
::std::future<int32_t> PsLocalClient::Load(uint32_t table_id, ::std::future<int32_t> PsLocalClient::Load(uint32_t table_id,
const std::string& epoch, const std::string& epoch,
const std::string& mode) { const std::string& mode) {
// TODO // TODO // NOLINT
auto* table_ptr = GetTable(table_id); auto* table_ptr = GetTable(table_id);
table_ptr->Load(epoch, mode); table_ptr->Load(epoch, mode);
return done(); return done();
...@@ -59,7 +59,7 @@ int32_t PsLocalClient::Initialize() { ...@@ -59,7 +59,7 @@ int32_t PsLocalClient::Initialize() {
::std::future<int32_t> PsLocalClient::Save(const std::string& epoch, ::std::future<int32_t> PsLocalClient::Save(const std::string& epoch,
const std::string& mode) { const std::string& mode) {
// TODO // TODO // NOLINT
for (auto& it : _table_map) { for (auto& it : _table_map) {
Save(it.first, epoch, mode); Save(it.first, epoch, mode);
} }
...@@ -68,7 +68,7 @@ int32_t PsLocalClient::Initialize() { ...@@ -68,7 +68,7 @@ int32_t PsLocalClient::Initialize() {
::std::future<int32_t> PsLocalClient::Save(uint32_t table_id, ::std::future<int32_t> PsLocalClient::Save(uint32_t table_id,
const std::string& epoch, const std::string& epoch,
const std::string& mode) { const std::string& mode) {
// TODO // TODO // NOLINT
auto* table_ptr = GetTable(table_id); auto* table_ptr = GetTable(table_id);
table_ptr->Flush(); table_ptr->Flush();
table_ptr->Save(epoch, mode); table_ptr->Save(epoch, mode);
...@@ -76,11 +76,11 @@ int32_t PsLocalClient::Initialize() { ...@@ -76,11 +76,11 @@ int32_t PsLocalClient::Initialize() {
} }
::std::future<int32_t> PsLocalClient::Clear() { ::std::future<int32_t> PsLocalClient::Clear() {
// TODO // TODO // NOLINT
return done(); return done();
} }
::std::future<int32_t> PsLocalClient::Clear(uint32_t table_id) { ::std::future<int32_t> PsLocalClient::Clear(uint32_t table_id) {
// TODO // TODO // NOLINT
return done(); return done();
} }
...@@ -125,8 +125,10 @@ int32_t PsLocalClient::Initialize() { ...@@ -125,8 +125,10 @@ int32_t PsLocalClient::Initialize() {
while (shard_buffer_remain > 0 && region_idx < region_num) { while (shard_buffer_remain > 0 && region_idx < region_num) {
auto& region = regions[region_idx]; auto& region = regions[region_idx];
if (region.size - region_data_idx >= shard_buffer_remain) { if (region.size - region_data_idx >= shard_buffer_remain) {
memcpy((void*)(region.data + region_data_idx), memcpy(reinterpret_cast<void*>(region.data + region_data_idx),
(uint8_t*)(void*)(region_buffer.data()) + index, reinterpret_cast<uint8_t*>(
reinterpret_cast<void*>(region_buffer.data())) +
index,
shard_buffer_remain); shard_buffer_remain);
region_data_idx += shard_buffer_remain; region_data_idx += shard_buffer_remain;
shard_buffer_remain = 0; shard_buffer_remain = 0;
...@@ -134,8 +136,10 @@ int32_t PsLocalClient::Initialize() { ...@@ -134,8 +136,10 @@ int32_t PsLocalClient::Initialize() {
++region_idx; ++region_idx;
region_data_idx = 0; region_data_idx = 0;
} else { } else {
memcpy((void*)(region.data + region_data_idx), memcpy(reinterpret_cast<void*>(region.data + region_data_idx),
(uint8_t*)(void*)(region_buffer.data()) + index, reinterpret_cast<uint8_t*>(
reinterpret_cast<void*>(region_buffer.data())) +
index,
region.size - region_data_idx); region.size - region_data_idx);
shard_buffer_remain -= (region.size - region_data_idx); shard_buffer_remain -= (region.size - region_data_idx);
index += (region.size - region_data_idx); index += (region.size - region_data_idx);
...@@ -230,7 +234,7 @@ int32_t PsLocalClient::Initialize() { ...@@ -230,7 +234,7 @@ int32_t PsLocalClient::Initialize() {
return done(); return done();
} }
//::std::future<int32_t> PsLocalClient::PullSparse(float** select_values, // ::std::future<int32_t> PsLocalClient::PullSparse(float** select_values,
// size_t table_id, // size_t table_id,
// const uint64_t* keys, // const uint64_t* keys,
// size_t num) { // size_t num) {
...@@ -271,7 +275,7 @@ int32_t PsLocalClient::Initialize() { ...@@ -271,7 +275,7 @@ int32_t PsLocalClient::Initialize() {
// std::make_shared<CostTimer>("pslib_downpour_client_pull_sparse"); // std::make_shared<CostTimer>("pslib_downpour_client_pull_sparse");
// auto local_timer = // auto local_timer =
// std::make_shared<CostTimer>("pslib_downpour_client_pull_sparse_local"); // std::make_shared<CostTimer>("pslib_downpour_client_pull_sparse_local");
//将key拆分到各shard请求，并记录原始对应value指针 // 将key拆分到各shard请求，并记录原始对应value指针
auto* table_ptr = GetTable(table_id); auto* table_ptr = GetTable(table_id);
TableContext table_context; TableContext table_context;
......
paddle/fluid/distributed/ps/service/ps_local_client.h
浏览文件 @ 6f20a383
...@@ -32,26 +32,26 @@ class PsLocalClient : public PSClient { ...@@ -32,26 +32,26 @@ class PsLocalClient : public PSClient {
return 0; return 0;
} }
virtual ::std::future<int32_t> Shrink(uint32_t table_id, ::std::future<int32_t> Shrink(uint32_t table_id,
const std::string threshold) override; const std::string threshold) override;
virtual ::std::future<int32_t> Load(const std::string& epoch, ::std::future<int32_t> Load(const std::string& epoch,
const std::string& mode) override; const std::string& mode) override;
virtual ::std::future<int32_t> Load(uint32_t table_id, ::std::future<int32_t> Load(uint32_t table_id,
const std::string& epoch, const std::string& epoch,
const std::string& mode) override; const std::string& mode) override;
virtual ::std::future<int32_t> Save(const std::string& epoch, ::std::future<int32_t> Save(const std::string& epoch,
const std::string& mode) override; const std::string& mode) override;
virtual ::std::future<int32_t> Save(uint32_t table_id, ::std::future<int32_t> Save(uint32_t table_id,
const std::string& epoch, const std::string& epoch,
const std::string& mode) override; const std::string& mode) override;
virtual ::std::future<int32_t> Clear() override; ::std::future<int32_t> Clear() override;
virtual ::std::future<int32_t> Clear(uint32_t table_id) override; ::std::future<int32_t> Clear(uint32_t table_id) override;
virtual ::std::future<int32_t> StopServer() override; ::std::future<int32_t> StopServer() override;
virtual void FinalizeWorker() override {} void FinalizeWorker() override {}
virtual ::std::future<int32_t> PullDense(Region* regions, virtual ::std::future<int32_t> PullDense(Region* regions,
size_t region_num, size_t region_num,
size_t table_id); size_t table_id);
...@@ -102,7 +102,7 @@ class PsLocalClient : public PSClient { ...@@ -102,7 +102,7 @@ class PsLocalClient : public PSClient {
prom.set_value(0); prom.set_value(0);
return fut; return fut;
}; }
virtual std::future<int32_t> StopProfiler() { virtual std::future<int32_t> StopProfiler() {
std::promise<int32_t> prom; std::promise<int32_t> prom;
...@@ -147,8 +147,9 @@ class PsLocalClient : public PSClient { ...@@ -147,8 +147,9 @@ class PsLocalClient : public PSClient {
return 0; return 0;
} }
virtual ::std::future<int32_t> SendClient2ClientMsg( ::std::future<int32_t> SendClient2ClientMsg(int msg_type,
int msg_type, int to_client_id, const std::string& msg) override { int to_client_id,
const std::string& msg) override {
std::promise<int32_t> prom; std::promise<int32_t> prom;
std::future<int32_t> fut = prom.get_future(); std::future<int32_t> fut = prom.get_future();
prom.set_value(0); prom.set_value(0);
...@@ -157,20 +158,18 @@ class PsLocalClient : public PSClient { ...@@ -157,20 +158,18 @@ class PsLocalClient : public PSClient {
} }
virtual size_t GetServerNums() { return 1; } virtual size_t GetServerNums() { return 1; }
virtual std::future<int32_t> PushDenseRawGradient(int table_id, std::future<int32_t> PushDenseRawGradient(int table_id,
float* total_send_data, float* total_send_data,
size_t total_send_data_size, size_t total_send_data_size,
void* callback) override; void* callback) override;
virtual std::future<int32_t> PushSparseRawGradient( std::future<int32_t> PushSparseRawGradient(size_t table_id,
size_t table_id,
const uint64_t* keys, const uint64_t* keys,
const float** update_values, const float** update_values,
size_t num, size_t num,
void* callback) override; void* callback) override;
virtual std::future<int32_t> PushSparseRawGradientPartial( std::future<int32_t> PushSparseRawGradientPartial(size_t table_id,
size_t table_id,
const uint64_t* keys, const uint64_t* keys,
const float** update_values, const float** update_values,
uint32_t num, uint32_t num,
...@@ -183,7 +182,7 @@ class PsLocalClient : public PSClient { ...@@ -183,7 +182,7 @@ class PsLocalClient : public PSClient {
return fut; return fut;
} }
virtual std::future<int32_t> PushSparseParam(size_t table_id, std::future<int32_t> PushSparseParam(size_t table_id,
const uint64_t* keys, const uint64_t* keys,
const float** update_values, const float** update_values,
size_t num, size_t num,
...@@ -196,7 +195,7 @@ class PsLocalClient : public PSClient { ...@@ -196,7 +195,7 @@ class PsLocalClient : public PSClient {
} }
private: private:
virtual int32_t Initialize() override; int32_t Initialize() override;
std::future<int32_t> done() { std::future<int32_t> done() {
std::shared_ptr<std::promise<int32_t>> prom = std::shared_ptr<std::promise<int32_t>> prom =
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或