Method entry

src/msg/DispatchQueue.cc:158–216

* This function delivers incoming messages to the Messenger. * Connections with messages are kept in queues; when beginning a message * delivery the highest-priority queue is selected, the connection from the * front of the queue is removed, and its message read. If the connection * has remaining messages at that priority level, it is re-placed on to the * end of the queue. If the queue is em

 * The message is then delivered and the process starts again.
 */
void DispatchQueue::entry()
{
  std::unique_lock l{lock};
  while (true) {
    while (!mqueue.empty()) {
      QueueItem qitem = mqueue.dequeue();
      if (!qitem.is_code())
	remove_arrival(qitem);
      l.unlock();

      if (qitem.is_code()) {
	if (cct->_conf->ms_inject_internal_delays &&
	    cct->_conf->ms_inject_delay_probability &&
	    (rand() % 10000)/10000.0 < cct->_conf->ms_inject_delay_probability) {
	  utime_t t;
	  t.set_from_double(cct->_conf->ms_inject_internal_delays);
	  ldout(cct, 1) << "DispatchQueue::entry  inject delay of " << t
			<< dendl;
	  t.sleep();
	}
	switch (qitem.get_code()) {
	case D_BAD_REMOTE_RESET:
	  msgr->ms_deliver_handle_remote_reset(qitem.get_connection());
	  break;
	case D_CONNECT:
	  msgr->ms_deliver_handle_connect(qitem.get_connection());
	  break;
	case D_ACCEPT:
	  msgr->ms_deliver_handle_accept(qitem.get_connection());
	  break;
	case D_BAD_RESET:
	  msgr->ms_deliver_handle_reset(qitem.get_connection());
	  break;
	case D_CONN_REFUSED:
	  msgr->ms_deliver_handle_refused(qitem.get_connection());
	  break;
	default:
	  ceph_abort();
	}
      } else {
	const ref_t<Message>& m = qitem.get_message();
	if (stop) {
	  ldout(cct,10) << " stop flag set, discarding " << m << " " << *m << dendl;
	} else {
	  uint64_t msize = pre_dispatch(m);
	  msgr->ms_deliver_dispatch(m);
	  post_dispatch(m, msize);
	}
      }

      l.lock();
    }
    if (stop)
      break;

    // wait for something to be put on queue
    cond.wait(l);
  }