Do not expect second notification when cqe returns error

src/client/messenger.h

@@ -234,7 +234,6 @@ 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)
         {
-            destroy_client(cl);
+            delete 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)
         {
-            destroy_client(cl);
+            delete cl;
         }
         return;
     }

src/client/msgr_stop.cpp

@@ -85,13 +85,30 @@ 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)
     {
@@ -106,42 +123,19 @@ 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)
     {
-        destroy_client(cl);
+        delete 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);

src/util/ringloop.cpp

@@ -120,7 +120,7 @@ void ring_loop_t::loop()
         if (mt)
             mu.lock();
         struct ring_data_t *d = (struct ring_data_t*)cqe->user_data;
-        if (cqe->flags & IORING_CQE_F_MORE)
+        if ((cqe->flags & IORING_CQE_F_MORE) && cqe->res >= 0)
         {
             // There will be a second notification
             if (mt)