diff --git a/mindspore/ccsrc/ps/core/abstract_node.cc b/mindspore/ccsrc/ps/core/abstract_node.cc
index 0f124b2f69b..688f598a745 100644
--- a/mindspore/ccsrc/ps/core/abstract_node.cc
+++ b/mindspore/ccsrc/ps/core/abstract_node.cc
@@ -344,18 +344,15 @@ std::pair<uint32_t, uint64_t> AbstractNode::CollectiveReceiveAsync(const enum No
     MS_LOG(DEBUG) << "Receive data from rank id:" << rank_id << ", the rank request id is:" << rank_request_id;
   } else {
     receive_callbacks_[std::make_pair(rank_id, rank_request_id)] = [=]() mutable {
-      receive_callbacks_mutex_.lock();
       auto res = received_data_[std::make_pair(rank_id, rank_request_id)];
       if (*output != nullptr) {
         MS_LOG(WARNING) << "The output is not empty.";
-
       } else {
         *output = res;
         received_data_.erase(std::make_pair(rank_id, rank_request_id));
         receive_messages_done_[std::make_pair(rank_id, rank_request_id)] = true;
         MS_LOG(DEBUG) << "Receive data from rank id:" << rank_id << ", the rank request id is:" << rank_request_id;
       }
-      receive_callbacks_mutex_.unlock();
     };
   }
   receive_callbacks_mutex_.unlock();
