ACE/performance-tests/SCTP/SOCK_STREAM_clt.cpp

   1 #include "ace/INET_Addr.h"
   2 #include "ace/SOCK_Stream.h"
   3 #include "ace/SOCK_Connector.h"
   4 #include "ace/Log_Msg.h"
   5 #include "ace/CDR_Stream.h"
   6 #include "ace/High_Res_Timer.h"
   7 #include "ace/OS_Memory.h"
   8
   9 // FUZZ: disable check_for_streams_include
  10 #include "ace/streams.h"
  11
  12 #include "ace/os_include/netinet/os_tcp.h"
  13 #include "ace/os_include/arpa/os_inet.h"
  14
  15 // make sure that the code compiles cleanly even if SCTP is not
  16 // available. If SCTP is not installed, program will exit early in
  17 // main() with an error message
  18 #ifdef ACE_HAS_SCTP
  19 extern "C" {
  20 #include <netinet/sctp.h>
  21 };
  22 #else
  23 #ifndef IPPROTO_SCTP
  24 #define IPPROTO_SCTP 132
  25 #endif
  26 #define SCTP_NODELAY -1
  27 #endif
  28
  29 // class that manages setting of options
  30 #include "Options_Manager.h"
  31
  32 // histogram code
  33 #include "hist.h"
  34
  35 // global constants
  36 ACE_CDR::UShort const primerIterations = 100;
  37 // ace/High_Res_Timer.h describes the need for the following.
  38 ACE_High_Res_Timer::global_scale_factor_type const microsec_clock_scale_factor =
  39   ACE_High_Res_Timer::global_scale_factor();
  40
  41 // forward declations of functions. Bodies follow main() to improve
  42 // file readability.
  43 ACE_SCTP::HIST runTest(ACE_SOCK_Stream &);
  44
  45 int ACE_TMAIN (int argc, ACE_TCHAR **argv){
  46   // Initialize the options manager
  47   Options_Manager optsMgr(argc, argv, ACE_TEXT ("client-opts"));
  48
  49   // show usage if requested
  50   if (optsMgr._usage) {
  51     optsMgr._show_usage(stderr, ACE_TEXT ("client-opts"));
  52     return 1;
  53   }
  54
  55   // If SCTP is not installed then terminate the program, unless TCP
  56   // was specified.
  57 #ifndef ACE_HAS_SCTP
  58   if (optsMgr.test_transport_protocol == IPPROTO_SCTP)
  59     ACE_ERROR_RETURN((LM_ERROR,
  60                       ACE_TEXT ("SCTP was NOT installed when this binary was compiled.\nSOCK_STREAM_clt may still be run using TCP via the '-t tcp' option.\n")),
  61                      1);
  62 #endif
  63
  64   // check that valid options were specified
  65   if (optsMgr._error) {
  66     ACE_OS::fprintf (stderr, "ERROR: %s\n", ACE_TEXT_ALWAYS_CHAR (optsMgr._error_message));
  67     return 1;
  68   }
  69
  70
  71   // Create the address that we want the client to connect to
  72   ACE_INET_Addr serverAddr(Options_Manager::server_port,
  73                            Options_Manager::server_host);
  74
  75   // Create the address that we want the client to connect FROM
  76   ACE_INET_Addr clientAddr(Options_Manager::client_port,
  77                            Options_Manager::client_connect_addr);
  78
  79   // object that manages the connection to the server
  80   ACE_SOCK_Connector connector;
  81
  82   // object that manages the data xfer between the client and server
  83   ACE_SOCK_Stream dataStream;
  84
  85   // connect to the server
  86   if (connector.connect (dataStream,
  87                          serverAddr,
  88                          0,clientAddr, 0, 0, 0, // ALL DEFAULT ARGUMENTS
  89                          Options_Manager::test_transport_protocol) == -1)
  90     ACE_ERROR_RETURN ((LM_ERROR,
  91                        ACE_TEXT ("(%P|%t) %p\n"),
  92                        ACE_TEXT ("connection failed")),
  93                       1);
  94
  95   // run the test
  96   ACE_SCTP::HIST testResultsHistogram = 0;
  97   // connection is closed by runTest* functions
  98   testResultsHistogram = runTest(dataStream);
  99
 100   // report out the test statistics.
 101   // all histograms created are placed onto a linked list that report()
 102   // can access. So the histogram created in one of the *_test() functions
 103   // will be reported out by the report() call
 104   if (testResultsHistogram)
 105     ACE_SCTP::report();
 106
 107   return 0;
 108 }
 109
 110   // create a histogram to store test results
 111   ACE_SCTP::HIST createHistogram(ACE_CDR::ULong messageSize){
 112   // The histogram created below lives beyond the scope of this
 113   // function. So the memory allocated here cannot be cleaned up when
 114   // this function goes out of scope. Unfortunately the histogram
 115   // implementation does not clean it up either. So it is a small
 116   // memory leak.
 117   char * histName = (char *) malloc(200);
 118
 119   ACE_OS::sprintf(histName, "%s Unmarshalled Msg Synchronous Latency Test\n\t\t\t\t\t(Message Size %u, Message Type octet)",
 120                   "ACE", messageSize);
 121
 122   // actually create the histogram
 123   ACE_SCTP::HIST createdHist = ACE_SCTP::histogram(histName,
 124                                Options_Manager::histogram_bin_count,
 125                                Options_Manager::histogram_min_bin,
 126                                Options_Manager::histogram_max_bin);
 127
 128   // set the maximum number of outliers to maintain in the histogram
 129   ACE_SCTP::set_outer(Options_Manager::histogram_num_outliers, createdHist);
 130
 131   return (createdHist);
 132 }
 133
 134 // send the test header (only contains number of iterations)
 135 int sendHeader(ACE_SOCK_Stream & stream) {
 136   // create an ACE CDR output stream and place the header information
 137   // into it
 138   ACE_OutputCDR hdrCDR;
 139     hdrCDR << ACE_OutputCDR::from_boolean (ACE_CDR_BYTE_ORDER);
 140     hdrCDR << ACE_CDR::ULong(Options_Manager::test_iterations+primerIterations);
 141   if (!hdrCDR.good_bit())
 142     return (0);
 143
 144   // send the header to the server (HEADER IS 8 BYTES LONG)
 145   size_t bt;
 146   if (stream.send_n(hdrCDR.begin(), 0,  &bt) == -1)
 147     return 0;
 148
 149   return 1;
 150 }
 151
 152 // conduct the UnMarshalled Octet performance test using separate
 153 // send_n calls with Nagle's algorithm disabled
 154 ACE_SCTP::HIST runUnmarshalledOctetTest(ACE_CDR::Octet *buf, size_t seqLen, ACE_SOCK_Stream & stream){
 155   ACE_CDR::ULong const testIterations = Options_Manager::test_iterations;
 156
 157   size_t bt;
 158   ACE_CDR::ULong cnt = 0;
 159   // variables for the timing measurements
 160   ACE_hrtime_t startTime, endTime;
 161   ACE_CDR::Double messageLatency_usec = 0.0;
 162   ACE_CDR::ULong msgLen = seqLen*ACE_CDR::OCTET_SIZE;
 163
 164   // explicity configure Nagling. Default is
 165   // Options_Manager::test_enable_nagle=0 so default configurations is
 166   // NO NAGLING
 167   ACE_CDR::Long nagle;
 168   if (Options_Manager::test_enable_nagle)
 169     nagle=0;
 170   else
 171     nagle=1;
 172   if (Options_Manager::test_transport_protocol == IPPROTO_SCTP){
 173     // default - sctp case
 174     if (-1 == stream.set_option(IPPROTO_SCTP, SCTP_NODELAY, &nagle, sizeof nagle))
 175       ACE_ERROR_RETURN ((LM_ERROR,
 176                          "%p\n",
 177                          "set_option"),
 178                         0);
 179   } else {
 180     // tcp case
 181     if (-1 == stream.set_option(IPPROTO_TCP, TCP_NODELAY, &nagle, sizeof nagle))
 182       ACE_ERROR_RETURN ((LM_ERROR,
 183                          "%p\n",
 184                          "set_option"),
 185                         0);
 186   }
 187
 188   // prime the client and server before starting the test
 189   for(cnt=0;cnt<primerIterations;++cnt){
 190     // send message size
 191     // TODO : The message length should be CDR encoded
 192     ACE_CDR::ULong msgLenExpressed = ACE_HTONL(msgLen);
 193     if (-1 == stream.send_n (&msgLenExpressed, ACE_CDR::LONG_SIZE, 0, &bt))
 194       ACE_ERROR_RETURN ((LM_ERROR,
 195                          "%p\n",
 196                          "send_n"),
 197                         0);
 198
 199     // send a message
 200     if (-1 == stream.send_n (buf, msgLen, 0, &bt))
 201       ACE_ERROR_RETURN ((LM_ERROR,
 202                          "%p\n",
 203                          "send_n"),
 204                         0);
 205
 206     // block for a Short reply
 207     ACE_CDR::Short reply;
 208     if ((stream.recv_n(&reply, ACE_CDR::SHORT_SIZE, 0, &bt)) == -1)
 209       ACE_ERROR_RETURN((LM_ERROR,
 210                         "%p\n",
 211                         "recv_n"),
 212                        0);
 213   }
 214
 215   // AFTER PRIMING THE PUMP CREATE THE HISTOGRAM
 216   ACE_SCTP::HIST aceStream_hist = 0;
 217   aceStream_hist = createHistogram(msgLen);
 218   if (0 == aceStream_hist)
 219     ACE_ERROR_RETURN((LM_ERROR,
 220                       "%p\n",
 221                       "histogram create failed"),
 222                      0);
 223
 224   iovec iov[2];
 225   // PERFORMANCE TEST LOOP
 226   for (cnt = 0; cnt < testIterations; ++cnt){
 227     // get the start time
 228     startTime = ACE_OS::gethrtime();
 229     if (!startTime)
 230       ACE_ERROR_RETURN((LM_ERROR,
 231                         "%p\n",
 232                         "ACE_OS::gethrtime()"),
 233                        0);
 234
 235
 236     ACE_CDR::ULong msgLenExpressed = ACE_HTONL(msgLen);
 237     iov[0].iov_base = reinterpret_cast<char *> (&msgLenExpressed);
 238     iov[0].iov_len = ACE_CDR::LONG_SIZE;
 239     iov[1].iov_base = reinterpret_cast<char *> (buf);
 240     iov[1].iov_len = msgLen;
 241
 242     if (-1 == stream.sendv_n (iov, 2))
 243       ACE_ERROR_RETURN ((LM_ERROR,
 244                          "%p\n",
 245                          "send_n"),
 246                         0);
 247
 248     // block for a Short reply
 249     ACE_CDR::Short reply;
 250     if ((stream.recv_n(&reply, ACE_CDR::SHORT_SIZE, 0, &bt)) == -1)
 251       ACE_ERROR_RETURN((LM_ERROR,
 252                         "%p\n",
 253                         "recv_n"),
 254                        0);
 255
 256     // get the end time
 257     endTime = ACE_OS::gethrtime();
 258     if (!endTime)
 259       ACE_ERROR_RETURN((LM_ERROR,
 260                         "%p\n",
 261                         "ACE_OS::gethrtime()"),
 262                        0);
 263
 264     // compute the message latency in micro-seconds
 265     messageLatency_usec =
 266
 267       (static_cast<double> (ACE_UINT64_DBLCAST_ADAPTER(endTime)) -
 268        static_cast<double> (ACE_UINT64_DBLCAST_ADAPTER(startTime)))
 269
 270       / microsec_clock_scale_factor;
 271
 272     // record the message latency in the histogram
 273     ACE_SCTP::record(messageLatency_usec, aceStream_hist);
 274   }
 275
 276   // THE HEADER MESSAGE SENT TO THE SERVER CONTAINED THE NUMBER OF
 277   // PRIMER AND TEST MESSAGES TO BE SENT AFTER WHICH THE SERVER WILL
 278   // CLOSE THE STREAM SO ONCE WE REACH THIS POINT THE STREAM IS NO
 279   // LONGER VALID AND WE CLOSE IT.
 280   stream.close();
 281
 282   // allocated by runTest
 283   delete[] buf;
 284   return aceStream_hist;
 285 }
 286
 287 // sends the test information to the server and calls the correct test
 288 // function.
 289 ACE_SCTP::HIST runTest(ACE_SOCK_Stream & stream)
 290 {
 291   size_t msgLen = 1;
 292   for (int i=1; i <= Options_Manager::payload_size_power_of_2; i++)
 293     msgLen *= 2;
 294
 295   // send a header to the server that contains test parameters
 296   if (sendHeader(stream) < 0)
 297     ACE_ERROR_RETURN((LM_ERROR,
 298                       "Could NOT Send CDR Encoded Header"),
 299                      0);
 300
 301   // create the forward message buffer
 302
 303   ACE_CDR::Octet *o_arr;
 304   ACE_CDR::ULong seqLen = msgLen/ACE_CDR::OCTET_SIZE;
 305   ACE_NEW_RETURN(o_arr,
 306                  ACE_CDR::Octet[seqLen],
 307                  0);
 308   // o_arr is deleted in run_octet_test
 309   return(runUnmarshalledOctetTest(o_arr, seqLen, stream));
 310 }