TAO/orbsvcs/tests/Notify/performance-tests/RedGreen/RedGreen_Test.cpp

   1 // -*- C++ -*-
   2
   3 #include "RedGreen_Test.h"
   4 #include "ace/Arg_Shifter.h"
   5 #include "ace/Get_Opt.h"
   6 #include "ace/OS_NS_unistd.h"
   7 #include "orbsvcs/Time_Utilities.h"
   8 #include "tao/debug.h"
   9
  10 #define NOTIFY_FACTORY_NAME "NotifyEventChannelFactory"
  11 #define NAMING_SERVICE_NAME "NameService"
  12
  13 #define DOMAIN_GREEN "DOMAIN_GREEN"
  14 #define DOMAIN_RED "DOMAIN_RED"
  15
  16 #define TYPE_GREEN "TYPE_GREEN"
  17 #define TYPE_RED "TYPE_RED"
  18
  19 ACE_Atomic_Op <TAO_SYNCH_MUTEX, int> g_result_count = 0;
  20 ACE_hrtime_t g_throughput_start_;
  21
  22 static bool consumer_is_done = false;
  23
  24 int
  25 RedGreen_Test::parse_args (int argc,
  26                            ACE_TCHAR *argv[])
  27 {
  28   ACE_Arg_Shifter arg_shifter (argc, argv);
  29
  30     const ACE_TCHAR *current_arg = 0;
  31
  32     while (arg_shifter.is_anything_left ())
  33     {
  34       if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-burst_size"))))
  35         {
  36           this->burst_size_ = ACE_OS::atoi (current_arg);
  37           ACE_DEBUG ((LM_DEBUG,
  38                       "Burst size = %d\n",
  39                       burst_size_));
  40           // The number of events to send/receive.
  41           arg_shifter.consume_arg ();
  42         }
  43       else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT ("-?")) == 0)
  44         {
  45           ACE_DEBUG((LM_DEBUG,
  46                      "usage: %s "
  47                      "-burst_size [count]\n",
  48                      argv[0],
  49                      argv[0]));
  50
  51           arg_shifter.consume_arg ();
  52
  53           return -1;
  54         }
  55       else
  56         {
  57           arg_shifter.ignore_arg ();
  58         }
  59     }
  60
  61   return 0;
  62 }
  63
  64 RedGreen_Test::RedGreen_Test ()
  65   : burst_size_ (10),
  66     nthreads_ (2)
  67 {
  68   // No-Op.
  69   ifgop_ = CosNotifyChannelAdmin::OR_OP;
  70 }
  71
  72 void
  73 RedGreen_Test::destroy_ec ()
  74 {
  75   if (!CORBA::is_nil (ec_.in ()))
  76     {
  77       // Even though we still have a reference, there's no guarantee
  78       // the EC is still around.  So, trap exceptions.
  79       try
  80         {
  81           this->ec_->destroy ();
  82         }
  83       catch (const CORBA::COMM_FAILURE&)
  84         {
  85           // Silently swallow this b/c this could mean the EC is gone
  86           // or that the network is hosed.  Either way, we're not waiting
  87           // around to figure out the problem.  Report the incident to the
  88           // log and be done with it.
  89           ACE_DEBUG ((LM_INFO,
  90                       "INFO: Got a COMM_FAILURE exception while trying to\n"
  91                       "      invoke `destroy()' on the Event Channel in the\n"
  92                       "      RedGreen destroy_ec. This is likely not a problem.\n"));
  93         }
  94       catch (const CORBA::Exception& ex)
  95         {
  96           ex._tao_print_exception ("in RedGreen destructor; swallowing.\n");
  97         }
  98
  99       this->ec_ = CosNotifyChannelAdmin::EventChannel::_nil ();
 100     }
 101 }
 102
 103
 104 void
 105 RedGreen_Test::init (int argc,
 106                      ACE_TCHAR *argv [])
 107 {
 108   this->init_ORB (argc,
 109                   argv);
 110   this->resolve_naming_service ();
 111   this->resolve_Notify_factory ();
 112   this->create_EC ();
 113   this->create_supplieradmin ();
 114   this->create_consumeradmin ();
 115   this->create_consumers ();
 116   this->create_suppliers ();
 117 }
 118
 119 void
 120 RedGreen_Test::run ()
 121 {
 122   try
 123   {
 124     this->send_events ();
 125
 126     while (! consumer_is_done)
 127     {
 128       ACE_Time_Value tv(0, 10 * 1000);
 129       this->orb_->run(tv);
 130     }
 131
 132     this->destroy_ec ();
 133     this->orb_->shutdown (false);
 134   }
 135   catch (const CORBA::Exception& ex)
 136   {
 137     ex._tao_print_exception ("Supplier:");
 138     throw;
 139   }
 140
 141   worker_.thr_mgr ()->wait ();
 142 }
 143
 144 void
 145 RedGreen_Test::done ()
 146 {
 147   dump_results ();
 148   worker_.done ();
 149 }
 150
 151 void
 152 RedGreen_Test::init_ORB (int argc, ACE_TCHAR *argv [])
 153 {
 154   this->orb_ = CORBA::ORB_init (argc, argv);
 155
 156   CORBA::Object_ptr poa_object  =
 157     this->orb_->resolve_initial_references("RootPOA");
 158
 159   if (CORBA::is_nil (poa_object))
 160     {
 161       ACE_ERROR ((LM_ERROR,
 162                   " (%P|%t) Unable to initialize the POA.\n"));
 163       return;
 164     }
 165   this->root_poa_ =
 166     PortableServer::POA::_narrow (poa_object);
 167
 168   PortableServer::POAManager_var poa_manager =
 169     root_poa_->the_POAManager ();
 170
 171   poa_manager->activate ();
 172
 173   worker_.orb (this->orb_.in ());
 174
 175   if (worker_.activate (THR_NEW_LWP | THR_JOINABLE,
 176                         this->nthreads_) != 0)
 177     {
 178       ACE_ERROR ((LM_ERROR,
 179                   "Cannot activate client threads\n"));
 180     }
 181 }
 182
 183 void
 184 RedGreen_Test::resolve_naming_service ()
 185 {
 186   CORBA::Object_var naming_obj =
 187     this->orb_->resolve_initial_references (NAMING_SERVICE_NAME);
 188
 189   // Need to check return value for errors.
 190   if (CORBA::is_nil (naming_obj.in ()))
 191     {
 192       throw CORBA::UNKNOWN ();
 193     }
 194
 195   this->naming_context_ =
 196     CosNaming::NamingContext::_narrow (naming_obj.in ());
 197 }
 198
 199 void
 200 RedGreen_Test::resolve_Notify_factory ()
 201 {
 202   CosNaming::Name name (1);
 203   name.length (1);
 204   name[0].id = CORBA::string_dup (NOTIFY_FACTORY_NAME);
 205
 206   CORBA::Object_var obj =
 207     this->naming_context_->resolve (name);
 208
 209   this->notify_factory_ =
 210     CosNotifyChannelAdmin::EventChannelFactory::_narrow (obj.in ());
 211 }
 212
 213 void
 214 RedGreen_Test::create_EC ()
 215 {
 216   CosNotifyChannelAdmin::ChannelID id;
 217
 218   this->ec_ = notify_factory_->create_channel (this->initial_qos_,
 219                                                this->initial_admin_,
 220                                                id);
 221
 222   ACE_ASSERT (!CORBA::is_nil (ec_.in ()));
 223 }
 224
 225 void
 226 RedGreen_Test::create_supplieradmin ()
 227 {
 228   CosNotifyChannelAdmin::AdminID adminid;
 229
 230   supplier_admin_ =
 231     ec_->new_for_suppliers (this->ifgop_,
 232                             adminid);
 233
 234   ACE_ASSERT (!CORBA::is_nil (supplier_admin_.in ()));
 235 }
 236
 237 void
 238 RedGreen_Test::create_consumeradmin ()
 239 {
 240   CosNotifyChannelAdmin::AdminID adminid;
 241
 242   consumer_admin_ =
 243     ec_->new_for_consumers (this->ifgop_,
 244                             adminid);
 245
 246   ACE_ASSERT (!CORBA::is_nil (consumer_admin_.in ()));
 247 }
 248
 249 void
 250 RedGreen_Test::create_consumers ()
 251 {
 252   ACE_NEW (this->normal_consumer_,
 253            RedGreen_Test_StructuredPushConsumer (this));
 254   this->normal_consumer_->connect (this->consumer_admin_.in ());
 255
 256   ACE_NEW (this->slow_consumer_,
 257            SlowConsumer (this));
 258   this->slow_consumer_->connect (this->consumer_admin_.in ());
 259 }
 260
 261 void
 262 RedGreen_Test::create_suppliers ()
 263 {
 264   ACE_NEW (this->supplier_,
 265            RedGreen_Test_StructuredPushSupplier ());
 266   this->supplier_->connect (this->supplier_admin_.in ());
 267 }
 268
 269 void
 270 RedGreen_Test::send_events ()
 271 {
 272   // Setup the Consumer 1 to receive
 273   //event_type : "DOMAIN_GREEN", "DOMAIN_GREEN".
 274   CosNotification::EventTypeSeq added_1(1);
 275   added_1.length (1);
 276   CosNotification::EventTypeSeq removed_1 (0);
 277   removed_1.length (0);
 278
 279   added_1[0].domain_name =  CORBA::string_dup (DOMAIN_GREEN);
 280   added_1[0].type_name = CORBA::string_dup (TYPE_GREEN);
 281
 282   this->normal_consumer_->get_proxy_supplier ()->subscription_change (added_1, removed_1);
 283
 284   // Setup the Consumer 2 to receive event_type : "DOMAIN_RED", "TYPE_RED"
 285   CosNotification::EventTypeSeq added_2(1);
 286   added_2.length (1);
 287   CosNotification::EventTypeSeq removed_2 (0);
 288   removed_2.length (0);
 289
 290   added_2[0].domain_name =  CORBA::string_dup (DOMAIN_RED);
 291   added_2[0].type_name = CORBA::string_dup (TYPE_RED);
 292
 293   this->slow_consumer_->get_proxy_supplier ()->subscription_change (added_2, removed_2);
 294
 295   // Create the events - one of each type
 296
 297   // Event 2
 298   CosNotification::StructuredEvent green_event;
 299   green_event.header.fixed_header.event_type.domain_name =
 300     CORBA::string_dup(DOMAIN_GREEN);
 301   green_event.header.fixed_header.event_type.type_name =
 302     CORBA::string_dup(TYPE_GREEN);
 303   green_event.header.fixed_header.event_name = CORBA::string_dup ("");
 304   green_event.header.variable_header.length (0); // put nothing here
 305   green_event.filterable_data.length (0);
 306   green_event.remainder_of_body <<= (CORBA::Long)10;
 307
 308   // event 3
 309   CosNotification::StructuredEvent red_event;
 310   red_event.header.fixed_header.event_type.domain_name =
 311     CORBA::string_dup(DOMAIN_RED);
 312   red_event.header.fixed_header.event_type.type_name =
 313     CORBA::string_dup(TYPE_RED);
 314   red_event.header.fixed_header.event_name = CORBA::string_dup("");
 315   red_event.header.variable_header.length (0); // put nothing here
 316   red_event.filterable_data.length (0);
 317   red_event.remainder_of_body <<= (CORBA::Long)10;
 318
 319   g_throughput_start_ = ACE_OS::gethrtime ();
 320
 321   // let supplier 1 send all these events
 322   for (int i = 0; i < this->burst_size_; ++i)
 323     {
 324       this->supplier_->send_event (red_event);
 325
 326       this->supplier_->send_event (green_event);
 327     }
 328 }
 329
 330
 331 void
 332 RedGreen_Test::dump_results ()
 333 {
 334   ACE_Throughput_Stats throughput;
 335   ACE_High_Res_Timer::global_scale_factor_type gsf =
 336     ACE_High_Res_Timer::global_scale_factor ();
 337   ACE_TCHAR buf[BUFSIZ];
 338
 339   ACE_OS::sprintf (buf,
 340                    ACE_TEXT ("Normal Consumer [%02d]"),
 341                    1);
 342   normal_consumer_->dump_stats (buf,
 343                                 gsf);
 344   normal_consumer_->accumulate_into (throughput);
 345
 346   ACE_OS::sprintf (buf,
 347                    ACE_TEXT ("Slow Consumer [%02d]"),
 348                    2);
 349   slow_consumer_->dump_stats (buf,
 350                               gsf);
 351   slow_consumer_->accumulate_into (throughput);
 352
 353   ACE_DEBUG ((LM_DEBUG,
 354               "\n"));
 355
 356   ACE_Throughput_Stats suppliers;
 357
 358   ACE_OS::sprintf (buf,
 359                    ACE_TEXT ("Supplier [%02d]"),
 360                    1);
 361
 362   this->supplier_->dump_stats (buf, gsf);
 363   this->supplier_->accumulate_into (suppliers);
 364
 365   ACE_DEBUG ((LM_DEBUG, "\nTotals:\n"));
 366   throughput.dump_results (ACE_TEXT("Notify_Consumer/totals"), gsf);
 367
 368   ACE_DEBUG ((LM_DEBUG, "\n"));
 369   suppliers.dump_results (ACE_TEXT("Notify_Supplier/totals"), gsf);
 370 }
 371
 372 // *****************************************************************
 373
 374 RedGreen_Test_StructuredPushConsumer::RedGreen_Test_StructuredPushConsumer (
 375     RedGreen_Test* RedGreen_Test
 376   )
 377   : RedGreen_Test_ (RedGreen_Test),
 378     push_count_ (0)
 379 {
 380 }
 381
 382 RedGreen_Test_StructuredPushConsumer::~RedGreen_Test_StructuredPushConsumer (
 383     void
 384   )
 385 {
 386 }
 387
 388 void
 389 RedGreen_Test_StructuredPushConsumer::accumulate_into (
 390     ACE_Throughput_Stats &throughput) const
 391 {
 392   throughput.accumulate (this->throughput_);
 393 }
 394
 395 void
 396 RedGreen_Test_StructuredPushConsumer::dump_stats (
 397   const ACE_TCHAR* msg,
 398   ACE_High_Res_Timer::global_scale_factor_type gsf)
 399 {
 400   this->throughput_.dump_results (msg, gsf);
 401 }
 402
 403 void
 404 RedGreen_Test_StructuredPushConsumer::connect (
 405     CosNotifyChannelAdmin::ConsumerAdmin_ptr consumer_admin
 406   )
 407 {
 408   // Activate the consumer with the default_POA_.
 409   CosNotifyComm::StructuredPushConsumer_var objref =
 410     this->_this ();
 411
 412   CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
 413     consumer_admin->obtain_notification_push_supplier (
 414         CosNotifyChannelAdmin::STRUCTURED_EVENT,
 415         proxy_supplier_id_);
 416
 417   ACE_ASSERT (!CORBA::is_nil (proxysupplier.in ()));
 418
 419   // narrow
 420   this->proxy_supplier_ =
 421     CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow (
 422         proxysupplier.in ());
 423
 424   ACE_ASSERT (!CORBA::is_nil (proxy_supplier_.in ()));
 425
 426   proxy_supplier_->connect_structured_push_consumer (objref.in ());
 427 }
 428
 429 void
 430 RedGreen_Test_StructuredPushConsumer::disconnect ()
 431 {
 432   this->proxy_supplier_->
 433     disconnect_structured_push_supplier ();
 434 }
 435
 436 void
 437 RedGreen_Test_StructuredPushConsumer::offer_change (
 438     const CosNotification::EventTypeSeq & /*added*/,
 439     const CosNotification::EventTypeSeq & /*removed*/
 440   )
 441 {
 442   // No-Op.
 443 }
 444
 445 void
 446 RedGreen_Test_StructuredPushConsumer::push_structured_event (
 447     const CosNotification::StructuredEvent & notification
 448   )
 449 {
 450   ACE_GUARD (TAO_SYNCH_MUTEX,
 451              ace_mon,
 452              this->lock_);
 453   const char* domain_name =
 454     notification.header.fixed_header.event_type.domain_name;
 455
 456   const char* type_name =
 457     notification.header.fixed_header.event_type.type_name;
 458
 459   if (TAO_debug_level)
 460     ACE_DEBUG ((LM_DEBUG,
 461                 "Consumer %d event, domain = %s, type =  %s\n",
 462                 this->proxy_supplier_id_,
 463                 domain_name,
 464                 type_name));
 465
 466   TimeBase::TimeT latency_base_recorded;
 467   ACE_hrtime_t latency_base;
 468
 469   notification.filterable_data[0].value >>= latency_base_recorded;
 470
 471   ORBSVCS_Time::TimeT_to_hrtime (latency_base,
 472                                  latency_base_recorded);
 473
 474   ++this->push_count_;
 475
 476   // Grab timestamp again.
 477   ACE_hrtime_t now = ACE_OS::gethrtime ();
 478
 479   // Record statistics.
 480   this->throughput_.sample (now - g_throughput_start_,
 481                             now - latency_base);
 482
 483
 484   if (++g_result_count == 2*RedGreen_Test_->burst_size_)
 485     {
 486       RedGreen_Test_->done ();
 487     }
 488 }
 489
 490 void
 491 RedGreen_Test_StructuredPushConsumer::disconnect_structured_push_consumer (
 492   )
 493 {
 494   // No-Op.
 495 }
 496
 497 CosNotifyChannelAdmin::StructuredProxyPushSupplier_ptr
 498 RedGreen_Test_StructuredPushConsumer::get_proxy_supplier ()
 499 {
 500   return this->proxy_supplier_.in ();
 501 }
 502
 503 // *****************************************************************
 504
 505 SlowConsumer::SlowConsumer (RedGreen_Test* RedGreen_Test)
 506   : RedGreen_Test_StructuredPushConsumer (RedGreen_Test)
 507 {
 508 }
 509
 510 void
 511 SlowConsumer::push_structured_event (
 512     const CosNotification::StructuredEvent & notification
 513   )
 514 {
 515   // Slow it down ...
 516   ACE_OS::sleep (1);
 517
 518   RedGreen_Test_StructuredPushConsumer::push_structured_event (notification);
 519 }
 520
 521 // *****************************************************************
 522
 523 RedGreen_Test_StructuredPushSupplier::RedGreen_Test_StructuredPushSupplier ()
 524 {
 525 }
 526
 527 RedGreen_Test_StructuredPushSupplier::~RedGreen_Test_StructuredPushSupplier ()
 528 {
 529 }
 530
 531 void
 532 RedGreen_Test_StructuredPushSupplier::accumulate_into (
 533     ACE_Throughput_Stats &throughput) const
 534 {
 535   throughput.accumulate (this->throughput_);
 536 }
 537
 538 void
 539 RedGreen_Test_StructuredPushSupplier::dump_stats (
 540   const ACE_TCHAR* msg,
 541   ACE_High_Res_Timer::global_scale_factor_type gsf)
 542 {
 543   this->throughput_.dump_results (msg, gsf);
 544 }
 545
 546 void
 547 RedGreen_Test_StructuredPushSupplier::connect (
 548     CosNotifyChannelAdmin::SupplierAdmin_ptr supplier_admin
 549   )
 550 {
 551   CosNotifyComm::StructuredPushSupplier_var objref =
 552     this->_this ();
 553
 554   CosNotifyChannelAdmin::ProxyConsumer_var proxyconsumer =
 555     supplier_admin->obtain_notification_push_consumer (
 556         CosNotifyChannelAdmin::STRUCTURED_EVENT,
 557         proxy_consumer_id_);
 558
 559   ACE_ASSERT (!CORBA::is_nil (proxyconsumer.in ()));
 560
 561   // narrow
 562   this->proxy_consumer_ =
 563     CosNotifyChannelAdmin::StructuredProxyPushConsumer::_narrow (
 564         proxyconsumer.in ());
 565
 566   ACE_ASSERT (!CORBA::is_nil (proxy_consumer_.in ()));
 567
 568   proxy_consumer_->connect_structured_push_supplier (objref.in ());
 569 }
 570
 571 void
 572 RedGreen_Test_StructuredPushSupplier::disconnect ()
 573 {
 574   ACE_ASSERT (!CORBA::is_nil (this->proxy_consumer_.in ()));
 575
 576   this->proxy_consumer_->disconnect_structured_push_consumer ();
 577 }
 578
 579 void
 580 RedGreen_Test_StructuredPushSupplier::subscription_change (
 581     const CosNotification::EventTypeSeq & /*added*/,
 582     const CosNotification::EventTypeSeq &) /*removed */
 583 {
 584   //No-Op.
 585 }
 586
 587 void
 588 RedGreen_Test_StructuredPushSupplier::send_event (
 589     CosNotification::StructuredEvent& event)
 590 {
 591   event.filterable_data.length (1);
 592   event.filterable_data[0].name = CORBA::string_dup("latency_base");
 593
 594   // Record current time.
 595   ACE_hrtime_t start = ACE_OS::gethrtime ();
 596
 597   TimeBase::TimeT latency_base;
 598   ORBSVCS_Time::hrtime_to_TimeT (latency_base,
 599                                  start);
 600   // any
 601   event.filterable_data[0].value <<= latency_base;
 602   proxy_consumer_->push_structured_event (event);
 603
 604   ACE_hrtime_t end = ACE_OS::gethrtime ();
 605
 606   this->throughput_.sample (end - g_throughput_start_,
 607                             end - start);
 608 }
 609
 610 void
 611 RedGreen_Test_StructuredPushSupplier::disconnect_structured_push_supplier ()
 612 {
 613   // No-Op.
 614 }
 615
 616 //*****************************************************************
 617
 618 Worker::Worker ()
 619 {
 620 }
 621
 622 void
 623 Worker::orb (CORBA::ORB_ptr orb)
 624 {
 625   orb_ = CORBA::ORB::_duplicate (orb);
 626 }
 627
 628 void
 629 Worker::done ()
 630 {
 631   consumer_is_done = true;
 632 }
 633
 634 int
 635 Worker::svc ()
 636 {
 637   try
 638   {
 639     this->orb_->run ();
 640   }
 641   catch (const CORBA::Exception& ex)
 642   {
 643     ex._tao_print_exception ("Consumer:");
 644   }
 645
 646   return 0;
 647 }