WFMO_Reactor.cpp

Go to the documentation of this file.

// WFMO_Reactor.cpp,v 4.110 2006/06/07 03:03:54 schmidt Exp

#include "ace/WFMO_Reactor.h"

#if defined (ACE_WIN32)

#include "ace/Handle_Set.h"
#include "ace/Timer_Heap.h"
#include "ace/Thread.h"
#include "ace/OS_NS_errno.h"
#include "ace/Null_Condition.h"

#if !defined (__ACE_INLINE__)
#include "ace/WFMO_Reactor.inl"
#endif /* __ACE_INLINE__ */

ACE_RCSID(ace, WFMO_Reactor, "WFMO_Reactor.cpp,v 4.110 2006/06/07 03:03:54 schmidt Exp")

#include "ace/Auto_Ptr.h"

ACE_BEGIN_VERSIONED_NAMESPACE_DECL

ACE_WFMO_Reactor_Handler_Repository::ACE_WFMO_Reactor_Handler_Repository (ACE_WFMO_Reactor &wfmo_reactor)
  : wfmo_reactor_ (wfmo_reactor)
{
}

int
ACE_WFMO_Reactor_Handler_Repository::open (size_t size)
{
  if (size > MAXIMUM_WAIT_OBJECTS)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_LIB_TEXT ("%d exceeds MAXIMUM_WAIT_OBJECTS (%d)\n"),
                       size,
                       MAXIMUM_WAIT_OBJECTS),
                      -1);

  // Dynamic allocation
  ACE_NEW_RETURN (this->current_handles_,
                  ACE_HANDLE[size],
                  -1);
  ACE_NEW_RETURN (this->current_info_,
                  Current_Info[size],
                  -1);
  ACE_NEW_RETURN (this->current_suspended_info_,
                  Suspended_Info[size],
                  -1);
  ACE_NEW_RETURN (this->to_be_added_info_,
                  To_Be_Added_Info[size],
                  -1);

  // Initialization
  this->max_size_ = size;
  this->max_handlep1_ = 0;
  this->suspended_handles_ = 0;
  this->handles_to_be_added_ = 0;
  this->handles_to_be_deleted_ = 0;
  this->handles_to_be_suspended_ = 0;
  this->handles_to_be_resumed_ = 0;

  for (size_t i = 0; i < size; ++i)
    this->current_handles_[i] = ACE_INVALID_HANDLE;

  return 0;
}

ACE_WFMO_Reactor_Handler_Repository::~ACE_WFMO_Reactor_Handler_Repository (void)
{
  // Free up dynamically allocated space
  delete [] this->current_handles_;
  delete [] this->current_info_;
  delete [] this->current_suspended_info_;
  delete [] this->to_be_added_info_;
}

ACE_Reactor_Mask
ACE_WFMO_Reactor_Handler_Repository::bit_ops (long &existing_masks,
                                              ACE_Reactor_Mask change_masks,
                                              int operation)
{
  // Find the old reactor masks.  This automatically does the work of
  // the GET_MASK operation.

  ACE_Reactor_Mask old_masks = ACE_Event_Handler::NULL_MASK;

  if (ACE_BIT_ENABLED (existing_masks, FD_READ)
      || ACE_BIT_ENABLED (existing_masks, FD_CLOSE))
    ACE_SET_BITS (old_masks, ACE_Event_Handler::READ_MASK);

  if (ACE_BIT_ENABLED (existing_masks, FD_WRITE))
    ACE_SET_BITS (old_masks, ACE_Event_Handler::WRITE_MASK);

  if (ACE_BIT_ENABLED (existing_masks, FD_OOB))
    ACE_SET_BITS (old_masks, ACE_Event_Handler::EXCEPT_MASK);

  if (ACE_BIT_ENABLED (existing_masks, FD_ACCEPT))
    ACE_SET_BITS (old_masks, ACE_Event_Handler::ACCEPT_MASK);

  if (ACE_BIT_ENABLED (existing_masks, FD_CONNECT))
    ACE_SET_BITS (old_masks, ACE_Event_Handler::CONNECT_MASK);

  if (ACE_BIT_ENABLED (existing_masks, FD_QOS))
    ACE_SET_BITS (old_masks, ACE_Event_Handler::QOS_MASK);

  if (ACE_BIT_ENABLED (existing_masks, FD_GROUP_QOS))
    ACE_SET_BITS (old_masks, ACE_Event_Handler::GROUP_QOS_MASK);

  switch (operation)
    {
    case ACE_Reactor::CLR_MASK:
      // For the CLR_MASK operation, clear only the specific masks.

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::READ_MASK))
        {
          ACE_CLR_BITS (existing_masks, FD_READ);
          ACE_CLR_BITS (existing_masks, FD_CLOSE);
        }

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::WRITE_MASK))
        ACE_CLR_BITS (existing_masks, FD_WRITE);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::EXCEPT_MASK))
        ACE_CLR_BITS (existing_masks, FD_OOB);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::ACCEPT_MASK))
        ACE_CLR_BITS (existing_masks, FD_ACCEPT);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::CONNECT_MASK))
        ACE_CLR_BITS (existing_masks, FD_CONNECT);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::QOS_MASK))
        ACE_CLR_BITS (existing_masks, FD_QOS);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::GROUP_QOS_MASK))
        ACE_CLR_BITS (existing_masks, FD_GROUP_QOS);

      break;

    case ACE_Reactor::SET_MASK:
      // If the operation is a set, first reset any existing masks

      existing_masks = 0;
      /* FALLTHRU */

    case ACE_Reactor::ADD_MASK:
      // For the ADD_MASK and the SET_MASK operation, add only the
      // specific masks.

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::READ_MASK))
        {
          ACE_SET_BITS (existing_masks, FD_READ);
          ACE_SET_BITS (existing_masks, FD_CLOSE);
        }

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::WRITE_MASK))
        ACE_SET_BITS (existing_masks, FD_WRITE);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::EXCEPT_MASK))
        ACE_SET_BITS (existing_masks, FD_OOB);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::ACCEPT_MASK))
        ACE_SET_BITS (existing_masks, FD_ACCEPT);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::CONNECT_MASK))
        ACE_SET_BITS (existing_masks, FD_CONNECT);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::QOS_MASK))
        ACE_SET_BITS (existing_masks, FD_QOS);

      if (ACE_BIT_ENABLED (change_masks, ACE_Event_Handler::GROUP_QOS_MASK))
        ACE_SET_BITS (existing_masks, FD_GROUP_QOS);

      break;

    case ACE_Reactor::GET_MASK:

      // The work for this operation is done in all cases at the
      // begining of the function.

      ACE_UNUSED_ARG (change_masks);

      break;
    }

  return old_masks;
}

int
ACE_WFMO_Reactor_Handler_Repository::unbind_i (ACE_HANDLE handle,
                                               ACE_Reactor_Mask mask,
                                               int &changes_required)
{
  int error = 0;

  // Remember this value; only if it changes do we need to wakeup
  // the other threads
  size_t const original_handle_count = this->handles_to_be_deleted_;
  int result = 0;
  size_t i;

  // Go through all the handles looking for <handle>.  Even if we find
  // it, we continue through the rest of the list since <handle> could
  // appear multiple times. All handles are checked.

  // First check the current entries
  for (i = 0; i < this->max_handlep1_ && error == 0; ++i)
    // Since the handle can either be the event or the I/O handle,
    // we have to check both
    if ((this->current_handles_[i] == handle
         || this->current_info_[i].io_handle_ == handle)
        && // Make sure that it is not already marked for deleted
        !this->current_info_[i].delete_entry_)
      {
        result = this->remove_handler_i (i,
                                         mask);
        if (result == -1)
          error = 1;
      }

  // Then check the suspended entries
  for (i = 0; i < this->suspended_handles_ && error == 0; ++i)
    // Since the handle can either be the event or the I/O handle, we
    // have to check both
    if ((this->current_suspended_info_[i].io_handle_ == handle
         || this->current_suspended_info_[i].event_handle_ == handle)
        &&
        // Make sure that it is not already marked for deleted
        !this->current_suspended_info_[i].delete_entry_)
      {
        result = this->remove_suspended_handler_i (i,
                                                   mask);
        if (result == -1)
          error = 1;
      }

  // Then check the to_be_added entries
  for (i = 0; i < this->handles_to_be_added_ && error == 0; ++i)
    // Since the handle can either be the event or the I/O handle,
    // we have to check both
    if ((this->to_be_added_info_[i].io_handle_ == handle
         || this->to_be_added_info_[i].event_handle_ == handle)
        &&
        // Make sure that it is not already marked for deleted
        !this->to_be_added_info_[i].delete_entry_)
      {
        result = this->remove_to_be_added_handler_i (i,
                                                     mask);
        if (result == -1)
          error = 1;
      }

  // Only if the number of handlers to be deleted changes do we need
  // to wakeup the other threads
  if (original_handle_count < this->handles_to_be_deleted_)
    changes_required = 1;

  return error ? -1 : 0;
}

int
ACE_WFMO_Reactor_Handler_Repository::remove_handler_i (size_t slot,
                                                       ACE_Reactor_Mask to_be_removed_masks)
{
  // I/O entries
  if (this->current_info_[slot].io_entry_)
    {
      // See if there are other events that the <Event_Handler> is
      // interested in
      this->bit_ops (this->current_info_[slot].network_events_,
                     to_be_removed_masks,
                     ACE_Reactor::CLR_MASK);

      // Disassociate/Reassociate the event from/with the I/O handle.
      // This will depend on the value of remaining set of network
      // events that the <event_handler> is interested in. I don't
      // think we can do anything about errors here, so I will not
      // check this.
      ::WSAEventSelect ((SOCKET) this->current_info_[slot].io_handle_,
                        this->current_handles_[slot],
                        this->current_info_[slot].network_events_);
    }
  // Normal event entries.
  else if (ACE_BIT_ENABLED (to_be_removed_masks, ACE_Event_Handler::DONT_CALL))
    // Preserve DONT_CALL
    to_be_removed_masks = ACE_Event_Handler::DONT_CALL;
  else
    // Make sure that the <to_be_removed_masks> is the NULL_MASK
    to_be_removed_masks = ACE_Event_Handler::NULL_MASK;

  // If this event was marked for suspension, undo the suspension flag
  // and reduce the to be suspended count.
  if (this->current_info_[slot].suspend_entry_)
    {
      // Undo suspension
      this->current_info_[slot].suspend_entry_ = 0;
      // Decrement the handle count
      --this->handles_to_be_suspended_;
    }

  // If there are no more events that the <Event_Handler> is
  // interested in, or this is a non-I/O entry, schedule the
  // <Event_Handler> for removal
  if (this->current_info_[slot].network_events_ == 0)
    {
      // Mark to be deleted
      this->current_info_[slot].delete_entry_ = 1;
      // Remember the mask
      this->current_info_[slot].close_masks_ = to_be_removed_masks;
      // Increment the handle count
      ++this->handles_to_be_deleted_;
    }

  // Since it is not a complete removal, we'll call handle_close
  // for all the masks that were removed.  This does not change
  // the internal state of the reactor.
  //
  // Note: this condition only applies to I/O entries
  else if (ACE_BIT_ENABLED (to_be_removed_masks, ACE_Event_Handler::DONT_CALL) == 0)
    {
      ACE_HANDLE handle = this->current_info_[slot].io_handle_;
      this->current_info_[slot].event_handler_->handle_close (handle,
                                                              to_be_removed_masks);
    }

  return 0;
}

