From a48722c07815da7f6f4bac62d870147a47704840 Mon Sep 17 00:00:00 2001 From: twc Date: Mon, 10 Jan 2022 18:52:30 +0800 Subject: [PATCH] fix pclint --- mindspore/ccsrc/ps/core/abstract_node.cc | 13 +- mindspore/ccsrc/ps/core/abstract_node.h | 16 +- mindspore/ccsrc/ps/core/cluster_config.h | 2 +- mindspore/ccsrc/ps/core/comm_util.cc | 7 +- .../ps/core/communicator/communicator_base.cc | 2 +- .../ps/core/communicator/communicator_base.h | 2 +- .../ccsrc/ps/core/communicator/http_server.cc | 2 +- .../ccsrc/ps/core/communicator/tcp_client.cc | 8 +- .../ps/core/communicator/tcp_msg_handler.cc | 2 +- .../ps/core/communicator/tcp_msg_handler.h | 2 +- mindspore/ccsrc/ps/core/node_manager.cc | 68 +++--- mindspore/ccsrc/ps/core/node_manager.h | 38 ++-- mindspore/ccsrc/ps/core/scheduler_node.cc | 38 ++-- mindspore/ccsrc/ps/core/scheduler_recovery.cc | 27 ++- scripts/fl_restful_tool.py | 194 ++++++++++-------- 15 files changed, 221 insertions(+), 200 deletions(-) diff --git a/mindspore/ccsrc/ps/core/abstract_node.cc b/mindspore/ccsrc/ps/core/abstract_node.cc index 64bdae98f57..df05d16c417 100644 --- a/mindspore/ccsrc/ps/core/abstract_node.cc +++ b/mindspore/ccsrc/ps/core/abstract_node.cc @@ -439,13 +439,13 @@ void AbstractNode::RegisterFollowerScalerHandlerAfterScaleIn(const std::string & PersistentState AbstractNode::persistent_state() const { return persistent_state_; } void AbstractNode::set_persistent_state(PersistentState persistent_state) { persistent_state_ = persistent_state; } -int32_t AbstractNode::worker_num() const { return worker_num_; } +uint32_t AbstractNode::worker_num() const { return worker_num_; } -int32_t AbstractNode::server_num() const { return server_num_; } +uint32_t AbstractNode::server_num() const { return server_num_; } -void AbstractNode::set_worker_num(const int32_t &worker_num) { worker_num_ = worker_num; } +void AbstractNode::set_worker_num(const uint32_t &worker_num) { worker_num_ = worker_num; } -void AbstractNode::set_server_num(const int32_t &server_num) { server_num_ = server_num; } +void AbstractNode::set_server_num(const uint32_t &server_num) { server_num_ = server_num; } std::string AbstractNode::scheduler_ip() const { return scheduler_ip_; } @@ -1063,7 +1063,8 @@ void AbstractNode::ProcessSendData(const std::shared_ptr &conn, c std::shared_ptr res(new unsigned char[size], std::default_delete()); #else if (size < 0) { - MS_LOG(EXCEPTION) << "size < 0"; + MS_LOG(ERROR) << "size < 0"; + return; } std::shared_ptr res(new unsigned char[size]); #endif @@ -1293,7 +1294,7 @@ void AbstractNode::CreateTcpServer() { } void AbstractNode::UpdateClusterState(const ClusterState &state) { - std::lock_guard lk(cluster_state_mutex_); + std::lock_guard lock(cluster_state_mutex_); MS_LOG(INFO) << "[state]: Cluster state change from:" << CommUtil::ClusterStateToString(current_cluster_state_) << " to " << CommUtil::ClusterStateToString(state); current_cluster_state_ = state; diff --git a/mindspore/ccsrc/ps/core/abstract_node.h b/mindspore/ccsrc/ps/core/abstract_node.h index d8821eaa4da..49e1adcedb3 100644 --- a/mindspore/ccsrc/ps/core/abstract_node.h +++ b/mindspore/ccsrc/ps/core/abstract_node.h @@ -47,8 +47,8 @@ class AbstractNode : public Node { client_to_server_(nullptr), server_(nullptr), server_thread_(nullptr), - worker_num_(-1), - server_num_(-1), + worker_num_(0), + server_num_(0), is_connected_to_scheduler_(false), is_current_node_scale_in_(false), follower_scaler_(nullptr), @@ -125,11 +125,11 @@ class AbstractNode : public Node { PersistentState persistent_state() const; void set_persistent_state(PersistentState persistent_state); - int32_t worker_num() const; - int32_t server_num() const; + uint32_t worker_num() const; + uint32_t server_num() const; - void set_worker_num(const int32_t &worker_num); - void set_server_num(const int32_t &server_num); + void set_worker_num(const uint32_t &worker_num); + void set_server_num(const uint32_t &server_num); std::string scheduler_ip() const; void set_scheduler_ip(const std::string &scheduler_ip); @@ -264,8 +264,8 @@ class AbstractNode : public Node { std::shared_ptr server_; std::unique_ptr server_thread_; - int32_t worker_num_; - int32_t server_num_; + uint32_t worker_num_; + uint32_t server_num_; std::atomic is_connected_to_scheduler_; // Identify whether the current node is a scale in node. std::atomic is_current_node_scale_in_; diff --git a/mindspore/ccsrc/ps/core/cluster_config.h b/mindspore/ccsrc/ps/core/cluster_config.h index 63eb0ebf6a3..7e325c994ef 100644 --- a/mindspore/ccsrc/ps/core/cluster_config.h +++ b/mindspore/ccsrc/ps/core/cluster_config.h @@ -61,7 +61,7 @@ struct ClusterConfig { uint16_t scheduler_port; // The timeout for worker node and server node sending heartbeat packets to scheduler node is 30 seconds. uint32_t heartbeat_timeout; - // Timeout period for cluster preparation is 300 seconds. + // Timeout period for cluster preparation is 900 seconds. uint32_t cluster_available_timeout; // The timeout period for the client to connect to the server is 3000ms. uint32_t connect_interval; diff --git a/mindspore/ccsrc/ps/core/comm_util.cc b/mindspore/ccsrc/ps/core/comm_util.cc index 0513927d1bd..424d4f4c1bf 100644 --- a/mindspore/ccsrc/ps/core/comm_util.cc +++ b/mindspore/ccsrc/ps/core/comm_util.cc @@ -253,7 +253,6 @@ bool CommUtil::CreateDirectory(const std::string &directoryPath) { } std::string CommUtil::ClusterStateToString(const ClusterState &state) { - MS_LOG(DEBUG) << "The cluster state:" << state; if (state < SizeToInt(kClusterState.size())) { return kClusterState.at(state); } else { @@ -430,11 +429,11 @@ bool CommUtil::verifyCertKeyID(const X509 *caCert, const X509 *subCert) { ASN1_OCTET_STRING *skid = reinterpret_cast(X509_get_ext_d2i(caCert, NID_subject_key_identifier, &crit, NULL)); MS_EXCEPTION_IF_NULL(skid); - const int keyidLen = 512; + const size_t keyidLen = 512; char subject_keyid[keyidLen] = {0}; for (int i = 0; i < skid->length; i++) { char keyid[8] = {0}; - int base = keyidLen; + size_t base = keyidLen; (void)sprintf_s(keyid, sizeof(keyid), "%x ", (uint32_t)skid->data[i]); int ret = strcat_s(subject_keyid, base, keyid); if (ret == -1) { @@ -449,7 +448,7 @@ bool CommUtil::verifyCertKeyID(const X509 *caCert, const X509 *subCert) { char issuer_keyid[keyidLen] = {0}; for (int i = 0; i < akeyid->keyid->length; i++) { char keyid[8] = {0}; - int base = keyidLen; + size_t base = keyidLen; (void)sprintf_s(keyid, sizeof(keyid), "%x ", (uint32_t)(akeyid->keyid->data[i])); int ret = strcat_s(issuer_keyid, base, keyid); if (ret == -1) { diff --git a/mindspore/ccsrc/ps/core/communicator/communicator_base.cc b/mindspore/ccsrc/ps/core/communicator/communicator_base.cc index 5b0eedd8972..f8d0feccf0b 100644 --- a/mindspore/ccsrc/ps/core/communicator/communicator_base.cc +++ b/mindspore/ccsrc/ps/core/communicator/communicator_base.cc @@ -43,7 +43,7 @@ void CommunicatorBase::Join() { return; } -bool CommunicatorBase::running() { return running_; } +bool CommunicatorBase::running() const { return running_; } } // namespace core } // namespace ps } // namespace mindspore diff --git a/mindspore/ccsrc/ps/core/communicator/communicator_base.h b/mindspore/ccsrc/ps/core/communicator/communicator_base.h index a18b92251e9..41785ac2f74 100644 --- a/mindspore/ccsrc/ps/core/communicator/communicator_base.h +++ b/mindspore/ccsrc/ps/core/communicator/communicator_base.h @@ -85,7 +85,7 @@ class CommunicatorBase { bool SendResponse(const void *rsp_data, size_t rsp_len, const std::shared_ptr &msg_handler); - bool running(); + bool running() const; protected: std::unordered_map msg_callbacks_; diff --git a/mindspore/ccsrc/ps/core/communicator/http_server.cc b/mindspore/ccsrc/ps/core/communicator/http_server.cc index 1d598b5c011..13c1c4e0a4b 100644 --- a/mindspore/ccsrc/ps/core/communicator/http_server.cc +++ b/mindspore/ccsrc/ps/core/communicator/http_server.cc @@ -68,7 +68,7 @@ bool HttpServer::InitServer() { return false; } - fd_ = ::socket(AF_INET, SOCK_STREAM, 0); + fd_ = ::socket(static_cast(AF_INET), static_cast(SOCK_STREAM), 0); if (fd_ < 0) { MS_LOG(ERROR) << "Socker error!"; return false; diff --git a/mindspore/ccsrc/ps/core/communicator/tcp_client.cc b/mindspore/ccsrc/ps/core/communicator/tcp_client.cc index 78eb9c01358..1736ac66686 100644 --- a/mindspore/ccsrc/ps/core/communicator/tcp_client.cc +++ b/mindspore/ccsrc/ps/core/communicator/tcp_client.cc @@ -159,7 +159,7 @@ void TcpClient::Stop() { void TcpClient::SetTcpNoDelay(const evutil_socket_t &fd) { const int one = 1; - int ret = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(int)); + int ret = setsockopt(fd, static_cast(IPPROTO_TCP), static_cast(TCP_NODELAY), &one, sizeof(int)); if (ret < 0) { MS_LOG(EXCEPTION) << "Set socket no delay failed!"; } @@ -193,10 +193,10 @@ void TcpClient::ReadCallbackInner(struct bufferevent *bev, void *const ctx) { auto tcp_client = reinterpret_cast(ctx); char read_buffer[kMessageChunkLength]; - int read = 0; + size_t read = 0; - while ((read = bufferevent_read(bev, &read_buffer, SizeToInt(sizeof(read_buffer)))) > 0) { - tcp_client->OnReadHandler(read_buffer, IntToSize(read)); + while ((read = bufferevent_read(bev, &read_buffer, sizeof(read_buffer))) > 0) { + tcp_client->OnReadHandler(read_buffer, read); } } diff --git a/mindspore/ccsrc/ps/core/communicator/tcp_msg_handler.cc b/mindspore/ccsrc/ps/core/communicator/tcp_msg_handler.cc index 91320bb1568..1134e417ea2 100644 --- a/mindspore/ccsrc/ps/core/communicator/tcp_msg_handler.cc +++ b/mindspore/ccsrc/ps/core/communicator/tcp_msg_handler.cc @@ -21,7 +21,7 @@ namespace mindspore { namespace ps { namespace core { TcpMsgHandler::TcpMsgHandler(AbstractNode *abstract_node, const std::shared_ptr &conn, - const std::shared_ptr &meta, DataPtr data, size_t size) + const std::shared_ptr &meta, const DataPtr data, size_t size) : abstract_node_(abstract_node), tcp_conn_(conn), meta_(meta), data_ptr_(data), data_(nullptr), len_(size) { if (data_ptr_ != nullptr) { data_ = data_ptr_.get(); diff --git a/mindspore/ccsrc/ps/core/communicator/tcp_msg_handler.h b/mindspore/ccsrc/ps/core/communicator/tcp_msg_handler.h index 7d5c577c7fa..c2ab21bd71c 100644 --- a/mindspore/ccsrc/ps/core/communicator/tcp_msg_handler.h +++ b/mindspore/ccsrc/ps/core/communicator/tcp_msg_handler.h @@ -29,7 +29,7 @@ namespace core { class TcpMsgHandler : public MessageHandler { public: TcpMsgHandler(AbstractNode *abstract_node, const std::shared_ptr &conn, - const std::shared_ptr &meta, DataPtr data, size_t size); + const std::shared_ptr &meta, const DataPtr data, size_t size); ~TcpMsgHandler() override = default; void *data() const override; diff --git a/mindspore/ccsrc/ps/core/node_manager.cc b/mindspore/ccsrc/ps/core/node_manager.cc index 548670ed451..badc74552ba 100644 --- a/mindspore/ccsrc/ps/core/node_manager.cc +++ b/mindspore/ccsrc/ps/core/node_manager.cc @@ -28,8 +28,7 @@ void NodeManager::InitNode() { total_node_num_ = initial_total_node_num_; } -uint32_t NodeManager::checkIfRankIdExist(const RegisterMessage ®ister_message, - const std::shared_ptr &meta) { +uint32_t NodeManager::checkIfRankIdExist(const RegisterMessage ®ister_message) { uint32_t rank_id = UINT_MAX; const std::string &node_id = register_message.node_id(); if (registered_nodes_info_.find(node_id) != registered_nodes_info_.end()) { @@ -70,11 +69,11 @@ uint32_t NodeManager::NextRankId(const RegisterMessage ®ister_message, const MS_EXCEPTION_IF_NULL(meta); MS_EXCEPTION_IF_NULL(meta_data_); std::lock_guard lock(assign_rank_id_mutex_); - uint32_t rank_id = checkIfRankIdExist(register_message, meta); + uint32_t rank_id = checkIfRankIdExist(register_message); if (rank_id != UINT_MAX) { return rank_id; } - if (total_node_num_ == SizeToInt(registered_nodes_info_.size())) { + if (total_node_num_ == SizeToUint(registered_nodes_info_.size())) { MS_LOG(WARNING) << "There are enough nodes registering to scheduler."; return UINT_MAX; } @@ -94,11 +93,12 @@ uint32_t NodeManager::NextRankId(const RegisterMessage ®ister_message, const return res; }); if (rank_it == registered_nodes_info_.end()) { - if (meta->rank_id() != UINT32_MAX && UintToInt(meta->rank_id()) <= next_server_rank_id_) { + if (meta->rank_id() != UINT32_MAX && meta->rank_id() < next_server_rank_id_) { rank_id = meta->rank_id(); MS_LOG(INFO) << "Use the old rank id:" << rank_id; } else { - rank_id = ++next_server_rank_id_; + rank_id = next_server_rank_id_; + ++next_server_rank_id_; } } else { registered_nodes_info_.erase((*rank_it).first); @@ -134,11 +134,12 @@ uint32_t NodeManager::NextRankId(const RegisterMessage ®ister_message, const return res; }); if (worker_rank_it == registered_nodes_info_.end()) { - if (meta->rank_id() != UINT32_MAX && UintToInt(meta->rank_id()) <= next_worker_rank_id_) { + if (meta->rank_id() != UINT32_MAX && meta->rank_id() < next_worker_rank_id_) { rank_id = meta->rank_id(); MS_LOG(INFO) << "Use the old rank id:" << rank_id; } else { - rank_id = ++next_worker_rank_id_; + rank_id = next_worker_rank_id_; + ++next_worker_rank_id_; } } else { registered_nodes_info_.erase((*worker_rank_it).first); @@ -224,7 +225,7 @@ void NodeManager::UpdateCluster() { if (onPersist) { onPersist(); } - } else if (SizeToInt(heartbeats_.size()) == total_node_num_) { + } else if (SizeToUint(heartbeats_.size()) == total_node_num_) { if (cluster_state_ == ClusterState::NODE_TIMEOUT) { for (auto it = registered_nodes_info_.begin(); it != registered_nodes_info_.end(); ++it) { if (registered_nodes_info_.count(it->first)) { @@ -239,23 +240,11 @@ void NodeManager::UpdateCluster() { } // 2. update cluster finish state - if (SizeToInt(finish_nodes_id_.size()) == total_node_num_ || - SizeToInt(finish_nodes_id_.size()) == current_node_num_) { + if (SizeToUint(finish_nodes_id_.size()) == total_node_num_) { UpdateClusterState(ClusterState::CLUSTER_EXIT); } } -void NodeManager::CheckClusterTimeout() { - if (total_node_num_ != SizeToInt(registered_nodes_info_.size())) { - MS_LOG(WARNING) << "The cluster is not ready after " - << PSContext::instance()->cluster_config().cluster_available_timeout - << " seconds,so finish the cluster, and change total node number from " << total_node_num_ << " to " - << registered_nodes_info_.size(); - current_node_num_ = SizeToInt(registered_nodes_info_.size()); - UpdateClusterState(ClusterState::NODE_TIMEOUT); - } -} - void NodeManager::AddFinishNode(const std::string &finish_message) { finish_nodes_id_.insert(finish_message); } void NodeManager::AddScaleOutDoneNode(const std::string &node_id) { scale_out_done_nodes_id_.insert(node_id); } @@ -263,18 +252,20 @@ void NodeManager::AddScaleOutDoneNode(const std::string &node_id) { scale_out_do void NodeManager::AddScaleInDoneNode(const std::string &node_id) { scale_in_done_nodes_id_.insert(node_id); } bool NodeManager::IsAllNodesRegistered() const { - int32_t num = std::count_if(registered_nodes_info_.begin(), registered_nodes_info_.end(), - [](auto item) { return item.second.is_alive == true; }); + uint32_t num = std::count_if(registered_nodes_info_.begin(), registered_nodes_info_.end(), + [](auto item) { return item.second.is_alive == true; }); return num == total_node_num_; } -bool NodeManager::IsAllNodesFinished() const { return SizeToInt(finish_nodes_id_.size()) == total_node_num_; } +bool NodeManager::IsAllNodesFinished() const { return SizeToUint(finish_nodes_id_.size()) == total_node_num_; } bool NodeManager::IsAllNodesScaleOutDone() const { - return SizeToInt(scale_out_done_nodes_id_.size()) == total_node_num_; + return SizeToUint(scale_out_done_nodes_id_.size()) == total_node_num_; } -bool NodeManager::IsAllNodesScaleInDone() const { return SizeToInt(scale_in_done_nodes_id_.size()) == total_node_num_; } +bool NodeManager::IsAllNodesScaleInDone() const { + return SizeToUint(scale_in_done_nodes_id_.size()) == total_node_num_; +} const std::unordered_map &NodeManager::nodes_info() const { return nodes_info_; } @@ -362,7 +353,6 @@ bool NodeManager::IsNodeRegistered(const std::string &node_id) { const NodeInfo NodeManager::QueryNodeInfo(const std::string &node_id) const { auto iter = registered_nodes_info_.find(node_id); if (iter == registered_nodes_info_.end()) { - MS_LOG(DEBUG) << "Cannot find node of id: " << node_id; return NodeInfo(); } return iter->second; @@ -376,33 +366,33 @@ void NodeManager::AddPersistingNode(const std::string &node_id) { nodes_persisti bool NodeManager::IsAllNodeInPersisting() { // The worker role does not support disaster recovery currently. - if (nodes_persisting_.size() == IntToSize(server_num())) { + if (SizeToUint(nodes_persisting_.size()) == server_num()) { nodes_persisting_.clear(); return true; } return false; } -void NodeManager::set_total_node_num(const int32_t &node_num) { total_node_num_ = node_num; } +void NodeManager::set_total_node_num(const uint32_t &node_num) { total_node_num_ = node_num; } -const int32_t &NodeManager::total_node_num() const { return total_node_num_; } +const uint32_t &NodeManager::total_node_num() const { return total_node_num_; } -void NodeManager::set_worker_num(const int32_t &worker_num) { meta_data_->worker_num = IntToUint(worker_num); } +void NodeManager::set_worker_num(const uint32_t &worker_num) { meta_data_->worker_num = worker_num; } -void NodeManager::set_server_num(const int32_t &server_num) { meta_data_->server_num = IntToUint(server_num); } +void NodeManager::set_server_num(const uint32_t &server_num) { meta_data_->server_num = server_num; } -int32_t NodeManager::worker_num() const { return UintToInt(meta_data_->worker_num); } +uint32_t NodeManager::worker_num() const { return meta_data_->worker_num; } -int32_t NodeManager::server_num() const { return UintToInt(meta_data_->server_num); } +uint32_t NodeManager::server_num() const { return meta_data_->server_num; } -int32_t NodeManager::next_worker_rank_id() const { return next_worker_rank_id_.load(); } +uint32_t NodeManager::next_worker_rank_id() const { return next_worker_rank_id_.load(); } -int32_t NodeManager::next_server_rank_id() const { return next_server_rank_id_.load(); } +uint32_t NodeManager::next_server_rank_id() const { return next_server_rank_id_.load(); } -void NodeManager::set_next_worker_rank_id(const int32_t &next_worker_rank_id) { +void NodeManager::set_next_worker_rank_id(const uint32_t &next_worker_rank_id) { this->next_worker_rank_id_ = next_worker_rank_id; } -void NodeManager::set_next_server_rank_id(const int32_t &next_server_rank_id) { +void NodeManager::set_next_server_rank_id(const uint32_t &next_server_rank_id) { this->next_server_rank_id_ = next_server_rank_id; } void NodeManager::setPersistCallback(const OnPersist &onPersist) { this->onPersist = onPersist; } diff --git a/mindspore/ccsrc/ps/core/node_manager.h b/mindspore/ccsrc/ps/core/node_manager.h index 4e973399928..30e9f799ada 100644 --- a/mindspore/ccsrc/ps/core/node_manager.h +++ b/mindspore/ccsrc/ps/core/node_manager.h @@ -45,10 +45,9 @@ class NodeManager { public: NodeManager() : initial_total_node_num_(0), - total_node_num_(-1), - current_node_num_(-1), - next_worker_rank_id_(-1), - next_server_rank_id_(-1), + total_node_num_(0), + next_worker_rank_id_(0), + next_server_rank_id_(0), meta_data_(nullptr), node_state_(NodeState::NODE_STARTING), cluster_state_(ClusterState::ClUSTER_STARTING) {} @@ -57,7 +56,7 @@ class NodeManager { // When initializing nodes, the initial number of nodes will be assigned to the total number of nodes. void InitNode(); uint32_t NextRankId(const RegisterMessage ®ister_message, const std::shared_ptr &meta); - uint32_t checkIfRankIdExist(const RegisterMessage ®ister_message, const std::shared_ptr &meta); + uint32_t checkIfRankIdExist(const RegisterMessage ®ister_message); void UpdateHeartbeat(const std::string &node_id); std::vector FetchServersMeta(); @@ -65,7 +64,6 @@ class NodeManager { std::vector FetchAllNodesMeta(); void UpdateCluster(); - void CheckClusterTimeout(); void AddFinishNode(const std::string &finish_message); // After the scheduler receives the scale_out_done node, it will save this node. @@ -92,15 +90,15 @@ class NodeManager { // After all the nodes are registered successfully, the nodes info can be updated. void UpdateNodesInfo(); - void set_total_node_num(const int32_t &node_num); - const int32_t &total_node_num() const; - void set_worker_num(const int32_t &worker_num); - void set_server_num(const int32_t &server_num); - int32_t worker_num() const; - int32_t server_num() const; + void set_total_node_num(const uint32_t &node_num); + const uint32_t &total_node_num() const; + void set_worker_num(const uint32_t &worker_num); + void set_server_num(const uint32_t &server_num); + uint32_t worker_num() const; + uint32_t server_num() const; - int32_t next_worker_rank_id() const; - int32_t next_server_rank_id() const; + uint32_t next_worker_rank_id() const; + uint32_t next_server_rank_id() const; void UpdateNodeState(const NodeState &state); void UpdateClusterState(const ClusterState &state); @@ -119,8 +117,8 @@ class NodeManager { bool IsNodeRegistered(const std::string &node_id); void set_registered_nodes_info(const std::unordered_map registered_nodes_info); - void set_next_worker_rank_id(const int32_t &next_worker_rank_id); - void set_next_server_rank_id(const int32_t &next_server_rank_id); + void set_next_worker_rank_id(const uint32_t &next_worker_rank_id); + void set_next_server_rank_id(const uint32_t &next_server_rank_id); void setPersistCallback(const OnPersist &onPersist); // Query node information by node id. @@ -140,11 +138,11 @@ class NodeManager { std::mutex cluster_mutex_; uint32_t initial_total_node_num_; - int32_t total_node_num_; - int32_t current_node_num_; + uint32_t total_node_num_; + uint32_t current_node_num_; - std::atomic next_worker_rank_id_; - std::atomic next_server_rank_id_; + std::atomic next_worker_rank_id_; + std::atomic next_server_rank_id_; // Whenever a node is registered, it will be stored in this map. std::unordered_map registered_nodes_info_; diff --git a/mindspore/ccsrc/ps/core/scheduler_node.cc b/mindspore/ccsrc/ps/core/scheduler_node.cc index 1a1d2588d76..44edde0b87c 100644 --- a/mindspore/ccsrc/ps/core/scheduler_node.cc +++ b/mindspore/ccsrc/ps/core/scheduler_node.cc @@ -137,14 +137,13 @@ void SchedulerNode::ProcessHeartbeat(const std::shared_ptr &server, CHECK_RETURN_TYPE(heartbeat_message.ParseFromArray(data, SizeToInt(size))); std::string node_id = heartbeat_message.node_id(); - node_manager_.UpdateHeartbeat(node_id); MS_LOG(DEBUG) << "The scheduler get a heartbeat from node id :" << heartbeat_message.node_id(); HeartbeatRespMessage heartbeat_resp_message; heartbeat_resp_message.set_persistent_cmd(PersistentCommand::DEFAULT); NodeInfo nodeInfo = node_manager_.QueryNodeInfo(node_id); - if (nodeInfo.node_id_ != "") { + if (!nodeInfo.node_id_.empty()) { // The worker role does not support disaster recovery for the time being. NodeRole node_role = nodeInfo.node_role_; if (node_role == NodeRole::SERVER && persistent_cmd_ == PersistentCommand::BEGIN_PERSIST) { @@ -156,6 +155,7 @@ void SchedulerNode::ProcessHeartbeat(const std::shared_ptr &server, persistent_cmd_ = PersistentCommand::DEFAULT; } } + node_manager_.UpdateHeartbeat(node_id); } MS_LOG(DEBUG) << "The cluster state:" << CommUtil::ClusterStateToString(node_manager_.GetClusterState()); @@ -223,14 +223,13 @@ void SchedulerNode::CreateTcpServer() { const auto client_disconn = [&](const TcpServer &, const TcpConnection &conn) { int fd = conn.GetFd(); - if (register_connection_fd_.count(fd) <= 0) { - return; + if (register_connection_fd_.count(fd) > 0) { + MS_LOG(WARNING) << "remove client fd:" << fd << ", remove client id:" << register_connection_fd_[fd]; + (void)register_connection_fd_.erase(fd); + MS_LOG(INFO) << "Register node number is:" << register_connection_fd_.size() + << ", total node num is:" << node_manager_.total_node_num() + << ", scale in node size is: " << scale_in_node_ids_.size(); } - MS_LOG(WARNING) << "remove client fd:" << fd << ", remove client id:" << register_connection_fd_[fd]; - register_connection_fd_.erase(fd); - MS_LOG(WARNING) << "Register node number is:" << register_connection_fd_.size() - << ", total node num is:" << node_manager_.total_node_num() - << ", scale in node size is: " << scale_in_node_ids_.size(); }; server_->SetServerCallback(nullptr, client_disconn, nullptr); server_->Init(); @@ -306,7 +305,7 @@ void SchedulerNode::ProcessRegister(const std::shared_ptr &server, auto node_infos = node_manager_.nodes_info(); bool res = SendPrepareBuildingNetwork(node_infos); if (!res) { - MS_LOG(WARNING) << "Prepare for building network failed!"; + MS_LOG(ERROR) << "Prepare for building network failed!"; return; } MS_LOG(INFO) << "Prepare for building network success."; @@ -461,23 +460,22 @@ void SchedulerNode::ProcessSendEvent(const std::shared_ptr &server, } bool SchedulerNode::SendPrepareBuildingNetwork(const std::unordered_map &node_infos) { - std::string timeoutNodeId = ""; + uint64_t request_id = AddMessageTrack(node_infos.size()); for (const auto &kvs : node_infos) { auto client = GetOrCreateClient(kvs.second); + MS_ERROR_IF_NULL_W_RET_VAL(client, false); auto message_meta = std::make_shared(); MS_EXCEPTION_IF_NULL(message_meta); + message_meta->set_request_id(request_id); message_meta->set_cmd(NodeCommand::PREPARE_BUILDING_NETWORK); - SendMetadataMessage send_metadata_message; - send_metadata_message.set_rank_id(kvs.second.rank_id_); - if (!SendMessageSync(client, message_meta, Protos::PROTOBUF, send_metadata_message.SerializeAsString().data(), - send_metadata_message.ByteSizeLong(), kCommTimeoutInThreeSeconds)) { + std::string req_data; + if (!client->SendMessage(message_meta, Protos::RAW, req_data.data(), req_data.length())) { MS_LOG(ERROR) << "The node role:" << CommUtil::NodeRoleToString(kvs.second.node_role_) << " the node id:" << kvs.second.node_id_ << " prepare building network timeout!"; - timeoutNodeId += kvs.second.node_id_ + " "; } } - return timeoutNodeId.empty(); + return Wait(request_id); } void SchedulerNode::SendMetadata(const std::shared_ptr &client, uint32_t rank_id) { @@ -588,7 +586,7 @@ void SchedulerNode::StartUpdateClusterStateTimer() { // 1. update cluster timeout if (!is_ready_ && (std::chrono::steady_clock::now() - start_time > std::chrono::seconds(PSContext::instance()->cluster_config().cluster_available_timeout))) { - node_manager_.CheckClusterTimeout(); + node_manager_.UpdateClusterState(ClusterState::CLUSTER_EXIT); } std::this_thread::sleep_for(std::chrono::seconds(PSContext::instance()->cluster_config().heartbeat_interval)); node_manager_.UpdateCluster(); @@ -631,7 +629,7 @@ const std::shared_ptr &SchedulerNode::GetOrCreateClient(const NodeInf } std::string ip = node_info.ip_; uint16_t port = node_info.port_; - MS_LOG(DEBUG) << "ip:" << ip << ", port:" << port << ", node id:" << node_info.node_id_; + MS_LOG(INFO) << "ip:" << ip << ", port:" << port << ", node id:" << node_info.node_id_; auto client = std::make_shared(ip, port, config_.get()); MS_EXCEPTION_IF_NULL(client); client->SetMessageCallback( @@ -1344,7 +1342,7 @@ void SchedulerNode::PersistMetaData() { } bool SchedulerNode::CheckIfNodeDisconnected() const { - return UintToInt(register_connection_fd_.size()) != node_manager_.total_node_num(); + return UlongToUint(register_connection_fd_.size()) != node_manager_.total_node_num(); } void SchedulerNode::BroadcastTimeoutEvent() { diff --git a/mindspore/ccsrc/ps/core/scheduler_recovery.cc b/mindspore/ccsrc/ps/core/scheduler_recovery.cc index e0422f1d5ee..90dc049169c 100644 --- a/mindspore/ccsrc/ps/core/scheduler_recovery.cc +++ b/mindspore/ccsrc/ps/core/scheduler_recovery.cc @@ -32,16 +32,16 @@ bool SchedulerRecovery::Recover() { // 1. recover worker num if (recovery_storage_->Exists(kRecoveryWorkerNum)) { - clusterConfig.initial_worker_num = - std::strtol(recovery_storage_->Get(kRecoveryWorkerNum, "").c_str(), nullptr, kBase); + uint32_t initial_worker_num = std::strtol(recovery_storage_->Get(kRecoveryWorkerNum, "").c_str(), nullptr, kBase); + clusterConfig.initial_worker_num = IntToUint(initial_worker_num); } else { clusterConfig.initial_worker_num = PSContext::instance()->initial_worker_num(); } // 2. recover server num if (recovery_storage_->Exists(kRecoveryServerNum)) { - clusterConfig.initial_server_num = - std::strtol(recovery_storage_->Get(kRecoveryServerNum, "").c_str(), nullptr, kBase); + uint32_t initial_server_num = std::strtol(recovery_storage_->Get(kRecoveryServerNum, "").c_str(), nullptr, kBase); + clusterConfig.initial_server_num = IntToUint(initial_server_num); } else { clusterConfig.initial_server_num = PSContext::instance()->initial_server_num(); } @@ -55,8 +55,8 @@ bool SchedulerRecovery::Recover() { // 4. recover scheduler port if (recovery_storage_->Exists(kRecoverySchedulerPort)) { - clusterConfig.scheduler_port = - std::strtol(recovery_storage_->Get(kRecoverySchedulerPort, "").c_str(), nullptr, kBase); + uint16_t scheduler_port = std::strtol(recovery_storage_->Get(kRecoverySchedulerPort, "").c_str(), nullptr, kBase); + clusterConfig.scheduler_port = scheduler_port; } else { clusterConfig.scheduler_port = PSContext::instance()->scheduler_port(); } @@ -72,20 +72,23 @@ bool SchedulerRecovery::Recover() { } // 5. recover total node num if (scheduler_recovery_storage_->Exists(kRecoveryTotalNodeNum)) { - clusterConfig.initial_total_node_num = + uint32_t initial_total_node_num = std::strtol(scheduler_recovery_storage_->Get(kRecoveryTotalNodeNum, "").c_str(), nullptr, kBase); + clusterConfig.initial_total_node_num = initial_total_node_num; } // 6. recover next worker rank id if (scheduler_recovery_storage_->Exists(kRecoveryNextWorkerRankId)) { - clusterConfig.initial_next_worker_rank_id = + uint32_t initial_next_worker_rank_id = std::strtol(scheduler_recovery_storage_->Get(kRecoveryNextWorkerRankId, "").c_str(), nullptr, kBase); + clusterConfig.initial_next_worker_rank_id = initial_next_worker_rank_id; } // 7. recover next server rank id if (scheduler_recovery_storage_->Exists(kRecoveryNextServerRankId)) { - clusterConfig.initial_next_server_rank_id = + uint32_t initial_next_server_rank_id = std::strtol(scheduler_recovery_storage_->Get(kRecoveryNextServerRankId, "").c_str(), nullptr, kBase); + clusterConfig.initial_next_server_rank_id = initial_next_server_rank_id; } // 8. recover register nodes info @@ -98,9 +101,11 @@ bool SchedulerRecovery::Recover() { NodeInfo node_info; node_info.ip_ = elem.at("ip"); - node_info.port_ = std::strtol(port.c_str(), nullptr, kBase); + uint16_t uint_port = std::strtol(port.c_str(), nullptr, kBase); + node_info.port_ = uint_port; node_info.node_id_ = elem.at("node_id"); - node_info.rank_id_ = std::strtol(rank_id.c_str(), nullptr, kBase); + uint32_t uint_rank_id = std::strtol(rank_id.c_str(), nullptr, kBase); + node_info.rank_id_ = uint_rank_id; node_info.is_alive = CommUtil::StringToBool(elem.at("alive")); node_info.node_role_ = CommUtil::StringToNodeRole(elem.at("role")); diff --git a/scripts/fl_restful_tool.py b/scripts/fl_restful_tool.py index 1a1d2d7272c..c519d57380c 100644 --- a/scripts/fl_restful_tool.py +++ b/scripts/fl_restful_tool.py @@ -1,5 +1,3 @@ -#!/usr/bin/env python3 -# coding=UTF-8 # Copyright 2020 Huawei Technologies Co., Ltd # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -24,30 +22,34 @@ import argparse import json import os import warnings +from enum import Enum import requests -class Status: - success = "0" - failed = "1" +class Status(Enum): + """ + Response Status + """ + SUCCESS = "0" + FAILED = "1" -class Restful: +class Restful(Enum): """ Define restful interface constant """ + SCALE = "scale" + SCALE_OUT = "scaleout" + SCALE_IN = "scalein" + NODES = "nodes" + GET_INSTANCE_DETAIL = "getInstanceDetail" + NEW_INSTANCE = "newInstance" + QUERY_INSTANCE = "queryInstance" + ENABLE_FLS = "enableFLS" + DISABLE_FLS = "disableFLS" + STATE = "state" + SCALE_OUT_ROLLBACK = "scaleoutRollback" - scale = "scale" - scaleout = "scaleout" - scalein = "scalein" - nodes = "nodes" - getInstanceDetail = "getInstanceDetail" - newInstance = "newInstance" - queryInstance = "queryInstance" - enableFLS = "enableFLS" - disableFLS = "disableFLS" - state = "state" - scaleoutRollback = "scaleoutRollback" warnings.filterwarnings('ignore') @@ -55,11 +57,9 @@ parser = argparse.ArgumentParser() parser.add_argument("--http_type", type=str, default="http", help="http or https") parser.add_argument("--ip", type=str, default="127.0.0.1") parser.add_argument("--port", type=int, default=6666) -# scaleout scalein nodes parser.add_argument("--request_name", type=str, default="") parser.add_argument("--server_num", type=int, default=0) -# "start_fl_job_threshold=20,start_fl_job_time_window=2000..." parser.add_argument("--instance_param", type=str, default="") parser.add_argument("--metrics_file_path", type=str, default="/opt/huawei/mindspore/hybrid_albert/metrics.json") @@ -82,12 +82,12 @@ def call_scale(): call cluster scale out or scale in """ if server_num == 0: - return process_self_define_json(Status.failed, "error. server_num is 0") + return process_self_define_json(Status.FAILED.value, "error. server_num is 0") node_ids = json.loads(call_nodes())["result"] cluster_abstract_node_num = len(node_ids) if cluster_abstract_node_num == 0: - return process_self_define_json(Status.failed, "error. cluster abstract node num is 0") + return process_self_define_json(Status.FAILED.value, "error. cluster abstract node num is 0") cluster_server_node_num = 0 cluster_worker_node_num = 0 @@ -103,7 +103,7 @@ def call_scale(): else: pass if cluster_server_node_num == server_num: - return process_self_define_json(Status.failed, "error. cluster server num is same with server_num.") + return process_self_define_json(Status.FAILED.value, "error. cluster server num is same with server_num.") if cluster_server_node_num > server_num: scale_in_len = cluster_server_node_num - server_num scale_in_node_ids = [] @@ -115,56 +115,67 @@ def call_scale(): def call_scaleout(scale_out_server_num, scale_out_worker_num=0): - url = base_url + "scaleout" + """ + call scaleout + """ + url = base_url + Restful.SCALE_OUT.value data = {"server_num": scale_out_server_num, "worker_num": scale_out_worker_num} res = session.post(url, headers=headers, verify=False, data=json.dumps(data)) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) result = "scale out server num is " + str(scale_out_server_num) - return process_result_json(Status.success, res_json["message"], result) + return process_result_json(Status.SUCCESS.value, res_json["message"], result) + def call_scaleout_rollback(): - url = base_url + Restful.scaleoutRollback + """ + call scaleout rollback + """ + url = base_url + Restful.SCALE_OUT_ROLLBACK.value res = session.get(url, verify=False) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) - return process_self_define_json(Status.success, res_json["message"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) + return process_self_define_json(Status.SUCCESS.value, res_json["message"]) + def call_scalein(scale_in_node_ids): """ call cluster to scale in """ if not scale_in_node_ids: - return process_self_define_json(Status.failed, "error. node ids is empty.") + return process_self_define_json(Status.FAILED.value, "error. node ids is empty.") - url = base_url + "scalein" + url = base_url + Restful.SCALE_IN.value data = {"node_ids": scale_in_node_ids} res = session.post(url, headers=headers, verify=False, data=json.dumps(data)) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) result = "scale in node ids is " + str(scale_in_node_ids) - return process_result_json(Status.success, res_json["message"], result) + return process_result_json(Status.SUCCESS.value, res_json["message"], result) def call_nodes(): - url = base_url + Restful.nodes + """ + get nodes info + """ + url = base_url + Restful.NODES.value res = session.get(url, verify=False) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) - return process_result_json(Status.success, res_json["message"], res_json["nodeIds"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) + return process_result_json(Status.SUCCESS.value, res_json["message"], res_json["nodeIds"]) def call_get_instance_detail(): """ - call cluster get instance detail + get cluster instance detail """ if not os.path.exists(metrics_file_path): - return process_self_define_json(Status.failed, "error. metrics file is not existed.") + return process_self_define_json(Status.FAILED.value, "error. metrics file is not existed.") ans_json_obj = {} joined_client_num_list = [] @@ -177,7 +188,7 @@ def call_get_instance_detail(): metrics_list = f.readlines() if not metrics_list: - return process_self_define_json(Status.failed, "error. metrics file has no content") + return process_self_define_json(Status.FAILED.value, "error. metrics file has no content") for metrics in metrics_list: json_obj = json.loads(metrics) @@ -190,18 +201,19 @@ def call_get_instance_detail(): last_metrics = metrics_list[len(metrics_list) - 1] last_metrics_obj = json.loads(last_metrics) - ans_json_obj["code"] = Status.success + ans_json_obj["code"] = Status.SUCCESS.value ans_json_obj["describe"] = "get instance metrics detail successful." ans_json_obj["result"] = {} - ans_json_obj["result"]['currentIteration'] = last_metrics_obj['currentIteration'] - ans_json_obj["result"]['flIterationNum'] = last_metrics_obj['flIterationNum'] - ans_json_obj["result"]['flName'] = last_metrics_obj['flName'] - ans_json_obj["result"]['instanceStatus'] = last_metrics_obj['instanceStatus'] - ans_json_obj["result"]['iterationExecutionTime'] = iteration_execution_time_list - ans_json_obj["result"]['joinedClientNum'] = joined_client_num_list - ans_json_obj["result"]['rejectedClientNum'] = rejected_client_num_list - ans_json_obj["result"]['metricsAuc'] = metrics_auc_list - ans_json_obj["result"]['metricsLoss'] = metrics_loss_list + ans_json_result = ans_json_obj.get("result") + ans_json_result['currentIteration'] = last_metrics_obj['currentIteration'] + ans_json_result['flIterationNum'] = last_metrics_obj['flIterationNum'] + ans_json_result['flName'] = last_metrics_obj['flName'] + ans_json_result['instanceStatus'] = last_metrics_obj['instanceStatus'] + ans_json_result['iterationExecutionTime'] = iteration_execution_time_list + ans_json_result['joinedClientNum'] = joined_client_num_list + ans_json_result['rejectedClientNum'] = rejected_client_num_list + ans_json_result['metricsAuc'] = metrics_auc_list + ans_json_result['metricsLoss'] = metrics_loss_list return json.dumps(ans_json_obj) @@ -211,11 +223,11 @@ def call_new_instance(): call cluster new instance """ if instance_param == "": - return process_self_define_json(Status.failed, "error. instance_param is empty.") + return process_self_define_json(Status.FAILED.value, "error. instance_param is empty.") instance_param_list = instance_param.split(sep=",") instance_param_json_obj = {} - url = base_url + Restful.newInstance + url = base_url + Restful.NEW_INSTANCE.value for cur in instance_param_list: pair = cur.split(sep="=") instance_param_json_obj[pair[0]] = float(pair[1]) @@ -223,84 +235,102 @@ def call_new_instance(): data = json.dumps(instance_param_json_obj) res = session.post(url, verify=False, data=data) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) - return process_self_define_json(Status.success, res_json["message"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) + return process_self_define_json(Status.SUCCESS.value, res_json["message"]) def call_query_instance(): - url = base_url + Restful.queryInstance + """ + query cluster instance + """ + url = base_url + Restful.QUERY_INSTANCE.value res = session.post(url, verify=False) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) - return process_result_json(Status.success, res_json["message"], res_json["result"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) + return process_result_json(Status.SUCCESS.value, res_json["message"], res_json["result"]) def call_enable_fls(): - url = base_url + Restful.enableFLS + """ + enable cluster fls + """ + url = base_url + Restful.ENABLE_FLS.value res = session.post(url, verify=False) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) - return process_self_define_json(Status.success, res_json["message"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) + return process_self_define_json(Status.SUCCESS.value, res_json["message"]) def call_disable_fls(): - url = base_url + Restful.disableFLS + """ + disable cluster fls + """ + url = base_url + Restful.DISABLE_FLS.value res = session.post(url, verify=False) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) - return process_self_define_json(Status.success, res_json["message"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) + return process_self_define_json(Status.SUCCESS.value, res_json["message"]) def call_state(): - url = base_url + Restful.state + """ + get cluster state + """ + url = base_url + Restful.STATE.value res = session.get(url, verify=False) res_json = json.loads(res.text) - if res_json["code"] == Status.failed: - return process_self_define_json(Status.failed, res_json["error_message"]) + if res_json["code"] == Status.FAILED.value: + return process_self_define_json(Status.FAILED.value, res_json["error_message"]) result = res_json['cluster_state'] - return process_result_json(Status.success, res_json["message"], result) + return process_result_json(Status.SUCCESS.value, res_json["message"], result) def process_result_json(code, describe, result): + """ + process result json + """ result_dict = {"code": code, "describe": describe, "result": result} return json.dumps(result_dict) def process_self_define_json(code, describe): + """ + process self define json + """ result_dict = {"code": code, "describe": describe} return json.dumps(result_dict) if __name__ == '__main__': - if request_name == Restful.scale: + if request_name == Restful.SCALE.value: print(call_scale()) - elif request_name == Restful.nodes: + elif request_name == Restful.NODES.value: print(call_nodes()) - elif request_name == Restful.getInstanceDetail: + elif request_name == Restful.GET_INSTANCE_DETAIL.value: print(call_get_instance_detail()) - elif request_name == Restful.newInstance: + elif request_name == Restful.NEW_INSTANCE.value: print(call_new_instance()) - elif request_name == Restful.queryInstance: + elif request_name == Restful.QUERY_INSTANCE.value: print(call_query_instance()) - elif request_name == Restful.enableFLS: + elif request_name == Restful.ENABLE_FLS.value: print(call_enable_fls()) - elif request_name == Restful.disableFLS: + elif request_name == Restful.DISABLE_FLS.value: print(call_disable_fls()) - elif request_name == Restful.state: + elif request_name == Restful.STATE.value: print(call_state()) - elif request_name == Restful.scaleoutRollback: + elif request_name == Restful.SCALE_OUT_ROLLBACK.value: print(call_scaleout_rollback()) else: