int
Supplier_Task::open (void *)
{
  // Create the pipe.
  int result;

  result = this->pipe_.open ();
  ACE_TEST_ASSERT (result != -1);

  // Register the pipe's write handle with the <Reactor> for writing.
  // This should mean that it's always "active."
  if (long_timeout_ == 0)
    {
      result = ACE_Reactor::instance ()->register_handler
        (this->pipe_.write_handle (),
         this,
         ACE_Event_Handler::WRITE_MASK);
      ACE_TEST_ASSERT (result != -1);
    }

  // Make this an Active Object.
  result = this->activate (THR_BOUND | THR_DETACHED);
  ACE_TEST_ASSERT (result != -1);
  return 0;
}

int
Supplier_Task::close (u_long)
{
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%t) Supplier_Task::close\n")));

  int result;

  if (long_timeout_ == 0)
    {
      result = ACE_Reactor::instance ()->remove_handler
        (this->pipe_.write_handle (),
         ACE_Event_Handler::WRITE_MASK);
      ACE_TEST_ASSERT (result != -1);
    }
  else
    {
      // Wait to be told to shutdown by the main thread.

      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("(%t) waiting to be shutdown by main thread\n")));
      result = this->waiter_.acquire ();
      ACE_TEST_ASSERT (result != -1);
    }
  return 0;
}

static void *
child (void * = 0)
{
  int result;

  // Wait for the parent to be initialized.
  result = synchronizer->acquire ();
  ACE_TEST_ASSERT (result != -1);

  const char *t = ACE_ALPHABET;
  ACE_Shared_Memory_MM shm_child;

  result = shm_child.open (shm_key);
  ACE_TEST_ASSERT (result != -1);

  char *shm = (char *) shm_child.malloc ();

  ACE_TEST_ASSERT (shm != 0);

  for (char *s = shm; *s != '\0'; s++)
    {
      ACE_TEST_ASSERT (*t == s[0]);
      t++;
    }

  // Indicate to the parent that we're done.
  *shm = '*';

  return 0;
}

static void *
parent (void * = 0)
{
  int result;
  ACE_Shared_Memory_MM shm_parent;

  result = shm_parent.open (shm_key, SHMSZ);
  ACE_TEST_ASSERT (result != -1);

  char *shm = (char *) shm_parent.malloc ();

  ACE_TEST_ASSERT (shm != 0);

  char *s = shm;

  for (const char *c = ACE_ALPHABET; *c != '\0'; c++)
    *s++ = *c;

  *s = '\0';

  // Allow the child to proceed.
  result = synchronizer->release ();
  ACE_TEST_ASSERT (result != -1);

  // Perform a "busy wait" until the child sets the character to '*'.
  while (*shm != '*')
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P) spinning in parent!\n")));

  result = shm_parent.remove ();
  ACE_TEST_ASSERT (result != -1);

  ACE_OS::unlink (shm_key);
  return 0;
}

static void
purge_test_hash_cache (HASH_MAP_CACHE &cache)
{
  // Get the number of entries in the container.
  size_t current_map_size = cache.current_size ();

  // Find the number of entries which will get purged.
  size_t entries_to_remove = size_t ((double (purge_percent) / 100 * current_map_size) + 0.5);

  // Tell the caching strategy how much to purge.
  cache.caching_strategy ().purge_percent (purge_percent);

  // Purge from cache.
  int result = cache.purge ();
  ACE_TEST_ASSERT (result != -1);
  ACE_UNUSED_ARG (result);

  size_t resultant_size = 0;
  if (caching_strategy_type == ACE_NULL)
    resultant_size = current_map_size;
  else
    resultant_size = current_map_size - entries_to_remove;

  // Make sure the purge took out the appropriate number of entries.
  ACE_TEST_ASSERT (cache.current_size () == resultant_size);
  ACE_UNUSED_ARG (resultant_size);
}

static void
run_reverse_iterator_hash_cache (HASH_MAP_CACHE &cache)
{
  size_t counter = cache.current_size ();
  HASH_MAP_CACHE::reverse_iterator rend = cache.rend ();

  for (HASH_MAP_CACHE::reverse_iterator iter = cache.rbegin ();
       iter != rend;
       ++iter)
    {
      ACE_TEST_ASSERT ((*iter).first () == (*iter).second ());

      // Debugging info.
      if (debug)
        ACE_DEBUG ((LM_DEBUG,
                    ACE_TEXT ("(%d|%d)"),
                    (*iter).first (),
                    (*iter).second ()));
      --counter;
    }

  if (debug)
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("\n")));

  ACE_TEST_ASSERT (counter == 0);
}

