Merge pull request #29882 from cyx1231st/wip-seastar-msgr-test-lossless-crimson

crimson/net: bug fixes from v2 failover tests

Reviewed-by: Kefu Chai <kchai@redhat.com>

src/crimson/net/Connection.h

@@ -57,7 +57,11 @@ class Connection : public seastar::enable_shared_from_this<Connection> {
   bool peer_is_client() const { return peer_name.is_client(); }
 
   /// true if the handshake has completed and no errors have been encountered
-  virtual seastar::future<bool> is_connected() = 0;
+  virtual bool is_connected() const = 0;
+
+#ifdef UNIT_TESTS_BUILT
+  virtual bool is_closed() const = 0;
+#endif
 
   /// send a message over a connection that has completed its handshake
   virtual seastar::future<> send(MessageRef msg) = 0;

src/crimson/net/Protocol.cc

@@ -122,6 +122,7 @@ void Protocol::requeue_sent()
                     std::make_move_iterator(conn.sent.begin()),
                     std::make_move_iterator(conn.sent.end()));
   conn.sent.clear();
+  write_event();
 }
 
 void Protocol::requeue_up_to(seq_num_t seq)
@@ -169,98 +170,93 @@ void Protocol::ack_writes(seq_num_t seq)
   }
 }
 
-seastar::future<stop_t> Protocol::do_write_dispatch_sweep()
+seastar::future<> Protocol::do_write_dispatch_sweep()
 {
-  switch (write_state) {
-   case write_state_t::open: {
-    size_t num_msgs = conn.out_q.size();
-    // we must have something to write...
-    ceph_assert(is_queued());
-    assert(!open_write);
-    open_write = true;
+  return seastar::repeat([this] {
+    switch (write_state) {
+     case write_state_t::open: {
+      size_t num_msgs = conn.out_q.size();
+      // we must have something to write...
+      ceph_assert(is_queued());
+      assert(!open_write);
+      open_write = true;
 
-    conn.pending_q.clear();
-    conn.pending_q.swap(conn.out_q);
-    if (!conn.policy.lossy) {
-      conn.sent.insert(conn.sent.end(),
-                       conn.pending_q.begin(),
-                       conn.pending_q.end());
+      conn.pending_q.clear();
+      conn.pending_q.swap(conn.out_q);
+      if (!conn.policy.lossy) {
+        conn.sent.insert(conn.sent.end(),
+                         conn.pending_q.begin(),
+                         conn.pending_q.end());
+      }
+      auto acked = ack_left;
+      assert(acked == 0 || conn.in_seq > 0);
+      // sweep all pending writes with the concrete Protocol
+      return socket->write(do_sweep_messages(
+          conn.pending_q, num_msgs, need_keepalive, keepalive_ack, acked > 0)
+      ).then([this, prv_keepalive_ack=keepalive_ack, acked] {
+        need_keepalive = false;
+        if (keepalive_ack == prv_keepalive_ack) {
+          keepalive_ack = std::nullopt;
+        }
+        assert(ack_left >= acked);
+        ack_left -= acked;
+        if (!is_queued()) {
+          // good, we have nothing pending to send now.
+          return socket->flush().then([this] {
+            if (!is_queued()) {
+              // still nothing pending to send after flush,
+              // the dispatching can ONLY stop now
+              ceph_assert(write_dispatching);
+              write_dispatching = false;
+              assert(open_write);
+              open_write = false;
+              return seastar::make_ready_future<stop_t>(stop_t::yes);
+            } else {
+              // something is pending to send during flushing
+              assert(open_write);
+              open_write = false;
+              return seastar::make_ready_future<stop_t>(stop_t::no);
+            }
+          });
+        } else {
+          // messages were enqueued during socket write
+          assert(open_write);
+          open_write = false;
+          return seastar::make_ready_future<stop_t>(stop_t::no);
+        }
+      });
+     }
+     case write_state_t::delay:
+      // delay dispatching writes until open
+      return state_changed.get_shared_future()
+      .then([] { return stop_t::no; });
+     case write_state_t::drop:
+      ceph_assert(write_dispatching);
+      write_dispatching = false;
+      return seastar::make_ready_future<stop_t>(stop_t::yes);
+     default:
+      ceph_assert(false);
     }
-    auto acked = ack_left;
-    assert(acked == 0 || conn.in_seq > 0);
-    // sweep all pending writes with the concrete Protocol
-    return socket->write(do_sweep_messages(
-        conn.pending_q, num_msgs, need_keepalive, keepalive_ack, acked > 0)
-    ).then([this, prv_keepalive_ack=keepalive_ack, acked] {
-      need_keepalive = false;
-      if (keepalive_ack == prv_keepalive_ack) {
-        keepalive_ack = std::nullopt;
-      }
-      assert(ack_left >= acked);
-      ack_left -= acked;
-      if (!is_queued()) {
-        // good, we have nothing pending to send now.
-        return socket->flush().then([this] {
-          if (!is_queued()) {
-            // still nothing pending to send after flush,
-            // the dispatching can ONLY stop now
-            ceph_assert(write_dispatching);
-            write_dispatching = false;
-            assert(open_write);
-            open_write = false;
-            return seastar::make_ready_future<stop_t>(stop_t::yes);
-          } else {
-            // something is pending to send during flushing
-            assert(open_write);
-            open_write = false;
-            return seastar::make_ready_future<stop_t>(stop_t::no);
-          }
-        });
-      } else {
-        // messages were enqueued during socket write
-        assert(open_write);
-        open_write = false;
-        return seastar::make_ready_future<stop_t>(stop_t::no);
-      }
-    }).handle_exception_type([this] (const std::system_error& e) {
-      if (e.code() != error::broken_pipe &&
-          e.code() != error::connection_reset) {
-        logger().error("{} do_write_dispatch_sweep(): unexpected error {}",
-                       conn, e);
-        ceph_abort();
-      }
-
-      logger().debug("{} do_write_dispatch_sweep() fault: {}", conn, e);
-      assert(open_write);
-      open_write = false;
-      if (exit_open) {
-        exit_open->set_value();
-        exit_open = std::nullopt;
-      }
-      socket->shutdown();
-      if (write_state == write_state_t::open) {
-        write_state = write_state_t::delay;
-      }
-      return seastar::make_ready_future<stop_t>(stop_t::no);
-    }).handle_exception([this] (std::exception_ptr eptr) {
-      logger().error("{} do_write_dispatch_sweep(): unexpected exception {}",
-                     conn, eptr);
+  }).handle_exception_type([this] (const std::system_error& e) {
+    if (e.code() != error::broken_pipe &&
+        e.code() != error::connection_reset) {
+      logger().error("{} do_write_dispatch_sweep(): unexpected error {}",
+                     conn, e);
       ceph_abort();
-      return seastar::make_ready_future<stop_t>(stop_t::no);
-    });
-   }
-   case write_state_t::delay: {
-    // delay dispatching writes until open
-    return state_changed.get_shared_future()
-    .then([] { return stop_t::no; });
-   }
-   case write_state_t::drop:
-    ceph_assert(write_dispatching);
-    write_dispatching = false;
-    return seastar::make_ready_future<stop_t>(stop_t::yes);
-   default:
-    ceph_assert(false);
-  }
+    }
+    logger().debug("{} do_write_dispatch_sweep() fault: {}", conn, e);
+    assert(open_write);
+    open_write = false;
+    if (exit_open) {
+      exit_open->set_value();
+      exit_open = std::nullopt;
+    }
+    socket->shutdown();
+    if (write_state == write_state_t::open) {
+      write_state = write_state_t::delay;
+    }
+    return do_write_dispatch_sweep();
+  });
 }
 
 void Protocol::write_event()
@@ -276,13 +272,17 @@ void Protocol::write_event()
      [[fallthrough]];
    case write_state_t::delay:
     seastar::with_gate(pending_dispatch, [this] {
-      return seastar::repeat([this] {
-        return do_write_dispatch_sweep();
+      return do_write_dispatch_sweep(
+      ).handle_exception([this] (std::exception_ptr eptr) {
+        logger().error("{} do_write_dispatch_sweep(): unexpected exception {}",
+                       conn, eptr);
+        ceph_abort();
       });
     });
     return;
    case write_state_t::drop:
     write_dispatching = false;
+    return;
    default:
     ceph_assert(false);
   }

src/crimson/net/Protocol.h

@@ -24,6 +24,8 @@ class Protocol {
 
   bool is_connected() const;
 
+  bool is_closed() const { return closed; }
+
   // Reentrant closing
   seastar::future<> close();
 
@@ -138,7 +140,7 @@ class Protocol {
   // stopped or failed.
   std::optional<seastar::shared_promise<>> exit_open;
 
-  seastar::future<stop_t> do_write_dispatch_sweep();
+  seastar::future<> do_write_dispatch_sweep();
   void write_event();
 };
 

src/crimson/net/ProtocolV2.cc

@@ -157,10 +157,12 @@ void ProtocolV2::start_connect(const entity_addr_t& _peer_addr,
   conn.target_addr = _peer_addr;
   conn.set_peer_type(_peer_type);
   conn.policy = messenger.get_policy(_peer_type);
-  logger().info("{} ProtocolV2::start_connect(): peer_addr={}, peer_type={},"
+  client_cookie = generate_client_cookie();
+  logger().info("{} ProtocolV2::start_connect(): peer_addr={}, peer_type={}, cc={}"
                 " policy(lossy={}, server={}, standby={}, resetcheck={})",
-                conn, _peer_addr, ceph_entity_type_name(_peer_type), conn.policy.lossy,
-                conn.policy.server, conn.policy.standby, conn.policy.resetcheck);
+                conn, _peer_addr, ceph_entity_type_name(_peer_type), client_cookie,
+                conn.policy.lossy, conn.policy.server,
+                conn.policy.standby, conn.policy.resetcheck);
   messenger.register_conn(
     seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
   execute_connecting();
@@ -404,7 +406,9 @@ void ProtocolV2::fault(bool backoff)
   if (conn.policy.lossy) {
     dispatch_reset();
     close();
-  } else if (conn.policy.server || (conn.policy.standby && !is_queued())) {
+  } else if (conn.policy.server ||
+             (conn.policy.standby &&
+              (!is_queued() && conn.sent.empty()))) {
     execute_standby();
   } else if (backoff) {
     execute_wait(false);
@@ -417,11 +421,10 @@ void ProtocolV2::dispatch_reset()
 {
   seastar::with_gate(pending_dispatch, [this] {
     return dispatcher.ms_handle_reset(
-        seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()))
-    .handle_exception([this] (std::exception_ptr eptr) {
-      logger().error("{} ms_handle_reset caught exception: {}", conn, eptr);
-      ceph_abort("unexpected exception from ms_handle_reset()");
-    });
+        seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
+  }).handle_exception([this] (std::exception_ptr eptr) {
+    logger().error("{} ms_handle_reset caught exception: {}", conn, eptr);
+    ceph_abort("unexpected exception from ms_handle_reset()");
   });
 }
 
@@ -436,11 +439,10 @@ void ProtocolV2::reset_session(bool full)
     reset_write();
     seastar::with_gate(pending_dispatch, [this] {
       return dispatcher.ms_handle_remote_reset(
-          seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()))
-      .handle_exception([this] (std::exception_ptr eptr) {
-        logger().error("{} ms_handle_remote_reset caught exception: {}", conn, eptr);
-        ceph_abort("unexpected exception from ms_handle_remote_reset()");
-      });
+          seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
+    }).handle_exception([this] (std::exception_ptr eptr) {
+      logger().error("{} ms_handle_remote_reset caught exception: {}", conn, eptr);
+      ceph_abort("unexpected exception from ms_handle_remote_reset()");
     });
   }
 }
@@ -759,14 +761,7 @@ ProtocolV2::client_connect()
             server_cookie = 0;
           }
 
-          return dispatcher.ms_handle_connect(
-              seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()))
-          .handle_exception([this] (std::exception_ptr eptr) {
-            logger().error("{} ms_handle_connect caught exception: {}", conn, eptr);
-            ceph_abort("unexpected exception from ms_handle_connect()");
-          });
-        }).then([this] {
-          return next_step_t::ready;
+          return seastar::make_ready_future<next_step_t>(next_step_t::ready);
         });
       default: {
         unexpected_tag(tag, conn, "post_client_connect");
@@ -834,15 +829,7 @@ ProtocolV2::client_reconnect()
           logger().debug("{} GOT ReconnectOkFrame: msg_seq={}",
                          conn, reconnect_ok.msg_seq());
           requeue_up_to(reconnect_ok.msg_seq());
-          return dispatcher.ms_handle_connect(
-              seastar::static_pointer_cast<SocketConnection>(
-                conn.shared_from_this()))
-          .handle_exception([this] (std::exception_ptr eptr) {
-            logger().error("{} ms_handle_connect caught exception: {}", conn, eptr);
-            ceph_abort("unexpected exception from ms_handle_connect()");
-          });
-        }).then([this] {
-          return next_step_t::ready;
+          return seastar::make_ready_future<next_step_t>(next_step_t::ready);
         });
       default: {
         unexpected_tag(tag, conn, "post_client_reconnect");
@@ -863,17 +850,15 @@ void ProtocolV2::execute_connecting()
       conn.set_ephemeral_port(0, SocketConnection::side_t::none);
       return messenger.get_global_seq().then([this] (auto gs) {
           global_seq = gs;
+          assert(client_cookie != 0);
           if (!conn.policy.lossy && server_cookie != 0) {
-            assert(client_cookie != 0);
             ++connect_seq;
             logger().debug("{} UPDATE: gs={}, cs={} for reconnect",
                            conn, global_seq, connect_seq);
-          } else {
+          } else { // conn.policy.lossy || server_cookie == 0
             assert(connect_seq == 0);
             assert(server_cookie == 0);
-            client_cookie = generate_client_cookie();
-            logger().debug("{} UPDATE: gs={}, cc={} for connect",
-                           conn, global_seq, client_cookie);
+            logger().debug("{} UPDATE: gs={} for connect", conn, global_seq);
           }
 
           return wait_write_exit();
@@ -937,6 +922,13 @@ void ProtocolV2::execute_connecting()
         }).then([this] (next_step_t next) {
           switch (next) {
            case next_step_t::ready: {
+            seastar::with_gate(pending_dispatch, [this] {
+              return dispatcher.ms_handle_connect(
+                  seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
+            }).handle_exception([this] (std::exception_ptr eptr) {
+              logger().error("{} ms_handle_connect caught exception: {}", conn, eptr);
+              ceph_abort("unexpected exception from ms_handle_connect()");
+            });
             logger().info("{} connected: gs={}, pgs={}, cs={},"
                           " client_cookie={}, server_cookie={}, in_seq={}, out_seq={}",
                           conn, global_seq, peer_global_seq, connect_seq,
@@ -962,7 +954,9 @@ void ProtocolV2::execute_connecting()
             return;
           }
 
-          if (conn.policy.server || (conn.policy.standby && !is_queued())) {
+          if (conn.policy.server ||
+              (conn.policy.standby &&
+               (!is_queued() && conn.sent.empty()))) {
             execute_standby();
           } else {
             execute_wait(false);
@@ -1120,19 +1114,16 @@ ProtocolV2::handle_existing_connection(SocketConnectionRef existing_conn)
                  existing_proto->client_cookie,
                  existing_proto->server_cookie);
 
-  if (existing_proto->state == state_t::CLOSING) {
-    logger().warn("{} existing connection {} already closed.", conn, *existing_conn);
-    return send_server_ident();
-  }
-
   if (existing_proto->state == state_t::REPLACING) {
-    logger().warn("{} racing replace happened while replacing existing connection {}",
+    logger().warn("{} server_connect: racing replace happened while "
+                  " replacing existing connection {}, send wait.",
                   conn, *existing_conn);
     return send_wait();
   }
 
   if (existing_proto->peer_global_seq > peer_global_seq) {
-    logger().warn("{} this is a stale connection, because peer_global_seq({})"
+    logger().warn("{} server_connect:"
+                  " this is a stale connection, because peer_global_seq({})"
                   "< existing->peer_global_seq({}), close this connection"
                   " in favor of existing connection {}",
                   conn, peer_global_seq,
@@ -1142,7 +1133,8 @@ ProtocolV2::handle_existing_connection(SocketConnectionRef existing_conn)
 
   if (existing_conn->policy.lossy) {
     // existing connection can be thrown out in favor of this one
-    logger().warn("{} existing connection {} is a lossy channel. Close existing in favor of"
+    logger().warn("{} server_connect:"
+                  " existing connection {} is a lossy channel. Close existing in favor of"
                   " this connection", conn, *existing_conn);
     existing_proto->dispatch_reset();
     existing_proto->close();
@@ -1154,14 +1146,20 @@ ProtocolV2::handle_existing_connection(SocketConnectionRef existing_conn)
       // Found previous session
       // peer has reset and we're going to reuse the existing connection
       // by replacing the socket
-      logger().warn("{} found previous session with existing {}, peer must have reset",
-                    conn, *existing_conn);
+      logger().warn("{} server_connect:"
+                    " found new session (cs={})"
+                    " when existing {} is with stale session (cs={}, ss={}),"
+                    " peer must have reset",
+                    conn, client_cookie,
+                    *existing_conn, existing_proto->client_cookie,
+                    existing_proto->server_cookie);
       return reuse_connection(existing_proto, conn.policy.resetcheck);
     } else {
       // session establishment interrupted between client_ident and server_ident,
       // continuing...
-      logger().warn("{} found previous session with existing {}, continuing session establishment",
-                    conn, *existing_conn);
+      logger().warn("{} server_connect: found client session with existing {}"
+                    " matched (cs={}, ss={}), continuing session establishment",
+                    conn, *existing_conn, client_cookie, existing_proto->server_cookie);
       return reuse_connection(existing_proto);
     }
   } else {
@@ -1170,18 +1168,22 @@ ProtocolV2::handle_existing_connection(SocketConnectionRef existing_conn)
     if (existing_proto->client_cookie != client_cookie) {
       if (conn.peer_addr < messenger.get_myaddr() || existing_conn->policy.server) {
         // this connection wins
-        logger().warn("{} connection race detected and win, reusing existing {}",
-                      conn, *existing_conn);
+        logger().warn("{} server_connect: connection race detected (cs={}, e_cs={}, ss=0)"
+                      " and win, reusing existing {}",
+                      conn, client_cookie, existing_proto->client_cookie, *existing_conn);
         return reuse_connection(existing_proto);
       } else {
         // the existing connection wins
-        logger().warn("{} connection race detected and lose to existing {}",
-                      conn, *existing_conn);
+        logger().warn("{} server_connect: connection race detected (cs={}, e_cs={}, ss=0)"
+                      " and lose to existing {}, ask client to wait",
+                      conn, client_cookie, existing_proto->client_cookie, *existing_conn);
         existing_conn->keepalive();
         return send_wait();
       }
     } else {
-      logger().warn("{} found previous client session with existing {}, continuing session establishment");
+      logger().warn("{} server_connect: found client session with existing {}"
+                    " matched (cs={}, ss={}), continuing session establishment",
+                    conn, *existing_conn, client_cookie, existing_proto->server_cookie);
       return reuse_connection(existing_proto);
     }
   }
@@ -1272,6 +1274,8 @@ ProtocolV2::server_connect()
       }
     }
 
+    // TODO: atomically register & unaccept the connecton with lookup_conn()
+
     // if everything is OK reply with server identification
     return send_server_ident();
   });
@@ -1336,10 +1340,20 @@ ProtocolV2::server_reconnect()
 
     // can peer_addrs be changed on-the-fly?
     // TODO: change peer_addr to entity_addrvec_t
-    if (conn.peer_addr != reconnect.addrs().front()) {
+    entity_addr_t paddr = reconnect.addrs().front();
+    if (paddr.is_msgr2() || paddr.is_any()) {
+      // good
+    } else {
+      logger().warn("{} peer's address {} is not v2", conn, paddr);
+      throw std::system_error(
+          make_error_code(ceph::net::error::bad_peer_address));
+    }
+    if (conn.peer_addr == entity_addr_t()) {
+      conn.peer_addr = paddr;
+    } else if (conn.peer_addr != paddr) {
       logger().error("{} peer identifies as {}, while conn.peer_addr={},"
                      " reconnect failed",
-                     conn, reconnect.addrs().front(), conn.peer_addr);
+                     conn, paddr, conn.peer_addr);
       throw std::system_error(
           make_error_code(ceph::net::error::bad_peer_address));
     }
@@ -1350,8 +1364,8 @@ ProtocolV2::server_reconnect()
     if (!existing_conn) {
       // there is no existing connection therefore cannot reconnect to previous
       // session
-      logger().warn("{} server_reconnect: no existing connection,"
-                    " reseting client", conn);
+      logger().warn("{} server_reconnect: no existing connection from address {},"
+                    " reseting client", conn, conn.peer_addr);
       return send_reset(true);
     }
 
@@ -1401,15 +1415,15 @@ ProtocolV2::server_reconnect()
       //   - connection fault
       //   - b reconnects to a with cookie X, connect_seq=1
       //   - a has cookie==0
-      logger().warn("{} server_reconnect: I was a client and didn't received the"
+      logger().warn("{} server_reconnect: I was a client (cc={}) and didn't received the"
                     " server_ident with existing connection {}."
                     " Asking peer to resume session establishment",
-                    conn, *existing_conn);
+                    conn, existing_proto->client_cookie, *existing_conn);
       return send_reset(false);
     }
 
     if (existing_proto->peer_global_seq > reconnect.global_seq()) {
-      logger().warn("{} server_reconnect: stale global_seq: exist_pgs={} peer_gs={},"
+      logger().warn("{} server_reconnect: stale global_seq: exist_pgs({}) > peer_gs({}),"
                     " with existing connection {},"
                     " ask client to retry global",
                     conn, existing_proto->peer_global_seq,
@@ -1418,34 +1432,34 @@ ProtocolV2::server_reconnect()
     }
 
     if (existing_proto->connect_seq > reconnect.connect_seq()) {
-      logger().warn("{} server_reconnect: stale connect_seq exist_cs={} peer_cs={},"
-                    " with existing connection {},"
-                    " ask client to retry",
-                    conn, existing_proto->connect_seq, reconnect.connect_seq(),
-                    *existing_conn);
+      logger().warn("{} server_reconnect: stale peer connect_seq peer_cs({}) < exist_cs({}),"
+                    " with existing connection {}, ask client to retry",
+                    conn, reconnect.connect_seq(),
+                    existing_proto->connect_seq, *existing_conn);
       return send_retry(existing_proto->connect_seq);
     } else if (existing_proto->connect_seq == reconnect.connect_seq()) {
       // reconnect race: both peers are sending reconnect messages
       if (existing_conn->peer_addr > messenger.get_myaddrs().msgr2_addr() &&
           !existing_conn->policy.server) {
         // the existing connection wins
-        logger().warn("{} server_reconnect: reconnect race detected,"
-                      " this connection loses to existing connection {},"
-                      " ask client to wait", conn, *existing_conn);
+        logger().warn("{} server_reconnect: reconnect race detected (cs={})"
+                      " and lose to existing {}, ask client to wait",
+                      conn, reconnect.connect_seq(), *existing_conn);
         return send_wait();
       } else {
         // this connection wins
-        logger().warn("{} server_reconnect: reconnect race detected,"
-                      " replacing existing connection {}"
-                      " socket by this connection's socket",
-                      conn, *existing_conn);
+        logger().warn("{} server_reconnect: reconnect race detected (cs={})"
+                      " and win, reusing existing {}",
+                      conn, reconnect.connect_seq(), *existing_conn);
         return reuse_connection(
             existing_proto, false,
             true, reconnect.connect_seq(), reconnect.msg_seq());
       }
     } else { // existing_proto->connect_seq < reconnect.connect_seq()
-      logger().warn("{} server_reconnect: stale exsiting connection {},"
-                    " replacing", conn, *existing_conn);
+      logger().warn("{} server_reconnect: stale exsiting connect_seq exist_cs({}) < peer_cs({}),"
+                    " reusing existing {}",
+                    conn, existing_proto->connect_seq,
+                    reconnect.connect_seq(), *existing_conn);
       return reuse_connection(
           existing_proto, false,
           true, reconnect.connect_seq(), reconnect.msg_seq());
@@ -1455,14 +1469,16 @@ ProtocolV2::server_reconnect()
 
 void ProtocolV2::execute_accepting()
 {
-  trigger_state(state_t::ACCEPTING, write_state_t::none, false);
+  // TODO: change to write_state_t::none
+  trigger_state(state_t::ACCEPTING, write_state_t::delay, false);
   seastar::with_gate(pending_dispatch, [this] {
-      auth_meta = seastar::make_lw_shared<AuthConnectionMeta>();
-      session_stream_handlers = { nullptr, nullptr };
-      enable_recording();
-      return banner_exchange()
-        .then([this] (entity_type_t _peer_type,
-                      entity_addr_t _my_addr_from_peer) {
+      return seastar::futurize_apply([this] {
+          auth_meta = seastar::make_lw_shared<AuthConnectionMeta>();
+          session_stream_handlers = { nullptr, nullptr };
+          enable_recording();
+          return banner_exchange();
+        }).then([this] (entity_type_t _peer_type,
+                        entity_addr_t _my_addr_from_peer) {
           ceph_assert(conn.get_peer_type() == 0);
           conn.set_peer_type(_peer_type);
 
@@ -1498,6 +1514,13 @@ void ProtocolV2::execute_accepting()
         }).then([this] (next_step_t next) {
           switch (next) {
            case next_step_t::ready: {
+            seastar::with_gate(pending_dispatch, [this] {
+              return dispatcher.ms_handle_accept(
+                  seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()));
+            }).handle_exception([this] (std::exception_ptr eptr) {
+              logger().error("{} ms_handle_accept caught exception: {}", conn, eptr);
+              ceph_abort("unexpected exception from ms_handle_accept()");
+            });
             messenger.register_conn(
               seastar::static_pointer_cast<SocketConnection>(
                 conn.shared_from_this()));
@@ -1605,16 +1628,6 @@ ProtocolV2::send_server_ident()
 
     conn.set_features(connection_features);
 
-    // notify
-    seastar::with_gate(pending_dispatch, [this] {
-      return dispatcher.ms_handle_accept(
-          seastar::static_pointer_cast<SocketConnection>(conn.shared_from_this()))
-      .handle_exception([this] (std::exception_ptr eptr) {
-        logger().error("{} ms_handle_accept caught exception: {}", conn, eptr);
-        ceph_abort("unexpected exception from ms_handle_accept()");
-      });
-    });
-
     return write_frame(server_ident);
   }).then([] {
     return next_step_t::ready;
@@ -1654,7 +1667,7 @@ void ProtocolV2::trigger_replacing(bool reconnect,
         reset_session(true);
       }
       protocol_timer.cancel();
-      return std::move(execution_done);
+      return execution_done.get_future();
     }).then([this,
              reconnect,
              new_socket = std::move(new_socket),
@@ -1851,17 +1864,17 @@ seastar::future<> ProtocolV2::read_message(utime_t throttle_stamp)
     // TODO: change MessageRef with seastar::shared_ptr
     auto msg_ref = MessageRef{message, false};
     seastar::with_gate(pending_dispatch, [this, msg = std::move(msg_ref)] {
-      return dispatcher.ms_dispatch(&conn, std::move(msg))
-	.handle_exception([this] (std::exception_ptr eptr) {
-        logger().error("{} ms_dispatch caught exception: {}", conn, eptr);
-        ceph_abort("unexpected exception from ms_dispatch()");
-      });
+      return dispatcher.ms_dispatch(&conn, std::move(msg));
+    }).handle_exception([this] (std::exception_ptr eptr) {
+      logger().error("{} ms_dispatch caught exception: {}", conn, eptr);
+      ceph_abort("unexpected exception from ms_dispatch()");
     });
   });
 }
 
 void ProtocolV2::execute_ready()
 {
+  assert(conn.policy.lossy || (client_cookie != 0 && server_cookie != 0));
   trigger_state(state_t::READY, write_state_t::open, false);
   execution_done = seastar::with_gate(pending_dispatch, [this] {
     protocol_timer.cancel();

src/crimson/net/ProtocolV2.h

@@ -76,7 +76,7 @@ class ProtocolV2 final : public Protocol {
   uint64_t peer_global_seq = 0;
   uint64_t connect_seq = 0;
 
-  seastar::future<> execution_done = seastar::now();
+  seastar::shared_future<> execution_done = seastar::now();
 
   class Timer {
     double last_dur_ = 0.0;

src/crimson/net/SocketConnection.cc

@@ -41,13 +41,20 @@ SocketConnection::get_messenger() const {
   return &messenger;
 }
 
-seastar::future<bool> SocketConnection::is_connected()
+bool SocketConnection::is_connected() const
 {
-  return seastar::smp::submit_to(shard_id(), [this] {
-      return protocol->is_connected();
-    });
+  ceph_assert(seastar::engine().cpu_id() == shard_id());
+  return protocol->is_connected();
 }
 
+#ifdef UNIT_TESTS_BUILT
+bool SocketConnection::is_closed() const
+{
+  ceph_assert(seastar::engine().cpu_id() == shard_id());
+  return protocol->is_closed();
+}
+#endif
+
 seastar::future<> SocketConnection::send(MessageRef msg)
 {
   // Cannot send msg from another core now, its ref counter can be contaminated!

src/crimson/net/SocketConnection.h

@@ -79,7 +79,11 @@ class SocketConnection : public Connection {
 
   Messenger* get_messenger() const override;
 
-  seastar::future<bool> is_connected() override;
+  bool is_connected() const override;
+
+#ifdef UNIT_TESTS_BUILT
+  bool is_closed() const override;
+#endif
 
   seastar::future<> send(MessageRef msg) override;