1 handle lock when close all pubsub conns

2 rename some funs and varibles
OpenAtomFoundation · Aug 11, 2024 · b45a402 · b45a402
1 parent a2d7949
commit b45a402
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 16 deletions.
diff --git a/src/net/include/net_pubsub.h b/src/net/include/net_pubsub.h
@@ -77,7 +77,7 @@ class PubSubThread : public Thread {
   bool IsReady(int fd);
   int ClientPubSubChannelSize(const std::shared_ptr<NetConn>& conn);
   int ClientPubSubChannelPatternSize(const std::shared_ptr<NetConn>& conn);
-  void NotifyToCloseAllConns();
+  void NotifyCloseAllConns();
 
  private:
   void RemoveConn(const std::shared_ptr<NetConn>& conn);

diff --git a/src/net/src/net_pubsub.cc b/src/net/src/net_pubsub.cc
@@ -151,26 +151,32 @@ void PubSubThread::RemoveConn(const std::shared_ptr<NetConn>& conn) {
 }
 
 void PubSubThread::CloseConn(const std::shared_ptr<NetConn>& conn) {
-  CloseFd(conn);
   net_multiplexer_->NetDelEvent(conn->fd(), 0);
+  CloseFd(conn);
   {
     std::lock_guard l(rwlock_);
     conns_.erase(conn->fd());
   }
 }
 
 void PubSubThread::CloseAllConns() {
-  std::lock_guard l(rwlock_);
-
-  pubsub_channel_.clear();
-  pubsub_pattern_.clear();
-
-  for (auto& pair : conns_) {
-    net_multiplexer_->NetDelEvent(pair.second->conn->fd(), 0);
-    CloseFd(pair.second->conn);
+  {
+    std::lock_guard l(channel_mutex_);
+    pubsub_channel_.clear();
+  }
+  {
+    std::lock_guard l(pattern_mutex_);
+    pubsub_pattern_.clear();
+  }
+  {
+    std::lock_guard l(rwlock_);
+    for (auto& pair : conns_) {
+      net_multiplexer_->NetDelEvent(pair.second->conn->fd(), 0);
+      CloseFd(pair.second->conn);
+    }
+    std::map<int, std::shared_ptr<ConnHandle>> tmp;
+    conns_.swap(tmp);
   }
-  std::map<int, std::shared_ptr<ConnHandle>> empty_conns;
-  conns_.swap(empty_conns);
 }
 
 int PubSubThread::Publish(const std::string& channel, const std::string& msg) {
@@ -605,7 +611,7 @@ void PubSubThread::Cleanup() {
   }
   conns_.clear();
 }
-void PubSubThread::NotifyToCloseAllConns() {
+void PubSubThread::NotifyCloseAllConns() {
   close_all_conn_sig_.store(true);
 }
 };  // namespace net
diff --git a/src/pika_server.cc b/src/pika_server.cc
@@ -858,11 +858,10 @@ void PikaServer::KeyScanTaskSchedule(net::TaskFunc func, void* arg) {
 
 void PikaServer::ClientKillAll() {
   pika_dispatch_thread_->ClientKillAll();
-  pika_pubsub_thread_->NotifyToCloseAllConns();
+  pika_pubsub_thread_->NotifyCloseAllConns();
 }
 
-void PikaServer::ClientKillPubSub() {
-  pika_pubsub_thread_->NotifyToCloseAllConns();
+void PikaServer::ClientKillPubSub() { pika_pubsub_thread_->NotifyCloseAllConns();
 }
 
 void PikaServer::ClientKillAllNormal() {