int
MCT_Task::open (void *)
{
  MCT_Event_Handler *handler;

  ACE_INET_Addr addr = this->config_.group_start ();
  int groups = this->config_.groups ();
  for (int i = 0; i < groups; ++i)
    {
      ACE_NEW_RETURN (handler,
                      MCT_Event_Handler (this->config_.options ()), -1);
      // We subscribe to all groups for the first one and one each for
      // all the others.
      if (i == 0)
        {
          // go ahead and hide the other one since we want our own.
          ACE_INET_Addr addr = this->config_.group_start ();
          for (int j = 0; j < groups; ++j)
            {
              // If OPT_BINDADDR_YES is set, this will fail after the first
              // join, so just break and keep on going, otherwise it's a
              // real error.
              if (j > 0
                  && ACE_BIT_ENABLED (ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_YES,
                                      this->config_.options ()))
                break;

              if (handler->join (addr) == -1)
                ACE_ERROR_RETURN ((LM_ERROR,
                                   ACE_TEXT ("MCT_Task::open - join error\n")),
                                  -1);
              advance_addr (addr);
            }
        }
      else
        {
          if (handler->join (addr) == -1)
            ACE_ERROR_RETURN ((LM_ERROR,
                               ACE_TEXT ("MCT_Task::open - join error\n")),
                              -1);
        }

      advance_addr (addr);

      if (this->reactor ()->register_handler (handler, READ_MASK) == -1)
        ACE_ERROR_RETURN ((LM_ERROR,
                           ACE_TEXT ("MCT_Task::open - cannot register ")
                           ACE_TEXT ("handler\n")),
                          -1);
    }

  if (this->activate (THR_NEW_LWP) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("%p\n"),
                       ACE_TEXT ("MCT_TASK:open - activate failed")),
                      -1);
  return 0;
}

int
run_test (u_long unload_mask = 0)
{
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Running test with mask = %s|%s\n"),
              ACE_BIT_ENABLED(unload_mask, ACE_DLL_UNLOAD_POLICY_PER_DLL) == 0
              ? ACE_TEXT ("PER_PROCESS") : ACE_TEXT ("PER_DLL"),
              ACE_BIT_ENABLED(unload_mask, ACE_DLL_UNLOAD_POLICY_LAZY) == 0
              ? ACE_TEXT ("EAGER") : ACE_TEXT ("LAZY")));

  ACE_DLL_Manager::instance ()->unload_policy (unload_mask);

  // Now, let the ACE Service Configurator framework load our service from a
  // dll, which contains a singleton.
  ACE_Service_Config::open (ACE_TEXT ("Framework_Component_Test"),
                            ACE_DEFAULT_LOGGER_KEY,
                            1, 1, 1);

  // Now add server 1.
  ACE_Service_Config::process_directive (ADD_SERVICE(1));

  // And unload the first one, could unload the dll.
  ACE_Service_Config::process_directive (REMOVE_SERVICE(1));

  // Now readd server 1.
  ACE_Service_Config::process_directive (ADD_SERVICE(1));

  // And load another service from the same library.
  ACE_Service_Config::process_directive (ADD_SERVICE(2));

  // Unload the first one again, should *not* unload the dll this time.
  ACE_Service_Config::process_directive (REMOVE_SERVICE(1));

  // And unload the second service.  Since the ACE_DLL_Handle will no longer
  // have any references, the ACE_DLL_Manager will apply it's current unloading
  // strategy and either call ACE_OS::dlclose() immediately, schedule a timeout
  // the the reactor to call dlclose() some time in the future, or keep the
  // dll loaded until program termination.
  ACE_Service_Config::process_directive (REMOVE_SERVICE(2));

  // Force unloading so we'll be ready for the next test.
  ACE_DLL_Manager::instance ()->unload_policy (ACE_DLL_UNLOAD_POLICY_DEFAULT);

  ACE_Service_Config::close ();
  return 0;
}

