TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.cpp

   1 #include "ECT_Consumer_Driver.h"
   2
   3 #include "orbsvcs/CosNamingC.h"
   4 #include "orbsvcs/Event_Utilities.h"
   5 #include "orbsvcs/Event_Service_Constants.h"
   6 #include "orbsvcs/Time_Utilities.h"
   7
   8 #include "tao/Timeprobe.h"
   9 #include "tao/debug.h"
  10
  11 #include "ace/Get_Opt.h"
  12 #include "ace/Auto_Ptr.h"
  13 #include "ace/Sched_Params.h"
  14 #include "ace/OS_NS_errno.h"
  15 #include "ace/OS_NS_unistd.h"
  16
  17 int
  18 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  19 {
  20   ECT_Consumer_Driver driver;
  21   return driver.run (argc, argv);
  22 }
  23
  24 // ****************************************************************
  25
  26 ECT_Consumer_Driver::ECT_Consumer_Driver (void)
  27   : n_consumers_ (1),
  28     n_suppliers_ (1),
  29     type_start_ (ACE_ES_EVENT_UNDEFINED),
  30     type_count_ (1),
  31     stall_length_(0),
  32     shutdown_event_channel_ (1),
  33     pid_file_name_ (0),
  34     active_count_ (0)
  35 {
  36 }
  37
  38 ECT_Consumer_Driver::~ECT_Consumer_Driver (void)
  39 {
  40 }
  41
  42 int
  43 ECT_Consumer_Driver::run (int argc, ACE_TCHAR* argv[])
  44 {
  45   try
  46     {
  47       this->orb_ =
  48         CORBA::ORB_init (argc, argv);
  49
  50       CORBA::Object_var poa_object =
  51         this->orb_->resolve_initial_references("RootPOA");
  52
  53       if (CORBA::is_nil (poa_object.in ()))
  54         ACE_ERROR_RETURN ((LM_ERROR,
  55                            " (%P|%t) Unable to initialize the POA.\n"),
  56                           1);
  57
  58       PortableServer::POA_var root_poa =
  59         PortableServer::POA::_narrow (poa_object.in ());
  60
  61       PortableServer::POAManager_var poa_manager =
  62         root_poa->the_POAManager ();
  63
  64       if (this->parse_args (argc, argv))
  65         return 1;
  66
  67       if (TAO_debug_level > 0)
  68         {
  69           ACE_DEBUG ((LM_DEBUG,
  70                       "Execution parameters:\n"
  71                       "  consumers = <%d>\n"
  72                       "  suppliers = <%d>\n"
  73                       "  type start = <%d>\n"
  74                       "  type count = <%d>\n"
  75                       "  pid file name = <%s>\n",
  76
  77                       this->n_consumers_,
  78                       this->n_suppliers_,
  79                       this->type_start_,
  80                       this->type_count_,
  81
  82                       this->pid_file_name_?this->pid_file_name_:ACE_TEXT("nil")) );
  83         }
  84
  85       if (this->pid_file_name_ != 0)
  86         {
  87           FILE* pid = ACE_OS::fopen (this->pid_file_name_, "w");
  88           if (pid != 0)
  89             {
  90               ACE_OS::fprintf (pid, "%ld\n",
  91                                static_cast<long> (ACE_OS::getpid ()));
  92               ACE_OS::fclose (pid);
  93             }
  94         }
  95
  96       int min_priority =
  97         ACE_Sched_Params::priority_min (ACE_SCHED_FIFO);
  98         // Enable FIFO scheduling, e.g., RT scheduling class on Solaris.
  99
 100       if (ACE_OS::sched_params (ACE_Sched_Params (ACE_SCHED_FIFO,
 101                                                   min_priority,
 102                                                   ACE_SCOPE_PROCESS)) != 0)
 103         {
 104           if (ACE_OS::last_error () == EPERM)
 105             ACE_DEBUG ((LM_DEBUG,
 106                         "%s: user is not superuser, "
 107                         "so remain in time-sharing class\n", argv[0]));
 108           else
 109             ACE_ERROR ((LM_ERROR,
 110                         "%s: ACE_OS::sched_params failed\n", argv[0]));
 111         }
 112
 113       if (ACE_OS::thr_setprio (min_priority) == -1)
 114         {
 115           ACE_ERROR ((LM_ERROR, "(%P|%t) main thr_setprio failed,"
 116                       "no real-time features\n"));
 117         }
 118
 119       CORBA::Object_var naming_obj =
 120         this->orb_->resolve_initial_references ("NameService");
 121
 122       if (CORBA::is_nil (naming_obj.in ()))
 123         ACE_ERROR_RETURN ((LM_ERROR,
 124                            " (%P|%t) Unable to get the Naming Service.\n"),
 125                           1);
 126
 127       CosNaming::NamingContext_var naming_context =
 128         CosNaming::NamingContext::_narrow (naming_obj.in ());
 129
 130       CosNaming::Name schedule_name (1);
 131       schedule_name.length (1);
 132       schedule_name[0].id = CORBA::string_dup ("ScheduleService");
 133
 134       CORBA::Object_var sched_obj =
 135         naming_context->resolve (schedule_name);
 136       if (CORBA::is_nil (sched_obj.in ()))
 137         return 1;
 138       RtecScheduler::Scheduler_var scheduler =
 139         RtecScheduler::Scheduler::_narrow (sched_obj.in ());
 140
 141       CosNaming::Name name (1);
 142       name.length (1);
 143       name[0].id = CORBA::string_dup ("EventService");
 144
 145       CORBA::Object_var ec_obj =
 146         naming_context->resolve (name);
 147
 148       RtecEventChannelAdmin::EventChannel_var channel;
 149       if (CORBA::is_nil (ec_obj.in ()))
 150         channel = RtecEventChannelAdmin::EventChannel::_nil ();
 151       else
 152         channel = RtecEventChannelAdmin::EventChannel::_narrow (ec_obj.in ());
 153
 154       poa_manager->activate ();
 155
 156       this->connect_consumers (scheduler.in (), channel.in ());
 157
 158       ACE_DEBUG ((LM_DEBUG, "connected consumer(s)\n"));
 159
 160       ACE_DEBUG ((LM_DEBUG, "running the test\n"));
 161       for (;;)
 162         {
 163           ACE_Time_Value tv (1, 0);
 164           this->orb_->perform_work (tv);
 165           ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->lock_, 1);
 166           if (this->active_count_ <= 0)
 167             break;
 168         }
 169       ACE_DEBUG ((LM_DEBUG, "event loop finished\n"));
 170
 171       this->dump_results ();
 172
 173       this->disconnect_consumers ();
 174
 175       if (this->shutdown_event_channel_ != 0)
 176         {
 177           channel->destroy ();
 178         }
 179
 180       root_poa->destroy (1, 1);
 181
 182       this->orb_->destroy ();
 183     }
 184   catch (const CORBA::SystemException& sys_ex)
 185     {
 186       sys_ex._tao_print_exception ("SYS_EX");
 187     }
 188   catch (const CORBA::Exception& ex)
 189     {
 190       ex._tao_print_exception ("NON SYS EX");
 191     }
 192   return 0;
 193 }
 194
 195 void
 196 ECT_Consumer_Driver::shutdown_consumer (void*)
 197 {
 198   // int ID =
 199   //   (reinterpret_cast<Test_Consumer**> (consumer_cookie)
 200   //    - this->consumers_);
 201   //
 202   // ACE_DEBUG ((LM_DEBUG, "(%t) events received by consumer %d\n", ID));
 203
 204   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 205   this->active_count_--;
 206 }
 207
 208 void
 209 ECT_Consumer_Driver::connect_consumers
 210      (RtecScheduler::Scheduler_ptr scheduler,
 211       RtecEventChannelAdmin::EventChannel_ptr channel)
 212 {
 213   {
 214     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 215     this->active_count_ = this->n_consumers_;
 216   }
 217   for (int i = 0; i < this->n_consumers_; ++i)
 218     {
 219       char buf[BUFSIZ];
 220       ACE_OS::sprintf (buf, "consumer_%02d", i);
 221
 222       ACE_NEW (this->consumers_[i],
 223                Test_Consumer (this,
 224                               this->consumers_ + i,
 225                               this->n_suppliers_,
 226                               this->stall_length_));
 227
 228       this->consumers_[i]->connect (scheduler,
 229                                     buf,
 230                                     this->type_start_,
 231                                     this->type_count_,
 232                                     channel);
 233     }
 234 }
 235
 236 void
 237 ECT_Consumer_Driver::dump_results (void)
 238 {
 239   ACE_High_Res_Timer::global_scale_factor_type gsf =
 240     ACE_High_Res_Timer::global_scale_factor ();
 241
 242   ACE_Throughput_Stats throughput;
 243   for (int i = 0; i < this->n_consumers_; ++i)
 244     {
 245       ACE_TCHAR buf[BUFSIZ];
 246       ACE_OS::sprintf (buf, ACE_TEXT("consumer_%02d"), i);
 247
 248       this->consumers_[i]->dump_results (buf, gsf);
 249       this->consumers_[i]->accumulate (throughput);
 250     }
 251   throughput.dump_results (ACE_TEXT("ECT_Consumer/totals"), gsf);
 252 }
 253
 254 void
 255 ECT_Consumer_Driver::disconnect_consumers (void)
 256 {
 257   for (int i = 0; i < this->n_consumers_; ++i)
 258     {
 259       this->consumers_[i]->disconnect ();
 260
 261       delete this->consumers_[i];
 262       this->consumers_[i] = 0;
 263     }
 264 }
 265
 266 int
 267 ECT_Consumer_Driver::parse_args (int argc, ACE_TCHAR *argv [])
 268 {
 269   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT("xdc:s:h:p:o:"));
 270   int opt;
 271
 272   while ((opt = get_opt ()) != EOF)
 273     {
 274       switch (opt)
 275         {
 276         case 'x':
 277           this->shutdown_event_channel_ = 0;
 278           break;
 279
 280         case 'c':
 281           this->n_consumers_ = ACE_OS::atoi (get_opt.opt_arg ());
 282           break;
 283
 284         case 's':
 285           this->n_suppliers_ = ACE_OS::atoi (get_opt.opt_arg ());
 286           break;
 287
 288         case 'h':
 289           {
 290             char* aux;
 291                 char* arg = ACE_OS::strtok_r (ACE_TEXT_ALWAYS_CHAR(get_opt.opt_arg ()), ",", &aux);
 292
 293             this->type_start_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
 294                 arg = ACE_OS::strtok_r (0, ",", &aux);
 295             this->type_count_ = ACE_OS::atoi (arg);
 296           }
 297           break;
 298
 299         case 'p':
 300           this->pid_file_name_ = get_opt.opt_arg ();
 301           break;
 302
 303         case 'o':
 304           this->stall_length_ = ACE_OS::atoi (get_opt.opt_arg ());
 305           break;
 306
 307         case '?':
 308         default:
 309           ACE_DEBUG ((LM_DEBUG,
 310                       "Usage: %s "
 311                       "[ORB options] "
 312                       "-d -x "
 313                       "-c <n_consumers> "
 314                       "-s <n_suppliers> "
 315                       "-h <type_start,type_count> "
 316                       "-p <pid file name> "
 317                       "\n",
 318                       argv[0]));
 319           return -1;
 320         }
 321     }
 322
 323   if (this->n_suppliers_ <= 0)
 324     {
 325       ACE_DEBUG ((LM_DEBUG,
 326                   "%s: number of suppliers (%d) is out of range, "
 327                   "reset to default (%d)\n",
 328                   argv[0], this->n_suppliers_,
 329                   1));
 330       this->n_suppliers_ = 1;
 331     }
 332
 333   if (this->n_consumers_ <= 0)
 334     {
 335       ACE_ERROR_RETURN ((LM_ERROR,
 336                          "%s: number of consumers or "
 337                          "suppliers out of range\n", argv[0]), -1);
 338     }
 339
 340   if (this->type_count_ <= 0)
 341     {
 342       this->type_count_ = 1;
 343       ACE_ERROR_RETURN ((LM_ERROR,
 344                          "%s: number of event types "
 345                          "suppliers out of range, reset to default (1)\n",
 346                          argv[0]), -1);
 347     }
 348
 349   return 0;
 350 }