ssize_t AsyncConnection::write_message(Message *m, bufferlist& bl, bool more)
{
  FUNCTRACE();
  assert(center->in_thread());
  m->set_seq(++out_seq);

  if (msgr->crcflags & MSG_CRC_HEADER)
    m->calc_header_crc();

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

  // TODO: let sign_message could be reentry?
  // Now that we have all the crcs calculated, handle the
  // digital signature for the message, if the AsyncConnection has session
  // security set up.  Some session security options do not
  // actually calculate and check the signature, but they should
  // handle the calls to sign_message and check_signature.  PLR
  if (session_security.get() == NULL) {
    ldout(async_msgr->cct, 20) << __func__ << " no session security" << dendl;
  } else {
    if (session_security->sign_message(m)) {
      ldout(async_msgr->cct, 20) << __func__ << " failed to sign m="
                                 << m << "): sig = " << footer.sig << dendl;
    } else {
      ldout(async_msgr->cct, 20) << __func__ << " signed m=" << m
                                 << "): sig = " << footer.sig << dendl;
    }
  }
  
  unsigned original_bl_len = outcoming_bl.length();

  outcoming_bl.append(CEPH_MSGR_TAG_MSG);

  if (has_feature(CEPH_FEATURE_NOSRCADDR)) {
    outcoming_bl.append((char*)&header, sizeof(header));
  } else {
    ceph_msg_header_old oldheader;
    memcpy(&oldheader, &header, sizeof(header));
    oldheader.src.name = header.src;
    oldheader.src.addr = get_peer_addr();
    oldheader.orig_src = oldheader.src;
    oldheader.reserved = header.reserved;
    oldheader.crc = ceph_crc32c(0, (unsigned char*)&oldheader,
                                sizeof(oldheader) - sizeof(oldheader.crc));
    outcoming_bl.append((char*)&oldheader, sizeof(oldheader));
  }

  ldout(async_msgr->cct, 20) << __func__ << " sending message type=" << header.type
                             << " src " << entity_name_t(header.src)
                             << " front=" << header.front_len
                             << " data=" << header.data_len
                             << " off " << header.data_off << dendl;

  if ((bl.length() <= ASYNC_COALESCE_THRESHOLD) && (bl.buffers().size() > 1)) {
    for (const auto &pb : bl.buffers()) {
      outcoming_bl.append((char*)pb.c_str(), pb.length());
    }
  } else {
    outcoming_bl.claim_append(bl);  
  }

  // send footer; if receiver doesn't support signatures, use the old footer format
  ceph_msg_footer_old old_footer;
  if (has_feature(CEPH_FEATURE_MSG_AUTH)) {
    outcoming_bl.append((char*)&footer, sizeof(footer));
  } else {
    if (msgr->crcflags & MSG_CRC_HEADER) {
      old_footer.front_crc = footer.front_crc;
      old_footer.middle_crc = footer.middle_crc;
      old_footer.data_crc = footer.data_crc;
    } else {
       old_footer.front_crc = old_footer.middle_crc = 0;
    }
    old_footer.data_crc = msgr->crcflags & MSG_CRC_DATA ? footer.data_crc : 0;
    old_footer.flags = footer.flags;
    outcoming_bl.append((char*)&old_footer, sizeof(old_footer));
  }

  m->trace.event("async writing message");
  ldout(async_msgr->cct, 20) << __func__ << " sending " << m->get_seq()
                             << " " << m << dendl;
  ssize_t total_send_size = outcoming_bl.length();
  ssize_t rc = _try_send(more);
  if (rc < 0) {
    ldout(async_msgr->cct, 1) << __func__ << " error sending " << m << ", "
                              << cpp_strerror(rc) << dendl;
  } else if (rc == 0) {
    logger->inc(l_msgr_send_bytes, total_send_size - original_bl_len);
    ldout(async_msgr->cct, 10) << __func__ << " sending " << m << " done." << dendl;
  } else {
    logger->inc(l_msgr_send_bytes, total_send_size - outcoming_bl.length());
    ldout(async_msgr->cct, 10) << __func__ << " sending " << m << " continuely." << dendl;
  }
  if (m->get_type() == CEPH_MSG_OSD_OP)
    OID_EVENT_TRACE_WITH_MSG(m, "SEND_MSG_OSD_OP_END", false);
  else if (m->get_type() == CEPH_MSG_OSD_OPREPLY)
    OID_EVENT_TRACE_WITH_MSG(m, "SEND_MSG_OSD_OPREPLY_END", false);
  m->put();

  return rc;
}