CtPtr ProtocolV2::handle_reconnect(ceph::bufferlist &payload)
{
  ldout(cct, 20) << __func__
		 << " payload.length()=" << payload.length() << dendl;

  if (state != SESSION_ACCEPTING) {
    lderr(cct) << __func__ << " not in session accept state!" << dendl;
    return _fault();
  }

  auto reconnect = ReconnectFrame::Decode(payload);

  ldout(cct, 5) << __func__
                << " received reconnect:" 
                << " client_cookie=" << std::hex << reconnect.client_cookie()
                << " server_cookie=" << reconnect.server_cookie() << std::dec
                << " gs=" << reconnect.global_seq()
                << " cs=" << reconnect.connect_seq()
                << " ms=" << reconnect.msg_seq()
		            << dendl;

  // Should we check if one of the ident.addrs match connection->target_addr
  // as we do in ProtocolV1?
  connection->set_peer_addrs(reconnect.addrs());
  connection->target_addr = connection->_infer_target_addr(reconnect.addrs());
  peer_global_seq = reconnect.global_seq();

  connection->lock.unlock();
  AsyncConnectionRef existing = messenger->lookup_conn(*connection->peer_addrs);

  if (existing &&
      existing->protocol->proto_type != 2) {
    ldout(cct,1) << __func__ << " existing " << existing << " proto "
		 << existing->protocol.get() << " version is "
		 << existing->protocol->proto_type << ", marking down" << dendl;
    existing->mark_down();
    existing = nullptr;
  }

  connection->inject_delay();

  connection->lock.lock();
  if (state != SESSION_ACCEPTING) {
    ldout(cct, 1) << __func__
                  << " state changed while accept, it must be mark_down"
                  << dendl;
    ceph_assert(state == CLOSED);
    return _fault();
  }

  if (!existing) {
    // there is no existing connection therefore cannot reconnect to previous
    // session
    ldout(cct, 0) << __func__
                  << " no existing connection exists, reseting client" << dendl;
    auto reset = ResetFrame::Encode(true);
    return WRITE(reset, "session reset", read_frame);
  }

  std::lock_guard<std::mutex> l(existing->lock);

  ProtocolV2 *exproto = dynamic_cast<ProtocolV2 *>(existing->protocol.get());
  if (!exproto) {
    ldout(cct, 1) << __func__ << " existing=" << existing << dendl;
    ceph_assert(false);
  }

  if (exproto->state == CLOSED) {
    ldout(cct, 5) << __func__ << " existing " << existing
                  << " already closed. Reseting client" << dendl;
    auto reset = ResetFrame::Encode(true);
    return WRITE(reset, "session reset", read_frame);
  }

  if (exproto->replacing) {
    ldout(cct, 1) << __func__
                  << " existing racing replace happened while replacing."
                  << " existing=" << existing << dendl;
    auto retry = RetryGlobalFrame::Encode(exproto->peer_global_seq);
    return WRITE(retry, "session retry", read_frame);
  }

  if (exproto->client_cookie != reconnect.client_cookie()) {
    ldout(cct, 1) << __func__ << " existing=" << existing
                  << " client cookie mismatch, I must have reseted:"
                  << " cc=" << std::hex << exproto->client_cookie
                  << " rcc=" << reconnect.client_cookie()
                  << ", reseting client." << std::dec
                  << dendl;
    auto reset = ResetFrame::Encode(connection->policy.resetcheck);
    return WRITE(reset, "session reset", read_frame);
  } else if (exproto->server_cookie == 0) {
    // this happens when:
    //   - a connects to b
    //   - a sends client_ident
    //   - b gets client_ident, sends server_ident and sets cookie X
    //   - connection fault
    //   - b reconnects to a with cookie X, connect_seq=1
    //   - a has cookie==0
    ldout(cct, 1) << __func__ << " I was a client and didn't received the"
                  << " server_ident. Asking peer to resume session"
                  << " establishment" << dendl;
    auto reset = ResetFrame::Encode(false);
    return WRITE(reset, "session reset", read_frame);
  }

  if (exproto->peer_global_seq > reconnect.global_seq()) {
    ldout(cct, 5) << __func__
                  << " stale global_seq: sgs=" << exproto->peer_global_seq
                  << " cgs=" << reconnect.global_seq()
                  << ", ask client to retry global" << dendl;
    auto retry = RetryGlobalFrame::Encode(exproto->peer_global_seq);

    INTERCEPT(18);

    return WRITE(retry, "session retry", read_frame);
  }

  if (exproto->connect_seq > reconnect.connect_seq()) {
    ldout(cct, 5) << __func__
                  << " stale connect_seq scs=" << exproto->connect_seq
                  << " ccs=" << reconnect.connect_seq()
                  << " , ask client to retry" << dendl;
    auto retry = RetryFrame::Encode(exproto->connect_seq);
    return WRITE(retry, "session retry", read_frame);
  }

  if (exproto->connect_seq == reconnect.connect_seq()) {
    // reconnect race: both peers are sending reconnect messages
    if (existing->peer_addrs->msgr2_addr() >
            messenger->get_myaddrs().msgr2_addr() &&
        !existing->policy.server) {
      // the existing connection wins
      ldout(cct, 1)
          << __func__
          << " reconnect race detected, this connection loses to existing="
          << existing << dendl;

      auto wait = WaitFrame::Encode();
      return WRITE(wait, "wait", read_frame);
    } else {
      // this connection wins
      ldout(cct, 1) << __func__
                    << " reconnect race detected, replacing existing="
                    << existing << " socket by this connection's socket"
                    << dendl;
    }
  }

  ldout(cct, 1) << __func__ << " reconnect to existing=" << existing << dendl;

  reconnecting = true;

  // everything looks good
  exproto->connect_seq = reconnect.connect_seq();
  exproto->message_seq = reconnect.msg_seq();

  return reuse_connection(existing, exproto);
}