TAO/tests/RTCORBA/Linear_Priority/client.cpp

   1 #include "ace/Get_Opt.h"
   2 #include "ace/Task.h"
   3 #include "testC.h"
   4 #include "tao/RTCORBA/RTCORBA.h"
   5 #include "tao/ORB_Core.h"
   6 #include "tao/Policy_ManagerC.h"
   7 #include "../check_supported_priorities.cpp"
   8 #include "../common_args.cpp"
   9
  10 static int iterations = 5;
  11 static int shutdown_server = 0;
  12 static int debug = 1;
  13
  14 static const ACE_TCHAR *ior = ACE_TEXT ("file://ior");
  15
  16 static const ACE_TCHAR *invocation_priorities_file = ACE_TEXT ("invocation_priorities");
  17 static const ACE_TCHAR *bands_file = ACE_TEXT ("empty_file");
  18
  19 static int
  20 parse_args (int argc, ACE_TCHAR **argv)
  21 {
  22   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("b:d:p:k:i:x"));
  23   int c;
  24
  25   while ((c = get_opts ()) != -1)
  26     switch (c)
  27       {
  28       case 'k':
  29         ior = get_opts.opt_arg ();
  30         break;
  31
  32       case 'i':
  33         iterations = ACE_OS::atoi (get_opts.opt_arg ());
  34         break;
  35
  36       case 'd':
  37         debug = ACE_OS::atoi (get_opts.opt_arg ());
  38         break;
  39
  40       case 'x':
  41         shutdown_server = 1;
  42         break;
  43
  44       case 'p':
  45         invocation_priorities_file = get_opts.opt_arg ();
  46         break;
  47
  48       case 'b':
  49         bands_file = get_opts.opt_arg ();
  50         break;
  51
  52       case '?':
  53       default:
  54         ACE_ERROR_RETURN ((LM_ERROR,
  55                            "usage:  %s "
  56                            "-b <bands_file> "
  57                            "-d <debug> "
  58                            "-p <invocation_priorities_file> "
  59                            "-k ior "
  60                            "-i iterations "
  61                            "-x shutdown server "
  62                            "\n",
  63                            argv [0]),
  64                           -1);
  65       }
  66
  67   return 0;
  68 }
  69
  70 class Worker_Thread : public ACE_Task_Base
  71 {
  72 public:
  73   Worker_Thread (ACE_Thread_Manager &thread_manager,
  74                  test_ptr test,
  75                  RTCORBA::Current_ptr current,
  76                  CORBA::Short priority);
  77
  78   int svc ();
  79
  80   void validate_connection ();
  81
  82 private:
  83   test_var test_;
  84   RTCORBA::Current_var current_;
  85   CORBA::Short priority_;
  86 };
  87
  88 Worker_Thread::Worker_Thread (ACE_Thread_Manager &thread_manager,
  89                               test_ptr test,
  90                               RTCORBA::Current_ptr current,
  91                               CORBA::Short priority)
  92   : ACE_Task_Base (&thread_manager),
  93     test_ (test::_duplicate (test)),
  94     current_ (RTCORBA::Current::_duplicate (current)),
  95     priority_ (priority)
  96 {
  97 }
  98
  99 void
 100 Worker_Thread::validate_connection ()
 101 {
 102   // Try to validate the connection several times, ignoring transient
 103   // exceptions.  If the connection can still not be setup, return
 104   // failure.
 105   CORBA::PolicyList_var inconsistent_policies;
 106   int max_attempts = 10;
 107   int current_attempt = 0;
 108   for (;;)
 109     {
 110       try
 111         {
 112           ++current_attempt;
 113           this->test_->_validate_connection (inconsistent_policies.out ());
 114
 115           // If successful, we are done.
 116           return;
 117         }
 118       catch (const CORBA::TRANSIENT& )
 119         {
 120           // If we have reach our maximum number of tries, throw exception.
 121           if (current_attempt == max_attempts)
 122             throw;
 123           // Otherwise, ignore...
 124         }
 125       catch (const CORBA::Exception&)
 126         {
 127           // Rethrow any other exceptions.
 128           throw;
 129         }
 130     }
 131 }
 132
 133 int
 134 Worker_Thread::svc ()
 135 {
 136   try
 137     {
 138       this->current_->the_priority (this->priority_);
 139
 140       this->validate_connection ();
 141
 142       for (int i = 0; i < iterations; i++)
 143         {
 144           this->test_->method ();
 145         }
 146     }
 147   catch (const CORBA::Exception& ex)
 148     {
 149       ex._tao_print_exception ("Worker Thread exception:");
 150     }
 151   return 0;
 152 }
 153
 154 class Task : public ACE_Task_Base
 155 {
 156 public:
 157   Task (ACE_Thread_Manager &thread_manager,
 158         CORBA::ORB_ptr orb);
 159
 160   int svc ();
 161
 162   CORBA::ORB_var orb_;
 163 };
 164
 165 Task::Task (ACE_Thread_Manager &thread_manager,
 166             CORBA::ORB_ptr orb)
 167   : ACE_Task_Base (&thread_manager),
 168     orb_ (CORBA::ORB::_duplicate (orb))
 169 {
 170 }
 171
 172 int
 173 Task::svc ()
 174 {
 175   try
 176     {
 177       CORBA::Object_var object =
 178         this->orb_->resolve_initial_references ("RTORB");
 179
 180       RTCORBA::RTORB_var rt_orb =
 181         RTCORBA::RTORB::_narrow (object.in ());
 182
 183       object =
 184         this->orb_->resolve_initial_references ("RTCurrent");
 185
 186       RTCORBA::Current_var current =
 187         RTCORBA::Current::_narrow (object.in ());
 188
 189       current->the_priority (0);
 190
 191       object =
 192         this->orb_->resolve_initial_references ("ORBPolicyManager");
 193
 194       CORBA::PolicyManager_var policy_manager =
 195         CORBA::PolicyManager::_narrow (object.in ());
 196
 197       object =
 198         this->orb_->string_to_object (ior);
 199
 200       test_var test =
 201         test::_narrow (object.in ());
 202
 203       ULong_Array priorities;
 204       int result =
 205         get_values ("client",
 206                     invocation_priorities_file,
 207                     "invocation priorities",
 208                     priorities,
 209                     debug);
 210       if (result != 0)
 211         return result;
 212
 213       CORBA::PolicyList policies;
 214
 215       result =
 216         get_priority_bands ("client",
 217                             bands_file,
 218                             rt_orb.in (),
 219                             policies,
 220                             debug);
 221       if (result != 0)
 222         return result;
 223
 224       policy_manager->set_policy_overrides (policies,
 225                                             CORBA::SET_OVERRIDE);
 226
 227       u_long i = 0;
 228
 229       // Thread Manager for managing workers.
 230       ACE_Thread_Manager thread_manager;
 231
 232       // Workers.
 233       Worker_Thread **workers = 0;
 234
 235       ACE_NEW_RETURN (workers,
 236                       Worker_Thread *[priorities.size ()],
 237                       -1);
 238
 239       for (i = 0;
 240            i < priorities.size ();
 241            ++i)
 242         {
 243           ACE_NEW_RETURN (workers[i],
 244                           Worker_Thread (thread_manager,
 245                                          test.in (),
 246                                          current.in (),
 247                                          priorities[i]),
 248                           -1);
 249
 250           long flags =
 251             THR_NEW_LWP |
 252             THR_JOINABLE |
 253             this->orb_->orb_core ()->orb_params ()->thread_creation_flags ();
 254
 255           result =
 256             workers[i]->activate (flags);
 257           if (result != 0)
 258             return result;
 259         }
 260
 261       thread_manager.wait ();
 262
 263       for (i = 0;
 264            i < priorities.size ();
 265            ++i)
 266         {
 267           delete workers[i];
 268         }
 269       delete[] workers;
 270
 271       if (shutdown_server)
 272         {
 273           test->shutdown ();
 274         }
 275     }
 276   catch (const CORBA::Exception& ex)
 277     {
 278       ex._tao_print_exception ("Unexpected exception!");
 279       return -1;
 280     }
 281
 282   return 0;
 283 }
 284
 285 int
 286 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 287 {
 288   try
 289     {
 290       CORBA::ORB_var orb =
 291         CORBA::ORB_init (argc, argv);
 292
 293       int result =
 294         parse_args (argc, argv);
 295       if (result != 0)
 296         return result;
 297
 298       // Make sure we can support multiple priorities that are required
 299       // for this test.
 300       if (!check_supported_priorities (orb.in ()))
 301         return 2;
 302
 303       // Thread Manager for managing task.
 304       ACE_Thread_Manager thread_manager;
 305
 306       // Create task.
 307       Task task (thread_manager,
 308                  orb.in ());
 309
 310       // Task activation flags.
 311       long flags =
 312         THR_NEW_LWP |
 313         THR_JOINABLE |
 314         orb->orb_core ()->orb_params ()->thread_creation_flags ();
 315
 316       // Activate task.
 317       result =
 318         task.activate (flags);
 319       if (result == -1)
 320         {
 321           if (errno == EPERM)
 322             {
 323               ACE_ERROR_RETURN ((LM_ERROR,
 324                                  "Cannot create thread with scheduling policy %C\n"
 325                                  "because the user does not have the appropriate privileges, terminating program....\n"
 326                                  "Check svc.conf options and/or run as root\n",
 327                                  sched_policy_name (orb->orb_core ()->orb_params ()->ace_sched_policy ())),
 328                                 2);
 329             }
 330           else
 331             // Unexpected error.
 332             ACE_ASSERT (0);
 333         }
 334
 335       // Wait for task to exit.
 336       result =
 337         thread_manager.wait ();
 338       ACE_ASSERT (result != -1);
 339     }
 340   catch (const CORBA::Exception& ex)
 341     {
 342       ex._tao_print_exception ("Unexpected exception!");
 343       return -1;
 344     }
 345
 346   return 0;
 347 }