1 files changed, 399 insertions, 0 deletions
diff --git a/ACE/tests/Priority_Reactor_Test.cpp b/ACE/tests/Priority_Reactor_Test.cpp
new file mode 100644
index 00000000000..463b43a22e7
--- /dev/null
+++ b/ACE/tests/Priority_Reactor_Test.cpp
@@ -0,0 +1,399 @@
+// $Id$
+
+// ============================================================================
+//
+// = LIBRARY
+//    tests
+//
+// = FILENAME
+//    Priority_Reactor_Test.cpp
+//
+// = DESCRIPTION
+//     This is a test of the <ACE_Priority_Reactor>.  The test forks
+//     two processes (for a total of three processes) which connect to
+//     the main process and The clients send data to a connector,
+//     interestingly enough the acceptor will give more priority to
+//     the second connection, which should run always before the first
+//     one.
+//
+//     The test itself is interesting, it shows how to write very
+//     simple <ACE_Svc_Handler>, <ACE_Connectors> and <ACE_Acceptors>.
+//
+// = AUTHOR
+//    Carlos O'Ryan <coryan@cs.wustl.edu>
+//
+// ============================================================================
+
+#include "test_config.h"
+#include "ace/Get_Opt.h"
+#include "ace/SOCK_Connector.h"
+#include "ace/SOCK_Acceptor.h"
+#include "ace/Acceptor.h"
+#include "ace/Handle_Set.h"
+#include "ace/Connector.h"
+#include "ace/Auto_Ptr.h"
+#include "ace/Priority_Reactor.h"
+#include "Priority_Reactor_Test.h"
+#include "ace/OS_NS_sys_wait.h"
+#include "ace/OS_NS_unistd.h"
+
+ACE_RCSID(tests, Priority_Reactor_Test, "$Id$")
+
+static const char ACE_ALPHABET[] = "abcdefghijklmnopqrstuvwxyz";
+
+// The number of children to run, it can be changed using the -c
+// option.
+static int opt_nchildren = 10;
+
+// The number of loops per children, it can be changed using the -l
+// option.
+static int opt_nloops = 200;
+
+// If not set use the normal reactor, it can be changed using the -d
+// option.
+static int opt_priority_reactor = 1;
+
+// Maximum time to wait for the test termination (-t)
+static int opt_max_duration = 60;
+
+// Maximum number of retries to connect, it can be changed using the
+// -m option.
+static int max_retries = 5;
+
+typedef ACE_Connector<Write_Handler, ACE_SOCK_CONNECTOR>
+        CONNECTOR;
+typedef ACE_Acceptor<Read_Handler, ACE_SOCK_ACCEPTOR>
+        ACCEPTOR;
+
+int Read_Handler::waiting_ = 0;
+int Read_Handler::started_ = 0;
+
+void
+Read_Handler::set_countdown (int nchildren)
+{
+  Read_Handler::waiting_ = nchildren;
+}
+
+int
+Read_Handler::get_countdown (void)
+{
+  return Read_Handler::waiting_;
+}
+
+int
+Read_Handler::open (void *)
+{
+  if (this->peer ().enable (ACE_NONBLOCK) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%P|%t) Read_Handler::open, ")
+                       ACE_TEXT ("cannot set non blocking mode")),
+                      -1);
+
+  if (reactor ()->register_handler (this, READ_MASK) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%P|%t) Read_Handler::open, ")
+                       ACE_TEXT ("cannot register handler")),
+                      -1);
+
+  // A number larger than the actual number of priorities, so some
+  // clients are misbehaved, hence pusnished.
+  const int max_priority = 15;
+
+  this->priority (ACE_Event_Handler::LO_PRIORITY + started_ % max_priority);
+  started_++;
+
+  ACE_DEBUG ((LM_DEBUG,
+              ACE_TEXT ("(%P|%t) created svc_handler for handle %d ")
+              ACE_TEXT ("with priority %d\n"),
+              get_handle (),
+              priority ()));
+  return 0;
+}
+
+int
+Read_Handler::handle_input (ACE_HANDLE h)
+{
+  // ACE_DEBUG((LM_DEBUG,
+  // "(%P|%t) Read_Handler::handle_input (%d)\n", h));
+  ACE_UNUSED_ARG (h);
+
+  char buf[BUFSIZ];
+
+  ssize_t result = this->peer ().recv (buf, sizeof (buf));
+
+  if (result <= 0)
+    {
+      if (result < 0 && errno == EWOULDBLOCK)
+        return 0;
+
+      if (result != 0)
+        ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) %p\n"),
+                    ACE_TEXT ("Read_Handler::handle_input")));
+      waiting_--;
+
+      if (waiting_ == 0)
+        {
+          ACE_DEBUG ((LM_DEBUG,
+                      ACE_TEXT ("Last svc_handler closed, shutting down\n")));
+          ACE_Reactor::instance()->end_reactor_event_loop();
+        }
+
+      ACE_DEBUG ((LM_DEBUG,
+                  ACE_TEXT ("(%P|%t) Read_Handler::handle_input closing down\n")));
+      return -1;
+    }
+
+  // ACE_DEBUG((LM_DEBUG,
+  // "(%P|%t) read %d bytes from handle %d, priority %d\n",
+  // result, h, priority ()));
+  return 0;
+}
+
+int
+Write_Handler::open (void *)
+{
+  return 0;
+}
+
+int
+Write_Handler::svc (void)
+{
+  // Send several short messages, doing pauses between each message.
+  // The number of messages can be controlled from the command line.
+  ACE_Time_Value pause (0, 1000);
+
+  for (int i = 0; i < opt_nloops; ++i)
+    {
+      if (this->peer ().send_n (ACE_ALPHABET,
+                                sizeof (ACE_ALPHABET) - 1) == -1)
+          ACE_ERROR ((LM_ERROR,
+                      ACE_TEXT ("(%P|%t) %p\n"),
+                      ACE_TEXT ("send_n")));
+      ACE_OS::sleep (pause);
+    }
+
+  return 0;
+}
+
+#if !defined (ACE_LACKS_FORK) || defined (ACE_HAS_THREADS)
+
+// Execute the client tests.
+static void *
+client (void *arg)
+{
+  ACE_INET_Addr *connection_addr =
+    reinterpret_cast<ACE_INET_Addr *> (arg);
+  ACE_DEBUG ((LM_DEBUG,
+              ACE_TEXT ("(%P|%t) running client\n")));
+  CONNECTOR connector;
+
+  Write_Handler *writer = 0;
+
+  // Do exponential backoff connections
+  ACE_Synch_Options options = ACE_Synch_Options::synch;
+
+  // Start with one msec timeouts.
+  ACE_Time_Value msec (0, 1000);
+  options.timeout (msec);
+
+  // Try up to <max_retries> to connect to the server.
+  for (int i = 0; i < max_retries; i++)
+    {
+      if (connector.connect (writer,
+                             *connection_addr,
+                             options) == -1)
+        {
+          // Double the timeout...
+          ACE_Time_Value tmp = options.timeout ();
+          tmp += options.timeout ();
+          options.timeout (tmp);
+          writer = 0;
+          ACE_DEBUG ((LM_DEBUG,
+                      ACE_TEXT ("(%P|%t) still trying to connect\n")));
+        }
+      else
+        {
+          // Let the new Svc_Handler to its job...
+          writer->svc ();
+
+          // then close the connection and release the Svc_Handler.
+          writer->destroy ();
+
+          ACE_DEBUG ((LM_DEBUG,
+                      ACE_TEXT ("(%P|%t) finishing client\n")));
+          return 0;
+        }
+    }
+
+  ACE_ERROR ((LM_ERROR,
+              ACE_TEXT ("(%P|%t) failed to connect after %d retries\n"),
+              max_retries));
+  return 0;
+}
+
+#endif
+
+int
+run_main (int argc, ACE_TCHAR *argv[])
+{
+  ACE_START_TEST (ACE_TEXT ("Priority_Reactor_Test"));
+
+  //FUZZ: disable check_for_lack_ACE_OS
+  ACE_Get_Opt getopt (argc, argv, ACE_TEXT ("dc:l:m:t:"));
+
+  for (int c; (c = getopt ()) != -1; )
+    switch (c)
+      {
+  //FUZZ: enable check_for_lack_ACE_OS
+      case 'd':
+        opt_priority_reactor = 0;
+        break;
+      case 'c':
+        opt_nchildren = ACE_OS::atoi (getopt.opt_arg ());
+        break;
+      case 'l':
+        opt_nloops = ACE_OS::atoi (getopt.opt_arg ());
+        break;
+      case 'm':
+        max_retries = ACE_OS::atoi (getopt.opt_arg ());
+        break;
+      case 't':
+        opt_max_duration = ACE_OS::atoi (getopt.opt_arg ());
+        break;
+      case '?':
+      default:
+        ACE_ERROR_RETURN ((LM_ERROR,
+                           ACE_TEXT ("Usage: Priority_Reactor_Test ")
+                           ACE_TEXT ("   [-d] (disable priority reactor)\n")
+                           ACE_TEXT ("   [-c nchildren] (number of threads/processes)\n")
+                           ACE_TEXT ("   [-l loops] (number of loops per child)\n")
+                           ACE_TEXT ("   [-m maxretries] (attempts to connect)\n")
+                           ACE_TEXT ("   [-t max_time] (limits test duration)\n")),
+                          -1);
+        ACE_NOTREACHED (break);
+      }
+
+  // Manage Reactor memory automagically.
+  // Note:  If opt_priority_reactor is false, the default ACE_Reactor is used
+  // and we don't need to set one up.
+  ACE_Reactor *orig_reactor = 0;
+  auto_ptr<ACE_Reactor> reactor;
+
+  if (opt_priority_reactor)
+    {
+      ACE_Select_Reactor *impl_ptr;
+      ACE_NEW_RETURN (impl_ptr, ACE_Priority_Reactor, -1);
+      auto_ptr<ACE_Select_Reactor> auto_impl (impl_ptr);
+
+      ACE_Reactor *reactor_ptr;
+      ACE_NEW_RETURN (reactor_ptr, ACE_Reactor (impl_ptr, 1), -1);
+      auto_impl.release ();   // ACE_Reactor dtor will take it from here
+      auto_ptr<ACE_Reactor> auto_reactor (reactor_ptr);
+      reactor = auto_reactor;
+      orig_reactor = ACE_Reactor::instance (reactor_ptr);
+    }
+
+  Read_Handler::set_countdown (opt_nchildren);
+
+  // Acceptor
+  ACCEPTOR acceptor;
+
+  acceptor.priority (ACE_Event_Handler::HI_PRIORITY);
+  ACE_INET_Addr server_addr;
+
+  // Bind acceptor to any port and then find out what the port was.
+  if (acceptor.open (ACE_sap_any_cast (const ACE_INET_Addr &)) == -1
+      || acceptor.acceptor ().get_local_addr (server_addr) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%P|%t) %p\n"),
+                       ACE_TEXT ("open")),
+                      -1);
+
+  ACE_DEBUG ((LM_DEBUG,
+              ACE_TEXT ("(%P|%t) starting server at port %d\n"),
+              server_addr.get_port_number ()));
+
+  ACE_INET_Addr connection_addr (server_addr.get_port_number (),
+                                 ACE_DEFAULT_SERVER_HOST);
+
+  int i;
+
+#if defined (ACE_HAS_THREADS)
+  for (i = 0; i < opt_nchildren; ++i)
+    {
+      if (ACE_Thread_Manager::instance ()->spawn
+          (ACE_THR_FUNC (client),
+           (void *) &connection_addr,
+           THR_NEW_LWP | THR_DETACHED) == -1)
+        ACE_ERROR ((LM_ERROR,
+                    ACE_TEXT ("(%P|%t) %p\n%a"),
+                    ACE_TEXT ("thread create failed"),
+                    1));
+    }
+#elif !defined (ACE_LACKS_FORK)
+  for (i = 0; i < opt_nchildren; ++i)
+    {
+      switch (ACE_OS::fork ("child"))
+        {
+        case -1:
+          ACE_ERROR ((LM_ERROR,
+                      ACE_TEXT ("(%P|%t) %p\n%a"),
+                      ACE_TEXT ("fork failed"),
+                      1));
+          ACE_OS::exit (-1);
+          /* NOTREACHED */
+        case 0:
+          client (&connection_addr);
+          ACE_OS::exit (0);
+          break;
+          /* NOTREACHED */
+        default:
+          break;
+          /* NOTREACHED */
+        }
+    }
+#else
+  ACE_ERROR ((LM_INFO,
+              ACE_TEXT ("(%P|%t) ")
+              ACE_TEXT ("only one thread may be run ")
+              ACE_TEXT ("in a process on this platform\n")));
+#endif /* ACE_HAS_THREADS */
+
+  ACE_Time_Value tv (opt_max_duration);
+
+  ACE_Reactor::instance()->register_handler
+    (&acceptor, ACE_Event_Handler::READ_MASK);
+  ACE_Reactor::instance()->run_reactor_event_loop (tv);
+
+  if (Read_Handler::get_countdown () != 0)
+    {
+      ACE_DEBUG ((LM_DEBUG,
+                  ACE_TEXT ("(%P|%t) running out of time, ")
+                  ACE_TEXT ("probably due to failed connections.\n")));
+    }
+
+  ACE_DEBUG ((LM_DEBUG,
+              ACE_TEXT ("(%P|%t) waiting for the children...\n")));
+
+#if defined (ACE_HAS_THREADS)
+  ACE_Thread_Manager::instance ()->wait ();
+#elif !defined (ACE_WIN32) && !defined (VXWORKS)
+  for (i = 0; i < opt_nchildren; ++i)
+    {
+      pid_t pid = ACE_OS::wait();
+      ACE_DEBUG ((LM_DEBUG,
+                  ACE_TEXT ("(%P|%t) child %d terminated\n"),
+                  pid));
+    }
+#else
+  /* NOTREACHED */
+  // We aborted on the previous #ifdef
+#endif /* ACE_HAS_THREADS */
+
+  if (orig_reactor != 0)
+    ACE_Reactor::instance (orig_reactor);
+
+  ACE_END_TEST;
+  return 0;
+}
+