TAO/examples/RTScheduling/DT_Creator.cpp

   1 #include "DT_Creator.h"
   2 #include "Thread_Task.h"
   3 #include "Task_Stats.h"
   4 #include "DT_Creator.h"
   5
   6 #include "tao/ORB_Core.h"
   7 #include "tao/RTScheduling/Current.h"
   8
   9 #include "ace/High_Res_Timer.h"
  10
  11 ACE_Atomic_Op<TAO_SYNCH_MUTEX, long> guid_counter;
  12
  13 int
  14 DT_Creator::dt_task_init (ACE_Arg_Shifter& arg_shifter)
  15 {
  16   static int dt_index = 0;
  17   time_t start_time = 0;
  18   int load = 0;
  19   int iter = 0;
  20   int importance = 0;
  21   char *job_name = 0;
  22   int dist = 0;
  23   const ACE_TCHAR* current_arg = 0;
  24
  25   if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-Importance")) == 0)
  26     {
  27       arg_shifter.consume_arg ();
  28       current_arg = arg_shifter.get_current ();
  29       importance = ACE_OS::atoi (current_arg);
  30       arg_shifter.consume_arg ();
  31     }
  32
  33   if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-Start_Time"))))
  34     {
  35       start_time = ACE_OS::atoi (current_arg);
  36       arg_shifter.consume_arg ();
  37     }
  38
  39   if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-Iter"))))
  40     {
  41       iter = ACE_OS::atoi (current_arg);
  42       arg_shifter.consume_arg ();
  43     }
  44
  45   if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-Load"))))
  46     {
  47       load = ACE_OS::atoi (current_arg);
  48       arg_shifter.consume_arg ();
  49     }
  50
  51   if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-JobName"))))
  52     {
  53       job_name = (char *)current_arg;
  54       dist = 1;
  55       arg_shifter.consume_arg ();
  56     }
  57
  58   dt_list_ [dt_index++] = this->create_thr_task (importance,
  59              start_time,
  60              load,
  61              iter,
  62              dist,
  63              job_name);
  64
  65   return 0;
  66 }
  67
  68 int log_index = 0;
  69
  70 int
  71 DT_Creator::init (int argc, ACE_TCHAR *argv [])
  72 {
  73   gsf_ = ACE_High_Res_Timer::global_scale_factor ();
  74   state_lock_ = new ACE_Lock_Adapter <TAO_SYNCH_MUTEX>;
  75   shutdown_lock_ = new ACE_Lock_Adapter <TAO_SYNCH_MUTEX>;
  76   active_dt_count_ = 0;
  77   active_job_count_ = 0;
  78   ACE_NEW_RETURN (log, char*[BUFSIZ * 100],-1);
  79
  80   ACE_Arg_Shifter arg_shifter (argc, argv);
  81
  82   const ACE_TCHAR* current_arg = 0;
  83
  84   dt_count_ = 0;
  85   poa_count_ = 0;
  86   int poa_count = 0;
  87   job_count_ = 0;
  88   int job_count = 0;
  89   while (arg_shifter.is_anything_left ())
  90     {
  91       if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-GuidSeed"))))
  92         {
  93           guid_counter = (long) ACE_OS::atoi (current_arg);
  94           arg_shifter.consume_arg ();
  95         }
  96       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-DT_Count"))))
  97         {
  98           dt_count_ = ACE_OS::atoi (current_arg);
  99           ACE_NEW_RETURN (dt_list_, Thread_Task*[dt_count_], -1);
 100           arg_shifter.consume_arg ();
 101         }
 102       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-POA_Count"))))
 103         {
 104           poa_count_ = ACE_OS::atoi (current_arg);
 105           ACE_NEW_RETURN (poa_list_, POA_Holder*[poa_count_], -1);
 106           arg_shifter.consume_arg ();
 107         }
 108       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-JOB_Count"))))
 109         {
 110           job_count_ = ACE_OS::atoi (current_arg);
 111           ACE_NEW_RETURN (job_list_, Job_i*[job_count_], -1);
 112           arg_shifter.consume_arg ();
 113         }
 114       else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-DT_Task")) == 0)
 115         {
 116           arg_shifter.consume_arg ();
 117           dt_task_init (arg_shifter);
 118         }
 119       else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-POA")) == 0)
 120         {
 121           arg_shifter.consume_arg ();
 122
 123           ACE_NEW_RETURN (this->poa_list_[poa_count], POA_Holder (), -1);
 124
 125           if (this->poa_list_[poa_count]->init (arg_shifter) == -1)
 126             {
 127               delete this->poa_list_[poa_count];
 128               return -1;
 129             }
 130           else
 131             {
 132               poa_count++;
 133             }
 134         }
 135       else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-Job")) == 0)
 136         {
 137           arg_shifter.consume_arg ();
 138
 139           ACE_NEW_RETURN (this->job_list_[job_count], Job_i (this), -1);
 140
 141           if (this->job_list_[job_count]->init (arg_shifter) == -1)
 142             {
 143               delete this->job_list_[job_count];
 144               return -1;
 145             }
 146           else
 147             {
 148               job_count++;
 149             }
 150         }
 151       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-OutFile"))))
 152         {
 153           file_name_ = current_arg;
 154           arg_shifter.consume_arg ();
 155         }
 156       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-LogFile"))))
 157         {
 158           log_file_name_ = current_arg;
 159           arg_shifter.consume_arg ();
 160         }
 161       else
 162         {
 163           arg_shifter.ignore_arg ();
 164         }
 165     }
 166
 167   return 0;
 168 }
 169
 170 void
 171 DT_Creator::register_synch_obj ()
 172 {
 173   CosNaming::Name name (1);
 174   name.length (1);
 175
 176   CosNaming::NamingContext_var synch_context;
 177
 178   try
 179     {
 180       // Try binding the sender context in the NS
 181       name [0].id =
 182         CORBA::string_dup ("Synch");
 183
 184       synch_context = this->naming_->bind_new_context (name);
 185
 186       //
 187       // We reach here if there was no exception raised in
 188       // <bind_new_context>.  We then create a receiver context.
 189       //
 190
 191     }
 192   catch (const CosNaming::NamingContext::AlreadyBound& )
 193     {
 194       //
 195       // The synch context already exists, probably created by the
 196       // receiver(s).
 197       //
 198
 199       // Get the synch context.
 200       name [0].id =
 201         CORBA::string_dup ("Synch");
 202
 203       CORBA::Object_var object =
 204         this->naming_->resolve (name);
 205
 206       synch_context = CosNaming::NamingContext::_narrow (object.in ());
 207     }
 208
 209   ACE_CString synch_name ("Synch");
 210   ACE_Time_Value timestamp = ACE_OS::gettimeofday ();
 211
 212   char buf [BUFSIZ];
 213   ACE_OS::sprintf(buf, "%lu", static_cast<unsigned long> (timestamp.sec ()));
 214   synch_name += buf;
 215
 216   name [0].id =
 217     CORBA::string_dup (synch_name.c_str ());
 218
 219   ACE_DEBUG ((LM_DEBUG,
 220         "Synch Name %C\n\n",
 221         synch_name.c_str ()));
 222
 223   ACE_NEW (synch_,
 224      Synch_i);
 225
 226   Synch_var synch = synch_->_this ();
 227
 228   // Register the synch object with the Synch context.
 229   synch_context->rebind (name,
 230        synch.in ());
 231 }
 232
 233
 234 int
 235 DT_Creator::activate_root_poa ()
 236 {
 237   CORBA::Object_var object =
 238     this->orb_->resolve_initial_references ("RootPOA");
 239
 240   PortableServer::POA_var root_poa =
 241     PortableServer::POA::_narrow (object.in ());
 242
 243   PortableServer::POAManager_var poa_manager =
 244     root_poa->the_POAManager ();
 245
 246   poa_manager->activate ();
 247
 248   return 0;
 249 }
 250
 251 void
 252 DT_Creator::activate_poa_list ()
 253 {
 254   if (TAO_debug_level > 0)
 255     ACE_DEBUG ((LM_DEBUG,
 256     "DT_Creator::activate_poa_list\n"));
 257
 258   if (poa_count_ > 0)
 259     {
 260       CORBA::Object_var object =
 261         this->orb_->resolve_initial_references ("RTORB");
 262
 263       RTCORBA::RTORB_var rt_orb = RTCORBA::RTORB::_narrow (object.in ());
 264
 265       object = this->orb_->resolve_initial_references ("RootPOA");
 266
 267       PortableServer::POA_var root_poa =
 268         PortableServer::POA::_narrow (object.in ());
 269
 270       for (int i = 0; i < poa_count_; ++i)
 271         {
 272           poa_list_[i]->activate (rt_orb.in(), root_poa.in ());
 273         }
 274     }
 275 }
 276
 277 void
 278 DT_Creator::activate_job_list ()
 279 {
 280   if (TAO_debug_level > 0)
 281     ACE_DEBUG ((LM_DEBUG,
 282     "DT_Creator::activate_job_list\n"));
 283
 284   Job_i* job;
 285
 286   CORBA::Object_var object =
 287     this->orb_->resolve_initial_references ("RootPOA");
 288
 289   PortableServer::POA_var root_poa =
 290     PortableServer::POA::_narrow (object.in ());
 291
 292   for (int i = 0; i < job_count_; ++i)
 293     {
 294       ++active_job_count_;
 295
 296       job = job_list_[i];
 297
 298       if (TAO_debug_level > 0)
 299         ACE_DEBUG ((LM_DEBUG, "Activating job:%C\n", job->name ().c_str ()));
 300
 301       // find your poa
 302       PortableServer::POA_var host_poa =
 303         root_poa->find_POA (job->poa ().c_str (), 0);
 304
 305       PortableServer::ServantBase_var servant_var (job);
 306
 307       // Register with poa.
 308       PortableServer::ObjectId_var id;
 309
 310       id = host_poa->activate_object (job);
 311
 312       CORBA::Object_var server =
 313         host_poa->id_to_reference (id.in ());
 314
 315       CORBA::String_var ior =
 316         orb_->object_to_string (server.in ());
 317
 318       const ACE_CString &job_name = job->name ();
 319
 320       CosNaming::Name_var name =
 321         this->naming_->to_name (job_name.c_str ());
 322
 323       this->naming_->rebind (name.in (),
 324                              server.in ());
 325     } /* while */
 326
 327   ACE_DEBUG ((LM_DEBUG,
 328               "Activated Job List\n"));
 329 }
 330
 331 void
 332 DT_Creator::activate_schedule ()
 333 {
 334   if (TAO_debug_level > 0)
 335     ACE_DEBUG ((LM_DEBUG,
 336     "Activating schedule, task count = %d\n",
 337     dt_count_));
 338
 339   Thread_Task* task;
 340
 341   for (int i = 0; i < dt_count_; ++i)
 342     {
 343       task = dt_list_[i];
 344
 345       if (task->dist ())
 346   {
 347     // resolve the object from the naming service
 348     CosNaming::Name name (1);
 349     name.length (1);
 350     name[0].id = CORBA::string_dup (task->job ());
 351
 352     CORBA::Object_var obj =
 353       this->naming_->resolve (name);
 354
 355     Job_var job = Job::_narrow (obj.in ());
 356
 357     //    if (TAO_debug_level > 0)
 358     // {
 359         // Check that the object is configured with some
 360         // PriorityModelPolicy.
 361         CORBA::Policy_var policy =
 362     job->_get_policy (RTCORBA::PRIORITY_MODEL_POLICY_TYPE);
 363
 364         RTCORBA::PriorityModelPolicy_var priority_policy =
 365     RTCORBA::PriorityModelPolicy::_narrow (policy.in ());
 366
 367         if (CORBA::is_nil (priority_policy.in ()))
 368     ACE_DEBUG ((LM_DEBUG,
 369           "ERROR: Priority Model Policy not exposed!\n"));
 370         else
 371     {
 372                   /*
 373       RTCORBA::PriorityModel priority_model =
 374         priority_policy->priority_model ();
 375
 376       if (priority_model == RTCORBA::CLIENT_PROPAGATED)
 377         ACE_DEBUG ((LM_DEBUG,
 378         "%C priority_model = RTCORBA::CLIENT_PROPAGATED\n", task->job ()));
 379       else
 380         ACE_DEBUG ((LM_DEBUG,
 381         "%C priority_model = RTCORBA::SERVER_DECLARED\n", task->job ()));
 382       */
 383     }
 384         //} /*  if (TAO_debug_level > 0) */
 385
 386     task->job (job.in ());
 387   }
 388     }
 389   if (TAO_debug_level > 0 && dt_count_ > 0)
 390     ACE_DEBUG ((LM_DEBUG,
 391     "Activated schedule, task count = %d\n",
 392     dt_count_));
 393 }
 394
 395 int
 396 DT_Creator::resolve_naming_service ()
 397 {
 398   CORBA::Object_var naming_obj =
 399     this->orb_->resolve_initial_references ("NameService");
 400
 401   // Need to check return value for errors.
 402   if (CORBA::is_nil (naming_obj.in ()))
 403     ACE_ERROR_RETURN ((LM_ERROR,
 404                        " (%P|%t) Unable to resolve the Naming Service.\n"),
 405                       -1);
 406
 407   this->naming_ =
 408     CosNaming::NamingContextExt::_narrow (naming_obj.in ());
 409
 410   return 0;
 411 }
 412
 413 void
 414 DT_Creator::create_distributable_threads (RTScheduling::Current_ptr current)
 415 {
 416   long flags;
 417   flags = THR_NEW_LWP | THR_JOINABLE;
 418   flags |=
 419     orb_->orb_core ()->orb_params ()->scope_policy () |
 420     orb_->orb_core ()->orb_params ()->sched_policy ();
 421
 422   ACE_DEBUG ((LM_DEBUG,
 423               "Waiting to Synch\n"));
 424
 425   while (!this->synch ()->synched ())
 426     {
 427       try
 428         {
 429           this->orb_->perform_work ();
 430         }
 431       catch (const CORBA::Exception &)
 432         {
 433           return;
 434         }
 435     }
 436
 437   CORBA::Policy_var sched_param;
 438   sched_param = this->sched_param (100);
 439   const char * name = 0;
 440   current->begin_scheduling_segment (name,
 441               sched_param.in (),
 442               sched_param.in ());
 443
 444   ACE_NEW (base_time_,
 445      ACE_Time_Value (*(this->synch ()->base_time ())));
 446
 447   for (int i = 0; i < this->dt_count_; i++)
 448     {
 449       ACE_Time_Value now (ACE_OS::gettimeofday ());
 450
 451       ACE_Time_Value elapsed_time =  now - *base_time_;
 452
 453       char buf [BUFSIZ];
 454       ACE_OS::sprintf (buf, "elapsed time = %d\n now = %d\n base_time_ = %d\n",
 455            (int) elapsed_time.sec (),
 456            (int) now.sec (),
 457            (int) base_time_->sec());
 458
 459       log [log_index++] = ACE_OS::strdup (buf) ;
 460
 461       ACE_hthread_t curr_thr;
 462       ACE_Thread::self (curr_thr);
 463
 464       if (dt_list_ [i]->start_time () != 0 &&
 465           (elapsed_time.sec () < dt_list_[i]->start_time ()))
 466         {
 467           time_t suspension_time = dt_list_[i]->start_time () - elapsed_time.sec ();
 468           ACE_OS::sprintf (buf, "suspension_time = %lu\n",
 469                            static_cast<unsigned long> (suspension_time));
 470           log [log_index++] = ACE_OS::strdup (buf);
 471           yield (suspension_time, dt_list_[i]);
 472         }
 473
 474       sched_param = this->sched_param (dt_list_ [i]->importance ());
 475       dt_list_ [i]->activate_task (current,
 476            sched_param.in (),
 477            flags,
 478            base_time_);
 479
 480       ++active_dt_count_;
 481     }
 482
 483   this->wait ();
 484
 485   current->end_scheduling_segment (name);
 486
 487   this->check_ifexit ();
 488 }
 489
 490 void
 491 DT_Creator::dt_ended ()
 492 {
 493   {
 494     ACE_GUARD (ACE_Lock, ace_mon, *state_lock_);
 495     --active_dt_count_;
 496     if (TAO_debug_level > 0)
 497       ACE_DEBUG ((LM_DEBUG, "Active dt count = %d\n",active_dt_count_));
 498     char buf [BUFSIZ];
 499     ACE_OS::sprintf (buf,"Active dt count = %d\n",active_dt_count_);
 500     log [log_index++] = ACE_OS::strdup (buf);
 501   }
 502   this->check_ifexit ();
 503 }
 504
 505 void
 506 DT_Creator::job_ended ()
 507 {
 508   {
 509     ACE_GUARD (ACE_Lock, ace_mon, *state_lock_);
 510     --active_job_count_;
 511     if (TAO_debug_level > 0)
 512       ACE_DEBUG ((LM_DEBUG, "Active job count = %d\n",active_job_count_));
 513     char buf [BUFSIZ];
 514     ACE_OS::sprintf (buf,"Active job count = %d\n",active_job_count_);
 515     log [log_index++] = ACE_OS::strdup (buf);
 516   }
 517
 518  this->check_ifexit ();
 519 }
 520
 521 void
 522 DT_Creator::check_ifexit ()
 523 {
 524   if (TAO_debug_level > 0)
 525     ACE_DEBUG ((LM_DEBUG,
 526     "Checking exit status Job# = %d DT# = %d\n",
 527     active_job_count_,
 528     active_dt_count_));
 529
 530   static int shutdown = 0;
 531
 532   {
 533     ACE_GUARD (ACE_Lock, ace_mon, *shutdown_lock_);
 534
 535     if (!shutdown)
 536       {
 537   // All tasks have finished and all jobs have been shutdown.
 538   if (active_dt_count_ == 0 && active_job_count_ == 0)
 539     {
 540       ACE_DEBUG ((LM_DEBUG, "Shutdown in progress ...\n"));
 541
 542       /*
 543       for (int i = 0; i < dt_count_; i++)
 544         {
 545     dt_list_[i]->dump_stats ();
 546         }
 547
 548       for (int i = 0; i < job_count_; i ++)
 549         {
 550     job_list_[i]->dump_stats ();
 551         }
 552       */
 553       TASK_STATS::instance ()->dump_samples (file_name_.c_str (),
 554                ACE_TEXT("#Schedule Output"));
 555
 556       shutdown = 1;
 557
 558       FILE* log_file = ACE_OS::fopen (log_file_name_.c_str (), "w");
 559
 560       if (log_file != 0)
 561         {
 562     // first dump what the caller has to say.
 563     ACE_OS::fprintf (log_file, "Log File\n");
 564
 565     for (int i = 0; i < log_index; i++)
 566       {
 567         ACE_OS::fprintf (log_file, "%s\n", log [i]);
 568       }
 569
 570     ACE_OS::fclose (log_file);
 571         }
 572       ACE_DEBUG ((LM_DEBUG,
 573       "Log File Ready\n"));
 574     }
 575       }
 576   }
 577 }
 578
 579 int
 580 DT_Creator::dt_count ()
 581 {
 582   return dt_count_;
 583 }
 584
 585 DT_Creator::DT_Creator ()
 586  : dt_list_ (0),
 587    poa_list_ (0),
 588    job_list_ (0),
 589    dt_count_ (0),
 590    poa_count_ (0),
 591    job_count_ (0),
 592    state_lock_ (0),
 593    shutdown_lock_ (0),
 594    active_dt_count_ (0),
 595    active_job_count_ (0),
 596    log (0),
 597    base_time_ (0),
 598    file_name_ (),
 599    log_file_name_ (),
 600    gsf_ (0),
 601    synch_ (0)
 602 {
 603 }
 604
 605 DT_Creator::~DT_Creator ()
 606 {
 607   for (int i = 0; i < log_index; ++i)
 608     ACE_OS::free (log[i]);
 609   delete[] log;
 610
 611   for (int i = 0; i < this->dt_count_; ++i)
 612     delete this->dt_list_[i];
 613   delete[] dt_list_;
 614   for (int i = 0; i < this->poa_count_; ++i)
 615     delete this->poa_list_[i];
 616   delete[] poa_list_;
 617   delete[] job_list_;
 618
 619   delete base_time_;
 620
 621   delete state_lock_;
 622   delete shutdown_lock_;
 623 }
 624
 625
 626 void
 627 DT_Creator::log_msg (char* msg)
 628 {
 629   log [log_index++] = ACE_OS::strdup (msg);
 630 }
 631
 632 CORBA::ORB_ptr
 633 DT_Creator::orb ()
 634 {
 635   return this->orb_.in ();
 636 }
 637
 638 void
 639 DT_Creator::orb (CORBA::ORB_ptr orb)
 640 {
 641   this->orb_ = CORBA::ORB::_duplicate (orb);
 642 }
 643
 644 ACE_Time_Value*
 645 DT_Creator::base_time ()
 646 {
 647   return this->base_time_;
 648 }
 649
 650 void
 651 DT_Creator::base_time (ACE_Time_Value* base_time)
 652 {
 653   this->base_time_ = base_time;
 654 }
 655
 656
 657 Synch_i*
 658 DT_Creator::synch ()
 659 {
 660   return this->synch_.in ();
 661 }