static void
run_iterator_hash_cache (HASH_MAP_CACHE &cache)
{
  size_t iterations = cache.current_size ();
  size_t counter = 0;
  HASH_MAP_CACHE::iterator end = cache.end ();

  for (HASH_MAP_CACHE::iterator iter = cache.begin ();
       iter != end;
       ++iter)
    {
      // Debugging info.
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("(%d|%d)"),
                  (*iter).first (),
                  (*iter).second ()));

      ACE_TEST_ASSERT ((*iter).first () == (*iter).second ());
      ++counter;
    }

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

  ACE_UNUSED_ARG (iterations);
  ACE_TEST_ASSERT (counter == iterations);
}

static void *
connector (void *)
{
  ACE_UPIPE_Stream c_stream;

  ACE_OS::sleep (5);

  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) connector starting connect\n")));
  ACE_UPIPE_Connector con;

  if (con.connect (c_stream, addr) == -1)
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) connector ACE_UPIPE_Connector failed\n")));

  ACE_Message_Block *mb = 0;

  ACE_NEW_RETURN (mb, ACE_Message_Block (sizeof ("hello thanks") * sizeof (char)), 0);

  mb->copy ("hello");

  if (c_stream.send (mb) == -1)
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) error connector send\n")));

  if (c_stream.recv (mb) == -1)
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) error connector recv\n")));

  ACE_TEST_ASSERT (ACE_OS::strcmp (mb->rd_ptr (), "thanks") == 0);

  // Free up the memory block.
  mb->release ();

  // Now try the send()/recv() interface.
  char mytext[] = "This string is sent by connector as a buffer";

  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) connector sending text\n")));
  if (c_stream.send (mytext, sizeof (mytext)) == -1)
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%t) buffer send from connector failed\n")));

  char conbuf[BUFSIZ];  // Buffer to receive response.

  int i = 0;

  for (char c = ' '; c != '!'; i++)
    {
      if (c_stream.recv (&c, 1) == -1)
        ACE_DEBUG ((LM_DEBUG,
                    ACE_TEXT ("(%t) buffer recv from connector failed\n")));
      else
        conbuf[i] = c;
    }

  conbuf[i] = '\0';
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) conbuf = %s\n"), conbuf));
  ACE_TEST_ASSERT (ACE_OS::strcmp (conbuf, "this is the acceptor response!") == 0);

  c_stream.close ();
  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) exiting thread\n")));
  return 0;
}

int
run_main (int, ACE_TCHAR *[])
{
  ACE_START_TEST (ACE_TEXT ("Reactors_Test"));

#if defined (ACE_HAS_THREADS)
  ACE_TEST_ASSERT (ACE_LOG_MSG->op_status () != -1);

  thr_mgr = ACE_Thread_Manager::instance ();

  ACE_Reactor reactor;
  ACE_TEST_ASSERT (ACE_LOG_MSG->op_status () != -1);

  Test_Task tt1[MAX_TASKS];
  Test_Task tt2[MAX_TASKS];

  // Activate all of the Tasks.

  for (int i = 0; i < MAX_TASKS; i++)
    {
      tt1[i].open (ACE_Reactor::instance ());
      tt2[i].open (&reactor);
    }

  // Spawn two threads each running a different reactor.

  if (ACE_Thread_Manager::instance ()->spawn
      (ACE_THR_FUNC (worker),
       (void *) ACE_Reactor::instance (),
       THR_BOUND | THR_DETACHED) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("%p\n"),
                       ACE_TEXT ("spawn")),
                      -1);

  else if (ACE_Thread_Manager::instance ()->spawn
      (ACE_THR_FUNC (worker), (void *) &reactor,
       THR_BOUND | THR_DETACHED) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("%p\n"),
                       ACE_TEXT ("spawn")),
                      -1);

  if (ACE_Thread_Manager::instance ()->wait () == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("%p\n"),
                       ACE_TEXT ("wait")),
                      -1);

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%t) all threads are finished\n")));

#else
  ACE_ERROR ((LM_INFO,
              ACE_TEXT ("threads not supported on this platform\n")));
#endif /* ACE_HAS_THREADS */
  ACE_END_TEST;
  return 0;
}

