skip processing command while associated connection has been closed (#3111)

Co-authored-by: wangshaoyi <[email protected]>

Author: wangshao1

src/net/include/net_conn.h

@@ -9,6 +9,7 @@
 #include <sys/time.h>
 #include <sstream>
 #include <string>
+#include <atomic>
 
 #ifdef __ENABLE_SSL
 #  include <openssl/err.h>
@@ -70,7 +71,7 @@ class NetConn : public std::enable_shared_from_this<NetConn>, public pstd::nonco
   std::string name() { return name_; }
   void set_name(std::string name) { name_ = std::move(name); }
 
-  bool IsClose() { return close_; }
+  bool IsClose() { return close_.load(); }
   void SetClose(bool close);
 
   void set_last_interaction(const struct timeval& now) { last_interaction_ = now; }
@@ -100,7 +101,7 @@ class NetConn : public std::enable_shared_from_this<NetConn>, public pstd::nonco
   std::string ip_port_;
   bool is_reply_ = false;
   bool is_writable_ = false;
-  bool close_ = false;
+  std::atomic_bool close_ = false;
   struct timeval last_interaction_;
   int flags_ = 0;
   std::string name_;

src/net/src/net_conn.cc

@@ -23,6 +23,7 @@ NetConn::NetConn(const int fd, std::string  ip_port, Thread* thread, NetMultiple
 #ifdef __ENABLE_SSL
       ssl_(nullptr),
 #endif
+      close_(false),
       thread_(thread),
       net_multiplexer_(net_mpx) {
   gettimeofday(&last_interaction_, nullptr);
@@ -36,7 +37,7 @@ NetConn::~NetConn() {
 #endif
 
 void NetConn::SetClose(bool close) {
-  close_ = close;
+  close_.store(close);
 }
 
 bool NetConn::SetNonblock() {

src/net/src/worker_thread.cc

@@ -144,6 +144,7 @@ void* WorkerThread::ThreadMain() {
               } else if (ti.notify_type() == kNotiEpollin) {
                 net_multiplexer_->NetModEvent(ti.fd(), 0, kReadable);
               } else if (ti.notify_type() == kNotiEpolloutAndEpollin) {
+                // TODO(wangshaoyi): conn_fd might be closed already, so epoll_ctl might return non-zero
                 net_multiplexer_->NetModEvent(ti.fd(), 0, kReadable | kWritable);
               } else if (ti.notify_type() == kNotiWait) {
                 // do not register events
@@ -209,6 +210,7 @@ void* WorkerThread::ThreadMain() {
           // eg. in_conn are being transferred to net_pubsub
           // while peer client closing this connection
           CloseFd(in_conn);
+          in_conn->SetClose(true);
           in_conn = nullptr;
           {
             std::lock_guard lock(rwlock_);

src/pika_client_conn.cc

@@ -338,6 +338,10 @@ void PikaClientConn::DoBackgroundTask(void* arg) {
   std::unique_ptr<BgTaskArg> bg_arg(static_cast<BgTaskArg*>(arg));
   std::shared_ptr<PikaClientConn> conn_ptr = bg_arg->conn_ptr;
   conn_ptr->time_stat_->dequeue_ts_ = pstd::NowMicros();
+  if (conn_ptr->IsClose()) {
+    LOG(INFO) << "conn " << conn_ptr->ip_port() << " has already been closed, skip processing command";
+    return;
+  }
   if (bg_arg->redis_cmds.empty()) {
     conn_ptr->NotifyEpoll(false);
     return;