int
ACE_WFMO_Reactor_Handler_Repository::remove_suspended_handler_i (size_t slot,
                                                                 ACE_Reactor_Mask to_be_removed_masks)
{
  // I/O entries
  if (this->current_suspended_info_[slot].io_entry_)
    {
      // See if there are other events that the <Event_Handler> is
      // interested in
      this->bit_ops (this->current_suspended_info_[slot].network_events_,
                     to_be_removed_masks,
                     ACE_Reactor::CLR_MASK);

      // Disassociate/Reassociate the event from/with the I/O handle.
      // This will depend on the value of remaining set of network
      // events that the <event_handler> is interested in. I don't
      // think we can do anything about errors here, so I will not
      // check this.
      ::WSAEventSelect ((SOCKET) this->current_suspended_info_[slot].io_handle_,
                        this->current_suspended_info_[slot].event_handle_,
                        this->current_suspended_info_[slot].network_events_);
    }
  // Normal event entries.
  else if (ACE_BIT_ENABLED (to_be_removed_masks, ACE_Event_Handler::DONT_CALL))
    // Preserve DONT_CALL
    to_be_removed_masks = ACE_Event_Handler::DONT_CALL;
  else
    // Make sure that the <to_be_removed_masks> is the NULL_MASK
    to_be_removed_masks = ACE_Event_Handler::NULL_MASK;

  // If this event was marked for resumption, undo the resumption flag
  // and reduce the to be resumed count.
  if (this->current_suspended_info_[slot].resume_entry_)
    {
      // Undo resumption
      this->current_suspended_info_[slot].resume_entry_ = 0;
      // Decrement the handle count
      --this->handles_to_be_resumed_;
    }

  // If there are no more events that the <Event_Handler> is
  // interested in, or this is a non-I/O entry, schedule the
  // <Event_Handler> for removal
  if (this->current_suspended_info_[slot].network_events_ == 0)
    {
      // Mark to be deleted
      this->current_suspended_info_[slot].delete_entry_ = 1;
      // Remember the mask
      this->current_suspended_info_[slot].close_masks_ = to_be_removed_masks;
      // Increment the handle count
      ++this->handles_to_be_deleted_;
    }
  // Since it is not a complete removal, we'll call handle_close for
  // all the masks that were removed.  This does not change the
  // internal state of the reactor.
  //
  // Note: this condition only applies to I/O entries
  else if (ACE_BIT_ENABLED (to_be_removed_masks, ACE_Event_Handler::DONT_CALL) == 0)
    {
      ACE_HANDLE handle = this->current_suspended_info_[slot].io_handle_;
      this->current_suspended_info_[slot].event_handler_->handle_close (handle,
                                                                        to_be_removed_masks);
    }

  return 0;
}

int
ACE_WFMO_Reactor_Handler_Repository::remove_to_be_added_handler_i (size_t slot,
                                                                   ACE_Reactor_Mask to_be_removed_masks)
{
  // I/O entries
  if (this->to_be_added_info_[slot].io_entry_)
    {
      // See if there are other events that the <Event_Handler> is
      // interested in
      this->bit_ops (this->to_be_added_info_[slot].network_events_,
                     to_be_removed_masks,
                     ACE_Reactor::CLR_MASK);

      // Disassociate/Reassociate the event from/with the I/O handle.
      // This will depend on the value of remaining set of network
      // events that the <event_handler> is interested in. I don't
      // think we can do anything about errors here, so I will not
      // check this.
      ::WSAEventSelect ((SOCKET) this->to_be_added_info_[slot].io_handle_,
                        this->to_be_added_info_[slot].event_handle_,
                        this->to_be_added_info_[slot].network_events_);
    }
  // Normal event entries.
  else if (ACE_BIT_ENABLED (to_be_removed_masks, ACE_Event_Handler::DONT_CALL))
    // Preserve DONT_CALL
    to_be_removed_masks = ACE_Event_Handler::DONT_CALL;
  else
    // Make sure that the <to_be_removed_masks> is the NULL_MASK
    to_be_removed_masks = ACE_Event_Handler::NULL_MASK;

  // If this event was marked for suspension, undo the suspension flag
  // and reduce the to be suspended count.
  if (this->to_be_added_info_[slot].suspend_entry_)
    {
      // Undo suspension
      this->to_be_added_info_[slot].suspend_entry_ = 0;
      // Decrement the handle count
      --this->handles_to_be_suspended_;
    }

  // If there are no more events that the <Event_Handler> is
  // interested in, or this is a non-I/O entry, schedule the
  // <Event_Handler> for removal
  if (this->to_be_added_info_[slot].network_events_ == 0)
    {
      // Mark to be deleted
      this->to_be_added_info_[slot].delete_entry_ = 1;
      // Remember the mask
      this->to_be_added_info_[slot].close_masks_ = to_be_removed_masks;
      // Increment the handle count
      ++this->handles_to_be_deleted_;
    }
  // Since it is not a complete removal, we'll call handle_close
  // for all the masks that were removed.  This does not change
  // the internal state of the reactor.
  //
  // Note: this condition only applies to I/O entries
  else if (ACE_BIT_ENABLED (to_be_removed_masks, ACE_Event_Handler::DONT_CALL) == 0)
    {
      ACE_HANDLE handle = this->to_be_added_info_[slot].io_handle_;
      this->to_be_added_info_[slot].event_handler_->handle_close (handle,
                                                                  to_be_removed_masks);
    }

  return 0;
}

int
ACE_WFMO_Reactor_Handler_Repository::suspend_handler_i (ACE_HANDLE handle,
                                                        int &changes_required)
{
  size_t i = 0;

  // Go through all the handles looking for <handle>.  Even if we find
  // it, we continue through the rest of the list since <handle> could
  // appear multiple times. All handles are checked.

  // Check the current entries first.
  for (i = 0; i < this->max_handlep1_; ++i)
    // Since the handle can either be the event or the I/O handle,
    // we have to check both
    if ((this->current_handles_[i] == handle ||
         this->current_info_[i].io_handle_ == handle) &&
        // Make sure that it is not already marked for suspension
        !this->current_info_[i].suspend_entry_)
      {
        // Mark to be suspended
        this->current_info_[i].suspend_entry_ = 1;
        // Increment the handle count
        ++this->handles_to_be_suspended_;
        // Changes will be required
        changes_required = 1;
      }

  // Then check the suspended entries.
  for (i = 0; i < this->suspended_handles_; ++i)
    // Since the handle can either be the event or the I/O handle,
    // we have to check both
    if ((this->current_suspended_info_[i].event_handle_ == handle ||
         this->current_suspended_info_[i].io_handle_ == handle) &&
        // Make sure that the resumption is not already undone
        this->current_suspended_info_[i].resume_entry_)
      {
        // Undo resumption
        this->current_suspended_info_[i].resume_entry_ = 0;
        // Decrement the handle count
        --this->handles_to_be_resumed_;
        // Changes will be required
        changes_required = 1;
      }

  // Then check the to_be_added entries.
  for (i = 0; i < this->handles_to_be_added_; ++i)
    // Since the handle can either be the event or the I/O handle,
    // we have to check both
    if ((this->to_be_added_info_[i].io_handle_ == handle ||
         this->to_be_added_info_[i].event_handle_ == handle) &&
        // Make sure that it is not already marked for suspension
        !this->to_be_added_info_[i].suspend_entry_)
      {
        // Mark to be suspended
        this->to_be_added_info_[i].suspend_entry_ = 1;
        // Increment the handle count
        ++this->handles_to_be_suspended_;
        // Changes will be required
        changes_required = 1;
      }

  return 0;
}

int
ACE_WFMO_Reactor_Handler_Repository::resume_handler_i (ACE_HANDLE handle,
                                                       int &changes_required)
{
  size_t i = 0;

  // Go through all the handles looking for <handle>.  Even if we find
  // it, we continue through the rest of the list since <handle> could
  // appear multiple times. All handles are checked.

  // Check the current entries first.
  for (i = 0; i < this->max_handlep1_; ++i)
    // Since the handle can either be the event or the I/O handle,
    // we have to check both
    if ((this->current_handles_[i] == handle ||
         this->current_info_[i].io_handle_ == handle) &&
        // Make sure that the suspension is not already undone
        this->current_info_[i].suspend_entry_)
      {
        // Undo suspension
        this->current_info_[i].suspend_entry_ = 0;
        // Decrement the handle count
        --this->handles_to_be_suspended_;
        // Changes will be required
        changes_required = 1;
      }

  // Then check the suspended entries.
  for (i = 0; i < this->suspended_handles_; ++i)
    // Since the handle can either be the event or the I/O handle,
    // we have to check both
    if ((this->current_suspended_info_[i].event_handle_ == handle ||
         this->current_suspended_info_[i].io_handle_ == handle) &&
        // Make sure that it is not already marked for resumption
        !this->current_suspended_info_[i].resume_entry_)
      {
        // Mark to be resumed
        this->current_suspended_info_[i].resume_entry_ = 1;
        // Increment the handle count
        ++this->handles_to_be_resumed_;
        // Changes will be required
        changes_required = 1;
      }

  // Then check the to_be_added entries.
  for (i = 0; i < this->handles_to_be_added_; ++i)
    // Since the handle can either be the event or the I/O handle,
    // we have to check both
    if ((this->to_be_added_info_[i].io_handle_ == handle ||
         this->to_be_added_info_[i].event_handle_ == handle) &&
        // Make sure that the suspension is not already undone
        this->to_be_added_info_[i].suspend_entry_)
      {
        // Undo suspension
        this->to_be_added_info_[i].suspend_entry_ = 0;
        // Decrement the handle count
        --this->handles_to_be_suspended_;
        // Changes will be required
        changes_required = 1;
      }

  return 0;
}

void
ACE_WFMO_Reactor_Handler_Repository::unbind_all (void)
{
  {
    ACE_GUARD (ACE_Process_Mutex, ace_mon, this->wfmo_reactor_.lock_);

    int dummy;
    size_t i;

    // Remove all the current handlers
    for (i = 0; i < this->max_handlep1_; ++i)
      this->unbind_i (this->current_handles_[i],
                      ACE_Event_Handler::ALL_EVENTS_MASK,
                      dummy);

    // Remove all the suspended handlers
    for (i = 0; i < this->suspended_handles_; ++i)
      this->unbind_i (this->current_suspended_info_[i].event_handle_,
                      ACE_Event_Handler::ALL_EVENTS_MASK,
                      dummy);

    // Remove all the to_be_added handlers
    for (i = 0; i < this->handles_to_be_added_; ++i)
      this->unbind_i (this->to_be_added_info_[i].event_handle_,
                      ACE_Event_Handler::ALL_EVENTS_MASK,
                      dummy);

  }

  // The guard is released here

  // Wake up all threads in WaitForMultipleObjects so that they can
  // reconsult the handle set
  this->wfmo_reactor_.wakeup_all_threads ();
}

