TAO/orbsvcs/tests/Event/Basic/Observer.cpp

   1 #include "Observer.h"
   2 #include "Consumer.h"
   3 #include "Supplier.h"
   4 #include "orbsvcs/Event/EC_Event_Channel.h"
   5 #include "orbsvcs/Event/EC_Default_Factory.h"
   6 #include "ace/Arg_Shifter.h"
   7 #include "ace/High_Res_Timer.h"
   8
   9 int
  10 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  11 {
  12   TAO_EC_Default_Factory::init_svcs ();
  13   EC_Master master;
  14   return master.run (argc, argv);
  15 }
  16
  17 // ****************************************************************
  18
  19 EC_Master::EC_Master (void)
  20   :  seed_ (0),
  21      n_channels_ (4),
  22      channels_ (0)
  23 {
  24 }
  25
  26 EC_Master::~EC_Master (void)
  27 {
  28   if (this->channels_ != 0)
  29     {
  30       for (int i = 0; i < this->n_channels_; ++i)
  31         delete this->channels_[i];
  32       delete[] this->channels_;
  33     }
  34 }
  35
  36 int
  37 EC_Master::run (int argc, ACE_TCHAR* argv[])
  38 {
  39   try
  40     {
  41       // Calibrate the high resolution timer *before* starting the
  42       // test.
  43       ACE_High_Res_Timer::calibrate ();
  44
  45       this->seed_ = static_cast<unsigned int> (ACE_OS::time (0));
  46
  47       this->initialize_orb_and_poa (argc, argv);
  48
  49       if (this->parse_args (argc, argv))
  50         return 1;
  51
  52       ACE_DEBUG ((LM_DEBUG,
  53                   "The seed value is %d\n", this->seed_));
  54
  55       ACE_NEW_RETURN (this->channels_,
  56                       EC_Observer*[this->n_channels_],
  57                       1);
  58
  59       {
  60         for (int i = 0; i != this->n_channels_; ++i)
  61           {
  62             ACE_OS::rand_r (&this->seed_);
  63             ACE_NEW_RETURN (this->channels_[i],
  64                             EC_Observer (this,
  65                                          this->seed_,
  66                                          this->orb_.in (),
  67                                          this->root_poa_.in (),
  68                                          i),
  69                             1);
  70           }
  71       }
  72
  73       {
  74         ACE_TCHAR** targv;
  75         ACE_NEW_RETURN (targv, ACE_TCHAR*[argc], 1);
  76
  77         for (int i = 0; i != this->n_channels_; ++i)
  78           {
  79             int targc = argc;
  80             for (int j = 0; j < targc; ++j)
  81               targv[j] = argv[j];
  82             this->channels_[i]->run_init (targc, targv);
  83           }
  84         delete[] targv;
  85       }
  86
  87       {
  88         for (int i = 0; i != this->n_channels_; ++i)
  89           {
  90             this->channels_[i]->execute_test ();
  91           }
  92       }
  93
  94       if (ACE_Thread_Manager::instance ()->wait () == -1)
  95         {
  96           ACE_ERROR ((LM_ERROR,
  97                       "EC_Master (%P|%t) thread manager wait failed\n"));
  98           return 1;
  99         }
 100
 101       {
 102         for (int i = 0; i != this->n_channels_; ++i)
 103           {
 104             this->channels_[i]->dump_results ();
 105           }
 106       }
 107
 108       {
 109         for (int i = 0; i != this->n_channels_; ++i)
 110           {
 111             this->channels_[i]->run_cleanup ();
 112           }
 113       }
 114
 115       {
 116         for (int i = 0; i != this->n_channels_; ++i)
 117           {
 118             this->channels_[i]->disconnect_clients ();
 119             this->channels_[i]->shutdown_clients ();
 120             this->channels_[i]->destroy_ec ();
 121             this->channels_[i]->deactivate_ec ();
 122             this->channels_[i]->cleanup_tasks ();
 123             this->channels_[i]->cleanup_suppliers ();
 124             this->channels_[i]->cleanup_consumers ();
 125             this->channels_[i]->cleanup_ec ();
 126           }
 127       }
 128
 129       this->root_poa_->destroy (1,
 130                                 1);
 131
 132       this->orb_->destroy ();
 133    }
 134   catch (const CORBA::Exception& ex)
 135     {
 136       ex._tao_print_exception ("EC_Driver::run");
 137     }
 138   catch (...)
 139     {
 140       ACE_ERROR ((LM_ERROR, "EC_Driver (%P|%t) non-corba exception raised\n"));
 141     }
 142   return 0;
 143 }
 144
 145 void
 146 EC_Master::initialize_orb_and_poa (int &argc, ACE_TCHAR* argv[])
 147 {
 148   this->orb_ =
 149     CORBA::ORB_init (argc, argv);
 150
 151   CORBA::Object_var poa_object =
 152     this->orb_->resolve_initial_references("RootPOA");
 153
 154   if (CORBA::is_nil (poa_object.in ()))
 155     {
 156       ACE_ERROR ((LM_ERROR,
 157                   "EC_Driver (%P|%t) Unable to initialize the POA.\n"));
 158       return;
 159     }
 160
 161   this->root_poa_ =
 162     PortableServer::POA::_narrow (poa_object.in ());
 163
 164   PortableServer::POAManager_var poa_manager =
 165     this->root_poa_->the_POAManager ();
 166
 167   poa_manager->activate ();
 168 }
 169
 170 int
 171 EC_Master::parse_args (int &argc, ACE_TCHAR *argv [])
 172 {
 173   ACE_Arg_Shifter arg_shifter (argc, argv);
 174
 175   while (arg_shifter.is_anything_left ())
 176     {
 177       const ACE_TCHAR *arg = arg_shifter.get_current ();
 178
 179       if (ACE_OS::strcmp (arg, ACE_TEXT("-channels")) == 0)
 180         {
 181           arg_shifter.consume_arg ();
 182           this->n_channels_ = ACE_OS::atoi (arg_shifter.get_current ());
 183         }
 184       else if (ACE_OS::strcmp (arg, ACE_TEXT("-seed")) == 0)
 185         {
 186           arg_shifter.consume_arg ();
 187           this->seed_ = ACE_OS::atoi (arg_shifter.get_current ());
 188         }
 189
 190       arg_shifter.ignore_arg ();
 191     }
 192   return 0;
 193 }
 194
 195 int
 196 EC_Master::channel_count (void) const
 197 {
 198   return this->n_channels_;
 199 }
 200
 201 EC_Observer*
 202 EC_Master::channel (int i) const
 203 {
 204   return this->channels_[i];
 205 }
 206
 207 // ****************************************************************
 208
 209 EC_Observer::EC_Observer (EC_Master *master,
 210                           unsigned int seed,
 211                           CORBA::ORB_ptr orb,
 212                           PortableServer::POA_ptr root_poa,
 213                           int id)
 214   : master_ (master),
 215     seed_ (seed),
 216     id_ (id),
 217     gwys_ (0)
 218 {
 219   this->orb_ = CORBA::ORB::_duplicate (orb);
 220   this->root_poa_ = PortableServer::POA::_duplicate (root_poa);
 221 }
 222
 223 EC_Observer::~EC_Observer (void)
 224 {
 225   if (this->gwys_ != 0)
 226     delete[] this->gwys_;
 227 }
 228
 229 void
 230 EC_Observer::initialize_orb_and_poa (int&, ACE_TCHAR*[])
 231 {
 232 }
 233
 234 int
 235 EC_Observer::parse_args (int& argc, ACE_TCHAR* argv[])
 236 {
 237   return this->EC_Driver::parse_args (argc, argv);
 238 }
 239
 240 void
 241 EC_Observer::print_args (void) const
 242 {
 243   this->EC_Driver::print_args ();
 244 }
 245
 246 void
 247 EC_Observer::print_usage (void)
 248 {
 249   this->EC_Driver::print_usage ();
 250 }
 251
 252 void
 253 EC_Observer::execute_test (void)
 254 {
 255   int peer_count = this->master_->channel_count ();
 256   ACE_NEW (this->gwys_, TAO_EC_Gateway_IIOP[peer_count]);
 257
 258   for (int i = 0; i != peer_count; ++i)
 259     {
 260       if (i == this->id_)
 261         continue;
 262
 263       RtecEventChannelAdmin::EventChannel_ptr rmt_ec =
 264         this->master_->channel (i)->event_channel_.in ();
 265
 266       this->gwys_[i].init (rmt_ec,
 267                            this->event_channel_.in ());
 268
 269       RtecEventChannelAdmin::Observer_var obs =
 270         this->gwys_[i]._this ();
 271
 272       RtecEventChannelAdmin::Observer_Handle h =
 273         rmt_ec->append_observer (obs.in ());
 274
 275       this->gwys_[i].observer_handle (h);
 276
 277     }
 278
 279   if (this->allocate_tasks () == -1)
 280     return;
 281
 282   this->activate_tasks ();
 283
 284   if (this->verbose ())
 285     ACE_DEBUG ((LM_DEBUG, "EC_Observer[%d] (%P|%t) suppliers are active\n",
 286                 this->id_));
 287 }
 288
 289 void
 290 EC_Observer::run_cleanup (void)
 291 {
 292   for (int j = 0; j != this->master_->channel_count (); ++j)
 293     {
 294       if (j == this->id_)
 295         continue;
 296
 297       RtecEventChannelAdmin::EventChannel_ptr rmt_ec =
 298         this->master_->channel (j)->event_channel_.in ();
 299       rmt_ec->remove_observer (this->gwys_[j].observer_handle ());
 300
 301       this->gwys_[j].shutdown ();
 302     }
 303 }
 304
 305 void
 306 EC_Observer::dump_results (void)
 307 {
 308   ACE_DEBUG ((LM_DEBUG, "===== Results for %d =====\n", this->id_));
 309
 310   ACE_Throughput_Stats throughput;
 311   ACE_High_Res_Timer::global_scale_factor_type gsf =
 312     ACE_High_Res_Timer::global_scale_factor ();
 313   for (int j = 0; j < this->n_consumers_; ++j)
 314     {
 315       this->consumers_[j]->accumulate (throughput);
 316     }
 317   ACE_DEBUG ((LM_DEBUG, "\n"));
 318
 319   ACE_Throughput_Stats suppliers;
 320   for (int i = 0; i < this->n_suppliers_; ++i)
 321     {
 322       this->suppliers_[i]->accumulate (suppliers);
 323     }
 324
 325   ACE_DEBUG ((LM_DEBUG, "\nTotals:\n"));
 326   throughput.dump_results (ACE_TEXT("EC_Consumer/totals"), gsf);
 327
 328   ACE_DEBUG ((LM_DEBUG, "\n"));
 329   suppliers.dump_results (ACE_TEXT("EC_Supplier/totals"), gsf);
 330 }
 331
 332 void
 333 EC_Observer::connect_consumer (
 334     RtecEventChannelAdmin::ConsumerAdmin_ptr consumer_admin,
 335     int i)
 336 {
 337   if (i == 0)
 338     {
 339       this->EC_Driver::connect_consumer (consumer_admin, i);
 340       return;
 341     }
 342   unsigned int x = ACE_OS::rand_r (&this->seed_);
 343   if (x < RAND_MAX / 8)
 344     this->EC_Driver::connect_consumer (consumer_admin, i);
 345 }
 346
 347 void
 348 EC_Observer::consumer_push (void*,
 349                             const RtecEventComm::EventSet&)
 350 {
 351   unsigned int x = ACE_OS::rand_r (&this->seed_);
 352   if (x < (RAND_MAX / 64))
 353     {
 354       if (this->verbose ())
 355         ACE_DEBUG ((LM_DEBUG,
 356                     "EC_Observer[%d] (%P|%t) reconnecting\n", this->id_));
 357
 358       RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
 359         this->event_channel_->for_consumers ();
 360
 361       for (int i = 1; i < this->n_consumers_; ++i)
 362         {
 363           ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 364
 365           if (this->consumers_[i]->connected ())
 366             {
 367               this->consumers_[i]->disconnect ();
 368             }
 369           else
 370             {
 371               this->EC_Driver::connect_consumer (consumer_admin.in (),
 372                                                  i);
 373             }
 374         }
 375     }
 376 }