Method process

src/msg/async/AsyncConnection.cc:386–501

}

void AsyncConnection::process() {
  std::lock_guard<std::mutex> l(lock);
  last_active = ceph::coarse_mono_clock::now();
  recv_start_time = ceph::mono_clock::now();

  ldout(async_msgr->cct, 20) << __func__ << dendl;

  switch (state) {
    case STATE_NONE: {
      ldout(async_msgr->cct, 20) << __func__ << " enter none state" << dendl;
      return;
    }
    case STATE_CLOSED: {
      ldout(async_msgr->cct, 20) << __func__ << " socket closed" << dendl;
      return;
    }
    case STATE_CONNECTING: {
      ceph_assert(!policy.server);

      // clear timer (if any) since we are connecting/re-connecting
      if (last_tick_id) {
        center->delete_time_event(last_tick_id);
      }
      last_connect_started = ceph::coarse_mono_clock::now();
      last_tick_id = center->create_time_event(
          connect_timeout_us, tick_handler);

      if (cs) {
        center->delete_file_event(cs.fd(), EVENT_READABLE | EVENT_WRITABLE);
        cs.close();
      }

      SocketOptions opts;
      opts.priority = async_msgr->get_socket_priority();
      if (async_msgr->cct->_conf->mon_use_min_delay_socket) {
          if (async_msgr->get_mytype() == CEPH_ENTITY_TYPE_MON &&
              peer_is_mon()) {
            opts.priority = SOCKET_PRIORITY_MIN_DELAY;
          }
      }
      opts.connect_bind_addr = msgr->get_myaddrs().front();
      ssize_t r = worker->connect(target_addr, opts, &cs);
      if (r < 0) {
        protocol->fault();
        return;
      }

      center->create_file_event(cs.fd(), EVENT_READABLE, read_handler);
      state = STATE_CONNECTING_RE;
    }
    case STATE_CONNECTING_RE: {
      ssize_t r = cs.is_connected();
      if (r < 0) {
        ldout(async_msgr->cct, 1) << __func__ << " reconnect failed to "
                                  << target_addr << dendl;
        if (r == -ECONNREFUSED) {
          ldout(async_msgr->cct, 2)
              << __func__ << " connection refused!" << dendl;