int
ACE_WFMO_Reactor_Handler_Repository::bind_i (int io_entry,
                                             ACE_Event_Handler *event_handler,
                                             long network_events,
                                             ACE_HANDLE io_handle,
                                             ACE_HANDLE event_handle,
                                             int delete_event)
{
  // Make sure that the <handle> is valid
  if (event_handle == ACE_INVALID_HANDLE)
    event_handle = event_handler->get_handle ();
  if (this->invalid_handle (event_handle))
    return -1;

  size_t current_size = this->max_handlep1_ +
    this->handles_to_be_added_ -
    this->handles_to_be_deleted_ +
    this->suspended_handles_;

  // Make sure that there's room in the table and that total pending
  // additions should not exceed what the <to_be_added_info_> array
  // can hold.
  if (current_size < this->max_size_ &&
      this->handles_to_be_added_ < this->max_size_)
    {
      // Cache this set into the <to_be_added_info_>, till we come
      // around to actually adding this to the <current_info_>
      this->to_be_added_info_[this->handles_to_be_added_].set (event_handle,
                                                               io_entry,
                                                               event_handler,
                                                               io_handle,
                                                               network_events,
                                                               delete_event);

      ++this->handles_to_be_added_;

      event_handler->add_reference ();

      // Wake up all threads in WaitForMultipleObjects so that they can
      // reconsult the handle set
      this->wfmo_reactor_.wakeup_all_threads ();
    }
  else
    {
      errno = EMFILE;   // File descriptor table is full (better than nothing)
      return -1;
    }

  return 0;
}

int
ACE_WFMO_Reactor_Handler_Repository::make_changes_in_current_infos (void)
{
  // Go through the entire valid array and check for all handles that
  // have been schedule for deletion
  if (this->handles_to_be_deleted_ > 0 || this->handles_to_be_suspended_ > 0)
    {
      size_t i = 0;
      while (i < this->max_handlep1_)
        {
          // This stuff is necessary here, since we should not make
          // the upcall until all the internal data structures have
          // been updated.  This is to protect against upcalls that
          // try to deregister again.
          ACE_HANDLE handle = ACE_INVALID_HANDLE;
          ACE_Reactor_Mask masks = ACE_Event_Handler::NULL_MASK;
          ACE_Event_Handler *event_handler = 0;

          // See if this entry is scheduled for deletion
          if (this->current_info_[i].delete_entry_)
            {
              // Calling the <handle_close> method here will ensure that we
              // will only call it once per deregistering <Event_Handler>.
              // This is essential in the case when the <Event_Handler> will
              // do something like delete itself and we have multiple
              // threads in WFMO_Reactor.
              //
              // Make sure that the DONT_CALL mask is not set
              masks = this->current_info_[i].close_masks_;
              if (ACE_BIT_ENABLED (masks, ACE_Event_Handler::DONT_CALL) == 0)
                {
                  // Grab the correct handle depending on the type entry
                  if (this->current_info_[i].io_entry_)
                    handle = this->current_info_[i].io_handle_;
                  else
                    handle = this->current_handles_[i];

                  // Event handler
                  event_handler = this->current_info_[i].event_handler_;
                }

              // If <WFMO_Reactor> created the event, we need to clean it up
              if (this->current_info_[i].delete_event_)
                ACE_OS::event_destroy (&this->current_handles_[i]);

              // Reduce count by one
              --this->handles_to_be_deleted_;
            }

          // See if this entry is scheduled for suspension
          else if (this->current_info_[i].suspend_entry_)
            {
              this->current_suspended_info_ [this->suspended_handles_].set (this->current_handles_[i],
                                                                            this->current_info_[i]);
              // Increase number of suspended handles
              ++this->suspended_handles_;

              // Reduce count by one
              --this->handles_to_be_suspended_;
            }

          // See if this entry is scheduled for deletion or suspension
          // If so we need to clean up
          if (this->current_info_[i].delete_entry_ ||
              this->current_info_[i].suspend_entry_  )
            {
              size_t last_valid_slot = this->max_handlep1_ - 1;
              // If this is the last handle in the set, no need to swap
              // places. Simply remove it.
              if (i < last_valid_slot)
                // Swap this handle with the last valid handle
                {
                  // Struct copy
                  this->current_info_[i] =
                    this->current_info_[last_valid_slot];
                  this->current_handles_[i] =
                    this->current_handles_[last_valid_slot];
                }
              // Reset the info in this slot
              this->current_info_[last_valid_slot].reset ();
              this->current_handles_[last_valid_slot] = ACE_INVALID_HANDLE;
              --this->max_handlep1_;
            }
          else
            {
              // This current entry is not up for deletion or
              // suspension.  Proceed to the next entry in the current
              // handles.
              ++i;
            }

          // Now that all internal structures have been updated, make
          // the upcall.
          if (event_handler != 0)
            {
              int requires_reference_counting =
                event_handler->reference_counting_policy ().value () ==
                ACE_Event_Handler::Reference_Counting_Policy::ENABLED;

              event_handler->handle_close (handle, masks);

              if (requires_reference_counting)
                {
                  event_handler->remove_reference ();
                }
            }
        }
    }

  return 0;
}

int
ACE_WFMO_Reactor_Handler_Repository::make_changes_in_suspension_infos (void)
{
  // Go through the <suspended_handle> array
  if (this->handles_to_be_deleted_ > 0 || this->handles_to_be_resumed_ > 0)
    {
      size_t i = 0;
      while (i < this->suspended_handles_)
        {
          // This stuff is necessary here, since we should not make
          // the upcall until all the internal data structures have
          // been updated.  This is to protect against upcalls that
          // try to deregister again.
          ACE_HANDLE handle = ACE_INVALID_HANDLE;
          ACE_Reactor_Mask masks = ACE_Event_Handler::NULL_MASK;
          ACE_Event_Handler *event_handler = 0;

          // See if this entry is scheduled for deletion
          if (this->current_suspended_info_[i].delete_entry_)
            {
              // Calling the <handle_close> method here will ensure that we
              // will only call it once per deregistering <Event_Handler>.
              // This is essential in the case when the <Event_Handler> will
              // do something like delete itself and we have multiple
              // threads in WFMO_Reactor.
              //
              // Make sure that the DONT_CALL mask is not set
              masks = this->current_suspended_info_[i].close_masks_;
              if (ACE_BIT_ENABLED (masks, ACE_Event_Handler::DONT_CALL) == 0)
                {
                  // Grab the correct handle depending on the type entry
                  if (this->current_suspended_info_[i].io_entry_)
                    handle = this->current_suspended_info_[i].io_handle_;
                  else
                    handle = this->current_suspended_info_[i].event_handle_;

                  // Upcall
                  event_handler = this->current_suspended_info_[i].event_handler_;
                }

              // If <WFMO_Reactor> created the event, we need to clean it up
              if (this->current_suspended_info_[i].delete_event_)
                ACE_OS::event_destroy (&this->current_suspended_info_[i].event_handle_);

              // Reduce count by one
              --this->handles_to_be_deleted_;
            }

          else if (this->current_suspended_info_[i].resume_entry_)
            {
              // Add to the end of the current handles set
              this->current_handles_[this->max_handlep1_] = this->current_suspended_info_[i].event_handle_;
              // Struct copy
              this->current_info_[this->max_handlep1_].set (this->current_suspended_info_[i]);
              ++this->max_handlep1_;

              // Reduce count by one
              --this->handles_to_be_resumed_;
            }

          // If an entry needs to be removed, either because it
          // was deleted or resumed, remove it now before doing
          // the upcall.
          if (this->current_suspended_info_[i].resume_entry_ ||
              this->current_suspended_info_[i].delete_entry_)
            {
              size_t last_valid_slot = this->suspended_handles_ - 1;
              // Net effect is that we're removing an entry and
              // compressing the list from the end.  So, if removing
              // an entry from the middle, copy the last valid one to the
              // removed slot.  Reset the end and decrement the number
              // of suspended handles.
              if (i < last_valid_slot)
                // Struct copy
                this->current_suspended_info_[i] =
                  this->current_suspended_info_[last_valid_slot];
              this->current_suspended_info_[last_valid_slot].reset ();
              --this->suspended_handles_;
            }
          else
            {
              // This current entry is not up for deletion or
              // resumption.  Proceed to the next entry in the
              // suspended handles.
              ++i;
            }

          // Now that all internal structures have been updated, make
          // the upcall.
          if (event_handler != 0)
            {
              int requires_reference_counting =
                event_handler->reference_counting_policy ().value () ==
                ACE_Event_Handler::Reference_Counting_Policy::ENABLED;

              event_handler->handle_close (handle, masks);

              if (requires_reference_counting)
                {
                  event_handler->remove_reference ();
                }
            }
        }
    }

  return 0;
}

int
ACE_WFMO_Reactor_Handler_Repository::make_changes_in_to_be_added_infos (void)
{
  // Go through the <to_be_added_*> arrays
  for (size_t i = 0; i < this->handles_to_be_added_; ++i)
    {
      // This stuff is necessary here, since we should not make
      // the upcall until all the internal data structures have
      // been updated.  This is to protect against upcalls that
      // try to deregister again.
      ACE_HANDLE handle = ACE_INVALID_HANDLE;
      ACE_Reactor_Mask masks = ACE_Event_Handler::NULL_MASK;
      ACE_Event_Handler *event_handler = 0;

      // See if this entry is scheduled for deletion
      if (this->to_be_added_info_[i].delete_entry_)
        {
          // Calling the <handle_close> method here will ensure that we
          // will only call it once per deregistering <Event_Handler>.
          // This is essential in the case when the <Event_Handler> will
          // do something like delete itself and we have multiple
          // threads in WFMO_Reactor.
          //
          // Make sure that the DONT_CALL mask is not set
          masks = this->to_be_added_info_[i].close_masks_;
          if (ACE_BIT_ENABLED (masks, ACE_Event_Handler::DONT_CALL) == 0)
            {
              // Grab the correct handle depending on the type entry
              if (this->to_be_added_info_[i].io_entry_)
                handle = this->to_be_added_info_[i].io_handle_;
              else
                handle = this->to_be_added_info_[i].event_handle_;

              // Upcall
              event_handler = this->to_be_added_info_[i].event_handler_;
            }

          // If <WFMO_Reactor> created the event, we need to clean it up
          if (this->to_be_added_info_[i].delete_event_)
            ACE_OS::event_destroy (&this->to_be_added_info_[i].event_handle_);

          // Reduce count by one
          --this->handles_to_be_deleted_;
        }

      // See if this entry is scheduled for suspension
      else if (this->to_be_added_info_[i].suspend_entry_)
        {
          this->current_suspended_info_ [this->suspended_handles_].set (this->to_be_added_info_[i].event_handle_,
                                                                        this->to_be_added_info_[i]);
          // Increase number of suspended handles
          ++this->suspended_handles_;

          // Reduce count by one
          --this->handles_to_be_suspended_;
        }

      // If neither of the two flags are on, add to current
      else
        {
          // Add to the end of the current handles set
          this->current_handles_[this->max_handlep1_] = this->to_be_added_info_[i].event_handle_;
          // Struct copy
          this->current_info_[this->max_handlep1_].set (this->to_be_added_info_[i]);
          ++this->max_handlep1_;
        }

      // Reset the <to_be_added_info_>
      this->to_be_added_info_[i].reset ();

      // Now that all internal structures have been updated, make the
      // upcall.
      if (event_handler != 0)
        {
          int requires_reference_counting =
            event_handler->reference_counting_policy ().value () ==
            ACE_Event_Handler::Reference_Counting_Policy::ENABLED;

          event_handler->handle_close (handle, masks);

          if (requires_reference_counting)
            {
              event_handler->remove_reference ();
            }
        }
    }

  // Since all to be added handles have been taken care of, reset the
  // counter
  this->handles_to_be_added_ = 0;

  return 0;
}

