TAO/tests/RTCORBA/Thread_Pool/server.cpp

   1 #include "test_i.h"
   2 #include "ace/Get_Opt.h"
   3 #include "ace/OS_NS_unistd.h"
   4 #include "tao/ORB_Core.h"
   5 #include "ace/Task.h"
   6 #include "tao/RTPortableServer/RTPortableServer.h"
   7 #include "../check_supported_priorities.cpp"
   8
   9 const ACE_TCHAR *ior_output_file = ACE_TEXT("ior");
  10 int ior_count = 1;
  11 CORBA::ULong static_threads = 2;
  12 CORBA::ULong dynamic_threads = 2;
  13 long nap_time = 1000;
  14
  15 int
  16 parse_args (int argc, ACE_TCHAR *argv[])
  17 {
  18   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("o:s:d:t:"));
  19   int c;
  20
  21   while ((c = get_opts ()) != -1)
  22     switch (c)
  23       {
  24       case 'o':
  25         ior_output_file = get_opts.opt_arg ();
  26         break;
  27
  28       case 's':
  29         static_threads = ACE_OS::atoi (get_opts.opt_arg ());
  30         break;
  31
  32       case 'd':
  33         dynamic_threads = ACE_OS::atoi (get_opts.opt_arg ());
  34         break;
  35
  36       case 't':
  37         nap_time = ACE_OS::atoi (get_opts.opt_arg ());
  38         break;
  39
  40       case '?':
  41       default:
  42         ACE_ERROR_RETURN ((LM_ERROR,
  43                            "usage:  %s "
  44                            "-o <iorfile> "
  45                            "-s <static_threads> "
  46                            "-d <dynamic_threads> "
  47                            "-t <nap_time> "
  48                            "\n",
  49                            argv [0]),
  50                           -1);
  51       }
  52
  53   // Indicates successful parsing of the command line
  54   return 0;
  55 }
  56
  57 int
  58 write_ior_to_file (CORBA::ORB_ptr orb,
  59                    test_ptr test)
  60 {
  61   CORBA::String_var ior =
  62     orb->object_to_string (test);
  63
  64   char filename[BUFSIZ];
  65   ACE_OS::sprintf (filename,
  66                    "%s_%d",
  67                    ACE_TEXT_ALWAYS_CHAR (ior_output_file),
  68                    ior_count++);
  69
  70   FILE *output_file =
  71     ACE_OS::fopen (filename,
  72                    "w");
  73
  74   if (output_file == 0)
  75     ACE_ERROR_RETURN ((LM_ERROR,
  76                        "Cannot open output file for writing IOR: %s",
  77                        filename),
  78                       -1);
  79
  80   ACE_OS::fprintf (output_file,
  81                    "%s",
  82                    ior.in ());
  83
  84   ACE_OS::fclose (output_file);
  85
  86   return 0;
  87 }
  88
  89 int
  90 create_POA_and_register_servant (CORBA::Policy_ptr threadpool_policy,
  91                                  const char *poa_name,
  92                                  PortableServer::POAManager_ptr poa_manager,
  93                                  PortableServer::POA_ptr root_poa,
  94                                  CORBA::ORB_ptr orb,
  95                                  RTCORBA::RTORB_ptr rt_orb)
  96 {
  97   // Policies for the firstPOA to be created.
  98   CORBA::PolicyList policies (3); policies.length (3);
  99
 100   // Implicit_activation policy.
 101   policies[0] =
 102     root_poa->create_implicit_activation_policy
 103     (PortableServer::IMPLICIT_ACTIVATION);
 104
 105   // Thread pool policy.
 106   policies[1] =
 107     CORBA::Policy::_duplicate (threadpool_policy);
 108
 109   // Priority Model policy.
 110   policies[2] =
 111     rt_orb->create_priority_model_policy (RTCORBA::CLIENT_PROPAGATED, 0);
 112
 113   // Create the POA under the RootPOA.
 114   PortableServer::POA_var poa =
 115     root_poa->create_POA (poa_name,
 116                           poa_manager,
 117                           policies);
 118
 119   // Creation of POAs is over. Destroy the Policy objects.
 120   for (CORBA::ULong i = 0;
 121        i < policies.length ();
 122        ++i)
 123     {
 124       policies[i]->destroy ();
 125     }
 126
 127   test_i *servant =
 128     new test_i (orb,
 129                 poa.in (),
 130                 nap_time);
 131
 132   PortableServer::ServantBase_var safe_servant (servant);
 133   ACE_UNUSED_ARG (safe_servant);
 134
 135   PortableServer::ObjectId_var id =
 136     poa->activate_object (servant);
 137
 138   CORBA::Object_var object = poa->id_to_reference (id.in ());
 139
 140   test_var test = test::_narrow (object.in ());
 141
 142   int const result = write_ior_to_file (orb, test.in ());
 143
 144   return result;
 145 }
 146
 147 class Task : public ACE_Task_Base
 148 {
 149 public:
 150   Task (ACE_Thread_Manager &thread_manager,
 151         CORBA::ORB_ptr orb);
 152
 153   int svc ();
 154
 155   CORBA::ORB_var orb_;
 156 };
 157
 158 Task::Task (ACE_Thread_Manager &thread_manager,
 159             CORBA::ORB_ptr orb)
 160   : ACE_Task_Base (&thread_manager),
 161     orb_ (CORBA::ORB::_duplicate (orb))
 162 {
 163 }
 164
 165 int
 166 Task::svc ()
 167 {
 168   try
 169     {
 170       CORBA::Object_var object =
 171         this->orb_->resolve_initial_references ("RootPOA");
 172
 173       PortableServer::POA_var root_poa =
 174         PortableServer::POA::_narrow (object.in ());
 175
 176       PortableServer::POAManager_var poa_manager =
 177         root_poa->the_POAManager ();
 178
 179       object =
 180         this->orb_->resolve_initial_references ("RTORB");
 181
 182       RTCORBA::RTORB_var rt_orb =
 183         RTCORBA::RTORB::_narrow (object.in ());
 184
 185       object =
 186         this->orb_->resolve_initial_references ("RTCurrent");
 187
 188       RTCORBA::Current_var current =
 189         RTCORBA::Current::_narrow (object.in ());
 190
 191       RTCORBA::Priority default_thread_priority =
 192         get_implicit_thread_CORBA_priority (this->orb_.in ());
 193
 194       test_i servant (this->orb_.in (),
 195                       root_poa.in (),
 196                       nap_time);
 197       PortableServer::ObjectId_var id =
 198         root_poa->activate_object (&servant);
 199
 200       CORBA::Object_var object_act = root_poa->id_to_reference (id.in ());
 201
 202       test_var test =
 203         test::_narrow (object_act.in ());
 204
 205       int result =
 206         write_ior_to_file (this->orb_.in (),
 207                            test.in ());
 208
 209       if (result != 0)
 210         return result;
 211
 212       poa_manager->activate ();
 213
 214       CORBA::ULong stacksize = 0;
 215       CORBA::Boolean allow_request_buffering = 0;
 216       CORBA::ULong max_buffered_requests = 0;
 217       CORBA::ULong max_request_buffer_size = 0;
 218
 219       RTCORBA::ThreadpoolId threadpool_id_1 =
 220         rt_orb->create_threadpool (stacksize,
 221                                    static_threads,
 222                                    dynamic_threads,
 223                                    default_thread_priority,
 224                                    allow_request_buffering,
 225                                    max_buffered_requests,
 226                                    max_request_buffer_size);
 227
 228       CORBA::Policy_var threadpool_policy_1 =
 229         rt_orb->create_threadpool_policy (threadpool_id_1);
 230
 231       CORBA::Boolean allow_borrowing = 0;
 232       RTCORBA::ThreadpoolLanes lanes (1);
 233       lanes.length (1);
 234
 235       lanes[0].lane_priority = default_thread_priority;
 236       lanes[0].static_threads = static_threads;
 237       lanes[0].dynamic_threads = dynamic_threads;
 238
 239       RTCORBA::ThreadpoolId threadpool_id_2 =
 240         rt_orb->create_threadpool_with_lanes (stacksize,
 241                                               lanes,
 242                                               allow_borrowing,
 243                                               allow_request_buffering,
 244                                               max_buffered_requests,
 245                                               max_request_buffer_size);
 246
 247       CORBA::Policy_var threadpool_policy_2 =
 248         rt_orb->create_threadpool_policy (threadpool_id_2);
 249
 250       result =
 251         create_POA_and_register_servant (threadpool_policy_1.in (),
 252                                          "first_poa",
 253                                          poa_manager.in (),
 254                                          root_poa.in (),
 255                                          this->orb_.in (),
 256                                          rt_orb.in ());
 257       if (result != 0)
 258         return result;
 259
 260       result =
 261         create_POA_and_register_servant (threadpool_policy_2.in (),
 262                                          "second_poa",
 263                                          poa_manager.in (),
 264                                          root_poa.in (),
 265                                          this->orb_.in (),
 266                                          rt_orb.in ());
 267       if (result != 0)
 268         return result;
 269
 270       this->orb_->run ();
 271
 272       this->orb_->destroy ();
 273     }
 274   catch (const CORBA::Exception& ex)
 275     {
 276       ex._tao_print_exception ("Exception caught:");
 277       return -1;
 278     }
 279
 280   return 0;
 281 }
 282
 283 int
 284 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
 285 {
 286   try
 287     {
 288       CORBA::ORB_var orb = CORBA::ORB_init (argc, argv);
 289
 290       int result =
 291         parse_args (argc, argv);
 292       if (result != 0)
 293         return result;
 294
 295       // Make sure we can support multiple priorities that are required
 296       // for this test.
 297       if (!check_supported_priorities (orb.in ()))
 298         return 2;
 299
 300       // Thread Manager for managing task.
 301       ACE_Thread_Manager thread_manager;
 302
 303       // Create task.
 304       Task task (thread_manager,
 305                  orb.in ());
 306
 307       // Task activation flags.
 308       long flags =
 309         THR_NEW_LWP |
 310         THR_JOINABLE |
 311         orb->orb_core ()->orb_params ()->thread_creation_flags ();
 312
 313       // Activate task.
 314       result =
 315         task.activate (flags);
 316       if (result == -1)
 317         {
 318           if (errno == EPERM)
 319             {
 320               ACE_ERROR_RETURN ((LM_ERROR,
 321                                  "Cannot create thread with scheduling policy %s\n"
 322                                  "because the user does not have the appropriate privileges, terminating program....\n"
 323                                  "Check svc.conf options and/or run as root\n",
 324                                  sched_policy_name (orb->orb_core ()->orb_params ()->ace_sched_policy ())),
 325                                 2);
 326             }
 327           else
 328             // Unexpected error.
 329             ACE_ASSERT (0);
 330         }
 331
 332       // Wait for task to exit.
 333       result =
 334         thread_manager.wait ();
 335       ACE_ASSERT (result != -1);
 336     }
 337   catch (const CORBA::Exception& ex)
 338     {
 339       ex._tao_print_exception ("Exception caught");
 340       return -1;
 341     }
 342
 343   return 0;
 344 }