ACE/performance-tests/UDP/udp_test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    udp_test.cpp
   5  *
   6  *  Measures UDP round-trip performance.
   7  *
   8  *  @author Fred Kuhns and David L. Levine
   9  */
  10 //=============================================================================
  11
  12
  13 #include "ace/OS_main.h"
  14 #include "ace/Reactor.h"
  15 #include "ace/SOCK_Dgram.h"
  16 #include "ace/INET_Addr.h"
  17 #include "ace/ACE.h"
  18 #include "ace/Get_Opt.h"
  19 #include "ace/High_Res_Timer.h"
  20 #include "ace/Log_Msg.h"
  21 #include "ace/OS_NS_stdio.h"
  22 #include "ace/OS_NS_ctype.h"
  23 #include "ace/OS_NS_arpa_inet.h"
  24 #include "ace/OS_NS_string.h"
  25 #include "ace/os_include/os_netdb.h"
  26 #include "ace/OS_NS_unistd.h"
  27
  28 // FUZZ: disable check_for_math_include
  29 #include <math.h>
  30
  31 // Global variables (evil).
  32 static const u_short DEFPORT = 5050;
  33 static const int MAXPKTSZ = 65536;
  34 static const int DEFPKTSZ = 64;
  35 static const int DEFITERATIONS = 1000;
  36 static const int DEFINTERVAL = 1000; // 1000 usecs.
  37 static const int DEFWINDOWSZ = 10; // 10 microsecond.
  38 static char SendBuf[MAXPKTSZ];
  39 static char RxBuf[MAXPKTSZ];
  40 static ACE_TCHAR **cmd;
  41 static ACE_TCHAR datafile[MAXHOSTNAMELEN];
  42
  43 static ACE_UINT32 nsamples = DEFITERATIONS;
  44 static int usdelay = DEFINTERVAL;
  45 static int bufsz = DEFPKTSZ;
  46 static int window = DEFWINDOWSZ;
  47 static int VERBOSE = 0;
  48 static int logfile = 0;
  49 static int server = 0;
  50 static int client = 0;
  51 static u_int use_reactor = 0;
  52 ACE_hrtime_t max_allow = 0;
  53 ACE_hrtime_t total_ltime;
  54 ACE_hrtime_t ltime;
  55
  56 static void
  57 usage ()
  58 {
  59   ACE_ERROR ((LM_ERROR,
  60               "%s\n"
  61               "[-w window_size]\n"
  62               "  [-f datafile] (creates datafile.samp and datafile.dist)\n"
  63               "  [-v]          (Verbose)\n"
  64               "  [-b send_bufsz]\n"
  65               "  [-n nsamples]\n"
  66               "  [-I usdelay]\n"
  67               "  [-s so_bufsz]\n"
  68               "  [-p port]\n"
  69               "  [-t]\n"
  70               "  [-r]\n"
  71               "  [-x max_sample_allowed]\n"
  72               "  [-a to use the ACE reactor]\n"
  73               "  targethost\n",
  74               *cmd));
  75 }
  76
  77 static ACE_hrtime_t *Samples;
  78 static u_int *Dist;
  79 static ACE_TCHAR sumfile[30];
  80 static ACE_TCHAR distfile[30];
  81 static ACE_TCHAR sampfile[30];
  82
  83 class Client : public ACE_Event_Handler
  84 {
  85 public:
  86   Client (const ACE_INET_Addr &addr,
  87           const ACE_INET_Addr &remote_addr);
  88
  89   virtual ~Client ();
  90
  91   // = Override <ACE_Event_Handler> methods.
  92   virtual ACE_HANDLE get_handle () const;
  93   virtual int handle_input (ACE_HANDLE);
  94   virtual int handle_close (ACE_HANDLE handle,
  95                             ACE_Reactor_Mask close_mask);
  96
  97   //FUZZ: disable check_for_lack_ACE_OS
  98   /// Send the <buf> to the server.
  99   int send (const char *buf, size_t len);
 100   //FUZZ: enable check_for_lack_ACE_OS
 101
 102   /// Wait for the response.
 103   int get_response (char *buf, size_t len);
 104
 105   /// Send messages to server and record statistics.
 106   int run ();
 107
 108   //FUZZ: disable check_for_lack_ACE_OS
 109   /// Send shutdown message to server.
 110   int shutdown ();
 111   //FUZZ: enable check_for_lack_ACE_OS
 112
 113 private:
 114   /// To send messages and receive responses.
 115   ACE_SOCK_Dgram endpoint_;
 116
 117   /// The address to send messages to.
 118   ACE_INET_Addr remote_addr_;
 119
 120   Client () = delete;
 121   Client (const Client &) = delete;
 122   Client &operator= (const Client &) = delete;
 123 };
 124
 125 Client::Client (const ACE_INET_Addr &addr,
 126                 const ACE_INET_Addr &remote_addr)
 127   : endpoint_ (addr),
 128     remote_addr_ (remote_addr)
 129 {
 130   if (use_reactor)
 131     {
 132       if (ACE_Reactor::instance ()->register_handler
 133           (this, ACE_Event_Handler::READ_MASK) == -1)
 134         ACE_ERROR ((LM_ERROR,
 135                     "ACE_Reactor::register_handler: Client\n"));
 136     }
 137 }
 138
 139 Client::~Client ()
 140 {
 141 }
 142
 143 ACE_HANDLE
 144 Client::get_handle () const
 145 {
 146   return endpoint_.get_handle ();
 147 }
 148
 149 int
 150 Client::handle_input (ACE_HANDLE)
 151 {
 152   char buf[BUFSIZ];
 153   ACE_INET_Addr from_addr;
 154
 155   ssize_t n = endpoint_.recv (buf, sizeof buf, from_addr);
 156
 157   if (n == -1)
 158     ACE_ERROR ((LM_ERROR,
 159                 "%p\n",
 160                 "handle_input"));
 161   else
 162     ACE_DEBUG ((LM_DEBUG,
 163                 "(%P|%t) buf of size %d = %*s\n",
 164                 n,
 165                 n,
 166                 buf));
 167
 168   return 0;
 169 }
 170
 171 int
 172 Client::handle_close (ACE_HANDLE,
 173                       ACE_Reactor_Mask)
 174 {
 175   this->endpoint_.close ();
 176   return 0;
 177 }
 178
 179 int
 180 Client::send (const char *buf, size_t len)
 181 {
 182   return this->endpoint_.send (buf, len, remote_addr_);
 183 }
 184
 185 int
 186 Client::get_response (char *buf, size_t len)
 187 {
 188   ACE_INET_Addr addr;
 189   return this->endpoint_.recv (buf, len, addr);
 190 }
 191
 192 int
 193 Client::run ()
 194 {
 195   int ndist = 0;
 196   int i;
 197   int j;
 198   int n;
 199   int maxindx = 0;
 200   int minindx = 0;
 201   char *sbuf = SendBuf;
 202   char *rbuf = RxBuf;
 203
 204   ACE_High_Res_Timer timer;
 205   ACE_hrtime_t sample;
 206
 207   int d;
 208   double std_dev = 0.0;
 209   double std_err = 0.0;
 210   double sample_mean = 0.0;
 211
 212   int                tracking_last_over = 0;
 213   ACE_High_Res_Timer since_over;
 214   ACE_hrtime_t psum = 0;
 215   ACE_hrtime_t sum = 0;
 216   ACE_hrtime_t max = 0;
 217   ACE_hrtime_t min = (ACE_hrtime_t) (u_int) -1;
 218   FILE *sumfp = 0;
 219   FILE *distfp = 0;
 220   FILE *sampfp = 0;
 221   pid_t *pid = (pid_t *) sbuf;
 222   int *seq = (int *) (sbuf + sizeof (int));
 223
 224   ACE_OS::memset (sbuf, 0, bufsz);
 225   ACE_OS::memset (rbuf, 0, bufsz);
 226
 227   *pid = ACE_OS::getpid ();
 228   *seq = 0;
 229
 230   ACE_DEBUG ((LM_DEBUG,
 231               "PID = %d, Starting SEQ = %d\n",
 232               *pid,
 233               *seq));
 234
 235   // Allocate memory to hold samples.
 236   Samples = (ACE_hrtime_t *) ACE_OS::calloc (nsamples,
 237                                              sizeof (ACE_hrtime_t));
 238
 239   for (i = -1, *seq = 0, j = 0;
 240        i < (ACE_INT32) nsamples;
 241        (*seq)++, i++, j++, timer.reset ())
 242     {
 243       timer.start ();
 244       if (this->send (sbuf, bufsz) <= 0)
 245         ACE_ERROR_RETURN ((LM_ERROR, "(%P) %p\n", "send"), -1);
 246
 247       if ((n = get_response (rbuf, bufsz)) <= 0)
 248         ACE_ERROR_RETURN ((LM_ERROR, "(%P) %p\n", "get_response"), -1);
 249
 250       timer.stop ();
 251
 252       if (n <= 0)
 253         ACE_ERROR_RETURN ((LM_ERROR,
 254                            "\nTrouble receiving from socket!\n\n"),
 255                           -1);
 256
 257       timer.elapsed_time (sample);     // in nanoseconds.
 258
 259       if (i < 0 )
 260         {
 261           ACE_DEBUG ((LM_DEBUG,
 262                       "Ignoring first sample of %u usecs\n",
 263                        (ACE_UINT32) (sample)));
 264           continue;
 265         }
 266       else if (max_allow > 0  &&  sample > max_allow)
 267         {
 268           ACE_DEBUG ((LM_DEBUG, "Sample # %i = "
 269                       "%u msec is over the limit (%u)!\n",
 270                       i,
 271                       (ACE_UINT32) (sample / (ACE_UINT32) 1000000),
 272                       (ACE_UINT32) (max_allow / (ACE_UINT32) 1000000)));
 273
 274           if (tracking_last_over)
 275             {
 276               since_over.stop ();
 277               ACE_Time_Value over_time;
 278               since_over.elapsed_time (over_time);
 279               ACE_DEBUG ((LM_DEBUG,
 280                           "\tTime since last over = %u msec!\n",
 281                           over_time.msec ()));
 282               since_over.reset ();
 283             }
 284
 285           tracking_last_over = 1;
 286           since_over.start ();
 287           i--;
 288           continue;
 289         }
 290
 291       Samples[i] = sample;
 292       sum += sample;
 293
 294       if (min == (ACE_hrtime_t) (u_int) -1)
 295         {
 296           min = sample;
 297           minindx = i;
 298         }
 299       if (sample > max)
 300         {
 301           max = sample;
 302           maxindx = i;
 303         }
 304       if (sample < min)
 305         {
 306           min = sample;
 307           minindx = i;
 308         }
 309
 310       if (VERBOSE)
 311         {
 312           psum += sample;
 313           if (j == 500)
 314             {
 315               ACE_DEBUG ((LM_DEBUG,
 316                           "(%i) Partial (running) mean %u usecs\n",
 317                           i + 1,
 318                           (ACE_UINT32) (psum / (ACE_UINT32) (1000 * 500))));
 319               j = 0;
 320               psum = 0;
 321             }
 322         }
 323     }
 324
 325   sample_mean = ((double) ACE_U64_TO_U32 (sum)) / (double) nsamples;
 326
 327   if (logfile)
 328     {
 329       ACE_OS::sprintf (sumfile, ACE_TEXT("%s.sum"), datafile);
 330       ACE_OS::sprintf (distfile, ACE_TEXT("%s.dist"), datafile);
 331       ACE_OS::sprintf (sampfile, ACE_TEXT("%s.samp"), datafile);
 332
 333       distfp = ACE_OS::fopen(distfile, ACE_TEXT("w"));
 334
 335       if (distfp == 0)
 336         {
 337           ACE_DEBUG ((LM_DEBUG,
 338                       "Unable to open dist file!\n\n"));
 339           logfile = 0;
 340         }
 341       if (logfile && (sampfp = ACE_OS::fopen (sampfile, ACE_TEXT("w"))) == 0)
 342         {
 343           ACE_OS::fclose (distfp);
 344           ACE_DEBUG ((LM_DEBUG,
 345                       "Unable to open sample file!\n\n"));
 346           logfile = 0;
 347         }
 348       if (logfile && (sumfp = ACE_OS::fopen (sumfile, ACE_TEXT("w"))) == 0)
 349         {
 350           ACE_OS::fclose (distfp);
 351           ACE_OS::fclose (sampfp);
 352           ACE_DEBUG ((LM_DEBUG,
 353                       "Unable to open sample file!\n\n"));
 354           logfile = 0;
 355         }
 356     }
 357
 358   window = window * 1000; // convert to nsec.
 359   ndist = (int)((max-min) / window) + 1;
 360   Dist = (u_int *) ACE_OS::calloc (ndist,
 361                                    sizeof (u_int));
 362
 363   for (i = 0; i < (ACE_INT32) nsamples; i++)
 364     {
 365       std_dev += ((double) ACE_U64_TO_U32 (Samples[i]) - sample_mean) *
 366         ((double) ACE_U64_TO_U32 (Samples[i]) - sample_mean);
 367       d = (int)((Samples[i] - min)/window);
 368
 369       if (d < 0 || d > ndist)
 370         {
 371           ACE_DEBUG ((LM_DEBUG,
 372                       "\nError indexing into dist array %d (%d)\n\n",
 373                       d,
 374                       ndist));
 375           ACE_OS::exit (1);
 376         }
 377       Dist[d] += 1;
 378       if (logfile)
 379         ACE_OS::fprintf (sampfp,
 380                          "%u\n",
 381                          ACE_U64_TO_U32 (Samples[i]));
 382     }
 383
 384   if (logfile)
 385     {
 386       ACE_hrtime_t tmp;
 387       tmp = min + (window / 2);
 388
 389       for (i = 0; i < ndist; i++)
 390         {
 391           ACE_OS::fprintf (distfp,
 392                            "%u %d\n",
 393                            (ACE_UINT32) (tmp / (ACE_UINT32) 1000),
 394                            Dist[i]);
 395           tmp += window;
 396         }
 397     }
 398
 399   std_dev = (double) sqrt (std_dev / (double) (nsamples - 1.0));
 400   std_err = (double) std_dev / sqrt ((double) nsamples);
 401
 402   ACE_DEBUG ((LM_DEBUG,
 403               "\nResults for %i samples (usec):\n"
 404               "\tSample Mean = %f,\n"
 405               "\tSample Max = %u, Max index = %d,\n"
 406               "\tSample Min = %u, Min index = %d,\n"
 407               "\tStandard Deviation = %f,\n"
 408               "\tStandard Error = %f\n",
 409               nsamples,
 410               sample_mean / 1000.0,
 411               (u_int) (max / (ACE_UINT32) 1000),
 412               maxindx,
 413               (u_int) (min / (ACE_UINT32) 1000),
 414               minindx,
 415               std_dev / 1000.0,
 416               std_err / 1000.0));
 417
 418   if (logfile)
 419     {
 420       ACE_OS::fprintf (sumfp,
 421                        "Command executed:\n");
 422       for (; *cmd; cmd++)
 423         ACE_OS::fprintf (sumfp,
 424                          "%s ",
 425                          ACE_TEXT_ALWAYS_CHAR (*cmd));
 426       ACE_OS::fprintf (sumfp,
 427                        "\n");
 428
 429       ACE_OS::fprintf (sumfp,
 430                        "\nResults for %i samples (usec):"
 431                        "\tSample Mean = %f,\n"
 432                        "\tSample Max = %u, Max index = %d,\n"
 433                        "\tSample Min = %u, Min index = %d,\n"
 434                        "\tStandard Deviation = %f,\n"
 435                        "\tStandard Error = %f\n",
 436                        nsamples,
 437                        sample_mean / 1000.0,
 438                        (ACE_UINT32) (max / (ACE_UINT32) 1000),
 439                        maxindx,
 440                        (ACE_UINT32) (min / (ACE_UINT32) 1000),
 441                        minindx,
 442                        std_dev / 1000.0,
 443                        std_err / 1000.0);
 444     }
 445
 446   return 0;
 447 }
 448
 449 int
 450 Client::shutdown ()
 451 {
 452   const char buf = 'S';
 453   const int n = endpoint_.send (&buf, 1u, remote_addr_);
 454
 455   if (use_reactor)
 456     {
 457       if (ACE_Reactor::instance ()->remove_handler
 458           (this, ACE_Event_Handler::READ_MASK) == -1)
 459         ACE_ERROR_RETURN ((LM_ERROR,
 460                            "ACE_Reactor::remove_handler: Client\n"),
 461                           -1);
 462     }
 463
 464   return n;
 465 }
 466
 467 class Server : public ACE_Event_Handler
 468 {
 469 public:
 470   Server (const ACE_INET_Addr &addr);
 471
 472   virtual ~Server ();
 473
 474   // = Override <ACE_Event_Handler> methods.
 475   virtual ACE_HANDLE get_handle () const;
 476   virtual int handle_input (ACE_HANDLE);
 477   virtual int handle_close (ACE_HANDLE handle,
 478                             ACE_Reactor_Mask close_mask);
 479
 480 private:
 481   /// Receives datagrams.
 482   ACE_SOCK_Dgram endpoint_;
 483
 484   Server () = delete;
 485   Server (const Server &) = delete;
 486   Server &operator= (const Server &) = delete;
 487 };
 488
 489 Server::Server (const ACE_INET_Addr &addr)
 490   :  endpoint_ (addr)
 491 {
 492   if (use_reactor)
 493     {
 494       if (ACE_Reactor::instance ()->register_handler
 495           (this,
 496            ACE_Event_Handler::READ_MASK) == -1)
 497         ACE_ERROR ((LM_ERROR,
 498                     "ACE_Reactor::register_handler: Server\n"));
 499     }
 500 }
 501
 502 Server::~Server ()
 503 {
 504 }
 505
 506 ACE_HANDLE
 507 Server::get_handle () const
 508 {
 509   return endpoint_.get_handle ();
 510 }
 511
 512 int
 513 Server::handle_input (ACE_HANDLE)
 514 {
 515   char buf[BUFSIZ];
 516   ACE_INET_Addr from_addr;
 517
 518   ssize_t n = endpoint_.recv (buf, sizeof buf, from_addr);
 519
 520   if (n == -1)
 521     ACE_DEBUG ((LM_ERROR,
 522                 "%p\n",
 523                 "handle_input: recv"));
 524
 525   // Send the message back as the response.
 526   if (endpoint_.send (buf, n, from_addr) == n)
 527     {
 528       if (n == 1 && buf[0] == 'S')
 529         {
 530           if (use_reactor)
 531             {
 532               if (ACE_Reactor::instance ()->remove_handler
 533                   (this, ACE_Event_Handler::READ_MASK) == -1)
 534                 ACE_ERROR_RETURN ((LM_ERROR,
 535                                    "ACE_Reactor::remove_handler: server\n"),
 536                                   -1);
 537
 538               ACE_Reactor::end_event_loop ();
 539             }
 540           else
 541             {
 542               // Indicate done by returning 1.
 543               return 1;
 544             }
 545         }
 546
 547       return 0;
 548     }
 549   else
 550     {
 551       ACE_DEBUG ((LM_ERROR,
 552                   "%p\n",
 553                   "handle_input: send"));
 554       return -1;
 555     }
 556 }
 557
 558 int
 559 Server::handle_close (ACE_HANDLE,
 560                       ACE_Reactor_Mask)
 561 {
 562   endpoint_.close ();
 563
 564   return 0;
 565 }
 566
 567 int
 568 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 569 {
 570   int c, dstport = DEFPORT;
 571   int so_bufsz = 0;
 572
 573   cmd = argv;
 574
 575   //FUZZ: disable check_for_lack_ACE_OS
 576   ACE_Get_Opt getopt (argc, argv, ACE_TEXT("x:w:f:vs:I:p:rtn:b:a"));
 577
 578   while ((c = getopt ()) != -1)
 579     {
 580   //FUZZ: enable check_for_lack_ACE_OS
 581       switch ((char) c)
 582         {
 583         case 'x':
 584           max_allow = ACE_OS::atoi (getopt.opt_arg ());
 585           break;
 586         case 'w':
 587           window = ACE_OS::atoi (getopt.opt_arg ());
 588           if (window < 2)
 589             ACE_ERROR_RETURN ((LM_ERROR,
 590                                "Invalid window!\n\n"),
 591                               1);
 592           break;
 593         case 'f':
 594           ACE_OS::strcpy (datafile, getopt.opt_arg ());
 595           logfile = 1;
 596           break;
 597         case 'v':
 598           VERBOSE = 1;
 599           break;
 600         case 'b':
 601           bufsz = ACE_OS::atoi (getopt.opt_arg ());
 602
 603           if (bufsz <= 0)
 604             ACE_ERROR_RETURN ((LM_ERROR,
 605                                "\nBuffer size must be greater than 0!\n\n"),
 606                               1);
 607           break;
 608         case 'n':
 609           nsamples = ACE_OS::atoi (getopt.opt_arg ());
 610           if (nsamples <= 0)
 611             ACE_ERROR_RETURN ((LM_ERROR,
 612                                "\nIterations must be greater than 0!\n\n"),
 613                               1);
 614           break;
 615         case 'a':
 616           use_reactor = 1;
 617           break;
 618         case 's':
 619           so_bufsz = ACE_OS::atoi (getopt.opt_arg ());
 620
 621           if (so_bufsz <= 0)
 622             ACE_ERROR_RETURN ((LM_ERROR,
 623                                "\nInvalid socket buffer size!\n\n"),
 624                               1);
 625           break;
 626         case 'I':
 627           usdelay = ACE_OS::atoi (getopt.opt_arg ());
 628
 629           if (usdelay == 0)
 630             ACE_ERROR_RETURN ((LM_ERROR,
 631                                "%s: bad usdelay: %s\n",
 632                                argv[0],
 633                                getopt.opt_arg ()),
 634                               1);
 635           break;
 636         case 'p':
 637           dstport = ACE_OS::atoi (getopt.opt_arg ());
 638           if (dstport <= 0)
 639             ACE_ERROR_RETURN ((LM_ERROR,
 640                                "\nInvalid port number!\n\n"),
 641                               1);
 642           break;
 643         case 't':
 644           server = 0;
 645           client = 1;
 646           break;
 647         case 'r':
 648           client = 0;
 649           server = 1;
 650           break;
 651         default:
 652           usage ();
 653           return 1;
 654         }
 655     }
 656
 657   if ((getopt.opt_ind () >= argc && client != 0) || argc == 1)
 658     {
 659       usage ();
 660       return 1;
 661     }
 662
 663   ACE_INET_Addr addr (server ? dstport : dstport + 1);
 664
 665   if (server)
 666     {
 667       Server server (addr);
 668
 669       if (use_reactor)
 670         {
 671           ACE_Reactor::run_event_loop ();
 672         }
 673       else
 674         {
 675           // Handle input in the current thread.
 676           while (server.handle_input (0) != 1)
 677             {
 678               continue;
 679             }
 680         }
 681     }
 682   else
 683     {
 684       if ((u_int) bufsz < sizeof (ACE_hrtime_t))
 685         ACE_ERROR_RETURN ((LM_ERROR,
 686                            "\nbufsz must be >= %d\n",
 687                            sizeof (ACE_hrtime_t)),
 688                           1);
 689       ACE_INET_Addr remote_addr;
 690
 691       if (ACE_OS::ace_isdigit(argv[getopt.opt_ind ()][0]))
 692         {
 693           if (remote_addr.set (dstport,
 694                                (ACE_UINT32) ACE_OS::inet_addr
 695                                  (ACE_TEXT_ALWAYS_CHAR(argv[getopt.opt_ind ()]))) == -1)
 696             ACE_ERROR_RETURN ((LM_ERROR,
 697                                "invalid IP address: %s\n",
 698                                argv[getopt.opt_ind ()]),
 699                               1);
 700         }
 701       else
 702         {
 703           if (remote_addr.set (dstport, argv[getopt.opt_ind ()]) == -1)
 704             ACE_ERROR_RETURN ((LM_ERROR,
 705                                "invalid IP address: %s\n",
 706                                argv[getopt.opt_ind ()]),
 707                               1);
 708         }
 709       getopt.opt_ind ()++;
 710
 711       Client client (addr, remote_addr);
 712
 713       ACE_DEBUG ((LM_DEBUG,
 714                   "\nSending %d byte packets to %s:%d "
 715                   "with so_bufsz = %d\n\n",
 716                   bufsz,
 717                   addr.get_host_name (),
 718                   dstport,
 719                   so_bufsz));
 720
 721       client.run ();
 722       client.shutdown ();
 723     }
 724
 725   return 0;
 726 }