ACE/tests/TP_Reactor_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    TP_Reactor_Test.cpp
   5  *
   6  *  This program illustrates how the <ACE_TP_Reactor> can be used to
   7  *  implement an application that does various  operations.
   8  *  usage: TP_Reactor_Test
   9  *         -n number threads in the TP_Reactor thread pool
  10  *         -d duplex mode 1 (full-duplex) vs. 0 (half-duplex)
  11  *         -p port to listen(Server)/connect(Client)
  12  *         -h host to connect (Client mode)
  13  *         -s number of sender's instances ( Client mode)
  14  *         -b run client and server (both modes ) at the same time
  15  *         -v log level
  16  *            0 - log all messages
  17  *            1 - log only errors and unusual cases
  18  *         -i time to run in seconds
  19  *         -u show this message
  20  *
  21  *   The main differences between Thread_Pool_Reactor_Test.cpp and
  22  *   this test are:
  23  *
  24  *   1. Thread_Pool_Reactor_Test.cpp tests only handle_input()
  25  *   events on the server, whereas this one tests both handle_input() and
  26  *   handle_output() on both server and client, i.e., the receiver
  27  *   and sender are completely event-driven.
  28  *
  29  *   2. The receiver and sender in this test can work in full duplex
  30  *   mode, i.e., input and ouput events are processed independently.
  31  *   Half-duplex mode (request-reply) is also supported.
  32  *
  33  *   This test is therefore a bit more stressful than the
  34  *   Thread_Pool_Reactor.cpp for the ACE_TP_Reactor since same
  35  *   thread pool is shared between client and server.
  36  *
  37  *   This test is a "twin" of the Proactor_Test.cpp, so it can help for
  38  *   developers to provide independent of Reactor/Proactor solutions.
  39  *
  40  *  @author   Alexander Libman <alibman@ihug.com.au>
  41  *  @author <alexl@rumblgroup.com>
  42  */
  43 //=============================================================================
  44
  45
  46 #include "test_config.h"
  47
  48 #if defined(ACE_HAS_THREADS)
  49
  50 #include "TP_Reactor_Test.h"
  51
  52 #include "ace/Signal.h"
  53 #include "ace/Service_Config.h"
  54 #include "ace/Get_Opt.h"
  55
  56 #include "ace/Reactor.h"
  57 #include "ace/TP_Reactor.h"
  58 #include "ace/OS_NS_signal.h"
  59 #include "ace/OS_NS_stdio.h"
  60 #include "ace/OS_NS_string.h"
  61 #include "ace/OS_NS_unistd.h"
  62 #include "ace/Synch_Traits.h"
  63 #include "ace/Thread_Semaphore.h"
  64
  65 //  Some debug helper functions
  66 static int disable_signal (int sigmin, int sigmax);
  67
  68 // both: 0 run client or server / depends on host
  69 //     != 0 run client and server
  70 static int both = 0;
  71
  72 // Host that we're connecting to.
  73 static const ACE_TCHAR *host = 0;
  74
  75 // number of Senders instances
  76 static int senders = 1;
  77
  78 // duplex mode: == 0 half-duplex
  79 //              != 0 full duplex
  80 static int duplex = 0;
  81
  82 // number threads in the TP_Reactor thread pool
  83 static int threads = 1;
  84
  85 // Port that we're receiving connections on.
  86 static u_short port = ACE_DEFAULT_SERVER_PORT;
  87
  88 // Log options
  89 static int loglevel = 1; // 0 full , 1 only errors
  90
  91 static const size_t MIN_TIME = 1;    // min 1 sec
  92 static const size_t MAX_TIME = 3600; // max 1 hour
  93 static u_int seconds = 2;  // default time to run - 2 seconds
  94
  95 static char data[] =
  96   "GET / HTTP/1.1\r\n"
  97   "Accept: */*\r\n"
  98   "Accept-Language: C++\r\n"
  99   "Accept-Encoding: gzip, deflate\r\n"
 100   "User-Agent: TPReactor_Test/1.0 (non-compatible)\r\n"
 101   "Connection: Keep-Alive\r\n"
 102   "\r\n" ;
 103
 104 // *************************************************************
 105
 106 class LogLocker
 107 {
 108 public:
 109   LogLocker () { ACE_LOG_MSG->acquire (); }
 110   virtual ~LogLocker () { ACE_LOG_MSG->release (); }
 111 };
 112 // *************************************************************
 113
 114 /**
 115  * @class MyTask
 116  *
 117  * MyTask plays role for TP_Reactor threads pool
 118  *
 119  * MyTask is ACE_Task resposible for:
 120  * 1. Creation and deletion of TP_Reactor and TP_Reactor thread pool
 121  * 2. Running TP_Reactor event loop
 122  */
 123 class MyTask : public ACE_Task<ACE_MT_SYNCH>
 124 {
 125 public:
 126   MyTask (): sem_ ((unsigned int) 0),
 127                  my_reactor_ (0) {}
 128
 129   ~MyTask () override { stop (); }
 130
 131   int svc () override;
 132
 133   int start (int num_threads);
 134   int stop ();
 135
 136 private:
 137   int  create_reactor ();
 138   int  delete_reactor ();
 139
 140   ACE_SYNCH_RECURSIVE_MUTEX lock_;
 141   ACE_Thread_Semaphore sem_;
 142   ACE_Reactor *my_reactor_;
 143 };
 144
 145 int
 146 MyTask::create_reactor ()
 147 {
 148   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX,
 149                     monitor,
 150                     this->lock_,
 151                     -1);
 152
 153   ACE_TEST_ASSERT (this->my_reactor_ == 0);
 154
 155   ACE_TP_Reactor * pImpl = 0;
 156
 157   ACE_NEW_RETURN (pImpl,ACE_TP_Reactor, -1);
 158
 159   ACE_NEW_RETURN (my_reactor_,
 160                    ACE_Reactor (pImpl ,1),
 161                    -1);
 162
 163   ACE_DEBUG ((LM_DEBUG,
 164               ACE_TEXT (" (%t) Create TP_Reactor\n")));
 165
 166   ACE_Reactor::instance (this->my_reactor_);
 167
 168   this->reactor (my_reactor_);
 169
 170   return 0;
 171 }
 172
 173 int
 174 MyTask::delete_reactor ()
 175 {
 176   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX,
 177                     monitor,
 178                     this->lock_,
 179                     -1);
 180
 181   ACE_DEBUG ((LM_DEBUG,
 182               ACE_TEXT (" (%t) Delete TP_Reactor\n")));
 183
 184   delete this->my_reactor_;
 185   ACE_Reactor::instance ((ACE_Reactor *) 0);
 186   this->my_reactor_ = 0;
 187   this->reactor (0);
 188
 189   return 0;
 190 }
 191
 192 int
 193 MyTask::start (int num_threads)
 194 {
 195   if (this->create_reactor () == -1)
 196     ACE_ERROR_RETURN ((LM_ERROR,
 197                        ACE_TEXT ("%p.\n"),
 198                        ACE_TEXT ("unable to create reactor")),
 199                       -1);
 200
 201   if (this->activate (THR_NEW_LWP, num_threads) == -1)
 202     ACE_ERROR_RETURN ((LM_ERROR,
 203                        ACE_TEXT ("%p.\n"),
 204                        ACE_TEXT ("unable to activate thread pool")),
 205                       -1);
 206
 207   for (; num_threads > 0 ; num_threads--)
 208     sem_.acquire ();
 209
 210   return 0;
 211 }
 212
 213
 214 int
 215 MyTask::stop ()
 216 {
 217   if (this->my_reactor_ != 0)
 218     {
 219       ACE_DEBUG ((LM_DEBUG,
 220                   ACE_TEXT ("End TP_Reactor event loop\n")));
 221
 222       ACE_Reactor::instance()->end_reactor_event_loop ();
 223     }
 224
 225   if (this->wait () == -1)
 226     ACE_ERROR ((LM_ERROR,
 227                 ACE_TEXT ("%p.\n"),
 228                 ACE_TEXT ("unable to stop thread pool")));
 229
 230   if (this->delete_reactor () == -1)
 231     ACE_ERROR ((LM_ERROR,
 232                 ACE_TEXT ("%p.\n"),
 233                 ACE_TEXT ("unable to delete reactor")));
 234
 235   return 0;
 236 }
 237
 238 int
 239 MyTask::svc ()
 240 {
 241   ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) MyTask started\n")));
 242
 243   disable_signal (SIGPIPE, SIGPIPE);
 244
 245   // signal that we are ready
 246   sem_.release (1);
 247
 248   while (ACE_Reactor::instance()->reactor_event_loop_done () == 0)
 249     ACE_Reactor::instance()->run_reactor_event_loop ();
 250
 251   ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) MyTask finished\n")));
 252   return 0;
 253 }
 254
 255 // *************************************************************
 256
 257 Acceptor::Acceptor ()
 258   : ACE_Acceptor<Receiver,ACE_SOCK_ACCEPTOR> ((ACE_Reactor *) 0),
 259     sessions_ (0),
 260     total_snd_(0),
 261     total_rcv_(0),
 262     total_w_  (0),
 263     total_r_  (0)
 264 {
 265   ACE_GUARD (ACE_Recursive_Thread_Mutex, locker, this->mutex_);
 266
 267   for (size_t i = 0; i < MAX_RECEIVERS; ++i)
 268      this->list_receivers_[i] =0;
 269 }
 270
 271 Acceptor::~Acceptor ()
 272 {
 273   this->reactor (0);
 274   stop ();
 275 }
 276
 277 void
 278 Acceptor::stop ()
 279 {
 280   // this method can be called only after reactor event loop id done
 281   // in all threads
 282
 283   ACE_GUARD (ACE_Recursive_Thread_Mutex, locker, this->mutex_);
 284
 285   for (size_t i = 0; i < MAX_RECEIVERS; ++i)
 286     {
 287       delete this->list_receivers_[i];
 288       this->list_receivers_[i] =0;
 289     }
 290 }
 291
 292 void
 293 Acceptor::on_new_receiver (Receiver &rcvr)
 294 {
 295   ACE_GUARD (ACE_Recursive_Thread_Mutex, locker, this->mutex_);
 296   this->sessions_++;
 297   this->list_receivers_[rcvr.index_] = & rcvr;
 298   ACE_DEBUG ((LM_DEBUG,
 299               "Receiver::CTOR sessions_=%d\n",
 300               this->sessions_));
 301 }
 302
 303 void
 304 Acceptor::on_delete_receiver (Receiver &rcvr)
 305 {
 306   ACE_GUARD (ACE_Recursive_Thread_Mutex, locker, this->mutex_);
 307
 308   this->sessions_--;
 309
 310   this->total_snd_ += rcvr.get_total_snd ();
 311   this->total_rcv_ += rcvr.get_total_rcv ();
 312   this->total_w_   += rcvr.get_total_w ();
 313   this->total_r_   += rcvr.get_total_r ();
 314
 315   if (rcvr.index_ < MAX_RECEIVERS
 316       && this->list_receivers_[rcvr.index_] == &rcvr)
 317     this->list_receivers_[rcvr.index_] = 0;
 318
 319   ACE_TCHAR bufs [256];
 320   ACE_TCHAR bufr [256];
 321
 322   ACE_OS::snprintf (bufs, 256, ACE_TEXT ("%ld(%ld)"),
 323                     rcvr.get_total_snd (),
 324                     rcvr.get_total_w ());
 325
 326   ACE_OS::snprintf (bufr, 256, ACE_TEXT ("%ld(%ld)"),
 327                     rcvr.get_total_rcv (),
 328                     rcvr.get_total_r ());
 329
 330   ACE_DEBUG ((LM_DEBUG,
 331               ACE_TEXT ("Receiver::~DTOR index=%d snd=%s rcv=%s sessions_=%d\n"),
 332               rcvr.index_,
 333               bufs,
 334               bufr,
 335               this->sessions_));
 336 }
 337
 338 int
 339 Acceptor::start (const ACE_INET_Addr &addr)
 340 {
 341   if (ACE_Acceptor<Receiver,ACE_SOCK_ACCEPTOR>::open (addr,
 342                                                       ACE_Reactor::instance (),
 343                                                       ACE_NONBLOCK) < 0)
 344     ACE_ERROR_RETURN ((LM_ERROR,
 345                        ACE_TEXT("%p\n"),
 346                        ACE_TEXT("Acceptor::start () - open failed")),
 347                        0);
 348   return 1;
 349 }
 350
 351 int
 352 Acceptor::make_svc_handler (Receiver *&sh)
 353 {
 354   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);
 355
 356   if (sessions_ >= MAX_RECEIVERS)
 357     return -1;
 358
 359   for (size_t i = 0; i < MAX_RECEIVERS; ++i)
 360     if (this->list_receivers_ [i] == 0)
 361       {
 362         ACE_NEW_RETURN (sh,
 363                         Receiver (this , i),
 364                         -1);
 365         return 0;
 366       }
 367   return -1;
 368 }
 369
 370 // *************************************************************
 371
 372 Receiver::Receiver (Acceptor * acceptor, size_t index)
 373   : acceptor_ (acceptor),
 374     index_ (index),
 375     flg_mask_ (ACE_Event_Handler::NULL_MASK),
 376     total_snd_(0),
 377     total_rcv_(0),
 378     total_w_  (0),
 379     total_r_  (0)
 380 {
 381   if (acceptor_ != 0)
 382     acceptor_->on_new_receiver (*this);
 383 }
 384
 385
 386 Receiver::~Receiver ()
 387 {
 388   this->reactor (0);
 389   if (acceptor_ != 0)
 390     acceptor_->on_delete_receiver (*this);
 391
 392   this->index_ = 0;
 393
 394   for (; ;)
 395     {
 396       ACE_Time_Value tv = ACE_Time_Value::zero;
 397       ACE_Message_Block *mb = 0;
 398
 399       if (this->getq (mb, &tv) < 0)
 400         break;
 401
 402       ACE_Message_Block::release (mb);
 403     }
 404 }
 405
 406 int
 407 Receiver::check_destroy ()
 408 {
 409   if (flg_mask_ == ACE_Event_Handler::NULL_MASK)
 410     return -1;
 411
 412   return 0;
 413 }
 414
 415 int
 416 Receiver::open (void *)
 417 {
 418   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);
 419
 420   ACE_Reactor *TPReactor = ACE_Reactor::instance ();
 421
 422   this->reactor (TPReactor);
 423
 424   flg_mask_ = ACE_Event_Handler::NULL_MASK ;
 425
 426   if (TPReactor->register_handler (this, flg_mask_) == -1)
 427     return -1;
 428
 429   initiate_io (ACE_Event_Handler::READ_MASK);
 430
 431   return check_destroy ();
 432 }
 433
 434 int
 435 Receiver::initiate_io (ACE_Reactor_Mask mask)
 436 {
 437   if (ACE_BIT_ENABLED (flg_mask_, mask))
 438     return 0;
 439
 440   if (ACE_Reactor::instance ()->schedule_wakeup  (this, mask) == -1)
 441     return -1;
 442
 443   ACE_SET_BITS (flg_mask_, mask);
 444   return 0;
 445 }
 446
 447 int
 448 Receiver::terminate_io (ACE_Reactor_Mask mask)
 449 {
 450   if (ACE_BIT_DISABLED (flg_mask_, mask))
 451     return 0;
 452
 453   if (ACE_Reactor::instance ()->cancel_wakeup (this, mask) == -1)
 454     return -1;
 455
 456   ACE_CLR_BITS (flg_mask_, mask);
 457   return 0;
 458 }
 459
 460 int
 461 Receiver::handle_close (ACE_HANDLE, ACE_Reactor_Mask)
 462 {
 463   ACE_Reactor *TPReactor = ACE_Reactor::instance ();
 464
 465   TPReactor->remove_handler (this,
 466                              ACE_Event_Handler::ALL_EVENTS_MASK |
 467                              ACE_Event_Handler::DONT_CALL);  // Don't call handle_close
 468   this->reactor (0);
 469   this->destroy ();
 470   return 0;
 471 }
 472
 473 int
 474 Receiver::handle_input (ACE_HANDLE h)
 475 {
 476   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);
 477
 478   ACE_Message_Block *mb = 0;
 479   ACE_NEW_RETURN (mb,
 480                   ACE_Message_Block (BUFSIZ),
 481                   -1);
 482
 483   int err = 0;
 484   ssize_t res = this->peer ().recv (mb->rd_ptr (), BUFSIZ-1);
 485
 486   this->total_r_++;
 487
 488   if (res >= 0)
 489     {
 490       mb->wr_ptr (res);
 491       this->total_rcv_ += res;
 492     }
 493   else
 494     err = errno ;
 495
 496   mb->wr_ptr ()[0] = '\0';
 497
 498   if (loglevel == 0 || res <= 0 || err!= 0)
 499     {
 500       LogLocker log_lock;
 501
 502       ACE_DEBUG ((LM_DEBUG, "**** Receiver::handle_input () SessionId=%d****\n", index_));
 503       ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "bytes_to_read", BUFSIZ));
 504       ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "handle", h));
 505       ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "bytes_transferred", res));
 506       ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "error", err));
 507       ACE_DEBUG ((LM_DEBUG, "%C = %s\n", "message_block", mb->rd_ptr ()));
 508       ACE_DEBUG ((LM_DEBUG, "**** end of message ****************\n"));
 509     }
 510
 511   if (err == EWOULDBLOCK)
 512     {
 513       err=0;
 514       res=0;
 515       return check_destroy ();
 516     }
 517
 518   if (err !=0  || res <= 0)
 519     {
 520       ACE_Message_Block::release (mb);
 521       return -1;
 522     }
 523
 524   ACE_Time_Value tv = ACE_Time_Value::zero;
 525
 526   int qcount = this->putq (mb, & tv);
 527
 528   if (qcount <= 0)  // failed to putq
 529     {
 530       ACE_Message_Block::release (mb);
 531       return -1 ;
 532     }
 533
 534   int rc = 0;
 535
 536   if (duplex == 0)      // half-duplex , stop read
 537     rc = this->terminate_io (ACE_Event_Handler::READ_MASK);
 538   else                     // full duplex
 539     {
 540       if (qcount >= 20 )   // flow control, stop read
 541         rc = this->terminate_io (ACE_Event_Handler::READ_MASK);
 542       else
 543         rc = this->initiate_io (ACE_Event_Handler::READ_MASK);
 544     }
 545
 546   if (rc == -1)
 547     return -1;
 548
 549   //initiate write
 550   if (this->initiate_io (ACE_Event_Handler::WRITE_MASK) != 0)
 551     return -1;
 552
 553   return check_destroy ();
 554 }
 555
 556 int
 557 Receiver::handle_output (ACE_HANDLE h)
 558 {
 559   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);
 560
 561   ACE_Time_Value tv = ACE_Time_Value::zero;
 562   ACE_Message_Block *mb = 0;
 563
 564   int     err = 0;
 565   ssize_t res = 0;
 566   size_t  bytes = 0;
 567
 568   int     qcount = this->getq (mb, &tv);
 569
 570   if (mb != 0)  // qcount >= 0)
 571     {
 572       bytes = mb->length ();
 573       res = this->peer ().send (mb->rd_ptr (), bytes);
 574
 575       this->total_w_++;
 576
 577       if (res < 0)
 578         err = errno ;
 579       else
 580         this->total_snd_ += res;
 581
 582
 583       if (loglevel == 0 || res <= 0 || err!= 0)
 584         {
 585           LogLocker log_lock;
 586
 587           ACE_DEBUG ((LM_DEBUG, "**** Receiver::handle_output () SessionId=%d****\n", index_));
 588           ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "bytes_to_write", bytes));
 589           ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "handle", h));
 590           ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "bytes_transferred", res));
 591           ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "error", err));
 592           ACE_DEBUG ((LM_DEBUG, "%C = %s\n", "message_block", mb->rd_ptr ()));
 593           ACE_DEBUG ((LM_DEBUG, "**** end of message ****************\n"));
 594         }
 595     }
 596
 597   ACE_Message_Block::release (mb);
 598
 599   if (err != 0  || res < 0)
 600     return -1;
 601
 602   if (qcount <= 0)  // no more message blocks in queue
 603     {
 604       if (this->terminate_io (ACE_Event_Handler::WRITE_MASK) != 0)
 605         return -1;
 606
 607       if (this->initiate_io (ACE_Event_Handler::READ_MASK) != 0)
 608         return -1;
 609     }
 610
 611   return check_destroy ();
 612 }
 613
 614 // *************************************************************
 615
 616 Connector::Connector ()
 617   : ACE_Connector<Sender,ACE_SOCK_CONNECTOR> ((ACE_Reactor *) 0),
 618     sessions_ (0),
 619     total_snd_(0),
 620     total_rcv_(0),
 621     total_w_  (0),
 622     total_r_  (0)
 623 {
 624   ACE_GUARD (ACE_Recursive_Thread_Mutex, locker, this->mutex_);
 625
 626   for (size_t i = 0; i < MAX_SENDERS; ++i)
 627      this->list_senders_[i] = 0;
 628 }
 629
 630 Connector::~Connector ()
 631 {
 632   this->reactor (0);
 633   stop ();
 634 }
 635
 636 void
 637 Connector::stop ()
 638 {
 639   // this method can be called only
 640   // after reactor event loop id done
 641   // in all threads
 642
 643   ACE_GUARD (ACE_Recursive_Thread_Mutex, locker, this->mutex_);
 644
 645   for (size_t i = 0; i < MAX_SENDERS; ++i)
 646     {
 647       delete this->list_senders_[i];
 648       this->list_senders_[i] =0;
 649     }
 650 }
 651
 652 void
 653 Connector::on_new_sender (Sender & sndr)
 654 {
 655   ACE_GUARD (ACE_Recursive_Thread_Mutex, locker, this->mutex_);
 656   this->sessions_++;
 657   this->list_senders_[sndr.index_] = &sndr;
 658   ACE_DEBUG ((LM_DEBUG,
 659               "Sender::CTOR sessions_=%d\n",
 660               this->sessions_));
 661 }
 662
 663 void
 664 Connector::on_delete_sender (Sender & sndr)
 665 {
 666   ACE_GUARD (ACE_Recursive_Thread_Mutex, locker, this->mutex_);
 667
 668   this->sessions_--;
 669   this->total_snd_ += sndr.get_total_snd();
 670   this->total_rcv_ += sndr.get_total_rcv();
 671   this->total_w_   += sndr.get_total_w();
 672   this->total_r_   += sndr.get_total_r();
 673
 674   if (sndr.index_ < MAX_SENDERS
 675       && this->list_senders_[sndr.index_] == &sndr)
 676     this->list_senders_[sndr.index_] = 0;
 677
 678   ACE_TCHAR bufs [256];
 679   ACE_TCHAR bufr [256];
 680
 681   ACE_OS::snprintf (bufs, 256, ACE_TEXT ("%ld(%ld)"),
 682                     sndr.get_total_snd (),
 683                     sndr.get_total_w ());
 684
 685   ACE_OS::snprintf (bufr, 256, ACE_TEXT ("%ld(%ld)"),
 686                     sndr.get_total_rcv (),
 687                     sndr.get_total_r ());
 688
 689   ACE_DEBUG ((LM_DEBUG,
 690               ACE_TEXT ("Sender::~DTOR index=%d snd=%s rcv=%s sessions_=%d\n"),
 691               sndr.index_,
 692               bufs,
 693               bufr,
 694               this->sessions_));
 695 }
 696
 697 int
 698 Connector::start (const ACE_INET_Addr & addr, int num)
 699 {
 700   if (ACE_Connector<Sender,ACE_SOCK_CONNECTOR>::open (ACE_Reactor::instance (),
 701                                                       ACE_NONBLOCK) < 0)
 702     ACE_ERROR_RETURN
 703       ((LM_ERROR,
 704         ACE_TEXT("%p\n"),
 705         ACE_TEXT("Connector::start () - open failed")),
 706        0);
 707
 708   int rc = 0;
 709
 710   for (int i = 0 ; i < num ; i++)
 711     {
 712       Sender * sender = 0;
 713
 714       if (ACE_Connector<Sender,ACE_SOCK_CONNECTOR>::connect (sender, addr) < 0)
 715         ACE_ERROR_RETURN
 716           ((LM_ERROR,
 717             ACE_TEXT("%p\n"),
 718             ACE_TEXT("Connector::start () - connect failed")),
 719            rc);
 720     }
 721
 722   return rc;
 723 }
 724
 725 int
 726 Connector::make_svc_handler (Sender * & sh)
 727 {
 728   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);
 729
 730   if (sessions_ >= MAX_SENDERS)
 731     return -1;
 732
 733   for (size_t i = 0; i < MAX_SENDERS; ++i)
 734     if (this->list_senders_ [i] == 0)
 735       {
 736         ACE_NEW_RETURN (sh,
 737                         Sender (this , i),
 738                         -1);
 739         return 0;
 740       }
 741
 742   return -1;
 743 }
 744
 745 // *************************************************************
 746
 747 Sender::Sender (Connector* connector, size_t index)
 748   : connector_ (connector),
 749     index_ (index),
 750     flg_mask_ (ACE_Event_Handler::NULL_MASK),
 751     total_snd_(0),
 752     total_rcv_(0),
 753     total_w_  (0),
 754     total_r_  (0)
 755 {
 756   if (connector_ != 0)
 757     connector_->on_new_sender (*this);
 758
 759   ACE_OS::snprintf (send_buf_, 1024, "%s", data);
 760 }
 761
 762
 763 Sender::~Sender ()
 764 {
 765   this->reactor (0);
 766   if (connector_ != 0)
 767     connector_->on_delete_sender (*this);
 768
 769   this->index_ = 0;
 770
 771   for (; ;)
 772     {
 773       ACE_Time_Value tv = ACE_Time_Value::zero;
 774       ACE_Message_Block *mb = 0;
 775
 776       if (this->getq (mb, &tv) < 0)
 777         break;
 778
 779       ACE_Message_Block::release (mb);
 780     }
 781 }
 782
 783 int
 784 Sender::check_destroy ()
 785 {
 786   if (flg_mask_ == ACE_Event_Handler::NULL_MASK)
 787     return -1;
 788
 789   return 0;
 790 }
 791
 792 int Sender::open (void *)
 793 {
 794   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);
 795
 796   ACE_Reactor * TPReactor = ACE_Reactor::instance ();
 797
 798   this->reactor (TPReactor);
 799
 800   flg_mask_ = ACE_Event_Handler::NULL_MASK ;
 801
 802   if (TPReactor->register_handler (this,flg_mask_) == -1)
 803     return -1;
 804
 805   if (this->initiate_write () == -1)
 806     return -1;
 807
 808   if (duplex != 0)
 809     initiate_io (ACE_Event_Handler::READ_MASK);
 810
 811   return check_destroy ();
 812 }
 813
 814 int
 815 Sender::initiate_write ()
 816 {
 817   if ( this->msg_queue ()->message_count () < 20) // flow control
 818     {
 819       size_t nbytes = ACE_OS::strlen (send_buf_);
 820
 821       ACE_Message_Block *mb = 0;
 822       ACE_NEW_RETURN (mb,
 823                       ACE_Message_Block (nbytes+8),
 824                       -1);
 825
 826       mb->init (send_buf_, nbytes);
 827       mb->rd_ptr (mb->base ());
 828       mb->wr_ptr (mb->base ());
 829       mb->wr_ptr (nbytes);
 830
 831       ACE_Time_Value tv = ACE_Time_Value::zero;
 832
 833       int qcount =this->putq (mb, & tv);
 834
 835       if (qcount <= 0)
 836         {
 837           ACE_Message_Block::release (mb);
 838           return -1;
 839         }
 840     }
 841
 842   return initiate_io (ACE_Event_Handler::WRITE_MASK);
 843 }
 844
 845 int
 846 Sender::initiate_io (ACE_Reactor_Mask mask)
 847 {
 848   if (ACE_BIT_ENABLED (flg_mask_, mask))
 849     return 0;
 850
 851   if (ACE_Reactor::instance ()->schedule_wakeup (this, mask) == -1)
 852     return -1;
 853
 854   ACE_SET_BITS (flg_mask_, mask);
 855   return 0;
 856 }
 857
 858 int
 859 Sender::terminate_io (ACE_Reactor_Mask mask)
 860 {
 861   if (ACE_BIT_DISABLED (flg_mask_, mask))
 862     return 0;
 863
 864   if (ACE_Reactor::instance ()->cancel_wakeup (this, mask) == -1)
 865     return -1;
 866
 867   ACE_CLR_BITS (flg_mask_, mask);
 868   return 0;
 869 }
 870
 871 int
 872 Sender::handle_close (ACE_HANDLE, ACE_Reactor_Mask)
 873 {
 874   ACE_Reactor * TPReactor = ACE_Reactor::instance ();
 875
 876   TPReactor->remove_handler (this,
 877                              ACE_Event_Handler::ALL_EVENTS_MASK |
 878                              ACE_Event_Handler::DONT_CALL);  // Don't call handle_close
 879   this->reactor (0);
 880   this->destroy ();
 881   return 0;
 882 }
 883
 884 int
 885 Sender::handle_input (ACE_HANDLE h)
 886 {
 887   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);
 888
 889   ACE_Message_Block *mb = 0;
 890   ACE_NEW_RETURN (mb,
 891                   ACE_Message_Block (BUFSIZ),
 892                   -1);
 893
 894   int     err = 0;
 895   ssize_t res = this->peer ().recv (mb->rd_ptr (),
 896                                     BUFSIZ-1);
 897   this->total_r_++;
 898
 899   if (res >= 0)
 900     {
 901       mb->wr_ptr (res);
 902       this->total_rcv_ += res;
 903     }
 904   else
 905     err = errno ;
 906
 907   mb->wr_ptr ()[0] = '\0';
 908
 909   if (loglevel == 0 || res <= 0 || err!= 0)
 910     {
 911       LogLocker log_lock;
 912
 913       ACE_DEBUG ((LM_DEBUG, "**** Sender::handle_input () SessionId=%d****\n", index_));
 914       ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "bytes_to_read", BUFSIZ));
 915       ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "handle", h));
 916       ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "bytes_transferred", res));
 917       ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "error", err));
 918       ACE_DEBUG ((LM_DEBUG, "%C = %s\n", "message_block", mb->rd_ptr ()));
 919       ACE_DEBUG ((LM_DEBUG, "**** end of message ****************\n"));
 920     }
 921
 922   ACE_Message_Block::release (mb);
 923
 924   if (err == EWOULDBLOCK)
 925     {
 926       err=0;
 927       res=0;
 928       return check_destroy ();
 929     }
 930
 931   if (err !=0  || res <= 0)
 932     return -1;
 933
 934   int rc = 0;
 935
 936   if (duplex != 0)  // full duplex, continue read
 937     rc = initiate_io (ACE_Event_Handler::READ_MASK);
 938   else
 939     rc = terminate_io (ACE_Event_Handler::READ_MASK);
 940
 941   if (rc != 0)
 942     return -1 ;
 943
 944   rc = initiate_write ();
 945   if (rc != 0)
 946     return -1;
 947
 948   return check_destroy ();
 949 }
 950
 951 int
 952 Sender::handle_output (ACE_HANDLE h)
 953 {
 954   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);
 955
 956   ACE_Time_Value tv = ACE_Time_Value::zero;
 957   ACE_Message_Block *mb = 0;
 958
 959   int     err=0;
 960   ssize_t res=0;
 961   size_t  bytes=0;
 962
 963   int     qcount = this->getq (mb , & tv);
 964
 965   if (mb != 0)  // qcount >= 0
 966     {
 967       bytes = mb->length ();
 968       res = this->peer ().send (mb->rd_ptr (), bytes);
 969
 970       this->total_w_++;
 971
 972       if (res < 0)
 973         err = errno ;
 974       else
 975         this->total_snd_ += res;
 976
 977       if (loglevel == 0 || res <= 0 || err!= 0)
 978         {
 979           LogLocker log_lock;
 980
 981           ACE_DEBUG ((LM_DEBUG, "**** Sender::handle_output () SessionId=%d****\n", index_));
 982           ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "bytes_to_write", bytes));
 983           ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "handle", h));
 984           ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "bytes_transferred", res));
 985           ACE_DEBUG ((LM_DEBUG, "%C = %d\n", "error", err));
 986           ACE_DEBUG ((LM_DEBUG, "%C = %s\n", "message_block", mb->rd_ptr ()));
 987           ACE_DEBUG ((LM_DEBUG, "**** end of message ****************\n"));
 988         }
 989     }
 990
 991   ACE_Message_Block::release (mb);
 992
 993   if (err != 0  || res < 0)
 994     return -1;
 995
 996   int rc = 0;
 997
 998   if (qcount <= 0)  // no more message blocks in queue
 999     {
1000       if (duplex != 0 &&   // full duplex, continue write
1001           (this->total_snd_ - this->total_rcv_ ) < 1024*32 )  // flow control
1002         rc = initiate_write ();
1003       else
1004         rc = terminate_io (ACE_Event_Handler::WRITE_MASK);
1005
1006       if (rc == -1)
1007         return -1;
1008     }
1009
1010   rc = initiate_io (ACE_Event_Handler::READ_MASK);
1011   if (rc == -1)
1012     return -1;
1013
1014   return check_destroy ();
1015 }
1016
1017
1018 // *************************************************************
1019 //   Configuration helpers
1020 // *************************************************************
1021 int
1022 print_usage (int /* argc */, ACE_TCHAR *argv[])
1023 {
1024   ACE_ERROR
1025     ((LM_ERROR,
1026       ACE_TEXT ("\nusage: %s")
1027       ACE_TEXT ("\n-n <number threads in the thread pool>")
1028       ACE_TEXT ("\n-d <duplex mode 1-on/0-off>")
1029       ACE_TEXT ("\n-p <port to listen/connect>")
1030       ACE_TEXT ("\n-h <host> for Sender mode")
1031       ACE_TEXT ("\n-s <number of sender's instances>")
1032       ACE_TEXT ("\n-b run client and server at the same time")
1033       ACE_TEXT ("\n-v log level")
1034       ACE_TEXT ("\n    0 - log all messages")
1035       ACE_TEXT ("\n    1 - log only errors and unusual cases")
1036       ACE_TEXT ("\n-i time to run in seconds")
1037       ACE_TEXT ("\n-u show this message")
1038       ACE_TEXT ("\n"),
1039       argv[0]
1040       ));
1041   return -1;
1042 }
1043
1044 int
1045 parse_args (int argc, ACE_TCHAR *argv[])
1046 {
1047   if (argc == 1) // no arguments , so one button test
1048     {
1049       both = 1;               // client and server simultaneosly
1050       duplex = 1;             // full duplex is on
1051       host = ACE_LOCALHOST;   // server to connect
1052       port = ACE_DEFAULT_SERVER_PORT; // port to connect/listen
1053       threads = 3;            // size of Proactor thread pool
1054       senders = 20;           // number of senders
1055       loglevel = 1;           // log level : 0 full/ 1 only errors
1056       seconds = 20;           // time to run in seconds
1057 #if defined(SOMAXCONN) // The test is invalid if senders > SOMAXCONN
1058       if(SOMAXCONN < senders)
1059         senders = SOMAXCONN;
1060 #endif
1061       return 0;
1062     }
1063
1064   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT ("i:n:p:d:h:s:v:ub"));
1065   int c;
1066
1067   while ((c = get_opt ()) != EOF)
1068     {
1069       switch (c)
1070         {
1071         case 'i':  // time to run
1072           seconds = ACE_OS::atoi (get_opt.opt_arg());
1073           if (seconds < MIN_TIME)
1074             seconds = MIN_TIME;
1075           if (seconds > MAX_TIME)
1076             seconds = MAX_TIME;
1077           break;
1078         case 'b':  // both client and server
1079           both = 1;
1080           break;
1081         case 'v':  // log level
1082           loglevel = ACE_OS::atoi (get_opt.opt_arg());
1083           break;
1084         case 'd':         // duplex
1085           duplex = ACE_OS::atoi (get_opt.opt_arg());
1086           break;
1087         case 'h':         // host for sender
1088           host = get_opt.opt_arg();
1089           break;
1090         case 'p':         // port number
1091           port = ACE_OS::atoi (get_opt.opt_arg());
1092           break;
1093         case 'n':         // thread pool size
1094           threads = ACE_OS::atoi (get_opt.opt_arg());
1095           break;
1096         case 's':     // number of senders
1097           senders = ACE_OS::atoi (get_opt.opt_arg());
1098           if (size_t (senders) > MAX_SENDERS)
1099             senders = MAX_SENDERS;
1100           break;
1101         case 'u':
1102         default:
1103           return print_usage (argc,argv);
1104         } // switch
1105     } // while
1106
1107   return 0;
1108 }
1109
1110 static int
1111 disable_signal (int sigmin, int sigmax)
1112 {
1113 #if !defined (ACE_LACKS_UNIX_SIGNALS)
1114   sigset_t signal_set;
1115   if (ACE_OS::sigemptyset (&signal_set) == - 1)
1116     ACE_ERROR ((LM_ERROR,
1117                 ACE_TEXT("Error: (%P | %t):%p\n"),
1118                 ACE_TEXT("sigemptyset failed")));
1119
1120   for (int i = sigmin; i <= sigmax; i++)
1121     ACE_OS::sigaddset (&signal_set, i);
1122
1123   // Put the <signal_set>.
1124 # if defined (ACE_LACKS_PTHREAD_THR_SIGSETMASK)
1125   // In multi-threaded application this is not POSIX compliant
1126   // but let's leave it just in case.
1127   if (ACE_OS::sigprocmask (SIG_BLOCK, &signal_set, 0) != 0)
1128 # else
1129   if (ACE_OS::thr_sigsetmask (SIG_BLOCK, &signal_set, 0) != 0)
1130 # endif /* ACE_LACKS_PTHREAD_THR_SIGSETMASK */
1131     ACE_ERROR_RETURN ((LM_ERROR,
1132                        ACE_TEXT ("Error: (%P|%t): %p\n"),
1133                        ACE_TEXT ("SIG_BLOCK failed")),
1134                       -1);
1135 #else
1136   ACE_UNUSED_ARG(sigmin);
1137   ACE_UNUSED_ARG(sigmax);
1138 #endif /* ACE_LACKS_UNIX_SIGNALS */
1139
1140   return 0;
1141 }
1142
1143 #endif /* ACE_HAS_THREADS */
1144
1145 int
1146 run_main (int argc, ACE_TCHAR *argv[])
1147 {
1148   ACE_START_TEST (ACE_TEXT ("TP_Reactor_Test"));
1149
1150 #if defined(ACE_HAS_THREADS) && !defined ACE_LACKS_ACCEPT
1151   if (::parse_args (argc, argv) == -1)
1152     return -1;
1153
1154   disable_signal (SIGPIPE, SIGPIPE);
1155
1156   MyTask    task1;
1157   Acceptor  acceptor;
1158   Connector connector;
1159
1160   if (task1.start (threads) == 0)
1161     {
1162       int rc = 0;
1163
1164       ACE_INET_Addr addr (port);
1165       if (both != 0 || host == 0) // Acceptor
1166         rc += acceptor.start (addr);
1167
1168       if (both != 0 || host != 0)
1169         {
1170           if (host == 0)
1171             host = ACE_LOCALHOST;
1172
1173           if (addr.set (port, host, 1, addr.get_type ()) == -1)
1174             ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), host));
1175           rc += connector.start (addr, senders);
1176         }
1177
1178       if (rc > 0)
1179         ACE_OS::sleep (seconds);
1180     }
1181
1182   task1.stop ();
1183
1184   ACE_DEBUG ((LM_DEBUG,
1185               ACE_TEXT ("\nNumber of Receivers objects = %d\n")
1186               ACE_TEXT ("\nNumber of Sender objects = %d\n"),
1187               acceptor.get_number_sessions (),
1188               connector.get_number_sessions ()));
1189
1190   // As Reactor event loop now is inactive it is safe to destroy all
1191   // senders
1192
1193   connector.stop ();
1194   acceptor.stop ();
1195
1196   //Print statistic
1197   ACE_TCHAR bufs [256];
1198   ACE_TCHAR bufr [256];
1199
1200   ACE_OS::snprintf (bufs, 256, ACE_TEXT ("%ld(%ld)"),
1201                     connector.get_total_snd (),
1202                     connector.get_total_w ());
1203
1204   ACE_OS::snprintf (bufr, 256, ACE_TEXT ("%ld(%ld)"),
1205                     connector.get_total_rcv (),
1206                     connector.get_total_r ());
1207
1208   ACE_DEBUG ((LM_DEBUG,
1209               ACE_TEXT ("Connector/Senders  total bytes: snd=%s rcv=%s\n"),
1210               bufs,
1211               bufr
1212             ));
1213
1214   ACE_OS::snprintf (bufs, 256, ACE_TEXT ("%ld(%ld)"),
1215                     acceptor.get_total_snd (),
1216                     acceptor.get_total_w ());
1217
1218   ACE_OS::snprintf (bufr, 256, ACE_TEXT ("%ld(%ld)"),
1219                     acceptor.get_total_rcv (),
1220                     acceptor.get_total_r ());
1221
1222   ACE_DEBUG ((LM_DEBUG,
1223               ACE_TEXT ("Acceptor/Receivers total bytes: snd=%s rcv=%s\n"),
1224               bufs,
1225               bufr
1226             ));
1227
1228 #else /* ACE_HAS_THREADS */
1229    ACE_UNUSED_ARG( argc );
1230    ACE_UNUSED_ARG( argv );
1231 #endif /* ACE_HAS_THREADS */
1232
1233   ACE_END_TEST;
1234
1235   return 0;
1236 }