ACE/tests/Bug_2740_Regression_Test.cpp

   1 /**
   2  * @file Bug_2740_Regression_Test.cpp
   3  *
   4  * Reproduces the problems reported in bug 2740
   5  *   http://bugzilla.dre.vanderbilt.edu/show_bug.cgi?id=2740
   6  *
   7  * @author paolo.volpi@tvblob.com
   8  */
   9
  10 #include "test_config.h"
  11
  12 #if defined (ACE_HAS_DEV_POLL) || defined (ACE_HAS_EVENT_POLL)
  13
  14 #include "ace/Atomic_Op.h"
  15 #include "ace/Reactor.h"
  16 #include "ace/Dev_Poll_Reactor.h"
  17 #include "ace/Connector.h"
  18 #include "ace/SOCK_Connector.h"
  19 #include "ace/Acceptor.h"
  20 #include "ace/SOCK_Acceptor.h"
  21 #include "ace/Svc_Handler.h"
  22 #include "ace/Signal.h"
  23 #include "ace/Time_Value.h"
  24 #include "ace/OS_NS_string.h"
  25
  26 const ACE_Time_Value MAX_CLIENT_TIMEOUT(30);  // 30 secs
  27
  28 // The number of clients we'll run. When the ClientsLeft gets back to 0 (it's
  29 // decremented when a ClientSvcHandler is deleted) the event loop can stop.
  30 const long NrClients = 10;
  31 static ACE_Atomic_Op<ACE_Thread_Mutex, long> ClientsLeft (10);
  32
  33
  34 class ClientSvcHandler : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>
  35 {
  36 public:
  37   // Default ctor needed to satisfy ACE_Connector template, but not used.
  38   ClientSvcHandler ()
  39   { ACE_ERROR((LM_ERROR, ACE_TEXT ("%t: Wrong ClientSvcHandler ctor!!\n"))); }
  40
  41   // Each client will send/recv 'echo_cnt' times, close/reopen the socket,
  42   // then echo, etc. for ACE_MAX_ITERATIONS times.
  43   ClientSvcHandler (const ACE_INET_Addr &addr, int echo_cnt);
  44   ~ClientSvcHandler () override;
  45
  46   int open (void* factory) override;
  47   int handle_input (ACE_HANDLE handle = ACE_INVALID_HANDLE) override;
  48   int handle_timeout (const ACE_Time_Value &now, const void *act = 0) override;
  49   int handle_close (ACE_HANDLE handle = ACE_INVALID_HANDLE,
  50                     ACE_Reactor_Mask mask = 0) override;
  51
  52 protected:
  53   static const char *send_str;
  54
  55   ACE_INET_Addr server_addr_;
  56   // Know when we've received 'echo_cnt' echoes of the data. Don't want to
  57   // get into record-marking, short-reads/writes, etc. so just count the number
  58   // of bytes it takes to echo the data. Don't care about verifying the
  59   // content - this is mostly a stress test for multithreaded reactor.
  60   size_t echo_cnt_;
  61   size_t iter_;
  62   unsigned long timer_;
  63 };
  64
  65 const char *ClientSvcHandler::send_str =
  66   "1234567890123456789012345678901234567890";
  67
  68 class ServerSvcHandler : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>
  69 {
  70 public:
  71   int open (void* factory) override;
  72   int handle_input (ACE_HANDLE handle = ACE_INVALID_HANDLE) override;
  73   int handle_close (ACE_HANDLE handle = ACE_INVALID_HANDLE,
  74                     ACE_Reactor_Mask mask = 0) override;
  75 };
  76
  77
  78 ClientSvcHandler::ClientSvcHandler (const ACE_INET_Addr &addr, int echo_cnt)
  79   : ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>(),
  80     server_addr_ (addr),
  81     echo_cnt_ (ACE_OS::strlen (ClientSvcHandler::send_str) * echo_cnt),
  82     iter_ (0)
  83 {
  84 }
  85
  86 ClientSvcHandler::~ClientSvcHandler ()
  87 {
  88   if (--ClientsLeft == 0)
  89     {
  90       ACE_DEBUG ((LM_DEBUG,
  91                   ACE_TEXT ("%t: No more clients; ending reactor loop\n")));
  92       this->reactor ()->end_reactor_event_loop ();
  93     }
  94 }
  95
  96 int
  97 ClientSvcHandler::open (void* factory)
  98 {
  99   ++iter_;
 100   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: client h %d, open for iteration %B\n"),
 101               this->peer ().get_handle (), this->iter_));
 102
 103   if (ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::open (factory) == 0)
 104     {
 105       this->timer_ = this->reactor ()->schedule_timer (this,
 106                                                        0,
 107                                                        MAX_CLIENT_TIMEOUT);
 108
 109       size_t send_len = ACE_OS::strlen (ClientSvcHandler::send_str);
 110       if (-1 == this->peer().send_n (ClientSvcHandler::send_str, send_len))
 111         ACE_ERROR_RETURN ((LM_ERROR,
 112                            ACE_TEXT ("%t: client h %d, %p\n"),
 113                            this->peer ().get_handle(),
 114                            ACE_TEXT ("send")),
 115                           -1);
 116       return 0;
 117     }
 118
 119   ACE_ERROR_RETURN ((LM_ERROR,
 120                      ACE_TEXT ("%t: %p\n"),
 121                      ACE_TEXT ("Client open")),
 122                     -1);
 123 }
 124
 125 int
 126 ClientSvcHandler::handle_input (ACE_HANDLE handle)
 127 {
 128     // Get socket data
 129     char buffer[ACE_DEFAULT_MAX_SOCKET_BUFSIZ];
 130     ssize_t bc;
 131     bc = this->peer ().recv (buffer, sizeof (buffer));
 132     ACE_DEBUG ((LM_DEBUG,
 133                 ACE_TEXT ("%t: client h %d recv %b bytes\n"),
 134                 handle,
 135                 bc));
 136     if (bc > 0)
 137       {
 138         // Reset Client timeout timer
 139         this->reactor ()->cancel_timer (this->timer_);
 140
 141         // Reduce the echo count by what was received; if not there, send
 142         // more data and await more echo.
 143         size_t ubc = (size_t)bc;
 144         if (ubc <= this->echo_cnt_)
 145           this->echo_cnt_ -= ubc;
 146         if (this->echo_cnt_ == 0)
 147           {
 148             ACE_DEBUG ((LM_DEBUG,
 149                         ACE_TEXT ("%t: client h %d end of echo\n"),
 150                         handle));
 151             return -1;
 152           }
 153
 154         // Not done yet; send more data and resched timer.
 155         size_t send_len = ACE_OS::strlen (ClientSvcHandler::send_str);
 156         if (-1 == this->peer().send_n (ClientSvcHandler::send_str, send_len))
 157           ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%t: client h %d: %p\n"),
 158                              ACE_TEXT ("resending")),
 159                             -1);
 160         this->timer_ = reactor ()->schedule_timer (this, 0, MAX_CLIENT_TIMEOUT);
 161       }
 162     else if (bc == 0) // Socket was closed by server
 163       {
 164         ACE_ERROR_RETURN ((LM_ERROR,
 165                            ACE_TEXT ("%t: client h %d: closed by server\n"),
 166                            handle),
 167                           -1);
 168       }
 169     else if (errno == EWOULDBLOCK) // no data ready on socket
 170       {
 171         ACE_ERROR_RETURN ((LM_DEBUG,
 172                            ACE_TEXT ("%t: client h %d: recv no data\n"),
 173                            handle),
 174                           0);
 175       }
 176     else
 177       ACE_ERROR_RETURN ((LM_ERROR,
 178                          ACE_TEXT ("%t: client h %d: %p\n"),
 179                          handle,
 180                          ACE_TEXT ("send")),
 181                         -1);
 182
 183     return 0;
 184 }
 185
 186 int
 187 ClientSvcHandler::handle_timeout (const ACE_Time_Value &, const void*)
 188 {
 189   ACE_DEBUG ((LM_DEBUG,
 190               ACE_TEXT ("%t: client h %d: timeout\n"),
 191               this->peer ().get_handle ()));
 192   reactor()->remove_handler (this, ACE_Event_Handler::ALL_EVENTS_MASK);
 193   return 0;
 194 }
 195
 196 int
 197 ClientSvcHandler::handle_close (ACE_HANDLE handle, ACE_Reactor_Mask mask)
 198 {
 199   ACE_DEBUG ((LM_DEBUG,
 200               ACE_TEXT ("%t: client h %d handle_close\n"),
 201               handle));
 202
 203   // If not done iterating, just close the socket and reopen the connection.
 204   // Else shut down and delete.
 205   if (this->iter_ >= ACE_MAX_ITERATIONS)
 206     return ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::handle_close(handle,
 207                                                                           mask);
 208   this->shutdown ();
 209   ClientSvcHandler *p = const_cast<ClientSvcHandler *>(this);
 210   ACE_Connector<ClientSvcHandler, ACE_SOCK_CONNECTOR>
 211     connector (this->reactor (), ACE_NONBLOCK);
 212   connector.connect (p, this->server_addr_);
 213   return 0;
 214 }
 215
 216 int
 217 ServerSvcHandler::open (void* factory)
 218 {
 219   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: server h %d open\n"),
 220               this->peer ().get_handle ()));
 221
 222   return ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::open (factory);
 223 }
 224
 225 int
 226 ServerSvcHandler::handle_input (ACE_HANDLE handle)
 227 {
 228   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: server h %d input\n"), handle));
 229
 230   // Receive whatever is here and send it back.
 231   char buffer[ACE_DEFAULT_MAX_SOCKET_BUFSIZ];
 232   ssize_t bc;
 233   bc = this->peer ().recv (buffer, sizeof (buffer));
 234   if (bc > 0)
 235     {
 236       ACE_DEBUG ((LM_DEBUG,
 237                   ACE_TEXT ("%t: server h %d recv %b bytes\n"),
 238                   handle,
 239                   bc));
 240       if (-1 == this->peer ().send_n (buffer, bc))
 241         ACE_ERROR_RETURN ((LM_ERROR,
 242                            ACE_TEXT ("%t: server h %d: %p\n"),
 243                            handle, ACE_TEXT ("send")),
 244                           -1);
 245     }
 246   else if (bc == 0 || errno == ECONNRESET) // Socket was closed by client
 247     {
 248       ACE_ERROR_RETURN ((LM_DEBUG,
 249                          ACE_TEXT ("%t: server h %d: closed by client\n"),
 250                          handle),
 251                         -1);
 252     }
 253   else if (errno == EWOULDBLOCK) // no data ready on socket
 254     {
 255       // The reactor shouldn't call back for input if there's no data, so
 256       // log an error, but keep running.
 257       ACE_ERROR_RETURN ((LM_ERROR,
 258                          ACE_TEXT ("%t: server h %d: recv no data\n"),
 259                          handle),
 260                         0);
 261     }
 262   else
 263     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%t: server h %d: %p\n"),
 264                        handle,
 265                        ACE_TEXT ("recv")),
 266                       -1);
 267
 268   return 0;
 269 }
 270
 271 int
 272 ServerSvcHandler::handle_close(ACE_HANDLE handle, ACE_Reactor_Mask mask)
 273 {
 274   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: server h %d: handle_close\n"), handle));
 275   return ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::handle_close (handle,
 276                                                                          mask);
 277 }
 278
 279 static int
 280 disable_signal (int sigmin, int sigmax)
 281 {
 282 #if !defined (ACE_LACKS_UNIX_SIGNALS)
 283   sigset_t signal_set;
 284   if (ACE_OS::sigemptyset (&signal_set) == - 1)
 285     ACE_ERROR ((LM_ERROR,
 286                 ACE_TEXT ("Error: (%P|%t):%p\n"),
 287                 ACE_TEXT ("sigemptyset failed")));
 288
 289   for (int i = sigmin; i <= sigmax; i++)
 290     ACE_OS::sigaddset (&signal_set, i);
 291
 292   // Put the <signal_set>.
 293 # if defined (ACE_LACKS_PTHREAD_THR_SIGSETMASK)
 294   // In multi-threaded application this is not POSIX compliant
 295   // but let's leave it just in case.
 296   if (ACE_OS::sigprocmask (SIG_BLOCK, &signal_set, 0) != 0)
 297 # else
 298   if (ACE_OS::thr_sigsetmask (SIG_BLOCK, &signal_set, 0) != 0)
 299 # endif /* ACE_LACKS_PTHREAD_THR_SIGSETMASK */
 300     ACE_ERROR_RETURN ((LM_ERROR,
 301                        ACE_TEXT ("Error: (%P|%t): %p\n"),
 302                        ACE_TEXT ("SIG_BLOCK failed")),
 303                       -1);
 304 #else
 305   ACE_UNUSED_ARG (sigmin);
 306   ACE_UNUSED_ARG (sigmax);
 307 #endif /* ACE_LACKS_UNIX_SIGNALS */
 308
 309   return 0;
 310 }
 311
 312 ACE_THR_FUNC_RETURN
 313 event_loop(void *arg)
 314 {
 315   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: Starting reactor event loop\n")));
 316
 317   disable_signal (SIGPIPE, SIGPIPE);
 318
 319   ACE_Reactor *reactor = static_cast<ACE_Reactor*>(arg);
 320   int s = reactor->run_reactor_event_loop();
 321
 322   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%t: reactor loop done; status %d\n"), s));
 323   return 0;
 324 }
 325
 326
 327 int run_main(int, ACE_TCHAR *[])
 328 {
 329   ACE_START_TEST (ACE_TEXT ("Bug_2740_Regression_Test"));
 330
 331   // Make sure we ignore SIGPIPE
 332   disable_signal (SIGPIPE, SIGPIPE);
 333
 334   ACE_Dev_Poll_Reactor dp_reactor;
 335   dp_reactor.restart (1);          // Restart on EINTR
 336   ACE_Reactor reactor (&dp_reactor);
 337
 338   // Bind acceptor to any port and then find out what the port was.
 339   ACE_Acceptor<ServerSvcHandler, ACE_SOCK_Acceptor> acceptor (&reactor,
 340                                                               ACE_NONBLOCK);
 341   ACE_SOCK_Acceptor::PEER_ADDR server_addr;
 342   if (acceptor.open (ACE_sap_any_cast (const ACE_INET_Addr &), &reactor) == -1
 343       || acceptor.acceptor ().get_local_addr (server_addr) == -1)
 344     {
 345       ACE_ERROR_RETURN ((LM_ERROR,
 346                          ACE_TEXT ("(%P|%t) %p\n"),
 347                          ACE_TEXT ("acceptor open")),
 348                         1);
 349     }
 350
 351   ACE_DEBUG ((LM_DEBUG,
 352               ACE_TEXT ("(%P|%t) starting server at port %d\n"),
 353               server_addr.get_port_number ()));
 354
 355   ACE_Thread_Manager::instance ()->spawn_n (ACE_MAX_THREADS,
 356                                             event_loop,
 357                                             &reactor);
 358
 359   ACE_Connector<ClientSvcHandler, ACE_SOCK_CONNECTOR> connector (&reactor,
 360                                                                  ACE_NONBLOCK);
 361   // Spin up clients to connect and iterate
 362   long i;
 363   for (i = 0; i < NrClients; ++i)
 364     {
 365       ClientSvcHandler *c = new ClientSvcHandler (server_addr, i % 4);
 366       connector.connect(c, server_addr);
 367     }
 368
 369   ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%d clients connected\n"), (int)i));
 370   ACE_Thread_Manager::instance ()->wait ();
 371   ACE_END_TEST;
 372   return 0;
 373 }
 374
 375 #else
 376
 377 int
 378 run_main (int, ACE_TCHAR *[])
 379 {
 380   ACE_START_TEST (ACE_TEXT ("Bug_2740_Regression_Test"));
 381   ACE_ERROR ((LM_INFO,
 382               ACE_TEXT ("Dev Poll and Event Poll are not supported ")
 383               ACE_TEXT ("on this platform\n")));
 384   ACE_END_TEST;
 385   return 0;
 386 }
 387
 388 #endif  /* ACE_HAS_DEV_POLL || ACE_HAS_EVENT_POLL */