TAO/examples/CSD_Strategy/ThreadPool5/ServerApp.cpp

   1 #include "ServerApp.h"
   2 #include "OrbTask.h"
   3 #include "FooServantList.h"
   4 #include "ClientTask.h"
   5 #include "OrbShutdownTask.h"
   6 #include "ace/Get_Opt.h"
   7 #include "tao/CSD_ThreadPool/CSD_TP_Strategy.h"
   8 #include "tao/Intrusive_Ref_Count_Handle_T.h"
   9 // To force static load the service.
  10 #include "tao/PI/PI.h"
  11 #include "tao/CSD_ThreadPool/CSD_ThreadPool.h"
  12
  13
  14 ServerApp::ServerApp()
  15   : ior_filename_(ACE_TEXT("foo")),
  16     num_servants_(1),
  17     num_csd_threads_ (1),
  18     num_clients_(1),
  19     num_orb_threads_ (1),
  20     collocated_test_ (0),
  21     servant_to_deactivate_ (-1)
  22 {
  23 }
  24
  25
  26 ServerApp::~ServerApp()
  27 {
  28 }
  29
  30
  31 int
  32 ServerApp::run (int argc, ACE_TCHAR* argv[])
  33 {
  34   CORBA::ORB_var orb = CORBA::ORB_init (argc, argv);
  35
  36   // Parse the command-line args for this application.
  37   // * Raises -1 if problems are encountered.
  38   // * Returns 1 if the usage statement was explicitly requested.
  39   // * Returns 0 otherwise.
  40   int result = this->parse_args (argc, argv);
  41   if (result != 0)
  42     {
  43       return result;
  44     }
  45
  46   TheOrbShutdownTask::instance()->orb (orb.in ());
  47
  48   CORBA::Object_var obj
  49     = orb->resolve_initial_references("RootPOA");
  50
  51   if (CORBA::is_nil(obj.in()))
  52     {
  53       ACE_ERROR((LM_ERROR,
  54                  "(%P|%t) Failed to resolve initial ref for 'RootPOA'.\n"));
  55       throw TestException();
  56     }
  57
  58   PortableServer::POA_var root_poa
  59     = PortableServer::POA::_narrow(obj.in());
  60
  61   if (CORBA::is_nil(root_poa.in()))
  62     {
  63       ACE_ERROR((LM_ERROR,
  64                  "(%P|%t) Failed to narrow obj ref to POA interface.\n"));
  65       throw TestException();
  66     }
  67
  68   PortableServer::POAManager_var poa_manager
  69     = root_poa->the_POAManager();
  70
  71   // Create the child POA.
  72   CORBA::PolicyList policies(1);
  73   policies.length(1);
  74
  75   policies[0]
  76     = root_poa->create_id_assignment_policy(PortableServer::USER_ID);
  77
  78   PortableServer::POA_var child_poa
  79     = root_poa->create_POA("ChildPoa",
  80                            poa_manager.in(),
  81                            policies);
  82
  83   if (CORBA::is_nil(child_poa.in()))
  84     {
  85       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR [ServerApp::run()]: "
  86                  "Failed to create the child POA.\n"));
  87       throw TestException();
  88     }
  89
  90   policies[0]->destroy ();
  91
  92   // Create the thread pool servant dispatching strategy object, and
  93   // hold it in a (local) smart pointer variable.
  94   TAO_Intrusive_Ref_Count_Handle<TAO::CSD::TP_Strategy> csd_tp_strategy =
  95                                                  new TAO::CSD::TP_Strategy();
  96
  97   csd_tp_strategy->set_num_threads(this->num_csd_threads_);
  98
  99   // Tell the strategy to apply itself to the child poa.
 100   if (csd_tp_strategy->apply_to(child_poa.in()) == false)
 101     {
 102       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR [ServerApp::run()]: "
 103                  "Failed to apply custom dispatching strategy to child poa.\n"));
 104       throw TestException();
 105     }
 106
 107   FooServantList servants(this->ior_filename_.c_str(),
 108                           this->num_servants_,
 109                           this->num_clients_,
 110                           this->collocated_test_,
 111                           this->servant_to_deactivate_,
 112                           orb.in());
 113
 114   // Activate the POA Manager before start the ClientTask thread so that
 115   // we do not need coordinate the ClientTask and main thread for the
 116   // collocated test.
 117   poa_manager->activate();
 118
 119   servants.create_and_activate(orb.in (),
 120                                child_poa.in());
 121
 122   ACE_DEBUG((LM_DEBUG,
 123              "(%P|%t) ServerApp is ready.\n"));
 124
 125   // If the num_orb_threads_ is exactly one, then just use the current
 126   // (mainline) thread to run the ORB event loop.
 127   if (this->num_orb_threads_ == 1)
 128     {
 129       // Since the num_orb_threads_ is exactly one, we just use the current
 130       // (mainline) thread to run the ORB event loop.
 131       orb->run();
 132     }
 133   else
 134     {
 135       // The num_orb_threads_ is greater than 1, so we will use an OrbTask
 136       // (active object) to run the ORB event loop in (num_orb_threads_ - 1)
 137       // threads.  We use the current (mainline) thread as the other thread
 138       // running the ORB event loop.
 139       OrbTask orb_task(orb.in(), this->num_orb_threads_ - 1);
 140
 141       // Activate the OrbTask worker threads
 142       if (orb_task.open() != 0)
 143         {
 144           ACE_ERROR((LM_ERROR,
 145                      "(%P|%t) Failed to open the OrbTask.\n"));
 146           throw TestException();
 147         }
 148
 149       // This will use the current (mainline) thread to run the ORB event loop.
 150       orb->run();
 151
 152       // Now that the current thread has unblocked from running the orb,
 153       // make sure to wait for all of the worker threads to complete.
 154       orb_task.wait();
 155     }
 156
 157   ACE_DEBUG((LM_DEBUG,
 158              "(%P|%t) ServerApp is waiting for OrbShutdownTask.\n"));
 159   TheOrbShutdownTask::instance()->wait ();
 160
 161   // Sleep for 2 second to let the done() two-way call complete
 162   // before cleanup.
 163   ACE_OS::sleep (2);
 164
 165   if (collocated_test_)
 166     {
 167       servants.collocated_client ()->wait ();
 168     }
 169
 170   ACE_DEBUG((LM_DEBUG,
 171              "(%P|%t) ServerApp is destroying the Root POA.\n"));
 172
 173   // Tear-down the root poa and orb.
 174   root_poa->destroy(1, 1);
 175
 176   ACE_DEBUG((LM_DEBUG,
 177              "(%P|%t) ServerApp is destroying the ORB.\n"));
 178
 179   orb->destroy();
 180
 181   ACE_DEBUG((LM_DEBUG,
 182              "(%P|%t) ServerApp has completed running successfully.\n"));
 183
 184   return 0;
 185 }
 186
 187
 188 int
 189 ServerApp::parse_args(int argc, ACE_TCHAR* argv[])
 190 {
 191   this->exe_name_ = argv[0];
 192
 193   ACE_Get_Opt get_opts(argc, argv, ACE_TEXT("p:s:c:t:l:d:n:"));
 194
 195   int c;
 196   int tmp;
 197
 198   while ((c = get_opts()) != -1)
 199     {
 200       int parse_error = 0;
 201
 202       switch (c)
 203       {
 204         case 'p':
 205           this->ior_filename_ = get_opts.opt_arg();
 206           break;
 207
 208         case 's':
 209           tmp = ACE_OS::atoi(get_opts.opt_arg());
 210           if (tmp < 1)
 211             {
 212               ACE_ERROR((LM_ERROR,
 213                          "(%P|%t) Error.  -s must be followed by an integer "
 214                          "value greater than 0.\n"));
 215               parse_error = 1;
 216             }
 217
 218           this->num_servants_ = tmp;
 219           break;
 220
 221         case 'c':
 222           tmp = ACE_OS::atoi(get_opts.opt_arg());
 223           if (tmp < 1)
 224             {
 225               ACE_ERROR((LM_ERROR,
 226                          "(%P|%t) Error.  -c must be followed by an integer "
 227                          "value greater than 0.\n"));
 228               parse_error = 1;
 229             }
 230
 231           this->num_clients_ = tmp;
 232           break;
 233
 234         case 't':
 235           tmp = ACE_OS::atoi(get_opts.opt_arg());
 236           if (tmp < 1)
 237             {
 238               ACE_ERROR((LM_ERROR,
 239                          "(%P|%t) Error.  -t must be followed by an integer "
 240                          "value greater than 0.\n"));
 241               parse_error = 1;
 242             }
 243
 244           this->num_orb_threads_ = tmp;
 245           break;
 246
 247         case 'n':
 248           tmp = ACE_OS::atoi(get_opts.opt_arg());
 249           if (tmp < 1)
 250             {
 251               ACE_ERROR((LM_ERROR,
 252                          "(%P|%t) Error.  -n must be followed by an integer "
 253                          "value greater than 0.\n"));
 254               parse_error = 1;
 255             }
 256
 257           this->num_csd_threads_ = tmp;
 258           break;
 259
 260         case 'l':
 261           tmp = ACE_OS::atoi(get_opts.opt_arg());
 262           if (tmp < 0)
 263             {
 264               ACE_ERROR((LM_ERROR,
 265                          "(%P|%t) Error.  -l must be followed by an integer "
 266                          "value greater than -1.\n"));
 267               parse_error = 1;
 268             }
 269
 270           this->collocated_test_ = tmp;
 271           break;
 272
 273         case 'd':
 274           tmp = ACE_OS::atoi(get_opts.opt_arg());
 275           if (tmp < 0)
 276             {
 277               ACE_ERROR((LM_ERROR,
 278                          "(%P|%t) Error.  -d must be followed by an integer "
 279                          "value >= 0.\n"));
 280               parse_error = 1;
 281             }
 282
 283           this->servant_to_deactivate_ = tmp;
 284           break;
 285
 286         case '?':
 287           this->usage_statement();
 288           return 1;
 289
 290         default:
 291           this->usage_statement();
 292           return -1;
 293       }
 294
 295       if (parse_error != 0)
 296         {
 297           this->usage_statement();
 298           return parse_error;
 299         }
 300     }
 301
 302   // The deadlock will happen with the collocated callback test
 303   // when we have one working thread, so create at least one more
 304   // working thread would resolve the deadlock.
 305   if (this->collocated_test_ == 1 && this->num_csd_threads_ == 1)
 306     {
 307       ACE_ERROR((LM_ERROR,
 308                  "(%P|%t) Error. The num_csd_threads_ should be "
 309                  ">= 1.\n"));
 310       return -1;
 311     }
 312
 313   return 0;
 314 }
 315
 316
 317 void
 318 ServerApp::usage_statement()
 319 {
 320   ACE_ERROR((LM_ERROR,
 321               "(%P|%t) usage:  %s\n"
 322               "\t[-p <ior_filename_prefix>]\n"
 323               "\t[-s <num_servants>]\n"
 324               "\t[-c <num_clients>]\n"
 325               "\t[-n <num_csd_threads>]\n"
 326               "\t[-t <num_orb_threads>]\n"
 327               "\t[-l <collocation_test>]\n"
 328               "\t[-d <servant_to_deactivate>]\n"
 329               "Default ior_filename_prefix is 'foo'.\n"
 330               "Default num_servants is 1.\n"
 331               "Default num_clients is 1.\n"
 332               "Default num_orb_threads is 1.\n"
 333               "Default collocation_test flag is 0.\n"
 334               "Default servant_to_deactivate is -1 means not deactivate servant.\n"
 335               "   0  means deactivate all servant.\n"
 336               "   >0 means the index (servant_to_deactivate-1) of the servant in the servant list.\n",
 337               this->exe_name_.c_str ()));
 338 }