Remove peer_fd from the event loop only after clearing cl->refs

src/client/messenger.h

@@ -234,6 +234,7 @@ public:
     void parse_config(const json11::Json & config);
     void connect_peer(uint64_t osd_num, json11::Json peer_state);
     void stop_client(int peer_fd, bool force = false, bool force_delete = false);
+    void destroy_client(osd_client_t *cl);
     void outbox_push(osd_op_t *cur_op);
     std::function<void(osd_op_t*)> exec_op;
     std::function<void(osd_num_t)> repeer_pgs;

src/client/msgr_receive.cpp

@@ -78,7 +78,7 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl)
     {
         if (cl->refs <= 0)
         {
-            delete cl;
+            destroy_client(cl);
         }
         return false;
     }

src/client/msgr_send.cpp

@@ -274,7 +274,7 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
     {
         if (cl->refs <= 0)
         {
-            delete cl;
+            destroy_client(cl);
         }
         return;
     }

src/client/msgr_stop.cpp

@@ -85,30 +85,13 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
             osd_peer_fds.erase(osd_it);
         }
     }
+    // Do not remove socket from the event loop as it may have refs > 0 and we want to clear them
 #ifndef __MOCK__
-    // Then remove FD from the eventloop so we don't accidentally read something
-    tfd->set_fd_handler(peer_fd, false, NULL);
     if (cl->connect_timeout_id >= 0)
     {
         tfd->clear_timer(cl->connect_timeout_id);
         cl->connect_timeout_id = -1;
     }
-    for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++)
-    {
-        if (*rit == peer_fd)
-        {
-            read_ready_clients.erase(rit);
-            break;
-        }
-    }
-    for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++)
-    {
-        if (*wit == peer_fd)
-        {
-            write_ready_clients.erase(wit);
-            break;
-        }
-    }
 #endif
     if (cl->in_osd_num && break_pg_locks)
     {
@@ -123,19 +106,42 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete)
         // so do not repeer on it.
         repeer_pgs(cl->osd_num);
     }
-    // Find the item again because it can be invalidated at this point
-    it = clients.find(peer_fd);
-    if (it != clients.end())
-    {
-        clients.erase(it);
-    }
     cl->refs--;
     if (cl->refs <= 0 || force_delete)
     {
-        delete cl;
+        destroy_client(cl);
     }
 }
 
+void osd_messenger_t::destroy_client(osd_client_t *cl)
+{
+#ifndef __MOCK__
+    if (cl->peer_fd >= 0)
+    {
+        // Remove FD from the eventloop
+        tfd->set_fd_handler(cl->peer_fd, false, NULL);
+        for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++)
+        {
+            if (*rit == cl->peer_fd)
+            {
+                read_ready_clients.erase(rit);
+                break;
+            }
+        }
+        for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++)
+        {
+            if (*wit == cl->peer_fd)
+            {
+                write_ready_clients.erase(wit);
+                break;
+            }
+        }
+    }
+#endif
+    clients.erase(cl->peer_fd);
+    delete cl;
+}
+
 osd_client_t::~osd_client_t()
 {
     free(in_buf);