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

  if (state != HELLO_CONNECTING && state != HELLO_ACCEPTING) {
    lderr(cct) << __func__ << " not in hello exchange state!" << dendl;
    return _fault();
  }

  auto hello = HelloFrame::Decode(payload);

  ldout(cct, 5) << __func__ << " received hello:"
                << " peer_type=" << (int)hello.entity_type()
                << " peer_addr_for_me=" << hello.peer_addr() << dendl;

  sockaddr_storage ss;
  socklen_t len = sizeof(ss);
  getsockname(connection->cs.fd(), (sockaddr *)&ss, &len);
  ldout(cct, 5) << __func__ << " getsockname says I am " << (sockaddr *)&ss
		<< " when talking to " << connection->target_addr << dendl;

  if (connection->get_peer_type() == -1) {
    connection->set_peer_type(hello.entity_type());

    ceph_assert(state == HELLO_ACCEPTING);
    connection->policy = messenger->get_policy(hello.entity_type());
    ldout(cct, 10) << __func__ << " accept of host_type "
                   << (int)hello.entity_type()
                   << ", policy.lossy=" << connection->policy.lossy
                   << " policy.server=" << connection->policy.server
                   << " policy.standby=" << connection->policy.standby
                   << " policy.resetcheck=" << connection->policy.resetcheck
                   << dendl;
  } else {
    ceph_assert(state == HELLO_CONNECTING);
    if (connection->get_peer_type() != hello.entity_type()) {
      ldout(cct, 1) << __func__ << " connection peer type does not match what"
                    << " peer advertises " << connection->get_peer_type()
                    << " != " << (int)hello.entity_type() << dendl;
      stop();
      connection->dispatch_queue->queue_reset(connection);
      return nullptr;
    }
  }

  if (messenger->get_myaddrs().empty() ||
      messenger->get_myaddrs().front().is_blank_ip()) {
    entity_addr_t a;
    if (cct->_conf->ms_learn_addr_from_peer) {
      ldout(cct, 1) << __func__ << " peer " << connection->target_addr
		    << " says I am " << hello.peer_addr() << " (socket says "
		    << (sockaddr*)&ss << ")" << dendl;
      a = hello.peer_addr();
    } else {
      ldout(cct, 1) << __func__ << " socket to  " << connection->target_addr
		    << " says I am " << (sockaddr*)&ss
		    << " (peer says " << hello.peer_addr() << ")" << dendl;
      a.set_sockaddr((sockaddr *)&ss);
    }
    a.set_type(entity_addr_t::TYPE_MSGR2); // anything but NONE; learned_addr ignores this
    a.set_port(0);
    connection->lock.unlock();
    messenger->learned_addr(a);
    if (cct->_conf->ms_inject_internal_delays &&
        cct->_conf->ms_inject_socket_failures) {
      if (rand() % cct->_conf->ms_inject_socket_failures == 0) {
        ldout(cct, 10) << __func__ << " sleep for "
                       << cct->_conf->ms_inject_internal_delays << dendl;
        utime_t t;
        t.set_from_double(cct->_conf->ms_inject_internal_delays);
        t.sleep();
      }
    }
    connection->lock.lock();
    if (state != HELLO_CONNECTING) {
      ldout(cct, 1) << __func__
                    << " state changed while learned_addr, mark_down or "
                    << " replacing must be happened just now" << dendl;
      return nullptr;
    }
  }



  CtPtr callback;
  callback = bannerExchangeCallback;
  bannerExchangeCallback = nullptr;
  ceph_assert(callback);
  return callback;
}