huawei
/
mindspore2022
0
0
Fork 53
Code Issues Pull Requests 29 Projects Releases Wiki Activity

Fix pc lint

This commit is contained in:
ZPaC 2021-07-22 20:56:38 +08:00
parent 349d3f85cb
commit 08c81b3cec
21 changed files with 199 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
mindspore/ccsrc/fl/server/collective_ops_impl.cc
View File

@ -62,6 +62,7 @@ bool CollectiveOpsImpl::RingAllReduce(const void *sendbuff, void *recvbuff, size
// Ring ReduceScatter.
MS_LOG(DEBUG) << "Start Ring ReduceScatter.";
std::unique_ptr<T[]> tmp_recv_chunk = std::make_unique<T[]>(chunk_sizes[0]);
MS_EXCEPTION_IF_NULL(tmp_recv_chunk);
for (size_t i = 0; i < rank_size - 1; i++) {
// Step 1: Async send data to next rank.
size_t send_chunk_index = (local_rank_ - i + rank_size) % rank_size;
@ -147,6 +148,7 @@ bool CollectiveOpsImpl::ReduceBroadcastAllReduce(const void *sendbuff, void *rec
MS_LOG(DEBUG) << "Start Reduce to rank 0 process.";
if (local_rank_ == 0) {
std::unique_ptr<T[]> tmp_recv_buff = std::make_unique<T[]>(count);
MS_EXCEPTION_IF_NULL(tmp_recv_buff);
for (uint32_t i = 1; i < rank_size; i++) {
std::shared_ptr<std::vector<unsigned char>> recv_str;
MS_LOG(DEBUG) << "Reduce rank 0 receive from rank " << i;

2
mindspore/ccsrc/fl/server/consistent_hash_ring.cc
View File

@ -37,7 +37,7 @@ bool ConsistentHashRing::Insert(uint32_t rank) {
bool ConsistentHashRing::Erase(uint32_t rank) {
for (auto iterator = ring_.begin(); iterator != ring_.end();) {
if (iterator->second == rank) {
ring_.erase(iterator++);
(void)ring_.erase(iterator++);
}
}
return true;

34
mindspore/ccsrc/fl/server/distributed_count_service.cc
View File

@ -82,7 +82,7 @@ bool DistributedCountService::Count(const std::string &name, const std::string &
}
MS_LOG(INFO) << "Leader server increase count for " << name << " of " << id;
global_current_count_[name].insert(id);
(void)global_current_count_[name].insert(id);
if (!TriggerCounterEvent(name, reason)) {
MS_LOG(ERROR) << "Leader server trigger count event failed.";
return false;
@ -190,7 +190,11 @@ void DistributedCountService::HandleCountRequest(const std::shared_ptr<ps::core:
count_rsp.set_result(false);
count_rsp.set_reason(reason);
MS_LOG(ERROR) << reason;
communicator_->SendResponse(count_rsp.SerializeAsString().data(), count_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(count_rsp.SerializeAsString().data(), count_rsp.SerializeAsString().size(),
message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}
@ -201,13 +205,17 @@ void DistributedCountService::HandleCountRequest(const std::shared_ptr<ps::core:
count_rsp.set_result(false);
count_rsp.set_reason(reason);
MS_LOG(ERROR) << reason;
communicator_->SendResponse(count_rsp.SerializeAsString().data(), count_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(count_rsp.SerializeAsString().data(), count_rsp.SerializeAsString().size(),
message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}
// Insert the id for the counter, which means the count for the name is increased.
MS_LOG(INFO) << "Leader server increase count for " << name << " of " << id;
global_current_count_[name].insert(id);
(void)global_current_count_[name].insert(id);
std::string reason = "success";
if (!TriggerCounterEvent(name, &reason)) {
count_rsp.set_result(false);
@ -216,7 +224,11 @@ void DistributedCountService::HandleCountRequest(const std::shared_ptr<ps::core:
count_rsp.set_result(true);
count_rsp.set_reason(reason);
}
communicator_->SendResponse(count_rsp.SerializeAsString().data(), count_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(count_rsp.SerializeAsString().data(), count_rsp.SerializeAsString().size(),
message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}
@ -239,8 +251,11 @@ void DistributedCountService::HandleCountReachThresholdRequest(
CountReachThresholdResponse count_reach_threshold_rsp;
count_reach_threshold_rsp.set_is_enough(global_current_count_[name].size() == global_threshold_count_[name]);
communicator_->SendResponse(count_reach_threshold_rsp.SerializeAsString().data(),
count_reach_threshold_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(count_reach_threshold_rsp.SerializeAsString().data(),
count_reach_threshold_rsp.SerializeAsString().size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}
@ -253,7 +268,10 @@ void DistributedCountService::HandleCounterEvent(const std::shared_ptr<ps::core:
// Respond as soon as possible so the leader server won't wait for each follower servers to finish calling the
// callbacks.
std::string couter_event_rsp_msg = "success";
communicator_->SendResponse(couter_event_rsp_msg.data(), couter_event_rsp_msg.size(), message);
if (!communicator_->SendResponse(couter_event_rsp_msg.data(), couter_event_rsp_msg.size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
CounterEvent counter_event;
(void)counter_event.ParseFromArray(message->data(), SizeToInt(message->len()));

10
mindspore/ccsrc/fl/server/distributed_metadata_store.cc
View File

@ -195,7 +195,10 @@ void DistributedMetadataStore::HandleUpdateMetadataRequest(const std::shared_ptr
} else {
update_meta_rsp_msg = "Success";
}
(void)communicator_->SendResponse(update_meta_rsp_msg.data(), update_meta_rsp_msg.size(), message);
if (!communicator_->SendResponse(update_meta_rsp_msg.data(), update_meta_rsp_msg.size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}
@ -213,7 +216,10 @@ void DistributedMetadataStore::HandleGetMetadataRequest(const std::shared_ptr<ps
std::unique_lock<std::mutex> lock(mutex_[name]);
PBMetadata stored_meta = metadata_[name];
std::string getting_meta_rsp_msg = stored_meta.SerializeAsString();
(void)communicator_->SendResponse(getting_meta_rsp_msg.data(), getting_meta_rsp_msg.size(), message);
if (!communicator_->SendResponse(getting_meta_rsp_msg.data(), getting_meta_rsp_msg.size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}

60
mindspore/ccsrc/fl/server/iteration.cc
View File

@ -63,15 +63,15 @@ void Iteration::InitRounds(const std::vector<std::shared_ptr<ps::core::Communica
return;
}
std::for_each(communicators.begin(), communicators.end(),
[&](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) {
for (auto &round : rounds_) {
if (round == nullptr) {
continue;
}
round->Initialize(communicator, timeout_cb, finish_iteration_cb);
}
});
(void)std::for_each(communicators.begin(), communicators.end(),
[&](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) {
for (auto &round : rounds_) {
if (round == nullptr) {
continue;
}
round->Initialize(communicator, timeout_cb, finish_iteration_cb);
}
});
// The time window for one iteration, which will be used in some round kernels.
size_t iteration_time_window = std::accumulate(rounds_.begin(), rounds_.end(), IntToSize(0),
@ -203,7 +203,10 @@ void Iteration::HandleSyncIterationRequest(const std::shared_ptr<ps::core::Messa
SyncIterationResponse sync_iter_rsp;
sync_iter_rsp.set_iteration(iteration_num_);
std::string sync_iter_rsp_msg = sync_iter_rsp.SerializeAsString();
(void)communicator_->SendResponse(sync_iter_rsp_msg.data(), sync_iter_rsp_msg.size(), message);
if (!communicator_->SendResponse(sync_iter_rsp_msg.data(), sync_iter_rsp_msg.size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
}
bool Iteration::IsMoveToNextIterRequestReentrant(uint64_t iteration_num) {
@ -238,8 +241,11 @@ void Iteration::HandleNotifyLeaderMoveToNextIterRequest(const std::shared_ptr<ps
NotifyLeaderMoveToNextIterResponse notify_leader_to_next_iter_rsp;
notify_leader_to_next_iter_rsp.set_result("success");
(void)communicator_->SendResponse(notify_leader_to_next_iter_rsp.SerializeAsString().data(),
notify_leader_to_next_iter_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(notify_leader_to_next_iter_rsp.SerializeAsString().data(),
notify_leader_to_next_iter_rsp.SerializeAsString().size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
NotifyLeaderMoveToNextIterRequest notify_leader_to_next_iter_req;
(void)notify_leader_to_next_iter_req.ParseFromArray(message->data(), SizeToInt(message->len()));
@ -286,7 +292,7 @@ bool Iteration::BroadcastPrepareForNextIterRequest(bool is_last_iter_valid, cons
}
// Retry sending to offline servers to notify them to prepare.
std::for_each(offline_servers.begin(), offline_servers.end(), [&](uint32_t rank) {
(void)std::for_each(offline_servers.begin(), offline_servers.end(), [&](uint32_t rank) {
// Should avoid endless loop if the server communicator is stopped.
while (communicator_->running() &&
!communicator_->SendPbRequest(prepare_next_iter_req, rank, ps::core::TcpUserCommand::kPrepareForNextIter)) {
@ -313,8 +319,11 @@ void Iteration::HandlePrepareForNextIterRequest(const std::shared_ptr<ps::core::
PrepareForNextIterResponse prepare_next_iter_rsp;
prepare_next_iter_rsp.set_result("success");
(void)communicator_->SendResponse(prepare_next_iter_rsp.SerializeAsString().data(),
prepare_next_iter_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(prepare_next_iter_rsp.SerializeAsString().data(),
prepare_next_iter_rsp.SerializeAsString().size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
}
void Iteration::PrepareForNextIter() {
@ -347,8 +356,11 @@ void Iteration::HandleMoveToNextIterRequest(const std::shared_ptr<ps::core::Mess
MoveToNextIterResponse proceed_to_next_iter_rsp;
proceed_to_next_iter_rsp.set_result("success");
(void)communicator_->SendResponse(proceed_to_next_iter_rsp.SerializeAsString().data(),
proceed_to_next_iter_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(proceed_to_next_iter_rsp.SerializeAsString().data(),
proceed_to_next_iter_rsp.SerializeAsString().size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
MoveToNextIterRequest proceed_to_next_iter_req;
(void)proceed_to_next_iter_req.ParseFromArray(message->data(), SizeToInt(message->len()));
@ -413,8 +425,11 @@ void Iteration::HandleEndLastIterRequest(const std::shared_ptr<ps::core::Message
std::to_string(iteration_num_) + ", iteration to be ended is " + std::to_string(last_iter_num);
EndLastIterResponse end_last_iter_rsp;
end_last_iter_rsp.set_result(reason);
(void)communicator_->SendResponse(end_last_iter_rsp.SerializeAsString().data(),
end_last_iter_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(end_last_iter_rsp.SerializeAsString().data(),
end_last_iter_rsp.SerializeAsString().size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}
@ -422,8 +437,11 @@ void Iteration::HandleEndLastIterRequest(const std::shared_ptr<ps::core::Message
EndLastIterResponse end_last_iter_rsp;
end_last_iter_rsp.set_result("success");
(void)communicator_->SendResponse(end_last_iter_rsp.SerializeAsString().data(),
end_last_iter_rsp.SerializeAsString().size(), message);
if (!communicator_->SendResponse(end_last_iter_rsp.SerializeAsString().data(),
end_last_iter_rsp.SerializeAsString().size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
}
void Iteration::EndLastIter() {

2
mindspore/ccsrc/fl/server/kernel/round/get_model_kernel.cc
View File

@ -123,7 +123,7 @@ void GetModelKernel::BuildGetModelRsp(const std::shared_ptr<FBBuilder> &fbb, con
auto fbs_feature_maps_vector = fbb->CreateVector(fbs_feature_maps);
schema::ResponseGetModelBuilder rsp_get_model_builder(*(fbb.get()));
rsp_get_model_builder.add_retcode(retcode);
rsp_get_model_builder.add_retcode(static_cast<int>(retcode));
rsp_get_model_builder.add_reason(fbs_reason);
rsp_get_model_builder.add_iteration(static_cast<int>(iter));
rsp_get_model_builder.add_feature_map(fbs_feature_maps_vector);

4
mindspore/ccsrc/fl/server/kernel/round/round_kernel.cc
View File

@ -49,7 +49,7 @@ RoundKernel::RoundKernel() : name_(""), current_count_(0), required_count_(0), e
}
// Manually release unique_ptr data.
heap_data_[addr_ptr].reset(nullptr);
heap_data_.erase(heap_data_.find(addr_ptr));
(void)heap_data_.erase(heap_data_.find(addr_ptr));
}
});
}
@ -124,7 +124,7 @@ void RoundKernel::GenerateOutput(const std::vector<AddressPtr> &outputs, const v
outputs[0]->size = len;
std::unique_lock<std::mutex> lock(heap_data_mtx_);
heap_data_.insert(std::make_pair(outputs[0], std::move(output_data)));
(void)heap_data_.insert(std::make_pair(outputs[0], std::move(output_data)));
return;
}
} // namespace kernel

6
mindspore/ccsrc/fl/server/kernel/round/start_fl_job_kernel.cc
View File

@ -79,6 +79,10 @@ bool StartFLJobKernel::Launch(const std::vector<AddressPtr> &inputs, const std::
}
const schema::RequestFLJob *start_fl_job_req = flatbuffers::GetRoot<schema::RequestFLJob>(req_data);
if (start_fl_job_req == nullptr) {
MS_LOG(ERROR) << "RequestFLJob is nullptr.";
return false;
}
DeviceMeta device_meta = CreateDeviceMetadata(start_fl_job_req);
result_code = ReadyForStartFLJob(fbb, device_meta);
if (result_code != ResultCode::kSuccess) {
@ -240,7 +244,7 @@ void StartFLJobKernel::BuildStartFLJobRsp(const std::shared_ptr<FBBuilder> &fbb,
auto fbs_feature_maps_vector = fbb->CreateVector(fbs_feature_maps);
schema::ResponseFLJobBuilder rsp_fl_job_builder(*(fbb.get()));
rsp_fl_job_builder.add_retcode(retcode);
rsp_fl_job_builder.add_retcode(static_cast<int>(retcode));
rsp_fl_job_builder.add_reason(fbs_reason);
rsp_fl_job_builder.add_iteration(SizeToInt(LocalMetaStore::GetInstance().curr_iter_num()));
rsp_fl_job_builder.add_is_selected(is_selected);

11
mindspore/ccsrc/fl/server/kernel/round/update_model_kernel.cc
View File

@ -159,7 +159,14 @@ ResultCode UpdateModelKernel::UpdateModel(const schema::RequestUpdateModel *upda
for (auto weight : feature_map) {
weight.second[kNewDataSize].addr = &data_size;
weight.second[kNewDataSize].size = sizeof(size_t);
executor_->HandleModelUpdate(weight.first, weight.second);
if (!executor_->HandleModelUpdate(weight.first, weight.second)) {
std::string reason = "Updating weight " + weight.first + " failed.";
BuildUpdateModelRsp(
fbb, schema::ResponseCode_OutOfTime, reason,
std::to_string(LocalMetaStore::GetInstance().value<uint64_t>(kCtxIterationNextRequestTimestamp)));
MS_LOG(ERROR) << reason;
return ResultCode::kFail;
}
}
FLId fl_id;
@ -220,7 +227,7 @@ void UpdateModelKernel::BuildUpdateModelRsp(const std::shared_ptr<FBBuilder> &fb
auto fbs_next_req_time = fbb->CreateString(next_req_time);
schema::ResponseUpdateModelBuilder rsp_update_model_builder(*(fbb.get()));
rsp_update_model_builder.add_retcode(retcode);
rsp_update_model_builder.add_retcode(static_cast<int>(retcode));
rsp_update_model_builder.add_reason(fbs_reason);
rsp_update_model_builder.add_next_req_time(fbs_next_req_time);
auto rsp_update_model = rsp_update_model_builder.Finish();

2
mindspore/ccsrc/fl/server/local_meta_store.cc
View File

@ -22,7 +22,7 @@ namespace server {
void LocalMetaStore::remove_value(const std::string &name) {
std::unique_lock<std::mutex> lock(mtx_);
if (key_to_meta_.count(name) != 0) {
key_to_meta_.erase(key_to_meta_.find(name));
(void)key_to_meta_.erase(key_to_meta_.find(name));
}
}

2
mindspore/ccsrc/fl/server/memory_register.cc
View File

@ -21,7 +21,7 @@ namespace mindspore {
namespace fl {
namespace server {
void MemoryRegister::RegisterAddressPtr(const std::string &name, const AddressPtr &address) {
addresses_.try_emplace(name, address);
(void)addresses_.try_emplace(name, address);
}
void MemoryRegister::StoreFloatArray(std::unique_ptr<float[]> *array) { float_arrays_.push_back(std::move(*array)); }

2
mindspore/ccsrc/fl/server/model_store.cc
View File

@ -62,7 +62,7 @@ bool ModelStore::StoreModelByIterNum(size_t iteration, const std::map<std::strin
MS_LOG(ERROR) << "Earliest model is nullptr.";
return false;
}
iteration_to_model_.erase(iteration_to_model_.begin());
(void)iteration_to_model_.erase(iteration_to_model_.begin());
}
// Copy new model data to the the stored model.

4
mindspore/ccsrc/fl/server/parameter_aggregator.cc
View File

@ -323,9 +323,9 @@ std::vector<std::string> ParameterAggregator::SelectAggregationAlgorithm(const C
std::vector<std::string> aggregation_algorithm = {};
if (ps::PSContext::instance()->server_mode() == ps::kServerModeFL ||
ps::PSContext::instance()->server_mode() == ps::kServerModeHybrid) {
aggregation_algorithm.push_back("FedAvg");
(void)aggregation_algorithm.emplace_back("FedAvg");
} else if (ps::PSContext::instance()->server_mode() == ps::kServerModePS) {
aggregation_algorithm.push_back("DenseGradAccum");
(void)aggregation_algorithm.emplace_back("DenseGradAccum");
} else {
MS_LOG(ERROR) << "Server doesn't support mode " << ps::PSContext::instance()->server_mode();
}

15
mindspore/ccsrc/fl/server/round.cc
View File

@ -116,7 +116,10 @@ void Round::LaunchRoundKernel(const std::shared_ptr<ps::core::MessageHandler> &m
if (Server::GetInstance().IsSafeMode()) {
MS_LOG(WARNING) << "The cluster is still in process of scaling, please retry " << name_ << " later.";
std::string reason = "The cluster is in safemode.";
(void)communicator_->SendResponse(reason.c_str(), reason.size(), message);
if (!communicator_->SendResponse(reason.c_str(), reason.size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}
@ -128,10 +131,16 @@ void Round::LaunchRoundKernel(const std::shared_ptr<ps::core::MessageHandler> &m
if (output->size == 0) {
std::string reason = "The output of the round " + name_ + " is empty.";
MS_LOG(WARNING) << reason;
(void)communicator_->SendResponse(reason.c_str(), reason.size(), message);
if (!communicator_->SendResponse(reason.c_str(), reason.size(), message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
return;
}
if (!communicator_->SendResponse(output->addr, output->size, message)) {
MS_LOG(ERROR) << "Sending response failed.";
return;
}
(void)communicator_->SendResponse(output->addr, output->size, message);
kernel_->Release(output);
// Must send response back no matter what value Launch method returns.

33
mindspore/ccsrc/fl/server/server.cc
View File

@ -85,8 +85,8 @@ void Server::Run() {
lock.unlock();
// Wait communicators to stop so the main thread is blocked.
std::for_each(communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) { communicator->Join(); });
(void)std::for_each(communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) { communicator->Join(); });
communicator_with_server_->Join();
MsException::Instance().CheckException();
return;
@ -303,8 +303,9 @@ void Server::RegisterExceptionEventCallback(const std::shared_ptr<ps::core::TcpC
communicator->RegisterEventCallback(ps::core::ClusterEvent::SCHEDULER_TIMEOUT, [&]() {
MS_LOG(ERROR) << "Event SCHEDULER_TIMEOUT is captured. This is because scheduler node is finalized or crashed.";
safemode_ = true;
std::for_each(communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) { (void)communicator->Stop(); });
(void)std::for_each(
communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) { (void)communicator->Stop(); });
(void)communicator_with_server_->Stop();
});
@ -313,8 +314,9 @@ void Server::RegisterExceptionEventCallback(const std::shared_ptr<ps::core::TcpC
<< "Event NODE_TIMEOUT is captured. This is because some server nodes are finalized or crashed after the "
"network building phase.";
safemode_ = true;
std::for_each(communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) { (void)communicator->Stop(); });
(void)std::for_each(
communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) { (void)communicator->Stop(); });
(void)communicator_with_server_->Stop();
});
}
@ -373,12 +375,12 @@ void Server::StartCommunicator() {
MS_LOG(INFO) << "This server rank is " << server_node_->rank_id();
MS_LOG(INFO) << "Start communicator with worker.";
std::for_each(communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) {
if (!communicator->Start()) {
MS_LOG(EXCEPTION) << "Starting communicator with worker failed.";
}
});
(void)std::for_each(communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) {
if (!communicator->Start()) {
MS_LOG(EXCEPTION) << "Starting communicator with worker failed.";
}
});
}
void Server::ProcessBeforeScalingOut() {
@ -424,9 +426,10 @@ void Server::ProcessAfterScalingIn() {
if (server_node_->rank_id() == UINT32_MAX) {
MS_LOG(WARNING) << "This server the one to be scaled in. Server exiting.";
std::for_each(communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) { (void)communicator->Stop(); });
communicator_with_server_->Stop();
(void)std::for_each(
communicators_with_worker_.begin(), communicators_with_worker_.end(),
[](const std::shared_ptr<ps::core::CommunicatorBase> &communicator) { (void)communicator->Stop(); });
(void)communicator_with_server_->Stop();
return;
}

5
mindspore/ccsrc/fl/worker/fl_worker.cc
View File

@ -67,7 +67,10 @@ void FLWorker::Run() {
});
InitializeFollowerScaler();
worker_node_->Start();
if (!worker_node_->Start()) {
MS_LOG(EXCEPTION) << "Starting worker node failed.";
return;
}
rank_id_ = worker_node_->rank_id();
std::this_thread::sleep_for(std::chrono::milliseconds(kWorkerSleepTimeForNetworking));

8
mindspore/ccsrc/ps/core/follower_scaler.cc
View File

@ -160,19 +160,19 @@ void FollowerScaler::ProcessAfterScaleIn() {
}
void FollowerScaler::RegisterBarrierBeforeScaleOut(const std::string &module, const BarrierBeforeScaleOut &barrier) {
barriers_before_scale_out_.try_emplace(module, barrier);
(void)barriers_before_scale_out_.try_emplace(module, barrier);
}
void FollowerScaler::RegisterBarrierBeforeScaleIn(const std::string &module, const BarrierBeforeScaleIn &barrier) {
barriers_before_scale_in_.try_emplace(module, barrier);
(void)barriers_before_scale_in_.try_emplace(module, barrier);
}
void FollowerScaler::RegisterHandlerAfterScaleOut(const std::string &module, const HandlerAfterScaleOut &handler) {
handlers_after_scale_out_.try_emplace(module, handler);
(void)handlers_after_scale_out_.try_emplace(module, handler);
}
void FollowerScaler::RegisterHandlerAfterScaleIn(const std::string &module, const HandlerAfterScaleIn &handler) {
handlers_after_scale_in_.try_emplace(module, handler);
(void)handlers_after_scale_in_.try_emplace(module, handler);
}
} // namespace core
} // namespace ps

5
mindspore/ccsrc/ps/optimizer_info.cc
View File

@ -85,6 +85,7 @@ void DenseOptimInfo::Accumulate(const Values &values, const Lengths &lengths) {
grad_offset += IntToSize(lengths[i]);
}
float *grad_data = const_cast<float *>(values.data()) + grad_offset;
MS_EXCEPTION_IF_NULL(grad_data);
#define google mindspore_private
CHECK_EQ(size, IntToSize(lengths[grad_index]));
#undef google
@ -132,6 +133,7 @@ void SparseOptimInfo::Accumulate(const Values &values, const Lengths &lengths) {
void *dst_data = accum_grad_data + grads_offset_;
void *src_data = incr_grad_data;
MS_EXCEPTION_IF_NULL(dst_data);
MS_EXCEPTION_IF_NULL(src_data);
int64_t ret = memcpy_s(dst_data, dst_size, src_data, src_size);
if (ret != 0) {
MS_LOG(EXCEPTION) << "memcpy_s error, errorno(" << ret << ")";
@ -151,9 +153,9 @@ void SparseOptimInfo::Accumulate(const Values &values, const Lengths &lengths) {
}
void *incr_indice_data_temp = const_cast<float *>(values.data()) + indice_offset;
int *incr_indice_data = reinterpret_cast<int *>(incr_indice_data_temp);
MS_EXCEPTION_IF_NULL(incr_indice_data_temp);
MS_EXCEPTION_IF_NULL(incr_indice_data);
size_t incr_indice_size = lengths[indices_index];
size_t incr_indice_data_size = incr_indice_size * sizeof(int);
@ -162,6 +164,7 @@ void SparseOptimInfo::Accumulate(const Values &values, const Lengths &lengths) {
dst_data = accum_indices_data + indices_offset_;
src_data = incr_indice_data;
MS_EXCEPTION_IF_NULL(dst_data);
MS_EXCEPTION_IF_NULL(src_data);
auto ret2 = memcpy_s(dst_data, dst_size, src_data, src_size);
if (ret2 != 0) {
MS_LOG(EXCEPTION) << "memcpy_s error, errorno(" << ret2 << ")";

79
mindspore/ccsrc/ps/optimizer_info_builder.cc
View File

@ -17,6 +17,7 @@
#include "ps/optimizer_info_builder.h"
#include <vector>
#include <memory>
#include <utility>
#include <functional>
#include "backend/kernel_compiler/cpu/ps/sparse_apply_ftrl_ps_kernel.h"
@ -43,8 +44,11 @@ void OptimizerInfoBuilder::BuildWorkspaces(OptimizerInfo *info, const std::vecto
size_t size = ws_sizes[i];
AddressPtr workspace = std::make_shared<kernel::Address>();
MS_EXCEPTION_IF_NULL(workspace);
workspace->addr = new float[size];
auto unique_buffer = std::make_unique<char[]>(sizeof(float) * size);
MS_EXCEPTION_IF_NULL(unique_buffer);
workspace->addr = unique_buffer.get();
MS_EXCEPTION_IF_NULL(workspace->addr);
(void)arrays_.emplace_back(std::move(unique_buffer));
workspace->size = size;
info->AddWorkspace(workspace);
}
@ -78,6 +82,7 @@ AddressPtr OptimizerInfoBuilder::GenInputAddrPtr(const std::string &optim_type,
// addr_data_size should be calculated by inputs_shape if it's passed.
size_t origin_index = origin_input_map.at(input_name);
EXC_IF_VEC_IDX_OOB((*inputs_shape), origin_index);
MS_EXCEPTION_IF_NULL((*inputs_shape)[origin_index]);
auto shape = *((*inputs_shape)[origin_index]);
addr_data_size = std::accumulate(shape.begin(), shape.end(), worker_num_, std::multiplies<size_t>());
} else {
@ -88,12 +93,14 @@ AddressPtr OptimizerInfoBuilder::GenInputAddrPtr(const std::string &optim_type,
IntToSize(std::accumulate(ps_lens.begin(), ps_lens.begin() + SizeToInt(ps_index), 0, std::plus<int>()));
// The size in ps_lens instead of addr_data_size is the size of real data.
T *buffer = new T[addr_data_size];
auto unique_buffer = std::make_unique<char[]>(sizeof(T) * addr_data_size);
MS_EXCEPTION_IF_NULL(unique_buffer);
addr_ptr->size = IntToSize(ps_lens[ps_index]) * sizeof(T);
addr_ptr->addr = buffer;
addr_ptr->addr = unique_buffer.get();
(void)arrays_.emplace_back(std::move(unique_buffer));
size_t dst_size = addr_ptr->size;
size_t src_size = addr_ptr->size;
size_t src_size = IntToSize(ps_lens[ps_index]) * sizeof(T);
void *dst_data = addr_ptr->addr;
void *src_data = reinterpret_cast<T *>(ps_data) + addr_data_offset;
MS_EXCEPTION_IF_NULL(dst_data);
@ -101,8 +108,6 @@ AddressPtr OptimizerInfoBuilder::GenInputAddrPtr(const std::string &optim_type,
int64_t ret = memcpy_s(dst_data, dst_size, src_data, src_size);
if (ret != 0) {
MS_LOG(EXCEPTION) << "memcpy_s error, errorno(" << ret << ")";
delete[] buffer;
buffer = nullptr;
return nullptr;
}
return addr_ptr;
@ -118,20 +123,26 @@ OptimizerInfo *MomentumOptimInfoBuilder::BuildInputs(const WeightPtr &weight, co
AddressPtr accumulate = std::make_shared<kernel::Address>();
MS_EXCEPTION_IF_NULL(accumulate);
accumulate->addr = new float[weight->size()];
auto unique_buffer = std::make_unique<char[]>(sizeof(float) * weight->size());
MS_EXCEPTION_IF_NULL(unique_buffer);
accumulate->addr = unique_buffer.get();
MS_EXCEPTION_IF_NULL(accumulate->addr);
accumulate->size = weight->size() * sizeof(float);
(void)arrays_.emplace_back(std::move(unique_buffer));
accumulate->size = sizeof(float) * weight->size();
int64_t ret = memset_s(accumulate->addr, accumulate->size, 0x00, accumulate->size);
if (ret != 0) {
MS_LOG(EXCEPTION) << "memset_s error, errorno(" << ret << ")";
delete[] reinterpret_cast<float *>(accumulate->addr);
accumulate->addr = nullptr;
return nullptr;
}
AddressPtr learning_rate = GenInputAddrPtr<float>(kApplyMomentum, "lr", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(learning_rate);
AddressPtr gradient = GenInputAddrPtr<float>(kApplyMomentum, "grad", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(gradient);
AddressPtr momentum = GenInputAddrPtr<float>(kApplyMomentum, "momentum", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(momentum);
return new MomentumOptimInfo(weight_addr, accumulate, learning_rate, gradient, momentum);
}
@ -145,41 +156,53 @@ OptimizerInfo *SparseAdamOptimInfoBuilder::BuildInputs(const WeightPtr &weight,
AddressPtr m = std::make_shared<kernel::Address>();
MS_EXCEPTION_IF_NULL(m);
m->addr = new float[weight->size()];
auto unique_buffer = std::make_unique<char[]>(sizeof(float) * weight->size());
MS_EXCEPTION_IF_NULL(unique_buffer);
m->addr = unique_buffer.get();
MS_EXCEPTION_IF_NULL(m->addr);
(void)arrays_.emplace_back(std::move(unique_buffer));
m->size = weight->size() * sizeof(float);
int64_t ret = memset_s(m->addr, m->size, 0x00, m->size);
if (ret != 0) {
MS_LOG(EXCEPTION) << "memset_s error, errorno(" << ret << ")";
delete[] reinterpret_cast<float *>(m->addr);
m->addr = nullptr;
return nullptr;
}
AddressPtr v = std::make_shared<kernel::Address>();
MS_EXCEPTION_IF_NULL(v);
v->addr = new float[weight->size()];
unique_buffer = std::make_unique<char[]>(sizeof(float) * weight->size());
MS_EXCEPTION_IF_NULL(unique_buffer);
v->addr = unique_buffer.get();
MS_EXCEPTION_IF_NULL(v->addr);
(void)arrays_.emplace_back(std::move(unique_buffer));
v->size = weight->size() * sizeof(float);
ret = memset_s(v->addr, v->size, 0x00, v->size);
if (ret != 0) {
MS_LOG(EXCEPTION) << "memset_s error, errorno(" << ret << ")";
delete[] reinterpret_cast<float *>(v->addr);
v->addr = nullptr;
delete[] reinterpret_cast<float *>(m->addr);
m->addr = nullptr;
return nullptr;
}
AddressPtr beta1_power = GenInputAddrPtr<float>(kSparseAdam, "beta1_power", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(beta1_power);
AddressPtr beta2_power = GenInputAddrPtr<float>(kSparseAdam, "beta2_power", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(beta2_power);
AddressPtr learning_rate = GenInputAddrPtr<float>(kSparseAdam, "lr", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(learning_rate);
AddressPtr beta1 = GenInputAddrPtr<float>(kSparseAdam, "beta1", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(beta1);
AddressPtr beta2 = GenInputAddrPtr<float>(kSparseAdam, "beta2", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(beta2);
AddressPtr epsilon = GenInputAddrPtr<float>(kSparseAdam, "eps", const_cast<float *>(values.data()), lens);
MS_EXCEPTION_IF_NULL(epsilon);
AddressPtr grad = GenInputAddrPtr<float>(kSparseAdam, "grad", const_cast<float *>(values.data()), lens, inputs_shape);
MS_EXCEPTION_IF_NULL(grad);
AddressPtr indices =
GenInputAddrPtr<float>(kSparseAdam, "indices", const_cast<float *>(values.data()), lens, inputs_shape);
MS_EXCEPTION_IF_NULL(indices);
return new SparseAdamOptimInfo(weight_addr, m, v, beta1_power, beta2_power, learning_rate, beta1, beta2, epsilon,
grad, indices, sharded);
}
@ -196,8 +219,13 @@ OptimizerInfo *SparseFtrlOptimInfoBuilder::BuildInputs(const WeightPtr &weight,
AddressPtr accum = std::make_shared<kernel::Address>();
MS_EXCEPTION_IF_NULL(accum);
accum->addr = new float[weight->size()];
auto unique_buffer = std::make_unique<char[]>(sizeof(float) * weight->size());
MS_EXCEPTION_IF_NULL(unique_buffer);
accum->addr = unique_buffer.get();
MS_EXCEPTION_IF_NULL(accum->addr);
(void)arrays_.emplace_back(std::move(unique_buffer));
accum->size = weight->size() * sizeof(float);
for (size_t i = 0; i < weight->size(); i++) {
float *tmp = reinterpret_cast<float *>(accum->addr);
@ -206,20 +234,25 @@ OptimizerInfo *SparseFtrlOptimInfoBuilder::BuildInputs(const WeightPtr &weight,
AddressPtr linear = std::make_shared<kernel::Address>();
MS_EXCEPTION_IF_NULL(linear);
linear->addr = new float[weight->size()];
unique_buffer = std::make_unique<char[]>(sizeof(float) * weight->size());
MS_EXCEPTION_IF_NULL(unique_buffer);
linear->addr = unique_buffer.get();
MS_EXCEPTION_IF_NULL(linear->addr);
(void)arrays_.emplace_back(std::move(unique_buffer));
linear->size = weight->size() * sizeof(float);
int64_t ret = memset_s(linear->addr, weight->size() * sizeof(float), 0x00, weight->size() * sizeof(float));
if (ret != 0) {
MS_LOG(EXCEPTION) << "memset_s error, errorno(" << ret << ")";
delete[] reinterpret_cast<float *>(linear->addr);
linear->addr = nullptr;
return nullptr;
}
linear->size = weight->size() * sizeof(float);
AddressPtr grad = GenInputAddrPtr<float>(kSparseFtrl, "grad", const_cast<float *>(values.data()), lens, inputs_shape);
MS_EXCEPTION_IF_NULL(grad);
AddressPtr indices =
GenInputAddrPtr<float>(kSparseFtrl, "indices", const_cast<float *>(values.data()), lens, inputs_shape);
MS_EXCEPTION_IF_NULL(indices);
return new SparseFtrlOptimInfo(weight_addr, accum, linear, grad, indices, sharded);
}
} // namespace ps

2
mindspore/ccsrc/ps/optimizer_info_builder.h
View File

@ -48,6 +48,8 @@ class OptimizerInfoBuilder {
template <typename T>
AddressPtr GenInputAddrPtr(const std::string &optim_type, const std::string &input_name, void *ps_data,
const Lengths &lens, const InputsShapePtr &inputs_shape = nullptr);
std::vector<std::unique_ptr<char[]>> arrays_;
size_t worker_num_;
};

2
mindspore/ccsrc/ps/util.cc
View File

@ -166,7 +166,7 @@ void Util::DoFusion(const FuncGraphPtr &func_graph, const std::string &cnode_nam
MS_EXCEPTION_IF_NULL(prim);
std::vector<AnfNodePtr> fused_node_inputs = {};
fused_node_inputs.push_back(NewValueNode(prim));
std::for_each(single_nodes.begin(), single_nodes.end(), [&](AnfNodePtr node) {
(void)std::for_each(single_nodes.begin(), single_nodes.end(), [&](AnfNodePtr node) {
fused_node_inputs.push_back(AnfAlgo::GetInputNode(node->cast<CNodePtr>(), 0));
});