Ceph » RADOS

From Neha:

• http://telemetry.front.sepia.ceph.com:4000/d/jByk5HaMz/crash-spec-x-ray?var-sig_v2=12eed3bdd041d05365ec4192181b37d22054f1fec42e589c2e1fc8e6993ed467&orgId=1
• http://telemetry.front.sepia.ceph.com:4000/d/jByk5HaMz/crash-spec-x-ray?var-sig_v2=5532f5c395e705ebfcbe87dae6ae4269e074636a7dae01956a69d44d02f5f401&orgId=1
• http://telemetry.front.sepia.ceph.com:4000/d/Nvj6XTaMk/spec-search?orgId=1&var-substr_1=&var-substr_2=&var-substr_3=&var-majors_affected=&var-minors_affected=&var-assert_function=&var-assert_condition=m_descs.empty()&var-total_results=2&var-sig_v1=&var-sig_v2=&var-daemons=&var-only_new_fingerprints=false&var-status_description=All&var-only_open=false

The number of elements in FrameAssembler::m_desc can be altered only by:

1.

bufferlist FrameAssembler::assemble_frame(Tag tag, bufferlist segment_bls[],
                                          const uint16_t segment_aligns[],
                                          size_t segment_count) {
  m_flags = 0;
  m_descs.resize(calc_num_segments(segment_bls, segment_count));
  // ...
}

2.

void set_is_rev1(bool is_rev1) {
  m_descs.clear();
  m_flags = 0;
  m_is_rev1 = is_rev1;
}

set_is_rev1() can be called from:
1. ProtocolV2::_handle_peer_banner_payload() (the hello frame sequence) or
2. ProtocolV2::reuse_connection().

Guess which one is the culprit! ;-)

It looks reuse_connection() holds the

CtPtr ProtocolV2::reuse_connection(const AsyncConnectionRef& existing,
                                   ProtocolV2 *exproto) {
  // ...
  std::lock_guard<std::mutex> l(existing->write_lock);

  // ...
  if (!reconnecting) {
    exproto->peer_supported_features = peer_supported_features;
    exproto->tx_frame_asm.set_is_rev1(tx_frame_asm.get_is_rev1());
    exproto->rx_frame_asm.set_is_rev1(rx_frame_asm.get_is_rev1());

    // ...
  }
  // ...
}

Let's see the caller of write_message():

void ProtocolV2::write_event() {
      // ...
      connection->write_lock.unlock();

      // send_message or requeue messages may not encode message
      if (!out_entry.is_prepared) {
        prepare_send_message(connection->get_features(), out_entry.m);
      }

      if (out_entry.m->queue_start != ceph::mono_time()) {
        connection->logger->tinc(l_msgr_send_messages_queue_lat,
                                 ceph::mono_clock::now() -
                                 out_entry.m->queue_start);
      }

      r = write_message(out_entry.m, more);

      connection->write_lock.lock();

write_message() only takes the AsyncConnection::lock which is not enough. Oops!

ssize_t ProtocolV2::write_message(Message *m, bool more) {
  FUNCTRACE(cct);
  ceph_assert(connection->center->in_thread());
  m->set_seq(++out_seq);

  connection->lock.lock();
  uint64_t ack_seq = in_seq;
  ack_left = 0;
  connection->lock.unlock();

  ceph_msg_header &header = m->get_header();
  ceph_msg_footer &footer = m->get_footer();

  ceph_msg_header2 header2{header.seq,        header.tid,
                           header.type,       header.priority,
                           header.version,
                           ceph_le32(0),      header.data_off,
                           ceph_le64(ack_seq),
                           footer.flags,      header.compat_version,
                           header.reserved};

  auto message = MessageFrame::Encode(
                             header2,
                             m->get_payload(),
                             m->get_middle(),
                             m->get_data());
  if (!append_frame(message)) {
    m->put();
    return -EILSEQ;
  }

Perhaps we should move into deactivate_existing part of reuse_connection() where we hold both locks the same time.