ACE/tests/SSL/Thread_Pool_Reactor_SSL_Test.cpp

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