From b73b7344cc3405047a8b0b337b8ec1ae5b9296be Mon Sep 17 00:00:00 2001 From: ZPaC Date: Fri, 13 Aug 2021 14:58:13 +0800 Subject: [PATCH] Docking with cloud platform --- .../cpu/fl/fused_pull_weight_kernel.h | 12 +- .../cpu/fl/fused_push_weight_kernel.h | 16 +- mindspore/ccsrc/fl/server/common.h | 1 + mindspore/ccsrc/fl/server/executor.cc | 12 + mindspore/ccsrc/fl/server/executor.h | 5 +- mindspore/ccsrc/fl/server/iteration.cc | 290 +++++++++++++++++- mindspore/ccsrc/fl/server/iteration.h | 95 +++++- .../ccsrc/fl/server/iteration_metrics.cc | 2 +- .../fl/server/kernel/aggregation_kernel.h | 2 + .../ccsrc/fl/server/kernel/fed_avg_kernel.h | 6 + mindspore/ccsrc/fl/server/model_store.cc | 1 - .../ccsrc/fl/server/parameter_aggregator.cc | 15 + .../ccsrc/fl/server/parameter_aggregator.h | 3 + mindspore/ccsrc/fl/server/round.cc | 54 +++- mindspore/ccsrc/fl/server/round.h | 6 + mindspore/ccsrc/fl/server/server.cc | 115 +++++++ mindspore/ccsrc/fl/server/server.h | 22 ++ mindspore/ccsrc/fl/worker/fl_worker.cc | 5 +- mindspore/ccsrc/fl/worker/fl_worker.h | 3 +- mindspore/ccsrc/pipeline/jit/action.cc | 3 +- mindspore/ccsrc/ps/constants.h | 1 + 21 files changed, 631 insertions(+), 38 deletions(-) diff --git a/mindspore/ccsrc/backend/kernel_compiler/cpu/fl/fused_pull_weight_kernel.h b/mindspore/ccsrc/backend/kernel_compiler/cpu/fl/fused_pull_weight_kernel.h index 5c9483d41a4..e295aa76a71 100644 --- a/mindspore/ccsrc/backend/kernel_compiler/cpu/fl/fused_pull_weight_kernel.h +++ b/mindspore/ccsrc/backend/kernel_compiler/cpu/fl/fused_pull_weight_kernel.h @@ -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 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; } diff --git a/mindspore/ccsrc/backend/kernel_compiler/cpu/fl/fused_push_weight_kernel.h b/mindspore/ccsrc/backend/kernel_compiler/cpu/fl/fused_push_weight_kernel.h index 56d09bf956c..fcdc00a2f41 100644 --- a/mindspore/ccsrc/backend/kernel_compiler/cpu/fl/fused_push_weight_kernel.h +++ b/mindspore/ccsrc/backend/kernel_compiler/cpu/fl/fused_push_weight_kernel.h @@ -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 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; } diff --git a/mindspore/ccsrc/fl/server/common.h b/mindspore/ccsrc/fl/server/common.h index a4cb3db47e9..a35ecb00244 100644 --- a/mindspore/ccsrc/fl/server/common.h +++ b/mindspore/ccsrc/fl/server/common.h @@ -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"; diff --git a/mindspore/ccsrc/fl/server/executor.cc b/mindspore/ccsrc/fl/server/executor.cc index cf87a3513eb..ddaf30ca7d3 100644 --- a/mindspore/ccsrc/fl/server/executor.cc +++ b/mindspore/ccsrc/fl/server/executor.cc @@ -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) { diff --git a/mindspore/ccsrc/fl/server/executor.h b/mindspore/ccsrc/fl/server/executor.h index bc0963cb519..3bc90288d5f 100644 --- a/mindspore/ccsrc/fl/server/executor.h +++ b/mindspore/ccsrc/fl/server/executor.h @@ -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. diff --git a/mindspore/ccsrc/fl/server/iteration.cc b/mindspore/ccsrc/fl/server/iteration.cc index 5200f440fa5..1a6d654c5ca 100644 --- a/mindspore/ccsrc/fl/server/iteration.cc +++ b/mindspore/ccsrc/fl/server/iteration.cc @@ -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 &communicator) { MS_EXCEPTION_IF_NULL(communicator); communicator_ = communicator; @@ -79,9 +87,27 @@ void Iteration::InitRounds(const std::vector 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 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(ps::CustomEvent::kIterationRunning)); } + + std::unique_lock 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(ps::CustomEvent::kIterationCompleted)); } + + std::unique_lock 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 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 &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> &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 &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 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 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 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::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 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( + 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()); + continue; + } + if (key == "start_fl_job_time_window") { + ps::PSContext::instance()->set_start_fl_job_time_window(item.value().get()); + continue; + } + if (key == "update_model_ratio") { + ps::PSContext::instance()->set_update_model_ratio(item.value().get()); + continue; + } + if (key == "update_model_time_window") { + ps::PSContext::instance()->set_update_model_time_window(item.value().get()); + continue; + } + if (key == "fl_iteration_num") { + ps::PSContext::instance()->set_fl_iteration_num(item.value().get()); + continue; + } + if (key == "client_epoch_num") { + ps::PSContext::instance()->set_client_epoch_num(item.value().get()); + continue; + } + if (key == "client_batch_size") { + ps::PSContext::instance()->set_client_batch_size(item.value().get()); + continue; + } + if (key == "client_learning_rate") { + ps::PSContext::instance()->set_client_learning_rate(item.value().get()); + 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(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 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 diff --git a/mindspore/ccsrc/fl/server/iteration.h b/mindspore/ccsrc/fl/server/iteration.h index d89689ce4b5..7caef124241 100644 --- a/mindspore/ccsrc/fl/server/iteration.h +++ b/mindspore/ccsrc/fl/server/iteration.h @@ -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> &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 &updated_rounds_config); + const std::vector> &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 &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 server_node_; std::shared_ptr communicator_; @@ -145,7 +202,11 @@ class Iteration { std::vector> rounds_; // The iteration is either running or completed at any time. + std::mutex iteration_state_mtx_; + std::condition_variable iteration_state_cv_; std::atomic 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 metrics_; + + // The state for current instance. + std::atomic 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 diff --git a/mindspore/ccsrc/fl/server/iteration_metrics.cc b/mindspore/ccsrc/fl/server/iteration_metrics.cc index 504859ac1ee..9bc5f693f69 100644 --- a/mindspore/ccsrc/fl/server/iteration_metrics.cc +++ b/mindspore/ccsrc/fl/server/iteration_metrics.cc @@ -14,11 +14,11 @@ * limitations under the License. */ +#include "fl/server/iteration_metrics.h" #include #include #include "debug/common.h" #include "ps/constants.h" -#include "fl/server/iteration_metrics.h" namespace mindspore { namespace fl { diff --git a/mindspore/ccsrc/fl/server/kernel/aggregation_kernel.h b/mindspore/ccsrc/fl/server/kernel/aggregation_kernel.h index a0c41771163..aae59210a1c 100644 --- a/mindspore/ccsrc/fl/server/kernel/aggregation_kernel.h +++ b/mindspore/ccsrc/fl/server/kernel/aggregation_kernel.h @@ -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 &input_names() { return params_info_.inputs_names(); } diff --git a/mindspore/ccsrc/fl/server/kernel/fed_avg_kernel.h b/mindspore/ccsrc/fl/server/kernel/fed_avg_kernel.h index b201fa83d92..8e05ea540b8 100644 --- a/mindspore/ccsrc/fl/server/kernel/fed_avg_kernel.h +++ b/mindspore/ccsrc/fl/server/kernel/fed_avg_kernel.h @@ -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."; diff --git a/mindspore/ccsrc/fl/server/model_store.cc b/mindspore/ccsrc/fl/server/model_store.cc index 8444798a614..4d2f66c1d40 100644 --- a/mindspore/ccsrc/fl/server/model_store.cc +++ b/mindspore/ccsrc/fl/server/model_store.cc @@ -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> &ModelStore::iteration_to_model() { diff --git a/mindspore/ccsrc/fl/server/parameter_aggregator.cc b/mindspore/ccsrc/fl/server/parameter_aggregator.cc index 9a5cf531821..0ef6f5569ad 100644 --- a/mindspore/ccsrc/fl/server/parameter_aggregator.cc +++ b/mindspore/ccsrc/fl/server/parameter_aggregator.cc @@ -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 &new_data) { std::map &name_to_addr = memory_register_->addresses(); for (const auto &data : new_data) { diff --git a/mindspore/ccsrc/fl/server/parameter_aggregator.h b/mindspore/ccsrc/fl/server/parameter_aggregator.h index 4fc3fe60f0c..8bf68143b6d 100644 --- a/mindspore/ccsrc/fl/server/parameter_aggregator.h +++ b/mindspore/ccsrc/fl/server/parameter_aggregator.h @@ -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 &new_data); diff --git a/mindspore/ccsrc/fl/server/round.cc b/mindspore/ccsrc/fl/server/round.cc index 0b578814b29..a88e81563f7 100644 --- a/mindspore/ccsrc/fl/server/round.cc +++ b/mindspore/ccsrc/fl/server/round.cc @@ -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) { MS_EXCEPTION_IF_NULL(kernel); kernel_ = kernel; @@ -114,10 +129,9 @@ void Round::LaunchRoundKernel(const std::shared_ptr &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 &m return; } + Iteration::GetInstance().running_round_num_++; AddressPtr input = std::make_shared
(); AddressPtr output = std::make_shared
(); MS_ERROR_IF_NULL_WO_RET_VAL(input); @@ -133,7 +148,7 @@ void Round::LaunchRoundKernel(const std::shared_ptr &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 &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 &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 diff --git a/mindspore/ccsrc/fl/server/round.h b/mindspore/ccsrc/fl/server/round.h index 1aae7b560d7..cbd868b1f43 100644 --- a/mindspore/ccsrc/fl/server/round.h +++ b/mindspore/ccsrc/fl/server/round.h @@ -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); @@ -63,6 +66,9 @@ class Round { void OnFirstCountEvent(const std::shared_ptr &message); void OnLastCountEvent(const std::shared_ptr &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 diff --git a/mindspore/ccsrc/fl/server/server.cc b/mindspore/ccsrc/fl/server/server.cc index 69ad3fe52f1..8db8a01713e 100644 --- a/mindspore/ccsrc/fl/server/server.cc +++ b/mindspore/ccsrc/fl/server/server.cc @@ -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 &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 iteration_metrics = + std::make_shared(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 &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(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 &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(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 &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(communicator_with_server_); + MS_ERROR_IF_NULL_WO_RET_VAL(tcp_comm); + + std::string hyper_params_str(static_cast(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 &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(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 diff --git a/mindspore/ccsrc/fl/server/server.h b/mindspore/ccsrc/fl/server/server.h index bd0a3c6aa68..4e175fc4df1 100644 --- a/mindspore/ccsrc/fl/server/server.h +++ b/mindspore/ccsrc/fl/server/server.h @@ -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 &communicator); + // Register message callbacks. These messages are mainly from scheduler. + void RegisterMessageCallback(const std::shared_ptr &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 &message); + void HandleDisableServerRequest(const std::shared_ptr &message); + + // Finish current instance and start a new one. FLPlan could be changed in this method. + void HandleNewInstanceRequest(const std::shared_ptr &message); + + // Query current instance information. + void HandleQueryInstanceRequest(const std::shared_ptr &message); + // The server node is initialized in Server. std::shared_ptr server_node_; diff --git a/mindspore/ccsrc/fl/worker/fl_worker.cc b/mindspore/ccsrc/fl/worker/fl_worker.cc index a004ba74042..8406d64bbad 100644 --- a/mindspore/ccsrc/fl/worker/fl_worker.cc +++ b/mindspore/ccsrc/fl/worker/fl_worker.cc @@ -123,8 +123,9 @@ bool FLWorker::SendToServer(uint32_t server_rank, const void *data, size_t size, return false; } - if (std::string(reinterpret_cast((*output)->data()), (*output)->size()) == ps::kClusterSafeMode) { - MS_LOG(INFO) << "The server " << server_rank << " is in safemode."; + std::string response_str = std::string(reinterpret_cast((*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; diff --git a/mindspore/ccsrc/fl/worker/fl_worker.h b/mindspore/ccsrc/fl/worker/fl_worker.h index 3950c2a96b9..0ca6f2f60fa 100644 --- a/mindspore/ccsrc/fl/worker/fl_worker.h +++ b/mindspore/ccsrc/fl/worker/fl_worker.h @@ -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 { diff --git a/mindspore/ccsrc/pipeline/jit/action.cc b/mindspore/ccsrc/pipeline/jit/action.cc index 1e28d5f4160..245a7eabea3 100644 --- a/mindspore/ccsrc/pipeline/jit/action.cc +++ b/mindspore/ccsrc/pipeline/jit/action.cc @@ -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(); diff --git a/mindspore/ccsrc/ps/constants.h b/mindspore/ccsrc/ps/constants.h index d153d536265..9b0a5e3e4f6 100644 --- a/mindspore/ccsrc/ps/constants.h +++ b/mindspore/ccsrc/ps/constants.h @@ -250,6 +250,7 @@ using HandlerAfterScaleOut = std::function; using HandlerAfterScaleIn = std::function; 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 };