void
ACE_WFMO_Reactor_Handler_Repository::dump (void) const
{
#if defined (ACE_HAS_DUMP)
  size_t i = 0;

  ACE_TRACE ("ACE_WFMO_Reactor_Handler_Repository::dump");

  ACE_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("Max size = %d\n"),
              this->max_size_));

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("Current info table\n\n")));
  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("\tSize = %d\n"),
              this->max_handlep1_));
  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("\tHandles to be suspended = %d\n"),
              this->handles_to_be_suspended_));

  for (i = 0; i < this->max_handlep1_; ++i)
    this->current_info_[i].dump (this->current_handles_[i]);

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("\n")));

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("To-be-added info table\n\n")));
  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("\tSize = %d\n"),
              this->handles_to_be_added_));

  for (i = 0; i < this->handles_to_be_added_; ++i)
    this->to_be_added_info_[i].dump ();

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("\n")));

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("Suspended info table\n\n")));
  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("\tSize = %d\n"),
              this->suspended_handles_));
  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("\tHandles to be resumed = %d\n"),
              this->handles_to_be_resumed_));

  for (i = 0; i < this->suspended_handles_; ++i)
    this->current_suspended_info_[i].dump ();

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("\n")));

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("Total handles to be deleted = %d\n"),
              this->handles_to_be_deleted_));

  ACE_DEBUG ((LM_DEBUG,
              ACE_END_DUMP));
#endif /* ACE_HAS_DUMP */
}

/************************************************************/

int
ACE_WFMO_Reactor::work_pending (const ACE_Time_Value &)
{
  ACE_NOTSUP_RETURN (-1);
}

ACE_WFMO_Reactor::ACE_WFMO_Reactor (ACE_Sig_Handler *sh,
                                    ACE_Timer_Queue *tq,
                                    ACE_Reactor_Notify *notify)
  : signal_handler_ (0),
    delete_signal_handler_ (0),
    timer_queue_ (0),
    delete_timer_queue_ (0),
    delete_handler_rep_ (0),
    delete_notify_handler_ (0),
    lock_adapter_ (lock_),
    handler_rep_ (*this),
    // this event is initially signaled
    ok_to_wait_ (1),
    // this event is initially unsignaled
    wakeup_all_threads_ (0),
    // this event is initially unsignaled
    waiting_to_change_state_ (0),
    active_threads_ (0),
    owner_ (ACE_Thread::self ()),
    new_owner_ (0),
    change_state_thread_ (0),
    open_for_business_ (0),
    deactivated_ (0)
{
  if (this->open (ACE_WFMO_Reactor::DEFAULT_SIZE, 0, sh, tq, 0, notify) == -1)
    ACE_ERROR ((LM_ERROR,
                ACE_LIB_TEXT ("%p\n"),
                ACE_LIB_TEXT ("WFMO_Reactor")));
}

ACE_WFMO_Reactor::ACE_WFMO_Reactor (size_t size,
                                    int unused,
                                    ACE_Sig_Handler *sh,
                                    ACE_Timer_Queue *tq,
                                    ACE_Reactor_Notify *notify)
  : signal_handler_ (0),
    delete_signal_handler_ (0),
    timer_queue_ (0),
    delete_timer_queue_ (0),
    delete_handler_rep_ (0),
    delete_notify_handler_ (0),
    lock_adapter_ (lock_),
    handler_rep_ (*this),
    // this event is initially signaled
    ok_to_wait_ (1),
    // this event is initially unsignaled
    wakeup_all_threads_ (0),
    // this event is initially unsignaled
    waiting_to_change_state_ (0),
    active_threads_ (0),
    owner_ (ACE_Thread::self ()),
    new_owner_ (0),
    change_state_thread_ (0),
    open_for_business_ (0),
    deactivated_ (0)
{
  ACE_UNUSED_ARG (unused);

  if (this->open (size, 0, sh, tq, 0, notify) == -1)
    ACE_ERROR ((LM_ERROR,
                ACE_LIB_TEXT ("%p\n"),
                ACE_LIB_TEXT ("WFMO_Reactor")));
}

int
ACE_WFMO_Reactor::current_info (ACE_HANDLE, size_t &)
{
  return -1;
}

int
ACE_WFMO_Reactor::open (size_t size,
                        int unused,
                        ACE_Sig_Handler *sh,
                        ACE_Timer_Queue *tq,
                        int disable_notify_pipe,
                        ACE_Reactor_Notify *notify)
{
  ACE_UNUSED_ARG (unused);
  ACE_UNUSED_ARG (disable_notify_pipe);

  // This GUARD is necessary since we are updating shared state.
  ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, this->lock_, -1);

  // If we are already open, return -1
  if (this->open_for_business_)
    return -1;

  // Timer Queue
  if (this->delete_timer_queue_)
    delete this->timer_queue_;

  if (tq == 0)
    {
      ACE_NEW_RETURN (this->timer_queue_,
                      ACE_Timer_Heap,
                      -1);
      this->delete_timer_queue_ = 1;
    }
  else
    {
      this->timer_queue_ = tq;
      this->delete_timer_queue_ = 0;
    }

  // Signal Handler
  if (this->delete_signal_handler_)
    delete this->signal_handler_;

  if (sh == 0)
    {
      ACE_NEW_RETURN (this->signal_handler_,
                      ACE_Sig_Handler,
                      -1);
      this->delete_signal_handler_ = 1;
    }
  else
    {
      this->signal_handler_ = sh;
      this->delete_signal_handler_ = 0;
    }

  // Setup the atomic wait array (used later in <handle_events>)
  this->atomic_wait_array_[0] = this->lock_.lock ().proc_mutex_;
  this->atomic_wait_array_[1] = this->ok_to_wait_.handle ();

  // This is to guard against reopens of WFMO_Reactor
  if (this->delete_handler_rep_)
    this->handler_rep_.~ACE_WFMO_Reactor_Handler_Repository ();

  // Open the handle repository.  Two additional handles for internal
  // purposes
  if (this->handler_rep_.open (size + 2) == -1)
    ACE_ERROR_RETURN ((LM_ERROR, ACE_LIB_TEXT ("%p\n"),
                       ACE_LIB_TEXT ("opening handler repository")),
                      -1);
  else
    this->delete_handler_rep_ = 1;

  if (this->notify_handler_ != 0
      && this->delete_notify_handler_ != 0)
    delete this->notify_handler_;

  this->notify_handler_ = notify;

  if (this->notify_handler_ == 0)
    {
      ACE_NEW_RETURN (this->notify_handler_,
                      ACE_WFMO_Reactor_Notify,
                      -1);

      if (this->notify_handler_ == 0)
        return -1;
      else
        this->delete_notify_handler_ = 1;
    }

  /* NOTE */
  // The order of the following two registrations is very important

  // Open the notification handler
  if (this->notify_handler_->open (this, this->timer_queue_) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_LIB_TEXT ("%p\n"),
                       ACE_LIB_TEXT ("opening notify handler ")),
                      -1);

  // Register for <wakeup_all_threads> event
  if (this->register_handler (&this->wakeup_all_threads_handler_,
                              this->wakeup_all_threads_.handle ()) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_LIB_TEXT ("%p\n"),
                       ACE_LIB_TEXT ("registering thread wakeup handler")),
                      -1);

  // Since we have added two handles into the handler repository,
  // update the <handler_repository_>
  if (this->handler_rep_.changes_required ())
    {
      // Make necessary changes to the handler repository
      this->handler_rep_.make_changes ();
      // Turn off <wakeup_all_threads_> since all necessary changes
      // have completed
      this->wakeup_all_threads_.reset ();
    }

  // We are open for business
  this->open_for_business_ = 1;

  return 0;
}

int
ACE_WFMO_Reactor::set_sig_handler (ACE_Sig_Handler *signal_handler)
{
  if (this->signal_handler_ != 0 && this->delete_signal_handler_ != 0)
    delete this->signal_handler_;
  this->signal_handler_ = signal_handler;
  this->delete_signal_handler_ = 0;
  return 0;
}

ACE_Timer_Queue *
ACE_WFMO_Reactor::timer_queue (void) const
{
  return this->timer_queue_;
}

int
ACE_WFMO_Reactor::timer_queue (ACE_Timer_Queue *tq)
{
  if (this->timer_queue_ != 0 && this->delete_timer_queue_ != 0)
    delete this->timer_queue_;
  this->timer_queue_ = tq;
  this->delete_timer_queue_ = 0;
  return 0;
}

int
ACE_WFMO_Reactor::close (void)
{
  // This GUARD is necessary since we are updating shared state.
  ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, this->lock_, -1);

  // If we are already closed, return error
  if (!this->open_for_business_)
    return -1;

  // We are now closed
  this->open_for_business_ = 0;
  // This will unregister all handles
  this->handler_rep_.close ();

  return 0;
}

ACE_WFMO_Reactor::~ACE_WFMO_Reactor (void)
{
  // Assumption: No threads are left in the Reactor when this method
  // is called (i.e., active_threads_ == 0)

  // Close down
  this->close ();

  // Make necessary changes to the handler repository that we caused
  // by <close>.
  this->handler_rep_.make_changes ();

  if (this->delete_timer_queue_)
    {
      delete this->timer_queue_;
      this->timer_queue_ = 0;
      this->delete_timer_queue_ = 0;
    }

  if (this->delete_signal_handler_)
    {
      delete this->signal_handler_;
      this->signal_handler_ = 0;
      this->delete_signal_handler_ = 0;
    }

  if (this->delete_notify_handler_)
    {
      delete this->notify_handler_;
      this->notify_handler_ = 0;
      this->delete_notify_handler_ = 0;
    }
}

