void perform_operations_for_descriptors(
      const Descriptor_Set& descriptors, op_queue<operation>& ops)
  {
    typename operations_map::iterator i = operations_.begin();
    while (i != operations_.end())
    {
      typename operations_map::iterator op_iter = i++;
      if (descriptors.is_set(op_iter->first))
      {
        while (reactor_op* op = op_iter->second.op_queue_.front())
        {
          if (op->perform())
          {
            op_iter->second.op_queue_.pop();
            ops.push(op);
          }
          else
          {
            break;
          }
        }

        if (op_iter->second.op_queue_.empty())
          operations_.erase(op_iter);
      }
    }
  }

 // Get all operations owned by the queue.
 void get_all_operations(op_queue<operation>& ops)
 {
   typename operations_map::iterator i = operations_.begin();
   while (i != operations_.end())
   {
     typename operations_map::iterator op_iter = i++;
     ops.push(op_iter->second.op_queue_);
     operations_.erase(op_iter);
   }
 }

 // Dequeue all timers not later than the current time.
 virtual void get_ready_timers(op_queue<operation>& ops)
 {
   const time_type now = Time_Traits::now();
   while (!heap_.empty() && !Time_Traits::less_than(now, heap_[0].time_))
   {
     per_timer_data* timer = heap_[0].timer_;
     ops.push(timer->op_queue_);
     remove_timer(*timer);
   }
 }

  // Dequeue all timers.
  virtual void get_all_timers(op_queue<operation>& ops)
  {
    while (timers_)
    {
      per_timer_data* timer = timers_;
      timers_ = timers_->next_;
      ops.push(timer->op_queue_);
      timer->next_ = 0;
      timer->prev_ = 0;
    }

    heap_.clear();
  }

  // Cancel all operations associated with the descriptor identified by the
  // supplied iterator. Any operations pending for the descriptor will be
  // cancelled. Returns true if any operations were cancelled, in which case
  // the reactor's event demultiplexing function may need to be interrupted and
  // restarted.
  bool cancel_operations(iterator i, op_queue<operation>& ops,
      const boost::system::error_code& ec =
        boost::asio::error::operation_aborted)
  {
    if (i != operations_.end())
    {
      while (reactor_op* op = i->second.front())
      {
        op->ec_ = ec;
        i->second.pop();
        ops.push(op);
      }
      operations_.erase(i);
      return true;
    }

    return false;
  }

 // Cancel and dequeue operations for the given timer.
 std::size_t cancel_timer(per_timer_data& timer, op_queue<operation>& ops,
     std::size_t max_cancelled = (std::numeric_limits<std::size_t>::max)())
 {
   std::size_t num_cancelled = 0;
   if (timer.prev_ != 0 || &timer == timers_)
   {
     while (wait_op* op = (num_cancelled != max_cancelled)
         ? timer.op_queue_.front() : 0)
     {
       op->ec_ = asio::error::operation_aborted;
       timer.op_queue_.pop();
       ops.push(op);
       ++num_cancelled;
     }
     if (timer.op_queue_.empty())
       remove_timer(timer);
   }
   return num_cancelled;
 }

  // Cancel all operations associated with the descriptor. Any operations
  // pending for the descriptor will be notified that they have been cancelled
  // next time perform_cancellations is called. Returns true if any operations
  // were cancelled, in which case the reactor's event demultiplexing function
  // may need to be interrupted and restarted.
  bool cancel_operations(Descriptor descriptor, op_queue<operation>& ops,
      const asio::error_code& ec =
        asio::error::operation_aborted)
  {
    typename operations_map::iterator i = operations_.find(descriptor);
    if (i != operations_.end())
    {
      while (reactor_op* op = i->second.op_queue_.front())
      {
        op->ec_ = ec;
        i->second.op_queue_.pop();
        ops.push(op);
      }
      operations_.erase(i);
      return true;
    }

    return false;
  }

 // Perform the operations corresponding to the descriptor identified by the
 // supplied iterator. Returns true if there are still unfinished operations
 // queued for the descriptor.
 bool perform_operations(iterator i, op_queue<operation>& ops)
 {
   if (i != operations_.end())
   {
     while (reactor_op* op = i->second.front())
     {
       if (op->perform())
       {
         i->second.pop();
         ops.push(op);
       }
       else
       {
         return true;
       }
     }
     operations_.erase(i);
   }
   return false;
 }

 // Perform the operations corresponding to the descriptor. Returns true if
 // there are still unfinished operations queued for the descriptor.
 bool perform_operations(Descriptor descriptor, op_queue<operation>& ops)
 {
   typename operations_map::iterator i = operations_.find(descriptor);
   if (i != operations_.end())
   {
     while (reactor_op* op = i->second.op_queue_.front())
     {
       if (op->perform())
       {
         i->second.op_queue_.pop();
         ops.push(op);
       }
       else
       {
         return true;
       }
     }
     operations_.erase(i);
   }
   return false;
 }