template <ACE_SYNCH_DECL> int
ACE_Stream_Head<ACE_SYNCH_USE>::canonical_flush (ACE_Message_Block *mb)
{
  ACE_TRACE ("ACE_Stream_Head<ACE_SYNCH_USE>::canonical_flush");
  char *cp = mb->rd_ptr ();

  if (ACE_BIT_ENABLED (*cp, ACE_Task_Flags::ACE_FLUSHR))
    {
      this->flush (ACE_Task_Flags::ACE_FLUSHALL);
      ACE_CLR_BITS (*cp, ACE_Task_Flags::ACE_FLUSHR);
    }

  if (ACE_BIT_ENABLED (*cp, ACE_Task_Flags::ACE_FLUSHW))
    return this->reply (mb);
  else
    mb->release ();
  return 0;
}

int
JAWS_IO_Reactive_Handler::handle_close (ACE_HANDLE, ACE_Reactor_Mask)
{
  if (this->completer_)
    {
      if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::WRITE_MASK))
        this->completer_->output_complete (this->io_result_, this->act_);
      else if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::READ_MASK))
        this->completer_->input_complete (this->io_result_, this->act_);
    }

  ACE_Reactor::instance ()
  ->remove_handler ( this
                   , ACE_Event_Handler::RWE_MASK|ACE_Event_Handler::DONT_CALL
                   );

  delete this;
  return 0;
}

void
ACE_Select_Reactor_Impl::clear_dispatch_mask (ACE_HANDLE handle,
                                              ACE_Reactor_Mask mask)
{
  ACE_TRACE ("ACE_Select_Reactor_Impl::clear_dispatch_mask");

  //  Use handle and mask in order to modify the sets
  // (wait/suspend/ready/dispatch), that way, the dispatch_io_set loop
  // will not be interrupt, and there will no reason to rescan the
  // wait_set and re-calling select function, which is *very*
  // expensive. It seems that wait/suspend/ready sets are getting
  // updated in register/remove bind/unbind etc functions.  The only
  // thing need to be updated is the dispatch_set (also can  be found
  // in that file code as dispatch_mask).  Because of that, we need
  // that dispatch_set to be member of the ACE_Select_Reactor_impl in
  // Select_Reactor_Base.h file  That way we will have access to that
  // member in that function.

  // We kind of invalidate the iterator in dispatch_io_set because its
  // an array and index built from the original dispatch-set. Take a
  // look at dispatch_io_set for more details.

  // We only need to clr_bit, because we are interested in clearing the
  // handles that was removed, so no dispatching to these handles will
  // occur.
  if (ACE_BIT_ENABLED (mask, ACE_Event_Handler::READ_MASK) ||
      ACE_BIT_ENABLED (mask, ACE_Event_Handler::ACCEPT_MASK))
    {
      this->dispatch_set_.rd_mask_.clr_bit (handle);
    }
  if (ACE_BIT_ENABLED (mask, ACE_Event_Handler::WRITE_MASK))
    {
      this->dispatch_set_.wr_mask_.clr_bit (handle);
    }
  if (ACE_BIT_ENABLED (mask, ACE_Event_Handler::EXCEPT_MASK))
    {
      this->dispatch_set_.ex_mask_.clr_bit (handle);
    }

  // That will make the dispatch_io_set iterator re-start and rescan
  // the dispatch set.
  this->state_changed_ = true;
}

int
JAWS_IO_Reactive_Handler::handle_timeout (const ACE_Time_Value &, const void *)
{
  if (this->was_active_)
    {
      this->was_active_ = 0;

      this->timer_id_ =
        ACE_Reactor::instance ()->schedule_timer (this, 0, this->tv_);

      return 0;
    }

  ACE_Reactor::instance ()
  ->remove_handler ( this
                   , ACE_Event_Handler::RWE_MASK|ACE_Event_Handler::DONT_CALL
                   );

  this->timer_id_ = -1;

  if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::WRITE_MASK))
    {
      JAWS_Event_Result io_result ( 0
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_SEND_TIMEOUT
                                  , ETIME
                                  );
      this->io_result_ = io_result;
    }
  else if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::READ_MASK))
    {
      JAWS_Event_Result io_result ( 0
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_RECV_TIMEOUT
                                  , ETIME
                                  );
      this->io_result_ = io_result;
    }

  return -1;
}