int
ACE_WFMO_Reactor::register_handler_i (ACE_HANDLE event_handle,
                                      ACE_HANDLE io_handle,
                                      ACE_Event_Handler *event_handler,
                                      ACE_Reactor_Mask new_masks)
{
  // If this is a Winsock 1 system, the underlying event assignment will
  // not work, so don't try. Winsock 1 must use ACE_Select_Reactor for
  // reacting to socket activity.

#if !defined (ACE_HAS_WINSOCK2) || (ACE_HAS_WINSOCK2 == 0)

  ACE_UNUSED_ARG (event_handle);
  ACE_UNUSED_ARG (io_handle);
  ACE_UNUSED_ARG (event_handler);
  ACE_UNUSED_ARG (new_masks);
  ACE_NOTSUP_RETURN (-1);

#else

  // Make sure that the <handle> is valid
  if (io_handle == ACE_INVALID_HANDLE)
    io_handle = event_handler->get_handle ();

  if (this->handler_rep_.invalid_handle (io_handle))
    {
      errno = ERROR_INVALID_HANDLE;
      return -1;
    }

  long new_network_events = 0;
  int delete_event = 0;
  auto_ptr <ACE_Auto_Event> event;

  // Look up the repository to see if the <event_handler> is already
  // there.
  ACE_Reactor_Mask old_masks;
  int found = this->handler_rep_.modify_network_events_i (io_handle,
                                                          new_masks,
                                                          old_masks,
                                                          new_network_events,
                                                          event_handle,
                                                          delete_event,
                                                          ACE_Reactor::ADD_MASK);

  // Check to see if the user passed us a valid event; If not then we
  // need to create one
  if (event_handle == ACE_INVALID_HANDLE)
    {
      // Note: don't change this since some C++ compilers have
      // <auto_ptr>s that don't work properly...
      auto_ptr<ACE_Auto_Event> tmp (new ACE_Auto_Event);
      event = tmp;
      event_handle = event->handle ();
      delete_event = 1;
    }

  int result = ::WSAEventSelect ((SOCKET) io_handle,
                                 event_handle,
                                 new_network_events);
  // If we had found the <Event_Handler> there is nothing more to do
  if (found)
    return result;
  else if (result != SOCKET_ERROR &&
           this->handler_rep_.bind_i (1,
                                      event_handler,
                                      new_network_events,
                                      io_handle,
                                      event_handle,
                                      delete_event) != -1)
    {
      // The <event_handler> was not found in the repository, add to
      // the repository.
      if (delete_event)
        {
          // Clear out the handle in the ACE_Auto_Event so that when
          // it is destroyed, the handle isn't closed out from under
          // the reactor. After setting it, running down the event
          // (via auto_ptr<> event, above) at function return will
          // cause an error because it'll try to close an invalid handle.
          // To avoid that smashing the errno value, save the errno
          // here, explicitly remove the event so the dtor won't do it
          // again, then restore errno.
          ACE_Errno_Guard guard (errno);
          event->handle (ACE_INVALID_HANDLE);
          event->remove ();
        }
      return 0;
    }
  else
    return -1;

#endif /* ACE_HAS_WINSOCK2 || ACE_HAS_WINSOCK2 == 0 */

}

int
ACE_WFMO_Reactor::mask_ops_i (ACE_HANDLE io_handle,
                              ACE_Reactor_Mask new_masks,
                              int operation)
{
  // Make sure that the <handle> is valid
  if (this->handler_rep_.invalid_handle (io_handle))
    return -1;

  long new_network_events = 0;
  int delete_event = 0;
  ACE_HANDLE event_handle = ACE_INVALID_HANDLE;

  // Look up the repository to see if the <Event_Handler> is already
  // there.
  ACE_Reactor_Mask old_masks;
  int found = this->handler_rep_.modify_network_events_i (io_handle,
                                                          new_masks,
                                                          old_masks,
                                                          new_network_events,
                                                          event_handle,
                                                          delete_event,
                                                          operation);
  if (found)
    {
      int result = ::WSAEventSelect ((SOCKET) io_handle,
                                     event_handle,
                                     new_network_events);
      if (result == 0)
        return old_masks;
      else
        return result;
    }
  else
    return -1;
}



int
ACE_WFMO_Reactor_Handler_Repository::modify_network_events_i (ACE_HANDLE io_handle,
                                                              ACE_Reactor_Mask new_masks,
                                                              ACE_Reactor_Mask &old_masks,
                                                              long &new_network_events,
                                                              ACE_HANDLE &event_handle,
                                                              int &delete_event,
                                                              int operation)
{
  long *modified_network_events = &new_network_events;
  int found = 0;
  size_t i;

  // First go through the current entries
  //
  // Look for all entries in the current handles for matching handle
  // (except those that have been scheduled for deletion)
  for (i = 0; i < this->max_handlep1_ && !found; ++i)
    if (io_handle == this->current_info_[i].io_handle_ &&
        !this->current_info_[i].delete_entry_)
      {
        found = 1;
        modified_network_events = &this->current_info_[i].network_events_;
        delete_event = this->current_info_[i].delete_event_;
        event_handle = this->current_handles_[i];
      }

  // Then pass through the suspended handles
  //
  // Look for all entries in the suspended handles for matching handle
  // (except those that have been scheduled for deletion)
  for (i = 0; i < this->suspended_handles_ && !found; ++i)
    if (io_handle == this->current_suspended_info_[i].io_handle_ &&
        !this->current_suspended_info_[i].delete_entry_)
      {
        found = 1;
        modified_network_events = &this->current_suspended_info_[i].network_events_;
        delete_event = this->current_suspended_info_[i].delete_event_;
        event_handle = this->current_suspended_info_[i].event_handle_;
      }

  // Then check the to_be_added handles
  //
  // Look for all entries in the to_be_added handles for matching
  // handle (except those that have been scheduled for deletion)
  for (i = 0; i < this->handles_to_be_added_ && !found; ++i)
    if (io_handle == this->to_be_added_info_[i].io_handle_ &&
        !this->to_be_added_info_[i].delete_entry_)
      {
        found = 1;
        modified_network_events = &this->to_be_added_info_[i].network_events_;
        delete_event = this->to_be_added_info_[i].delete_event_;
        event_handle = this->to_be_added_info_[i].event_handle_;
      }

  old_masks = this->bit_ops (*modified_network_events,
                             new_masks,
                             operation);

  new_network_events = *modified_network_events;

  return found;
}

ACE_Event_Handler *
ACE_WFMO_Reactor_Handler_Repository::find_handler (ACE_HANDLE handle)
{
  long existing_masks_ignored = 0;
  return this->handler (handle,
                        existing_masks_ignored);
}

ACE_Event_Handler *
ACE_WFMO_Reactor_Handler_Repository::handler (ACE_HANDLE handle,
                                              long &existing_masks)
{
  int found = 0;
  size_t i = 0;
  ACE_Event_Handler *event_handler = 0;
  existing_masks = 0;

  // Look for the handle first

  // First go through the current entries
  //
  // Look for all entries in the current handles for matching handle
  // (except those that have been scheduled for deletion)
  for (i = 0; i < this->max_handlep1_ && !found; ++i)
    if ((handle == this->current_info_[i].io_handle_ ||
         handle == this->current_handles_[i]) &&
        !this->current_info_[i].delete_entry_)
      {
        found = 1;
        event_handler = this->current_info_[i].event_handler_;
        existing_masks = this->current_info_[i].network_events_;
      }

  // Then pass through the suspended handles
  //
  // Look for all entries in the suspended handles for matching handle
  // (except those that have been scheduled for deletion)
  for (i = 0; i < this->suspended_handles_ && !found; ++i)
    if ((handle == this->current_suspended_info_[i].io_handle_ ||
         handle == this->current_suspended_info_[i].event_handle_) &&
        !this->current_suspended_info_[i].delete_entry_)
      {
        found = 1;
        event_handler = this->current_suspended_info_[i].event_handler_;
        existing_masks = this->current_suspended_info_[i].network_events_;
      }

  // Then check the to_be_added handles
  //
  // Look for all entries in the to_be_added handles for matching
  // handle (except those that have been scheduled for deletion)
  for (i = 0; i < this->handles_to_be_added_ && !found; ++i)
    if ((handle == this->to_be_added_info_[i].io_handle_ ||
         handle == this->to_be_added_info_[i].event_handle_) &&
        !this->to_be_added_info_[i].delete_entry_)
      {
        found = 1;
        event_handler = this->to_be_added_info_[i].event_handler_;
        existing_masks = this->to_be_added_info_[i].network_events_;
      }

  if (event_handler)
    event_handler->add_reference ();

  return event_handler;
}

int
ACE_WFMO_Reactor_Handler_Repository::handler (ACE_HANDLE handle,
                                              ACE_Reactor_Mask user_masks,
                                              ACE_Event_Handler **user_event_handler)
{
  long existing_masks = 0;
  int found = 0;

  ACE_Event_Handler_var safe_event_handler =
    this->handler (handle,
                   existing_masks);

  if (safe_event_handler.handler ())
    found = 1;

  if (!found)
    return -1;

  // Otherwise, make sure that the masks that the user is looking for
  // are on.
  if (found &&
      ACE_BIT_ENABLED (user_masks, ACE_Event_Handler::READ_MASK))
    if (!ACE_BIT_ENABLED (existing_masks, FD_READ) &&
        !ACE_BIT_ENABLED (existing_masks, FD_CLOSE))
      found = 0;

  if (found &&
      ACE_BIT_ENABLED (user_masks, ACE_Event_Handler::WRITE_MASK))
    if (!ACE_BIT_ENABLED (existing_masks, FD_WRITE))
      found = 0;

  if (found &&
      ACE_BIT_ENABLED (user_masks, ACE_Event_Handler::EXCEPT_MASK))
    if (!ACE_BIT_ENABLED (existing_masks, FD_OOB))
      found = 0;

  if (found &&
      ACE_BIT_ENABLED (user_masks, ACE_Event_Handler::ACCEPT_MASK))
    if (!ACE_BIT_ENABLED (existing_masks, FD_ACCEPT))
      found = 0;

  if (found &&
      ACE_BIT_ENABLED (user_masks, ACE_Event_Handler::CONNECT_MASK))
    if (!ACE_BIT_ENABLED (existing_masks, FD_CONNECT))
      found = 0;

  if (found &&
      ACE_BIT_ENABLED (user_masks, ACE_Event_Handler::QOS_MASK))
    if (!ACE_BIT_ENABLED (existing_masks, FD_QOS))
      found = 0;

  if (found &&
      ACE_BIT_ENABLED (user_masks, ACE_Event_Handler::GROUP_QOS_MASK))
    if (!ACE_BIT_ENABLED (existing_masks, FD_GROUP_QOS))
      found = 0;

  if (found &&
      user_event_handler)
    *user_event_handler = safe_event_handler.release ();

  if (found)
    return 0;
  else
    return -1;
}