Sender *
Invocation_Thread::create_connection (void)
{
  int result = 0;

  // Connector for creating new connections.
  Connector connector (this->thread_manager_,
                       this->reactor_,
                       this->nested_upcalls_);

  // <server_handle> is a global variable. It will be used later by
  // the Close_Socket_Thread.
  result =
    connector.connect (client_handle,
                       server_handle,
                       this->run_receiver_thread_);
  ACE_TEST_ASSERT (result == 0);
  ACE_UNUSED_ARG (result);

  // Create a new sender.
  Sender *sender =
    new Sender (client_handle,
                this->connection_cache_);

  // Register it with the cache.
  this->connection_cache_.add_connection (sender);

  //
  // There might be a race condition here. The sender has been added
  // to the cache and is potentially available to other threads
  // accessing the cache. Therefore, the other thread may use this
  // sender and potentially close the sender before it even gets
  // registered with the Reactor.
  //
  // This is resolved by marking the connection as busy when it is
  // first added to the cache. And only once the thread creating the
  // connection is done with it, it is marked a available in the
  // cache.
  //
  // This order of registration is important.
  //

  // Register the handle with the Reactor.
  result =
    this->reactor_.register_handler (client_handle,
                                     sender,
                                     ACE_Event_Handler::READ_MASK);
#if 0
  ACE_TEST_ASSERT (result == 0);
  ACE_UNUSED_ARG (result);
#else
  if (result != 0)
    ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%t) create_connection h %d, %p\n"),
                client_handle,
                ACE_TEXT ("register_handler")));
#endif
  return sender;
}

static void
run_test (ACE_THR_FUNC worker,
          long handle_signals_in_separate_thread,
          long handle_signals_synchronously)
{
#if defined (ACE_HAS_THREADS)
  if (handle_signals_synchronously)
    {
      // For the synchronous signal tests, block signals to prevent
      // asynchronous delivery to default handler (at least necessary
      // on linux and solaris; POSIX spec also states that signal(s)
      // should be blocked before call to sigwait())
      ACE_Sig_Guard guard;

      int result;

      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("(%P|%t) spawning worker thread\n")));
      result = ACE_Thread_Manager::instance ()->spawn
                (worker,
                  reinterpret_cast <void *> (handle_signals_synchronously),
                  THR_DETACHED);
      ACE_TEST_ASSERT (result != -1);

      if (handle_signals_in_separate_thread)
        {
          ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P|%t) spawning signal handler thread\n")));

          result = ACE_Thread_Manager::instance ()->spawn
            (synchronous_signal_handler,
             0,
             THR_DETACHED);
          ACE_TEST_ASSERT (result != -1);
        }
      else
        {
          synchronous_signal_handler (0);
        }

      // Wait for the thread(s) to finish.
      result = ACE_Thread_Manager::instance ()->wait ();
      ACE_TEST_ASSERT (result != -1);
    }
  else
#else
    // Don't remove this since otherwise some compilers give warnings
    // when ACE_HAS_THREADS is disabled!
    ACE_UNUSED_ARG (synchronous_signal_handler);
#endif /* ACE_HAS_THREADS */
    {
      ACE_UNUSED_ARG (handle_signals_in_separate_thread);
      // Arrange to handle signals asynchronously.
      asynchronous_signal_handler (0);
      (*worker) (reinterpret_cast <void *> (handle_signals_synchronously));
    }
}

static ACE_THR_FUNC_RETURN
worker_child (void *arg)
{
  long handle_signals_synchronously =
    reinterpret_cast <long> (arg);

  for (size_t i = 0; i < n_iterations; i++)
    {
      if (shut_down > 0)
        {
          ACE_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("(%P|%t) we've been shutdown!\n")));
          break;
        }

      // Every 100 iterations sleep for 2 seconds.
      if ((i % 100) == 0)
        {
          ACE_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("(%P|%t) sleeping for 2 seconds\n")));
          ACE_OS::sleep (2);
        }

      // After 1000 iterations sent a SIGHUP to our parent.
      if ((i % 1000) == 0)
        {
          ACE_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("(%P|%t) sending SIGHUP to parent process %d\n"),
                      parent_pid));
          int const result = ACE_OS::kill (parent_pid,
                                           SIGHUP);
          if (result == -1)
            {
              ACE_ERROR ((LM_ERROR,
                          ACE_TEXT ("(%P|%t) %p\n"),
                          ACE_TEXT ("kill")));
              ACE_TEST_ASSERT (result != -1);
            }
        }
    }

  if (handle_signals_synchronously)
    {
      if (!shut_down)
        {
          ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P|%t) sending SIGINT to ourselves\n")));
          // We need to do this to dislodge the signal handling thread if
          // it hasn't shut down on its own accord yet.
          int const result = ACE_OS::kill (ACE_OS::getpid (), SIGINT);
          ACE_TEST_ASSERT (result != -1);
        }
    }
  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%P|%t) finished running child\n")));
  return 0;
}