Connection_Handler *
Connection_Handler_Factory::make_connection_handler (const Connection_Config_Info &pci)
{
  Connection_Handler *connection_handler = 0;

  // The next few lines of code are dependent on whether we are making
  // a threaded/reactive Supplier_Handler/Consumer_Handler.

  if (pci.connection_role_ == 'C') // Configure a Consumer_Handler.
    {
      // Create a threaded Consumer_Handler.
      if (ACE_BIT_ENABLED (Options::instance ()->threading_strategy (),
                           Options::OUTPUT_MT))
        ACE_NEW_RETURN (connection_handler,
                        Thr_Consumer_Handler (pci),
                        0);

      // Create a reactive Consumer_Handler.
      else
        ACE_NEW_RETURN (connection_handler,
                        Consumer_Handler (pci),
                        0);
    }
  else // connection_role == 'S', so configure a Supplier_Handler.
    {
      // Create a threaded Supplier_Handler.
      if (ACE_BIT_ENABLED (Options::instance ()->threading_strategy (),
                           Options::INPUT_MT))
        ACE_NEW_RETURN (connection_handler,
                        Thr_Supplier_Handler (pci),
                        0);

      // Create a reactive Supplier_Handler.
      else
        ACE_NEW_RETURN (connection_handler,
                        Supplier_Handler (pci),
                        0);
    }

  return connection_handler;
}

static void
handle_data (size_t &n_handles)
{
  // Handle pending logging messages first (s_handle + 1 is guaranteed
  // to be lowest client descriptor).

  for (size_t index = 1; index < n_handles; index++)
    {
      if (ACE_BIT_ENABLED (poll_array[index].revents, POLLIN))
        {
          // First time in, we need to initialize the buffer.
          if (buffer_array[index].buf_ == 0
              && init_buffer (index) == -1)
            {
              ACE_ERROR ((LM_ERROR,
                          "(%P|%t) %p\n",
                          "init_buffer"));
              continue;
            }

          // Read data from client (terminate on error).

          ssize_t n = ACE::recv (poll_array[index].fd,
                                 buffer_array[index].buf_,
                                 buffer_array[index].len_);
          // <recv> will not block in this case!

          if (n == -1)
            ACE_ERROR ((LM_ERROR,
                        "%p\n",
                        "read failed"));
          else if (n == 0)
            {
              ACE_DEBUG ((LM_DEBUG,
                          "(%P|%t) closing oneway server at handle %d\n",
                          poll_array[index].fd));

              // Handle client connection shutdown.
              ACE_OS::close (poll_array[index].fd);
              poll_array[index].fd = poll_array[--n_handles].fd;

              ACE_OS::free ((void *) buffer_array[index].buf_);
              buffer_array[index].buf_ = 0;
              buffer_array[index].len_ = 0;
            }
          else if (verbose)
            ACE_DEBUG ((LM_DEBUG,
                        "(%P|%t) %*s",
                        n,
                        buffer_array[index].buf_));
        }
    }
}

int
Sender::initiate_io (ACE_Reactor_Mask mask)
{
  if (ACE_BIT_ENABLED (flg_mask_, mask))
    return 0;

  if (ACE_Reactor::instance ()->schedule_wakeup (this, mask) == -1)
    return -1;

  ACE_SET_BITS (flg_mask_, mask);
  return 0;
}

