TAO/performance-tests/Protocols/receiver.cpp

   1 #include "ace/Get_Opt.h"
   2 #include "ace/OS_NS_stdio.h"
   3 #include "ace/OS_NS_string.h"
   4 #include "ace/OS_NS_unistd.h"
   5 #include "ace/Sample_History.h"
   6 #include "ace/High_Res_Timer.h"
   7 #include "ace/Stats.h"
   8 #include "ace/Throughput_Stats.h"
   9 #include "tao/debug.h"
  10 #include "testS.h"
  11
  12 static const ACE_TCHAR *ior_file = ACE_TEXT ("receiver.ior");
  13 static int do_dump_history = 0;
  14 static int print_missed_invocations = 0;
  15 static ACE_High_Res_Timer::global_scale_factor_type gsf = 0;
  16
  17 static int
  18 parse_args (int argc, ACE_TCHAR **argv)
  19 {
  20   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("d:f:m:"));
  21   int c;
  22
  23   while ((c = get_opts ()) != -1)
  24     switch (c)
  25       {
  26       case 'd':
  27         do_dump_history = ACE_OS::atoi (get_opts.opt_arg ());
  28         break;
  29
  30       case 'f':
  31         ior_file = get_opts.opt_arg ();
  32         break;
  33
  34       case 'm':
  35         print_missed_invocations = ACE_OS::atoi (get_opts.opt_arg ());
  36         break;
  37
  38       default:
  39         ACE_ERROR_RETURN ((LM_ERROR,
  40                            "usage: %s\n"
  41                            "\t-d <show history> (defaults to %d)\n"
  42                            "\t-f <ior_file> (defaults to %s)\n"
  43                            "\t-m <print missed invocations for paced workers> (defaults to %d)\n"
  44                            "\n",
  45                            argv[0],
  46                            do_dump_history,
  47                            ior_file,
  48                            print_missed_invocations),
  49                           -1);
  50       }
  51
  52   return 0;
  53 }
  54
  55 class test_i :
  56   public POA_test
  57 {
  58 public:
  59   test_i (CORBA::ORB_ptr orb,
  60           PortableServer::POA_ptr poa);
  61
  62   ~test_i ();
  63
  64   void start_test (CORBA::Long session_id,
  65                    const char *protocol,
  66                    CORBA::ULong invocation_rate,
  67                    CORBA::ULong message_size,
  68                    CORBA::ULong iterations);
  69
  70   void end_test ();
  71
  72   void oneway_sync ();
  73
  74   void twoway_sync ();
  75
  76   void oneway_method (CORBA::Long session_id,
  77                       CORBA::ULong iteration,
  78                       const ::test::octets &payload);
  79
  80   void twoway_method (CORBA::Long &session_id,
  81                       CORBA::ULong &iteration,
  82                       ::test::octets &payload);
  83
  84   //FUZZ: disable check_for_lack_ACE_OS
  85   void shutdown ();
  86   //FUZZ: enable check_for_lack_ACE_OS
  87
  88   PortableServer::POA_ptr _default_POA ();
  89
  90 private:
  91   CORBA::ORB_var orb_;
  92   PortableServer::POA_var poa_;
  93   typedef ACE_Array_Base<CORBA::Boolean> Invocations;
  94   Invocations invocations_received_;
  95   ACE_hrtime_t time_of_last_call_;
  96   ACE_hrtime_t test_start_;
  97   ACE_hrtime_t test_end_;
  98   CORBA::Boolean first_invocation_;
  99   ACE_Sample_History *inter_arrival_times_;
 100   CORBA::ULong iterations_;
 101   CORBA::ULong number_of_invocations_received_;
 102   CORBA::Long session_id_;
 103 };
 104
 105 test_i::test_i (CORBA::ORB_ptr orb,
 106                 PortableServer::POA_ptr poa)
 107   : orb_ (CORBA::ORB::_duplicate (orb)),
 108     poa_ (PortableServer::POA::_duplicate (poa)),
 109     inter_arrival_times_ (0),
 110     iterations_ (0),
 111     session_id_ (-1)
 112 {
 113 }
 114
 115 test_i::~test_i ()
 116 {
 117 }
 118
 119 void
 120 test_i::start_test (CORBA::Long session_id,
 121                     const char *protocol,
 122                     CORBA::ULong invocation_rate,
 123                     CORBA::ULong message_size,
 124                     CORBA::ULong iterations)
 125 {
 126   if (TAO_debug_level > 0)
 127     {
 128       ACE_DEBUG ((LM_DEBUG,
 129                   "Session id starts %d\n",
 130                   session_id));
 131     }
 132
 133   ACE_DEBUG ((LM_DEBUG,
 134               "Protocol = %5s Invocation Rate = %3d Message Size = %5d Expected Latency = %4d ",
 135               ACE_TEXT_CHAR_TO_TCHAR (protocol),
 136               invocation_rate,
 137               message_size,
 138               1000 / invocation_rate));
 139
 140   // Remember test parameters.
 141   this->session_id_ = session_id;
 142   this->iterations_ = iterations;
 143   this->number_of_invocations_received_ = 0;
 144
 145   //
 146   // Initialize counters and tables.
 147   //
 148   this->inter_arrival_times_ =
 149     new ACE_Sample_History (iterations);
 150
 151   this->first_invocation_ = 1;
 152
 153   this->invocations_received_.size (iterations);
 154   for (CORBA::ULong i = 0;
 155        i < iterations;
 156        ++i)
 157     this->invocations_received_[i] = 0;
 158
 159   // Record start time.
 160   this->test_start_ =
 161     ACE_OS::gethrtime ();
 162 }
 163
 164 void
 165 test_i::end_test ()
 166 {
 167   // Record end time.
 168   this->test_end_ =
 169     ACE_OS::gethrtime ();
 170
 171   if (do_dump_history)
 172     {
 173       this->inter_arrival_times_->dump_samples (ACE_TEXT("Inter-arrival times"), gsf);
 174     }
 175
 176   ACE_Basic_Stats stats;
 177   this->inter_arrival_times_->collect_basic_stats (stats);
 178
 179   ACE_DEBUG ((LM_DEBUG,
 180               "Max Latency = %6d ",
 181               stats.max_ / gsf / 1000));
 182
 183   ACE_DEBUG ((LM_DEBUG,
 184               "Invocations expected / received / missed / missed %% = %6d / %6d / %6d / %5.2f\n",
 185               this->iterations_,
 186               this->number_of_invocations_received_,
 187               this->iterations_ - this->number_of_invocations_received_,
 188               (this->iterations_ - this->number_of_invocations_received_) / (double) this->iterations_ * 100));
 189
 190   if (print_missed_invocations)
 191     {
 192       ACE_DEBUG ((LM_DEBUG, "\nFollowing invocations were never received:\n"));
 193
 194       for (CORBA::ULong i = 0;
 195            i < this->iterations_;
 196            ++i)
 197         {
 198           if (this->invocations_received_[i] == 0)
 199             {
 200               ACE_DEBUG ((LM_DEBUG,
 201                           "%d ",
 202                           i));
 203             }
 204         }
 205
 206       ACE_DEBUG ((LM_DEBUG, "\n"));
 207     }
 208
 209   if (TAO_debug_level > 0)
 210     {
 211       ACE_DEBUG ((LM_DEBUG,
 212                   "Session id ends %d\n",
 213                   this->session_id_));
 214
 215       stats.dump_results (ACE_TEXT("Inter-arrival times"), gsf);
 216
 217       ACE_Throughput_Stats::dump_throughput (ACE_TEXT("Inter-arrival times"),
 218                                              gsf,
 219                                              this->test_end_ - this->test_start_,
 220                                              stats.samples_count ());
 221     }
 222
 223   this->session_id_ = -1;
 224
 225   delete this->inter_arrival_times_;
 226 }
 227
 228 void
 229 test_i::oneway_sync ()
 230 {
 231 }
 232
 233 void
 234 test_i::twoway_sync ()
 235 {
 236 }
 237
 238 void
 239 test_i::oneway_method (CORBA::Long session_id,
 240                        CORBA::ULong iteration,
 241                        const ::test::octets &payload)
 242 {
 243   if (this->session_id_ != session_id)
 244     {
 245       ACE_DEBUG ((LM_DEBUG,
 246                   "Late message with iteration id = %d: will not count message\n",
 247                   iteration));
 248       return;
 249     }
 250
 251   if (TAO_debug_level > 0)
 252     {
 253       ACE_DEBUG ((LM_DEBUG,
 254                   "test_i::oneway_method -> session id = %d iteration = %d payload size = %d\n",
 255                   session_id,
 256                   iteration,
 257                   payload.length ()));
 258     }
 259
 260   this->invocations_received_[iteration] = 1;
 261   ++this->number_of_invocations_received_;
 262
 263   ACE_hrtime_t time_of_current_call =
 264     ACE_OS::gethrtime ();
 265
 266   if (this->first_invocation_)
 267     this->first_invocation_ = 0;
 268   else
 269     this->inter_arrival_times_->sample (time_of_current_call - this->time_of_last_call_);
 270
 271   this->time_of_last_call_ = time_of_current_call;
 272 }
 273
 274 void
 275 test_i::twoway_method (CORBA::Long &session_id,
 276                        CORBA::ULong &iteration,
 277                        ::test::octets &payload)
 278 {
 279   if (this->session_id_ != session_id)
 280     {
 281       ACE_DEBUG ((LM_DEBUG,
 282                   "Late message with iteration id = %d: will not count message\n",
 283                   iteration));
 284       return;
 285     }
 286
 287   if (TAO_debug_level > 0)
 288     {
 289       ACE_DEBUG ((LM_DEBUG,
 290                   "test_i::twoway_method -> session id = %d iteration = %d payload size = %d\n",
 291                   session_id,
 292                   iteration,
 293                   payload.length ()));
 294     }
 295
 296   this->invocations_received_[iteration] = 1;
 297   ++this->number_of_invocations_received_;
 298
 299   ACE_hrtime_t time_of_current_call =
 300     ACE_OS::gethrtime ();
 301
 302   if (this->first_invocation_)
 303     this->first_invocation_ = 0;
 304   else
 305     this->inter_arrival_times_->sample (time_of_current_call - this->time_of_last_call_);
 306
 307   this->time_of_last_call_ = time_of_current_call;
 308 }
 309
 310 PortableServer::POA_ptr
 311 test_i::_default_POA ()
 312 {
 313   return PortableServer::POA::_duplicate (this->poa_.in ());
 314 }
 315
 316 void
 317 test_i::shutdown ()
 318 {
 319   ACE_DEBUG ((LM_DEBUG,
 320               "test_i::shutdown\n"));
 321
 322   this->orb_->shutdown (false);
 323 }
 324
 325 int
 326 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 327 {
 328   gsf = ACE_High_Res_Timer::global_scale_factor ();
 329
 330   try
 331     {
 332       CORBA::ORB_var orb =
 333         CORBA::ORB_init (argc, argv);
 334
 335       int parse_args_result =
 336         parse_args (argc, argv);
 337       if (parse_args_result != 0)
 338         return parse_args_result;
 339
 340       CORBA::Object_var object =
 341         orb->resolve_initial_references ("RootPOA");
 342
 343       PortableServer::POA_var root_poa =
 344         PortableServer::POA::_narrow (object.in ());
 345
 346       PortableServer::POAManager_var poa_manager =
 347         root_poa->the_POAManager ();
 348
 349       test_i *servant =
 350         new test_i (orb.in (),
 351         root_poa.in ());
 352
 353       PortableServer::ServantBase_var safe_servant (servant);
 354
 355       test_var test =
 356         servant->_this ();
 357
 358       CORBA::String_var ior =
 359         orb->object_to_string (test.in ());
 360
 361       FILE *output_file =
 362         ACE_OS::fopen (ior_file, "w");
 363       ACE_ASSERT (output_file != 0);
 364
 365       u_int result =
 366         ACE_OS::fprintf (output_file,
 367                          "%s",
 368                          ior.in ());
 369       ACE_ASSERT (result == ACE_OS::strlen (ior.in ()));
 370       ACE_UNUSED_ARG (result);
 371
 372       ACE_OS::fclose (output_file);
 373
 374       poa_manager->activate ();
 375
 376       orb->run ();
 377       ACE_OS::sleep(1);
 378     }
 379   catch (const CORBA::Exception& ex)
 380     {
 381       ex._tao_print_exception ("Exception caught");
 382       return -1;
 383     }
 384
 385   return 0;
 386 }