ACE/tests/Thread_Pool_Reactor_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Thread_Pool_Reactor_Test.cpp
   5  *
   6  *    This program is a torture test of thread pool reactors.  It
   7  *    starts by spawning several server threads waiting to handle
   8  *    events.  Several other client threads are spawned right after
   9  *    to initiate connections to server threads.  Each connection
  10  *    adds a new Svc_Handler into the TP_Reactor and sends out
  11  *    several "requests" to the server thread.  After the connection
  12  *    is closed, the Svc_Handler is removed from the TP_Reactor.
  13  *    Each message is treated as a separate request by the server so
  14  *    two consecutive requests might be serviced by two different
  15  *    threads.
  16  *
  17  *    Usage: Thread_Pool_Reactor_Test [-r <hostname:port#>]
  18  *              [-s <server thr#>] [-c <client thr#>] [-d <delay>]
  19  *              [-i <client conn attempt#>] [-n <client request# per conn>]
  20  *
  21  *    Default value:
  22  *        <hostname:port#>:       ACE_DEFAULT_RENDEZVOUS
  23  *        <server thr#>:          ACE_MAX_THREADS
  24  *        <client thr#>:          ACE_MAX_ITERATIONS
  25  *        <client conn attempt#>: ACE_MAX_ITERATIONS
  26  *        <client req# per conn>: ACE_MAX_THREADS
  27  *        <delay>:                50 usec
  28  *
  29  *  @author   Irfan Pyarali <irfan@cs.wustl.edu> and   Nanbor Wang <nanbor@cs.wustl.edu>
  30  */
  31 //=============================================================================
  32
  33
  34 #include "test_config.h"
  35 #include "ace/OS_NS_string.h"
  36 #include "ace/OS_NS_unistd.h"
  37 #include "ace/Get_Opt.h"
  38 #include "ace/SOCK_Connector.h"
  39 #include "ace/SOCK_Acceptor.h"
  40 #include "ace/Acceptor.h"
  41 #include "ace/Thread_Manager.h"
  42 #include "ace/TP_Reactor.h"
  43
  44
  45 #if defined (ACE_HAS_THREADS) && !defined ACE_LACKS_ACCEPT
  46
  47 #include "Thread_Pool_Reactor_Test.h"
  48 using ACCEPTOR = ACE_Strategy_Acceptor<Request_Handler, ACE_SOCK_Acceptor>;
  49
  50 // Accepting end point.  This is actually "localhost:10010", but some
  51 // platform couldn't resolve the name so we use the IP address
  52 // directly here.
  53 static const ACE_TCHAR *rendezvous = ACE_TEXT ("127.0.0.1:10010");
  54
  55 // Total number of server threads.
  56 static size_t svr_thrno = ACE_MAX_THREADS;
  57
  58 // Default network parameters (MAX_BINDS and system buffers) are too small
  59 // for full test on some platforms; add platforms that can't handle too many
  60 // connection simultaneously here.
  61 #if defined (ACE_VXWORKS)
  62 #define ACE_LOAD_FACTOR /2
  63 #else
  64 #define ACE_LOAD_FACTOR
  65 #endif
  66
  67 // Total number of client threads.
  68 static size_t cli_thrno = ACE_MAX_THREADS ACE_LOAD_FACTOR;
  69
  70 // Total connection attempts of a client thread.
  71 static size_t cli_conn_no = ACE_MAX_ITERATIONS ACE_LOAD_FACTOR;
  72
  73 // Total requests a client thread sends.
  74 static size_t cli_req_no = ACE_MAX_THREADS ACE_LOAD_FACTOR;
  75
  76 // Delay before a thread sending the next request (in msec.)
  77 static int req_delay = 50;
  78
  79 static void
  80 parse_arg (int argc, ACE_TCHAR *argv[])
  81 {
  82   //FUZZ: disable check_for_lack_ACE_OS
  83   ACE_Get_Opt getopt (argc, argv, ACE_TEXT ("r:s:c:d:i:n:"));
  84
  85   int c;
  86
  87   while ((c = getopt ()) != -1)
  88     {
  89   //FUZZ: enable check_for_lack_ACE_OS
  90       switch (c)
  91         {
  92         case 'r':                   // hostname:port
  93           rendezvous = getopt.opt_arg ();
  94           break;
  95         case 's':
  96           svr_thrno = ACE_OS::atoi (getopt.opt_arg ());
  97           break;
  98         case 'c':
  99           cli_thrno = ACE_OS::atoi (getopt.opt_arg ());
 100           break;
 101         case 'd':
 102           req_delay = ACE_OS::atoi (getopt.opt_arg ());
 103           break;
 104         case 'i':
 105           cli_conn_no = ACE_OS::atoi (getopt.opt_arg ());
 106           break;
 107         case 'n':
 108           cli_req_no = ACE_OS::atoi (getopt.opt_arg ());
 109           break;
 110         default:
 111           ACE_ERROR ((LM_ERROR,
 112                       "Usage: Thread_Pool_Reactor_Test [-r <hostname:port#>]"
 113                       "\t[-s <server thr#>] [-c <client thr#>] [-d <delay>]"
 114                       "\t[-i <client conn attempt#>]"
 115                       "[-n <client request# per conn>]\n"));
 116           break;
 117         }
 118     }
 119 }
 120
 121 Request_Handler::Request_Handler (ACE_Thread_Manager *thr_mgr)
 122   : ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_MT_SYNCH> (thr_mgr),
 123     nr_msgs_rcvd_(0)
 124 {
 125   // Make sure we use TP_Reactor with this class (that's the whole
 126   // point, right?)
 127   this->reactor (ACE_Reactor::instance ());
 128 }
 129
 130 int
 131 Request_Handler::handle_input (ACE_HANDLE fd)
 132 {
 133   ACE_TCHAR buffer[BUFSIZ];
 134   ACE_TCHAR len = 0;
 135   ssize_t result = this->peer ().recv (&len, sizeof (ACE_TCHAR));
 136
 137   if (result > 0
 138       && this->peer ().recv_n (buffer, len * sizeof (ACE_TCHAR))
 139           == static_cast<ssize_t> (len * sizeof (ACE_TCHAR)))
 140     {
 141       ++this->nr_msgs_rcvd_;
 142
 143       ACE_DEBUG ((LM_DEBUG,
 144                   "(%t) svr input; fd: 0x%x; input: %s\n",
 145                   fd,
 146                   buffer));
 147       if (ACE_OS::strcmp (buffer, ACE_TEXT ("shutdown")) == 0)
 148           ACE_Reactor::instance()->end_reactor_event_loop ();
 149       return 0;
 150     }
 151   else
 152     ACE_DEBUG ((LM_DEBUG,
 153                 "(%t) Request_Handler: 0x%x peer closed (0x%x)\n",
 154                 this, fd));
 155   return -1;
 156 }
 157
 158 int
 159 Request_Handler::handle_close (ACE_HANDLE fd, ACE_Reactor_Mask)
 160 {
 161   ACE_DEBUG ((LM_DEBUG,
 162               "(%t) svr close; fd: 0x%x, rcvd %d msgs\n",
 163               fd,
 164               this->nr_msgs_rcvd_));
 165   if (this->nr_msgs_rcvd_ != cli_req_no)
 166     ACE_ERROR((LM_ERROR,
 167                "(%t) Handler 0x%x: Expected %d messages; got %d\n",
 168                this,
 169                cli_req_no,
 170                this->nr_msgs_rcvd_));
 171   this->destroy ();
 172   return 0;
 173 }
 174
 175 static int
 176 reactor_event_hook (ACE_Reactor *)
 177 {
 178   ACE_DEBUG ((LM_DEBUG,
 179               "(%t) handling events ....\n"));
 180
 181   return 0;
 182 }
 183
 184 static ACE_THR_FUNC_RETURN
 185 svr_worker (void *)
 186 {
 187   // Server thread function.
 188   int result =
 189     ACE_Reactor::instance ()->run_reactor_event_loop (&reactor_event_hook);
 190
 191   if (result == -1)
 192     ACE_ERROR_RETURN ((LM_ERROR,
 193                        "(%t) %p\n",
 194                        "Error handling events"),
 195                       0);
 196
 197   ACE_DEBUG ((LM_DEBUG,
 198               "(%t) I am done handling events. Bye, bye\n"));
 199
 200   return 0;
 201 }
 202
 203 static ACE_THR_FUNC_RETURN
 204 cli_worker (void *arg)
 205 {
 206   // Client thread function.
 207   ACE_INET_Addr addr (rendezvous);
 208   ACE_SOCK_Stream stream;
 209   ACE_SOCK_Connector connect;
 210   ACE_Time_Value delay (0, req_delay);
 211   size_t len = * reinterpret_cast<ACE_TCHAR *> (arg);
 212
 213   for (size_t i = 0 ; i < cli_conn_no; i++)
 214     {
 215       if (connect.connect (stream, addr) < 0)
 216         {
 217           ACE_ERROR ((LM_ERROR,
 218                       "(%t) %p\n",
 219                       "connect"));
 220           continue;
 221         }
 222
 223       for (size_t j = 0; j < cli_req_no; j++)
 224         {
 225           ACE_DEBUG ((LM_DEBUG,
 226                       "(%t) conn_worker handle 0x%x, req %d\n",
 227                       stream.get_handle (),
 228                       j+1));
 229           if (stream.send_n (arg,
 230                              (len + 1) * sizeof (ACE_TCHAR)) == -1)
 231             {
 232               ACE_ERROR ((LM_ERROR,
 233                           "(%t) %p\n",
 234                           "send_n"));
 235               continue;
 236             }
 237           ACE_OS::sleep (delay);
 238         }
 239
 240       stream.close ();
 241     }
 242
 243   return 0;
 244 }
 245
 246 static ACE_THR_FUNC_RETURN
 247 worker (void *)
 248 {
 249   ACE_OS::sleep (3);
 250   const ACE_TCHAR *msg = ACE_TEXT ("Message from Connection worker");
 251   ACE_TCHAR buf [BUFSIZ];
 252   buf[0] = static_cast<ACE_TCHAR> ((ACE_OS::strlen (msg) + 1));
 253   ACE_OS::strcpy (&buf[1], msg);
 254
 255   ACE_INET_Addr addr (rendezvous);
 256
 257   ACE_DEBUG((LM_DEBUG,
 258              "(%t) Spawning %d client threads...\n",
 259              cli_thrno));
 260   int grp = ACE_Thread_Manager::instance ()->spawn_n (cli_thrno,
 261                                                       &cli_worker,
 262                                                       buf);
 263   ACE_TEST_ASSERT (grp != -1);
 264
 265   ACE_Thread_Manager::instance ()->wait_grp (grp);
 266
 267   ACE_DEBUG ((LM_DEBUG,
 268               "(%t) Client threads done; shutting down...\n"));
 269   ACE_SOCK_Stream stream;
 270   ACE_SOCK_Connector connect;
 271
 272   if (connect.connect (stream, addr) == -1)
 273     ACE_ERROR ((LM_ERROR,
 274                 "(%t) %p Error while connecting\n",
 275                 "connect"));
 276
 277   const ACE_TCHAR *sbuf = ACE_TEXT ("\011shutdown");
 278
 279   ACE_DEBUG ((LM_DEBUG,
 280               "shutdown stream handle = %x\n",
 281               stream.get_handle ()));
 282
 283   if (stream.send_n (sbuf, (ACE_OS::strlen (sbuf) + 1) * sizeof (ACE_TCHAR)) == -1)
 284     ACE_ERROR ((LM_ERROR,
 285                 "(%t) %p\n",
 286                 "send_n"));
 287
 288   stream.close ();
 289
 290   return 0;
 291 }
 292
 293 int
 294 run_main (int argc, ACE_TCHAR *argv[])
 295 {
 296   ACE_START_TEST (ACE_TEXT ("Thread_Pool_Reactor_Test"));
 297   parse_arg (argc, argv);
 298
 299   // Changed the default
 300   ACE_TP_Reactor sr;
 301   ACE_Reactor new_reactor (&sr);
 302   ACE_Reactor::instance (&new_reactor);
 303
 304   ACCEPTOR acceptor;
 305   ACE_INET_Addr accept_addr (rendezvous);
 306
 307   if (acceptor.open (accept_addr) == -1)
 308     ACE_ERROR_RETURN ((LM_ERROR,
 309                        ACE_TEXT ("%p\n"),
 310                        ACE_TEXT ("open")),
 311                       1);
 312
 313   ACE_DEBUG((LM_DEBUG,
 314              ACE_TEXT ("(%t) Spawning %d server threads...\n"),
 315              svr_thrno));
 316   ACE_Thread_Manager::instance ()->spawn_n (svr_thrno,
 317                                             svr_worker);
 318   ACE_Thread_Manager::instance ()->spawn (worker);
 319
 320   ACE_Thread_Manager::instance ()->wait ();
 321
 322   ACE_END_TEST;
 323   return 0;
 324 }
 325
 326 #else
 327 int
 328 run_main (int, ACE_TCHAR *[])
 329 {
 330   ACE_START_TEST (ACE_TEXT ("Thread_Pool_Reactor_Test"));
 331
 332   ACE_ERROR ((LM_INFO,
 333               "threads/accept not supported on this platform\n"));
 334
 335   ACE_END_TEST;
 336   return 0;
 337 }
 338 #endif /* ACE_HAS_THREADS */