int run_main (int, ACE_TCHAR *[])
{
  ACE_START_TEST (ACE_TEXT ("Intrusive_Auto_Ptr_Test"));


  One *theone (new One(0));

  {
    ACE_TEST_ASSERT (theone->has_refs (0));
    ACE_TEST_ASSERT (!One::was_released ());

    ACE_Intrusive_Auto_Ptr<One> ip2(theone);

    {
      ACE_TEST_ASSERT (theone->has_refs (1));
      ACE_TEST_ASSERT (!One::was_released ());

      ACE_Intrusive_Auto_Ptr<One> ip2(theone);
      ACE_TEST_ASSERT (theone->has_refs (2));
      ACE_TEST_ASSERT (!One::was_released ());
    }

    ACE_TEST_ASSERT (theone->has_refs (1));
    ACE_TEST_ASSERT (!One::was_released ());
  }

  ACE_TEST_ASSERT (One::was_released());

  ACE_END_TEST;
  return 0;
}

static void
acquire_release (void)
{
  ACE_Process_Mutex mutex (mutex_name);

  // Make sure the constructor succeeded
  ACE_TEST_ASSERT (ACE_LOG_MSG->op_status () == 0);

  // To see if we really are the only holder of the mutex below,
  // we'll try to create a file with exclusive access. If the file
  // already exists, we're not the only one holding the mutex.
  ACE_TCHAR mutex_check[MAXPATHLEN+1];
  ACE_OS::strncpy (mutex_check, mutex_name, MAXPATHLEN);
  ACE_OS::strncat (mutex_check, ACE_TEXT ("_checker"), MAXPATHLEN);

  // Grab the lock
  int mutex_acq = mutex.acquire ();
  ACE_TEST_ASSERT (mutex_acq == 0);

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%P) Mutex acquired %s\n"),
              mutex_name));

  ACE_HANDLE checker_handle = ACE_OS::open (mutex_check, O_CREAT | O_EXCL);
  if (checker_handle == ACE_INVALID_HANDLE)
    {
      ACE_DEBUG ((LM_WARNING, ACE_TEXT ("(%P): %p\n"),
                  ACE_TEXT ("checker file open")));
      ACE_TEST_ASSERT (errno != EEXIST);
    }
  else
    ACE_OS::close (checker_handle);

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%P) Working....\n")));

  // Do some "work", i.e., just sleep for a couple of seconds.
  ACE_OS::sleep (2);

  // Free up the check file for the next acquirer.
  ACE_OS::unlink (mutex_check);

  // Check if we need to release the mutex
  if (release_mutex == 1)
    {
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("(%P) Releasing the mutex %s\n"),
                  mutex_name));
      int mutex_release = mutex.release ();
      ACE_TEST_ASSERT (mutex_release == 0);
    }
}

static void *
consumer (void *args)
{
  ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue =
    reinterpret_cast<ACE_Message_Queue<ACE_MT_SYNCH> *> (args);

  u_long cur_priority = 27;
  ACE_UNUSED_ARG (cur_priority);
  // To suppress ghs warning about unused local variable
  // "cur_priority".

  int local_count = 0;

  // Keep looping, reading a message out of the queue, until we get a
  // message with a length == 0, which signals us to quit.
  for (char c = 'z'; ; c--)
    {
      ACE_Message_Block *mb = 0;

      int result = msg_queue->dequeue_head (mb);

      if (result == -1)
        break;

      local_count++;

      size_t length = mb->length ();

      if (length > 0)
        {
          // This isn't a "shutdown" message, so process it
          // "normally."
          ACE_TEST_ASSERT (c == *mb->rd_ptr ());
          ACE_TEST_ASSERT (mb->msg_priority () < cur_priority);
          cur_priority = mb->msg_priority ();
        }

      // Free up the buffer memory and the Message_Block. Note that
      // the destructor of Message Block will delete the the actual
      // buffer.
      mb->release ();

      if (length == 0)
        // This was a "shutdown" message, so break out of the loop.
        break;
    }

  ACE_TEST_ASSERT (local_count == message_count);
  return 0;
}

