ACE/tests/SSL/Bug_2912_Regression_Test.cpp

   1 /**
   2  * @file Bug_2912_Regression_Test.cpp
   3  *
   4  * Reproduces the problems reported in bug 2912:
   5  *   http://bugzilla.dre.vanderbilt.edu/show_bug.cgi?id=2912
   6  *
   7  * This test reproduces the following interactions:
   8  *
   9  * ACE_TMAIN          Client proactor          Server proactor
  10  * thread             dispatcher thread        dispatcher thread
  11  * ----------------   -------------------      --------------------
  12  *
  13  * init_ssl
  14  * connect
  15  *                    SH::open                 SH::open
  16  *                    SH::read                 SH::read
  17  *                    SH::write
  18  *                                             SH::handle_read_stream
  19  *                                             SH::write
  20  *                                             SH::read
  21  *                    SH::handle_read_stream
  22  *                    SH::read
  23  * SH::write (causes do_SSL_read to fail)
  24  *
  25  * @author Paul Daugherty <paul@nxicom.com>
  26  */
  27
  28 #include "ace/ACE.h"
  29 #include "../test_config.h"
  30 #include "ace/SSL/SSL_Asynch_Stream.h"
  31 #include "ace/Proactor.h"
  32 #include "ace/Task.h"
  33 #include "ace/Asynch_Acceptor.h"
  34 #include "ace/Asynch_Connector.h"
  35 #include "ace/Manual_Event.h"
  36 #include "ace/OS_NS_unistd.h"
  37
  38 /* Linux kernels can't hack multiple outstanding I/O, which this
  39    test requires */
  40 #if defined (ACE_HAS_THREADS) && \
  41     (defined (ACE_HAS_WIN32_OVERLAPPED_IO) || defined (ACE_HAS_AIO_CALLS)) && \
  42     !defined (ACE_LINUX)
  43
  44 /**
  45  * Data payload sent between client and server.  Test is not dependent
  46  * on payload characteristics.
  47  */
  48 #define DATA "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
  49 #define DATA_SIZE ACE_OS::strlen(DATA)
  50
  51 // Function to remove signals from the signal mask.
  52 static int
  53 disable_signal (int sigmin, int sigmax)
  54 {
  55 #if !defined (ACE_LACKS_UNIX_SIGNALS)
  56   sigset_t signal_set;
  57   if (ACE_OS::sigemptyset (&signal_set) == - 1)
  58     ACE_ERROR ((LM_ERROR,
  59                 ACE_TEXT ("Error: (%P|%t):%p\n"),
  60                 ACE_TEXT ("sigemptyset failed")));
  61
  62   for (int i = sigmin; i <= sigmax; i++)
  63     ACE_OS::sigaddset (&signal_set, i);
  64
  65   // Put the <signal_set>.
  66 # if defined (ACE_LACKS_PTHREAD_THR_SIGSETMASK)
  67   // In multi-threaded application this is not POSIX compliant
  68   // but let's leave it just in case.
  69   if (ACE_OS::sigprocmask (SIG_BLOCK, &signal_set, 0) != 0)
  70 # else
  71   if (ACE_OS::thr_sigsetmask (SIG_BLOCK, &signal_set, 0) != 0)
  72 # endif /* ACE_LACKS_PTHREAD_THR_SIGSETMASK */
  73     ACE_ERROR_RETURN ((LM_ERROR,
  74                        ACE_TEXT ("Error: (%P|%t): %p\n"),
  75                        ACE_TEXT ("SIG_BLOCK failed")),
  76                       -1);
  77 #else
  78   ACE_UNUSED_ARG (sigmin);
  79   ACE_UNUSED_ARG (sigmax);
  80 #endif /* ACE_LACKS_UNIX_SIGNALS */
  81
  82   return 0;
  83 }
  84
  85 /**
  86  * Client's proactor
  87  *
  88  * The client and server must use different proactors since this bug is
  89  * dependent on threads.
  90  */
  91 class Client_Proactor : public ACE_Proactor {};
  92 typedef ACE_Singleton<Client_Proactor, ACE_SYNCH_RECURSIVE_MUTEX>
  93     Client_Proactor_Singleton;
  94 #define CLIENT_PROACTOR Client_Proactor_Singleton::instance()
  95
  96 class Client_Proactor_Task : public ACE_Task_Base
  97 {
  98 public:
  99   virtual int svc ();
 100 };
 101
 102 typedef ACE_Singleton<Client_Proactor_Task, ACE_SYNCH_RECURSIVE_MUTEX>
 103     Client_Proactor_Task_Singleton;
 104 #define CLIENT_PROACTOR_TASK Client_Proactor_Task_Singleton::instance()
 105
 106 int
 107 Client_Proactor_Task::svc ()
 108 {
 109   // Keep RT signals on POSIX from killing us.
 110   disable_signal (ACE_SIGRTMIN, ACE_SIGRTMAX);
 111
 112   CLIENT_PROACTOR->proactor_reset_event_loop ();
 113   CLIENT_PROACTOR->proactor_run_event_loop ();
 114   return 0;
 115 }
 116
 117 /**
 118  * Server's proactor
 119  */
 120 class Server_Proactor : public ACE_Proactor {};
 121 typedef ACE_Singleton<Server_Proactor, ACE_SYNCH_RECURSIVE_MUTEX>
 122     Server_Proactor_Singleton;
 123 #define SERVER_PROACTOR Server_Proactor_Singleton::instance ()
 124
 125 class Server_Proactor_Task : public ACE_Task_Base
 126 {
 127 public:
 128   virtual int svc ();
 129 };
 130
 131 typedef ACE_Singleton<Server_Proactor_Task, ACE_SYNCH_RECURSIVE_MUTEX>
 132     Server_Proactor_Task_Singleton;
 133 #define SERVER_PROACTOR_TASK Server_Proactor_Task_Singleton::instance ()
 134
 135 int
 136 Server_Proactor_Task::svc ()
 137 {
 138   // Keep RT signals on POSIX from killing us.
 139   disable_signal (ACE_SIGRTMIN, ACE_SIGRTMAX);
 140
 141   SERVER_PROACTOR->proactor_reset_event_loop ();
 142   SERVER_PROACTOR->proactor_run_event_loop ();
 143   return 0;
 144 }
 145
 146 /**
 147  * This test depends on ADH since the error is related to a missing cert.pem
 148  */
 149 static DH* dh1024 = 0;
 150
 151 DH*
 152 get_dh1024 ()
 153 {
 154   static unsigned char dh1024_p[]={
 155         0xD7,0xFE,0xEC,0x06,0x28,0x03,0x34,0x96,0xB8,0x08,0x86,0x62,
 156             0xF1,0xA2,0xBA,0x84,0x7C,0xAF,0xA3,0x1F,0x6A,0x3D,0x03,0x20,
 157             0x81,0x8D,0x0E,0x43,0x3A,0x54,0x74,0x9F,0x83,0xD2,0xB7,0xE9,
 158             0x57,0xC1,0x67,0xE9,0x11,0x38,0x2B,0x8E,0x9B,0x1C,0x5D,0x14,
 159             0x18,0x7D,0x4F,0xEB,0xB1,0x4D,0xFA,0x6F,0x06,0xDD,0xDD,0x6D,
 160             0x9A,0xD0,0x9E,0x4F,0xE4,0x04,0x3E,0x86,0x6F,0x15,0x60,0x35,
 161             0x9B,0xA1,0xBA,0x53,0xBA,0x84,0xB5,0x06,0xB1,0xAD,0x94,0x25,
 162             0xD1,0xED,0xD2,0xF4,0xD7,0x02,0x2F,0x35,0x25,0xE7,0x2D,0x60,
 163             0xEE,0x7A,0x61,0xAD,0x98,0xA8,0x3D,0xAD,0xB1,0x8A,0x5E,0xCE,
 164             0xF0,0x09,0xD4,0x67,0x28,0x3D,0x52,0x64,0x78,0xBB,0xC3,0x9D,
 165             0x40,0xF4,0x72,0xDC,0xC9,0x31,0x0D,0xA3,
 166   };
 167   static unsigned char dh1024_g[]={
 168     0x02,
 169   };
 170   DH *dh;
 171
 172   if ((dh=DH_new()) == 0) return(0);
 173
 174   BIGNUM* p = BN_bin2bn(dh1024_p,sizeof(dh1024_p),0);
 175   BIGNUM* g = BN_bin2bn(dh1024_g,sizeof(dh1024_g),0);
 176
 177   if ((p == 0) || (g == 0))
 178     {
 179       DH_free(dh);
 180       return(0);
 181     }
 182
 183 #if (OPENSSL_VERSION_NUMBER < 0x10100000L)
 184   dh->p = p;
 185   dh->g = g;
 186 #else
 187   DH_set0_pqg(dh, p, 0, g);
 188 #endif
 189
 190   return(dh);
 191 }
 192
 193 extern "C" DH*
 194 tmp_dh_callback (SSL *s, int is_export, int keylength)
 195 {
 196   ACE_UNUSED_ARG(s);
 197   ACE_UNUSED_ARG(is_export);
 198
 199   switch (keylength) {
 200   case 1024:
 201     if (dh1024 == 0)
 202     {
 203       dh1024 = get_dh1024();
 204     }
 205     return dh1024;
 206   default:
 207     /* Generating a key on the fly is very costly, so use what is there */
 208     ACE_ERROR ((LM_ERROR,
 209       ACE_TEXT ("tmp_dh_callback, unsupported key length, %d\n"),
 210       keylength));
 211   }
 212   return(0);
 213 }
 214
 215 bool
 216 init_ssl ()
 217 {
 218   SSL_CTX_set_quiet_shutdown (ACE_SSL_Context::instance ()->context(), 1);
 219   SSL_CTX_set_options (ACE_SSL_Context::instance ()->context(),
 220       SSL_OP_SINGLE_DH_USE);
 221   SSL_CTX_set_tmp_dh_callback (ACE_SSL_Context::instance ()->context (),
 222       tmp_dh_callback);
 223
 224   if (SSL_CTX_set_cipher_list (ACE_SSL_Context::instance ()->context (), "ADH"))
 225   {
 226     return true;
 227   }
 228   else
 229   {
 230     ACE_DEBUG ((LM_ERROR,
 231       ACE_TEXT ("SSL_CTX_set_cipher_list failed\n")));
 232     return false;
 233   }
 234 }
 235
 236
 237 /**
 238  * Server's ACE_Service_Handler
 239  */
 240 class Server_Service_Handler : public ACE_Service_Handler
 241 {
 242 public:
 243   Server_Service_Handler ();
 244
 245   virtual ~Server_Service_Handler ();
 246
 247   //FUZZ: disable check_for_lack_ACE_OS
 248   virtual void open (ACE_HANDLE h, ACE_Message_Block&);
 249   //FUZZ: enable check_for_lack_ACE_OS
 250
 251   virtual void handle_read_stream (
 252     const ACE_Asynch_Read_Stream::Result &result);
 253
 254   virtual void handle_write_stream (
 255     const ACE_Asynch_Write_Stream::Result &result);
 256
 257   virtual void handle_wakeup ();
 258
 259   void cancel_and_close ();
 260
 261   int read_data ();
 262
 263   int write_data ();
 264
 265   //FUZZ: disable check_for_lack_ACE_OS
 266   int read (ACE_Message_Block &mb, size_t bytes_to_read);
 267
 268   int write (ACE_Message_Block &mb, size_t bytes_to_write);
 269   //FUZZ: enable check_for_lack_ACE_OS
 270
 271   int safe_to_delete () const;
 272
 273 private:
 274   mutable ACE_SYNCH_RECURSIVE_MUTEX  mtx_;
 275   ACE_SSL_Asynch_Stream ssl_stream_;
 276   int pending_writes_;
 277   int pending_reads_;
 278   int handle_wakeup_expected_;
 279   int handle_wakeup_received_;
 280   int closing_;
 281 };
 282
 283 Server_Service_Handler::Server_Service_Handler () :
 284   ssl_stream_ (ACE_SSL_Asynch_Stream::ST_SERVER),
 285   pending_writes_ (0),
 286   pending_reads_ (0),
 287   handle_wakeup_expected_ (0),
 288   handle_wakeup_received_ (0),
 289   closing_ (0)
 290 {
 291 }
 292
 293 Server_Service_Handler::~Server_Service_Handler ()
 294 {
 295   if (ACE_INVALID_HANDLE != this->handle ())
 296   {
 297     ACE_OS::closesocket (this->handle ());
 298     this->handle (ACE_INVALID_HANDLE);
 299   }
 300 }
 301
 302 void
 303 Server_Service_Handler::open (ACE_HANDLE h, ACE_Message_Block&)
 304 {
 305   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 306
 307   if (this->ssl_stream_.open (*this, h, 0, this->proactor ()) != 0)
 308   {
 309     //FUZZ: disable check_for_lack_ACE_OS
 310     ACE_DEBUG ((LM_DEBUG,
 311                 ACE_TEXT("Server_Service_Handler::open: ")
 312                 ACE_TEXT("ACE_SSL_Asynch_Stream::open failed, %d\n"),
 313                 (int)errno));
 314     //FUZZ: enable check_for_lack_ACE_OS
 315
 316     this->cancel_and_close ();
 317   }
 318   else
 319   {
 320     if (this->read_data () < 0)
 321     {
 322       this->cancel_and_close ();
 323     }
 324   }
 325 }
 326
 327 void
 328 Server_Service_Handler::handle_read_stream(
 329   const ACE_Asynch_Read_Stream::Result &result)
 330 {
 331   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 332
 333   this->pending_reads_--;
 334
 335   if (!result.success () || 0 == result.bytes_transferred ())
 336   {
 337     // Error or remote disconnect
 338
 339     result.message_block ().release ();
 340
 341     if (!this->closing_)
 342     {
 343       // No error message when shutting down
 344
 345       if (!result.success ())
 346       {
 347         ACE_DEBUG ((LM_DEBUG,
 348           ACE_TEXT ("Server_Service_Handler::handle_read_stream: error: %d\n"),
 349           result.error ()));
 350       }
 351       else
 352       {
 353         ACE_DEBUG ((LM_DEBUG,
 354           ACE_TEXT ("Server_Service_Handler::handle_read_stream: remote disconnect\n")));
 355       }
 356     }
 357
 358     this->cancel_and_close ();
 359   }
 360   else if (result.bytes_transferred () < result.bytes_to_read ())
 361   {
 362     // More to read...
 363
 364     if (this->read (result.message_block (),
 365       result.bytes_to_read () - result.bytes_transferred ()) < 0)
 366     {
 367       result.message_block ().release ();
 368
 369       this->cancel_and_close ();
 370     }
 371   }
 372   else
 373   {
 374     // Read it all
 375
 376     result.message_block ().release ();
 377
 378     // Send data back
 379
 380     if (this->write_data () < 0)
 381     {
 382       this->cancel_and_close ();
 383     }
 384
 385     // Next read
 386
 387     else if (this->read_data () < 0)
 388     {
 389       this->cancel_and_close ();
 390     }
 391   }
 392 }
 393
 394 void
 395 Server_Service_Handler::handle_write_stream (
 396   const ACE_Asynch_Write_Stream::Result &result)
 397 {
 398   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 399
 400   this->pending_writes_--;
 401
 402   if (!result.success () || 0 == result.bytes_transferred ())
 403   {
 404     // Error
 405
 406     result.message_block ().release ();
 407
 408     ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 409       "Server_Service_Handler::handle_write_stream: error: %d\n"),
 410       result.error ()));
 411
 412     this->cancel_and_close ();
 413   }
 414   else if (result.bytes_transferred () < result.bytes_to_write ())
 415   {
 416     // More to write...
 417
 418     if (this->write(result.message_block (),
 419       result.bytes_to_write () - result.bytes_transferred ()) < 0)
 420     {
 421       result.message_block ().release ();
 422
 423       this->cancel_and_close ();
 424     }
 425   }
 426   else
 427   {
 428     // Wrote it all
 429
 430     result.message_block().release ();
 431   }
 432 }
 433
 434 void
 435 Server_Service_Handler::handle_wakeup ()
 436 {
 437   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 438
 439   this->handle_wakeup_received_ = 1;
 440 }
 441
 442 void
 443 Server_Service_Handler::cancel_and_close ()
 444 {
 445   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 446
 447   this->closing_ = 1;
 448   this->ssl_stream_.cancel ();
 449   this->handle_wakeup_expected_ = -1 == this->ssl_stream_.close ();
 450 }
 451
 452 int
 453 Server_Service_Handler::read_data ()
 454 {
 455   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 456
 457   ACE_Message_Block *mb = 0;
 458   ACE_NEW_NORETURN(mb, ACE_Message_Block (DATA_SIZE));
 459
 460   int ret = this->read (*mb, DATA_SIZE);
 461   if (ret < 0)
 462   {
 463     mb->release ();
 464     return -1;
 465   }
 466   else
 467   {
 468     return 0;
 469   }
 470 }
 471
 472 int
 473 Server_Service_Handler::write_data ()
 474 {
 475   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 476
 477   ACE_Message_Block *mb = 0;
 478   ACE_NEW_NORETURN(mb, ACE_Message_Block (DATA_SIZE));
 479   ACE_OS::memcpy (mb->wr_ptr (), DATA, DATA_SIZE);
 480   mb->wr_ptr (DATA_SIZE);
 481
 482   int ret = this->write (*mb, DATA_SIZE);
 483   if (ret < 0)
 484   {
 485     mb->release ();
 486     return -1;
 487   }
 488   else
 489   {
 490     return 0;
 491   }
 492 }
 493
 494 int
 495 Server_Service_Handler::read (ACE_Message_Block &mb, size_t bytes_to_read)
 496 {
 497   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 498
 499   int ret;
 500   if ((ret = this->ssl_stream_.read (mb, bytes_to_read)) < 0)
 501   {
 502     ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 503       "Server_Service_Handler::read: read failed: %d\n"), (int)errno));
 504   }
 505   else
 506   {
 507     this->pending_reads_++;
 508   }
 509   return ret;
 510 }
 511
 512 int
 513 Server_Service_Handler::write (ACE_Message_Block &mb, size_t bytes_to_write)
 514 {
 515   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 516
 517   int ret;
 518   if ((ret = this->ssl_stream_.write (mb, bytes_to_write)) < 0)
 519   {
 520     ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 521       "Server_Service_Handler::write: write failed: %d\n"), (int)errno));
 522   }
 523   else
 524   {
 525     this->pending_writes_++;
 526   }
 527   return ret;
 528 }
 529
 530 int
 531 Server_Service_Handler::safe_to_delete () const
 532 {
 533   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 534
 535   return 0 == this->pending_writes_ &&
 536     0 == this->pending_reads_ &&
 537     (!this->handle_wakeup_expected_ || this->handle_wakeup_received_);
 538 }
 539
 540 /**
 541  * Server's acceptor
 542  */
 543 class Acceptor : public ACE_Asynch_Acceptor<Server_Service_Handler>
 544 {
 545 public:
 546   Acceptor ();
 547
 548   virtual ~Acceptor ();
 549
 550   virtual int cancel ();
 551
 552   virtual int validate_connection (const ACE_Asynch_Accept::Result& result,
 553     const ACE_INET_Addr &remote, const ACE_INET_Addr& local);
 554
 555   virtual Server_Service_Handler *make_handler ();
 556
 557   //FUZZ: disable check_for_lack_ACE_OS
 558   virtual int accept (size_t bytes_to_read = 0, const void *act = 0);
 559   //FUZZ: enable check_for_lack_ACE_OS
 560
 561   virtual void handle_accept (const ACE_Asynch_Accept::Result &result);
 562
 563   int safe_to_delete () const;
 564
 565   void prepare_for_connection (Server_Service_Handler *service_handler);
 566
 567   mutable ACE_SYNCH_RECURSIVE_MUTEX  mtx_;
 568   int accept_cnt_;
 569   int cancel_flag_;
 570   Server_Service_Handler *service_handler_;
 571 };
 572
 573 typedef ACE_Singleton<Acceptor, ACE_SYNCH_RECURSIVE_MUTEX> Acceptor_Singleton;
 574 #define ACCEPTOR Acceptor_Singleton::instance ()
 575
 576 Acceptor::Acceptor () :
 577   accept_cnt_ (0),
 578   cancel_flag_ (0),
 579   service_handler_ (0)
 580 {
 581 }
 582
 583 Acceptor::~Acceptor ()
 584 {
 585 }
 586
 587 int
 588 Acceptor::cancel ()
 589 {
 590   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 591
 592   this->cancel_flag_ = 1;
 593   this->reissue_accept (0);
 594
 595   ACE_HANDLE h = this->get_handle ();
 596   if (h != ACE_INVALID_HANDLE)
 597   {
 598     this->ACE_Asynch_Acceptor<Server_Service_Handler>::cancel ();
 599
 600     ACE_OS::closesocket (h);
 601     this->handle (ACE_INVALID_HANDLE);
 602   }
 603
 604   return 0;
 605 }
 606
 607 int
 608 Acceptor::safe_to_delete () const
 609 {
 610   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 611   return (this->cancel_flag_ != 0 && this->accept_cnt_ == 0) ? 1 : 0;
 612 }
 613
 614 void
 615 Acceptor::prepare_for_connection (Server_Service_Handler *service_handler)
 616 {
 617   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 618   this->service_handler_ = service_handler;
 619 }
 620
 621 int
 622 Acceptor::validate_connection (const ACE_Asynch_Accept::Result& result,
 623   const ACE_INET_Addr & /*remote*/, const ACE_INET_Addr& /*local*/)
 624 {
 625   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 626
 627   if (0 != this->service_handler_ && result.success ())
 628   {
 629     return 0;
 630   }
 631   else
 632   {
 633     return -1;
 634   }
 635 }
 636
 637 Server_Service_Handler*
 638 Acceptor::make_handler ()
 639 {
 640   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, 0);
 641
 642   ACE_ASSERT (0 != this->service_handler_);
 643   Server_Service_Handler *service_handler = this->service_handler_;
 644   this->service_handler_ = 0;
 645   return service_handler;
 646 }
 647
 648 int
 649 Acceptor::accept (size_t bytes_to_read, const void *act)
 650 {
 651   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 652
 653   if  (this->cancel_flag_!= 0)
 654     return -1;
 655
 656   this->accept_cnt_++;
 657   int rc = this->ACE_Asynch_Acceptor<Server_Service_Handler>::accept (
 658     bytes_to_read, act);
 659   if (rc != 0)
 660     this->accept_cnt_--;
 661
 662   return rc;
 663 }
 664
 665 void
 666 Acceptor::handle_accept (const ACE_Asynch_Accept::Result &result)
 667 {
 668   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 669
 670   this->ACE_Asynch_Acceptor<Server_Service_Handler>::handle_accept (result);
 671
 672   --this->accept_cnt_;
 673 }
 674
 675
 676 /**
 677  * Client's ACE_Service_Handler
 678  */
 679 class Client_Service_Handler : public ACE_Service_Handler
 680 {
 681 public:
 682   Client_Service_Handler ();
 683
 684   virtual ~Client_Service_Handler ();
 685
 686   //FUZZ: disable check_for_lack_ACE_OS
 687   virtual void open (ACE_HANDLE h, ACE_Message_Block&);
 688   //FUZZ: enable check_for_lack_ACE_OS
 689
 690   virtual void handle_read_stream (
 691     const ACE_Asynch_Read_Stream::Result &result);
 692
 693   virtual void handle_write_stream (
 694     const ACE_Asynch_Write_Stream::Result &result);
 695
 696   virtual void handle_wakeup ();
 697
 698   void cancel_and_close ();
 699
 700   int read_data ();
 701
 702   int write_data ();
 703
 704   //FUZZ: disable check_for_lack_ACE_OS
 705   int read (ACE_Message_Block &mb, size_t bytes_to_read);
 706
 707   int write (ACE_Message_Block &mb, size_t bytes_to_write);
 708   //FUZZ: enable check_for_lack_ACE_OS
 709
 710   int safe_to_delete () const;
 711
 712   int wait_for_external_write_queue (ACE_Time_Value *wait_time);
 713
 714   int wait_for_read_completed (ACE_Time_Value *wait_time);
 715
 716   int read_successful () const;
 717
 718 private:
 719   mutable ACE_SYNCH_RECURSIVE_MUTEX  mtx_;
 720   ACE_SSL_Asynch_Stream ssl_stream_;
 721   ACE_Manual_Event ready_for_external_write_;
 722   ACE_Manual_Event read_completed_;
 723   int read_successful_;
 724   int pending_writes_;
 725   int pending_reads_;
 726   int handle_wakeup_expected_;
 727   int handle_wakeup_received_;
 728   int completed_reads_;
 729   int closing_;
 730 };
 731
 732 Client_Service_Handler::Client_Service_Handler () :
 733   ssl_stream_ (ACE_SSL_Asynch_Stream::ST_CLIENT),
 734   read_successful_ (0),
 735   pending_writes_ (0),
 736   pending_reads_ (0),
 737   handle_wakeup_expected_ (0),
 738   handle_wakeup_received_ (0),
 739   completed_reads_ (0),
 740   closing_ (0)
 741 {
 742 }
 743
 744 Client_Service_Handler::~Client_Service_Handler ()
 745 {
 746   if (ACE_INVALID_HANDLE != this->handle ())
 747   {
 748     ACE_OS::closesocket (this->handle ());
 749     this->handle (ACE_INVALID_HANDLE);
 750   }
 751 }
 752
 753 void
 754 Client_Service_Handler::open (ACE_HANDLE h, ACE_Message_Block&)
 755 {
 756   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 757
 758   if (this->ssl_stream_.open (*this, h, 0, this->proactor ()) != 0)
 759   {
 760     ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 761       "Client_Service_Handler::open: ACE_SSL_Asynch_Stream::open failed, %d\n"),
 762       (int)errno));
 763     this->cancel_and_close ();
 764   }
 765   else
 766   {
 767     ACE_Message_Block *mb = 0;
 768     ACE_NEW_NORETURN(mb, ACE_Message_Block (DATA_SIZE));
 769
 770     if (this->read_data () < 0 || this->write_data () < 0)
 771     {
 772       this->cancel_and_close ();
 773     }
 774   }
 775 }
 776
 777 void
 778 Client_Service_Handler::handle_read_stream (
 779   const ACE_Asynch_Read_Stream::Result &result)
 780 {
 781   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 782
 783   this->pending_reads_--;
 784
 785   if (!result.success () || 0 == result.bytes_transferred ())
 786   {
 787     // Error or remote disconnect
 788
 789     result.message_block ().release ();
 790
 791     if (!this->closing_)
 792     {
 793       // No error message when shutting down
 794
 795       if (!result.success ())
 796       {
 797         ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 798           "Client_Service_Handler::handle_read_stream: error: %d\n"),
 799           result.error ()));
 800       }
 801       else
 802       {
 803         ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 804           "Client_Service_Handler::handle_read_stream: remote disconnect\n")));
 805       }
 806     }
 807
 808     this->read_completed_.signal ();
 809
 810     this->cancel_and_close ();
 811   }
 812   else if (result.bytes_transferred () < result.bytes_to_read ())
 813   {
 814     // More to read...
 815
 816     if (this->read (result.message_block(),
 817       result.bytes_to_read () - result.bytes_transferred ()) < 0)
 818     {
 819       result.message_block ().release ();
 820
 821       this->cancel_and_close ();
 822     }
 823   }
 824   else
 825   {
 826     // Read it all
 827
 828     this->completed_reads_++;
 829
 830     result.message_block ().release ();
 831
 832     // We now have sent and received data in the proactor thread.  Signal the
 833     // main thread to try sending data in the main thread.
 834     if (this->completed_reads_ == 1)
 835     {
 836       this->ready_for_external_write_.signal ();
 837     }
 838     else
 839     {
 840       // The main thread wrote data that was echoed back to us on our
 841       // second read.  If we get here, the test was successful in that
 842       // the main thread successfully sent data to the server.
 843       this->read_successful_ = 1;
 844       this->read_completed_.signal ();
 845     }
 846
 847     // Next read
 848     if (this->read_data () < 0)
 849     {
 850       this->cancel_and_close ();
 851     }
 852   }
 853 }
 854
 855 void
 856 Client_Service_Handler::handle_write_stream (
 857   const ACE_Asynch_Write_Stream::Result &result)
 858 {
 859   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 860
 861   this->pending_writes_--;
 862
 863   if (!result.success () || 0 == result.bytes_transferred ())
 864   {
 865     // Error
 866
 867     result.message_block ().release ();
 868
 869     ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 870       "Client_Service_Handler::handle_write_stream: error: %d\n"),
 871       result.error ()));
 872
 873     this->cancel_and_close ();
 874   }
 875   else if (result.bytes_transferred () < result.bytes_to_write ())
 876   {
 877     // More to write...
 878
 879     if (this->write (result.message_block(),
 880       result.bytes_to_write () - result.bytes_transferred ()) < 0)
 881     {
 882       result.message_block ().release ();
 883
 884       this->cancel_and_close ();
 885     }
 886   }
 887   else
 888   {
 889     // Wrote it all
 890
 891     result.message_block ().release ();
 892   }
 893 }
 894
 895 void
 896 Client_Service_Handler::handle_wakeup ()
 897 {
 898   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 899
 900   this->handle_wakeup_received_ = 1;
 901 }
 902
 903 void
 904 Client_Service_Handler::cancel_and_close ()
 905 {
 906   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
 907
 908   this->closing_ = 1;
 909   this->ssl_stream_.cancel ();
 910   this->handle_wakeup_expected_ = -1 == this->ssl_stream_.close ();
 911 }
 912
 913 int
 914 Client_Service_Handler::read_data ()
 915 {
 916   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 917
 918   ACE_Message_Block *mb = 0;
 919   ACE_NEW_NORETURN(mb, ACE_Message_Block (DATA_SIZE));
 920
 921   int ret = this->read (*mb, DATA_SIZE);
 922   if (ret < 0)
 923   {
 924     mb->release ();
 925     return -1;
 926   }
 927   else
 928   {
 929     return 0;
 930   }
 931 }
 932
 933 int
 934 Client_Service_Handler::write_data ()
 935 {
 936   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 937
 938   ACE_Message_Block *mb = 0;
 939   ACE_NEW_NORETURN (mb, ACE_Message_Block (DATA_SIZE));
 940   ACE_OS::memcpy (mb->wr_ptr (), DATA, DATA_SIZE);
 941   mb->wr_ptr (DATA_SIZE);
 942
 943   int ret = this->write (*mb, DATA_SIZE);
 944   if (ret < 0)
 945   {
 946     mb->release ();
 947     return -1;
 948   }
 949   else
 950   {
 951     return 0;
 952   }
 953 }
 954
 955 int
 956 Client_Service_Handler::read (ACE_Message_Block &mb, size_t bytes_to_read)
 957 {
 958   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 959
 960   int ret;
 961   if ((ret = this->ssl_stream_.read (mb, bytes_to_read)) < 0)
 962   {
 963     ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 964         "Client_Service_Handler::read: read failed: %d\n"), (int)errno));
 965   }
 966   else
 967   {
 968     this->pending_reads_++;
 969   }
 970   return ret;
 971 }
 972
 973 int
 974 Client_Service_Handler::write (ACE_Message_Block &mb, size_t bytes_to_write)
 975 {
 976   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 977
 978   int ret;
 979   if ((ret = this->ssl_stream_.write (mb, bytes_to_write)) < 0)
 980   {
 981     ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
 982         "Client_Service_Handler::write: write failed: %d\n"), (int)errno));
 983   }
 984   else
 985   {
 986     this->pending_writes_++;
 987   }
 988   return ret;
 989 }
 990
 991 int
 992 Client_Service_Handler::safe_to_delete () const
 993 {
 994   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
 995
 996   return 0 == this->pending_writes_ &&
 997     0 == this->pending_reads_ &&
 998     (!this->handle_wakeup_expected_ || this->handle_wakeup_received_);
 999 }
