CtPtr ProtocolV2::reuse_connection(const AsyncConnectionRef& existing,
                                   ProtocolV2 *exproto) {
  ldout(cct, 20) << __func__ << " existing=" << existing
                 << " reconnect=" << reconnecting << dendl;

  connection->inject_delay();

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

  connection->center->delete_file_event(connection->cs.fd(),
                                        EVENT_READABLE | EVENT_WRITABLE);

  if (existing->delay_state) {
    existing->delay_state->flush();
    ceph_assert(!connection->delay_state);
  }
  exproto->reset_recv_state();
  exproto->pre_auth.enabled = false;

  if (!reconnecting) {
    exproto->client_cookie = client_cookie;
    exproto->peer_name = peer_name;
    exproto->connection_features = connection_features;
    existing->set_features(connection_features);
  }
  exproto->peer_global_seq = peer_global_seq;

  ceph_assert(connection->center->in_thread());
  auto temp_cs = std::move(connection->cs);
  EventCenter *new_center = connection->center;
  Worker *new_worker = connection->worker;
  // we can steal the session_stream_handlers under the assumption
  // this happens in the event center's thread as there should be
  // no user outside its boundaries (simlarly to e.g. outgoing_bl).
  auto temp_stream_handlers = std::move(session_stream_handlers);
  exproto->auth_meta = auth_meta;

  ldout(messenger->cct, 5) << __func__ << " stop myself to swap existing"
                           << dendl;

  // avoid _stop shutdown replacing socket
  // queue a reset on the new connection, which we're dumping for the old
  stop();

  connection->dispatch_queue->queue_reset(connection);

  exproto->can_write = false;
  exproto->write_in_progress = false;
  exproto->reconnecting = reconnecting;
  exproto->replacing = true;
  existing->state_offset = 0;
  // avoid previous thread modify event
  exproto->state = NONE;
  existing->state = AsyncConnection::STATE_NONE;
  // Discard existing prefetch buffer in `recv_buf`
  existing->recv_start = existing->recv_end = 0;
  // there shouldn't exist any buffer
  ceph_assert(connection->recv_start == connection->recv_end);

  auto deactivate_existing = std::bind(
      [ existing,
        new_worker,
        new_center,
        exproto,
        temp_stream_handlers=std::move(temp_stream_handlers)
      ](ConnectedSocket &cs) mutable {
        // we need to delete time event in original thread
        {
          std::lock_guard<std::mutex> l(existing->lock);
          existing->write_lock.lock();
          exproto->requeue_sent();
          // XXX: do we really need the locking for `outgoing_bl`? There is
          // a comment just above its definition saying "lockfree, only used
          // in own thread". I'm following lockfull schema just in the case.
          // From performance point of view it should be fine – this happens
          // far away from hot paths.
          existing->outgoing_bl.clear();
          existing->open_write = false;
          exproto->session_stream_handlers = std::move(temp_stream_handlers);
          existing->write_lock.unlock();
          if (exproto->state == NONE) {
            existing->shutdown_socket();
            existing->cs = std::move(cs);
            existing->worker->references--;
            new_worker->references++;
            existing->logger = new_worker->get_perf_counter();
            existing->worker = new_worker;
            existing->center = new_center;
            if (existing->delay_state)
              existing->delay_state->set_center(new_center);
          } else if (exproto->state == CLOSED) {
            auto back_to_close = std::bind(
                [](ConnectedSocket &cs) mutable { cs.close(); }, std::move(cs));
            new_center->submit_to(new_center->get_id(),
                                  std::move(back_to_close), true);
            return;
          } else {
            ceph_abort();
          }
        }

        // Before changing existing->center, it may already exists some
        // events in existing->center's queue. Then if we mark down
        // `existing`, it will execute in another thread and clean up
        // connection. Previous event will result in segment fault
        auto transfer_existing = [existing, exproto]() mutable {
          std::lock_guard<std::mutex> l(existing->lock);
          if (exproto->state == CLOSED) return;
          ceph_assert(exproto->state == NONE);

          exproto->state = SESSION_ACCEPTING;
          // we have called shutdown_socket above
          ceph_assert(existing->last_tick_id == 0);
          // restart timer since we are going to re-build connection
          existing->last_connect_started = ceph::coarse_mono_clock::now();
          existing->last_tick_id = existing->center->create_time_event(
            existing->connect_timeout_us, existing->tick_handler);
          existing->state = AsyncConnection::STATE_CONNECTION_ESTABLISHED;
          existing->center->create_file_event(existing->cs.fd(), EVENT_READABLE,
                                              existing->read_handler);
          if (!exproto->reconnecting) {
            exproto->run_continuation(exproto->send_server_ident());
          } else {
            exproto->run_continuation(exproto->send_reconnect_ok());
          }
        };
        if (existing->center->in_thread())
          transfer_existing();
        else
          existing->center->submit_to(existing->center->get_id(),
                                      std::move(transfer_existing), true);
      },
      std::move(temp_cs));

  existing->center->submit_to(existing->center->get_id(),
                              std::move(deactivate_existing), true);
  return nullptr;
}