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 78f19b8e2f6563f43f7c10b73c45127442b78396..dbca8e74dc90d0d4a9007bf27639c8550871879a 100644 (file)
--- 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 43363371bc35d05eb342a4113881ee9936e8dc07..7373bada4b4ef8e12957bb6bca16f9084ec22208 100644 (file)
--- 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 a176fc2c808ac9eb8d465310781daf4eacc20b98..cf5d18b0c02317b1d1d92019c5eb3267464428ca 100644 (file)
--- 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;