crimson/net: don't return null from Connection::read_message()

SocketConnection::read_message() now loops until it has a message with
valid sequence number. this means SocketMessenger::dispatch() doesn't
have to handle the null message case

Signed-off-by: Casey Bodley <cbodley@redhat.com>

diff --git a/src/crimson/net/SocketConnection.cc b/src/crimson/net/SocketConnection.cc
index 2ad467b55d0b820d1f18c27b652a16da9ec60f83..9595cca9e1a050f372fd4ad318360b2d7ee58099 100644 (file)
--- a/src/crimson/net/SocketConnection.cc
+++ b/src/crimson/net/SocketConnection.cc
@@ -204,7 +204,7 @@ seastar::future<> SocketConnection::maybe_throttle()
   return policy.throttler_bytes->get(to_read);
 }
 
-seastar::future<MessageRef> SocketConnection::read_message()
+seastar::future<MessageRef> SocketConnection::do_read_message()
 {
   return on_message.get_future()
     .then([this] {
@@ -239,17 +239,26 @@ seastar::future<MessageRef> SocketConnection::read_message()
       ::decode(m.footer, p);
       auto msg = ::decode_message(nullptr, 0, m.header, m.footer,
                                   m.front, m.middle, m.data, nullptr);
+      // TODO: set time stamps
+      msg->set_byte_throttler(policy.throttler_bytes);
       constexpr bool add_ref = false; // Message starts with 1 ref
       return MessageRef{msg, add_ref};
-    }).then([this] (MessageRef msg) {
-      if (msg) {
-        // TODO: set time stamps
-        msg->set_byte_throttler(policy.throttler_bytes);
-        if (!update_rx_seq(msg->get_seq())) {
-          msg.reset();
-        }
-      }
-      return msg;
+    });
+}
+
+seastar::future<MessageRef> SocketConnection::read_message()
+{
+  namespace stdx = std::experimental;
+
+  return seastar::repeat_until_value([this] {
+      return do_read_message()
+        .then([this] (MessageRef msg) -> stdx::optional<MessageRef> {
+          if (!update_rx_seq(msg->get_seq())) {
+            // skip this request and read the next
+            return stdx::nullopt;
+          }
+          return msg;
+        });
     });
 }
 

diff --git a/src/crimson/net/SocketConnection.h b/src/crimson/net/SocketConnection.h
index b034685eb8443dc2a01db4a8a5ff40e9f5ddf08f..974c082ddcd2626eeed75b406b7b91239e413baa 100644 (file)
--- a/src/crimson/net/SocketConnection.h
+++ b/src/crimson/net/SocketConnection.h
@@ -123,6 +123,8 @@ class SocketConnection : public Connection {
   ///          false otherwise.
   bool update_rx_seq(seq_num_t seq);
 
+  seastar::future<MessageRef> do_read_message();
+
   std::unique_ptr<AuthSessionHandler> session_security;
 
   // messages to be resent after connection gets reset