forked from huawei/mindspore2022
!21793 Docking with cloud platform
Merge pull request !21793 from ZPaC/add-iteration-metrics
This commit is contained in:
Gitee
commit
e885ea1b2f
|
|
@ -30,7 +30,7 @@
|
|||
|
||||
namespace mindspore {
|
||||
namespace kernel {
|
||||
// The duration between two downloading requests when return code is ResponseCode_SucNotReady.
|
||||
// The duration between two PullWeights requests when return code is ResponseCode_SucNotReady.
|
||||
constexpr int kRetryDurationOfPullWeights = 200;
|
||||
template <typename T>
|
||||
class FusedPullWeightKernel : public CPUKernel {
|
||||
|
|
@ -51,9 +51,12 @@ class FusedPullWeightKernel : public CPUKernel {
|
|||
MS_EXCEPTION_IF_NULL(fbb);
|
||||
|
||||
total_iteration_++;
|
||||
uint64_t step_num_per_iteration = fl::worker::FLWorker::GetInstance().worker_step_num_per_iteration();
|
||||
// The worker has to train kWorkerTrainStepNum standalone iterations before it communicates with server.
|
||||
if (total_iteration_ % fl::worker::FLWorker::GetInstance().worker_step_num_per_iteration() !=
|
||||
fl::kTrainBeginStepNum) {
|
||||
MS_LOG(INFO) << "Try to pull weights. Local step number: " << total_iteration_
|
||||
<< ", step number needs to run per iteration: " << step_num_per_iteration;
|
||||
if (step_num_per_iteration != fl::kOneStepPerIteration &&
|
||||
total_iteration_ % step_num_per_iteration != fl::kTrainBeginStepNum) {
|
||||
return true;
|
||||
}
|
||||
|
||||
|
|
@ -77,6 +80,7 @@ class FusedPullWeightKernel : public CPUKernel {
|
|||
0, fbb->GetBufferPointer(), fbb->GetSize(), ps::core::TcpUserCommand::kPullWeight, &pull_weight_rsp_msg)) {
|
||||
MS_LOG(WARNING) << "Sending request for FusedPullWeight to server 0 failed. Retry later.";
|
||||
retcode = schema::ResponseCode_SucNotReady;
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kRetryDurationOfPullWeights));
|
||||
continue;
|
||||
}
|
||||
MS_EXCEPTION_IF_NULL(pull_weight_rsp_msg);
|
||||
|
|
@ -116,7 +120,7 @@ class FusedPullWeightKernel : public CPUKernel {
|
|||
return false;
|
||||
}
|
||||
}
|
||||
MS_LOG(INFO) << "Pull weights for " << weight_full_names_ << " succeed. Iteration: " << fl_iteration_;
|
||||
MS_LOG(INFO) << "Pull weights for " << weight_full_names_ << " success. Iteration: " << fl_iteration_;
|
||||
fl::worker::FLWorker::GetInstance().SetIterationRunning();
|
||||
return true;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@
|
|||
|
||||
namespace mindspore {
|
||||
namespace kernel {
|
||||
// The duration between two uploading requests when return code is ResponseCode_SucNotReady.
|
||||
// The duration between two PushWeights requests when return code is ResponseCode_SucNotReady.
|
||||
constexpr int kRetryDurationOfPushWeights = 200;
|
||||
template <typename T>
|
||||
class FusedPushWeightKernel : public CPUKernel {
|
||||
|
|
@ -49,9 +49,12 @@ class FusedPushWeightKernel : public CPUKernel {
|
|||
MS_EXCEPTION_IF_NULL(fbb);
|
||||
|
||||
total_iteration_++;
|
||||
uint64_t step_num_per_iteration = fl::worker::FLWorker::GetInstance().worker_step_num_per_iteration();
|
||||
// The worker has to train kWorkerTrainStepNum standalone iterations before it communicates with server.
|
||||
if (total_iteration_ % fl::worker::FLWorker::GetInstance().worker_step_num_per_iteration() !=
|
||||
fl::kTrainBeginStepNum) {
|
||||
MS_LOG(INFO) << "Try to push weights. Local step number: " << total_iteration_
|
||||
<< ", step number needs to run per iteration: " << step_num_per_iteration;
|
||||
if (step_num_per_iteration != fl::kOneStepPerIteration &&
|
||||
total_iteration_ % step_num_per_iteration != fl::kTrainEndStepNum) {
|
||||
return true;
|
||||
}
|
||||
|
||||
|
|
@ -76,9 +79,9 @@ class FusedPushWeightKernel : public CPUKernel {
|
|||
if (!fl::worker::FLWorker::GetInstance().SendToServer(i, fbb->GetBufferPointer(), fbb->GetSize(),
|
||||
ps::core::TcpUserCommand::kPushWeight,
|
||||
&push_weight_rsp_msg)) {
|
||||
MS_LOG(WARNING) << "Sending request for FusedPushWeight to server " << i
|
||||
<< " failed. This iteration is dropped.";
|
||||
MS_LOG(WARNING) << "Sending request for FusedPushWeight to server " << i << " failed.";
|
||||
retcode = schema::ResponseCode_SucNotReady;
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kRetryDurationOfPushWeights));
|
||||
continue;
|
||||
}
|
||||
MS_EXCEPTION_IF_NULL(push_weight_rsp_msg);
|
||||
|
|
@ -105,8 +108,7 @@ class FusedPushWeightKernel : public CPUKernel {
|
|||
}
|
||||
}
|
||||
|
||||
MS_LOG(INFO) << "Push weights for " << weight_full_names_ << " succeed. Iteration: " << fl_iteration_;
|
||||
fl::worker::FLWorker::GetInstance().SetIterationCompleted();
|
||||
MS_LOG(INFO) << "Push weights for " << weight_full_names_ << " success. Iteration: " << fl_iteration_;
|
||||
return true;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -187,6 +187,7 @@ constexpr size_t kCipherMgrMaxTaskNum = 64;
|
|||
constexpr size_t kExecutorThreadPoolSize = 32;
|
||||
constexpr size_t kExecutorMaxTaskNum = 32;
|
||||
constexpr int kHttpSuccess = 200;
|
||||
constexpr uint32_t kThreadSleepTime = 50;
|
||||
constexpr auto kPBProtocol = "PB";
|
||||
constexpr auto kFBSProtocol = "FBS";
|
||||
constexpr auto kSuccess = "Success";
|
||||
|
|
|
|||
|
|
@ -51,6 +51,18 @@ bool Executor::ReInitForScaling() {
|
|||
return true;
|
||||
}
|
||||
|
||||
bool Executor::ReInitForUpdatingHyperParams(size_t aggr_threshold) {
|
||||
aggregation_count_ = aggr_threshold;
|
||||
auto result = std::find_if(param_aggrs_.begin(), param_aggrs_.end(), [this](auto param_aggr) {
|
||||
return !param_aggr.second->ReInitForUpdatingHyperParams(aggregation_count_);
|
||||
});
|
||||
if (result != param_aggrs_.end()) {
|
||||
MS_LOG(ERROR) << "Reinitializing aggregator of " << result->first << " for scaling failed.";
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Executor::initialized() const { return initialized_; }
|
||||
|
||||
bool Executor::HandlePush(const std::string ¶m_name, const UploadData &upload_data) {
|
||||
|
|
|
|||
|
|
@ -33,8 +33,6 @@
|
|||
namespace mindspore {
|
||||
namespace fl {
|
||||
namespace server {
|
||||
constexpr int kThreadSleepTime = 5;
|
||||
|
||||
// Executor is the entrance for server to handle aggregation, optimizing, model querying, etc. It handles
|
||||
// logics relevant to kernel launching.
|
||||
class Executor {
|
||||
|
|
@ -53,6 +51,9 @@ class Executor {
|
|||
// Reinitialize parameter aggregators after scaling operations are done.
|
||||
bool ReInitForScaling();
|
||||
|
||||
// After hyper-parameters are updated, some parameter aggregators should be reinitialized.
|
||||
bool ReInitForUpdatingHyperParams(size_t aggr_threshold);
|
||||
|
||||
// Called in parameter server training mode to do Push operation.
|
||||
// For the same trainable parameter, HandlePush method must be called aggregation_count_ times before it's considered
|
||||
// as completed.
|
||||
|
|
|
|||
|
|
@ -26,6 +26,14 @@ namespace mindspore {
|
|||
namespace fl {
|
||||
namespace server {
|
||||
class Server;
|
||||
|
||||
Iteration::~Iteration() {
|
||||
move_to_next_thread_running_ = false;
|
||||
if (move_to_next_thread_.joinable()) {
|
||||
move_to_next_thread_.join();
|
||||
}
|
||||
}
|
||||
|
||||
void Iteration::RegisterMessageCallback(const std::shared_ptr<ps::core::TcpCommunicator> &communicator) {
|
||||
MS_EXCEPTION_IF_NULL(communicator);
|
||||
communicator_ = communicator;
|
||||
|
|
@ -79,9 +87,27 @@ void Iteration::InitRounds(const std::vector<std::shared_ptr<ps::core::Communica
|
|||
});
|
||||
LocalMetaStore::GetInstance().put_value(kCtxTotalTimeoutDuration, iteration_time_window);
|
||||
MS_LOG(INFO) << "Time window for one iteration is " << iteration_time_window;
|
||||
|
||||
// Initialize the thread which will handle the signal from round kernels.
|
||||
move_to_next_thread_ = std::thread([this]() {
|
||||
while (move_to_next_thread_running_.load()) {
|
||||
std::unique_lock<std::mutex> lock(next_iteration_mutex_);
|
||||
next_iteration_cv_.wait(lock);
|
||||
MoveToNextIteration(is_last_iteration_valid_, move_to_next_reason_);
|
||||
}
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
void Iteration::ClearRounds() { rounds_.clear(); }
|
||||
|
||||
void Iteration::NotifyNext(bool is_last_iter_valid, const std::string &reason) {
|
||||
std::unique_lock<std::mutex> lock(next_iteration_mutex_);
|
||||
is_last_iteration_valid_ = is_last_iter_valid;
|
||||
move_to_next_reason_ = reason;
|
||||
next_iteration_cv_.notify_one();
|
||||
}
|
||||
|
||||
void Iteration::MoveToNextIteration(bool is_last_iter_valid, const std::string &reason) {
|
||||
MS_LOG(INFO) << "Notify cluster starts to proceed to next iteration. Iteration is " << iteration_num_
|
||||
<< " validation is " << is_last_iter_valid << ". Reason: " << reason;
|
||||
|
|
@ -119,7 +145,10 @@ void Iteration::SetIterationRunning() {
|
|||
// This event helps worker/server to be consistent in iteration state.
|
||||
server_node_->BroadcastEvent(static_cast<uint32_t>(ps::CustomEvent::kIterationRunning));
|
||||
}
|
||||
|
||||
std::unique_lock<std::mutex> lock(iteration_state_mtx_);
|
||||
iteration_state_ = IterationState::kRunning;
|
||||
start_timestamp_ = LongToUlong(CURRENT_TIME_MILLI.count());
|
||||
}
|
||||
|
||||
void Iteration::SetIterationCompleted() {
|
||||
|
|
@ -129,13 +158,17 @@ void Iteration::SetIterationCompleted() {
|
|||
// This event helps worker/server to be consistent in iteration state.
|
||||
server_node_->BroadcastEvent(static_cast<uint32_t>(ps::CustomEvent::kIterationCompleted));
|
||||
}
|
||||
|
||||
std::unique_lock<std::mutex> lock(iteration_state_mtx_);
|
||||
iteration_state_ = IterationState::kCompleted;
|
||||
complete_timestamp_ = LongToUlong(CURRENT_TIME_MILLI.count());
|
||||
}
|
||||
|
||||
void Iteration::ScalingBarrier() {
|
||||
MS_LOG(INFO) << "Starting Iteration scaling barrier.";
|
||||
while (iteration_state_.load() != IterationState::kCompleted) {
|
||||
std::this_thread::yield();
|
||||
std::unique_lock<std::mutex> lock(iteration_state_mtx_);
|
||||
if (iteration_state_.load() != IterationState::kCompleted) {
|
||||
iteration_state_cv_.wait(lock);
|
||||
}
|
||||
MS_LOG(INFO) << "Ending Iteration scaling barrier.";
|
||||
}
|
||||
|
|
@ -156,14 +189,144 @@ bool Iteration::ReInitForScaling(uint32_t server_num, uint32_t server_rank) {
|
|||
return true;
|
||||
}
|
||||
|
||||
bool Iteration::ReInitForUpdatingHyperParams(const std::vector<RoundConfig> &updated_rounds_config) {
|
||||
for (const auto &updated_round : updated_rounds_config) {
|
||||
for (const auto &round : rounds_) {
|
||||
if (updated_round.name == round->name()) {
|
||||
MS_LOG(INFO) << "Reinitialize for round " << round->name();
|
||||
if (!round->ReInitForUpdatingHyperParams(updated_round.threshold_count, updated_round.time_window)) {
|
||||
MS_LOG(ERROR) << "Reinitializing for round " << round->name() << " failed.";
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
const std::vector<std::shared_ptr<Round>> &Iteration::rounds() const { return rounds_; }
|
||||
|
||||
bool Iteration::is_last_iteration_valid() const { return is_last_iteration_valid_; }
|
||||
|
||||
void Iteration::set_metrics(const std::shared_ptr<IterationMetrics> &metrics) { metrics_ = metrics; }
|
||||
|
||||
void Iteration::set_loss(float loss) { loss_ = loss; }
|
||||
|
||||
void Iteration::set_accuracy(float accuracy) { accuracy_ = accuracy; }
|
||||
|
||||
InstanceState Iteration::instance_state() const { return instance_state_.load(); }
|
||||
|
||||
bool Iteration::EnableServerInstance(std::string *result) {
|
||||
MS_ERROR_IF_NULL_W_RET_VAL(result, false);
|
||||
// Before enabling server instance, we should judge whether this request should be handled.
|
||||
std::unique_lock<std::mutex> lock(instance_mtx_);
|
||||
if (is_instance_being_updated_) {
|
||||
*result = "The instance is being updated. Please retry enabling server later.";
|
||||
MS_LOG(WARNING) << result;
|
||||
return false;
|
||||
}
|
||||
if (instance_state_.load() == InstanceState::kFinish) {
|
||||
*result = "The instance is completed. Please do not enabling server now.";
|
||||
MS_LOG(WARNING) << result;
|
||||
return false;
|
||||
}
|
||||
|
||||
// Start enabling server instance.
|
||||
is_instance_being_updated_ = true;
|
||||
|
||||
instance_state_ = InstanceState::kRunning;
|
||||
*result = "Enabling FL-Server succeeded.";
|
||||
|
||||
// End enabling server instance.
|
||||
is_instance_being_updated_ = false;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Iteration::DisableServerInstance(std::string *result) {
|
||||
MS_ERROR_IF_NULL_W_RET_VAL(result, false);
|
||||
// Before disabling server instance, we should judge whether this request should be handled.
|
||||
std::unique_lock<std::mutex> lock(instance_mtx_);
|
||||
if (is_instance_being_updated_) {
|
||||
*result = "The instance is being updated. Please retry disabling server later.";
|
||||
MS_LOG(WARNING) << *result;
|
||||
return false;
|
||||
}
|
||||
if (instance_state_.load() == InstanceState::kFinish) {
|
||||
*result = "The instance is completed. Please do not disabling server now.";
|
||||
MS_LOG(WARNING) << *result;
|
||||
return false;
|
||||
}
|
||||
if (instance_state_.load() == InstanceState::kDisable) {
|
||||
*result = "Disabling FL-Server succeeded.";
|
||||
MS_LOG(INFO) << *result;
|
||||
return false;
|
||||
}
|
||||
|
||||
// Start disabling server instance.
|
||||
is_instance_being_updated_ = true;
|
||||
|
||||
// If instance is running, we should drop current iteration and move to the next.
|
||||
instance_state_ = InstanceState::kDisable;
|
||||
if (!ForciblyMoveToNextIteration()) {
|
||||
*result = "Disabling instance failed. Can't drop current iteration and move to the next.";
|
||||
MS_LOG(ERROR) << result;
|
||||
return false;
|
||||
}
|
||||
|
||||
// End disabling server instance.
|
||||
is_instance_being_updated_ = false;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Iteration::NewInstance(const nlohmann::json &new_instance_json, std::string *result) {
|
||||
MS_ERROR_IF_NULL_W_RET_VAL(result, false);
|
||||
// Before new instance, we should judge whether this request should be handled.
|
||||
std::unique_lock<std::mutex> lock(instance_mtx_);
|
||||
if (is_instance_being_updated_) {
|
||||
*result = "The instance is being updated. Please retry new instance later.";
|
||||
MS_LOG(WARNING) << *result;
|
||||
return false;
|
||||
}
|
||||
|
||||
// Start new server instance.
|
||||
is_instance_being_updated_ = true;
|
||||
|
||||
// Reset current instance.
|
||||
instance_state_ = InstanceState::kFinish;
|
||||
MS_LOG(INFO) << "Proceed to a new instance.";
|
||||
WaitAllRoundsFinish();
|
||||
for (auto &round : rounds_) {
|
||||
MS_ERROR_IF_NULL_W_RET_VAL(round, false);
|
||||
round->Reset();
|
||||
}
|
||||
iteration_num_ = 1;
|
||||
LocalMetaStore::GetInstance().set_curr_iter_num(iteration_num_);
|
||||
ModelStore::GetInstance().Reset();
|
||||
|
||||
// Update the hyper-parameters on server and reinitialize rounds.
|
||||
if (!UpdateHyperParams(new_instance_json)) {
|
||||
*result = "Updating hyper-parameters failed.";
|
||||
return false;
|
||||
}
|
||||
if (!ReInitRounds()) {
|
||||
*result = "Reinitializing rounds failed.";
|
||||
return false;
|
||||
}
|
||||
|
||||
instance_state_ = InstanceState::kRunning;
|
||||
*result = "New FL-Server instance succeeded.";
|
||||
|
||||
// End new server instance.
|
||||
is_instance_being_updated_ = false;
|
||||
return true;
|
||||
}
|
||||
|
||||
void Iteration::WaitAllRoundsFinish() {
|
||||
while (running_round_num_.load() != 0) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kThreadSleepTime));
|
||||
}
|
||||
}
|
||||
|
||||
bool Iteration::SyncIteration(uint32_t rank) {
|
||||
MS_ERROR_IF_NULL_W_RET_VAL(communicator_, false);
|
||||
SyncIterationRequest sync_iter_req;
|
||||
|
|
@ -320,6 +483,7 @@ void Iteration::HandlePrepareForNextIterRequest(const std::shared_ptr<ps::core::
|
|||
void Iteration::PrepareForNextIter() {
|
||||
MS_LOG(INFO) << "Prepare for next iteration. Switch the server to safemode.";
|
||||
Server::GetInstance().SwitchToSafeMode();
|
||||
WaitAllRoundsFinish();
|
||||
}
|
||||
|
||||
bool Iteration::BroadcastMoveToNextIterRequest(bool is_last_iter_valid, const std::string &reason) {
|
||||
|
|
@ -436,24 +600,134 @@ void Iteration::HandleEndLastIterRequest(const std::shared_ptr<ps::core::Message
|
|||
|
||||
void Iteration::EndLastIter() {
|
||||
MS_LOG(INFO) << "End the last iteration " << iteration_num_;
|
||||
iteration_num_++;
|
||||
// After the job is done, reset the iteration to the initial number and reset ModelStore.
|
||||
if (iteration_num_ > ps::PSContext::instance()->fl_iteration_num()) {
|
||||
if (iteration_num_ == ps::PSContext::instance()->fl_iteration_num()) {
|
||||
MS_LOG(INFO) << "Iteration loop " << iteration_loop_count_
|
||||
<< " is completed. Iteration number: " << ps::PSContext::instance()->fl_iteration_num();
|
||||
iteration_num_ = 1;
|
||||
iteration_loop_count_++;
|
||||
ModelStore::GetInstance().Reset();
|
||||
instance_state_ = InstanceState::kFinish;
|
||||
}
|
||||
|
||||
std::unique_lock<std::mutex> lock(pinned_mtx_);
|
||||
pinned_iter_num_ = 0;
|
||||
lock.unlock();
|
||||
|
||||
SetIterationCompleted();
|
||||
SummarizeIteration();
|
||||
iteration_num_++;
|
||||
LocalMetaStore::GetInstance().set_curr_iter_num(iteration_num_);
|
||||
Server::GetInstance().CancelSafeMode();
|
||||
SetIterationCompleted();
|
||||
iteration_state_cv_.notify_all();
|
||||
MS_LOG(INFO) << "Move to next iteration:" << iteration_num_ << "\n";
|
||||
}
|
||||
|
||||
bool Iteration::ForciblyMoveToNextIteration() {
|
||||
NotifyNext(false, "Forcibly move to next iteration.");
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Iteration::SummarizeIteration() {
|
||||
// If the metrics_ is not initialized or the server is not the leader server, do not summarize.
|
||||
if (server_node_->rank_id() != kLeaderServerRank || metrics_ == nullptr) {
|
||||
MS_LOG(INFO) << "This server will not summarize for iteration.";
|
||||
return true;
|
||||
}
|
||||
|
||||
metrics_->set_fl_name(ps::PSContext::instance()->fl_name());
|
||||
metrics_->set_fl_iteration_num(ps::PSContext::instance()->fl_iteration_num());
|
||||
metrics_->set_cur_iteration_num(iteration_num_ - 1);
|
||||
metrics_->set_instance_state(instance_state_.load());
|
||||
metrics_->set_loss(loss_);
|
||||
metrics_->set_accuracy(accuracy_);
|
||||
// The joined client number is equal to the threshold of updateModel.
|
||||
size_t update_model_threshold = static_cast<size_t>(
|
||||
std::ceil(ps::PSContext::instance()->start_fl_job_threshold() * ps::PSContext::instance()->update_model_ratio()));
|
||||
metrics_->set_joined_client_num(update_model_threshold);
|
||||
// The rejected client number is equal to threshold of startFLJob minus threshold of updateModel.
|
||||
metrics_->set_rejected_client_num(ps::PSContext::instance()->start_fl_job_threshold() - update_model_threshold);
|
||||
|
||||
if (complete_timestamp_ < start_timestamp_) {
|
||||
MS_LOG(ERROR) << "The complete_timestamp_: " << complete_timestamp_ << ", start_timestamp_: " << start_timestamp_
|
||||
<< ". One of them is invalid.";
|
||||
metrics_->set_iteration_time_cost(UINT64_MAX);
|
||||
} else {
|
||||
metrics_->set_iteration_time_cost(complete_timestamp_ - start_timestamp_);
|
||||
}
|
||||
|
||||
metrics_->Summarize();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Iteration::UpdateHyperParams(const nlohmann::json &json) {
|
||||
for (const auto &item : json.items()) {
|
||||
std::string key = item.key();
|
||||
if (key == "start_fl_job_threshold") {
|
||||
ps::PSContext::instance()->set_start_fl_job_threshold(item.value().get<uint64_t>());
|
||||
continue;
|
||||
}
|
||||
if (key == "start_fl_job_time_window") {
|
||||
ps::PSContext::instance()->set_start_fl_job_time_window(item.value().get<uint64_t>());
|
||||
continue;
|
||||
}
|
||||
if (key == "update_model_ratio") {
|
||||
ps::PSContext::instance()->set_update_model_ratio(item.value().get<float>());
|
||||
continue;
|
||||
}
|
||||
if (key == "update_model_time_window") {
|
||||
ps::PSContext::instance()->set_update_model_time_window(item.value().get<uint64_t>());
|
||||
continue;
|
||||
}
|
||||
if (key == "fl_iteration_num") {
|
||||
ps::PSContext::instance()->set_fl_iteration_num(item.value().get<uint64_t>());
|
||||
continue;
|
||||
}
|
||||
if (key == "client_epoch_num") {
|
||||
ps::PSContext::instance()->set_client_epoch_num(item.value().get<uint64_t>());
|
||||
continue;
|
||||
}
|
||||
if (key == "client_batch_size") {
|
||||
ps::PSContext::instance()->set_client_batch_size(item.value().get<uint64_t>());
|
||||
continue;
|
||||
}
|
||||
if (key == "client_learning_rate") {
|
||||
ps::PSContext::instance()->set_client_learning_rate(item.value().get<float>());
|
||||
continue;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool Iteration::ReInitRounds() {
|
||||
size_t start_fl_job_threshold = ps::PSContext::instance()->start_fl_job_threshold();
|
||||
float update_model_ratio = ps::PSContext::instance()->update_model_ratio();
|
||||
size_t update_model_threshold = static_cast<size_t>(std::ceil(start_fl_job_threshold * update_model_ratio));
|
||||
uint64_t start_fl_job_time_window = ps::PSContext::instance()->start_fl_job_time_window();
|
||||
uint64_t update_model_time_window = ps::PSContext::instance()->update_model_time_window();
|
||||
std::vector<RoundConfig> new_round_config = {
|
||||
{"startFLJob", true, start_fl_job_time_window, true, start_fl_job_threshold},
|
||||
{"updateModel", true, update_model_time_window, true, update_model_threshold}};
|
||||
if (!Iteration::GetInstance().ReInitForUpdatingHyperParams(new_round_config)) {
|
||||
MS_LOG(ERROR) << "Reinitializing for updating hyper-parameters failed.";
|
||||
return false;
|
||||
}
|
||||
|
||||
size_t executor_threshold = 0;
|
||||
const std::string &server_mode = ps::PSContext::instance()->server_mode();
|
||||
uint32_t worker_num = ps::PSContext::instance()->initial_worker_num();
|
||||
if (server_mode == ps::kServerModeFL || server_mode == ps::kServerModeHybrid) {
|
||||
executor_threshold = update_model_threshold;
|
||||
} else if (server_mode == ps::kServerModePS) {
|
||||
executor_threshold = worker_num;
|
||||
} else {
|
||||
MS_LOG(ERROR) << "Server mode " << server_mode << " is not supported.";
|
||||
return false;
|
||||
}
|
||||
if (!Executor::GetInstance().ReInitForUpdatingHyperParams(executor_threshold)) {
|
||||
MS_LOG(ERROR) << "Reinitializing executor failed.";
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
} // namespace server
|
||||
} // namespace fl
|
||||
} // namespace mindspore
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@
|
|||
#include "fl/server/common.h"
|
||||
#include "fl/server/round.h"
|
||||
#include "fl/server/local_meta_store.h"
|
||||
#include "fl/server/iteration_metrics.h"
|
||||
|
||||
namespace mindspore {
|
||||
namespace fl {
|
||||
|
|
@ -38,6 +39,7 @@ enum class IterationState {
|
|||
// The time duration between retrying when sending prepare for next iteration request failed.
|
||||
constexpr uint32_t kRetryDurationForPrepareForNextIter = 500;
|
||||
|
||||
class IterationMetrics;
|
||||
// In server's logic, Iteration is the minimum execution unit. For each execution, it consists of multiple kinds of
|
||||
// Rounds, only after all the rounds are finished, this iteration is considered as completed.
|
||||
class Iteration {
|
||||
|
|
@ -60,6 +62,12 @@ class Iteration {
|
|||
void InitRounds(const std::vector<std::shared_ptr<ps::core::CommunicatorBase>> &communicators,
|
||||
const TimeOutCb &timeout_cb, const FinishIterCb &finish_iteration_cb);
|
||||
|
||||
// Release all the round objects in Iteration instance. Used for reinitializing round and round kernels.
|
||||
void ClearRounds();
|
||||
|
||||
// Notify move_to_next_thread_ to move to next iteration.
|
||||
void NotifyNext(bool is_last_iter_valid, const std::string &reason);
|
||||
|
||||
// This method will control servers to proceed to next iteration.
|
||||
// There's communication between leader and follower servers in this method.
|
||||
// The server moves to next iteration only after the last round finishes or the time expires.
|
||||
|
|
@ -79,28 +87,65 @@ class Iteration {
|
|||
// The server number after scaling is required in some rounds.
|
||||
bool ReInitForScaling(uint32_t server_num, uint32_t server_rank);
|
||||
|
||||
// After hyper-parameters are updated, some rounds and kernels should be reinitialized.
|
||||
bool ReInitForUpdatingHyperParams(const std::vector<RoundConfig> &updated_rounds_config);
|
||||
|
||||
const std::vector<std::shared_ptr<Round>> &rounds() const;
|
||||
|
||||
bool is_last_iteration_valid() const;
|
||||
|
||||
// Set the instance metrics which will be called for each iteration.
|
||||
void set_metrics(const std::shared_ptr<IterationMetrics> &metrics);
|
||||
void set_loss(float loss);
|
||||
void set_accuracy(float accuracy);
|
||||
|
||||
// Return state of current training job instance.
|
||||
InstanceState instance_state() const;
|
||||
|
||||
// Return whether current instance is being updated.
|
||||
bool IsInstanceBeingUpdated() const;
|
||||
|
||||
// EnableFLS/disableFLS the current training instance.
|
||||
bool EnableServerInstance(std::string *result);
|
||||
bool DisableServerInstance(std::string *result);
|
||||
|
||||
// Finish current instance and start a new one. FLPlan could be changed in this method.
|
||||
bool NewInstance(const nlohmann::json &new_instance_json, std::string *result);
|
||||
|
||||
// Query information of current instance.
|
||||
bool QueryInstance(std::string *result);
|
||||
|
||||
// Need to wait all the rounds to finish before proceed to next iteration.
|
||||
void WaitAllRoundsFinish();
|
||||
|
||||
// The round kernels whose Launch method has not returned yet.
|
||||
std::atomic_uint32_t running_round_num_;
|
||||
|
||||
private:
|
||||
Iteration()
|
||||
: server_node_(nullptr),
|
||||
: running_round_num_(0),
|
||||
server_node_(nullptr),
|
||||
communicator_(nullptr),
|
||||
iteration_state_(IterationState::kCompleted),
|
||||
start_timestamp_(0),
|
||||
complete_timestamp_(0),
|
||||
iteration_loop_count_(0),
|
||||
iteration_num_(1),
|
||||
is_last_iteration_valid_(true),
|
||||
move_to_next_reason_(""),
|
||||
move_to_next_thread_running_(true),
|
||||
pinned_iter_num_(0),
|
||||
metrics_(nullptr),
|
||||
instance_state_(InstanceState::kRunning),
|
||||
is_instance_being_updated_(false),
|
||||
loss_(0.0),
|
||||
accuracy_(0.0) {
|
||||
accuracy_(0.0),
|
||||
joined_client_num_(0),
|
||||
rejected_client_num_(0),
|
||||
time_cost_(0) {
|
||||
LocalMetaStore::GetInstance().set_curr_iter_num(iteration_num_);
|
||||
}
|
||||
~Iteration() = default;
|
||||
~Iteration();
|
||||
Iteration(const Iteration &) = delete;
|
||||
Iteration &operator=(const Iteration &) = delete;
|
||||
|
||||
|
|
@ -138,6 +183,18 @@ class Iteration {
|
|||
// The server end the last iteration. This method will increase the iteration number and cancel the safemode.
|
||||
void EndLastIter();
|
||||
|
||||
// Drop current iteration and move to the next immediately.
|
||||
bool ForciblyMoveToNextIteration();
|
||||
|
||||
// Summarize metrics for the completed iteration, including iteration time cost, accuracy, loss, etc.
|
||||
bool SummarizeIteration();
|
||||
|
||||
// Update server's hyper-parameters according to the given serialized json(hyper_params_data).
|
||||
bool UpdateHyperParams(const nlohmann::json &json);
|
||||
|
||||
// Reinitialize rounds and round kernels.
|
||||
bool ReInitRounds();
|
||||
|
||||
std::shared_ptr<ps::core::ServerNode> server_node_;
|
||||
std::shared_ptr<ps::core::TcpCommunicator> communicator_;
|
||||
|
||||
|
|
@ -145,7 +202,11 @@ class Iteration {
|
|||
std::vector<std::shared_ptr<Round>> rounds_;
|
||||
|
||||
// The iteration is either running or completed at any time.
|
||||
std::mutex iteration_state_mtx_;
|
||||
std::condition_variable iteration_state_cv_;
|
||||
std::atomic<IterationState> iteration_state_;
|
||||
uint64_t start_timestamp_;
|
||||
uint64_t complete_timestamp_;
|
||||
|
||||
// The count of iteration loops which are completed.
|
||||
size_t iteration_loop_count_;
|
||||
|
|
@ -153,18 +214,44 @@ class Iteration {
|
|||
// Server's current iteration number.
|
||||
size_t iteration_num_;
|
||||
|
||||
// Last iteration is successfully finished.
|
||||
// Whether last iteration is successfully finished and the reason.
|
||||
bool is_last_iteration_valid_;
|
||||
std::string move_to_next_reason_;
|
||||
|
||||
// It will be notified by rounds that the instance moves to the next iteration.
|
||||
std::thread move_to_next_thread_;
|
||||
std::atomic_bool move_to_next_thread_running_;
|
||||
std::mutex next_iteration_mutex_;
|
||||
std::condition_variable next_iteration_cv_;
|
||||
|
||||
// To avoid Next method is called multiple times in one iteration, we should mark the iteration number.
|
||||
uint64_t pinned_iter_num_;
|
||||
std::mutex pinned_mtx_;
|
||||
|
||||
std::shared_ptr<IterationMetrics> metrics_;
|
||||
|
||||
// The state for current instance.
|
||||
std::atomic<InstanceState> instance_state_;
|
||||
|
||||
// Every instance is not reentrant.
|
||||
// This flag represents whether the instance is being updated.
|
||||
std::mutex instance_mtx_;
|
||||
bool is_instance_being_updated_;
|
||||
|
||||
// The training loss after this federated learning iteration, passed by worker.
|
||||
float loss_;
|
||||
|
||||
// The evaluation result after this federated learning iteration, passed by worker.
|
||||
float accuracy_;
|
||||
|
||||
// The number of clients which join the federated aggregation.
|
||||
size_t joined_client_num_;
|
||||
|
||||
// The number of clients which are not involved in federated aggregation.
|
||||
size_t rejected_client_num_;
|
||||
|
||||
// The time cost in millisecond for this completed iteration.
|
||||
uint64_t time_cost_;
|
||||
};
|
||||
} // namespace server
|
||||
} // namespace fl
|
||||
|
|
|
|||
|
|
@ -14,11 +14,11 @@
|
|||
* limitations under the License.
|
||||
*/
|
||||
|
||||
#include "fl/server/iteration_metrics.h"
|
||||
#include <string>
|
||||
#include <fstream>
|
||||
#include "debug/common.h"
|
||||
#include "ps/constants.h"
|
||||
#include "fl/server/iteration_metrics.h"
|
||||
|
||||
namespace mindspore {
|
||||
namespace fl {
|
||||
|
|
|
|||
|
|
@ -67,6 +67,8 @@ class AggregationKernel : public CPUKernel {
|
|||
// Reinitialize aggregation kernel after scaling operations are done.
|
||||
virtual bool ReInitForScaling() { return true; }
|
||||
|
||||
virtual bool ReInitForUpdatingHyperParams(size_t) { return true; }
|
||||
|
||||
// Setter and getter of kernels parameters information.
|
||||
void set_params_info(const ParamsInfo ¶ms_info) { params_info_ = params_info; }
|
||||
const std::vector<std::string> &input_names() { return params_info_.inputs_names(); }
|
||||
|
|
|
|||
|
|
@ -178,6 +178,12 @@ class FedAvgKernel : public AggregationKernel {
|
|||
return true;
|
||||
}
|
||||
|
||||
bool ReInitForUpdatingHyperParams(size_t aggr_threshold) override {
|
||||
done_count_ = aggr_threshold;
|
||||
DistributedCountService::GetInstance().RegisterCounter(name_, done_count_, {first_cnt_handler_, last_cnt_handler_});
|
||||
return true;
|
||||
}
|
||||
|
||||
private:
|
||||
void GenerateReuseKernelNodeInfo() override {
|
||||
MS_LOG(INFO) << "FedAvg reuse 'weight' of the kernel node.";
|
||||
|
|
|
|||
|
|
@ -102,7 +102,6 @@ void ModelStore::Reset() {
|
|||
initial_model_ = iteration_to_model_.rbegin()->second;
|
||||
iteration_to_model_.clear();
|
||||
iteration_to_model_[kInitIterationNum] = initial_model_;
|
||||
iteration_to_model_[kResetInitIterNum] = initial_model_;
|
||||
}
|
||||
|
||||
const std::map<size_t, std::shared_ptr<MemoryRegister>> &ModelStore::iteration_to_model() {
|
||||
|
|
|
|||
|
|
@ -60,6 +60,21 @@ bool ParameterAggregator::ReInitForScaling() {
|
|||
return true;
|
||||
}
|
||||
|
||||
bool ParameterAggregator::ReInitForUpdatingHyperParams(size_t aggr_threshold) {
|
||||
required_push_count_ = aggr_threshold;
|
||||
required_pull_count_ = aggr_threshold;
|
||||
auto result = std::find_if(aggregation_kernel_parameters_.begin(), aggregation_kernel_parameters_.end(),
|
||||
[aggr_threshold](auto aggregation_kernel) {
|
||||
MS_ERROR_IF_NULL_W_RET_VAL(aggregation_kernel.first, true);
|
||||
return !aggregation_kernel.first->ReInitForUpdatingHyperParams(aggr_threshold);
|
||||
});
|
||||
if (result != aggregation_kernel_parameters_.end()) {
|
||||
MS_LOG(ERROR) << "Reinitializing aggregation kernel after scaling failed";
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool ParameterAggregator::UpdateData(const std::map<std::string, Address> &new_data) {
|
||||
std::map<std::string, AddressPtr> &name_to_addr = memory_register_->addresses();
|
||||
for (const auto &data : new_data) {
|
||||
|
|
|
|||
|
|
@ -68,6 +68,9 @@ class ParameterAggregator {
|
|||
// Reinitialize the parameter aggregator after scaling operations are done.
|
||||
bool ReInitForScaling();
|
||||
|
||||
// After hyper-parameters are updated, some parameter aggregators should be reinitialized.
|
||||
bool ReInitForUpdatingHyperParams(size_t aggr_threshold);
|
||||
|
||||
// Update old data stored in ParameterAggregator with new data.
|
||||
// The data could have many meanings: weights, gradients, learning_rate, momentum, etc.
|
||||
bool UpdateData(const std::map<std::string, Address> &new_data);
|
||||
|
|
|
|||
|
|
@ -102,6 +102,21 @@ bool Round::ReInitForScaling(uint32_t server_num) {
|
|||
return true;
|
||||
}
|
||||
|
||||
bool Round::ReInitForUpdatingHyperParams(size_t updated_threshold_count, size_t updated_time_window) {
|
||||
time_window_ = updated_time_window;
|
||||
threshold_count_ = updated_threshold_count;
|
||||
if (check_count_) {
|
||||
auto first_count_handler = std::bind(&Round::OnFirstCountEvent, this, std::placeholders::_1);
|
||||
auto last_count_handler = std::bind(&Round::OnLastCountEvent, this, std::placeholders::_1);
|
||||
DistributedCountService::GetInstance().RegisterCounter(name_, threshold_count_,
|
||||
{first_count_handler, last_count_handler});
|
||||
}
|
||||
|
||||
MS_ERROR_IF_NULL_W_RET_VAL(kernel_, false);
|
||||
kernel_->InitKernel(threshold_count_);
|
||||
return true;
|
||||
}
|
||||
|
||||
void Round::BindRoundKernel(const std::shared_ptr<kernel::RoundKernel> &kernel) {
|
||||
MS_EXCEPTION_IF_NULL(kernel);
|
||||
kernel_ = kernel;
|
||||
|
|
@ -114,10 +129,9 @@ void Round::LaunchRoundKernel(const std::shared_ptr<ps::core::MessageHandler> &m
|
|||
MS_ERROR_IF_NULL_WO_RET_VAL(message);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(kernel_);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(communicator_);
|
||||
// If the server is still in the process of scaling, refuse the request.
|
||||
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.";
|
||||
|
||||
std::string reason = "";
|
||||
if (!IsServerAvailable(&reason)) {
|
||||
if (!communicator_->SendResponse(reason.c_str(), reason.size(), message)) {
|
||||
MS_LOG(ERROR) << "Sending response failed.";
|
||||
return;
|
||||
|
|
@ -125,6 +139,7 @@ void Round::LaunchRoundKernel(const std::shared_ptr<ps::core::MessageHandler> &m
|
|||
return;
|
||||
}
|
||||
|
||||
Iteration::GetInstance().running_round_num_++;
|
||||
AddressPtr input = std::make_shared<Address>();
|
||||
AddressPtr output = std::make_shared<Address>();
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(input);
|
||||
|
|
@ -133,7 +148,7 @@ void Round::LaunchRoundKernel(const std::shared_ptr<ps::core::MessageHandler> &m
|
|||
input->size = message->len();
|
||||
bool ret = kernel_->Launch({input}, {}, {output});
|
||||
if (output->size == 0) {
|
||||
std::string reason = "The output of the round " + name_ + " is empty.";
|
||||
reason = "The output of the round " + name_ + " is empty.";
|
||||
MS_LOG(WARNING) << reason;
|
||||
if (!communicator_->SendResponse(reason.c_str(), reason.size(), message)) {
|
||||
MS_LOG(ERROR) << "Sending response failed.";
|
||||
|
|
@ -149,9 +164,10 @@ void Round::LaunchRoundKernel(const std::shared_ptr<ps::core::MessageHandler> &m
|
|||
|
||||
// Must send response back no matter what value Launch method returns.
|
||||
if (!ret) {
|
||||
std::string reason = "Launching round kernel of round " + name_ + " failed.";
|
||||
Iteration::GetInstance().MoveToNextIteration(false, reason);
|
||||
reason = "Launching round kernel of round " + name_ + " failed.";
|
||||
Iteration::GetInstance().NotifyNext(false, reason);
|
||||
}
|
||||
Iteration::GetInstance().running_round_num_--;
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -195,6 +211,30 @@ void Round::OnLastCountEvent(const std::shared_ptr<ps::core::MessageHandler> &me
|
|||
kernel_->OnLastCountEvent(message);
|
||||
return;
|
||||
}
|
||||
|
||||
bool Round::IsServerAvailable(std::string *reason) {
|
||||
MS_ERROR_IF_NULL_W_RET_VAL(reason, false);
|
||||
// After one instance is completed, the model should be accessed by clients.
|
||||
if (Iteration::GetInstance().instance_state() == InstanceState::kFinish && name_ == "getModel") {
|
||||
return true;
|
||||
}
|
||||
|
||||
// If the server state is Disable or Finish, refuse the request.
|
||||
if (Iteration::GetInstance().instance_state() == InstanceState::kDisable ||
|
||||
Iteration::GetInstance().instance_state() == InstanceState::kFinish) {
|
||||
MS_LOG(WARNING) << "The server's training job is disabled or finished, please retry " + name_ + " later.";
|
||||
*reason = ps::kJobNotAvailable;
|
||||
return false;
|
||||
}
|
||||
|
||||
// If the server is still in the process of scaling, reject the request.
|
||||
if (Server::GetInstance().IsSafeMode()) {
|
||||
MS_LOG(WARNING) << "The cluster is still in process of scaling, please retry " << name_ << " later.";
|
||||
*reason = ps::kClusterSafeMode;
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
} // namespace server
|
||||
} // namespace fl
|
||||
} // namespace mindspore
|
||||
|
|
|
|||
|
|
@ -43,6 +43,9 @@ class Round {
|
|||
// Reinitialize count service and round kernel of this round after scaling operations are done.
|
||||
bool ReInitForScaling(uint32_t server_num);
|
||||
|
||||
// After hyper-parameters are updated, some rounds and kernels should be reinitialized.
|
||||
bool ReInitForUpdatingHyperParams(size_t updated_threshold_count, size_t updated_time_window);
|
||||
|
||||
// Bind a round kernel to this Round. This method should be called after Initialize.
|
||||
void BindRoundKernel(const std::shared_ptr<kernel::RoundKernel> &kernel);
|
||||
|
||||
|
|
@ -63,6 +66,9 @@ class Round {
|
|||
void OnFirstCountEvent(const std::shared_ptr<ps::core::MessageHandler> &message);
|
||||
void OnLastCountEvent(const std::shared_ptr<ps::core::MessageHandler> &message);
|
||||
|
||||
// Judge whether the training service is available.
|
||||
bool IsServerAvailable(std::string *reason);
|
||||
|
||||
std::string name_;
|
||||
|
||||
// Whether this round needs to use timer. Most rounds in federated learning with mobile devices scenario need to set
|
||||
|
|
|
|||
|
|
@ -80,6 +80,7 @@ void Server::Run() {
|
|||
MS_LOG(INFO) << "Parameters for secure aggregation have been initiated.";
|
||||
}
|
||||
RegisterRoundKernel();
|
||||
InitMetrics();
|
||||
MS_LOG(INFO) << "Server started successfully.";
|
||||
safemode_ = false;
|
||||
lock.unlock();
|
||||
|
|
@ -306,6 +307,8 @@ void Server::RegisterCommCallbacks() {
|
|||
|
||||
// Set exception event callbacks for server.
|
||||
RegisterExceptionEventCallback(tcp_comm);
|
||||
// Set message callbacks for server.
|
||||
RegisterMessageCallback(tcp_comm);
|
||||
|
||||
if (!server_node_->InitFollowerScaler()) {
|
||||
MS_LOG(EXCEPTION) << "Initializing follower elastic scaler failed.";
|
||||
|
|
@ -354,6 +357,19 @@ void Server::RegisterExceptionEventCallback(const std::shared_ptr<ps::core::TcpC
|
|||
});
|
||||
}
|
||||
|
||||
void Server::RegisterMessageCallback(const std::shared_ptr<ps::core::TcpCommunicator> &communicator) {
|
||||
MS_EXCEPTION_IF_NULL(communicator);
|
||||
// Register handler for restful requests receviced by scheduler.
|
||||
communicator->RegisterMsgCallBack("enableFLS",
|
||||
std::bind(&Server::HandleEnableServerRequest, this, std::placeholders::_1));
|
||||
communicator->RegisterMsgCallBack("disableFLS",
|
||||
std::bind(&Server::HandleDisableServerRequest, this, std::placeholders::_1));
|
||||
communicator->RegisterMsgCallBack("newInstance",
|
||||
std::bind(&Server::HandleNewInstanceRequest, this, std::placeholders::_1));
|
||||
communicator->RegisterMsgCallBack("queryInstance",
|
||||
std::bind(&Server::HandleQueryInstanceRequest, this, std::placeholders::_1));
|
||||
}
|
||||
|
||||
void Server::InitExecutor() {
|
||||
MS_EXCEPTION_IF_NULL(func_graph_);
|
||||
if (executor_threshold_ == 0) {
|
||||
|
|
@ -392,6 +408,19 @@ void Server::RegisterRoundKernel() {
|
|||
return;
|
||||
}
|
||||
|
||||
void Server::InitMetrics() {
|
||||
if (server_node_->rank_id() == kLeaderServerRank) {
|
||||
MS_EXCEPTION_IF_NULL(iteration_);
|
||||
std::shared_ptr<IterationMetrics> iteration_metrics =
|
||||
std::make_shared<IterationMetrics>(ps::PSContext::instance()->config_file_path());
|
||||
if (!iteration_metrics->Initialize()) {
|
||||
MS_LOG(WARNING) << "Initializing metrics failed.";
|
||||
return;
|
||||
}
|
||||
iteration_->set_metrics(iteration_metrics);
|
||||
}
|
||||
}
|
||||
|
||||
void Server::StartCommunicator() {
|
||||
if (communicators_with_worker_.empty()) {
|
||||
MS_LOG(EXCEPTION) << "Communicators for communication with worker is empty.";
|
||||
|
|
@ -489,6 +518,92 @@ void Server::ProcessAfterScalingIn() {
|
|||
std::this_thread::sleep_for(std::chrono::milliseconds(kServerSleepTimeForNetworking));
|
||||
safemode_ = false;
|
||||
}
|
||||
|
||||
void Server::HandleEnableServerRequest(const std::shared_ptr<ps::core::MessageHandler> &message) {
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(message);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(iteration_);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(communicator_with_server_);
|
||||
auto tcp_comm = std::dynamic_pointer_cast<ps::core::TcpCommunicator>(communicator_with_server_);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(tcp_comm);
|
||||
|
||||
std::string result_message = "";
|
||||
bool result = iteration_->EnableServerInstance(&result_message);
|
||||
nlohmann::json response;
|
||||
response["result"] = result;
|
||||
response["message"] = result_message;
|
||||
if (!tcp_comm->SendResponse(response.dump().c_str(), response.dump().size(), message)) {
|
||||
MS_LOG(ERROR) << "Sending response failed.";
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
void Server::HandleDisableServerRequest(const std::shared_ptr<ps::core::MessageHandler> &message) {
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(message);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(iteration_);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(communicator_with_server_);
|
||||
auto tcp_comm = std::dynamic_pointer_cast<ps::core::TcpCommunicator>(communicator_with_server_);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(tcp_comm);
|
||||
|
||||
std::string result_message = "";
|
||||
bool result = iteration_->DisableServerInstance(&result_message);
|
||||
nlohmann::json response;
|
||||
response["result"] = result;
|
||||
response["message"] = result_message;
|
||||
if (!tcp_comm->SendResponse(response.dump().c_str(), response.dump().size(), message)) {
|
||||
MS_LOG(ERROR) << "Sending response failed.";
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
void Server::HandleNewInstanceRequest(const std::shared_ptr<ps::core::MessageHandler> &message) {
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(message);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(iteration_);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(communicator_with_server_);
|
||||
auto tcp_comm = std::dynamic_pointer_cast<ps::core::TcpCommunicator>(communicator_with_server_);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(tcp_comm);
|
||||
|
||||
std::string hyper_params_str(static_cast<const char *>(message->data()), message->len());
|
||||
nlohmann::json new_instance_json;
|
||||
nlohmann::json response;
|
||||
try {
|
||||
new_instance_json = nlohmann::json::parse(hyper_params_str);
|
||||
} catch (const std::exception &e) {
|
||||
response["result"] = false;
|
||||
response["message"] = "The hyper-parameter data is not in json format.";
|
||||
if (!tcp_comm->SendResponse(response.dump().c_str(), response.dump().size(), message)) {
|
||||
MS_LOG(ERROR) << "Sending response failed.";
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
std::string result_message = "";
|
||||
bool result = iteration_->NewInstance(new_instance_json, &result_message);
|
||||
response["result"] = result;
|
||||
response["message"] = result_message;
|
||||
if (!tcp_comm->SendResponse(response.dump().c_str(), response.dump().size(), message)) {
|
||||
MS_LOG(ERROR) << "Sending response failed.";
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
void Server::HandleQueryInstanceRequest(const std::shared_ptr<ps::core::MessageHandler> &message) {
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(message);
|
||||
nlohmann::json response;
|
||||
response["start_fl_job_threshold"] = ps::PSContext::instance()->start_fl_job_threshold();
|
||||
response["start_fl_job_time_window"] = ps::PSContext::instance()->start_fl_job_time_window();
|
||||
response["update_model_ratio"] = ps::PSContext::instance()->update_model_ratio();
|
||||
response["update_model_time_window"] = ps::PSContext::instance()->update_model_time_window();
|
||||
response["fl_iteration_num"] = ps::PSContext::instance()->fl_iteration_num();
|
||||
response["client_epoch_num"] = ps::PSContext::instance()->client_epoch_num();
|
||||
response["client_batch_size"] = ps::PSContext::instance()->client_batch_size();
|
||||
response["client_learning_rate"] = ps::PSContext::instance()->client_learning_rate();
|
||||
auto tcp_comm = std::dynamic_pointer_cast<ps::core::TcpCommunicator>(communicator_with_server_);
|
||||
MS_ERROR_IF_NULL_WO_RET_VAL(tcp_comm);
|
||||
if (!tcp_comm->SendResponse(response.dump().c_str(), response.dump().size(), message)) {
|
||||
MS_LOG(ERROR) << "Sending response failed.";
|
||||
return;
|
||||
}
|
||||
}
|
||||
} // namespace server
|
||||
} // namespace fl
|
||||
} // namespace mindspore
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@
|
|||
#include "ps/core/communicator/communicator_base.h"
|
||||
#include "ps/core/communicator/tcp_communicator.h"
|
||||
#include "ps/core/communicator/task_executor.h"
|
||||
#include "ps/core/file_configuration.h"
|
||||
#include "fl/server/common.h"
|
||||
#include "fl/server/executor.h"
|
||||
#include "fl/server/iteration.h"
|
||||
|
|
@ -56,6 +57,9 @@ class Server {
|
|||
void CancelSafeMode();
|
||||
bool IsSafeMode() const;
|
||||
|
||||
// Whether the training job of the server is enabled.
|
||||
InstanceState instance_state() const;
|
||||
|
||||
private:
|
||||
Server()
|
||||
: server_node_(nullptr),
|
||||
|
|
@ -88,6 +92,9 @@ class Server {
|
|||
// Load variables which is set by ps_context.
|
||||
void InitServerContext();
|
||||
|
||||
// Try to recover server config from persistent storage.
|
||||
void Recovery();
|
||||
|
||||
// Initialize the server cluster, server node and communicators.
|
||||
void InitCluster();
|
||||
bool InitCommunicatorWithServer();
|
||||
|
|
@ -103,6 +110,9 @@ class Server {
|
|||
// Register cluster exception callbacks. This method is called in RegisterCommCallbacks.
|
||||
void RegisterExceptionEventCallback(const std::shared_ptr<ps::core::TcpCommunicator> &communicator);
|
||||
|
||||
// Register message callbacks. These messages are mainly from scheduler.
|
||||
void RegisterMessageCallback(const std::shared_ptr<ps::core::TcpCommunicator> &communicator);
|
||||
|
||||
// Initialize executor according to the server mode.
|
||||
void InitExecutor();
|
||||
|
||||
|
|
@ -112,6 +122,8 @@ class Server {
|
|||
// Create round kernels and bind these kernels with corresponding Round.
|
||||
void RegisterRoundKernel();
|
||||
|
||||
void InitMetrics();
|
||||
|
||||
// The communicators should be started after all initializations are completed.
|
||||
void StartCommunicator();
|
||||
|
||||
|
|
@ -123,6 +135,16 @@ class Server {
|
|||
void ProcessAfterScalingOut();
|
||||
void ProcessAfterScalingIn();
|
||||
|
||||
// Handlers for enableFLS/disableFLS requests from the scheduler.
|
||||
void HandleEnableServerRequest(const std::shared_ptr<ps::core::MessageHandler> &message);
|
||||
void HandleDisableServerRequest(const std::shared_ptr<ps::core::MessageHandler> &message);
|
||||
|
||||
// Finish current instance and start a new one. FLPlan could be changed in this method.
|
||||
void HandleNewInstanceRequest(const std::shared_ptr<ps::core::MessageHandler> &message);
|
||||
|
||||
// Query current instance information.
|
||||
void HandleQueryInstanceRequest(const std::shared_ptr<ps::core::MessageHandler> &message);
|
||||
|
||||
// The server node is initialized in Server.
|
||||
std::shared_ptr<ps::core::ServerNode> server_node_;
|
||||
|
||||
|
|
|
|||
|
|
@ -123,8 +123,9 @@ bool FLWorker::SendToServer(uint32_t server_rank, const void *data, size_t size,
|
|||
return false;
|
||||
}
|
||||
|
||||
if (std::string(reinterpret_cast<char *>((*output)->data()), (*output)->size()) == ps::kClusterSafeMode) {
|
||||
MS_LOG(INFO) << "The server " << server_rank << " is in safemode.";
|
||||
std::string response_str = std::string(reinterpret_cast<char *>((*output)->data()), (*output)->size());
|
||||
if (response_str == ps::kClusterSafeMode || response_str == ps::kJobNotAvailable) {
|
||||
MS_LOG(INFO) << "The server " << server_rank << " is in safemode or finished.";
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kWorkerRetryDurationForSafeMode));
|
||||
} else {
|
||||
break;
|
||||
|
|
|
|||
|
|
@ -35,6 +35,7 @@ using FBBuilder = flatbuffers::FlatBufferBuilder;
|
|||
// The step number for worker to judge whether to communicate with server.
|
||||
constexpr uint32_t kTrainBeginStepNum = 1;
|
||||
constexpr uint32_t kTrainEndStepNum = 0;
|
||||
constexpr uint32_t kOneStepPerIteration = 1;
|
||||
|
||||
// The sleeping time of the worker thread before the networking is completed.
|
||||
constexpr uint32_t kWorkerSleepTimeForNetworking = 1000;
|
||||
|
|
@ -42,7 +43,7 @@ constexpr uint32_t kWorkerSleepTimeForNetworking = 1000;
|
|||
// The time duration between retrying when server is in safemode.
|
||||
constexpr uint32_t kWorkerRetryDurationForSafeMode = 500;
|
||||
|
||||
// The leader server rank.
|
||||
// The rank of the leader server.
|
||||
constexpr uint32_t kLeaderServerRank = 0;
|
||||
|
||||
enum class IterationState {
|
||||
|
|
|
|||
|
|
@ -747,7 +747,8 @@ bool StartServerAction(const ResourcePtr &res) {
|
|||
{"updateModel", true, update_model_time_window, true, update_model_threshold},
|
||||
{"getModel"},
|
||||
{"pullWeight"},
|
||||
{"pushWeight", false, 3000, true, server_num, true}};
|
||||
{"pushWeight", false, 3000, true, server_num, true},
|
||||
{"pushMetrics", false, 3000, true, 1}};
|
||||
|
||||
float share_secrets_ratio = ps::PSContext::instance()->share_secrets_ratio();
|
||||
uint64_t cipher_time_window = ps::PSContext::instance()->cipher_time_window();
|
||||
|
|
|
|||
|
|
@ -250,6 +250,7 @@ using HandlerAfterScaleOut = std::function<void(void)>;
|
|||
using HandlerAfterScaleIn = std::function<void(void)>;
|
||||
|
||||
constexpr char kClusterSafeMode[] = "The cluster is in safemode.";
|
||||
constexpr char kJobNotAvailable[] = "The server's training job is disabled or finished.";
|
||||
|
||||
enum class CustomEvent { kIterationRunning = 0, kIterationCompleted };
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue