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