TAO/examples/RTCORBA/Activity/Activity.cpp

   1 #include "Activity.h"
   2 #include "Thread_Task.h"
   3 #include "Job_i.h"
   4 #include "POA_Holder.h"
   5 #include "Builder.h"
   6
   7 #include "tao/ORB_Core.h"
   8 #include "tao/debug.h"
   9
  10 #include "ace/High_Res_Timer.h"
  11 #include "ace/Barrier.h"
  12 #include "ace/Timer_Heap.h"
  13 #include "ace/Service_Config.h"
  14 #include "ace/Arg_Shifter.h"
  15 #include "ace/Get_Opt.h"
  16 #include "ace/Argv_Type_Converter.h"
  17 #include "ace/Signal.h"
  18 #include "ace/Reactor.h"
  19
  20
  21 //***************************************************************//
  22 extern "C" void handler (int)
  23 {
  24   ACE_Service_Config::reconfig_occurred (1);
  25 }
  26
  27 //***************************************************************//
  28
  29 Activity::Activity ()
  30   :builder_ (0),
  31    barrier_ (0),
  32    active_task_count_ (0),
  33    active_job_count_ (0)
  34 {
  35   state_lock_ = new ACE_Lock_Adapter <TAO_SYNCH_MUTEX>;
  36 }
  37
  38 Activity::~Activity ()
  39 {
  40   delete state_lock_;
  41   delete barrier_;
  42 }
  43
  44 void
  45 Activity::builder (Builder* builder)
  46 {
  47   builder_ = builder;
  48 }
  49
  50 CORBA::ORB_ptr
  51 Activity::orb ()
  52 {
  53   return orb_.in ();
  54 }
  55
  56 RTCORBA::Current_ptr
  57 Activity::current ()
  58 {
  59   return current_.in ();
  60 }
  61
  62 int
  63 Activity::init (int& argc, ACE_TCHAR *argv [])
  64 {
  65   // Copy command line parameter.
  66   ACE_Argv_Type_Converter command_line(argc, argv);
  67
  68   this->orb_ = CORBA::ORB_init (command_line.get_argc(),
  69                                 command_line.get_TCHAR_argv());
  70
  71   CORBA::Object_var object =
  72     orb_->resolve_initial_references ("RootPOA");
  73
  74   root_poa_ =
  75     PortableServer::POA::_narrow (object.in ());
  76
  77   PortableServer::POAManager_var poa_manager =
  78     root_poa_->the_POAManager ();
  79
  80   object =
  81     orb_->resolve_initial_references ("RTORB");
  82
  83   this->rt_orb_ =
  84     RTCORBA::RTORB::_narrow (object.in ());
  85
  86   object =
  87     orb_->resolve_initial_references ("RTCurrent");
  88
  89   current_ =
  90     RTCORBA::Current::_narrow (object.in ());
  91
  92   poa_manager->activate ();
  93
  94   object = this->orb_->resolve_initial_references ("PriorityMappingManager");
  95   RTCORBA::PriorityMappingManager_var mapping_manager =
  96     RTCORBA::PriorityMappingManager::_narrow (object.in ());
  97
  98   this->priority_mapping_ = mapping_manager->mapping ();
  99
 100   return 0;
 101 }
 102
 103 int
 104 Activity::resolve_naming_service ()
 105 {
 106   CORBA::Object_var naming_obj =
 107     this->orb_->resolve_initial_references ("NameService");
 108
 109   // Need to check return value for errors.
 110   if (CORBA::is_nil (naming_obj.in ()))
 111     ACE_ERROR_RETURN ((LM_ERROR,
 112                        " (%P|%t) Unable to resolve the Naming Service.\n"),
 113                       -1);
 114
 115   this->naming_ =
 116     CosNaming::NamingContextExt::_narrow (naming_obj.in ());
 117
 118   return 0;
 119 }
 120
 121 void
 122 Activity::activate_poa_list ()
 123 {
 124   POA_LIST list;
 125   int count = builder_->poa_list (list);
 126
 127   for (int i = 0; i < count; ++i)
 128     {
 129       list[i]->activate (this->rt_orb_.in(), this->root_poa_.in ());
 130     }
 131 }
 132
 133 void
 134 Activity::activate_job_list ()
 135 {
 136   JOB_LIST list;
 137   int count = builder_->job_list (list);
 138   Job_i* job;
 139
 140   for (int i = 0; i < count; ++i)
 141     {
 142       job = list[i];
 143
 144       if (TAO_debug_level > 0)
 145         ACE_DEBUG ((LM_DEBUG, "Activating job:%C\n", job->name ().c_str ()));
 146
 147       // find your poa
 148       PortableServer::POA_var host_poa =
 149         root_poa_->find_POA (job->poa ().c_str (), 0);
 150
 151       PortableServer::ServantBase_var servant_var (job);
 152
 153       // Register with poa.
 154       PortableServer::ObjectId_var id;
 155
 156       id = host_poa->activate_object (job);
 157
 158       CORBA::Object_var server =
 159         host_poa->id_to_reference (id.in ());
 160
 161       CORBA::String_var ior =
 162         orb_->object_to_string (server.in ());
 163
 164       const ACE_CString &job_name = job->name ();
 165
 166       CosNaming::Name_var name =
 167         this->naming_->to_name (job_name.c_str ());
 168
 169       this->naming_->rebind (name.in (),
 170                              server.in ());
 171
 172       ACE_DEBUG ((LM_DEBUG,
 173                   "Registered %C with the naming service\n",
 174                   job_name.c_str ()));
 175
 176       active_job_count_++;
 177     } /* while */
 178 }
 179
 180 void
 181 Activity::activate_schedule ()
 182 {
 183   TASK_LIST list;
 184   int count = builder_->task_list (list);
 185
 186   if (TAO_debug_level > 0)
 187     ACE_DEBUG ((LM_DEBUG, "Activating schedule, task count = %d\n",
 188                 count));
 189
 190   ACE_NEW (barrier_, ACE_Barrier (count+1));
 191
 192   Periodic_Task* task;
 193
 194   for (int i = 0; i < count; ++i)
 195     {
 196       task = list[i];
 197
 198       // resolve the object from the naming service
 199       CosNaming::Name name (1);
 200       name.length (1);
 201       name[0].id = CORBA::string_dup (task->job ());
 202
 203       CORBA::Object_var obj =
 204         this->naming_->resolve (name);
 205
 206       Job_var job = Job::_narrow (obj.in ());
 207
 208       if (TAO_debug_level > 0)
 209         {
 210           // Check that the object is configured with some
 211           // PriorityModelPolicy.
 212           CORBA::Policy_var policy =
 213             job->_get_policy (RTCORBA::PRIORITY_MODEL_POLICY_TYPE);
 214
 215           RTCORBA::PriorityModelPolicy_var priority_policy =
 216             RTCORBA::PriorityModelPolicy::_narrow (policy.in ());
 217
 218           if (CORBA::is_nil (priority_policy.in ()))
 219             ACE_DEBUG ((LM_DEBUG,
 220                         "ERROR: Priority Model Policy not exposed!\n"));
 221           else
 222             {
 223               RTCORBA::PriorityModel priority_model =
 224                 priority_policy->priority_model ();
 225
 226               if (priority_model == RTCORBA::CLIENT_PROPAGATED)
 227                 ACE_DEBUG ((LM_DEBUG,
 228                             "%C priority_model = RTCORBA::CLIENT_PROPAGATED\n", task->job ()));
 229               else
 230                 ACE_DEBUG ((LM_DEBUG,
 231                             "%C priority_model = RTCORBA::SERVER_DECLARED\n", task->job ()));
 232             }
 233         } /*  if (TAO_debug_level > 0) */
 234
 235       task->job (job.in ());
 236       task->activate_task (this->barrier_, this->priority_mapping_);
 237       active_task_count_++;
 238
 239       ACE_DEBUG ((LM_DEBUG, "Job %C scheduled\n", task->job ()));
 240     }
 241
 242   ACE_DEBUG ((LM_DEBUG, "(%P,%t) Waiting for tasks to synch...\n"));
 243   barrier_->wait ();
 244   ACE_DEBUG ((LM_DEBUG, "(%P,%t) Tasks have synched...\n"));
 245 }
 246
 247 void
 248 Activity::task_ended (Periodic_Task* /*ended_task*/)
 249 {
 250   ACE_DEBUG ((LM_DEBUG, "Active task count = %d\n",active_task_count_));
 251   {
 252     ACE_GUARD (ACE_Lock, ace_mon, *state_lock_);
 253     --active_task_count_;
 254   }
 255
 256   this->check_ifexit ();
 257 }
 258
 259 void
 260 Activity::job_ended (Job_i* /*ended_job*/)
 261 {
 262   ACE_DEBUG ((LM_DEBUG, "Active job count = %d\n",active_job_count_));
 263   {
 264     ACE_GUARD (ACE_Lock, ace_mon, *state_lock_);
 265     --active_job_count_;
 266   }
 267
 268   this->check_ifexit ();
 269 }
 270
 271 void
 272 Activity::check_ifexit ()
 273 {
 274   // All tasks have finished and all jobs have been shutdown.
 275   if (active_task_count_ == 0 && active_job_count_ == 0)
 276     {
 277       ACE_DEBUG ((LM_DEBUG, "Shutdown in progress ...\n"));
 278       // ask all tasks to dump stats.
 279
 280       TASK_LIST task_list;
 281       int count = builder_->task_list (task_list);
 282
 283       ACE_TCHAR msg[BUFSIZ];
 284       ACE_OS::sprintf (msg, ACE_TEXT("# Stats generated on --\n"));
 285
 286       for (int i = 0; i < count; ++i)
 287         {
 288           task_list[i]->dump_stats (msg);
 289         }
 290
 291       // shutdown the ORB
 292       orb_->shutdown (false);
 293     }
 294 }
 295
 296 CORBA::Short
 297 Activity::get_server_priority (CORBA::Object_ptr server)
 298 {
 299   // Get the Priority Model Policy from the stub.
 300   CORBA::Policy_var policy =
 301     server->_get_policy (RTCORBA::PRIORITY_MODEL_POLICY_TYPE);
 302
 303   // Narrow down to correct type.
 304   RTCORBA::PriorityModelPolicy_var priority_policy =
 305     RTCORBA::PriorityModelPolicy::_narrow (policy.in ());
 306
 307   // Make sure that we have the SERVER_DECLARED priority model.
 308   RTCORBA::PriorityModel priority_model =
 309     priority_policy->priority_model ();
 310   if (priority_model != RTCORBA::SERVER_DECLARED)
 311     return -1;
 312
 313   // Return the server priority.
 314   return priority_policy->server_priority ();
 315 }
 316
 317 void
 318 Activity::run (int argc, ACE_TCHAR *argv[])
 319 {
 320   this->init (argc, argv);
 321
 322   if (this->resolve_naming_service () == -1)
 323     return;
 324
 325   this->activate_poa_list ();
 326
 327   this->activate_job_list ();
 328
 329   this->activate_schedule ();
 330
 331   this->create_started_flag_file (argc, argv);
 332
 333   orb_->run ();
 334
 335   ACE_Thread_Manager::instance ()->wait ();
 336
 337   CORBA::release (this->naming_);
 338
 339   // Hack for proper cleanup.
 340   this->builder_->fini ();
 341
 342   orb_->destroy ();
 343 }
 344
 345 void
 346 Activity::create_started_flag_file (int argc, ACE_TCHAR *argv[])
 347 {
 348   ACE_Arg_Shifter arg_shifter (argc, argv);
 349
 350   const ACE_TCHAR* current_arg = 0;
 351
 352   while (arg_shifter.is_anything_left ())
 353     {
 354       if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-Started_Flag"))))
 355         {
 356           FILE *file = ACE_OS::fopen (current_arg, ACE_TEXT("w"));
 357
 358           if (file == 0)
 359             ACE_ERROR ((LM_ERROR,
 360                         "Unable to open %s for writing: %p\n",
 361                         current_arg));
 362
 363           ACE_OS::fprintf (file, "ignore");
 364
 365           ACE_OS::fclose (file);
 366
 367           arg_shifter.consume_arg ();
 368         }
 369       else
 370         {
 371           arg_shifter.ignore_arg ();
 372         }
 373     }
 374 }
 375
 376 int
 377 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
 378 {
 379   ACE_Service_Config::static_svcs ()->insert (&ace_svc_desc_Builder);
 380
 381   ACE_Sig_Action sa ((ACE_SignalHandler) handler, SIGHUP);
 382
 383   ACE_Timer_Heap timer_queue;
 384   ACE_Reactor::instance ()->timer_queue (&timer_queue);
 385
 386   int rc = 0;
 387   try
 388     {
 389       ACTIVITY::instance()->run (argc, argv);
 390     }
 391   catch (const CORBA::Exception& ex)
 392     {
 393       ex._tao_print_exception ("Caught exception:");
 394       rc = 1;
 395     }
 396
 397   // reset stack based timer queue
 398   ACE_Reactor::instance ()->timer_queue (0);
 399
 400   return rc;
 401 }
 402
 403 ACE_SINGLETON_TEMPLATE_INSTANTIATE(ACE_Singleton, Activity, ACE_Null_Mutex);
 404