// Waits for and dispatches all events.  Returns -1 on error, 0 if
// max_wait_time expired, or the number of events that were dispatched.
int
ACE_WFMO_Reactor::event_handling (ACE_Time_Value *max_wait_time,
                                  int alertable)
{
  ACE_TRACE ("ACE_WFMO_Reactor::event_handling");

  // Make sure we are not closed
  if (!this->open_for_business_ || this->deactivated_)
    return -1;

  // Stash the current time -- the destructor of this object will
  // automatically compute how much time elapsed since this method was
  // called.
  ACE_Countdown_Time countdown (max_wait_time);

  int result;
  do
    {
      // Check to see if it is ok to enter ::WaitForMultipleObjects
      // This will acquire <this->lock_> on success On failure, the
      // lock will not be acquired
      result = this->ok_to_wait (max_wait_time, alertable);
      if (result != 1)
        return result;

      // Increment the number of active threads
      ++this->active_threads_;

      // Release the <lock_>
      this->lock_.release ();

      // Update the countdown to reflect time waiting to play with the
      // mut and event.
      countdown.update ();

      // Calculate timeout
      int timeout = this->calculate_timeout (max_wait_time);

      // Wait for event to happen
      DWORD wait_status = this->wait_for_multiple_events (timeout,
                                                          alertable);

      // Upcall
      result = this->safe_dispatch (wait_status);
      if (0 == result)
        {
          // wait_for_multiple_events timed out without dispatching
          // anything.  Because of rounding and conversion errors and
          // such, it could be that the wait loop timed out, but
          // the timer queue said it wasn't quite ready to expire a
          // timer. In this case, max_wait_time won't have quite been
          // reduced to 0, and we need to go around again. If max_wait_time
          // is all the way to 0, just return, as the entire time the
          // caller wanted to wait has been used up.
          countdown.update ();     // Reflect time waiting for events
          if (0 == max_wait_time || max_wait_time->usec () == 0)
            break;
        }
    }
  while (result == 0);

  return result;
}

int
ACE_WFMO_Reactor::ok_to_wait (ACE_Time_Value *max_wait_time,
                              int alertable)
{
  // Calculate the max time we should spend here
  //
  // Note: There is really no need to involve the <timer_queue_> here
  // because even if a timeout in the <timer_queue_> does expire we
  // will not be able to dispatch it
  int timeout = max_wait_time == 0 ? INFINITE : max_wait_time->msec ();

  // Atomically wait for both the <lock_> and <ok_to_wait_> event
  DWORD result = 0;
  while (1)
    {
#if defined (ACE_HAS_PHARLAP)
      // PharLap doesn't implement WaitForMultipleObjectsEx, and doesn't
      // do async I/O, so it's not needed in this case anyway.
      result = ::WaitForMultipleObjects (sizeof this->atomic_wait_array_ / sizeof (ACE_HANDLE),
                                         this->atomic_wait_array_,
                                         TRUE,
                                         timeout);

      if (result != WAIT_IO_COMPLETION)
        break;

#elif defined (ACE_HAS_WINCE)
      result = ::WaitForMultipleObjects (sizeof this->atomic_wait_array_ / sizeof (ACE_HANDLE),
                                         this->atomic_wait_array_,
                                         TRUE,
                                         timeout);
      break;  // CE does not have WAIT_IO_COMPLETION defined.
#else
      result = ::WaitForMultipleObjectsEx (sizeof this->atomic_wait_array_ / sizeof (ACE_HANDLE),
                                           this->atomic_wait_array_,
                                           TRUE,
                                           timeout,
                                           alertable);

      if (result != WAIT_IO_COMPLETION)
        break;

#endif /* ACE_HAS_PHARLAP */
    }

  switch (result)
    {
    case WAIT_TIMEOUT:
      errno = ETIME;
      return 0;
    case WAIT_FAILED:
    case WAIT_ABANDONED_0:
      ACE_OS::set_errno_to_last_error ();
      return -1;
    default:
      break;
    }

  // It is ok to enter ::WaitForMultipleObjects
  return 1;
}

DWORD
ACE_WFMO_Reactor::wait_for_multiple_events (int timeout,
                                            int alertable)
{
  // Wait for any of handles_ to be active, or until timeout expires.
  // If <alertable> is enabled allow asynchronous completion of
  // ReadFile and WriteFile operations.

#if defined (ACE_HAS_PHARLAP) || defined (ACE_HAS_WINCE)
  // PharLap doesn't do async I/O and doesn't implement
  // WaitForMultipleObjectsEx, so use WaitForMultipleObjects.
  ACE_UNUSED_ARG (alertable);
  return ::WaitForMultipleObjects (this->handler_rep_.max_handlep1 (),
                                   this->handler_rep_.handles (),
                                   FALSE,
                                   timeout);
#else
  return ::WaitForMultipleObjectsEx (this->handler_rep_.max_handlep1 (),
                                     this->handler_rep_.handles (),
                                     FALSE,
                                     timeout,
                                     alertable);
#endif /* ACE_HAS_PHARLAP */
}

DWORD
ACE_WFMO_Reactor::poll_remaining_handles (DWORD slot)
{
  return ::WaitForMultipleObjects (this->handler_rep_.max_handlep1 () - slot,
                                   this->handler_rep_.handles () + slot,
                                   FALSE,
                                   0);
}

int
ACE_WFMO_Reactor::calculate_timeout (ACE_Time_Value *max_wait_time)
{
  ACE_Time_Value *time = 0;
  if (this->owner_ == ACE_Thread::self ())
    time = this->timer_queue_->calculate_timeout (max_wait_time);
  else
    time = max_wait_time;

  if (time == 0)
    return INFINITE;
  else
    return time->msec ();
}


int
ACE_WFMO_Reactor::expire_timers (void)
{
  // If "owner" thread
  if (ACE_Thread::self () == this->owner_)
    // expire all pending timers.
    return this->timer_queue_->expire ();

  else
    // Nothing to expire
    return 0;
}

int
ACE_WFMO_Reactor::dispatch (DWORD wait_status)
{
  int handlers_dispatched = 0;

  // Expire timers
  handlers_dispatched += this->expire_timers ();

  switch (wait_status)
    {
    case WAIT_FAILED: // Failure.
      ACE_OS::set_errno_to_last_error ();
      return -1;

    case WAIT_TIMEOUT: // Timeout.
      errno = ETIME;
      return handlers_dispatched;

#ifndef ACE_HAS_WINCE
    case WAIT_IO_COMPLETION: // APC.
      return handlers_dispatched;
#endif  // ACE_HAS_WINCE

    default:  // Dispatch.
      // We'll let dispatch worry about abandoned mutes.
      handlers_dispatched += this->dispatch_handles (wait_status);
      return handlers_dispatched;
    }
}

// Dispatches any active handles from <handles_[slot]> to
// <handles_[max_handlep1_]>, polling through our handle set looking
// for active handles.
int
ACE_WFMO_Reactor::dispatch_handles (DWORD wait_status)
{
  // dispatch_slot is the absolute slot.  Only += is used to
  // increment it.
  DWORD dispatch_slot = 0;

  // Cache this value, this is the absolute value.
  DWORD max_handlep1 = this->handler_rep_.max_handlep1 ();

  // nCount starts off at <max_handlep1>, this is a transient count of
  // handles last waited on.
  DWORD nCount = max_handlep1;

  for (int number_of_handlers_dispatched = 1;
       ;
       ++number_of_handlers_dispatched)
    {
      const bool ok = (
#if ! (defined(__BORLANDC__) && (__BORLANDC__ >= 0x0530)) \
    && !defined (ghs) \
    && !defined (__MINGW32__) \
    && !(defined (_MSC_VER) && _MSC_VER >= 1300)
                 // wait_status is unsigned in Borland, Green Hills,
                 // mingw32 and MSVC++ >= 7.1.
                 // This >= is always true, with a warning.
                 wait_status >= WAIT_OBJECT_0 &&
#endif
                 wait_status <= (WAIT_OBJECT_0 + nCount));

      if (ok)
        dispatch_slot += wait_status - WAIT_OBJECT_0;
      else
        // Otherwise, a handle was abandoned.
        dispatch_slot += wait_status - WAIT_ABANDONED_0;

      // Dispatch handler
      if (this->dispatch_handler (dispatch_slot, max_handlep1) == -1)
        return -1;

      // Increment slot
      ++dispatch_slot;

      // We're done.
      if (dispatch_slot >= max_handlep1)
        return number_of_handlers_dispatched;

      // Readjust nCount
      nCount = max_handlep1 - dispatch_slot;

      // Check the remaining handles
      wait_status = this->poll_remaining_handles (dispatch_slot);
      switch (wait_status)
        {
        case WAIT_FAILED: // Failure.
          ACE_OS::set_errno_to_last_error ();
          /* FALLTHRU */
        case WAIT_TIMEOUT:
          // There are no more handles ready, we can return.
          return number_of_handlers_dispatched;
        }
    }
}

int
ACE_WFMO_Reactor::dispatch_handler (DWORD slot,
                                    DWORD max_handlep1)
{
  // Check if there are window messages that need to be dispatched
  if (slot == max_handlep1)
    return this->dispatch_window_messages ();

  // Dispatch the handler if it has not been scheduled for deletion.
  // Note that this is a very week test if there are multiple threads
  // dispatching this slot as no locks are held here. Generally, you
  // do not want to do something like deleting the this pointer in
  // handle_close() if you have registered multiple times and there is
  // more than one thread in WFMO_Reactor->handle_events().
  else if (!this->handler_rep_.scheduled_for_deletion (slot))
    {
      ACE_HANDLE event_handle = *(this->handler_rep_.handles () + slot);

      if (this->handler_rep_.current_info ()[slot].io_entry_)
        return this->complex_dispatch_handler (slot,
                                               event_handle);
      else
        return this->simple_dispatch_handler (slot,
                                              event_handle);
    }
  else
    // The handle was scheduled for deletion, so we will skip it.
    return 0;
}

int
ACE_WFMO_Reactor::simple_dispatch_handler (DWORD slot,
                                           ACE_HANDLE event_handle)
{
  // This dispatch is used for non-I/O entires

  // Assign the ``signaled'' HANDLE so that callers can get it.
  // siginfo_t is an ACE - specific fabrication. Constructor exists.
  siginfo_t sig (event_handle);

  ACE_Event_Handler *event_handler =
    this->handler_rep_.current_info ()[slot].event_handler_;

  int requires_reference_counting =
    event_handler->reference_counting_policy ().value () ==
    ACE_Event_Handler::Reference_Counting_Policy::ENABLED;

  if (requires_reference_counting)
    {
      event_handler->add_reference ();
    }

  // Upcall
  if (event_handler->handle_signal (0, &sig) == -1)
    this->handler_rep_.unbind (event_handle,
                               ACE_Event_Handler::NULL_MASK);

  // Call remove_reference() if needed.
  if (requires_reference_counting)
    {
      event_handler->remove_reference ();
    }

  return 0;
}

