void Monitor::_ms_dispatch(Message *m)
{
  if (is_shutdown()) {
    m->put();
    return;
  }

  MonOpRequestRef op = op_tracker.create_request<MonOpRequest>(m);
  bool src_is_mon = op->is_src_mon();
  op->mark_event("mon:_ms_dispatch");
  MonSession *s = op->get_session();
  if (s && s->closed) {
    return;
  }

  if (src_is_mon && s) {
    ConnectionRef con = m->get_connection();
    if (con->get_messenger() && con->get_features() != s->con_features) {
      // only update features if this is a non-anonymous connection
      dout(10) << __func__ << " feature change for " << m->get_source_inst()
               << " (was " << s->con_features
               << ", now " << con->get_features() << ")" << dendl;
      // connection features changed - recreate session.
      if (s->con && s->con != con) {
        dout(10) << __func__ << " connection for " << m->get_source_inst()
                 << " changed from session; mark down and replace" << dendl;
        s->con->mark_down();
      }
      if (s->item.is_on_list()) {
        // forwarded messages' sessions are not in the sessions map and
        // exist only while the op is being handled.
        remove_session(s);
      }
      s->put();
      s = nullptr;
    }
  }

  if (!s) {
    // if the sender is not a monitor, make sure their first message for a
    // session is an MAuth.  If it is not, assume it's a stray message,
    // and considering that we are creating a new session it is safe to
    // assume that the sender hasn't authenticated yet, so we have no way
    // of assessing whether we should handle it or not.
    if (!src_is_mon && (m->get_type() != CEPH_MSG_AUTH &&
			m->get_type() != CEPH_MSG_MON_GET_MAP &&
			m->get_type() != CEPH_MSG_PING)) {
      dout(1) << __func__ << " dropping stray message " << *m
	      << " from " << m->get_source_inst() << dendl;
      return;
    }

    ConnectionRef con = m->get_connection();
    {
      Mutex::Locker l(session_map_lock);
      s = session_map.new_session(m->get_source_inst(), con.get());
    }
    assert(s);
    con->set_priv(s->get());
    dout(10) << __func__ << " new session " << s << " " << *s
	     << " features 0x" << std::hex
	     << s->con_features << std::dec << dendl;
    op->set_session(s);

    logger->set(l_mon_num_sessions, session_map.get_size());
    logger->inc(l_mon_session_add);

    if (src_is_mon) {
      // give it monitor caps; the peer type has been authenticated
      dout(5) << __func__ << " setting monitor caps on this connection" << dendl;
      if (!s->caps.is_allow_all()) // but no need to repeatedly copy
        s->caps = *mon_caps;
    }
    s->put();
  } else {
    dout(20) << __func__ << " existing session " << s << " for " << s->inst
	     << dendl;
  }

  assert(s);

  s->session_timeout = ceph_clock_now();
  s->session_timeout += g_conf->mon_session_timeout;

  if (s->auth_handler) {
    s->entity_name = s->auth_handler->get_entity_name();
  }
  dout(20) << " caps " << s->caps.get_str() << dendl;

  if ((is_synchronizing() ||
       (s->global_id == 0 && !exited_quorum.is_zero())) &&
      !src_is_mon &&
      m->get_type() != CEPH_MSG_PING) {
    waitlist_or_zap_client(op);
  } else {
    dispatch_op(op);
  }
  return;
}