ACE/examples/IPC_SAP/SOCK_SAP/CPP-inclient.cpp

   1 // This tests the features of the <ACE_SOCK_Connector> and
   2 // <ACE_SOCK_Stream> classes.  In addition, it can be used to test the
   3 // oneway and twoway latency and throughput at the socket-level.  This
   4 // is useful as a baseline to compare against ORB-level performance
   5 // for the same types of data.
   6
   7 #include "CPP-inclient.h"
   8
   9 #include "ace/SOCK_Connector.h"
  10 #include "ace/INET_Addr.h"
  11 #include "ace/Thread_Manager.h"
  12 #include "ace/Singleton.h"
  13 #include "ace/Get_Opt.h"
  14 #include "ace/Truncate.h"
  15 #include "ace/High_Res_Timer.h"
  16 #include "ace/Basic_Types.h"
  17 #include "ace/OS_NS_string.h"
  18 #include "ace/OS_NS_unistd.h"
  19 #include "ace/OS_main.h"
  20
  21 Options::Options ()
  22   : host_ (ACE_DEFAULT_SERVER_HOST),
  23     port_ (ACE_DEFAULT_SERVER_PORT),
  24     sleep_time_ (0, 0), // By default, don't sleep between calls.
  25     threads_ (10),
  26     quit_string_ (ACE_TEXT("q")),
  27     message_len_ (0),
  28     message_buf_ (0),
  29     io_source_ (ACE_INVALID_HANDLE), // Defaults to using the generator.
  30     iterations_ (10000),
  31     oneway_ (1) // Make oneway calls the default.
  32 #if defined (ACE_MT_SAFE) && (ACE_MT_SAFE != 0)
  33     , barrier_ (0)
  34 #endif /* ACE_MT_SAFE */
  35 {
  36 }
  37
  38 Options::~Options ()
  39 {
  40   ACE_MT (delete this->barrier_);
  41   delete [] this->message_buf_;
  42 }
  43
  44 // Options Singleton.
  45 typedef ACE_Singleton<Options, ACE_SYNCH_RECURSIVE_MUTEX> OPTIONS;
  46
  47 int
  48 Options::init ()
  49 {
  50   ACE_DEBUG((LM_DEBUG,"Options::init, len = %d\n",this->message_len_));
  51
  52   // Check for default case.
  53   if (this->message_len_ == 0)
  54     this->message_len_ = ACE_OS::strlen ("TAO");
  55   ACE_DEBUG((LM_DEBUG,"Options::init, len = %d\n",this->message_len_));
  56
  57   this->message_len_ += sizeof (ACE_UINT32);
  58   ACE_DEBUG((LM_DEBUG,"Options::init, len = %d\n",this->message_len_));
  59
  60   ACE_NEW_RETURN (this->message_buf_,
  61                   char[this->message_len_],
  62                   -1);
  63
  64   // Copy the length into the beginning of the message.
  65   ACE_UINT32 length =
  66     ACE_NTOHL (ACE_Utils::truncate_cast<ACE_UINT32> (this->message_len_));
  67
  68   ACE_OS::memcpy ((void *) this->message_buf_,
  69                   (void *) &length,
  70                   sizeof length);
  71
  72   ACE_OS::memset ((void *) (this->message_buf_ + sizeof (ACE_UINT32)),
  73                   'a',
  74                   this->message_len_ - sizeof (ACE_UINT32));
  75
  76   // Allocate the barrier with the correct count.
  77   ACE_MT (ACE_NEW_RETURN (this->barrier_,
  78                           ACE_Barrier (
  79                             ACE_Utils::truncate_cast<unsigned int> (
  80                               this->threads_)),
  81                           -1));
  82   return 0;
  83 }
  84
  85 size_t
  86 Options::message_len () const
  87 {
  88   return this->message_len_;
  89 }
  90
  91 const void *
  92 Options::message_buf () const
  93 {
  94   return this->message_buf_;
  95 }
  96
  97 ssize_t
  98 Options::read (void *buf, size_t len, size_t &iteration)
  99 {
 100   ACE_UNUSED_ARG (len);
 101
 102   if (this->io_source_ == ACE_STDIN)
 103     return ACE_OS::read (ACE_STDIN, buf, len);
 104   else if (iteration >= this->iterations_)
 105     return 0;
 106   else
 107     {
 108       ACE_OS::memcpy (buf,
 109                       this->message_buf (),
 110                       len);
 111       iteration++;
 112       return ACE_Utils::truncate_cast<ssize_t> (len);
 113     }
 114 }
 115
 116 int
 117 Options::parse_args (int argc, ACE_TCHAR *argv[])
 118 {
 119   //FUZZ: disable check_for_lack_ACE_OS
 120   ACE_Get_Opt getopt (argc, argv, ACE_TEXT("2h:i:m:p:q:st:T:"), 1);
 121
 122   for (int c; (c = getopt ()) != -1; )
 123   //FUZZ: enable check_for_lack_ACE_OS
 124     switch (c)
 125       {
 126       case '2': // Disable the oneway client.
 127         this->oneway_ = 0;
 128         break;
 129       case 'h':
 130         this->host_ = getopt.opt_arg ();
 131         break;
 132       case 'i':
 133         this->iterations_ = ACE_OS::atoi (getopt.opt_arg ());
 134         break;
 135       case 'm':
 136         this->message_len_ = ACE_OS::atoi (getopt.opt_arg ());
 137         break;
 138       case 'p':
 139         this->port_ = ACE_OS::atoi (getopt.opt_arg ());
 140         break;
 141       case 'q':
 142         this->quit_string_ = getopt.opt_arg ();
 143         break;
 144       case 's':
 145         this->io_source_ = ACE_STDIN;
 146         break;
 147       case 't':
 148         this->threads_ = (size_t) ACE_OS::atoi (getopt.opt_arg ());
 149         break;
 150       case 'T':
 151         this->sleep_time_.set (0, ACE_OS::atoi (getopt.opt_arg ()));
 152         break;
 153       default:
 154         ACE_ERROR_RETURN ((LM_ERROR,
 155                            "(%P|%t) usage: %n [-2] [-h <host>] [-i iterations] [-m message-size] [-p <port>] [-q <quit string>] [-s] [-t <threads>] [-T <sleep_time>]\n"),
 156                           -1);
 157       }
 158
 159   return this->init ();
 160 }
 161
 162 u_short
 163 Options::port () const
 164 {
 165   return this->port_;
 166 }
 167
 168 const ACE_TCHAR *
 169 Options::host () const
 170 {
 171   return this->host_;
 172 }
 173
 174 const ACE_TCHAR *
 175 Options::quit_string () const
 176 {
 177   return this->quit_string_;
 178 }
 179
 180 size_t
 181 Options::threads () const
 182 {
 183   return this->threads_;
 184 }
 185
 186 const ACE_Time_Value &
 187 Options::sleep_time () const
 188 {
 189   return this->sleep_time_;
 190 }
 191
 192 char *
 193 Options::shared_client_test (u_short port,
 194                              ACE_SOCK_Stream &cli_stream)
 195 {
 196   ACE_INET_Addr remote_addr (port, this->host_);
 197
 198   ACE_SOCK_Connector con;
 199
 200   if (con.connect (cli_stream,
 201                    remote_addr) == -1)
 202     ACE_ERROR_RETURN ((LM_ERROR,
 203                        "(%P|%t) %p\n",
 204                        "connection failed"),
 205                       0);
 206   else
 207     ACE_DEBUG ((LM_DEBUG,
 208                 "(%P|%t) connected to %s at port %d\n",
 209                 remote_addr.get_host_name (),
 210                 remote_addr.get_port_number ()));
 211
 212   ACE_INT32 len =
 213     ACE_HTONL (ACE_Utils::truncate_cast<ACE_INT32> (this->message_len ()));
 214
 215   // Allocate the transmit buffer.
 216   char *buf;
 217   ACE_DEBUG((LM_DEBUG,"(%P|%t) allocating buffer, len = %d msglen = %d\n",
 218              len, message_len_));
 219
 220   ACE_NEW_RETURN (buf,
 221                   char[this->message_len()],
 222                   0);
 223
 224   ACE_DEBUG ((LM_DEBUG,
 225               "(%P|%t) waiting...\n"));
 226
 227   // Wait for all other threads to finish initialization.
 228   ACE_MT (this->barrier_->wait ());
 229   return buf;
 230 }
 231 // Static function entry point to the oneway client service.
 232
 233 void *
 234 Options::oneway_client_test (void *)
 235 {
 236   Options *options = OPTIONS::instance ();
 237   ACE_SOCK_Stream cli_stream;
 238
 239   ACE_DEBUG((LM_DEBUG,"options = %d, len = %d\n",options,options->message_len()));
 240
 241   // Add 1 to the port to trigger the oneway test!
 242   char *request = options->shared_client_test (options->port () + 1,
 243                                                cli_stream);
 244   if (request == 0)
 245     return 0;
 246
 247   // This variable is allocated off the stack to obviate the need for
 248   // locking.
 249   size_t iteration = 0;
 250
 251   // Keep track of return value.
 252   intptr_t result = 0;
 253   ACE_INT32 len =
 254     ACE_Utils::truncate_cast<ACE_INT32> (options->message_len ());
 255
 256   ACE_DEBUG ((LM_DEBUG,
 257               "(%P|%t) starting oneway transmission\n"));
 258
 259   // Perform oneway transmission of data to server (correctly handles
 260   // "incomplete writes").
 261
 262   for (ssize_t r_bytes;
 263        (r_bytes = options->read (request, len, iteration)) > 0;
 264        // Transmit at the proper rate.
 265        ACE_OS::sleep (options->sleep_time ()))
 266     if (ACE_OS::memcmp (request,
 267                         options->quit_string (),
 268                         ACE_OS::strlen (options->quit_string ())) == 0)
 269       break;
 270     else if (cli_stream.send_n (request, r_bytes) == -1)
 271       {
 272         ACE_ERROR ((LM_ERROR,
 273                     "(%P|%t) %p\n",
 274                     "send_n"));
 275         result = -1;
 276         break;
 277       }
 278
 279   // Close the connection.
 280   cli_stream.close ();
 281
 282   delete [] request;
 283   return reinterpret_cast<void *> (result);
 284 }
 285
 286 // Static function entry point to the twoway client service.
 287
 288 void *
 289 Options::twoway_client_test (void *)
 290 {
 291   Options *options = OPTIONS::instance ();
 292
 293   ACE_SOCK_Stream cli_stream;
 294
 295   char *request = options->shared_client_test (options->port (),
 296                                                cli_stream);
 297   if (request == 0)
 298     return 0;
 299
 300   // This variable is allocated off the stack to obviate the need for
 301   // locking.
 302   size_t iteration = 0;
 303
 304   // Keep track of return value.
 305   intptr_t result = 0;
 306
 307   // Timer business.
 308   ACE_High_Res_Timer timer;
 309
 310   ACE_INT32 len =
 311     ACE_Utils::truncate_cast<ACE_INT32> (options->message_len ());
 312
 313   ACE_DEBUG ((LM_DEBUG,
 314               "(%P|%t) starting twoway transmission\n"));
 315
 316   // Perform twoway transmission of data to server (correctly handles
 317   // "incomplete writes").
 318
 319   for (ssize_t r_bytes;
 320        (r_bytes = options->read (request, len, iteration)) > 0;
 321        // Transmit at the proper rate.
 322        ACE_OS::sleep (options->sleep_time ()))
 323     if (ACE_OS::memcmp (request,
 324                         options->quit_string (),
 325                         ACE_OS::strlen (options->quit_string ())) == 0)
 326       break;
 327
 328     // Transmit <request> to the server.
 329     else
 330       {
 331         // Note that we use the incremental feature of the
 332         // <ACE_High_Res_Timer> so that we don't get "charged" for the
 333         // <ACE_OS::sleep> used to control the rate at which requests
 334         // are sent.
 335         timer.start_incr ();
 336
 337         if (cli_stream.send_n (request, r_bytes) == -1)
 338           {
 339             ACE_ERROR ((LM_ERROR,
 340                         "(%P|%t) %p\n",
 341                         "send_n"));
 342             result = -1;
 343             break;
 344           }
 345         // Receive the reply from the server.  Normally, it just sends
 346         // back 24 bytes, which is typical for an IIOP reply.
 347         else if (cli_stream.recv (request, r_bytes) <= 0)
 348           {
 349             ACE_ERROR ((LM_ERROR,
 350                         "(%P|%t) %p\n",
 351                         "recv"));
 352             result = -1;
 353             break;
 354           }
 355
 356         timer.stop_incr ();
 357       }
 358
 359   ACE_Time_Value tv;
 360
 361   timer.elapsed_time_incr (tv);
 362   double real_time = (long) tv.sec () * ACE_ONE_SECOND_IN_USECS + tv.usec ();
 363   double messages_per_sec = iteration * double (ACE_ONE_SECOND_IN_USECS) / real_time;
 364
 365   ACE_DEBUG ((LM_DEBUG,
 366               ACE_TEXT ("(%t) messages = %d\n(%t) usec-per-message = %f\n(%t) messages-per-second = %0.00f\n"),
 367               iteration,
 368               real_time / double (iteration),
 369               messages_per_sec < 0 ? 0 : messages_per_sec));
 370
 371   // Close the connection.
 372   cli_stream.close ();
 373
 374   delete [] request;
 375   return reinterpret_cast<void *> (result);
 376 }
 377
 378 ACE_THR_FUNC
 379 Options::thr_func ()
 380 {
 381   ACE_DEBUG((LM_DEBUG,"(%P|%t) in thread func, mesg len = %d\n",this->message_len()));
 382   if (this->oneway_ == 0)
 383     return ACE_THR_FUNC (&Options::twoway_client_test);
 384   else
 385     return ACE_THR_FUNC (&Options::oneway_client_test);
 386 }
 387
 388 static int
 389 run_client ()
 390 {
 391   // Raise the socket handle limit to the maximum.
 392   ACE::set_handle_limit ();
 393
 394 #if defined (ACE_HAS_THREADS)
 395   ACE_DEBUG((LM_DEBUG,"(%P|%t) spawning client test thread options = %d\n",
 396   OPTIONS::instance()));
 397
 398   if (ACE_Thread_Manager::instance ()->spawn_n (OPTIONS::instance ()->threads (),
 399                                                 OPTIONS::instance ()->thr_func ()) == -1)
 400     ACE_ERROR_RETURN ((LM_ERROR,
 401                        "(%P|%t) %p\n",
 402                        "spawn_n"),
 403                       1);
 404   else
 405     ACE_Thread_Manager::instance ()->wait ();
 406 #else
 407   (void) *(OPTIONS::instance ()->thr_func) ();
 408 #endif /* ACE_HAS_THREADS */
 409   return 0;
 410 }
 411
 412 int
 413 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 414 {
 415   // Initialize the logger.
 416   ACE_LOG_MSG->open (argv[0]);
 417
 418   if (OPTIONS::instance ()->parse_args (argc, argv) == -1)
 419     return -1;
 420
 421   // Run the client
 422   run_client ();
 423
 424   return 0;
 425 }
 426
 427 ACE_SINGLETON_TEMPLATE_INSTANTIATE(ACE_Singleton, Options, ACE_SYNCH_RECURSIVE_MUTEX);
 428