void dev_poll_reactor::run()

in demo_example/asio/asio/detail/impl/dev_poll_reactor.ipp [257:380]
101 lines of code
26 McCabe index (conditional complexity)

void dev_poll_reactor::run(long usec, op_queue<operation>& ops)
{
  asio::detail::mutex::scoped_lock lock(mutex_);

  // We can return immediately if there's no work to do and the reactor is
  // not supposed to block.
  if (usec == 0 && op_queue_[read_op].empty() && op_queue_[write_op].empty()
      && op_queue_[except_op].empty() && timer_queues_.all_empty())
    return;

  // Write the pending event registration changes to the /dev/poll descriptor.
  std::size_t events_size = sizeof(::pollfd) * pending_event_changes_.size();
  if (events_size > 0)
  {
    errno = 0;
    int result = ::write(dev_poll_fd_,
        &pending_event_changes_[0], events_size);
    if (result != static_cast<int>(events_size))
    {
      asio::error_code ec = asio::error_code(
          errno, asio::error::get_system_category());
      for (std::size_t i = 0; i < pending_event_changes_.size(); ++i)
      {
        int descriptor = pending_event_changes_[i].fd;
        for (int j = 0; j < max_ops; ++j)
          op_queue_[j].cancel_operations(descriptor, ops, ec);
      }
    }
    pending_event_changes_.clear();
    pending_event_change_index_.clear();
  }

  // Calculate timeout.
  int timeout;
  if (usec == 0)
    timeout = 0;
  else
  {
    timeout = (usec < 0) ? -1 : ((usec - 1) / 1000 + 1);
    timeout = get_timeout(timeout);
  }
  lock.unlock();

  // Block on the /dev/poll descriptor.
  ::pollfd events[128] = { { 0, 0, 0 } };
  ::dvpoll dp = { 0, 0, 0 };
  dp.dp_fds = events;
  dp.dp_nfds = 128;
  dp.dp_timeout = timeout;
  int num_events = ::ioctl(dev_poll_fd_, DP_POLL, &dp);

  lock.lock();

  // Dispatch the waiting events.
  for (int i = 0; i < num_events; ++i)
  {
    int descriptor = events[i].fd;
    if (descriptor == interrupter_.read_descriptor())
    {
      interrupter_.reset();
    }
    else
    {
      bool more_reads = false;
      bool more_writes = false;
      bool more_except = false;

      // Exception operations must be processed first to ensure that any
      // out-of-band data is read before normal data.
      if (events[i].events & (POLLPRI | POLLERR | POLLHUP))
        more_except =
          op_queue_[except_op].perform_operations(descriptor, ops);
      else
        more_except = op_queue_[except_op].has_operation(descriptor);

      if (events[i].events & (POLLIN | POLLERR | POLLHUP))
        more_reads = op_queue_[read_op].perform_operations(descriptor, ops);
      else
        more_reads = op_queue_[read_op].has_operation(descriptor);

      if (events[i].events & (POLLOUT | POLLERR | POLLHUP))
        more_writes = op_queue_[write_op].perform_operations(descriptor, ops);
      else
        more_writes = op_queue_[write_op].has_operation(descriptor);

      if ((events[i].events & (POLLERR | POLLHUP)) != 0
            && !more_except && !more_reads && !more_writes)
      {
        // If we have an event and no operations associated with the
        // descriptor then we need to delete the descriptor from /dev/poll.
        // The poll operation can produce POLLHUP or POLLERR events when there
        // is no operation pending, so if we do not remove the descriptor we
        // can end up in a tight polling loop.
        ::pollfd ev = { 0, 0, 0 };
        ev.fd = descriptor;
        ev.events = POLLREMOVE;
        ev.revents = 0;
        ::write(dev_poll_fd_, &ev, sizeof(ev));
      }
      else
      {
        ::pollfd ev = { 0, 0, 0 };
        ev.fd = descriptor;
        ev.events = POLLERR | POLLHUP;
        if (more_reads)
          ev.events |= POLLIN;
        if (more_writes)
          ev.events |= POLLOUT;
        if (more_except)
          ev.events |= POLLPRI;
        ev.revents = 0;
        int result = ::write(dev_poll_fd_, &ev, sizeof(ev));
        if (result != sizeof(ev))
        {
          asio::error_code ec(errno,
              asio::error::get_system_category());
          for (int j = 0; j < max_ops; ++j)
            op_queue_[j].cancel_operations(descriptor, ops, ec);
        }
      }
    }
  }
  timer_queues_.get_ready_timers(ops);
}