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