void
MCT_Config::dump (void) const
{
  ACE_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" Dumping MCT_Config\n")));
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\tIP_MAX_MEMBERSHIPS = %d\n"),
              IP_MAX_MEMBERSHIPS));
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\tgroups_ = %d\n"),
              this->groups_));
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\trole_ = %s\n"),
              (ACE_BIT_ENABLED (this->role_, PRODUCER)
               && ACE_BIT_ENABLED (this->role_, CONSUMER))
                 ? ACE_TEXT ("PRODUCER/CONSUMER")
                 : ACE_BIT_ENABLED (this->role_, PRODUCER)
                     ? ACE_TEXT ("PRODUCER")
                     : ACE_TEXT ("CONSUMER")));
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\tsdm_options_ = %d\n"),
              this->sdm_opts_));
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\titerations_ = %d\n"),
              this->iterations_));
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\tttl_ = %d\n"),
              this->ttl_));
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\twait_ = %d\n"),
              this->wait_));
  // Note that this call to get_host_addr is the non-reentrant
  // version, but it's okay for us.
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("\tgroups_start_ = %s:%d\n"),
              this->group_start_.get_host_addr (),
              this->group_start_.get_port_number ()));

  ACE_DEBUG ((LM_DEBUG, ACE_END_DUMP));
}

template <ACE_SYNCH_DECL, class TIME_POLICY> int
ACE_Stream_Tail<ACE_SYNCH_USE, TIME_POLICY>::canonical_flush (ACE_Message_Block *mb)
{
  ACE_TRACE ("ACE_Stream_Tail<ACE_SYNCH_USE, TIME_POLICY>::canonical_flush");
  char *cp = mb->rd_ptr ();

  if (ACE_BIT_ENABLED (*cp, ACE_Task_Flags::ACE_FLUSHW))
    {
      this->flush (ACE_Task_Flags::ACE_FLUSHALL);
      ACE_CLR_BITS (*cp, ACE_Task_Flags::ACE_FLUSHW);
    }

  if (ACE_BIT_ENABLED (*cp, ACE_Task_Flags::ACE_FLUSHR))
    {
      this->sibling ()->flush (ACE_Task_Flags::ACE_FLUSHALL);
      return this->reply (mb);
    }
  else
    mb->release ();

  return 0;
}

int
ACE_Process_Manager::reap (pid_t pid,
                           ACE_exitcode *stat_loc,
                           int options)
{
  ACE_TRACE ("ACE_Process_Manager::reap");

  return this->wait (pid,
                     (ACE_BIT_ENABLED (options, WNOHANG)
                      ? ACE_Time_Value::zero
                      : ACE_Time_Value::max_time),
                     stat_loc);
}

template <ACE_SYNCH_DECL> int
ACE_Module<ACE_SYNCH_USE>::close_i (int which,
                                    int flags)
{
    ACE_TRACE ("ACE_Module<ACE_SYNCH_USE>::close_i");

    if (this->q_pair_[which] == 0)
        return 0;

    // Copy task pointer to prevent problems when ACE_Task::close
    // changes the task pointer
    ACE_Task<ACE_SYNCH_USE> *task = this->q_pair_[which];

    // Change so that close doesn't get called again from the task base.

    // Now close the task.
    int result = 0;

    if (task->module_closed () == -1)
        result = -1;

    task->flush ();
    task->next (0);

    // Should we also delete it ?
    if (flags != M_DELETE_NONE
            && ACE_BIT_ENABLED (flags_, which + 1))
    {
        // Only delete the Tasks if there aren't any more threads
        // running in them.
        task->wait ();

        // If this assert happens it is likely because the task was
        // activated with the THR_DETACHED flag, which means that we
        // can't join() with the thread.  Not using THR_DETACHED should
        // solve this problem.
        ACE_ASSERT (task->thr_count () == 0);

        delete task;
    }

    // Set the tasks pointer to 0 so that we don't try to close()
    // this object again if the destructor gets called.
    this->q_pair_[which] = 0;

    // Finally remove the delete bit.
    ACE_CLR_BITS (flags_, which + 1);

    return result;
}

  bool
  Eager_Transport_Queueing_Strategy::timer_check (
    const TAO::BufferingConstraint &buffering_constraint,
    const ACE_Time_Value &current_deadline,
    bool &set_timer,
    ACE_Time_Value &new_deadline) const
  {
    set_timer = false;

    if (!ACE_BIT_ENABLED (buffering_constraint.mode,
                          TAO::BUFFER_TIMEOUT))
      {
        return false;
      }

    // Compute the next deadline...
    ACE_Time_Value const now = ACE_OS::gettimeofday ();
    ACE_Time_Value timeout =
      this->time_conversion (buffering_constraint.timeout);
    new_deadline = now + timeout;

    // Check if the new deadline is more stringent, or if the deadline
    // has expired and thus must be reset anyway.
    if (current_deadline > new_deadline
        || current_deadline < now)
      {
        set_timer = true;
      }

    // ... if there is no deadline we don't want to schedule output (the
    // deadline will be set because set_timer is set to 1 in that case).
    // If there is a deadline but but it has not been reached, we
    // don't want to schedule any output either...
    if (current_deadline == ACE_Time_Value::zero
        || current_deadline >= now)
      {
        return false;
      }

    if (TAO_debug_level > 6)
      {
        TAOLIB_DEBUG ((LM_DEBUG,
                    "TAO (%P|%t) - TAO_Eager_Buffering_Sync_Strategy::timer_check, "
                    "Now = %u, Current = %u, New = %u\n",
                    now.msec (), current_deadline.msec (),
                    new_deadline.msec ()));
      }

    return true;
  }

