From: Vicente Cheng <vicente_cheng@bigtera.com>
Date: Fri, 10 Dec 2021 06:49:55 +0000 (+0000)
Subject: msg/async: fix outgoing_bl overflow and reset state_offset
X-Git-Tag: v18.0.0~1400^2
X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=refs%2Fpull%2F44289%2Fhead;p=ceph.git

msg/async: fix outgoing_bl overflow and reset state_offset

    - we should reset state_offset when read done.

    - check outgoing_bl before we try to write a message.

    In some environments, network would temporily block and return EAGAIN.
    For async msgr, we would callback the write event directly, but that still
    increase the outgoing_bl.

    Think about this case, the sender is in congestion or network driver
    has some problems. The data appended to outgoing_bl and outgoing_bl
    is not consumed up-to-date immediately.
    That size of outgoing_bl will increase with time then overflow.
    The wrong outgoing_bl would cause some problems so we need to wait
    for outgoing_bl before we appended another message.

Signed-off-by: Vicente Cheng <vicente_cheng@bigtera.com>
---

diff --git a/src/msg/async/AsyncConnection.cc b/src/msg/async/AsyncConnection.cc
index 78f19b8e2f656..dbca8e74dc90d 100644
--- a/src/msg/async/AsyncConnection.cc
+++ b/src/msg/async/AsyncConnection.cc
@@ -226,6 +226,7 @@ ssize_t AsyncConnection::read_until(unsigned len, char *p)
                                << " left is " << left << " buffer still has "
                                << recv_end - recv_start << dendl;
     if (left == 0) {
+      state_offset = 0;
       return 0;
     }
     state_offset += to_read;
diff --git a/src/msg/async/ProtocolV1.cc b/src/msg/async/ProtocolV1.cc
index 43363371bc35d..7373bada4b4ef 100644
--- a/src/msg/async/ProtocolV1.cc
+++ b/src/msg/async/ProtocolV1.cc
@@ -315,6 +315,13 @@ void ProtocolV1::write_event() {
     auto start = ceph::mono_clock::now();
     bool more;
     do {
+      if (connection->is_queued()) {
+	if (r = connection->_try_send(); r!= 0) {
+	  // either fails to send or not all queued buffer is sent
+	  break;
+	}
+      }
+
       ceph::buffer::list data;
       Message *m = _get_next_outgoing(&data);
       if (!m) {
diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index a176fc2c808ac..cf5d18b0c0231 100644
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -653,6 +653,13 @@ void ProtocolV2::write_event() {
     auto start = ceph::mono_clock::now();
     bool more;
     do {
+      if (connection->is_queued()) {
+	if (r = connection->_try_send(); r!= 0) {
+	  // either fails to send or not all queued buffer is sent
+	  break;
+	}
+      }
+
       const auto out_entry = _get_next_outgoing();
       if (!out_entry.m) {
         break;