@@ -571,7 +568,7 @@ void AbstractNode::ProcessSendMetadata(std::shared_ptr<TcpConnection> conn, std:
   nodes_address_.clear();
   for (const auto &it : send_meta_message.servers_meta()) {
     nodes_address_[std::make_pair(NodeRole::SERVER, it.rank_id())] = std::make_pair(it.ip(), it.port());
-    MS_LOG(INFO) << "The server ip is:" << it.ip() << ", the port is:" << it.port();
+    MS_LOG(INFO) << "The server ip is:" << it.ip() << ", the port is:" << it.port() << ", the rank id:" << it.rank_id();
   }
   server_->SendMessage(conn, meta, Protos::RAW, data, size);
   is_ready_ = true;
@@ -851,13 +848,11 @@ void AbstractNode::RunReceiveCallback(std::shared_ptr<MessageMeta> meta, const P
                 << ", the send request id is:" << meta->request_id() << " the size is:" << size;
   auto it = receive_callbacks_.find(std::make_pair(rank_id, rank_request_id));
   if (it != receive_callbacks_.end()) {
-    receive_callbacks_mutex_.unlock();
-
-    if (it->second) {
-      it->second();
+    if (receive_messages_done_[std::make_pair(rank_id, rank_request_id)] == false) {
+      if (it->second) {
+        it->second();
+      }
     }
-
-    receive_callbacks_mutex_.lock();
     receive_cond_.notify_all();
     receive_callbacks_.erase(it);
   }
diff --git a/mindspore/ccsrc/ps/core/node.h b/mindspore/ccsrc/ps/core/node.h
index 01652283834..2ed8a0c79b1 100644
--- a/mindspore/ccsrc/ps/core/node.h
+++ b/mindspore/ccsrc/ps/core/node.h
@@ -42,7 +42,7 @@ namespace mindspore {
 namespace ps {
 namespace core {
 constexpr int kTimeoutInSeconds = 30;
-constexpr int kCommTimeoutInSeconds = 30;
+constexpr int kCommTimeoutInSeconds = 3;
 class Node {
  public:
   Node()
diff --git a/mindspore/ccsrc/ps/core/node_manager.cc b/mindspore/ccsrc/ps/core/node_manager.cc
index 5eff34a9836..48d4b5722fd 100644
--- a/mindspore/ccsrc/ps/core/node_manager.cc
+++ b/mindspore/ccsrc/ps/core/node_manager.cc
@@ -45,7 +45,7 @@ uint32_t NodeManager::NextRankId(const RegisterMessage &register_message) {
     auto rank_it = std::find_if(registered_nodes_info_.begin(), registered_nodes_info_.end(), [&rank_id](auto item) {
       bool res = item.second.is_alive == false && item.second.node_role_ == NodeRole::SERVER;
       if (res) {
-        MS_LOG(INFO) << "The server node id:" << item.first << " rank id:" << rank_id << " is not alive.";
+        MS_LOG(INFO) << "The server node id:" << item.first << " rank id:" << item.second.rank_id_ << " is not alive.";
         rank_id = item.second.rank_id_;
       }
       return res;
diff --git a/mindspore/ccsrc/ps/server/collective_ops_impl.cc b/mindspore/ccsrc/ps/server/collective_ops_impl.cc
index 339b17b5191..e4aba56252f 100644
--- a/mindspore/ccsrc/ps/server/collective_ops_impl.cc
+++ b/mindspore/ccsrc/ps/server/collective_ops_impl.cc
@@ -77,7 +77,7 @@ bool CollectiveOpsImpl::RingAllReduce(const void *sendbuff, void *recvbuff, size
 
     std::shared_ptr<std::vector<unsigned char>> recv_str;
     auto recv_req_id = server_node_->CollectiveReceiveAsync(core::NodeRole::SERVER, recv_from_rank, &recv_str);
-    if (!server_node_->CollectiveWait(recv_req_id, 1)) {
+    if (!server_node_->CollectiveWait(recv_req_id)) {
       MS_LOG(ERROR) << "CollectiveWait " << recv_req_id << " failed.";
       return false;
     }
@@ -111,7 +111,7 @@ bool CollectiveOpsImpl::RingAllReduce(const void *sendbuff, void *recvbuff, size
     std::shared_ptr<std::vector<unsigned char>> recv_str;
     auto recv_req_id = server_node_->CollectiveReceiveAsync(core::NodeRole::SERVER, recv_from_rank, &recv_str);
 
-    if (!server_node_->CollectiveWait(recv_req_id, 1)) {
+    if (!server_node_->CollectiveWait(recv_req_id)) {
       MS_LOG(ERROR) << "CollectiveWait " << recv_req_id << " failed.";
       return false;
     }
@@ -144,7 +144,7 @@ bool CollectiveOpsImpl::ReduceBroadcastAllReduce(const void *sendbuff, void *rec
       std::shared_ptr<std::vector<unsigned char>> recv_str;
       MS_LOG(DEBUG) << "Reduce rank 0 receive from rank " << i;
       auto recv_req_id = server_node_->CollectiveReceiveAsync(core::NodeRole::SERVER, i, &recv_str);
-      if (!server_node_->CollectiveWait(recv_req_id, 1)) {
+      if (!server_node_->CollectiveWait(recv_req_id)) {
         MS_LOG(ERROR) << "CollectiveWait " << recv_req_id << " failed.";
         return false;
       }
@@ -156,7 +156,7 @@ bool CollectiveOpsImpl::ReduceBroadcastAllReduce(const void *sendbuff, void *rec
   } else {
     MS_LOG(DEBUG) << "Reduce send data to rank 0 process.";
     auto send_req_id = server_node_->CollectiveSendAsync(core::NodeRole::SERVER, 0, sendbuff, count * sizeof(T));
-    if (!server_node_->Wait(send_req_id, 1)) {
+    if (!server_node_->Wait(send_req_id)) {
       MS_LOG(ERROR) << "CollectiveWait " << send_req_id << " failed.";
       return false;
     }
@@ -169,7 +169,7 @@ bool CollectiveOpsImpl::ReduceBroadcastAllReduce(const void *sendbuff, void *rec
     for (uint32_t i = 1; i < rank_size; i++) {
       MS_LOG(DEBUG) << "Broadcast data to process " << i;
       auto send_req_id = server_node_->CollectiveSendAsync(core::NodeRole::SERVER, i, output_buff, count * sizeof(T));
-      if (!server_node_->Wait(send_req_id, 1)) {
+      if (!server_node_->Wait(send_req_id)) {
         MS_LOG(ERROR) << "CollectiveWait " << send_req_id << " failed.";
         return false;
       }
@@ -178,7 +178,7 @@ bool CollectiveOpsImpl::ReduceBroadcastAllReduce(const void *sendbuff, void *rec
     MS_LOG(DEBUG) << "Broadcast receive from rank 0.";
     std::shared_ptr<std::vector<unsigned char>> recv_str;
     auto recv_req_id = server_node_->CollectiveReceiveAsync(core::NodeRole::SERVER, 0, &recv_str);
-    if (!server_node_->CollectiveWait(recv_req_id, 1)) {
+    if (!server_node_->CollectiveWait(recv_req_id)) {
       MS_LOG(ERROR) << "CollectiveWait " << recv_req_id << " failed.";
       return false;
     }