int
ACE_WFMO_Reactor::complex_dispatch_handler (DWORD slot,
                                            ACE_HANDLE event_handle)
{
  // This dispatch is used for I/O entires.

  ACE_WFMO_Reactor_Handler_Repository::Current_Info &current_info =
    this->handler_rep_.current_info ()[slot];

  WSANETWORKEVENTS events;
  ACE_Reactor_Mask problems = ACE_Event_Handler::NULL_MASK;
  if (::WSAEnumNetworkEvents ((SOCKET) current_info.io_handle_,
                              event_handle,
                              &events) == SOCKET_ERROR)
    problems = ACE_Event_Handler::ALL_EVENTS_MASK;
  else
    {
      // Prepare for upcalls. Clear the bits from <events> representing
      // events the handler is not interested in. If there are any left,
      // do the upcall(s). upcall will replace events.lNetworkEvents
      // with bits representing any functions that requested a repeat
      // callback before checking handles again. In this case, continue
      // to call back unless the handler is unregistered as a result of
      // one of the upcalls. The way this is written, the upcalls will
      // keep being done even if one or more upcalls reported problems.
      // In practice this may turn out not so good, but let's see. If any
      // problems, please notify Steve Huston <shuston@riverace.com>
      // before or after you change this code.
      events.lNetworkEvents &= current_info.network_events_;
      while (events.lNetworkEvents != 0)
        {
          ACE_Event_Handler *event_handler =
            current_info.event_handler_;

          int reference_counting_required =
            event_handler->reference_counting_policy ().value () ==
            ACE_Event_Handler::Reference_Counting_Policy::ENABLED;

          // Call add_reference() if needed.
          if (reference_counting_required)
            {
              event_handler->add_reference ();
            }

          // Upcall
          problems |= this->upcall (current_info.event_handler_,
                                    current_info.io_handle_,
                                    events);

          // Call remove_reference() if needed.
          if (reference_counting_required)
            {
              event_handler->remove_reference ();
            }

          if (this->handler_rep_.scheduled_for_deletion (slot))
            break;
        }
    }

  if (problems != ACE_Event_Handler::NULL_MASK
      && !this->handler_rep_.scheduled_for_deletion (slot)  )
    this->handler_rep_.unbind (event_handle, problems);

  return 0;
}

ACE_Reactor_Mask
ACE_WFMO_Reactor::upcall (ACE_Event_Handler *event_handler,
                          ACE_HANDLE io_handle,
                          WSANETWORKEVENTS &events)
{
  // This method figures out what exactly has happened to the socket
  // and then calls appropriate methods.
  ACE_Reactor_Mask problems = ACE_Event_Handler::NULL_MASK;

  // Go through the events and do the indicated upcalls. If the handler
  // doesn't want to be called back, clear the bit for that event.
  // At the end, set the bits back to <events> to request a repeat call.

  long actual_events = events.lNetworkEvents;
  int action;

  if (ACE_BIT_ENABLED (actual_events, FD_WRITE))
    {
      action = event_handler->handle_output (io_handle);
      if (action <= 0)
        {
          ACE_CLR_BITS (actual_events, FD_WRITE);
          if (action == -1)
            ACE_SET_BITS (problems, ACE_Event_Handler::WRITE_MASK);
        }
    }

  if (ACE_BIT_ENABLED (actual_events, FD_CONNECT))
    {
      if (events.iErrorCode[FD_CONNECT_BIT] == 0)
        {
          // Successful connect
          action = event_handler->handle_output (io_handle);
          if (action <= 0)
            {
              ACE_CLR_BITS (actual_events, FD_CONNECT);
              if (action == -1)
                ACE_SET_BITS (problems,
                              ACE_Event_Handler::CONNECT_MASK);
            }
        }
      // Unsuccessful connect
      else
        {
          action = event_handler->handle_input (io_handle);
          if (action <= 0)
            {
              ACE_CLR_BITS (actual_events, FD_CONNECT);
              if (action == -1)
                ACE_SET_BITS (problems,
                              ACE_Event_Handler::CONNECT_MASK);
            }
        }
    }

  if (ACE_BIT_ENABLED (actual_events, FD_OOB))
    {
      action = event_handler->handle_exception (io_handle);
      if (action <= 0)
        {
          ACE_CLR_BITS (actual_events, FD_OOB);
          if (action == -1)
            ACE_SET_BITS (problems, ACE_Event_Handler::EXCEPT_MASK);
        }
    }

  if (ACE_BIT_ENABLED (actual_events, FD_READ))
    {
      action = event_handler->handle_input (io_handle);
      if (action <= 0)
        {
          ACE_CLR_BITS (actual_events, FD_READ);
          if (action == -1)
            ACE_SET_BITS (problems, ACE_Event_Handler::READ_MASK);
        }
    }

  if (ACE_BIT_ENABLED (actual_events, FD_CLOSE)
      && ACE_BIT_DISABLED (problems, ACE_Event_Handler::READ_MASK))
    {
      action = event_handler->handle_input (io_handle);
      if (action <= 0)
        {
          ACE_CLR_BITS (actual_events, FD_CLOSE);
          if (action == -1)
            ACE_SET_BITS (problems, ACE_Event_Handler::READ_MASK);
        }
    }

          if (ACE_BIT_ENABLED (actual_events, FD_ACCEPT))
    {
      action = event_handler->handle_input (io_handle);
      if (action <= 0)
        {
          ACE_CLR_BITS (actual_events, FD_ACCEPT);
          if (action == -1)
            ACE_SET_BITS (problems, ACE_Event_Handler::ACCEPT_MASK);
        }
    }

  if (ACE_BIT_ENABLED (actual_events, FD_QOS))
    {
      action = event_handler->handle_qos (io_handle);
      if (action <= 0)
        {
          ACE_CLR_BITS (actual_events, FD_QOS);
          if (action == -1)
            ACE_SET_BITS (problems, ACE_Event_Handler::QOS_MASK);
        }
    }

  if (ACE_BIT_ENABLED (actual_events, FD_GROUP_QOS))
    {
      action = event_handler->handle_group_qos (io_handle);
      if (action <= 0)
        {
          ACE_CLR_BITS (actual_events, FD_GROUP_QOS);
          if (action == -1)
            ACE_SET_BITS (problems, ACE_Event_Handler::GROUP_QOS_MASK);
        }
    }

  events.lNetworkEvents = actual_events;
  return problems;
}


int
ACE_WFMO_Reactor::update_state (void)
{
  // This GUARD is necessary since we are updating shared state.
  ACE_GUARD_RETURN (ACE_Process_Mutex, monitor, this->lock_, -1);

  // Decrement active threads
  --this->active_threads_;

  // Check if the state of the handler repository has changed or new
  // owner has to be set
  if (this->handler_rep_.changes_required () || this->new_owner ())
    {
      if (this->change_state_thread_ == 0)
        // Try to become the thread which will be responsible for the
        // changes
        {
          this->change_state_thread_ = ACE_Thread::self ();
          // Make sure no new threads are allowed to enter
          this->ok_to_wait_.reset ();

          if (this->active_threads_ > 0)
            // Check for other active threads
            {
              // Wake up all other threads
              this->wakeup_all_threads_.signal ();
              // Release <lock_>
              monitor.release ();
              // Go to sleep waiting for all other threads to get done
              this->waiting_to_change_state_.wait ();
              // Re-acquire <lock_> again
              monitor.acquire ();
            }

          // Note that make_changes() calls into user code which can
          // request other changes.  So keep looping until all
          // requested changes are completed.
          while (this->handler_rep_.changes_required ())
            // Make necessary changes to the handler repository
            this->handler_rep_.make_changes ();
          if (this->new_owner ())
            // Update the owner
            this->change_owner ();
          // Turn off <wakeup_all_threads_>
          this->wakeup_all_threads_.reset ();
          // Let everyone know that it is ok to go ahead
          this->ok_to_wait_.signal ();
          // Reset this flag
          this->change_state_thread_ = 0;
        }
      else if (this->active_threads_ == 0)
        // This thread did not get a chance to become the change
        // thread. If it is the last one out, it will wakeup the
        // change thread
        this->waiting_to_change_state_.signal ();
    }
  // This is if we were woken up explicitily by the user and there are
  // no state changes required.
  else if (this->active_threads_ == 0)
    // Turn off <wakeup_all_threads_>
    this->wakeup_all_threads_.reset ();

  return 0;
}

void
ACE_WFMO_Reactor::dump (void) const
{
#if defined (ACE_HAS_DUMP)
  ACE_TRACE ("ACE_WFMO_Reactor::dump");

  ACE_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("Count of currently active threads = %d\n"),
              this->active_threads_));

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("ID of owner thread = %d\n"),
              this->owner_));

  this->handler_rep_.dump ();
  this->signal_handler_->dump ();
  this->timer_queue_->dump ();

  ACE_DEBUG ((LM_DEBUG, ACE_END_DUMP));
#endif /* ACE_HAS_DUMP */
}

int
ACE_WFMO_Reactor_Notify::dispatch_notifications (int & /*number_of_active_handles*/,
                                                 ACE_Handle_Set & /*rd_mask*/)
{
  return -1;
}

int
ACE_WFMO_Reactor_Notify::is_dispatchable (ACE_Notification_Buffer & /*buffer*/)
{
  return 0;
}

ACE_HANDLE
ACE_WFMO_Reactor_Notify::notify_handle (void)
{
  return ACE_INVALID_HANDLE;
}

int
ACE_WFMO_Reactor_Notify::read_notify_pipe (ACE_HANDLE ,
                                           ACE_Notification_Buffer &)
{
  return 0;
}

int
ACE_WFMO_Reactor_Notify::dispatch_notify (ACE_Notification_Buffer &)
{
  return 0;
}

int
ACE_WFMO_Reactor_Notify::close (void)
{
  return -1;
}

ACE_WFMO_Reactor_Notify::ACE_WFMO_Reactor_Notify (size_t max_notifies)
  : timer_queue_ (0),
    message_queue_ (max_notifies * sizeof (ACE_Notification_Buffer),
                    max_notifies * sizeof (ACE_Notification_Buffer)),
    max_notify_iterations_ (-1)
{
}

int
ACE_WFMO_Reactor_Notify::open (ACE_Reactor_Impl *wfmo_reactor,
                               ACE_Timer_Queue *timer_queue,
                               int ignore_notify)
{
  ACE_UNUSED_ARG (ignore_notify);
  timer_queue_ = timer_queue;
  return wfmo_reactor->register_handler (this);
}

ACE_HANDLE
ACE_WFMO_Reactor_Notify::get_handle (void) const
{
  return this->wakeup_one_thread_.handle ();
}

