Method encode

src/msg/Message.cc:237–308

#define dout_subsys ceph_subsys_ms

void Message::encode(uint64_t features, int crcflags, bool skip_header_crc)
{
  // encode and copy out of *m
  if (empty_payload()) {
    ceph_assert(middle.length() == 0);
    encode_payload(features);

    if (byte_throttler) {
      byte_throttler->take(payload.length() + middle.length());
    }

    // if the encoder didn't specify past compatibility, we assume it
    // is incompatible.
    if (header.compat_version == 0)
      header.compat_version = header.version;
  }
  if (crcflags & MSG_CRC_HEADER)
    calc_front_crc();

  // update envelope
  header.front_len = get_payload().length();
  header.middle_len = get_middle().length();
  header.data_len = get_data().length();
  if (!skip_header_crc && (crcflags & MSG_CRC_HEADER))
    calc_header_crc();

  footer.flags = CEPH_MSG_FOOTER_COMPLETE;

  if (crcflags & MSG_CRC_DATA) {
    calc_data_crc();

#ifdef ENCODE_DUMP
    bufferlist bl;
    encode(get_header(), bl);

    // dump the old footer format
    ceph_msg_footer_old old_footer;
    old_footer.front_crc = footer.front_crc;
    old_footer.middle_crc = footer.middle_crc;
    old_footer.data_crc = footer.data_crc;
    old_footer.flags = footer.flags;
    encode(old_footer, bl);

    encode(get_payload(), bl);
    encode(get_middle(), bl);
    encode(get_data(), bl);

    // this is almost an exponential backoff, except because we count
    // bits we tend to sample things we encode later, which should be
    // more representative.
    static int i = 0;
    i++;
    int bits = 0;
    for (unsigned t = i; t; bits++)
      t &= t - 1;
    if (bits <= 2) {
      char fn[200];
      int status;