static void
open_pipe (ACE_Pipe &pipe,
           const char *name)
{
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("opening %C\n"), name));
    int result = pipe.open ();

    ACE_TEST_ASSERT (result != -1);
    result = pipe.read_handle () != ACE_INVALID_HANDLE
             && pipe.write_handle () != ACE_INVALID_HANDLE;
    ACE_TEST_ASSERT (result == 1);

    if (close_pipe)
        pipe.close ();
}

int
Supplier_Task::perform_notifications (int notifications)
{
  ACE_Reactor::instance ()->max_notify_iterations (notifications);

  size_t iterations = ACE_MAX_ITERATIONS;

  if (this->long_timeout_)
    {
      iterations *= (iterations * iterations * 2);
#if defined (ACE_VXWORKS)
      // scale down otherwise the test won't finish in time
      iterations /= 4;
#endif
    }

  for (size_t i = 0; i < iterations; i++)
    {
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("(%t) notifying reactor on iteration %d\n"),
                  i));

      int result;

      // Notify the Reactor, which will call <handle_exception>.
      result = ACE_Reactor::instance ()->notify (this);
      if (result == -1)
        {
          if (errno == ETIME)
            ACE_DEBUG ((LM_DEBUG,
                        ACE_TEXT ("(%t) %p\n"),
                        ACE_TEXT ("notify")));
          else
            ACE_TEST_ASSERT (result != -1);
        }

      // Wait for our <handle_exception> method to release the
      // semaphore.
      if (this->long_timeout_ == 0
          && this->disable_notify_pipe_ == 0)
        {
          result = this->waiter_.acquire ();
          ACE_TEST_ASSERT (result != -1);
        }
    }

  return 0;
}

static void
test_interval_timer (ACE_Timer_Queue *tq)
{
  /*
    The strategy:

    Set up a timer to fire on a 50ms interval.
  */
  Interval_Handler ih;
  ACE_Time_Value interval (0, 50 * 1000 /* number of usec in millisecond */);
  const unsigned NUM_INTERVAL_FIRINGS = 50;
  ACE_Time_Value loop_stop_time =
    tq->gettimeofday () + (NUM_INTERVAL_FIRINGS * interval);
  const unsigned EXPECTED_TRIP_COUNT =
    NUM_INTERVAL_FIRINGS + 1 /* for the first immediate firing */;

  long id = tq->schedule (&ih, 0 /* no act */, ACE_Time_Value::zero, interval);
  ACE_TEST_ASSERT (id != -1);

  do
    {
      tq->expire ();
    }
  while (tq->gettimeofday () < loop_stop_time);

  ACE_DEBUG((LM_DEBUG,
             ACE_TEXT("after interval loop, timer fired %d ")
             ACE_TEXT("times out of %d expected: %s\n"),
             ih.trip_count_, EXPECTED_TRIP_COUNT,
             ih.trip_count_ == EXPECTED_TRIP_COUNT
             ? ACE_TEXT ("success") : ACE_TEXT ("FAIL")
             ));

  tq->cancel (id);
}

int
MyTask::create_reactor (void)
{
  ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX,
                    monitor,
                    this->lock_,
                    -1);

  ACE_TEST_ASSERT (this->my_reactor_ == 0);

  ACE_TP_Reactor * pImpl = 0;

  ACE_NEW_RETURN (pImpl,ACE_TP_Reactor (0, create_timer_queue ()), -1);

  ACE_NEW_RETURN (my_reactor_,
                   ACE_Reactor (pImpl ,1),
                   -1);

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT (" (%t) Create TP_Reactor\n")));

  this->reactor (my_reactor_);

  return 0;
}

int
Time_Handler::handle_timeout (const ACE_Time_Value &tv,
                              const void *arg)
{
  long current_count = static_cast<long> (reinterpret_cast<size_t> (arg));
  if (current_count >= 0)
    ACE_TEST_ASSERT (current_count == the_count);

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("[%x] Timer id %d with count #%d|%d timed out at %d!\n"),
              this,
              this->timer_id (),
              the_count,
              current_count,
              tv.sec ()));

  if (current_count == long (ACE_MAX_TIMERS - 1))
    done = 1;
  else if (the_count == ACE_MAX_TIMERS - 1)
    {
      done = 1;
      return -1;
    }
  else if (current_count == -1)
    {
      int result = ACE_Reactor::instance ()->reset_timer_interval (this->timer_id (),
                                                                   ACE_Time_Value (the_count + 1));
      if (result == -1)
        ACE_ERROR ((LM_ERROR,
                    ACE_TEXT ("Error resetting timer interval\n")));
    }
  the_count += (1 + odd);
  return 0;
}