void Pipe::writer()
{
  pipe_lock.Lock();
  while (state != STATE_CLOSED) {// && state != STATE_WAIT) {
    ldout(msgr->cct,10) << "writer: state = " << get_state_name()
			<< " policy.server=" << policy.server << dendl;

    // standby?
    if (is_queued() && state == STATE_STANDBY && !policy.server)
      state = STATE_CONNECTING;

    // connect?
    if (state == STATE_CONNECTING) {
      assert(!policy.server);
      connect();
      continue;
    }
    
    if (state == STATE_CLOSING) {
      // write close tag
      ldout(msgr->cct,20) << "writer writing CLOSE tag" << dendl;
      char tag = CEPH_MSGR_TAG_CLOSE;
      state = STATE_CLOSED;
      state_closed = true;
      pipe_lock.Unlock();
      if (sd >= 0) {
	// we can ignore return value, actually; we don't care if this succeeds.
	int r = ::write(sd, &tag, 1);
	(void)r;
      }
      pipe_lock.Lock();
      continue;
    }

    if (state != STATE_CONNECTING && state != STATE_WAIT && state != STATE_STANDBY &&
	(is_queued() || in_seq > in_seq_acked)) {

      // keepalive?
      if (send_keepalive) {
	int rc;
	if (connection_state->has_feature(CEPH_FEATURE_MSGR_KEEPALIVE2)) {
	  pipe_lock.Unlock();
	  rc = write_keepalive2(CEPH_MSGR_TAG_KEEPALIVE2,
				ceph_clock_now());
	} else {
	  pipe_lock.Unlock();
	  rc = write_keepalive();
	}
	pipe_lock.Lock();
	if (rc < 0) {
	  ldout(msgr->cct,2) << "writer couldn't write keepalive[2], "
			     << cpp_strerror(errno) << dendl;
	  fault();
 	  continue;
	}
	send_keepalive = false;
      }
      if (send_keepalive_ack) {
	utime_t t = keepalive_ack_stamp;
	pipe_lock.Unlock();
	int rc = write_keepalive2(CEPH_MSGR_TAG_KEEPALIVE2_ACK, t);
	pipe_lock.Lock();
	if (rc < 0) {
	  ldout(msgr->cct,2) << "writer couldn't write keepalive_ack, " << cpp_strerror(errno) << dendl;
	  fault();
	  continue;
	}
	send_keepalive_ack = false;
      }

      // send ack?
      if (in_seq > in_seq_acked) {
	uint64_t send_seq = in_seq;
	pipe_lock.Unlock();
	int rc = write_ack(send_seq);
	pipe_lock.Lock();
	if (rc < 0) {
	  ldout(msgr->cct,2) << "writer couldn't write ack, " << cpp_strerror(errno) << dendl;
	  fault();
 	  continue;
	}
	in_seq_acked = send_seq;
      }

      // grab outgoing message
      Message *m = _get_next_outgoing();
      if (m) {
	m->set_seq(++out_seq);
	if (!policy.lossy) {
	  // put on sent list
	  sent.push_back(m); 
	  m->get();
	}

	// associate message with Connection (for benefit of encode_payload)
	m->set_connection(connection_state.get());

	uint64_t features = connection_state->get_features();

	if (m->empty_payload())
	  ldout(msgr->cct,20) << "writer encoding " << m->get_seq() << " features " << features
			      << " " << m << " " << *m << dendl;
	else
	  ldout(msgr->cct,20) << "writer half-reencoding " << m->get_seq() << " features " << features
			      << " " << m << " " << *m << dendl;

	// encode and copy out of *m
	m->encode(features, msgr->crcflags);

	// prepare everything
	const ceph_msg_header& header = m->get_header();
	const ceph_msg_footer& footer = m->get_footer();

	// Now that we have all the crcs calculated, handle the
	// digital signature for the message, if the pipe 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(msgr->cct, 20) << "writer no session security" << dendl;
	} else {
	  if (session_security->sign_message(m)) {
	    ldout(msgr->cct, 20) << "writer failed to sign seq # " << header.seq
				 << "): sig = " << footer.sig << dendl;
	  } else {
	    ldout(msgr->cct, 20) << "writer signed seq # " << header.seq
				 << "): sig = " << footer.sig << dendl;
	  }
	}

	bufferlist blist = m->get_payload();
	blist.append(m->get_middle());
	blist.append(m->get_data());

        pipe_lock.Unlock();

        m->trace.event("pipe writing message");

        ldout(msgr->cct,20) << "writer sending " << m->get_seq() << " " << m << dendl;
	int rc = write_message(header, footer, blist);

	pipe_lock.Lock();
	if (rc < 0) {
          ldout(msgr->cct,1) << "writer error sending " << m << ", "
		  << cpp_strerror(errno) << dendl;
	  fault();
        }
	m->put();
      }
      continue;
    }
    
    // wait
    ldout(msgr->cct,20) << "writer sleeping" << dendl;
    cond.Wait(pipe_lock);
  }
  
  ldout(msgr->cct,20) << "writer finishing" << dendl;

  // reap?
  writer_running = false;
  unlock_maybe_reap();
  ldout(msgr->cct,10) << "writer done" << dendl;
}