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