TAO/orbsvcs/performance-tests/RTEvent/Federated_Roundtrip/client.cpp

   1 #include "RT_Class.h"
   2 #include "ORB_Holder.h"
   3 #include "Servant_var.h"
   4 #include "RIR_Narrow.h"
   5 #include "RTServer_Setup.h"
   6 #include "Peer_Base.h"
   7 #include "Auto_Disconnect.h"
   8 #include "Send_Task.h"
   9 #include "Task_Activator.h"
  10 #include "Client_Pair.h"
  11
  12 #include "orbsvcs/Event_Service_Constants.h"
  13 #include "orbsvcs/Event/EC_Default_Factory.h"
  14
  15 #include "tao/Messaging/Messaging.h"
  16 #include "tao/Strategies/advanced_resource.h"
  17 #include "tao/RTCORBA/Priority_Mapping_Manager.h"
  18 #include "tao/RTCORBA/Continuous_Priority_Mapping.h"
  19 #include "tao/RTPortableServer/RTPortableServer.h"
  20 #include "ace/Get_Opt.h"
  21 #include "ace/High_Res_Timer.h"
  22 #include "ace/Sample_History.h"
  23 #include "ace/Basic_Stats.h"
  24 #include "ace/Stats.h"
  25 #include "ace/Sched_Params.h"
  26 #include "ace/Barrier.h"
  27
  28
  29 const ACE_TCHAR *ior = ACE_TEXT ("file://test.ior");
  30 int nthreads   = 0;
  31 int high_priority_period = 0;
  32 int high_priority_workload = 0;
  33 int low_priority_period  = 0;
  34 int low_priority_workload = 0;
  35 int disable_low_priority = 0;
  36 int use_rt_corba = 0;
  37
  38 class Roundtrip_Peer : public Peer_Base
  39 {
  40 public:
  41   Roundtrip_Peer (CORBA::ORB_ptr orb,
  42                   RTServer_Setup &rtserver_setup,
  43                   RT_Class &rt_class)
  44     : Peer_Base (orb, rtserver_setup)
  45     , rt_class_ (&rt_class)
  46   {
  47   }
  48
  49   //@{
  50   /** @name The Federated_Test::Peer methods
  51    */
  52   virtual Federated_Test::Experiment_Results *
  53       run_experiment (CORBA::Long experiment_id,
  54                       CORBA::Long iterations);
  55   //@}
  56
  57 private:
  58   RT_Class *rt_class_;
  59 };
  60
  61 int
  62 parse_args (int argc, ACE_TCHAR *argv[])
  63 {
  64   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("k:n:l:h:w:v:zr"));
  65   int c;
  66
  67   while ((c = get_opts ()) != -1)
  68     switch (c)
  69       {
  70       case 'k':
  71         ior = get_opts.opt_arg ();
  72         break;
  73
  74       case 'n':
  75         nthreads = ACE_OS::atoi (get_opts.opt_arg ());
  76         break;
  77
  78       case 'h':
  79         high_priority_period = ACE_OS::atoi (get_opts.opt_arg ());
  80         break;
  81
  82       case 'l':
  83         low_priority_period = ACE_OS::atoi (get_opts.opt_arg ());
  84         break;
  85
  86       case 'w':
  87         high_priority_workload = ACE_OS::atoi (get_opts.opt_arg ());
  88         break;
  89
  90       case 'v':
  91         low_priority_workload = ACE_OS::atoi (get_opts.opt_arg ());
  92         break;
  93
  94       case 'z':
  95         disable_low_priority = 1;
  96         break;
  97
  98       case 'r':
  99         use_rt_corba = 1;
 100         break;
 101
 102       case '?':
 103       default:
 104         ACE_ERROR_RETURN ((LM_ERROR,
 105                            "usage:  %s "
 106                            "-k <IOR> "
 107                            "-i iterations (iterations) "
 108                            "-h high_priority_period (usecs) "
 109                            "-l low_priority_period (usecs) "
 110                            "-w high_priority_workload (usecs) "
 111                            "-v low_priority_workload (usecs) "
 112                            "-r (enable RT-CORBA) "
 113                            "-n nthreads (low priority thread) "
 114                            "-z (disable low priority) "
 115                            "\n",
 116                            argv [0]),
 117                           -1);
 118       }
 119   // Indicates successful parsing of the command line
 120   return 0;
 121 }
 122
 123 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 124 {
 125   TAO_EC_Default_Factory::init_svcs();
 126   RT_Class rt_class;
 127
 128   try
 129     {
 130       ORB_Holder orb (argc, argv, "");
 131
 132       if (parse_args (argc, argv) != 0)
 133         return 1;
 134
 135       RTServer_Setup rtserver_setup (use_rt_corba,
 136                                      orb,
 137                                      rt_class,
 138                                      nthreads);
 139
 140       PortableServer::POA_var root_poa =
 141         RIR_Narrow<PortableServer::POA>::resolve (orb,
 142                                                   "RootPOA");
 143
 144       PortableServer::POAManager_var poa_manager =
 145         root_poa->the_POAManager ();
 146
 147       poa_manager->activate ();
 148
 149       ACE_DEBUG ((LM_DEBUG, "Finished ORB and POA configuration\n"));
 150
 151       Servant_var<Roundtrip_Peer> peer_impl (
 152           new Roundtrip_Peer (orb,
 153                               rtserver_setup,
 154                               rt_class)
 155           );
 156
 157       Federated_Test::Peer_var peer =
 158         peer_impl->_this ();
 159
 160       ACE_DEBUG ((LM_DEBUG, "Finished peer configuration and activation\n"));
 161
 162       CORBA::Object_var object =
 163         orb->string_to_object (ior);
 164
 165       Federated_Test::Control_var control =
 166         Federated_Test::Control::_narrow (object.in ());
 167
 168       control->join (peer.in ());
 169
 170       orb->run ();
 171     }
 172   catch (const CORBA::Exception& ex)
 173     {
 174       ex._tao_print_exception ("Exception caught:");
 175       return 1;
 176     }
 177
 178   return 0;
 179 }
 180
 181 Federated_Test::Experiment_Results *
 182 Roundtrip_Peer::run_experiment (CORBA::Long experiment_id,
 183                                 CORBA::Long iterations)
 184 {
 185   int thread_count = 1;
 186 #if 0
 187   if (disable_low_priority == 0)
 188     thread_count += nthreads;
 189 #endif
 190
 191   ACE_Barrier the_barrier (thread_count);
 192
 193   ACE_DEBUG ((LM_DEBUG, "Calibrating high res timer ...."));
 194   ACE_High_Res_Timer::calibrate ();
 195
 196   ACE_High_Res_Timer::global_scale_factor_type gsf =
 197     ACE_High_Res_Timer::global_scale_factor ();
 198   ACE_DEBUG ((LM_DEBUG, "Done (%d)\n", gsf));
 199
 200 #if 0
 201   Low_Priority_Setup<Client_Pair> low_priority_setup (
 202           nthreads,
 203           0, // no limit on the number of iterations
 204           1, // each client gets its own type
 205           experiment_id,
 206           ACE_ES_EVENT_UNDEFINED + 2,
 207           low_priority_workload,
 208           gsf,
 209           disable_low_priority ? 0 : 1,
 210           this->rt_class_.priority_low (),
 211           this->rt_class_.thr_sched_class (),
 212           low_priority_period,
 213           this->poa_.in (),
 214           this->poa_.in (),
 215           this->event_channel_.in (),
 216           &the_barrier);
 217 #endif
 218
 219   Client_Pair high_priority_group;
 220   high_priority_group.init (experiment_id,
 221                             ACE_ES_EVENT_UNDEFINED,
 222                             iterations,
 223                             high_priority_workload,
 224                             gsf,
 225                             this->poa_.in (),
 226                             this->poa_.in ());
 227   high_priority_group.connect (this->event_channel_.in ());
 228   Auto_Disconnect<Client_Pair> high_priority_disconnect (&high_priority_group);
 229
 230   Send_Task high_priority_task;
 231   high_priority_task.init (iterations,
 232                            high_priority_period,
 233                            0,
 234                            ACE_ES_EVENT_UNDEFINED,
 235                            experiment_id,
 236                            high_priority_group.supplier (),
 237                            &the_barrier);
 238   {
 239     // Artificial scope to wait for the high priority task...
 240     Task_Activator<Send_Task> high_priority_act (this->rt_class_->priority_high (),
 241                                                  this->rt_class_->thr_sched_class (),
 242                                                  1,
 243                                                  &high_priority_task);
 244   }
 245
 246   ACE_DEBUG ((LM_DEBUG, "(%P|%t) client - high priority task completed\n"));
 247
 248 #if 0
 249   if (disable_low_priority == 0)
 250     {
 251       low_priority_setup.stop_all_threads ();
 252     }
 253
 254   ACE_DEBUG ((LM_DEBUG, "(%P|%t) client - low priority task(s) stopped\n"));
 255 #endif
 256
 257   ACE_Sample_History &history =
 258     high_priority_group.consumer ()->sample_history ();
 259
 260   Federated_Test::Experiment_Results_var results (
 261      new Federated_Test::Experiment_Results (iterations)
 262      );
 263   results->length (iterations);
 264   for (CORBA::ULong i = 0; i != results->length (); ++i)
 265     {
 266       results[i] = history.get_sample (i) / gsf;
 267     }
 268
 269   return results._retn ();
 270 }