int
ACE_Service_Type_Impl::fini (void) const
{
  ACE_TRACE ("ACE_Service_Type_Impl::fini");

  delete [] const_cast <ACE_TCHAR *> (this->name_);
  (const_cast <ACE_Service_Type_Impl *> (this))->name_ = 0;

  if (ACE_BIT_ENABLED (this->flags_,
                       ACE_Service_Type::DELETE_OBJ))
    {
      if (gobbler_ != 0)
        gobbler_ (this->object ());
      else
        // Cast to remove const-ness.
        operator delete ((void *) this->object ());
    }

  if (ACE_BIT_ENABLED (this->flags_,
                       ACE_Service_Type::DELETE_THIS))
    delete const_cast <ACE_Service_Type_Impl *> (this);

  return 0;
}

int
ACE_XtReactor::compute_Xt_condition(ACE_HANDLE handle)
{
  ACE_TRACE ("ACE_XtReactor::compute_Xt_condition");

  // Retrieve current wait mask from base class.
  // The returned value is either a combination of READ/WRITE/EXCEPT_MASK
  // or -1.
  int mask =this->bit_ops(handle,
                          0,
                          this->wait_set_,
                          ACE_Reactor::GET_MASK);

  if (mask == -1) // No active mask.
    return 0;

 int condition = 0;

#if !defined ACE_WIN32
  if (ACE_BIT_ENABLED (mask, ACE_Event_Handler::READ_MASK))
    ACE_SET_BITS (condition, XtInputReadMask);
  if (ACE_BIT_ENABLED (mask, ACE_Event_Handler::WRITE_MASK))
    ACE_SET_BITS (condition, XtInputWriteMask);
  if (ACE_BIT_ENABLED (mask, ACE_Event_Handler::EXCEPT_MASK))
    ACE_SET_BITS (condition, XtInputExceptMask);
#else
  if (ACE_BIT_ENABLED (mask, ACE_Event_Handler::READ_MASK))
    ACE_SET_BITS (condition, XtInputReadWinsock);
  if (ACE_BIT_ENABLED (mask, ACE_Event_Handler::WRITE_MASK))
    ACE_SET_BITS (condition, XtInputWriteWinsock);
  // EXCEPT_MASK is not supported for WIN32. As this was
  // already handled in register_handler_i, no check here.
#endif /* !ACE_WIN32 */

  return condition;
}

