TAO/tests/RTCORBA/Profile_And_Endpoint_Selection/client.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file     client.cpp
   5  *
   6  *   This is a client implementation.
   7  *
   8  *  @author  Irfan Pyarali
   9  */
  10 //=============================================================================
  11
  12
  13 #include "ace/Get_Opt.h"
  14 #include "ace/Task.h"
  15 #include "tao/ORB_Core.h"
  16 #include "tao/RTCORBA/RTCORBA.h"
  17 #include "tao/Policy_ManagerC.h"
  18 #include "tao/Strategies/advanced_resource.h"
  19 #include "../check_supported_priorities.cpp"
  20 #include "../common_args.cpp"
  21 #include "testC.h"
  22
  23 static int iterations = 1;
  24 static int debug = 1;
  25 static int shutdown_server = 0;
  26 static const ACE_TCHAR *ior = ACE_TEXT ("file://ior");
  27 static const ACE_TCHAR *invocation_priorities_file = ACE_TEXT ("empty_file");
  28 static const ACE_TCHAR *protocols_file = ACE_TEXT ("empty_file");
  29 static const ACE_TCHAR *bands_file = ACE_TEXT ("empty_file");
  30
  31 static int
  32 parse_args (int argc, ACE_TCHAR **argv)
  33 {
  34   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("b:d:i:k:p:t:x"));
  35   int c;
  36
  37   while ((c = get_opts ()) != -1)
  38     switch (c)
  39       {
  40       case 'b':
  41         bands_file = get_opts.opt_arg ();
  42         break;
  43
  44       case 'd':
  45         debug = ACE_OS::atoi (get_opts.opt_arg ());
  46         break;
  47
  48       case 'i':
  49         iterations = ACE_OS::atoi (get_opts.opt_arg ());
  50         break;
  51
  52       case 'k':
  53         ior = get_opts.opt_arg ();
  54         break;
  55
  56       case 'p':
  57         protocols_file = get_opts.opt_arg ();
  58         break;
  59
  60       case 't':
  61         invocation_priorities_file = get_opts.opt_arg ();
  62         break;
  63
  64       case 'x':
  65         shutdown_server = 1;
  66         break;
  67
  68       case '?':
  69       default:
  70         ACE_ERROR_RETURN ((LM_ERROR,
  71                            "%s usage:\n"
  72                            "\t-b <bands file> (defaults to %s)\n"
  73                            "\t-i <iterations> (defaults to %d)\n"
  74                            "\t-k <ior> (defaults to %s)\n"
  75                            "\t-p <protocol file> (defaults to %s)\n"
  76                            "\t-t <invocation priorities file> (defaults to %s)\n"
  77                            "\t-x <shutdown server> (defaults to %d)\n"
  78                            "\n",
  79                            argv[0],
  80                            bands_file,
  81                            iterations,
  82                            ior,
  83                            protocols_file,
  84                            invocation_priorities_file,
  85                            shutdown_server),
  86                           -1);
  87       }
  88
  89   return 0;
  90 }
  91
  92 class Client
  93 {
  94 public:
  95   Client (test_ptr test,
  96           CORBA::ORB_ptr orb,
  97           RTCORBA::Current_ptr current,
  98           RTCORBA::RTORB_ptr rt_orb,
  99           CORBA::PolicyManager_ptr policy_manager);
 100
 101   void vanilla_invocations ();
 102
 103   void priority_invocations (int debug);
 104
 105   void set_client_protocols_policies (int debug);
 106
 107   void set_priority_bands (int debug);
 108
 109   void set_private_connection_policies ();
 110
 111   void reset_policies ();
 112
 113
 114 private:
 115   test_var test_;
 116   CORBA::ORB_var orb_;
 117   RTCORBA::Current_var current_;
 118   RTCORBA::RTORB_var rt_orb_;
 119   CORBA::PolicyManager_var policy_manager_;
 120 };
 121
 122 class Worker_Thread : public ACE_Task_Base
 123 {
 124 public:
 125   Worker_Thread (ACE_Thread_Manager &thread_manager,
 126                  Client &client,
 127                  test_ptr test,
 128                  RTCORBA::Current_ptr current,
 129                  CORBA::Short priority);
 130
 131   int svc ();
 132
 133   void validate_connection ();
 134
 135 private:
 136   Client client_;
 137   test_var test_;
 138   RTCORBA::Current_var current_;
 139   CORBA::Short priority_;
 140 };
 141
 142 Worker_Thread::Worker_Thread (ACE_Thread_Manager &thread_manager,
 143                               Client &client,
 144                               test_ptr test,
 145                               RTCORBA::Current_ptr current,
 146                               CORBA::Short priority)
 147   : ACE_Task_Base (&thread_manager),
 148     client_ (client),
 149     test_ (test::_duplicate (test)),
 150     current_ (RTCORBA::Current::_duplicate (current)),
 151     priority_ (priority)
 152 {
 153 }
 154
 155 void
 156 Worker_Thread::validate_connection ()
 157 {
 158   // Try to validate the connection several times, ignoring transient
 159   // exceptions.  If the connection can still not be setup, return
 160   // failure.
 161   CORBA::PolicyList_var inconsistent_policies;
 162   int max_attempts = 10;
 163   int current_attempt = 0;
 164   for (;;)
 165     {
 166       try
 167         {
 168           ++current_attempt;
 169           this->test_->_validate_connection (inconsistent_policies.out ());
 170
 171           // If successful, we are done.
 172           return;
 173         }
 174       catch (const CORBA::TRANSIENT& )
 175         {
 176           // If we have reach our maximum number of tries, throw exception.
 177           if (current_attempt == max_attempts)
 178             throw;
 179           // Otherwise, ignore...
 180         }
 181       catch (const CORBA::Exception&)
 182         {
 183           // Rethrow any other exceptions.
 184           throw;
 185         }
 186     }
 187 }
 188
 189 int
 190 Worker_Thread::svc ()
 191 {
 192   try
 193     {
 194       this->current_->the_priority (this->priority_);
 195
 196       this->validate_connection ();
 197
 198       this->client_.vanilla_invocations ();
 199     }
 200   catch (const CORBA::Exception& ex)
 201     {
 202       ex._tao_print_exception ("Worker Thread exception:");
 203     }
 204   return 0;
 205 }
 206
 207 Client::Client (test_ptr test,
 208                 CORBA::ORB_ptr orb,
 209                 RTCORBA::Current_ptr current,
 210                 RTCORBA::RTORB_ptr rt_orb,
 211                 CORBA::PolicyManager_ptr policy_manager)
 212   : test_ (test::_duplicate (test)),
 213     orb_ (CORBA::ORB::_duplicate (orb)),
 214     current_ (RTCORBA::Current::_duplicate (current)),
 215     rt_orb_ (RTCORBA::RTORB::_duplicate (rt_orb)),
 216     policy_manager_ (CORBA::PolicyManager::_duplicate (policy_manager))
 217 {
 218 }
 219
 220 void
 221 Client::vanilla_invocations ()
 222 {
 223   for (int i = 0; i < iterations; i++)
 224     {
 225       this->test_->method ();
 226     }
 227 }
 228
 229 void
 230 Client::priority_invocations (int debug)
 231 {
 232   ULong_Array priorities;
 233   int result =
 234     get_values ("client",
 235                 invocation_priorities_file,
 236                 "invocation priorities",
 237                 priorities,
 238                 debug);
 239   if (result != 0)
 240     {
 241       ACE_ERROR ((LM_ERROR,
 242                   "Error in parsing invocation priorities data file: %s\n",
 243                   invocation_priorities_file));
 244       return;
 245     }
 246
 247   u_long i = 0;
 248
 249   Worker_Thread **workers = 0;
 250
 251   ACE_NEW_THROW_EX (workers,
 252                     Worker_Thread *[priorities.size ()],
 253                     CORBA::NO_MEMORY ());
 254
 255   ACE_Thread_Manager thread_manager;
 256
 257   for (i = 0;
 258        i < priorities.size ();
 259        ++i)
 260     {
 261       ACE_NEW_THROW_EX (workers[i],
 262                         Worker_Thread (thread_manager,
 263                                        *this,
 264                                        this->test_.in (),
 265                                        this->current_.in (),
 266                                        priorities[i]),
 267                         CORBA::NO_MEMORY ());
 268
 269       long flags =
 270         THR_NEW_LWP |
 271         THR_JOINABLE |
 272         this->orb_->orb_core ()->orb_params ()->thread_creation_flags ();
 273
 274       result =
 275         workers[i]->activate (flags);
 276       if (result != 0)
 277         {
 278           ACE_ERROR ((LM_ERROR,
 279                       "Cannot activate thread\n"));
 280           return;
 281         }
 282     }
 283
 284   thread_manager.wait ();
 285
 286   for (i = 0;
 287        i < priorities.size ();
 288        ++i)
 289     {
 290       delete workers[i];
 291     }
 292   delete[] workers;
 293 }
 294
 295 void
 296 Client::set_client_protocols_policies (int debug)
 297 {
 298   CORBA::PolicyList policies;
 299
 300   int result =
 301     get_protocols ("client",
 302                    protocols_file,
 303                    this->rt_orb_.in (),
 304                    policies,
 305                    debug);
 306   if (result != 0)
 307     {
 308       ACE_ERROR ((LM_ERROR,
 309                   "Error in parsing protocols data file: %s\n",
 310                   protocols_file));
 311       return;
 312     }
 313
 314   this->policy_manager_->set_policy_overrides (policies,
 315                                                CORBA::ADD_OVERRIDE);
 316 }
 317
 318 void
 319 Client::set_priority_bands (int debug)
 320 {
 321   CORBA::PolicyList policies;
 322
 323   int result =
 324     get_priority_bands ("client",
 325                         bands_file,
 326                         this->rt_orb_.in (),
 327                         policies,
 328                         debug);
 329   if (result != 0)
 330     {
 331       ACE_ERROR ((LM_ERROR,
 332                   "Error in parsing bands data file: %s\n",
 333                   bands_file));
 334       return;
 335     }
 336
 337   this->policy_manager_->set_policy_overrides (policies,
 338                                                CORBA::ADD_OVERRIDE);
 339 }
 340
 341 void
 342 Client::set_private_connection_policies ()
 343 {
 344   CORBA::PolicyList policies;
 345   policies.length (1);
 346
 347   policies[0] =
 348     this->rt_orb_->create_private_connection_policy ();
 349
 350   this->policy_manager_->set_policy_overrides (policies,
 351                                                CORBA::ADD_OVERRIDE);
 352 }
 353
 354 void
 355 Client::reset_policies ()
 356 {
 357   CORBA::PolicyList empty_policies;
 358
 359   this->policy_manager_->set_policy_overrides (empty_policies,
 360                                                CORBA::SET_OVERRIDE);
 361 }
 362
 363 class Task : public ACE_Task_Base
 364 {
 365 public:
 366   Task (ACE_Thread_Manager &thread_manager,
 367         CORBA::ORB_ptr orb);
 368
 369   int svc ();
 370
 371   CORBA::ORB_var orb_;
 372 };
 373
 374 Task::Task (ACE_Thread_Manager &thread_manager,
 375             CORBA::ORB_ptr orb)
 376   : ACE_Task_Base (&thread_manager),
 377     orb_ (CORBA::ORB::_duplicate (orb))
 378 {
 379 }
 380
 381 int
 382 Task::svc ()
 383 {
 384   try
 385     {
 386       CORBA::Object_var object =
 387         this->orb_->resolve_initial_references ("RTORB");
 388
 389       RTCORBA::RTORB_var rt_orb =
 390         RTCORBA::RTORB::_narrow (object.in ());
 391
 392       object =
 393         this->orb_->resolve_initial_references ("ORBPolicyManager");
 394
 395       CORBA::PolicyManager_var policy_manager =
 396         CORBA::PolicyManager::_narrow (object.in ());
 397
 398       object =
 399         this->orb_->resolve_initial_references ("RTCurrent");
 400
 401       RTCORBA::Current_var current =
 402         RTCORBA::Current::_narrow (object.in ());
 403
 404       current->the_priority (0);
 405
 406       object =
 407         this->orb_->string_to_object (ior);
 408
 409       test_var test =
 410         test::_narrow (object.in ());
 411
 412       Client client (test.in (),
 413                      this->orb_.in (),
 414                      current.in (),
 415                      rt_orb.in (),
 416                      policy_manager.in ());
 417
 418       client.vanilla_invocations ();
 419
 420       client.set_private_connection_policies ();
 421
 422       client.vanilla_invocations ();
 423
 424       client.reset_policies ();
 425
 426       client.set_client_protocols_policies (debug);
 427
 428       client.vanilla_invocations ();
 429
 430       client.set_private_connection_policies ();
 431
 432       client.vanilla_invocations ();
 433
 434       client.reset_policies ();
 435
 436       client.set_priority_bands (debug);
 437
 438       client.priority_invocations (debug);
 439
 440       client.set_private_connection_policies ();
 441
 442       client.priority_invocations (debug);
 443
 444       client.reset_policies ();
 445
 446       client.set_priority_bands (debug);
 447
 448       client.set_client_protocols_policies (0);
 449
 450       client.priority_invocations (0);
 451
 452       client.set_private_connection_policies ();
 453
 454       client.priority_invocations (debug);
 455
 456       client.reset_policies ();
 457
 458       if (shutdown_server)
 459         {
 460           test->shutdown ();
 461         }
 462     }
 463   catch (const CORBA::Exception& ex)
 464     {
 465       ex._tao_print_exception ("Error!");
 466       return -1;
 467     }
 468
 469   return 0;
 470 }
 471
 472 int
 473 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 474 {
 475   try
 476     {
 477       CORBA::ORB_var orb =
 478         CORBA::ORB_init (argc, argv);
 479
 480       int parse_args_result =
 481         parse_args (argc, argv);
 482       if (parse_args_result != 0)
 483         return parse_args_result;
 484
 485       // Make sure we can support multiple priorities that are required
 486       // for this test.
 487       if (!check_supported_priorities (orb.in ()))
 488         return 2;
 489
 490       // Thread Manager for managing task.
 491       ACE_Thread_Manager thread_manager;
 492
 493       // Create task.
 494       Task task (thread_manager,
 495                  orb.in ());
 496
 497       // Task activation flags.
 498       long flags =
 499         THR_NEW_LWP |
 500         THR_JOINABLE |
 501         orb->orb_core ()->orb_params ()->thread_creation_flags ();
 502
 503       // Activate task.
 504       int result =
 505         task.activate (flags);
 506       if (result == -1)
 507         {
 508           if (errno == EPERM)
 509             {
 510               ACE_ERROR_RETURN ((LM_ERROR,
 511                                  "Cannot create thread with scheduling policy %C\n"
 512                                  "because the user does not have the appropriate privileges, terminating program....\n"
 513                                  "Check svc.conf options and/or run as root\n",
 514                                  sched_policy_name (orb->orb_core ()->orb_params ()->ace_sched_policy ())),
 515                                 2);
 516             }
 517           else
 518             // Unexpected error.
 519             ACE_ASSERT (0);
 520         }
 521
 522       // Wait for task to exit.
 523       result =
 524         thread_manager.wait ();
 525       ACE_ASSERT (result != -1);
 526     }
 527   catch (const CORBA::Exception& ex)
 528     {
 529       ex._tao_print_exception ("Error!");
 530       return -1;
 531     }
 532
 533   return 0;
 534 }