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 (void)
 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
 210   ACE_CString synch_name ("Synch");
 211   ACE_Time_Value timestamp = ACE_OS::gettimeofday ();
 212
 213   char buf [BUFSIZ];
 214   ACE_OS::sprintf(buf, "%lu", static_cast<unsigned long> (timestamp.sec ()));
 215   synch_name += buf;
 216
 217   name [0].id =
 218     CORBA::string_dup (synch_name.c_str ());
 219
 220   ACE_DEBUG ((LM_DEBUG,
 221         "Synch Name %C\n\n",
 222         synch_name.c_str ()));
 223
 224   ACE_NEW (synch_,
 225      Synch_i);
 226
 227   Synch_var synch = synch_->_this ();
 228
 229   // Register the synch object with the Synch context.
 230   synch_context->rebind (name,
 231        synch.in ());
 232
 233 }
 234
 235
 236 int
 237 DT_Creator::activate_root_poa (void)
 238 {
 239   CORBA::Object_var object =
 240     this->orb_->resolve_initial_references ("RootPOA");
 241
 242   PortableServer::POA_var root_poa =
 243     PortableServer::POA::_narrow (object.in ());
 244
 245   PortableServer::POAManager_var poa_manager =
 246     root_poa->the_POAManager ();
 247
 248   poa_manager->activate ();
 249
 250   return 0;
 251 }
 252
 253 void
 254 DT_Creator::activate_poa_list (void)
 255 {
 256   if (TAO_debug_level > 0)
 257     ACE_DEBUG ((LM_DEBUG,
 258     "DT_Creator::activate_poa_list\n"));
 259
 260   if (poa_count_ > 0)
 261     {
 262       CORBA::Object_var object =
 263         this->orb_->resolve_initial_references ("RTORB");
 264
 265       RTCORBA::RTORB_var rt_orb = RTCORBA::RTORB::_narrow (object.in ());
 266
 267       object = this->orb_->resolve_initial_references ("RootPOA");
 268
 269       PortableServer::POA_var root_poa =
 270         PortableServer::POA::_narrow (object.in ());
 271
 272       for (int i = 0; i < poa_count_; ++i)
 273         {
 274           poa_list_[i]->activate (rt_orb.in(), root_poa.in ());
 275         }
 276     }
 277 }
 278
 279 void
 280 DT_Creator::activate_job_list (void)
 281 {
 282
 283   if (TAO_debug_level > 0)
 284     ACE_DEBUG ((LM_DEBUG,
 285     "DT_Creator::activate_job_list\n"));
 286
 287   Job_i* job;
 288
 289   CORBA::Object_var object =
 290     this->orb_->resolve_initial_references ("RootPOA");
 291
 292   PortableServer::POA_var root_poa =
 293     PortableServer::POA::_narrow (object.in ());
 294
 295   for (int i = 0; i < job_count_; ++i)
 296     {
 297       ++active_job_count_;
 298
 299       job = job_list_[i];
 300
 301       if (TAO_debug_level > 0)
 302         ACE_DEBUG ((LM_DEBUG, "Activating job:%C\n", job->name ().c_str ()));
 303
 304       // find your poa
 305       PortableServer::POA_var host_poa =
 306         root_poa->find_POA (job->poa ().c_str (), 0);
 307
 308       PortableServer::ServantBase_var servant_var (job);
 309
 310       // Register with poa.
 311       PortableServer::ObjectId_var id;
 312
 313       id = host_poa->activate_object (job);
 314
 315       CORBA::Object_var server =
 316         host_poa->id_to_reference (id.in ());
 317
 318       CORBA::String_var ior =
 319         orb_->object_to_string (server.in ());
 320
 321       const ACE_CString &job_name = job->name ();
 322
 323       CosNaming::Name_var name =
 324         this->naming_->to_name (job_name.c_str ());
 325
 326       this->naming_->rebind (name.in (),
 327                              server.in ());
 328
 329
 330     } /* while */
 331
 332   ACE_DEBUG ((LM_DEBUG,
 333               "Activated Job List\n"));
 334 }
 335
 336 void
 337 DT_Creator::activate_schedule (void)
 338 {
 339   if (TAO_debug_level > 0)
 340     ACE_DEBUG ((LM_DEBUG,
 341     "Activating schedule, task count = %d\n",
 342     dt_count_));
 343
 344   Thread_Task* task;
 345
 346   for (int i = 0; i < dt_count_; ++i)
 347     {
 348       task = dt_list_[i];
 349
 350       if (task->dist ())
 351   {
 352     // resolve the object from the naming service
 353     CosNaming::Name name (1);
 354     name.length (1);
 355     name[0].id = CORBA::string_dup (task->job ());
 356
 357     CORBA::Object_var obj =
 358       this->naming_->resolve (name);
 359
 360     Job_var job = Job::_narrow (obj.in ());
 361
 362     //    if (TAO_debug_level > 0)
 363     // {
 364         // Check that the object is configured with some
 365         // PriorityModelPolicy.
 366         CORBA::Policy_var policy =
 367     job->_get_policy (RTCORBA::PRIORITY_MODEL_POLICY_TYPE);
 368
 369         RTCORBA::PriorityModelPolicy_var priority_policy =
 370     RTCORBA::PriorityModelPolicy::_narrow (policy.in ());
 371
 372         if (CORBA::is_nil (priority_policy.in ()))
 373     ACE_DEBUG ((LM_DEBUG,
 374           "ERROR: Priority Model Policy not exposed!\n"));
 375         else
 376     {
 377                   /*
 378       RTCORBA::PriorityModel priority_model =
 379         priority_policy->priority_model ();
 380
 381       if (priority_model == RTCORBA::CLIENT_PROPAGATED)
 382         ACE_DEBUG ((LM_DEBUG,
 383         "%C priority_model = RTCORBA::CLIENT_PROPAGATED\n", task->job ()));
 384       else
 385         ACE_DEBUG ((LM_DEBUG,
 386         "%C priority_model = RTCORBA::SERVER_DECLARED\n", task->job ()));
 387       */
 388     }
 389         //} /*  if (TAO_debug_level > 0) */
 390
 391     task->job (job.in ());
 392   }
 393     }
 394   if (TAO_debug_level > 0 && dt_count_ > 0)
 395     ACE_DEBUG ((LM_DEBUG,
 396     "Activated schedule, task count = %d\n",
 397     dt_count_));
 398
 399 }
 400
 401 int
 402 DT_Creator::resolve_naming_service (void)
 403 {
 404   CORBA::Object_var naming_obj =
 405     this->orb_->resolve_initial_references ("NameService");
 406
 407   // Need to check return value for errors.
 408   if (CORBA::is_nil (naming_obj.in ()))
 409     ACE_ERROR_RETURN ((LM_ERROR,
 410                        " (%P|%t) Unable to resolve the Naming Service.\n"),
 411                       -1);
 412
 413   this->naming_ =
 414     CosNaming::NamingContextExt::_narrow (naming_obj.in ());
 415
 416   return 0;
 417 }
 418
 419 void
 420 DT_Creator::create_distributable_threads (RTScheduling::Current_ptr current)
 421 {
 422   long flags;
 423   flags = THR_NEW_LWP | THR_JOINABLE;
 424   flags |=
 425     orb_->orb_core ()->orb_params ()->scope_policy () |
 426     orb_->orb_core ()->orb_params ()->sched_policy ();
 427
 428   ACE_DEBUG ((LM_DEBUG,
 429               "Waiting to Synch\n"));
 430
 431   while (!this->synch ()->synched ())
 432     {
 433       try
 434         {
 435           this->orb_->perform_work ();
 436         }
 437       catch (const CORBA::Exception &)
 438         {
 439           return;
 440         }
 441     }
 442
 443   CORBA::Policy_var sched_param;
 444   sched_param = this->sched_param (100);
 445   const char * name = 0;
 446   current->begin_scheduling_segment (name,
 447               sched_param.in (),
 448               sched_param.in ());
 449
 450   ACE_NEW (base_time_,
 451      ACE_Time_Value (*(this->synch ()->base_time ())));
 452
 453   for (int i = 0; i < this->dt_count_; i++)
 454     {
 455       ACE_Time_Value now (ACE_OS::gettimeofday ());
 456
 457       ACE_Time_Value elapsed_time =  now - *base_time_;
 458
 459       char buf [BUFSIZ];
 460       ACE_OS::sprintf (buf, "elapsed time = %d\n now = %d\n base_time_ = %d\n",
 461            (int) elapsed_time.sec (),
 462            (int) now.sec (),
 463            (int) base_time_->sec());
 464
 465       log [log_index++] = ACE_OS::strdup (buf) ;
 466
 467       ACE_hthread_t curr_thr;
 468       ACE_Thread::self (curr_thr);
 469
 470       if (dt_list_ [i]->start_time () != 0 &&
 471           (elapsed_time.sec () < dt_list_[i]->start_time ()))
 472         {
 473           time_t suspension_time = dt_list_[i]->start_time () - elapsed_time.sec ();
 474           ACE_OS::sprintf (buf, "suspension_time = %lu\n",
 475                            static_cast<unsigned long> (suspension_time));
 476           log [log_index++] = ACE_OS::strdup (buf);
 477           yield (suspension_time, dt_list_[i]);
 478         }
 479
 480       sched_param = this->sched_param (dt_list_ [i]->importance ());
 481       dt_list_ [i]->activate_task (current,
 482            sched_param.in (),
 483            flags,
 484            base_time_);
 485
 486       ++active_dt_count_;
 487     }
 488
 489   this->wait ();
 490
 491   current->end_scheduling_segment (name);
 492
 493   this->check_ifexit ();
 494 }
 495
 496 void
 497 DT_Creator::dt_ended (void)
 498 {
 499   {
 500     ACE_GUARD (ACE_Lock, ace_mon, *state_lock_);
 501     --active_dt_count_;
 502     if (TAO_debug_level > 0)
 503       ACE_DEBUG ((LM_DEBUG, "Active dt count = %d\n",active_dt_count_));
 504     char buf [BUFSIZ];
 505     ACE_OS::sprintf (buf,"Active dt count = %d\n",active_dt_count_);
 506     log [log_index++] = ACE_OS::strdup (buf);
 507   }
 508   this->check_ifexit ();
 509 }
 510
 511 void
 512 DT_Creator::job_ended (void)
 513 {
 514   {
 515     ACE_GUARD (ACE_Lock, ace_mon, *state_lock_);
 516     --active_job_count_;
 517     if (TAO_debug_level > 0)
 518       ACE_DEBUG ((LM_DEBUG, "Active job count = %d\n",active_job_count_));
 519     char buf [BUFSIZ];
 520     ACE_OS::sprintf (buf,"Active job count = %d\n",active_job_count_);
 521     log [log_index++] = ACE_OS::strdup (buf);
 522   }
 523
 524  this->check_ifexit ();
 525 }
 526
 527 void
 528 DT_Creator::check_ifexit (void)
 529 {
 530   if (TAO_debug_level > 0)
 531     ACE_DEBUG ((LM_DEBUG,
 532     "Checking exit status Job# = %d DT# = %d\n",
 533     active_job_count_,
 534     active_dt_count_));
 535
 536   static int shutdown = 0;
 537
 538   {
 539     ACE_GUARD (ACE_Lock, ace_mon, *shutdown_lock_);
 540
 541     if (!shutdown)
 542       {
 543   // All tasks have finished and all jobs have been shutdown.
 544   if (active_dt_count_ == 0 && active_job_count_ == 0)
 545     {
 546       ACE_DEBUG ((LM_DEBUG, "Shutdown in progress ...\n"));
 547
 548       /*
 549       for (int i = 0; i < dt_count_; i++)
 550         {
 551     dt_list_[i]->dump_stats ();
 552         }
 553
 554       for (int i = 0; i < job_count_; i ++)
 555         {
 556     job_list_[i]->dump_stats ();
 557         }
 558       */
 559       TASK_STATS::instance ()->dump_samples (file_name_.c_str (),
 560                ACE_TEXT("#Schedule Output"));
 561
 562       shutdown = 1;
 563
 564       FILE* log_file = ACE_OS::fopen (log_file_name_.c_str (), "w");
 565
 566       if (log_file != 0)
 567         {
 568     // first dump what the caller has to say.
 569     ACE_OS::fprintf (log_file, "Log File\n");
 570
 571     for (int i = 0; i < log_index; i++)
 572       {
 573         ACE_OS::fprintf (log_file, "%s\n", log [i]);
 574       }
 575
 576     ACE_OS::fclose (log_file);
 577         }
 578       ACE_DEBUG ((LM_DEBUG,
 579       "Log File Ready\n"));
 580
 581     }
 582       }
 583   }
 584 }
 585
 586 int
 587 DT_Creator::dt_count (void)
 588 {
 589   return dt_count_;
 590 }
 591
 592 DT_Creator::DT_Creator (void)
 593  : dt_list_ (0),
 594    poa_list_ (0),
 595    job_list_ (0),
 596    dt_count_ (0),
 597    poa_count_ (0),
 598    job_count_ (0),
 599    state_lock_ (0),
 600    shutdown_lock_ (0),
 601    active_dt_count_ (0),
 602    active_job_count_ (0),
 603    log (0),
 604    base_time_ (0),
 605    file_name_ (),
 606    log_file_name_ (),
 607    gsf_ (0),
 608    synch_ (0)
 609 {
 610 }
 611
 612 DT_Creator::~DT_Creator (void)
 613 {
 614   for (int i = 0; i < log_index; ++i)
 615     ACE_OS::free (log[i]);
 616   delete[] log;
 617
 618   for (int i = 0; i < this->dt_count_; ++i)
 619     delete this->dt_list_[i];
 620   delete[] dt_list_;
 621   for (int i = 0; i < this->poa_count_; ++i)
 622     delete this->poa_list_[i];
 623   delete[] poa_list_;
 624   delete[] job_list_;
 625
 626   delete base_time_;
 627
 628   delete state_lock_;
 629   delete shutdown_lock_;
 630 }
 631
 632
 633 void
 634 DT_Creator::log_msg (char* msg)
 635 {
 636   log [log_index++] = ACE_OS::strdup (msg);
 637 }
 638
 639 CORBA::ORB_ptr
 640 DT_Creator::orb (void)
 641 {
 642   return this->orb_.in ();
 643 }
 644
 645 void
 646 DT_Creator::orb (CORBA::ORB_ptr orb)
 647 {
 648   this->orb_ = CORBA::ORB::_duplicate (orb);
 649 }
 650
 651 ACE_Time_Value*
 652 DT_Creator::base_time (void)
 653 {
 654   return this->base_time_;
 655 }
 656
 657 void
 658 DT_Creator::base_time (ACE_Time_Value* base_time)
 659 {
 660   this->base_time_ = base_time;
 661 }
 662
 663
 664 Synch_i*
 665 DT_Creator::synch (void)
 666 {
 667   return this->synch_.in ();
 668 }