ACE/performance-tests/SCTP/SOCK_SEQPACK_clt.cpp

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