ACE/tests/SSL/SSL_Asynch_Stream_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    SSL_Asynch_Stream_Test.cpp
   5  *
   6  *    This program is a functionality test of ACE_SSL_Asynch_Stream.
   7  *    It demonstrates one proper use case of ACE_SSL_Asynch_Stream in the
   8  *    Proactor framework and validates its basic functionality.
   9  *
  10  *    Usage: SSL_Asynch_Stream_Test [-r <hostname:port#>]
  11  *              [-t <num threads>] [-d <delay>]
  12  *              [-i <client conn attempt#>] [-n <client request# per conn>]
  13  *
  14  *    Default value:
  15  *        <hostname:port#>:       ACE_DEFAULT_SERVER_HOST:ACE_DEFAULT_PORT
  16  *        <num threads>:          ACE_MAX_THREADS
  17  *        <client conn attempt#>: ACE_MAX_ITERATIONS
  18  *        <client req# per conn>: 20
  19  *        <delay>:                0 usec
  20  *
  21  *  @author   Steve Huston <shuston@riverace.com>
  22  */
  23 //=============================================================================
  24
  25
  26 #include "../test_config.h"
  27 #include "ace/Default_Constants.h"
  28 #include "ace/OS_NS_signal.h"
  29 #include "ace/OS_NS_string.h"
  30 #include "ace/Event_Handler.h"
  31 #include "ace/Get_Opt.h"
  32 #include "ace/Proactor.h"
  33 #include "ace/Reactor.h"
  34 #include "ace/Thread_Manager.h"
  35 #include "ace/INET_Addr.h"
  36 #include "ace/SSL/SSL_Asynch_Stream.h"
  37 #include "ace/SSL/SSL_SOCK_Connector.h"
  38 #include "ace/SSL/SSL_SOCK_Acceptor.h"
  39 #include "ace/SSL/SSL_SOCK_Stream.h"
  40
  41 #if defined (ACE_HAS_THREADS) && (defined (ACE_WIN32) || (defined (ACE_HAS_AIO_CALLS)))
  42   // This only works on Win32 platforms and on Unix platforms
  43   // supporting POSIX aio calls.
  44
  45 class Client_Handler : public ACE_Handler
  46 {
  47 public:
  48   Client_Handler ()
  49     : msgs_sent_ (0),
  50       stream_ (ACE_SSL_Asynch_Stream::ST_CLIENT),
  51       block_ (1024) {}
  52   ~Client_Handler ();
  53
  54   //FUZZ: disable check_for_lack_ACE_OS
  55   int open (ACE_HANDLE);
  56   //FUZZ: enable check_for_lack_ACE_OS
  57
  58 private:
  59   virtual void handle_write_stream (const ACE_Asynch_Write_Stream::Result &result);
  60
  61 private:
  62   size_t                  msgs_sent_;
  63   ACE_SSL_Asynch_Stream   stream_;
  64   ACE_Message_Block       block_;
  65 };
  66
  67 class Server_Handler : public ACE_Handler
  68 {
  69 public:
  70   Server_Handler ()
  71     : msgs_rcvd_ (0),
  72       stream_ (ACE_SSL_Asynch_Stream::ST_SERVER),
  73       block_ (1024) {}
  74   ~Server_Handler ();
  75
  76   //FUZZ: disable check_for_lack_ACE_OS
  77   int open (ACE_HANDLE);
  78   //FUZZ: enable check_for_lack_ACE_OS
  79
  80 private:
  81   virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result);
  82
  83 private:
  84   size_t                  msgs_rcvd_;
  85   ACE_SSL_Asynch_Stream   stream_;
  86   ACE_Message_Block       block_;
  87 };
  88
  89 class Server_Acceptor : public ACE_Event_Handler
  90 {
  91 public:
  92   //FUZZ: disable check_for_lack_ACE_OS
  93   int open (const ACE_INET_Addr &listen_addr);
  94   //FUZZ: enable check_for_lack_ACE_OS
  95
  96   // Get the I/O handle.
  97   virtual ACE_HANDLE get_handle () const;
  98
  99   // Called when a new connection is ready to accept.
 100   virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE);
 101
 102   virtual int handle_close (ACE_HANDLE handle,
 103                             ACE_Reactor_Mask close_mask);
 104
 105 private:
 106   ACE_SSL_SOCK_Acceptor acceptor_;
 107 };
 108
 109 // Accepting end point.  This is actually "localhost:10010", but some
 110 // platform couldn't resolve the name so we use the IP address
 111 // directly here.
 112 static const ACE_TCHAR *rendezvous = \
 113   ACE_DEFAULT_SERVER_HOST ACE_TEXT (":") ACE_DEFAULT_SERVER_PORT_STR;
 114
 115 // Total number of proactor threads.
 116 static size_t num_threads = ACE_MAX_THREADS;
 117
 118 // Number of client connections to attempt.
 119 static size_t cli_conn_no = ACE_MAX_ITERATIONS;
 120
 121 // Number of requests each client connection sends.
 122 static size_t cli_req_no = ACE_MAX_THREADS;
 123
 124 // Delay before a thread sending the next request (in msec.)
 125 static int req_delay = 0;
 126
 127 // This is the string sent from client to server.
 128 static const char *test_string = "SSL_Asynch_Stream_Test!";
 129
 130 // Function to remove signals from the signal mask.
 131 static int
 132 disable_signal (int sigmin, int sigmax)
 133 {
 134 #if !defined (ACE_LACKS_UNIX_SIGNALS)
 135   sigset_t signal_set;
 136   if (ACE_OS::sigemptyset (&signal_set) == - 1)
 137     ACE_ERROR ((LM_ERROR,
 138                 ACE_TEXT ("Error: (%P|%t):%p\n"),
 139                 ACE_TEXT ("sigemptyset failed")));
 140
 141   for (int i = sigmin; i <= sigmax; i++)
 142     ACE_OS::sigaddset (&signal_set, i);
 143
 144   // Put the <signal_set>.
 145 # if defined (ACE_LACKS_PTHREAD_THR_SIGSETMASK)
 146   // In multi-threaded application this is not POSIX compliant
 147   // but let's leave it just in case.
 148   if (ACE_OS::sigprocmask (SIG_BLOCK, &signal_set, 0) != 0)
 149 # else
 150   if (ACE_OS::thr_sigsetmask (SIG_BLOCK, &signal_set, 0) != 0)
 151 # endif /* ACE_LACKS_PTHREAD_THR_SIGSETMASK */
 152     ACE_ERROR_RETURN ((LM_ERROR,
 153                        ACE_TEXT ("Error: (%P|%t): %p\n"),
 154                        ACE_TEXT ("SIG_BLOCK failed")),
 155                       -1);
 156 #else
 157   ACE_UNUSED_ARG (sigmin);
 158   ACE_UNUSED_ARG (sigmax);
 159 #endif /* ACE_LACKS_UNIX_SIGNALS */
 160
 161   return 0;
 162 }
 163
 164 static void
 165 parse_args (int argc, ACE_TCHAR *argv[])
 166 {
 167   //FUZZ: disable check_for_lack_ACE_OS
 168   ACE_Get_Opt getopt (argc, argv, ACE_TEXT ("r:t:d:i:n:"));
 169
 170   int c;
 171
 172   while ((c = getopt ()) != -1)
 173     {
 174   //FUZZ: enable check_for_lack_ACE_OS
 175       switch (c)
 176         {
 177         case 'r':                   // hostname:port
 178           rendezvous = getopt.opt_arg ();
 179           break;
 180         case 't':
 181           num_threads = ACE_OS::atoi (getopt.opt_arg ());
 182           break;
 183         case 'd':
 184           req_delay = ACE_OS::atoi (getopt.opt_arg ());
 185           break;
 186         case 'i':
 187           cli_conn_no = ACE_OS::atoi (getopt.opt_arg ());
 188           break;
 189         case 'n':
 190           cli_req_no = ACE_OS::atoi (getopt.opt_arg ());
 191           break;
 192         default:
 193           ACE_ERROR ((LM_ERROR,
 194                       ACE_TEXT ("Usage: %s [-r <hostname:port#>]")
 195                       ACE_TEXT ("\t[-t <nr threads>] [-d <delay>]")
 196                       ACE_TEXT ("\t[-i <client conn attempt#>]")
 197                       ACE_TEXT ("\t[-n <client request# per conn>]\n"),
 198                       argv[0]));
 199           break;
 200         }
 201     }
 202 }
 203
 204 Client_Handler::~Client_Handler ()
 205 {
 206   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Client %@ handle %d down\n"),
 207               this, this->stream_.handle ()));
 208   if (this->stream_.handle () != ACE_INVALID_HANDLE)
 209     {
 210       if (this->msgs_sent_ != cli_req_no)
 211         ACE_ERROR ((LM_ERROR,
 212                     ACE_TEXT ("(%t) Client handle %d sent %d messages; ")
 213                     ACE_TEXT ("expected %d\n"),
 214                     this->stream_.handle (),
 215                     this->msgs_sent_,
 216                     cli_req_no));
 217       else
 218         ACE_DEBUG ((LM_DEBUG,
 219                     ACE_TEXT ("(%t) Client handle %d sent %d messages; ")
 220                     ACE_TEXT ("closing connection\n"),
 221                     this->stream_.handle (),
 222                     cli_req_no));
 223     }
 224   this->stream_.close ();
 225 }
 226
 227 int
 228 Client_Handler::open (ACE_HANDLE handle)
 229 {
 230   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Client %@ handle %d up\n"),
 231               this, handle));
 232   if (this->stream_.open (*this, handle) == -1)
 233     ACE_ERROR_RETURN ((LM_ERROR,
 234                        ACE_TEXT ("(%t) Client_Handler: %p\n"),
 235                        ACE_TEXT ("open")),
 236                       -1);
 237   this->block_.copy (test_string);
 238   if (this->stream_.write (this->block_, this->block_.length ()) == -1)
 239     ACE_ERROR_RETURN ((LM_ERROR,
 240                        ACE_TEXT ("(%t) Client_Handler: %p\n"),
 241                        ACE_TEXT ("initiate write")),
 242                       -1);
 243   return 0;
 244 }
 245
 246 void
 247 Client_Handler::handle_write_stream
 248   (const ACE_Asynch_Write_Stream::Result &result)
 249 {
 250   if (!result.success ())
 251     {
 252       errno = result.error ();
 253       ACE_ERROR ((LM_ERROR,
 254                   ACE_TEXT ("(%t) Client handle %d: %p\n"),
 255                   this->stream_.handle (),
 256                   ACE_TEXT ("write")));
 257       delete this;
 258       return;
 259     }
 260   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Client %@ handle %d sent %B of %B bytes\n"),
 261               this, this->stream_.handle (),
 262               result.bytes_transferred (), result.bytes_to_write ()));
 263   ACE_Message_Block &b = result.message_block ();
 264   bool send_again = true;
 265   if (b.length () == 0)
 266     {
 267       // All block's data sent; rewind the read pointer and send it again
 268       // until we've sent the configured number of times.
 269       ++this->msgs_sent_;
 270       if (this->msgs_sent_ == cli_req_no)
 271         send_again = false;   // All done
 272       else
 273         b.rd_ptr (b.base ());
 274     }
 275
 276   if (send_again)
 277     {
 278       if (this->stream_.write (this->block_, this->block_.length ()) == -1)
 279         {
 280           ACE_ERROR ((LM_ERROR,
 281                       ACE_TEXT ("(%t) Client_Handler: %p\n"),
 282                       ACE_TEXT ("initiate write")));
 283           delete this;
 284         }
 285     }
 286   else
 287     {
 288       ACE_DEBUG ((LM_DEBUG,
 289                   ACE_TEXT ("(%t) Client handle %d done sending\n"),
 290                   this->stream_.handle ()));
 291       delete this;
 292     }
 293   return;
 294 }
 295
 296 Server_Handler::~Server_Handler ()
 297 {
 298   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Server %@ handle %d down\n"),
 299               this, this->stream_.handle ()));
 300   if (this->stream_.handle () != ACE_INVALID_HANDLE)
 301     {
 302       if (this->msgs_rcvd_ != cli_req_no)
 303         ACE_ERROR ((LM_ERROR,
 304                     ACE_TEXT ("(%t) Server handle %d received %d messages; ")
 305                     ACE_TEXT ("expected %d\n"),
 306                     this->stream_.handle (),
 307                     this->msgs_rcvd_,
 308                     cli_req_no));
 309       else
 310         ACE_DEBUG ((LM_DEBUG,
 311                     ACE_TEXT ("(%t) Server handle %d received %d messages; ")
 312                     ACE_TEXT ("closing connection\n"),
 313                     this->stream_.handle (),
 314                     cli_req_no));
 315     }
 316   this->stream_.close ();
 317 }
 318
 319 int
 320 Server_Handler::open (ACE_HANDLE handle)
 321 {
 322   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Server %@ handle %d up\n"),
 323               this, handle));
 324   if (this->stream_.open (*this, handle) == -1)
 325     ACE_ERROR_RETURN ((LM_ERROR,
 326                        ACE_TEXT ("(%t) Server_Handler: %p\n"),
 327                        ACE_TEXT ("open")),
 328                       -1);
 329   if (this->stream_.read (this->block_, this->block_.space () - 1) == -1)
 330     ACE_ERROR_RETURN ((LM_ERROR,
 331                        ACE_TEXT ("(%t) Server_Handler: %p\n"),
 332                        ACE_TEXT ("read")),
 333                       -1);
 334   return 0;
 335 }
 336
 337 void
 338 Server_Handler::handle_read_stream
 339   (const ACE_Asynch_Read_Stream::Result &result)
 340 {
 341   if (!result.success ())
 342     {
 343       errno = result.error ();
 344       ACE_ERROR ((LM_ERROR,
 345                   ACE_TEXT ("(%t) Server handle %d: %p\n"),
 346                   this->stream_.handle (),
 347                   ACE_TEXT ("read")));
 348       delete this;
 349       return;
 350     }
 351   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Server %@ handle %d recv %B of %B bytes\n"),
 352               this, this->stream_.handle (),
 353               result.bytes_transferred (), result.bytes_to_read ()));
 354   if (result.bytes_transferred () == 0)
 355     {
 356       ACE_DEBUG ((LM_DEBUG,
 357                   ACE_TEXT ("(%t) Server handle %d closed by peer\n"),
 358                   this->stream_.handle ()));
 359       delete this;
 360       return;
 361     }
 362
 363   // Scan through the received data for the expected string. There may be
 364   // multiples and/or partials. Count up how many arrive before the connection
 365   // is closed.
 366   // Remember that the client side sends the terminating nul; in case the
 367   // whole thing didn't arrive, we add a nul to the end of the receive
 368   // block so we don't run off the end. When the recv into this buffer was
 369   // initiated, we left the last byte empty to facilitate this.
 370   ACE_Message_Block &b = result.message_block ();
 371   *(b.wr_ptr ()) = '\0';
 372   size_t test_string_len = ACE_OS::strlen (test_string);
 373   while (b.length () >= test_string_len)
 374     {
 375       if (0 != ACE_OS::strncmp (b.rd_ptr (), test_string, test_string_len))
 376         ACE_ERROR_BREAK ((LM_ERROR,
 377                           ACE_TEXT ("(%t) Read string: %C; expected: %C\n"),
 378                           b.rd_ptr (),
 379                           test_string));
 380       b.rd_ptr (test_string_len);
 381       // That ran up over the string; can we also consume the nul?
 382       if (b.length () > 0)
 383         b.rd_ptr (1);
 384       ++this->msgs_rcvd_;
 385     }
 386   b.crunch ();
 387   if (this->stream_.read (b, b.space () - 1) == -1)
 388     {
 389       ACE_ERROR ((LM_ERROR,
 390                   ACE_TEXT ("(%t) Server_Handler: %p\n"),
 391                   ACE_TEXT ("read")));
 392       delete this;
 393     }
 394   return;
 395 }
 396
 397
 398 int
 399 Server_Acceptor::open (const ACE_INET_Addr &listen_addr)
 400 {
 401   if (this->acceptor_.open (listen_addr) == -1)
 402     ACE_ERROR_RETURN ((LM_ERROR,
 403                        ACE_TEXT ("%p\n"),
 404                        ACE_TEXT ("listen")),
 405                       -1);
 406   return 0;
 407 }
 408
 409 ACE_HANDLE
 410 Server_Acceptor::get_handle () const
 411 {
 412   return this->acceptor_.get_handle ();
 413 }
 414
 415 int
 416 Server_Acceptor::handle_input (ACE_HANDLE)
 417 {
 418   ACE_SSL_SOCK_Stream new_stream;
 419   if (this->acceptor_.accept (new_stream) == -1)
 420     ACE_ERROR_RETURN ((LM_ERROR,
 421                        ACE_TEXT ("(%t) %p\n"),
 422                        ACE_TEXT ("accept")),
 423                       -1);
 424   Server_Handler *new_handler = 0;
 425   ACE_NEW_RETURN (new_handler, Server_Handler, -1);
 426   if (new_handler->open (new_stream.get_handle ()) != 0)
 427     delete new_handler;
 428
 429   return 0;
 430 }
 431
 432 int
 433 Server_Acceptor::handle_close (ACE_HANDLE, ACE_Reactor_Mask)
 434 {
 435   this->acceptor_.close ();
 436   return 0;
 437 }
 438
 439
 440 static ACE_THR_FUNC_RETURN
 441 proactor_loop (void *)
 442 {
 443   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Start handling events.\n")));
 444
 445   disable_signal (ACE_SIGRTMIN, ACE_SIGRTMAX);
 446   disable_signal (SIGPIPE, SIGPIPE);
 447
 448   int result =
 449     ACE_Proactor::instance ()->proactor_run_event_loop ();
 450   if (result == -1)
 451     ACE_ERROR_RETURN ((LM_ERROR,
 452                        ACE_TEXT ("(%t) %p\n"),
 453                        ACE_TEXT ("Error handling events")),
 454                       0);
 455
 456   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Done handling events.\n")));
 457
 458   return 0;
 459 }
 460
 461 static ACE_THR_FUNC_RETURN
 462 start_clients (void *)
 463 {
 464   // Client thread function.
 465   ACE_INET_Addr addr (rendezvous);
 466   ACE_SSL_SOCK_Connector connect;
 467
 468   disable_signal (ACE_SIGRTMIN, ACE_SIGRTMAX);
 469   disable_signal (SIGPIPE, SIGPIPE);
 470
 471   for (size_t i = 0 ; i < cli_conn_no; i++)
 472     {
 473       ACE_SSL_SOCK_Stream stream;
 474       if (connect.connect (stream, addr) < 0)
 475         {
 476           ACE_ERROR ((LM_ERROR,
 477                       ACE_TEXT ("(%t) %p\n"),
 478                       ACE_TEXT ("connect")));
 479           continue;
 480         }
 481
 482       Client_Handler *new_handler = 0;
 483       ACE_NEW_RETURN (new_handler, Client_Handler, (ACE_THR_FUNC_RETURN)-1);
 484       if (new_handler->open (stream.get_handle ()) != 0)
 485         {
 486           delete new_handler;
 487           stream.close ();
 488         }
 489     }
 490
 491   return 0;
 492 }
 493
 494 int
 495 run_main (int argc, ACE_TCHAR *argv[])
 496 {
 497   ACE_START_TEST (ACE_TEXT ("SSL_Asynch_Stream_Test"));
 498
 499   ACE_SSL_Context *context = ACE_SSL_Context::instance ();
 500   // Note - the next two strings are naked on purpose... the arguments to
 501   // the ACE_SSL_Context methods are const char *, not ACE_TCHAR *.
 502   context->certificate ("dummy.pem", SSL_FILETYPE_PEM);
 503   context->private_key ("key.pem", SSL_FILETYPE_PEM);
 504
 505   parse_args (argc, argv);
 506   disable_signal (ACE_SIGRTMIN, ACE_SIGRTMAX);
 507   disable_signal (SIGPIPE, SIGPIPE);
 508
 509   Server_Acceptor acceptor;
 510   ACE_INET_Addr accept_addr (rendezvous);
 511
 512   if (acceptor.open (accept_addr) == -1)
 513     return 1;
 514   if (-1 == ACE_Reactor::instance ()->register_handler (&acceptor,
 515                                                         ACE_Event_Handler::ACCEPT_MASK))
 516     {
 517       ACE_ERROR ((LM_ERROR,
 518                   ACE_TEXT ("(%t) %p; aborting\n"),
 519                   ACE_TEXT ("register_handler")));
 520       return 1;
 521     }
 522
 523   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Listening at %s\n"), rendezvous));
 524   ACE_DEBUG ((LM_DEBUG,
 525               ACE_TEXT ("(%t) Spawning %d proactor threads\n"),
 526               num_threads));
 527   ACE_Thread_Manager::instance ()->spawn_n (num_threads, proactor_loop);
 528   ACE_Thread_Manager::instance ()->spawn (start_clients);
 529
 530   ACE_Time_Value loop_limit (20);
 531   ACE_Reactor::instance ()->run_reactor_event_loop (loop_limit);
 532   ACE_Thread_Manager::instance ()->wait ();
 533
 534   // Check for num connections up/down.
 535
 536   ACE_END_TEST;
 537   return 0;
 538 }
 539
 540 #else
 541 int
 542 run_main (int, ACE_TCHAR *[])
 543 {
 544   ACE_START_TEST (ACE_TEXT ("SSL_Asynch_Stream_Test"));
 545
 546   ACE_ERROR ((LM_INFO,
 547               ACE_TEXT ("This test requires threads and AIO which are not ")
 548               ACE_TEXT ("supported on this platform\n")));
 549
 550   ACE_END_TEST;
 551   return 0;
 552 }
 553 #endif /* ACE_HAS_THREADS && (WIN32 || AIO) */