ACE/tests/Reactor_Performance_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Reactor_Performance_Test.cpp
   5  *
   6  *  This test is used to time the dispatching mechanisms of the
   7  *  <ACE_Reactor>s. Both the <ACE_WFMO_Reactor> and
   8  *  <ACE_Select_Reactor> can be tested.
   9  *
  10  *  @author Irfan Pyarali <irfan@cs.wustl.edu>
  11  */
  12 //=============================================================================
  13
  14 #include "test_config.h"
  15 #include "Reactor_Performance_Test.h"
  16 #include "ace/Profile_Timer.h"
  17 #include "ace/Get_Opt.h"
  18 #include "ace/SOCK_Connector.h"
  19 #include "ace/SOCK_Acceptor.h"
  20 #include "ace/Acceptor.h"
  21 #include "ace/Connector.h"
  22 #include "ace/Reactor.h"
  23 #include "ace/WFMO_Reactor.h"
  24 #include "ace/Select_Reactor.h"
  25 #include "ace/Auto_Ptr.h"
  26
  27 #if defined (ACE_HAS_THREADS) && !defined ACE_LACKS_ACCEPT
  28
  29 static const char ACE_ALPHABET[] = "abcdefghijklmnopqrstuvwxyz";
  30
  31 // Number of client (user) threads
  32 static int opt_nconnections = 5;
  33
  34 // Number of data exchanges
  35 static int opt_nloops = 200;
  36
  37 // Use the WFMO_Reactor
  38 static int opt_wfmo_reactor = 0;
  39
  40 // Use the Select_Reactor
  41 static int opt_select_reactor = 0;
  42
  43 // Extra debug messages
  44 static int opt_debug = 0;
  45
  46 int Read_Handler::waiting_ = 0;
  47
  48 void
  49 Read_Handler::set_countdown (int nconnections)
  50 {
  51   Read_Handler::waiting_ = nconnections;
  52 }
  53
  54 // Initialize the Svc_Handler
  55 int
  56 Read_Handler::open (void *)
  57 {
  58   if (this->peer ().enable (ACE_NONBLOCK) == -1)
  59     ACE_ERROR_RETURN ((LM_ERROR,
  60                        ACE_TEXT ("(%t) Read_Handler::open, cannot set non blocking mode\n")),
  61                       -1);
  62
  63   if (reactor ()->register_handler (this, READ_MASK) == -1)
  64     ACE_ERROR_RETURN ((LM_ERROR,
  65                        ACE_TEXT ("(%t) Read_Handler::open, cannot register handler\n")),
  66                       -1);
  67
  68   ACE_DEBUG ((LM_DEBUG,
  69               ACE_TEXT ("(%t) created svc_handler for handle %d\n"),
  70               get_handle ()));
  71   return 0;
  72 }
  73
  74 // Handle incoming data
  75 int
  76 Read_Handler::handle_input (ACE_HANDLE handle)
  77 {
  78   ACE_UNUSED_ARG (handle);
  79   char buf[BUFSIZ];
  80
  81   while (1)
  82     {
  83       ssize_t result = this->peer ().recv (buf, sizeof (buf) - 1);
  84
  85       if (result > 0)
  86         {
  87           if (opt_debug)
  88             {
  89               buf[result] = 0;
  90               ACE_DEBUG ((LM_DEBUG,
  91                           ACE_TEXT ("(%t) Read_Handler::handle_input: %s\n"),
  92                           buf));
  93             }
  94         }
  95       else if (result < 0)
  96         {
  97           if (errno == EWOULDBLOCK)
  98             return 0;
  99           else
 100             {
 101               ACE_ERROR ((LM_ERROR, ACE_TEXT ("handle_input: %p (errno: %d)\n"),
 102                           ACE_TEXT ("recv"), ACE_ERRNO_GET));
 103
 104               // This will cause handle_close to get called.
 105               return -1;
 106             }
 107         }
 108       else // result == 0
 109         {
 110           // This will cause handle_close to get called.
 111           return -1;
 112         }
 113     }
 114
 115   ACE_NOTREACHED (return 0);
 116 }
 117
 118 // Handle connection shutdown.
 119
 120 int
 121 Read_Handler::handle_close (ACE_HANDLE handle,
 122                             ACE_Reactor_Mask close_mask)
 123 {
 124   ACE_UNUSED_ARG (handle);
 125   ACE_UNUSED_ARG (close_mask);
 126
 127   // Reduce count.
 128   waiting_--;
 129
 130   // If no connections are open.
 131   if (waiting_ == 0)
 132     ACE_Reactor::instance ()->end_reactor_event_loop ();
 133
 134   ACE_DEBUG ((LM_DEBUG,
 135               ACE_TEXT ("(%t) Read_Handler::handle_close closing down\n")));
 136
 137   // Shutdown
 138   this->destroy ();
 139   return 0;
 140 }
 141
 142 int
 143 Write_Handler::open (void *)
 144 {
 145   return 0;
 146 }
 147
 148 int
 149 Write_Handler::send_data (void)
 150 {
 151   int send_size = sizeof (ACE_ALPHABET) - 1;
 152
 153   if (this->peer ().send_n (ACE_ALPHABET,
 154                             send_size) != send_size)
 155     ACE_ERROR_RETURN ((LM_ERROR,
 156                        ACE_TEXT ("(%t) %p\n"),
 157                        ACE_TEXT ("send_n")),
 158                       -1);
 159   return 0;
 160 }
 161
 162 // Connection factories
 163 typedef ACE_Connector<Write_Handler, ACE_SOCK_CONNECTOR> CONNECTOR;
 164 typedef ACE_Acceptor<Read_Handler, ACE_SOCK_ACCEPTOR> ACCEPTOR;
 165
 166 // Execute the client tests.
 167 void *
 168 client (void *arg)
 169 {
 170   ACE_DEBUG ((LM_DEBUG,
 171               ACE_TEXT ("(%t) running client\n")));
 172
 173   ACE_INET_Addr *connection_addr =
 174     reinterpret_cast<ACE_INET_Addr *> (arg);
 175   CONNECTOR connector;
 176
 177   int i;
 178
 179   // Automagic memory cleanup.
 180   Write_Handler **temp_writers = 0;
 181   ACE_NEW_RETURN (temp_writers,
 182                   Write_Handler *[opt_nconnections],
 183                   0);
 184   ACE_Auto_Basic_Array_Ptr <Write_Handler *> writers (temp_writers);
 185
 186   ACE_TCHAR *temp_failed = 0;
 187   ACE_NEW_RETURN (temp_failed,
 188                   ACE_TCHAR[opt_nconnections],
 189                   0);
 190   ACE_Auto_Basic_Array_Ptr <ACE_TCHAR> failed_svc_handlers (temp_failed);
 191
 192   // Automagic memory cleanup.
 193   ACE_INET_Addr *temp_addresses;
 194   ACE_NEW_RETURN (temp_addresses,
 195                   ACE_INET_Addr [opt_nconnections],
 196                   0);
 197   ACE_Auto_Array_Ptr <ACE_INET_Addr> addresses (temp_addresses);
 198
 199   // Initialize array.
 200   for (i = 0; i < opt_nconnections; i++)
 201     {
 202       writers[i] = 0;
 203       addresses[i] = *connection_addr;
 204     }
 205
 206   // Connection all <opt_nconnections> svc_handlers
 207   int result = connector.connect_n (opt_nconnections,
 208                                     writers.get (),
 209                                     addresses.get (),
 210                                     failed_svc_handlers.get ());
 211   if (result == -1)
 212     {
 213       // Print out the connections that failed...
 214       for (i = 0; i < opt_nconnections; i++)
 215         if (failed_svc_handlers.get ()[i])
 216           {
 217             ACE_INET_Addr failed_addr = addresses.get()[i];
 218             ACE_ERROR ((LM_ERROR,
 219                         ACE_TEXT ("(%t) connection failed to %s, %d\n"),
 220                         failed_addr.get_host_name (),
 221                         failed_addr.get_port_number ()));
 222           }
 223       return 0;
 224     }
 225
 226   // If no connections failed (result == 0) then there should be valid
 227   // ACE_Svc_handler pointers in each writers[] position.  Iterate to
 228   // send data
 229   for (int j = 0; j < opt_nloops; j++)
 230     for (i = 0; i < opt_nconnections; i++)
 231       if (writers[i]->send_data () == -1)
 232         ACE_ERROR_RETURN ((LM_ERROR,
 233                            ACE_TEXT ("(%t) %p\n"),
 234                            ACE_TEXT ("writer::send_data")),
 235                           0);
 236   // Cleanup
 237   for (i = 0; i < opt_nconnections; i++)
 238     writers[i]->destroy ();
 239
 240   ACE_DEBUG ((LM_DEBUG,
 241               ACE_TEXT ("(%t) finishing client\n")));
 242   return 0;
 243 }
 244
 245 // Sets up the correct reactor (based on platform and options).
 246
 247 void
 248 create_reactor (void)
 249 {
 250   ACE_Reactor_Impl *impl = 0;
 251
 252   if (opt_wfmo_reactor)
 253     {
 254 #if defined (ACE_HAS_WINSOCK2) && (ACE_HAS_WINSOCK2 == 1)
 255       ACE_NEW (impl,
 256                ACE_WFMO_Reactor);
 257 #endif /* ACE_HAS_WINSOCK2 == 1 */
 258     }
 259   else if (opt_select_reactor)
 260     ACE_NEW (impl,
 261              ACE_Select_Reactor);
 262
 263   ACE_Reactor *reactor = 0;
 264   ACE_NEW (reactor,
 265            ACE_Reactor (impl));
 266   ACE_Reactor::instance (reactor);
 267 }
 268
 269 // Print stats.
 270
 271 void
 272 print_results (ACE_Profile_Timer::ACE_Elapsed_Time &et)
 273 {
 274   const ACE_TCHAR *reactor_type = 0;
 275
 276   if (opt_wfmo_reactor)
 277     reactor_type = ACE_TEXT ("WFMO_Reactor");
 278   else if (opt_select_reactor)
 279     reactor_type = ACE_TEXT ("Select_Reactor");
 280   else
 281     reactor_type = ACE_TEXT ("Platform's default Reactor");
 282
 283   ACE_DEBUG ((LM_DEBUG,
 284               ACE_TEXT ("\n\tReactor_Performance Test statistics:\n\n")));
 285   ACE_DEBUG ((LM_DEBUG,
 286               ACE_TEXT ("\tReactor Type: %s\n"),
 287               reactor_type));
 288   ACE_DEBUG ((LM_DEBUG,
 289               ACE_TEXT ("\tConnections: %d\n"),
 290               opt_nconnections));
 291   ACE_DEBUG ((LM_DEBUG,
 292               ACE_TEXT ("\tIteration per connection: %d\n"),
 293               opt_nloops));
 294
 295   ACE_DEBUG ((LM_DEBUG,
 296               ACE_TEXT ("\n\tTiming results:\n")));
 297   ACE_DEBUG ((LM_DEBUG,
 298               ACE_TEXT ("\t\treal time = %f secs \n\t\tuser time = %f secs \n\t\tsystem time = %f secs\n\n"),
 299               et.real_time,
 300               et.user_time,
 301               et.system_time));
 302 }
 303
 304 int
 305 run_main (int argc, ACE_TCHAR *argv[])
 306 {
 307   ACE_START_TEST (ACE_TEXT ("Reactor_Performance_Test"));
 308
 309   //FUZZ: disable check_for_lack_ACE_OS
 310   ACE_Get_Opt getopt (argc, argv, ACE_TEXT ("dswc:l:"), 1);
 311   for (int c; (c = getopt ()) != -1; )
 312   //FUZZ: enble check_for_lack_ACE_OS
 313     switch (c)
 314       {
 315       case 's':
 316         opt_select_reactor = 1;
 317         break;
 318       case 'w':
 319         opt_wfmo_reactor = 1;
 320         break;
 321       case 'c':
 322         opt_nconnections = ACE_OS::atoi (getopt.opt_arg ());
 323         break;
 324       case 'l':
 325         opt_nloops = ACE_OS::atoi (getopt.opt_arg ());
 326         break;
 327       case 'd':
 328         opt_debug = 1;
 329         break;
 330       }
 331
 332   // Sets up the correct reactor (based on platform and options).
 333   create_reactor ();
 334
 335   // Manage memory automagically.
 336   auto_ptr<ACE_Reactor> reactor (ACE_Reactor::instance ());
 337   auto_ptr<ACE_Reactor_Impl> impl;
 338
 339   // If we are using other that the default implementation, we must
 340   // clean up.
 341   if (opt_select_reactor || opt_wfmo_reactor)
 342     {
 343       auto_ptr<ACE_Reactor_Impl> auto_impl (ACE_Reactor::instance ()->implementation ());
 344       impl = auto_impl;
 345     }
 346
 347   Read_Handler::set_countdown (opt_nconnections);
 348
 349   // Acceptor
 350   ACCEPTOR acceptor;
 351   ACE_INET_Addr server_addr;
 352
 353   // Bind acceptor to any port and then find out what the port was.
 354   ACE_INET_Addr local_addr (ACE_sap_any_cast (const ACE_INET_Addr &));
 355   if (acceptor.open (local_addr) == -1
 356       || acceptor.acceptor ().get_local_addr (server_addr) == -1)
 357     ACE_ERROR_RETURN ((LM_ERROR,
 358                        ACE_TEXT ("(%t) %p\n"),
 359                        ACE_TEXT ("open")),
 360                       -1);
 361
 362   ACE_DEBUG ((LM_DEBUG,
 363               ACE_TEXT ("(%t) starting server at port %d\n"),
 364               server_addr.get_port_number ()));
 365
 366   ACE_INET_Addr connection_addr (server_addr.get_port_number (),
 367                                  ACE_DEFAULT_SERVER_HOST);
 368
 369   if (ACE_Thread_Manager::instance ()->spawn
 370       (ACE_THR_FUNC (client),
 371        (void *) &connection_addr,
 372        THR_NEW_LWP | THR_DETACHED) == -1)
 373     ACE_ERROR ((LM_ERROR,
 374                 ACE_TEXT ("(%t) %p\n"),
 375                 ACE_TEXT ("thread create failed")));
 376
 377   ACE_Time_Value run_limit (opt_nloops / 10);
 378
 379   ACE_Profile_Timer timer;
 380   timer.start ();
 381   const int status =
 382     ACE_Reactor::instance ()->run_reactor_event_loop (run_limit);
 383   timer.stop ();
 384
 385   ACE_Profile_Timer::ACE_Elapsed_Time et;
 386   timer.elapsed_time (et);
 387
 388   // Print results
 389   print_results (et);
 390
 391   ACE_DEBUG ((LM_DEBUG,
 392               ACE_TEXT ("(%t) waiting for the client thread...\n")));
 393
 394   ACE_Thread_Manager::instance ()->wait ();
 395
 396   ACE_END_TEST;
 397   return status;
 398 }
 399
 400 #else
 401 int
 402 run_main (int, ACE_TCHAR *[])
 403 {
 404   ACE_START_TEST (ACE_TEXT ("Reactor_Performance_Test"));
 405
 406   ACE_ERROR ((LM_INFO,
 407               ACE_TEXT ("threads/accept not supported on this platform\n")));
 408
 409   ACE_END_TEST;
 410   return 0;
 411 }
 412 #endif /* ACE_HAS_THREADS && ! ACE_LACKS_ACCEPT */