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 : Resolved for ms_max_accept_failures
	src/common/options.cc : Resolved for ms_max_accept_failures
	src/msg/async/AsyncMessenger.cc : Resolved in accept
	src/msg/simple/Accepter.cc : Resolved in entry

diff --git a/src/common/legacy_config_opts.h b/src/common/legacy_config_opts.h
index 9c97aae8218152c2020d24f3e3a46ff1a914bc5d..a51870ef64289ba12163b40b4c0a02ec4792995f 100644 (file)
--- a/src/common/legacy_config_opts.h
+++ b/src/common/legacy_config_opts.h
@@ -169,6 +169,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 b0c7ccc252f336feca5f94c71fc7359af795b4bd..ff3bb1a1be193d7c4e66310a83f106cb24982ce8 100644 (file)
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -812,6 +812,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 1913e8f4c6eea8bb1f530a248094a21475e24925..abe0b474170ec42ca4139a9d6c892e27777ab43a 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;
@@ -185,7 +187,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;
@@ -193,7 +199,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 a2c26ef19718d12a328db7ae744d0c75331e64ed..f5e85336df1963caea585a75a542c3094a090d7d 100644 (file)
--- a/src/msg/simple/Accepter.cc
+++ b/src/msg/simple/Accepter.cc
@@ -293,7 +293,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;
@@ -302,7 +302,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
@@ -329,8 +329,10 @@ void *Accepter::entry()
       int e = errno;
       ldout(msgr->cct,0) << __func__ << " no incoming connection?  sd = " << sd
              << " errno " << e << " " << cpp_strerror(e) << 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();
+      }
     }
   }