// Handle all pending notifications.

int
ACE_WFMO_Reactor_Notify::handle_signal (int signum,
                                        siginfo_t *siginfo,
                                        ucontext_t *)
{
  ACE_UNUSED_ARG (signum);

  // Just check for sanity...
  if (siginfo->si_handle_ != this->wakeup_one_thread_.handle ())
    return -1;

  // This will get called when <WFMO_Reactor->wakeup_one_thread_> event
  // is signaled.
  //  ACE_DEBUG ((LM_DEBUG,
  //             ACE_LIB_TEXT ("(%t) waking up to handle internal notifications\n")));

  for (int i = 1; ; ++i)
    {
      ACE_Message_Block *mb = 0;
      // Copy ACE_Time_Value::zero since dequeue_head will modify it.
      ACE_Time_Value zero_timeout (ACE_Time_Value::zero);
      if (this->message_queue_.dequeue_head (mb, &zero_timeout) == -1)
        {
          if (errno == EWOULDBLOCK)
            // We've reached the end of the processing, return
            // normally.
            return 0;
          else
            return -1; // Something weird happened...
        }
      else
        {
          ACE_Notification_Buffer *buffer =
            reinterpret_cast <ACE_Notification_Buffer *> (mb->base ());

          // If eh == 0 then we've got major problems!  Otherwise, we
          // need to dispatch the appropriate handle_* method on the
          // ACE_Event_Handler pointer we've been passed.

          if (buffer->eh_ != 0)
            {
              ACE_Event_Handler *event_handler =
                buffer->eh_;

              int requires_reference_counting =
                event_handler->reference_counting_policy ().value () ==
                ACE_Event_Handler::Reference_Counting_Policy::ENABLED;

              int result = 0;

              switch (buffer->mask_)
                {
                case ACE_Event_Handler::READ_MASK:
                case ACE_Event_Handler::ACCEPT_MASK:
                  result = event_handler->handle_input (ACE_INVALID_HANDLE);
                  break;
                case ACE_Event_Handler::WRITE_MASK:
                  result = event_handler->handle_output (ACE_INVALID_HANDLE);
                  break;
                case ACE_Event_Handler::EXCEPT_MASK:
                  result = event_handler->handle_exception (ACE_INVALID_HANDLE);
                  break;
                case ACE_Event_Handler::QOS_MASK:
                  result = event_handler->handle_qos (ACE_INVALID_HANDLE);
                  break;
                case ACE_Event_Handler::GROUP_QOS_MASK:
                  result = event_handler->handle_group_qos (ACE_INVALID_HANDLE);
                  break;
                default:
                  ACE_ERROR ((LM_ERROR,
                              ACE_LIB_TEXT ("invalid mask = %d\n"),
                              buffer->mask_));
                  break;
                }

              if (result == -1)
                event_handler->handle_close (ACE_INVALID_HANDLE,
                                             ACE_Event_Handler::EXCEPT_MASK);

              if (requires_reference_counting)
                {
                  event_handler->remove_reference ();
                }
            }

          // Make sure to delete the memory regardless of success or
          // failure!
          mb->release ();

          // Bail out if we've reached the <max_notify_iterations_>.
          // Note that by default <max_notify_iterations_> is -1, so
          // we'll loop until we're done.
          if (i == this->max_notify_iterations_)
            {
              // If there are still notification in the queue, we need
              // to wake up again
              if (!this->message_queue_.is_empty ())
                this->wakeup_one_thread_.signal ();

              // Break the loop as we have reached max_notify_iterations_
              return 0;
            }
        }
    }
}

// Notify the WFMO_Reactor, potentially enqueueing the
// <ACE_Event_Handler> for subsequent processing in the WFMO_Reactor
// thread of control.

int
ACE_WFMO_Reactor_Notify::notify (ACE_Event_Handler *event_handler,
                                 ACE_Reactor_Mask mask,
                                 ACE_Time_Value *timeout)
{
  if (event_handler != 0)
    {
      ACE_Message_Block *mb = 0;
      ACE_NEW_RETURN (mb,
                      ACE_Message_Block (sizeof (ACE_Notification_Buffer)),
                      -1);

      ACE_Notification_Buffer *buffer =
        (ACE_Notification_Buffer *) mb->base ();
      buffer->eh_ = event_handler;
      buffer->mask_ = mask;

      // Convert from relative time to absolute time by adding the
      // current time of day.  This is what <ACE_Message_Queue>
      // expects.
      if (timeout != 0)
        *timeout += timer_queue_->gettimeofday ();

      if (this->message_queue_.enqueue_tail
          (mb, timeout) == -1)
        {
          mb->release ();
          return -1;
        }

      event_handler->add_reference ();
    }

  return this->wakeup_one_thread_.signal ();
}

void
ACE_WFMO_Reactor_Notify::max_notify_iterations (int iterations)
{
  ACE_TRACE ("ACE_WFMO_Reactor_Notify::max_notify_iterations");
  // Must always be > 0 or < 0 to optimize the loop exit condition.
  if (iterations == 0)
    iterations = 1;

  this->max_notify_iterations_ = iterations;
}

int
ACE_WFMO_Reactor_Notify::max_notify_iterations (void)
{
  ACE_TRACE ("ACE_WFMO_Reactor_Notify::max_notify_iterations");
  return this->max_notify_iterations_;
}

int
ACE_WFMO_Reactor_Notify::purge_pending_notifications (ACE_Event_Handler *eh,
                                                      ACE_Reactor_Mask mask)
{
  ACE_TRACE ("ACE_WFMO_Reactor_Notify::purge_pending_notifications");

  // Go over message queue and take out all the matching event
  // handlers.  If eh == 0, purge all. Note that reactor notifies (no
  // handler specified) are never purged, as this may lose a needed
  // notify the reactor queued for itself.

  if (this->message_queue_.is_empty ())
    return 0;

  // Guard against new and/or delivered notifications while purging.
  // WARNING!!! The use of the notification queue's lock object for
  // this guard makes use of the knowledge that on Win32, the mutex
  // protecting the queue is really a CriticalSection, which is
  // recursive. This is how we can get away with locking it down here
  // and still calling member functions on the queue object.
  ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, monitor, this->message_queue_.lock(), -1);

  // first, copy all to our own local queue. Since we've locked everyone out
  // of here, there's no need to use any synchronization on this queue.
  ACE_Message_Queue<ACE_NULL_SYNCH> local_queue;

  size_t queue_size  = this->message_queue_.message_count ();
  int number_purged = 0;

  size_t index;

  for (index = 0; index < queue_size; ++index)
    {
      ACE_Message_Block *mb = 0;
      if (-1 == this->message_queue_.dequeue_head (mb))
        return -1;        // This shouldn't happen...

      ACE_Notification_Buffer *buffer =
        reinterpret_cast<ACE_Notification_Buffer *> (mb->base ());

      // If this is not a Reactor notify (it is for a particular handler),
      // and it matches the specified handler (or purging all),
      // and applying the mask would totally eliminate the notification, then
      // release it and count the number purged.
      if ((0 != buffer->eh_) &&
          (0 == eh || eh == buffer->eh_) &&
          ACE_BIT_DISABLED (buffer->mask_, ~mask)) // the existing notification mask
                                                   // is left with nothing when
                                                   // applying the mask
        {
          ACE_Event_Handler *event_handler = buffer->eh_;

          event_handler->remove_reference ();

          mb->release ();
          ++number_purged;
        }
      else
        {
          // To preserve it, move it to the local_queue.  But first, if
          // this is not a Reactor notify (it is for a
          // particularhandler), and it matches the specified handler
          // (or purging all), then apply the mask
          if ((0 != buffer->eh_) &&
              (0 == eh || eh == buffer->eh_))
            ACE_CLR_BITS(buffer->mask_, mask);
          if (-1 == local_queue.enqueue_head (mb))
            return -1;
        }
    }

  if (this->message_queue_.message_count ())
    { // Should be empty!
      ACE_ASSERT (0);
      return -1;
    }

  // Now copy back from the local queue to the class queue, taking
  // care to preserve the original order...
  queue_size  = local_queue.message_count ();
  for (index = 0; index < queue_size; ++index)
    {
      ACE_Message_Block  *mb = 0;
      if (-1 == local_queue.dequeue_head (mb))
        {
          ACE_ASSERT (0);
          return -1;
        }

      if (-1 == this->message_queue_.enqueue_head (mb))
        {
          ACE_ASSERT (0);
          return -1;
        }
    }

  return number_purged;
}

void
ACE_WFMO_Reactor_Notify::dump (void) const
{
#if defined (ACE_HAS_DUMP)
  ACE_TRACE ("ACE_WFMO_Reactor_Notify::dump");
  ACE_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));
  this->timer_queue_->dump ();
  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT ("Max. iteration: %d\n"),
              this->max_notify_iterations_));
  ACE_DEBUG ((LM_DEBUG, ACE_END_DUMP));
#endif /* ACE_HAS_DUMP */
}

void
ACE_WFMO_Reactor::max_notify_iterations (int iterations)
{
  ACE_TRACE ("ACE_WFMO_Reactor::max_notify_iterations");
  ACE_GUARD (ACE_Process_Mutex, monitor, this->lock_);

  // Must always be > 0 or < 0 to optimize the loop exit condition.
  this->notify_handler_->max_notify_iterations (iterations);
}

int
ACE_WFMO_Reactor::max_notify_iterations (void)
{
  ACE_TRACE ("ACE_WFMO_Reactor::max_notify_iterations");
  ACE_GUARD_RETURN (ACE_Process_Mutex, monitor, this->lock_, -1);

  return this->notify_handler_->max_notify_iterations ();
}

int
ACE_WFMO_Reactor::purge_pending_notifications (ACE_Event_Handler *eh,
                                               ACE_Reactor_Mask mask)
{
  ACE_TRACE ("ACE_WFMO_Reactor::purge_pending_notifications");
  if (this->notify_handler_ == 0)
    return 0;
  else
    return this->notify_handler_->purge_pending_notifications (eh, mask);
}

int
ACE_WFMO_Reactor::resumable_handler (void)
{
  ACE_TRACE ("ACE_WFMO_Reactor::resumable_handler");
  return 0;
}


// No-op WinSOCK2 methods to help WFMO_Reactor compile
#if !defined (ACE_HAS_WINSOCK2) || (ACE_HAS_WINSOCK2 == 0)
int
WSAEventSelect (SOCKET /* s */,
                WSAEVENT /* hEventObject */,
                long /* lNetworkEvents */)
{
  return -1;
}

int
WSAEnumNetworkEvents (SOCKET /* s */,
                      WSAEVENT /* hEventObject */,
                      LPWSANETWORKEVENTS /* lpNetworkEvents */)
{
  return -1;
}
#endif /* !defined ACE_HAS_WINSOCK2 */

ACE_END_VERSIONED_NAMESPACE_DECL

#endif /* ACE_WIN32 */

---