msg: ceph_abort() when there are enough accepter errors in msg server

In some extrem cases(we have met one in our production cluster), when Accepter thread break out , new client can not connect to the osd. Because the former heartbeat connections are already connected, other osd can not detect failure then notify monitor to mark the failed osd down.
In the patch, we there are abnormal communication errors ,we just ceph_abort  so that osd can go down fastly and other osds can notify monitor to mark the failed osd down.
Signed-off-by: penglaiyxy@gmail.com <penglaiyxy@gmail.com>
(cherry picked from commit 00e0ab407b2e9659d9121be1217e95c8117c411e)

Conflicts:
	src/common/legacy_config_opts.h
	src/common/options.cc
	src/msg/async/AsyncMessenger.cc: trivial resolution

diff --git a/src/common/legacy_config_opts.h b/src/common/legacy_config_opts.h
index b76e089732dc3bb6ffe44f713d5c207bbe1e0aba..e2bb67fb13f00cf89e357d224647c1253911783a 100644 (file)
--- a/src/common/legacy_config_opts.h
+++ b/src/common/legacy_config_opts.h
@@ -166,6 +166,10 @@ OPTION(ms_async_rdma_roce_ver, OPT_INT)         // 0=RoCEv1, 1=RoCEv2, 2=RoCEv1.
 OPTION(ms_async_rdma_sl, OPT_INT)               // in RoCE, this means PCP
 OPTION(ms_async_rdma_dscp, OPT_INT)            // in RoCE, this means DSCP
 
+// when there are enough accept failures, indicating there are unrecoverable failures,
+// just do ceph_abort() . Here we make it configurable.
+OPTION(ms_max_accept_failures, OPT_INT)
+
 OPTION(ms_dpdk_port_id, OPT_INT)
 SAFE_OPTION(ms_dpdk_coremask, OPT_STR)        // it is modified in unittest so that use SAFE_OPTION to declare 
 OPTION(ms_dpdk_memory_channel, OPT_STR)

diff --git a/src/common/options.cc b/src/common/options.cc
index 6857b8aa0940066a63f4bef4f14298b3b61a8a40..197e44daab953a925e5f52a93b22bb038cb52619 100644 (file)
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -1044,6 +1044,11 @@ std::vector<Option> get_global_options() {
     .set_default(96)
     .set_description(""),
 
+    Option("ms_max_accept_failures", Option::TYPE_INT, Option::LEVEL_ADVANCED)
+    .set_default(4)
+    .set_description("The maximum number of consecutive failed accept() calls before "
+                     "considering the daemon is misconfigured and abort it."),
+
     Option("ms_dpdk_port_id", Option::TYPE_INT, Option::LEVEL_ADVANCED)
     .set_default(0)
     .set_description(""),

diff --git a/src/msg/async/AsyncMessenger.cc b/src/msg/async/AsyncMessenger.cc
index bcb78e67056afa6556e94ab9f8301530ee29f172..11508494efe29db27cbf9783c32c088565adc234 100644 (file)
--- a/src/msg/async/AsyncMessenger.cc
+++ b/src/msg/async/AsyncMessenger.cc
@@ -165,6 +165,8 @@ void Processor::accept()
   opts.nodelay = msgr->cct->_conf->ms_tcp_nodelay;
   opts.rcbuf_size = msgr->cct->_conf->ms_tcp_rcvbuf;
   opts.priority = msgr->get_socket_priority();
+  unsigned accept_error_num = 0;
+
   while (true) {
     entity_addr_t addr;
     ConnectedSocket cli_socket;
@@ -178,6 +180,7 @@ void Processor::accept()
       ldout(msgr->cct, 10) << __func__ << " accepted incoming on sd " << cli_socket.fd() << dendl;
 
       msgr->add_accept(w, std::move(cli_socket), addr);
+      accept_error_num = 0;
       continue;
     } else {
       if (r == -EINTR) {
@@ -187,7 +190,11 @@ void Processor::accept()
       } else if (r == -EMFILE || r == -ENFILE) {
         lderr(msgr->cct) << __func__ << " open file descriptions limit reached sd = " << listen_socket.fd()
                          << " errno " << r << " " << cpp_strerror(r) << dendl;
-        break;
+       if (++accept_error_num > msgr->cct->_conf->ms_max_accept_failures) {
+           lderr(msgr->cct) << "Proccessor accept has encountered enough error numbers, just do ceph_abort()." << dendl;
+           ceph_abort();
+       }
+        continue;
       } else if (r == -ECONNABORTED) {
         ldout(msgr->cct, 0) << __func__ << " it was closed because of rst arrived sd = " << listen_socket.fd()
                             << " errno " << r << " " << cpp_strerror(r) << dendl;
@@ -195,7 +202,11 @@ void Processor::accept()
       } else {
         lderr(msgr->cct) << __func__ << " no incoming connection?"
                          << " errno " << r << " " << cpp_strerror(r) << dendl;
-        break;
+       if (++accept_error_num > msgr->cct->_conf->ms_max_accept_failures) {
+         lderr(msgr->cct) << "Proccessor accept has encountered enough error numbers, just do ceph_abort()." << dendl;
+         ceph_abort();
+       }
+       continue;
       }
     }
   }

diff --git a/src/msg/simple/Accepter.cc b/src/msg/simple/Accepter.cc
index a95e046c7ba0870cea53464abe5c66791cd04cd6..db9e8795e6065497f8d1eedf9b1f1beb386d5843 100644 (file)
--- a/src/msg/simple/Accepter.cc
+++ b/src/msg/simple/Accepter.cc
@@ -313,7 +313,7 @@ void *Accepter::entry()
       }
       ldout(msgr->cct,1) << __func__ << " poll got error"  
                          << " errno " << errno << " " << cpp_strerror(errno) << dendl;
-      break;
+      ceph_abort();
     }
     ldout(msgr->cct,10) << __func__ << " poll returned oke: " << r << dendl;
     ldout(msgr->cct,20) << __func__ <<  " pfd.revents[0]=" << pfd[0].revents << dendl;
@@ -322,7 +322,7 @@ void *Accepter::entry()
     if (pfd[0].revents & (POLLERR | POLLNVAL | POLLHUP)) {
       ldout(msgr->cct,1) << __func__ << " poll got errors in revents "  
                         <<  pfd[0].revents << dendl;
-      break;
+      ceph_abort();
     }
     if (pfd[1].revents & (POLLIN | POLLERR | POLLNVAL | POLLHUP)) {
       // We got "signaled" to exit the poll
@@ -354,8 +354,10 @@ void *Accepter::entry()
     } else {
       ldout(msgr->cct,0) << __func__ << " no incoming connection?  sd = " << sd
              << " errno " << errno << " " << cpp_strerror(errno) << dendl;
-      if (++errors > 4)
-       break;
+      if (++errors > msgr->cct->_conf->ms_max_accept_failures) {
+        lderr(msgr->cct) << "accetper has encoutered enough errors, just do ceph_abort()." << dendl;
+        ceph_abort();
+      }
     }
   }