1000
1001 int
1002 Client_Service_Handler::wait_for_external_write_queue (ACE_Time_Value *wait_time)
1003 {
1004   return this->ready_for_external_write_.wait (wait_time, 0);
1005 }
1006
1007 int
1008 Client_Service_Handler::wait_for_read_completed (ACE_Time_Value *wait_time)
1009 {
1010   return this->read_completed_.wait (wait_time, 0);
1011 }
1012
1013 int
1014 Client_Service_Handler::read_successful () const
1015 {
1016   return this->read_successful_;
1017 }
1018
1019 /**
1020  * Client's connector
1021  */
1022 class Connector : public ACE_Asynch_Connector<Client_Service_Handler>
1023 {
1024 public:
1025   Connector ();
1026
1027   virtual ~Connector ();
1028
1029   //FUZZ: disable check_for_lack_ACE_OS
1030   virtual int connect (
1031     const ACE_INET_Addr &remote_sap,
1032     const ACE_INET_Addr &local_sap =
1033       (const ACE_INET_Addr &)ACE_Addr::sap_any,
1034     int reuse_addr = 1,
1035     const void *act = 0);
1036   //FUZZ: enable check_for_lack_ACE_OS
1037
1038   virtual int validate_connection (
1039     const ACE_Asynch_Connect::Result& result,
1040     const ACE_INET_Addr &remote,
1041     const ACE_INET_Addr& local);
1042
1043   int safe_to_delete () const;
1044
1045   void prepare_for_connection (Client_Service_Handler *service_handler);
1046
1047 protected:
1048   virtual void handle_connect (const ACE_Asynch_Connect::Result &result);
1049
1050   virtual Client_Service_Handler* make_handler ();
1051
1052   mutable ACE_SYNCH_RECURSIVE_MUTEX  mtx_;
1053   Client_Service_Handler *service_handler_;
1054   int connecting_;
1055 };
1056
1057 typedef ACE_Singleton<Connector, ACE_SYNCH_RECURSIVE_MUTEX> Connector_Singleton;
1058 #define CONNECTOR Connector_Singleton::instance ()
1059
1060 Connector::Connector () :
1061   service_handler_ (0)
1062 {
1063 }
1064
1065 Connector::~Connector ()
1066 {
1067 }
1068
1069 int
1070 Connector::connect (const ACE_INET_Addr &remote_sap,
1071   const ACE_INET_Addr &local_sap, int reuse_addr, const void *act)
1072 {
1073   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
1074
1075   this->connecting_ = 1;
1076
1077   return this->ACE_Asynch_Connector<Client_Service_Handler>::connect (
1078     remote_sap, local_sap, reuse_addr, act);
1079 }
1080
1081 int
1082 Connector::validate_connection (const ACE_Asynch_Connect::Result& result,
1083   const ACE_INET_Addr &remote, const ACE_INET_Addr& local)
1084 {
1085   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
1086
1087   ACE_UNUSED_ARG (result);
1088   ACE_UNUSED_ARG (remote);
1089   ACE_UNUSED_ARG (local);
1090
1091   if (!result.success ())
1092   {
1093     ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
1094       "Connector::validate_connection failed: %d\n"), result.error ()));
1095     return -1;
1096   }
1097   else
1098   {
1099     return 0;
1100   }
1101 }
1102
1103 void
1104 Connector::handle_connect (const ACE_Asynch_Connect::Result &result)
1105 {
1106   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
1107
1108   this->ACE_Asynch_Connector<Client_Service_Handler>::handle_connect (result);
1109
1110   this->connecting_ = 0;
1111 }
1112
1113 Client_Service_Handler*
1114 Connector::make_handler ()
1115 {
1116   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, 0);
1117
1118   ACE_ASSERT(0 != this->service_handler_);
1119   Client_Service_Handler *service_handler = this->service_handler_;
1120   this->service_handler_ = 0;
1121   return service_handler;
1122 }
1123
1124 int
1125 Connector::safe_to_delete () const
1126 {
1127   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_, -1);
1128
1129   return 0 == this->connecting_;
1130 }
1131
1132 void
1133 Connector::prepare_for_connection (Client_Service_Handler *service_handler)
1134 {
1135   ACE_GUARD (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->mtx_);
1136   this->service_handler_ = service_handler;
1137 }
1138
1139 int
1140 run_main (int, ACE_TCHAR *[])
1141 {
1142   ACE_START_TEST (ACE_TEXT ("Bug_2912_Regression_Test"));
1143
1144   // SSL_CTX_set_cipher_list, etc.
1145   init_ssl ();
1146
1147   // Keep RT signals on POSIX from killing us.
1148   disable_signal (ACE_SIGRTMIN, ACE_SIGRTMAX);
1149
1150   int ret = 0;
1151   Client_Service_Handler client_handler;
1152   Server_Service_Handler server_handler;
1153   ACE_Time_Value wait_time (10, 0);
1154
1155   // Client and Server will utilize different proactors since this test
1156   // depends on SSL thread error state behavior.
1157
1158   CLIENT_PROACTOR_TASK->activate ();
1159   SERVER_PROACTOR_TASK->activate ();
1160
1161   // Open server acceptor and client connector
1162
1163   if (0 == ret)
1164   {
1165     ret = ACCEPTOR->open (
1166       ACE_INET_Addr (ACE_DEFAULT_SERVER_PORT),
1167       0,
1168       0,
1169       ACE_DEFAULT_ASYNCH_BACKLOG,
1170       1,
1171       SERVER_PROACTOR,
1172       1);
1173
1174       if (-1 == ret)
1175       {
1176         ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
1177           "ACE_Asynch_Acceptor::open failed, %d\n"), (int)errno));
1178       }
1179   }
1180
1181   if (0 == ret)
1182   {
1183     ret = CONNECTOR->open (0, CLIENT_PROACTOR, 1);
1184
1185     if (-1 == ret)
1186     {
1187       ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
1188         "ACE_Asynch_Connector::open failed, %d\n"), (int)errno));
1189     }
1190   }
1191
1192   // Supply server_handler and client_handler to acceptor and connector and
1193   // connect client to the server.
1194
1195   if (0 == ret)
1196   {
1197     ACCEPTOR->prepare_for_connection (&server_handler);
1198     CONNECTOR->prepare_for_connection (&client_handler);
1199
1200     ret = CONNECTOR->connect (
1201       ACE_INET_Addr (ACE_DEFAULT_SERVER_PORT, ACE_LOCALHOST));
1202
1203     if (-1 == ret)
1204     {
1205       ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
1206         "ACE_Asynch_Connector::connect failed, %d\n"), (int)errno));
1207     }
1208   }
1209
1210   if (0 == ret)
1211   {
1212     ret = client_handler.wait_for_external_write_queue (&wait_time);
1213     if (-1 == ret)
1214     {
1215       ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
1216         "Timed out waiting for client's write readiness\n")));
1217     }
1218   }
1219
1220   // Client sends data to server
1221
1222   if (0 == ret)
1223   {
1224     ret = client_handler.write_data ();
1225   }
1226
1227   // Client waits for echo reply from server
1228
1229   if (0 == ret)
1230   {
1231     ret = client_handler.wait_for_read_completed (&wait_time);
1232
1233     if (-1 == ret)
1234     {
1235       ACE_DEBUG ((LM_DEBUG, ACE_TEXT (
1236         "Timed out waiting for client's read to complete\n")));
1237     }
1238   }
1239
1240   if (0 == ret)
1241   {
1242     if (client_handler.read_successful () == 1)
1243     {
1244       ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Success\n")));
1245       ret = 0;
1246     }
1247     else
1248     {
1249       ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Client's read failed\n")));
1250       ret = -1;
1251     }
1252   }
1253
1254   // Cleanup and shutdown
1255
1256   ACCEPTOR->cancel ();
1257   while (!ACCEPTOR->safe_to_delete ())
1258     ACE_OS::sleep (ACE_Time_Value (0, 500000));
1259
1260   CONNECTOR->cancel ();
1261   while (!CONNECTOR->safe_to_delete ())
1262     ACE_OS::sleep (ACE_Time_Value (0, 500000));
1263
1264   client_handler.cancel_and_close ();
1265   while (!client_handler.safe_to_delete ())
1266     ACE_OS::sleep (ACE_Time_Value (0, 500000));
1267
1268   server_handler.cancel_and_close ();
1269   while (!server_handler.safe_to_delete ())
1270     ACE_OS::sleep (ACE_Time_Value (0, 500000));
1271
1272   CLIENT_PROACTOR->proactor_end_event_loop ();
1273   CLIENT_PROACTOR_TASK->wait ();
1274
1275   SERVER_PROACTOR->proactor_end_event_loop ();
1276   SERVER_PROACTOR_TASK->wait ();
1277
1278   ACE_END_TEST;
1279
1280   return 0;
1281 }
1282
1283 #else
1284 int
1285 run_main (int, ACE_TCHAR *[])
1286 {
1287   ACE_START_TEST (ACE_TEXT ("Bug_2912_Regression_Test"));
1288
1289   ACE_DEBUG ((LM_INFO,
1290               ACE_TEXT ("threads or proactor not supported on this platform\n")));
1291
1292   ACE_END_TEST;
1293   return 0;
1294 }
1295 #endif /* ACE_HAS_THREADS && (ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS) && !linux */