diff --git a/mindspore/ccsrc/fl/server/iteration.cc b/mindspore/ccsrc/fl/server/iteration.cc index 5b666f2c50a..f548ea7cfeb 100644 --- a/mindspore/ccsrc/fl/server/iteration.cc +++ b/mindspore/ccsrc/fl/server/iteration.cc @@ -431,6 +431,7 @@ void Iteration::HandleNotifyLeaderMoveToNextIterRequest(const std::shared_ptr lock(iter_move_mtx_); NotifyLeaderMoveToNextIterRequest notify_leader_to_next_iter_req; (void)notify_leader_to_next_iter_req.ParseFromArray(message->data(), SizeToInt(message->len())); const auto &rank = notify_leader_to_next_iter_req.rank(); diff --git a/mindspore/ccsrc/fl/server/iteration.h b/mindspore/ccsrc/fl/server/iteration.h index 31c9166d575..48238d5c857 100644 --- a/mindspore/ccsrc/fl/server/iteration.h +++ b/mindspore/ccsrc/fl/server/iteration.h @@ -290,6 +290,9 @@ class Iteration { std::atomic iteration_result_; nlohmann::json new_instance_json_; + + // mutex for iter move to next, avoid core dump + std::mutex iter_move_mtx_; }; } // namespace server } // namespace fl diff --git a/mindspore/ccsrc/ps/core/communicator/tcp_client.cc b/mindspore/ccsrc/ps/core/communicator/tcp_client.cc index 010df9d3dbd..3385931b7c6 100644 --- a/mindspore/ccsrc/ps/core/communicator/tcp_client.cc +++ b/mindspore/ccsrc/ps/core/communicator/tcp_client.cc @@ -43,8 +43,8 @@ TcpClient::TcpClient(const std::string &address, std::uint16_t port, NodeRole pe server_address_(std::move(address)), server_port_(port), peer_role_(peer_role), - is_stop_(true), - is_connected_(false) { + disconnected_(false), + connected_(false) { message_handler_.SetCallback( [this](const std::shared_ptr &meta, const Protos &protos, const void *data, size_t size) { if (message_callback_) { @@ -86,16 +86,16 @@ std::string TcpClient::PeerRoleName() const { bool TcpClient::WaitConnected(const uint32_t &connected_timeout) { std::unique_lock lock(connection_mutex_); bool res = connection_cond_.wait_for(lock, std::chrono::seconds(connected_timeout), - [this] { return this->is_connected_.load(); }); + [this] { return this->connected_.load(); }); return res; } void TcpClient::Init() { - std::lock_guard lock(connection_mutex_); - if (buffer_event_) { - bufferevent_free(buffer_event_); - buffer_event_ = nullptr; + if (disconnected_) { + return; } + + std::lock_guard lock(connection_mutex_); if (!CommUtil::CheckIp(server_address_)) { MS_LOG(EXCEPTION) << "The tcp client ip:" << server_address_ << " is illegal!"; } @@ -117,10 +117,10 @@ void TcpClient::Init() { sin.sin_addr.s_addr = inet_addr(server_address_.c_str()); sin.sin_port = htons(server_port_); - if (!PSContext::instance()->enable_ssl()) { + if (!PSContext::instance()->enable_ssl() && buffer_event_ == nullptr) { MS_LOG(INFO) << "SSL is disable."; buffer_event_ = bufferevent_socket_new(event_base_, -1, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE); - } else { + } else if (buffer_event_ == nullptr) { if (!EstablishSSL()) { MS_LOG(WARNING) << "Establish SSL failed."; return; @@ -228,7 +228,7 @@ void TcpClient::TimerCallback(evutil_socket_t, int16_t, void *arg) { void TcpClient::NotifyConnected() { MS_LOG(INFO) << "Client connected to the server! Peer " << PeerRoleName() << " ip: " << server_address_ << ", port: " << server_port_; - is_connected_ = true; + connected_ = true; connection_cond_.notify_all(); } diff --git a/mindspore/ccsrc/ps/core/communicator/tcp_client.h b/mindspore/ccsrc/ps/core/communicator/tcp_client.h index 0c5493bb912..8484bb60cf3 100644 --- a/mindspore/ccsrc/ps/core/communicator/tcp_client.h +++ b/mindspore/ccsrc/ps/core/communicator/tcp_client.h @@ -72,6 +72,7 @@ class TcpClient { bool SendMessage(const std::shared_ptr &meta, const Protos &protos, const void *data, size_t size); void set_timer_callback(const OnTimer &timer); const event_base &eventbase() const; + void set_disconnected() { disconnected_ = true; } protected: static void SetTcpNoDelay(const evutil_socket_t &fd); @@ -109,8 +110,8 @@ class TcpClient { std::string server_address_; std::uint16_t server_port_; NodeRole peer_role_; - std::atomic is_stop_; - std::atomic is_connected_; + std::atomic disconnected_; + std::atomic connected_; // The Configuration file Configuration *config_; }; diff --git a/mindspore/ccsrc/ps/core/node_manager.cc b/mindspore/ccsrc/ps/core/node_manager.cc index a61373da93f..d529d0fb273 100644 --- a/mindspore/ccsrc/ps/core/node_manager.cc +++ b/mindspore/ccsrc/ps/core/node_manager.cc @@ -38,10 +38,10 @@ uint32_t NodeManager::checkIfRankIdExist(const RegisterMessage ®ister_message registered_nodes_info_[node_id].is_alive = true; registered_nodes_info_[node_id].ip_ = new_ip; registered_nodes_info_[node_id].port_ = static_cast(new_port); - MS_LOG(INFO) << "The node id: " << node_id << " is already assigned!" - << ", ip: " << register_message.ip() << ", port: " << register_message.port() - << ", rank id: " << rank_id << ", alive: " << registered_nodes_info_[node_id].is_alive - << ", the node_role:" << CommUtil::NodeRoleToString(registered_nodes_info_[node_id].node_role_); + MS_LOG(WARNING) << "The node id: " << node_id << " is already assigned!" + << ", ip: " << register_message.ip() << ", port: " << register_message.port() + << ", rank id: " << rank_id << ", alive: " << registered_nodes_info_[node_id].is_alive + << ", the node_role:" << CommUtil::NodeRoleToString(registered_nodes_info_[node_id].node_role_); return rank_id; } // This is for scheduler recovery diff --git a/mindspore/ccsrc/ps/core/server_node.cc b/mindspore/ccsrc/ps/core/server_node.cc index 9fd6851c0fe..e412d93f37a 100644 --- a/mindspore/ccsrc/ps/core/server_node.cc +++ b/mindspore/ccsrc/ps/core/server_node.cc @@ -94,6 +94,12 @@ bool ServerNode::Finish(const uint32_t &timeout) { return true; } + if (!is_connected_to_scheduler_) { + MS_LOG(INFO) << "[Server finish]: Not connect to scheduler, no need to disconnect!"; + return true; + } + client_to_scheduler_->set_disconnected(); + MS_LOG(INFO) << "[Server finish]: 1. Begin to finish server node!"; bool res = Disconnect(client_to_scheduler_, timeout); if (res) { diff --git a/mindspore/ccsrc/ps/core/worker_node.cc b/mindspore/ccsrc/ps/core/worker_node.cc index c7e67d6a5fc..40e0408dd26 100644 --- a/mindspore/ccsrc/ps/core/worker_node.cc +++ b/mindspore/ccsrc/ps/core/worker_node.cc @@ -95,6 +95,13 @@ bool WorkerNode::Finish(const uint32_t &timeout) { MS_LOG(INFO) << "The node is already stop."; return true; } + + if (!is_connected_to_scheduler_) { + MS_LOG(INFO) << "[Worker finish]: Not connect to scheduler, no need to disconnect!"; + return true; + } + client_to_scheduler_->set_disconnected(); + bool res = Disconnect(client_to_scheduler_, timeout); if (res) { MS_LOG(INFO) << "[Worker finish]: 2. Successfully finish worker node!";