TAO_Queued_Data *
TAO_Queued_Data::duplicate (TAO_Queued_Data &sqd)
{
  // Check to see if the underlying block is on the stack. If not it
  // is fine. If the datablock is on stack, try to make a copy of that
  // before doing a duplicate.
  // @@ todo: Theoretically this should be within the Message Block,
  // but we dont have much scope to do this in that mess. Probably in
  // the next stage of MB rewrite we should be okay
  ACE_Message_Block::Message_Flags fl =
    sqd.msg_block_->self_flags ();

  if (ACE_BIT_ENABLED (fl,
                       ACE_Message_Block::DONT_DELETE))
    (void) TAO_Queued_Data::replace_data_block (*sqd.msg_block_);


  TAO_Queued_Data *qd = 0;

  if (sqd.allocator_)
    {
      ACE_NEW_MALLOC_RETURN (qd,
                             static_cast<TAO_Queued_Data *> (
                               sqd.allocator_->malloc (sizeof (TAO_Queued_Data))),
                             TAO_Queued_Data (sqd),
                             0);

      return qd;
    }

  // No allocator, so use the global pool!
  // @@ TODO: We should be removing this at some point of time!
  if (TAO_debug_level == 4)
    {
      // This debug is for testing purposes!
      TAOLIB_DEBUG ((LM_DEBUG,
                  "TAO (%P|%t) - Queued_Data[%d]::duplicate\n",
                  "Using global pool for allocation\n"));
    }

  ACE_NEW_RETURN (qd,
                  TAO_Queued_Data (sqd),
                  0);

  return qd;
}

int
TRB_POSIX_Proactor::register_handle (ACE_HANDLE  handle,
                                     const void *completion_key,
                                     int         operations)
{
  TRB_POSIX_AIO_Processor * aio_processor = 
    this->find_processor(operations);

  if (aio_processor != 0)
     return aio_processor->register_handle (handle, 
                                            completion_key,
                                            operations);

  if (ACE_BIT_ENABLED(operations, TRB_Asynch_Result::OP_READ_MASK |
                                  TRB_Asynch_Result::OP_WRITE_MASK))
    {
      return -1;  // we need provider for these operations
    } 

  return 0;
}

int
ACE_Service_Object_Type::fini (void) const
{
    ACE_TRACE ("ACE_Service_Object_Type::fini");

    void *obj = this->object ();

    ACE_Service_Object *so = (ACE_Service_Object *) obj;

    if (so)
    {
        so->fini ();

#if 0
        if (ACE_BIT_ENABLED (this->flags_,
                             ACE_Service_Type::DELETE_OBJ))
            delete so;
#endif /* 1 */
    }

    return ACE_Service_Type_Impl::fini ();
}

int
ACE_QtReactor::set_enable_flag_by_mask (int flag_value,
                                        ACE_HANDLE handle,
                                        ACE_Reactor_Mask mask)
{
  QSocketNotifier *qs_not;

  if (ACE_BIT_ENABLED(mask, ACE_Event_Handler::READ_MASK) ||
      ACE_BIT_ENABLED( mask, ACE_Event_Handler::ACCEPT_MASK))
    {
      // Find the current notifier
      qs_not = 0;
      if ((this->read_notifier_.find (handle, qs_not) == -1))
        return -1;

      qs_not->setEnabled (flag_value);
    }

  if (ACE_BIT_ENABLED( mask, ACE_Event_Handler::WRITE_MASK) ||
      ACE_BIT_ENABLED( mask, ACE_Event_Handler::ACCEPT_MASK) ||
      ACE_BIT_ENABLED( mask, ACE_Event_Handler::CONNECT_MASK))
    {
      qs_not = 0;
      if ((this->write_notifier_.find (handle, qs_not) == -1))
        return -1;

      qs_not->setEnabled (flag_value);
    }

  if (ACE_BIT_ENABLED( mask,
                       ACE_Event_Handler::EXCEPT_MASK))
    {
      qs_not = 0;
      if ((this->exception_notifier_.find (handle, qs_not) == -1))
        return -1;

      qs_not->setEnabled (flag_value);
    }

  return 0;
}