TAO/performance-tests/Callback/client.cpp

   1 #include "Callback_i.h"
   2
   3 #include "tao/Messaging/Messaging.h"
   4 #include "tao/PortableServer/PortableServer.h"
   5 #include "tao/Strategies/advanced_resource.h"
   6 #include "tao/AnyTypeCode/Any.h"
   7 #include "ace/High_Res_Timer.h"
   8 #include "ace/Get_Opt.h"
   9 #include "ace/Basic_Stats.h"
  10 #include "ace/Stats.h"
  11 #include "ace/Throughput_Stats.h"
  12 #include "ace/Sched_Params.h"
  13 #include "ace/Task.h"
  14 #include "ace/OS_NS_errno.h"
  15
  16 int iterations = 1000;
  17 int do_dump_history = 0;
  18 const ACE_TCHAR *ior = ACE_TEXT("file://server.ior");
  19
  20 /// Parse the arguments.
  21 static int parse_args (int argc, ACE_TCHAR *argv[]);
  22
  23 /// Send events in another thread
  24 class Task : public ACE_Task_Base
  25 {
  26 public:
  27   /// Constructor
  28   Task (Test::Server_ptr server, int iterations);
  29
  30   /// Return 1 when all the messages have been sent
  31   int done ();
  32
  33   /// Run the experiment
  34   int svc ();
  35
  36 private:
  37   /// The consumer
  38   Test::Server_var server_;
  39
  40   /// Number of messages that have to be sent
  41   int remaining_messages_;
  42
  43   /// Synchronize access to remaining_messages_
  44   TAO_SYNCH_MUTEX mutex_;
  45 };
  46
  47 int
  48 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  49 {
  50   int priority =
  51     (ACE_Sched_Params::priority_min (ACE_SCHED_FIFO)
  52      + ACE_Sched_Params::priority_max (ACE_SCHED_FIFO)) / 2;
  53   priority = ACE_Sched_Params::next_priority (ACE_SCHED_FIFO, priority);
  54
  55   // Enable FIFO scheduling
  56   if (ACE_OS::sched_params (ACE_Sched_Params (ACE_SCHED_FIFO,
  57                                               priority,
  58                                               ACE_SCOPE_PROCESS)) != 0)
  59     {
  60       if (ACE_OS::last_error () == EPERM)
  61         {
  62           ACE_DEBUG ((LM_DEBUG,
  63                       "client (%P|%t): user is not superuser, "
  64                       "test runs in time-shared class\n"));
  65         }
  66       else
  67         ACE_ERROR ((LM_ERROR,
  68                     "client (%P|%t): sched_params failed\n"));
  69     }
  70
  71   try
  72     {
  73       CORBA::ORB_var orb =
  74         CORBA::ORB_init (argc, argv);
  75
  76       CORBA::Object_var manager_object =
  77         orb->resolve_initial_references ("ORBPolicyManager");
  78
  79       CORBA::PolicyManager_var policy_manager =
  80         CORBA::PolicyManager::_narrow (manager_object.in ());
  81
  82       CORBA::Any sync_scope;
  83       sync_scope <<= Messaging::SYNC_WITH_TARGET;
  84
  85       CORBA::PolicyList policy_list (1);
  86       policy_list.length (1);
  87       policy_list[0] =
  88         orb->create_policy (Messaging::SYNC_SCOPE_POLICY_TYPE,
  89                             sync_scope);
  90       policy_manager->set_policy_overrides (policy_list,
  91                                             CORBA::SET_OVERRIDE);
  92
  93       CORBA::Object_var poa_object =
  94         orb->resolve_initial_references("RootPOA");
  95
  96       if (CORBA::is_nil (poa_object.in ()))
  97         ACE_ERROR_RETURN ((LM_ERROR,
  98                            " (%P|%t) Unable to initialize the POA.\n"),
  99                           1);
 100
 101       PortableServer::POA_var root_poa =
 102         PortableServer::POA::_narrow (poa_object.in ());
 103
 104       PortableServer::POAManager_var poa_manager =
 105         root_poa->the_POAManager ();
 106
 107       poa_manager->activate ();
 108
 109       if (parse_args (argc, argv) != 0)
 110         return 1;
 111
 112       // Get the event channel object reference
 113       CORBA::Object_var object =
 114         orb->string_to_object (ior);
 115
 116       Test::Server_var server =
 117         Test::Server::_narrow (object.in ());
 118       if (CORBA::is_nil (server.in ()))
 119         {
 120           ACE_ERROR ((LM_ERROR,
 121                       "(%P|%t) Invalid or nil event channel\n"));
 122           return 1;
 123         }
 124
 125       ACE_DEBUG ((LM_DEBUG, "Resolved server\n"));
 126
 127       Callback_i callback_i (iterations);
 128
 129       Test::Callback_var callback = callback_i._this ();
 130
 131       server->set_callback (callback.in ());
 132
 133       ACE_DEBUG ((LM_DEBUG, "Connected callback\n"));
 134
 135       Task task (server.in (), iterations);
 136       task.activate ();
 137
 138       ACE_hrtime_t start = ACE_OS::gethrtime ();
 139       while (!callback_i.done () || !task.done ())
 140         {
 141           ACE_Time_Value tv (1, 0);
 142           orb->run (tv);
 143         }
 144       ACE_hrtime_t end = ACE_OS::gethrtime ();
 145
 146       ACE_Thread_Manager::instance ()->wait ();
 147
 148       // Calibrate the high resolution timer *before* starting the
 149       // test.
 150       ACE_DEBUG ((LM_DEBUG, "Calibrating high res timer ...."));
 151       ACE_High_Res_Timer::calibrate ();
 152
 153       ACE_High_Res_Timer::global_scale_factor_type gsf =
 154         ACE_High_Res_Timer::global_scale_factor ();
 155       ACE_DEBUG ((LM_DEBUG, "Done (%d)\n", gsf));
 156
 157       ACE_Sample_History &history =
 158         callback_i.sample_history ();
 159       if (do_dump_history)
 160         {
 161           history.dump_samples (ACE_TEXT("HISTORY"), gsf);
 162         }
 163
 164       ACE_Basic_Stats stats;
 165       history.collect_basic_stats (stats);
 166       stats.dump_results (ACE_TEXT("Total"), gsf);
 167
 168       ACE_Throughput_Stats::dump_throughput (ACE_TEXT("Total"), gsf,
 169                                              end - start,
 170                                              stats.samples_count ());
 171
 172       server->shutdown ();
 173
 174       PortableServer::ObjectId_var id =
 175         root_poa->servant_to_id (&callback_i);
 176       root_poa->deactivate_object (id.in ());
 177       orb->destroy ();
 178     }
 179   catch (const CORBA::Exception& ex)
 180     {
 181       ex._tao_print_exception (argv[0]);
 182     }
 183   return 0;
 184 }
 185
 186 // ****************************************************************
 187
 188 int
 189 parse_args (int argc, ACE_TCHAR *argv[])
 190 {
 191   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("hi:k:"));
 192   int c;
 193
 194   while ((c = get_opts ()) != -1)
 195     switch (c)
 196       {
 197       case 'h':
 198         do_dump_history = 1;
 199         break;
 200
 201       case 'i':
 202         iterations = ACE_OS::atoi (get_opts.opt_arg ());
 203         break;
 204
 205       case 'k':
 206         ior = get_opts.opt_arg ();
 207         break;
 208
 209       case '?':
 210       default:
 211         ACE_ERROR_RETURN ((LM_ERROR,
 212                            "usage:  %s "
 213                            "-h (dump full sample history) "
 214                            "-i <iterations> "
 215                            "-k <IOR> "
 216                            "\n",
 217                            argv [0]),
 218                           -1);
 219       }
 220   // Indicates successful parsing of the command line
 221   return 0;
 222 }
 223
 224 // ****************************************************************
 225
 226 Task::Task (Test::Server_ptr server,
 227             int iterations)
 228   : server_ (Test::Server::_duplicate (server))
 229   , remaining_messages_ (iterations)
 230 {
 231 }
 232
 233 int
 234 Task::done ()
 235 {
 236   ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->mutex_, 1);
 237   return this->remaining_messages_ == 0;
 238 }
 239
 240 int
 241 Task::svc ()
 242 {
 243   Test::Payload payload(1024); payload.length(1024);
 244   try
 245     {
 246       for (;;)
 247         {
 248           Test::TimeStamp creation = ACE_OS::gethrtime ();
 249           this->server_->request (creation, payload);
 250
 251           // ACE_Time_Value tv (0, 5000);
 252           // ACE_OS::sleep (tv);
 253
 254           ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->mutex_, -1);
 255
 256           this->remaining_messages_--;
 257           if (this->remaining_messages_ == 0)
 258             return 0;
 259         }
 260     }
 261   catch (const CORBA::SystemException& ex)
 262     {
 263       ex._tao_print_exception ("Task::svc");
 264       this->remaining_messages_ = 0;
 265     }
 266   return 0;
 267 }