TAO/orbsvcs/tests/EC_Throughput/ECT_Throughput.cpp

   1 #include "ECT_Throughput.h"
   2
   3 #include "orbsvcs/Event_Utilities.h"
   4 #include "orbsvcs/Event_Service_Constants.h"
   5 #include "orbsvcs/Scheduler_Factory.h"
   6 #include "orbsvcs/Time_Utilities.h"
   7 #include "orbsvcs/Sched/Config_Scheduler.h"
   8 #include "orbsvcs/Runtime_Scheduler.h"
   9 #include "orbsvcs/Event/EC_Event_Channel.h"
  10 #include "orbsvcs/Event/EC_Default_Factory.h"
  11
  12 #include "tao/Timeprobe.h"
  13 #include "tao/debug.h"
  14
  15 #include "ace/Get_Opt.h"
  16 #include "ace/Auto_Ptr.h"
  17 #include "ace/Sched_Params.h"
  18 #include "ace/High_Res_Timer.h"
  19 #include "ace/OS_NS_strings.h"
  20 #include "ace/OS_NS_errno.h"
  21 #include "ace/OS_NS_unistd.h"
  22
  23 int
  24 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  25 {
  26   TAO_EC_Default_Factory::init_svcs ();
  27
  28   ECT_Throughput driver;
  29   return driver.run (argc, argv);
  30 }
  31
  32 // ****************************************************************
  33
  34 ECT_Throughput::ECT_Throughput (void)
  35   : n_consumers_ (1),
  36     n_suppliers_ (1),
  37     burst_count_ (10),
  38     burst_size_ (100),
  39     event_size_ (128),
  40     burst_pause_ (100),
  41     consumer_type_start_ (ACE_ES_EVENT_UNDEFINED),
  42     consumer_type_count_ (1),
  43     consumer_type_shift_ (0),
  44     supplier_type_start_ (ACE_ES_EVENT_UNDEFINED),
  45     supplier_type_count_ (1),
  46     supplier_type_shift_ (0),
  47     pid_file_name_ (0),
  48     active_count_ (0),
  49     ec_concurrency_hwm_ (1),
  50     thr_create_flags_ (THR_NEW_LWP|THR_BOUND|THR_SCHED_FIFO)
  51 {
  52 }
  53
  54 ECT_Throughput::~ECT_Throughput (void)
  55 {
  56 }
  57
  58 int
  59 ECT_Throughput::run (int argc, ACE_TCHAR* argv[])
  60 {
  61   try
  62     {
  63       // Calibrate the high resolution timer *before* starting the
  64       // test.
  65       ACE_High_Res_Timer::calibrate ();
  66
  67       this->orb_ =
  68         CORBA::ORB_init (argc, argv);
  69
  70       CORBA::Object_var poa_object =
  71         this->orb_->resolve_initial_references("RootPOA");
  72
  73       if (CORBA::is_nil (poa_object.in ()))
  74         ACE_ERROR_RETURN ((LM_ERROR,
  75                            " (%P|%t) Unable to initialize the POA.\n"),
  76                           1);
  77
  78       PortableServer::POA_var root_poa =
  79         PortableServer::POA::_narrow (poa_object.in ());
  80
  81       PortableServer::POAManager_var poa_manager =
  82         root_poa->the_POAManager ();
  83
  84       poa_manager->activate ();
  85
  86       if (this->parse_args (argc, argv))
  87         return 1;
  88
  89       if (TAO_debug_level > 0)
  90         {
  91           ACE_DEBUG ((LM_DEBUG,
  92                       "Execution parameters:\n"
  93                       "  consumers = <%d>\n"
  94                       "  suppliers = <%d>\n"
  95                       "  burst count = <%d>\n"
  96                       "  burst size = <%d>\n"
  97                       "  event size = <%d>\n"
  98                       "  burst pause = <%d>\n"
  99                       "  consumer type start = <%d>\n"
 100                       "  consumer type count = <%d>\n"
 101                       "  consumer type shift = <%d>\n"
 102                       "  supplier type start = <%d>\n"
 103                       "  supplier type count = <%d>\n"
 104                       "  supplier type shift = <%d>\n"
 105                       "  pid file name = <%s>\n"
 106                       "  concurrency HWM = <%d>\n",
 107
 108                       this->n_consumers_,
 109                       this->n_suppliers_,
 110                       this->burst_count_,
 111                       this->burst_size_,
 112                       this->event_size_,
 113                       this->burst_pause_,
 114                       this->consumer_type_start_,
 115                       this->consumer_type_count_,
 116                       this->consumer_type_shift_,
 117                       this->supplier_type_start_,
 118                       this->supplier_type_count_,
 119                       this->supplier_type_shift_,
 120
 121                       this->pid_file_name_?this->pid_file_name_:ACE_TEXT("nil"),
 122                       this->ec_concurrency_hwm_
 123                       ) );
 124         }
 125
 126       if (this->pid_file_name_ != 0)
 127         {
 128           FILE* pid = ACE_OS::fopen (this->pid_file_name_, "w");
 129           if (pid != 0)
 130             {
 131               ACE_OS::fprintf (pid, "%ld\n",
 132                                static_cast<long> (ACE_OS::getpid ()));
 133               ACE_OS::fclose (pid);
 134             }
 135         }
 136
 137       int priority =
 138         (ACE_Sched_Params::priority_min (ACE_SCHED_FIFO)
 139          + ACE_Sched_Params::priority_max (ACE_SCHED_FIFO)) / 2;
 140       priority = ACE_Sched_Params::next_priority (ACE_SCHED_FIFO,
 141                                                   priority);
 142       // Enable FIFO scheduling, e.g., RT scheduling class on Solaris.
 143
 144       if (ACE_OS::sched_params (ACE_Sched_Params (ACE_SCHED_FIFO,
 145                                                   priority,
 146                                                   ACE_SCOPE_PROCESS)) != 0)
 147         {
 148           if (ACE_OS::last_error () == EPERM)
 149             {
 150               ACE_DEBUG ((LM_DEBUG,
 151                           "%s: user is not superuser, "
 152                           "so remain in time-sharing class\n", argv[0]));
 153               this->thr_create_flags_ = THR_NEW_LWP;
 154             }
 155           else
 156             ACE_ERROR ((LM_ERROR,
 157                         "%s: ACE_OS::sched_params failed\n", argv[0]));
 158         }
 159
 160       if (ACE_OS::thr_setprio (priority) == -1)
 161         {
 162           ACE_ERROR ((LM_ERROR, "(%P|%t) main thr_setprio failed,"
 163                       "no real-time features\n"));
 164         }
 165
 166 #if 1
 167       ACE_Config_Scheduler scheduler_impl;
 168 #else
 169 #include "ECT_Scheduler_Info.h"
 170       ACE_Runtime_Scheduler scheduler_impl (
 171         runtime_configs_size,
 172         runtime_configs,
 173         runtime_infos_size,
 174         runtime_infos);
 175 #endif
 176       RtecScheduler::Scheduler_var scheduler =
 177         scheduler_impl._this ();
 178
 179 #if 0
 180       CORBA::Object_var naming_obj =
 181         this->orb_->resolve_initial_references ("NameService");
 182
 183       if (CORBA::is_nil (naming_obj.in ()))
 184         ACE_ERROR_RETURN ((LM_ERROR,
 185                            " (%P|%t) Unable to get the Naming Service.\n"),
 186                           1);
 187
 188       CosNaming::NamingContext_var naming_context =
 189         CosNaming::NamingContext::_narrow (naming_obj.in ());
 190
 191       // This is the name we (potentially) register the Scheduling
 192       // Service in the Naming Service.
 193       CosNaming::Name schedule_name (1);
 194       schedule_name.length (1);
 195       schedule_name[0].id = CORBA::string_dup ("ScheduleService");
 196
 197       CORBA::String_var str =
 198         this->orb_->object_to_string (scheduler.in ());
 199       ACE_DEBUG ((LM_DEBUG, "The (local) scheduler IOR is <%s>\n",
 200                   str.in ()));
 201
 202       // Register the servant with the Naming Context....
 203       naming_context->rebind (schedule_name, scheduler.in ());
 204
 205       ACE_Scheduler_Factory::use_config (naming_context.in ());
 206 #endif /* 0 */
 207
 208       auto_ptr<POA_RtecEventChannelAdmin::EventChannel> ec_impl;
 209
 210       TAO_EC_Event_Channel_Attributes attr (root_poa.in (),
 211                                             root_poa.in ());
 212
 213       TAO_EC_Event_Channel *ec =
 214         new TAO_EC_Event_Channel (attr);
 215
 216       ec->activate ();
 217
 218       auto_ptr<POA_RtecEventChannelAdmin::EventChannel> auto_ec_impl (ec);
 219       ec_impl = auto_ec_impl;
 220
 221       RtecEventChannelAdmin::EventChannel_var channel =
 222         ec_impl->_this ();
 223
 224       this->connect_consumers (scheduler.in (),
 225                                channel.in ());
 226
 227       ACE_DEBUG ((LM_DEBUG, "connected consumer(s)\n"));
 228
 229       this->connect_suppliers (scheduler.in (),
 230                                channel.in ());
 231
 232       ACE_DEBUG ((LM_DEBUG, "connected supplier(s)\n"));
 233
 234       this->activate_suppliers ();
 235
 236       ACE_DEBUG ((LM_DEBUG, "suppliers are active\n"));
 237
 238       // Wait for the supplier threads...
 239       if (ACE_Thread_Manager::instance ()->wait () == -1)
 240         {
 241           ACE_ERROR ((LM_ERROR, "Thread_Manager wait failed\n"));
 242           return 1;
 243         }
 244
 245       ACE_DEBUG ((LM_DEBUG, "suppliers finished\n"));
 246
 247       this->dump_results ();
 248
 249       this->disconnect_consumers ();
 250
 251       ACE_DEBUG ((LM_DEBUG, "consumers disconnected\n"));
 252
 253       this->disconnect_suppliers ();
 254
 255       ACE_DEBUG ((LM_DEBUG, "suppliers disconnected\n"));
 256
 257       channel->destroy ();
 258
 259       ACE_DEBUG ((LM_DEBUG, "channel destroyed\n"));
 260
 261       {
 262         // Deactivate the EC
 263         PortableServer::POA_var poa =
 264           ec_impl->_default_POA ();
 265         PortableServer::ObjectId_var id =
 266           poa->servant_to_id (ec_impl.get ());
 267         poa->deactivate_object (id.in ());
 268
 269         ACE_DEBUG ((LM_DEBUG, "EC deactivated\n"));
 270       }
 271
 272       {
 273         // Deactivate the Scheduler
 274         PortableServer::POA_var poa =
 275           scheduler_impl._default_POA ();
 276         PortableServer::ObjectId_var id =
 277           poa->servant_to_id (&scheduler_impl);
 278         poa->deactivate_object (id.in ());
 279
 280         ACE_DEBUG ((LM_DEBUG, "scheduler deactivated\n"));
 281       }
 282     }
 283   catch (const CORBA::Exception& ex)
 284     {
 285       ex._tao_print_exception ("ECT_Throughput::run");
 286     }
 287   catch (...)
 288     {
 289       ACE_ERROR ((LM_ERROR, "non-corba exception raised\n"));
 290     }
 291
 292   return 0;
 293 }
 294
 295 void
 296 ECT_Throughput::shutdown_consumer (void*)
 297 {
 298   // int ID =
 299   //   (reinterpret_cast<Test_Consumer**> (consumer_cookie)
 300   //    - this->consumers_);
 301   //
 302   // ACE_DEBUG ((LM_DEBUG, "(%t) events received by consumer %d\n", ID));
 303
 304   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 305   this->active_count_--;
 306   if (this->active_count_ <= 0)
 307     {
 308       ACE_DEBUG ((LM_DEBUG,
 309                   "(%t) shutting down the ORB\n"));
 310       // Not needed: this->orb_->shutdown (0);
 311     }
 312 }
 313
 314 void
 315 ECT_Throughput::connect_consumers
 316      (RtecScheduler::Scheduler_ptr scheduler,
 317       RtecEventChannelAdmin::EventChannel_ptr channel)
 318 {
 319   {
 320     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 321     this->active_count_ = this->n_consumers_;
 322   }
 323   for (int i = 0; i < this->n_consumers_; ++i)
 324     {
 325       char buf[BUFSIZ];
 326       ACE_OS::sprintf (buf, "consumer_%02d", i);
 327
 328       ACE_NEW (this->consumers_[i],
 329                Test_Consumer (this,
 330                               this->consumers_ + i,
 331                               this->n_suppliers_));
 332
 333       int start = this->consumer_type_start_
 334         + i * this->consumer_type_shift_;
 335
 336       this->consumers_[i]->connect (scheduler,
 337                                     buf,
 338                                     start,
 339                                     this->consumer_type_count_,
 340                                     channel);
 341     }
 342 }
 343
 344 void
 345 ECT_Throughput::connect_suppliers
 346      (RtecScheduler::Scheduler_ptr scheduler,
 347       RtecEventChannelAdmin::EventChannel_ptr channel)
 348 {
 349   for (int i = 0; i < this->n_suppliers_; ++i)
 350     {
 351       char buf[BUFSIZ];
 352       ACE_OS::sprintf (buf, "supplier_%02d", i);
 353
 354       ACE_NEW (this->suppliers_[i], Test_Supplier (this));
 355
 356       int start = this->supplier_type_start_ + i*this->supplier_type_shift_;
 357       this->suppliers_[i]->connect (scheduler,
 358                                     buf,
 359                                     this->burst_count_,
 360                                     this->burst_size_,
 361                                     this->event_size_,
 362                                     this->burst_pause_,
 363                                     start,
 364                                     this->supplier_type_count_,
 365                                     channel);
 366     }
 367 }
 368
 369 void
 370 ECT_Throughput::activate_suppliers (void)
 371 {
 372   int priority =
 373     (ACE_Sched_Params::priority_min (ACE_SCHED_FIFO)
 374      + ACE_Sched_Params::priority_max (ACE_SCHED_FIFO)) / 2;
 375
 376   for (int i = 0; i < this->n_suppliers_; ++i)
 377     {
 378       if (this->suppliers_[i]->activate (this->thr_create_flags_,
 379                                          1, 0, priority) == -1)
 380         {
 381           ACE_ERROR ((LM_ERROR,
 382                       "Cannot activate thread for supplier %d\n",
 383                       i));
 384         }
 385     }
 386 }
 387
 388 void
 389 ECT_Throughput::disconnect_suppliers (void)
 390 {
 391   for (int i = 0; i < this->n_suppliers_; ++i)
 392     {
 393       this->suppliers_[i]->disconnect ();
 394     }
 395 }
 396
 397 void
 398 ECT_Throughput::disconnect_consumers (void)
 399 {
 400   for (int i = 0; i < this->n_consumers_; ++i)
 401     {
 402       this->consumers_[i]->disconnect ();
 403     }
 404 }
 405
 406 void
 407 ECT_Throughput::dump_results (void)
 408 {
 409   ACE_High_Res_Timer::global_scale_factor_type gsf =
 410     ACE_High_Res_Timer::global_scale_factor ();
 411
 412   ACE_Throughput_Stats consumers;
 413   for (int j = 0; j < this->n_consumers_; ++j)
 414     {
 415       ACE_TCHAR buf[BUFSIZ];
 416       ACE_OS::sprintf (buf, ACE_TEXT("consumer_%02d"), j);
 417
 418       this->consumers_[j]->dump_results (buf, gsf);
 419       this->consumers_[j]->accumulate (consumers);
 420     }
 421   consumers.dump_results (ACE_TEXT("ECT_Consumer/totals"), gsf);
 422
 423   ACE_Throughput_Stats suppliers;
 424   for (int i = 0; i < this->n_suppliers_; ++i)
 425     {
 426       ACE_TCHAR buf[BUFSIZ];
 427       ACE_OS::sprintf (buf, ACE_TEXT("supplier_%02d"), i);
 428
 429       this->suppliers_[i]->dump_results (buf, gsf);
 430       this->suppliers_[i]->accumulate (suppliers);
 431     }
 432   suppliers.dump_results (ACE_TEXT("ECT_Supplier/totals"), gsf);
 433 }
 434
 435 int
 436 ECT_Throughput::parse_args (int argc, ACE_TCHAR *argv [])
 437 {
 438   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT("dc:s:u:n:t:b:h:l:p:w:"));
 439   int opt;
 440
 441   while ((opt = get_opt ()) != EOF)
 442     {
 443       switch (opt)
 444         {
 445         case 'c':
 446           this->n_consumers_ = ACE_OS::atoi (get_opt.opt_arg ());
 447           break;
 448
 449         case 's':
 450           this->n_suppliers_ = ACE_OS::atoi (get_opt.opt_arg ());
 451           break;
 452
 453         case 'u':
 454           this->burst_count_ = ACE_OS::atoi (get_opt.opt_arg ());
 455           break;
 456
 457         case 'n':
 458           this->burst_size_ = ACE_OS::atoi (get_opt.opt_arg ());
 459           break;
 460
 461         case 'b':
 462           this->event_size_ = ACE_OS::atoi (get_opt.opt_arg ());
 463           break;
 464
 465         case 't':
 466           this->burst_pause_ = ACE_OS::atoi (get_opt.opt_arg ());
 467           break;
 468
 469         case 'h':
 470           {
 471             char* aux;
 472                 char* arg = ACE_OS::strtok_r (ACE_TEXT_ALWAYS_CHAR(get_opt.opt_arg ()), ",", &aux);
 473
 474             this->consumer_type_start_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
 475                 arg = ACE_OS::strtok_r (0, ",", &aux);
 476             this->consumer_type_count_ = ACE_OS::atoi (arg);
 477                 arg = ACE_OS::strtok_r (0, ",", &aux);
 478             this->consumer_type_shift_ = ACE_OS::atoi (arg);
 479           }
 480           break;
 481
 482         case 'l':
 483           {
 484             char* aux;
 485                 char* arg = ACE_OS::strtok_r (ACE_TEXT_ALWAYS_CHAR(get_opt.opt_arg ()), ",", &aux);
 486
 487             this->supplier_type_start_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
 488                 arg = ACE_OS::strtok_r (0, ",", &aux);
 489             this->supplier_type_count_ = ACE_OS::atoi (arg);
 490                 arg = ACE_OS::strtok_r (0, ",", &aux);
 491             this->supplier_type_shift_ = ACE_OS::atoi (arg);
 492           }
 493           break;
 494
 495         case 'p':
 496           this->pid_file_name_ = get_opt.opt_arg ();
 497           break;
 498
 499         case 'w':
 500           this->ec_concurrency_hwm_ = ACE_OS::atoi (get_opt.opt_arg ());
 501           break;
 502
 503         case '?':
 504         default:
 505           ACE_DEBUG ((LM_DEBUG,
 506                       "Usage: %s "
 507                       "[ORB options] "
 508                       "-r -d -x "
 509                       "-c <n_consumers> "
 510                       "-s <n_suppliers> "
 511                       "-u <burst count> "
 512                       "-n <burst size> "
 513                       "-b <event payload size> "
 514                       "-t <burst pause (usecs)> "
 515                       "-h <consumer_start,consumer_count,consumer_shift> "
 516                       "-l <supplier_start,supplier_count,supplier_shift> "
 517                       "-p <pid file name> "
 518                       "-w <concurrency HWM> "
 519                       "-r "
 520                       "\n",
 521                       argv[0]));
 522           return -1;
 523         }
 524     }
 525
 526   if (this->burst_count_ <= 0)
 527     {
 528       ACE_DEBUG ((LM_DEBUG,
 529                   "%s: burst count (%d) is out of range, "
 530                   "reset to default (%d)\n",
 531                   argv[0], this->burst_count_,
 532                   100));
 533       this->burst_count_ = 100;
 534     }
 535
 536   if (this->burst_size_ <= 0)
 537     {
 538       ACE_DEBUG ((LM_DEBUG,
 539                   "%s: burst size (%d) is out of range, "
 540                   "reset to default (%d)\n",
 541                   argv[0], this->burst_size_,
 542                   10));
 543       this->burst_size_ = 10;
 544     }
 545
 546   if (this->event_size_ < 0)
 547     {
 548       ACE_DEBUG ((LM_DEBUG,
 549                   "%s: event size (%d) is out of range, "
 550                   "reseting to default (%d)\n",
 551                   argv[0], this->event_size_,
 552                   128));
 553       this->event_size_ = 128;
 554     }
 555
 556   if (this->n_consumers_ < 0
 557       || this->n_consumers_ >= ECT_Throughput::MAX_CONSUMERS)
 558     {
 559       this->n_consumers_ = 1;
 560       ACE_ERROR_RETURN ((LM_ERROR,
 561                          "%s: number of consumers or "
 562                          "suppliers out of range, "
 563                          "reset to default (%d)\n",
 564                          argv[0], 1), -1);
 565     }
 566
 567   if (this->n_suppliers_ < 0
 568       || this->n_suppliers_ >= ECT_Throughput::MAX_SUPPLIERS)
 569     {
 570       this->n_suppliers_ = 1;
 571       ACE_ERROR_RETURN ((LM_ERROR,
 572                          "%s: number of suppliers out of range, "
 573                          "reset to default (%d)\n",
 574                          argv[0], 1), -1);
 575     }
 576
 577   if (this->n_suppliers_ == 0 && this->n_consumers_ == 0)
 578     {
 579       this->n_suppliers_ = 1;
 580       this->n_consumers_ = 1;
 581       ACE_ERROR_RETURN ((LM_ERROR,
 582                          "%s: no suppliers or consumers, "
 583                          "reset to default (%d of each)\n",
 584                          argv[0], 1), -1);
 585     }
 586
 587   if (this->ec_concurrency_hwm_ <= 0)
 588     {
 589       this->ec_concurrency_hwm_ = 1;
 590       ACE_ERROR_RETURN ((LM_ERROR,
 591                          "%s: invalid concurrency HWM, "
 592                          "reset to default (%d)\n",
 593                          argv[0], 1), -1);
 594     }
 595
 596   return 0;
 597 }