TAO/examples/Simulator/Event_Supplier/DualEC_Sup.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    DualEC_Sup.cpp
   5  *
   6  * Event Supplier for visualizing scheduling behavior, using arrival
   7  * and dispatch data logged by an event channel dispatch command object
   8  *
   9  *  @author Chris Gill (cdgill@cs.wustl.edu) Adapted from the DOVE simulation event supplier originally David Levine (levine@cs.wustl.edu) and Tim Harrison (harrison@cs.wustl.edu) modified Michael Kircher (mk1@cs.wustl.edu)
  10  */
  11 //=============================================================================
  12
  13
  14 #include "DualEC_Sup.h"
  15 #include "NavWeapC.h"
  16
  17 #include "orbsvcs/Event_Utilities.h"
  18 #include "orbsvcs/Event_Service_Constants.h"
  19 #include "orbsvcs/Event/EC_Event_Channel.h"
  20 #include "orbsvcs/Sched/Config_Scheduler.h"
  21 #include "orbsvcs/Runtime_Scheduler.h"
  22 #include "orbsvcs/RtecEventChannelAdminC.h"
  23
  24 #include "tao/Utils/ORB_Manager.h"
  25 #include "tao/ORB_Core.h"
  26
  27 #include "ace/Get_Opt.h"
  28 #include "ace/Sched_Params.h"
  29 #include "ace/OS_NS_errno.h"
  30 #include "ace/OS_NS_ctype.h"
  31
  32 //FUZZ: disable check_for_lack_ACE_OS
  33 static const char usage [] =
  34 "[[-?]\n"
  35 "                 -f <name of input data file>\n"
  36 "                 [-O[RBport] ORB port number]\n"
  37 "                 [-m <count> of messages to send (2000)]\n"
  38 "                 [-b <count> at which to break navigation event\n"
  39 "                             stream out onto its own channel (1000)]\n"
  40 "                 [-n <usec> pause between navigation events (100000)]\n"
  41 "                 [-w <usec> pause between weapons events (100000)]\n"
  42 "                 [-d to dump scheduler header files]\n"
  43 "                 [-s to suppress data updates by EC]\n"
  44 "                 [-r to use runtime schedulers]\n"
  45 "                 [-p to suppress prioritization of operations]\n";
  46 //FUZZ: enable check_for_lack_ACE_OS
  47
  48 DualEC_Supplier::DualEC_Supplier (int argc, ACE_TCHAR** argv)
  49 : nav_pause_ (0, 100000),
  50   weap_pause_ (0, 100000),
  51   channel_hi_name_ (1),
  52   channel_lo_name_ (1),
  53   sched_hi_name_ (1),
  54   sched_lo_name_ (1),
  55   sched_hi_impl_ (0),
  56   sched_lo_impl_ (0),
  57   ec_hi_impl_ (0),
  58   ec_lo_impl_ (0),
  59   argc_(argc),
  60   argv_(argv),
  61   total_messages_ (2000),
  62   break_count_(-1),
  63   input_file_name_(0),
  64   update_data_ (1),
  65   dump_schedule_headers_ (0),
  66   use_runtime_schedulers_ (0),
  67   suppress_priority_ (0),
  68   hi_schedule_file_name_ (ACE_TEXT("DualEC_Runtime_Hi.h")),
  69   lo_schedule_file_name_ (ACE_TEXT("DualEC_Runtime_Lo.h")),
  70   nav_roll_ (0),
  71   nav_pitch_ (0)
  72 {
  73   try
  74     {
  75       this->sched_hi_name_.length (1);
  76       this->sched_hi_name_[0].id = CORBA::string_dup ("DUAL_SCHED_HI");
  77
  78       this->sched_lo_name_.length (1);
  79       this->sched_lo_name_[0].id = CORBA::string_dup ("DUAL_SCHED_LO");
  80
  81       this->channel_hi_name_.length (1);
  82       this->channel_hi_name_[0].id = CORBA::string_dup ("DUAL_EC_HI");
  83
  84       this->channel_lo_name_.length (1);
  85       this->channel_lo_name_[0].id = CORBA::string_dup ("DUAL_EC_LO");
  86
  87       this->terminator_ = terminator_impl_._this ();
  88     }
  89   catch (const CORBA::Exception& ex)
  90   {
  91       ex._tao_print_exception (
  92         "DualEC_Supplier::DualEC_Supplier : could "
  93         "not resolve reference to terminator");
  94   }
  95
  96   // Initialize the high priority RT_Info data
  97   rt_info_nav_hi_.entry_point = "DUALEC_NAV_HI";
  98   rt_info_nav_hi_.criticality = RtecScheduler::VERY_HIGH_CRITICALITY;
  99   rt_info_nav_hi_.worst_case_execution_time = ORBSVCS_Time::zero ();
 100   rt_info_nav_hi_.typical_execution_time = ORBSVCS_Time::zero ();
 101   rt_info_nav_hi_.cached_execution_time = ORBSVCS_Time::zero ();
 102   rt_info_nav_hi_.period = 2500000;
 103   rt_info_nav_hi_.importance = RtecScheduler::VERY_HIGH_IMPORTANCE;
 104   rt_info_nav_hi_.quantum = ORBSVCS_Time::zero ();
 105   rt_info_nav_hi_.threads = 1;
 106   rt_info_nav_hi_.info_type = RtecScheduler::OPERATION;
 107   rt_info_weap_hi_ = rt_info_nav_hi_;
 108   rt_info_weap_hi_.entry_point = "DUALEC_WEAP_HI";
 109   rt_info_dummy_hi_ = rt_info_nav_hi_;
 110   rt_info_dummy_hi_.entry_point = "DUALEC_DUMMY_HI";
 111
 112   // Initialize the low priority RT_Info data
 113   rt_info_nav_lo_.entry_point = "DUALEC_NAV_LO";
 114   rt_info_nav_lo_.criticality = RtecScheduler::VERY_LOW_CRITICALITY;
 115   rt_info_nav_lo_.worst_case_execution_time = ORBSVCS_Time::zero ();
 116   rt_info_nav_lo_.typical_execution_time = ORBSVCS_Time::zero ();
 117   rt_info_nav_lo_.cached_execution_time = ORBSVCS_Time::zero ();
 118   rt_info_nav_lo_.period = 10000000;
 119   rt_info_nav_lo_.importance = RtecScheduler::VERY_LOW_IMPORTANCE;
 120   rt_info_nav_lo_.quantum = ORBSVCS_Time::zero ();
 121   rt_info_nav_lo_.threads = 1;
 122   rt_info_nav_lo_.info_type = RtecScheduler::OPERATION;
 123   rt_info_weap_lo_ = rt_info_nav_lo_;
 124   rt_info_weap_lo_.entry_point = "DUALEC_WEAP_LO";
 125   rt_info_dummy_lo_ = rt_info_nav_lo_;
 126   rt_info_dummy_lo_.entry_point = "DUALEC_DUMMY_LO";
 127 }
 128
 129 DualEC_Supplier::~DualEC_Supplier ()
 130 {
 131   try
 132     {
 133       this->navigation_Supplier_.disconnect ();
 134       this->weapons_Supplier_.disconnect ();
 135
 136       // Unbind the schedulers from the NS.
 137       this->naming_context_->unbind (this->sched_hi_name_);
 138       this->naming_context_->unbind (this->sched_lo_name_);
 139
 140       // Unbind the ECs from the NS.
 141       this->naming_context_->unbind (this->channel_hi_name_);
 142       this->naming_context_->unbind (this->channel_lo_name_);
 143     }
 144   catch (const CORBA::Exception& ex)
 145     {
 146       ex._tao_print_exception ("DualEC_Supplier::~DualEC_Supplier");
 147     }
 148
 149   // @@TBD - destroy the ECs
 150   // @@TBD - destroy the schedulers
 151 }
 152
 153 int
 154 DualEC_Supplier::init ()
 155 {
 156   this->get_options (argc_, argv_);
 157
 158   try
 159   {
 160     // Connect to the RootPOA.
 161     CORBA::Object_var poaObject_var =
 162       TAO_ORB_Core_instance()->orb()->resolve_initial_references("RootPOA");
 163
 164     if (CORBA::is_nil (poaObject_var.in ()))
 165       ACE_ERROR_RETURN ((LM_ERROR,
 166                         " (%P|%t) Unable to initialize the POA.\n"),
 167                         1);
 168
 169     this->root_POA_var_ =
 170       PortableServer::POA::_narrow (poaObject_var.in ());
 171
 172     this->poa_manager_ =
 173        root_POA_var_->the_POAManager ();
 174
 175     poa_manager_->activate ();
 176
 177     // Get the Naming Service object reference.
 178     CORBA::Object_var namingObj_var =
 179       TAO_ORB_Core_instance()->orb()->resolve_initial_references (
 180           "NameService");
 181
 182     if (CORBA::is_nil (namingObj_var.in ()))
 183       ACE_ERROR_RETURN ((LM_ERROR,
 184                         " (%P|%t) Unable to get the Naming Service.\n"),
 185                         -1);
 186
 187     this->naming_context_ =
 188       CosNaming::NamingContext::_narrow (namingObj_var.in ());
 189   }
 190   catch (const CORBA::Exception& ex)
 191   {
 192     ex._tao_print_exception ("DualEC_Supplier::init");
 193     return -1;
 194   }
 195
 196   // Create two scheduling service instances.
 197   if (this->create_schedulers () == -1)
 198     {
 199       ACE_ERROR_RETURN ((LM_ERROR,
 200                          "Could not create schedulers"),
 201                         -1);
 202     }
 203
 204   // Create two event channels.
 205   if (this->create_event_channels () == -1)
 206     {
 207       ACE_ERROR_RETURN ((LM_ERROR,
 208                          "Could not create event channels"),
 209                         -1);
 210     }
 211
 212   // Connect suppliers to the respective event channels.
 213   ACE_Scheduler_Factory::POD_RT_Info * rt_info_nav_hi =
 214     (suppress_priority_) ? 0 : &rt_info_nav_hi_;
 215   ACE_Scheduler_Factory::POD_RT_Info * rt_info_weap_hi =
 216     (suppress_priority_) ? 0 : &rt_info_weap_hi_;
 217   ACE_Scheduler_Factory::POD_RT_Info * rt_info_nav_lo =
 218     (suppress_priority_) ? 0 : &rt_info_nav_lo_;
 219   ACE_Scheduler_Factory::POD_RT_Info * rt_info_weap_lo =
 220     (suppress_priority_) ? 0 : &rt_info_weap_lo_;
 221
 222   if (this->navigation_Supplier_.connect ("MIB_unknown",
 223                                           "DUAL_EC_HI",
 224                                           "DUAL_SCHED_HI",
 225                                            rt_info_nav_hi) == -1)
 226     {
 227       ACE_ERROR_RETURN ((LM_ERROR,
 228                          "Could not connect navigation supplier to DUAL_EC_HI"),
 229                         -1);
 230     }
 231
 232  if (this->navigation_Supplier_.connect ("MIB_unknown",
 233                                           "DUAL_EC_LO",
 234                                           "DUAL_SCHED_LO",
 235                                            rt_info_nav_lo) == -1)
 236     {
 237       ACE_ERROR_RETURN ((LM_ERROR,
 238                          "Could not connect navigation supplier to DUAL_EC_LO"),
 239                         -1);
 240     }
 241
 242   if (this->weapons_Supplier_.connect ("MIB_unknown",
 243                                        "DUAL_EC_HI",
 244                                        "DUAL_SCHED_HI",
 245                                        rt_info_weap_hi) == -1)
 246     {
 247       ACE_ERROR_RETURN ((LM_ERROR,
 248                          "Could not connect weapons supplier to DUAL_EC_HI"),
 249                         -1);
 250     }
 251
 252  if (this->weapons_Supplier_.connect ("MIB_unknown",
 253                                       "DUAL_EC_LO",
 254                                       "DUAL_SCHED_LO",
 255                                       rt_info_weap_lo) == -1)
 256     {
 257       ACE_ERROR_RETURN ((LM_ERROR,
 258                          "Could not connect weapons supplier to DUAL_EC_LO"),
 259                         -1);
 260     }
 261
 262   return 0;
 263 }
 264
 265 // Private class that implements a termination servant.
 266
 267 void
 268 DualEC_Supplier::Terminator::shutdown ()
 269 {
 270   TAO_ORB_Core_instance ()->orb ()->shutdown ();
 271 }
 272
 273
 274 // Run the ORB event loop.
 275
 276 ACE_THR_FUNC_RETURN
 277 DualEC_Supplier::run_orb (void *)
 278 {
 279   TAO_ORB_Core_instance ()->orb ()->run ();
 280   return 0;
 281 }
 282
 283
 284 // Run navigation event generation thread.
 285
 286 ACE_THR_FUNC_RETURN
 287 DualEC_Supplier::run_nav_thread (void *arg)
 288 {
 289   DualEC_Supplier * sup =
 290     static_cast<DualEC_Supplier *> (arg);
 291
 292   try
 293     {
 294       ACE_Unbounded_Queue_Iterator<Navigation *>
 295         nav_iter (sup->navigation_data_);
 296
 297       if (nav_iter.done ())
 298         {
 299           ACE_ERROR_RETURN ((LM_ERROR,
 300                              "DualEC_Supplier::run_event_thread: "
 301                              "there is no navigation data\n"), 0);
 302         }
 303
 304       CORBA::Any any;
 305
 306       long total_sent = 0;
 307
 308       do
 309       {
 310         // Insert the event data
 311         Navigation **nav;
 312
 313         if ((nav_iter.next (nav)) && (nav) && (*nav))
 314           {
 315             any <<= *nav;
 316
 317             // Sleep briefly to avoid too much livelock (a little is good).
 318             ACE_OS::sleep (sup->nav_pause_);
 319
 320             // If the break count has been reached, change the
 321             // channel that is being used by the NAV supplier
 322             if (total_sent == sup->break_count_)
 323               {
 324                ACE_DEBUG ((LM_DEBUG,
 325                            "breaking out nav at event: %d\n",
 326                            sup->break_count_));
 327
 328                 sup->navigation_Supplier_.use_next_connection ();
 329               }
 330
 331             sup->navigation_Supplier_.notify (any);
 332           }
 333         else
 334           {
 335             ACE_ERROR ((LM_ERROR,
 336                         "DualEC_Supplier::run_nav_thread:"
 337                         "Could Not access navigation data"));
 338           }
 339
 340         if (total_sent < 5)
 341           ACE_DEBUG ((LM_DEBUG,
 342                       "Pushing event data.\n"));
 343         else if (total_sent == 5)
 344           ACE_DEBUG ((LM_DEBUG,
 345                       "Everything is running. Going to be mute.\n"));
 346
 347         nav_iter.advance ();
 348
 349         if (nav_iter.done ())
 350           nav_iter.first ();
 351       }
 352       while (++total_sent < sup->total_messages_);
 353     }
 354   catch (const CORBA::Exception&)
 355     {
 356     }
 357
 358   return 0;
 359 }
 360
 361
 362 // Run weapons event generation thread.
 363
 364 ACE_THR_FUNC_RETURN
 365 DualEC_Supplier::run_weap_thread (void *arg)
 366 {
 367   DualEC_Supplier * sup =
 368     static_cast<DualEC_Supplier *> (arg);
 369
 370   try
 371     {
 372       ACE_Unbounded_Queue_Iterator<Weapons *>
 373         weap_iter (sup->weapons_data_);
 374
 375       if (weap_iter.done ())
 376         {
 377           ACE_ERROR_RETURN ((LM_ERROR,
 378                              "DualEC_Supplier::run_event_thread: "
 379                              "there is no navigation data\n"), 0);
 380         }
 381
 382       CORBA::Any any;
 383
 384       long total_sent = 0;
 385
 386       do
 387       {
 388         // Insert the event data
 389         Weapons **weap;
 390
 391         if ((weap_iter.next (weap)) && (weap) && (*weap))
 392           {
 393             any <<= *weap;
 394
 395             // Sleep briefly to avoid too much livelock (a little is good).
 396             ACE_OS::sleep (sup->weap_pause_);
 397
 398             sup->weapons_Supplier_.notify (any);
 399           }
 400         else
 401           {
 402             ACE_ERROR ((LM_ERROR,
 403                         "DualEC_Supplier::run_weap_thread:"
 404                         "Could Not access weapons data"));
 405           }
 406
 407         if (total_sent < 5)
 408           ACE_DEBUG ((LM_DEBUG,
 409                       "Pushing event data.\n"));
 410         else if (total_sent == 5)
 411           ACE_DEBUG ((LM_DEBUG,
 412                       "Everything is running. Going to be mute.\n"));
 413
 414         weap_iter.advance ();
 415
 416         if (weap_iter.done ())
 417           weap_iter.first ();
 418       }
 419       while (++total_sent < sup->total_messages_);
 420     }
 421   catch (const CORBA::Exception&)
 422     {
 423     }
 424
 425   return 0;
 426 }
 427
 428
 429 // Create two scheduling service instances, register
 430 // them with the Naming Service.
 431
 432 int
 433 DualEC_Supplier::create_schedulers ()
 434 {
 435   // @@TBD - look at a command line modified setting,
 436   // create either a runtime or a config scheduler for
 437   // each instance
 438
 439   try
 440     {
 441       if (use_runtime_schedulers_)
 442         {
 443           ACE_ERROR_RETURN ((LM_ERROR,
 444                              " (%P|%t) Runtime Schedulers not implemented\n"),
 445                             -1);
 446         }
 447       else
 448         {
 449           // Create Event Service Implementations, passing in the respective
 450           // Scheduling Service Implementations (which must already be created).
 451
 452           ACE_NEW_RETURN (this->sched_hi_impl_,
 453                           ACE_Config_Scheduler,
 454                           -1);
 455
 456           this->sched_hi_ = sched_hi_impl_->_this ();
 457
 458           ACE_NEW_RETURN (this->sched_lo_impl_,
 459                           ACE_Config_Scheduler,
 460                           -1);
 461
 462           this->sched_lo_ = sched_lo_impl_->_this ();
 463
 464           // Register Scheduling Service Implementations with Naming Service
 465
 466           this->naming_context_->bind (this ->sched_hi_name_,
 467                                        this->sched_hi_.in ());
 468
 469           naming_context_->bind (this->sched_lo_name_,
 470                                  this->sched_lo_.in ());
 471
 472           // Register high and low priority rt_infos with the
 473           // schedulers to force priority differentiation.
 474
 475           this->sched_hi_rt_info_hi_ =
 476             this->sched_hi_->
 477               create (this->rt_info_dummy_hi_.entry_point);
 478
 479
 480           this->sched_hi_->
 481             set (this->sched_hi_rt_info_hi_,
 482                  static_cast<RtecScheduler::Criticality_t> (this->rt_info_dummy_hi_.criticality),
 483                  this->rt_info_dummy_hi_.worst_case_execution_time,
 484                  this->rt_info_dummy_hi_.typical_execution_time,
 485                  this->rt_info_dummy_hi_.cached_execution_time,
 486                  this->rt_info_dummy_hi_.period,
 487                  static_cast<RtecScheduler::Importance_t> (this->rt_info_dummy_hi_.importance),
 488                  this->rt_info_dummy_hi_.quantum,
 489                  this->rt_info_dummy_hi_.threads,
 490                  static_cast<RtecScheduler::Info_Type_t> (this->rt_info_dummy_hi_.info_type));
 491
 492
 493           this->sched_hi_rt_info_lo_ =
 494             this->sched_hi_->
 495               create (this->rt_info_dummy_lo_.entry_point);
 496
 497
 498           this->sched_hi_->
 499             set (this->sched_hi_rt_info_lo_,
 500                  static_cast<RtecScheduler::Criticality_t> (this->rt_info_dummy_lo_.criticality),
 501                  this->rt_info_dummy_lo_.worst_case_execution_time,
 502                  this->rt_info_dummy_lo_.typical_execution_time,
 503                  this->rt_info_dummy_lo_.cached_execution_time,
 504                  this->rt_info_dummy_lo_.period,
 505                  static_cast<RtecScheduler::Importance_t> (this->rt_info_dummy_lo_.importance),
 506                  this->rt_info_dummy_lo_.quantum,
 507                  this->rt_info_dummy_lo_.threads,
 508                  static_cast<RtecScheduler::Info_Type_t> (this->rt_info_dummy_lo_.info_type));
 509
 510
 511           this->sched_hi_rt_info_hi_ =
 512             this->sched_lo_->
 513               create (this->rt_info_dummy_hi_.entry_point);
 514
 515
 516           this->sched_lo_->
 517             set (this->sched_hi_rt_info_hi_,
 518                  static_cast<RtecScheduler::Criticality_t> (this->rt_info_dummy_hi_.criticality),
 519                  this->rt_info_dummy_hi_.worst_case_execution_time,
 520                  this->rt_info_dummy_hi_.typical_execution_time,
 521                  this->rt_info_dummy_hi_.cached_execution_time,
 522                  this->rt_info_dummy_hi_.period,
 523                  static_cast<RtecScheduler::Importance_t> (this->rt_info_dummy_hi_.importance),
 524                  this->rt_info_dummy_hi_.quantum,
 525                  this->rt_info_dummy_hi_.threads,
 526                  static_cast<RtecScheduler::Info_Type_t> (this->rt_info_dummy_hi_.info_type));
 527
 528
 529           this->sched_hi_rt_info_lo_ =
 530             this->sched_lo_->
 531               create (this->rt_info_dummy_lo_.entry_point);
 532
 533
 534           this->sched_lo_->
 535             set (this->sched_hi_rt_info_lo_,
 536                  static_cast<RtecScheduler::Criticality_t> (this->rt_info_dummy_lo_.criticality),
 537                  this->rt_info_dummy_lo_.worst_case_execution_time,
 538                  this->rt_info_dummy_lo_.typical_execution_time,
 539                  this->rt_info_dummy_lo_.cached_execution_time,
 540                  this->rt_info_dummy_lo_.period,
 541                  static_cast<RtecScheduler::Importance_t> (this->rt_info_dummy_lo_.importance),
 542                  this->rt_info_dummy_lo_.quantum,
 543                  this->rt_info_dummy_lo_.threads,
 544                  static_cast<RtecScheduler::Info_Type_t> (this->rt_info_dummy_lo_.info_type));
 545         }
 546     }
 547   catch (const CORBA::Exception& ex)
 548     {
 549       ex._tao_print_exception ("DualEC_Supplier::create_schedulers");
 550       return -1;
 551     }
 552
 553   return 0;
 554 }
 555
 556
 557 // Create two event service instances, registers
 558 // them with the Naming Service.
 559
 560 int
 561 DualEC_Supplier::create_event_channels ()
 562 {
 563   try
 564     {
 565       // Create Event Service Implementations, passing in the respective
 566       // Scheduling Service Implementations (which must already be created).
 567       TAO_EC_Event_Channel_Attributes attr_high (root_POA_var_.in (),
 568                                                  root_POA_var_.in ());
 569
 570       attr_high.scheduler = sched_hi_.in ();
 571
 572       ACE_NEW_RETURN (this->ec_hi_impl_,
 573                       TAO_EC_Event_Channel (attr_high),
 574                       -1);
 575
 576       this->ec_hi_ = ec_hi_impl_->_this ();
 577
 578       TAO_EC_Event_Channel_Attributes attr_low (root_POA_var_.in (),
 579                                                 root_POA_var_.in ());
 580
 581       attr_low.scheduler = sched_lo_.in ();
 582
 583       ACE_NEW_RETURN (this->ec_lo_impl_,
 584                       TAO_EC_Event_Channel (attr_low),
 585                       -1);
 586
 587       this->ec_lo_ = ec_lo_impl_->_this ();
 588
 589       // Register Event Service Implementations with Naming Service
 590
 591       naming_context_->bind (this->channel_hi_name_,
 592                              this->ec_hi_.in ());
 593
 594       naming_context_->bind (this->channel_lo_name_,
 595                                      this->ec_lo_.in ());
 596   }
 597   catch (const CORBA::Exception& ex)
 598   {
 599     ex._tao_print_exception ("DualEC_Supplier::create_event_channels");
 600     return -1;
 601   }
 602
 603   return 0;
 604 }
 605
 606 void
 607 DualEC_Supplier::compute_schedules ()
 608 {
 609   try
 610     {
 611       sched_hi_->compute_scheduling
 612         (ACE_Sched_Params::priority_min (ACE_SCHED_FIFO,
 613                                          ACE_SCOPE_THREAD),
 614          ACE_Sched_Params::priority_max (ACE_SCHED_FIFO,
 615                                          ACE_SCOPE_THREAD),
 616          this->infos_hi_.out (),
 617          this->deps_hi_.out (),
 618          this->configs_hi_.out (),
 619          this->anomalies_hi_.out ());
 620
 621       sched_lo_->compute_scheduling
 622         (ACE_Sched_Params::priority_min (ACE_SCHED_FIFO,
 623                                          ACE_SCOPE_THREAD),
 624          ACE_Sched_Params::priority_max (ACE_SCHED_FIFO,
 625                                          ACE_SCOPE_THREAD),
 626          this->infos_lo_.out (),
 627          this->deps_hi_.out (),
 628          this->configs_lo_.out (),
 629          this->anomalies_lo_.out ());
 630
 631       if (dump_schedule_headers_ && (this->hi_schedule_file_name_ != 0))
 632         {
 633           ACE_Scheduler_Factory::dump_schedule (infos_hi_.in (),
 634                                                 deps_hi_.in (),
 635                                                 configs_hi_.in (),
 636                                                 anomalies_hi_.in (),
 637                                                 this->hi_schedule_file_name_);
 638         }
 639
 640       if (dump_schedule_headers_ && (this->lo_schedule_file_name_ != 0))
 641         {
 642           ACE_Scheduler_Factory::dump_schedule (infos_lo_.in (),
 643                                                 deps_lo_.in (),
 644                                                 configs_lo_.in (),
 645                                                 anomalies_lo_.in (),
 646                                                 this->lo_schedule_file_name_);
 647         }
 648     }
 649   catch (const CORBA::Exception& ex)
 650     {
 651     ex._tao_print_exception ("SYS_EX");
 652     }
 653 }
 654
 655 void
 656 DualEC_Supplier::start_generating_events ()
 657 {
 658   try
 659     {
 660       // Spawn a thread that runs the orb event loop
 661       ACE_Thread_Manager orb_thread_manager;
 662       orb_thread_manager.spawn (DualEC_Supplier::run_orb);
 663
 664       // Compute the scheduling information, based on the provided RT_Infos.
 665       this->compute_schedules ();
 666
 667
 668       // Load the scheduling data for the simulation.
 669       this->load_schedule_data ();
 670
 671       // Sleep for 10 seconds to give time for registrations.
 672       ACE_DEBUG ((LM_DEBUG,
 673                   "\nDUAL_SCHED_HI, DUAL_SCHED_LO, DUAL_EC_HI and "
 674                   "DUAL_EC_LO are registered with the Naming Service.\n"
 675                   "Sleeping 10 seconds before generating events\n"));
 676       ACE_Time_Value tv (10, 0);
 677       ACE_OS::sleep (tv);
 678
 679       // Spawn thread to run over the navigation data and generate events.
 680       ACE_Thread_Manager event_thread_manager;
 681       event_thread_manager.spawn (DualEC_Supplier::run_nav_thread,
 682                                   this);
 683
 684       // Spawn thread to run over the weapons data and generate events.
 685       event_thread_manager.spawn (DualEC_Supplier::run_weap_thread,
 686                                   this);
 687
 688       // Wait for the threads that are generating events.
 689       event_thread_manager.wait ();
 690
 691       // Shut down the ORB via the termination servant
 692       this->terminator_->shutdown ();
 693
 694       // Wait for the thread that runs the orb event loop.
 695       orb_thread_manager.wait ();
 696
 697       // Clean up the navigation data.
 698       ACE_Unbounded_Queue_Iterator<Navigation *>
 699         nav_iter (this->navigation_data_);
 700       Navigation **nav_temp;
 701       for (nav_iter.first ();
 702            nav_iter.done () == 0;
 703            nav_iter.advance ())
 704         if (nav_iter.next (nav_temp) && nav_temp)
 705           delete (*nav_temp);
 706
 707       // Clean up the weapons data.
 708       ACE_Unbounded_Queue_Iterator<Weapons *>
 709         weap_iter (this->weapons_data_);
 710       Weapons **weap_temp;
 711       for (weap_iter.first ();
 712            weap_iter.done () == 0;
 713            weap_iter.advance ())
 714         if (weap_iter.next (weap_temp) && weap_temp)
 715           delete (*weap_temp);
 716     }
 717   catch (const CORBA::Exception& ex)
 718     {
 719       ex._tao_print_exception ("SYS_EX");
 720     }
 721
 722 }
 723
 724 void
 725 DualEC_Supplier::load_schedule_data ()
 726 {
 727   Navigation * nav = 0;
 728   Weapons * weap = 0;
 729
 730   Schedule_Viewer_Data data;
 731
 732   // constants for periods (in units of one hundred nanoseconds)
 733   const TimeBase::TimeT ONE_HZ_PERIOD = 10000000;
 734   // const TimeBase::TimeT FIVE_HZ_PERIOD = ONE_HZ_PERIOD / 5;
 735   // const TimeBase::TimeT TEN_HZ_PERIOD = ONE_HZ_PERIOD / 10;
 736   const TimeBase::TimeT TWENTY_HZ_PERIOD = ONE_HZ_PERIOD / 20;
 737
 738   if (this->input_file_name_)
 739     {
 740       // Open the scheduler data input file and read its contents into
 741       // a queue.
 742       FILE *input_file;
 743
 744       int scan_count = 0;
 745       input_file = ACE_OS::fopen(this->input_file_name_, "r");
 746
 747       if (input_file)
 748         {
 749           // Get a line at a time from the data file and parse it.
 750           char input_buf[BUFSIZ];
 751           while (ACE_OS::fgets (input_buf, BUFSIZ, input_file))
 752             {
 753               // Run through leading whitespace.
 754               char *temp = input_buf;
 755               while (*temp && ACE_OS::ace_isspace (*temp))
 756                 ++temp;
 757
 758               // If there is anything besides whitespace in the line
 759               // read, scan its fields into the scheduling data
 760               // structure.
 761               if (ACE_OS::strlen (temp) > 0)
 762                 {
 763                   scan_count = sscanf (temp, "%s %lf %lf %lu %lu %lu %lu",
 764                                        data.operation_name,
 765                                        &data.utilitzation,
 766                                        &data.overhead,
 767                                        &data.arrival_time,
 768                                        &data.deadline_time,
 769                                        &data.completion_time,
 770                                        &data.computation_time);
 771                   if (scan_count != 7)
 772                     {
 773                       ACE_ERROR ((LM_ERROR,
 774                                   "DOVE_Supplier::load_schedule_data: "
 775                                   "scanned incorrect number of data elements: %d\n", scan_count));
 776                       return;
 777                     }
 778
 779
 780                   if ((ACE_OS::strcmp(data.operation_name, "high_20") == 0) ||
 781                       (ACE_OS::strcmp(data.operation_name, "low_20") == 0))
 782                     {
 783                       ACE_NEW (weap, Weapons);
 784                       if (weap == 0)
 785                         {
 786                           ACE_ERROR ((LM_ERROR,
 787                                       "DOVE_Supplier::load_schedule_data: "
 788                                       "failed to allocate Weapons\n"));
 789                           return;
 790                         }
 791
 792                       weap->criticality = 1;
 793                       weap->deadline_time = TWENTY_HZ_PERIOD;
 794                       weap->number_of_weapons = 2;
 795                       weap->weapon1_identifier = CORBA::string_alloc (30);
 796                       ACE_OS::strcpy (weap->weapon1_identifier.inout (),"Photon Torpedoes");
 797                       weap->weapon1_status =(ACE_OS::rand() % 4) == 0 ? 0 : 1 ;
 798                       weap->weapon2_identifier = CORBA::string_alloc (30);
 799                       ACE_OS::strcpy (weap->weapon2_identifier.inout (),"Quantum Torpedoes");
 800                       weap->weapon2_status = (ACE_OS::rand() % 4) == 0 ? 0 : 1;
 801                       weap->weapon3_identifier = CORBA::string_alloc (1);
 802                       ACE_OS::strcpy (weap->weapon3_identifier.inout (), "");
 803                       weap->weapon3_status = 0;
 804                       weap->weapon4_identifier = CORBA::string_alloc (1);
 805                       ACE_OS::strcpy (weap->weapon4_identifier.inout (), "");
 806                       weap->weapon4_status = 0;
 807                       weap->weapon5_identifier = CORBA::string_alloc (1);
 808                       ACE_OS::strcpy (weap->weapon5_identifier.inout (), "");
 809                       weap->weapon5_status = 0;
 810                       weap->utilization =       0.0;
 811                       weap->overhead =          0.0;
 812                       weap->arrival_time =      ORBSVCS_Time::zero ();
 813                       weap->completion_time  =  ORBSVCS_Time::zero ();
 814                       weap->computation_time =  ORBSVCS_Time::zero ();
 815                       weap->update_data =       update_data_;
 816
 817                       // Insert the data into the queue.
 818                       weapons_data_.enqueue_tail (weap);
 819                     }
 820                   else
 821                     {
 822                       ACE_NEW (nav, Navigation);
 823                       if (nav == 0)
 824                         {
 825                           ACE_ERROR ((LM_ERROR,
 826                                       "DOVE_Supplier::load_schedule_data: "
 827                                   "failed to allocate Navigation\n"));
 828                           return;
 829                         }
 830
 831                       nav->criticality = 0;
 832                       nav->deadline_time = TWENTY_HZ_PERIOD;
 833                       nav->position_latitude = ACE_OS::rand() % 90;
 834                       nav->position_longitude = ACE_OS::rand() % 180;
 835                       nav->altitude = ACE_OS::rand() % 100;
 836                       nav->heading = ACE_OS::rand() % 180;
 837                       this->nav_roll_ = (this->nav_roll_ >= 180) ? -180 : this->nav_roll_ + 1;
 838                       nav->roll = this->nav_roll_;
 839                       this->nav_pitch_ =  (this->nav_pitch_ >= 90) ? -90 : this->nav_pitch_ + 1;
 840                       nav->pitch = this->nav_pitch_;
 841                       nav->utilization =       0.0;
 842                       nav->overhead =          0.0;
 843                       nav->arrival_time =      ORBSVCS_Time::zero ();
 844                       nav->completion_time =   ORBSVCS_Time::zero ();
 845                       nav->computation_time =  ORBSVCS_Time::zero ();
 846                       nav->update_data =       this->update_data_;
 847                       nav->utilization = (double) (20.0 + ACE_OS::rand() % 10);
 848                       nav->overhead = (double) (ACE_OS::rand() % 10);
 849
 850                       // Insert the data into the queue.
 851                       navigation_data_.enqueue_tail (nav);
 852                     }
 853                 }
 854             }
 855         }
 856       else
 857         {
 858           ACE_ERROR ((LM_ERROR,
 859                       "DOVE_Supplier::start_generating_events: "
 860                       "could not open input file [%s].\n",
 861                       this->input_file_name_));
 862           return;
 863         }
 864     }
 865 }
 866
 867
 868 // Get command line options.
 869
 870 unsigned int
 871 DualEC_Supplier::get_options (int argc, ACE_TCHAR *argv [])
 872 {
 873   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT("f:m:b:n:w:dsrp"));
 874   int opt;
 875   int temp;
 876
 877   while ((opt = get_opt ()) != -1)
 878     {
 879       switch (opt)
 880         {
 881         case 'f':
 882           this->input_file_name_ = get_opt.opt_arg ();
 883
 884           if (!this->input_file_name_ || ACE_OS::strlen (this->input_file_name_) > 0)
 885             ACE_DEBUG ((LM_DEBUG,"Reading file!\n"));
 886           else
 887             {
 888               this->input_file_name_ = 0;
 889               ACE_ERROR_RETURN ((LM_ERROR,
 890                                  "%s: file name must be specified with -f option",
 891                                  argv[0]),
 892                                 1);
 893             }
 894           break;
 895
 896         case 'm':
 897           temp = ACE_OS::atoi (get_opt.opt_arg ());
 898           if (temp > 0)
 899             {
 900               this->total_messages_ = (u_int) temp;
 901                ACE_DEBUG ((LM_DEBUG,
 902                            "Messages to send: %d\n",
 903                            this->total_messages_));
 904             }
 905           else
 906             ACE_ERROR_RETURN ((LM_ERROR,
 907                                "%s: message count must be > 0",
 908                                argv[0]),
 909                                1);
 910           break;
 911
 912         case 'b':
 913           temp = ACE_OS::atoi (get_opt.opt_arg ());
 914           if (temp > 0)
 915             {
 916               this->break_count_ = (u_int) temp;
 917                ACE_DEBUG ((LM_DEBUG,
 918                            "Break count: %d\n",
 919                            this->break_count_));
 920             }
 921           else
 922             ACE_ERROR_RETURN ((LM_ERROR,
 923                                "%s: break count must be > 0",
 924                                argv[0]),
 925                                1);
 926           break;
 927
 928         case 'n':
 929           temp = ACE_OS::atoi (get_opt.opt_arg ());
 930           if (temp >= 0)
 931             {
 932               this->nav_pause_ =
 933                 ACE_Time_Value(0, static_cast<long> (temp));
 934               ACE_DEBUG ((LM_DEBUG,
 935                           "Navigation pause: %d usec\n",
 936                           temp));
 937             }
 938           else
 939             ACE_ERROR_RETURN ((LM_ERROR,
 940                                "%s: navigation pause must be >= 0",
 941                                argv[0]),
 942                                1);
 943           break;
 944
 945
 946         case 'w':
 947           temp = ACE_OS::atoi (get_opt.opt_arg ());
 948           if (temp >= 0)
 949             {
 950               this->weap_pause_ =
 951                 ACE_Time_Value(0, static_cast<long> (temp));
 952               ACE_DEBUG ((LM_DEBUG,
 953                           "Weapons pause: %d usec\n",
 954                           temp));
 955             }
 956           else
 957             ACE_ERROR_RETURN ((LM_ERROR,
 958                                "%s: weapons pause must be >= 0",
 959                                argv[0]),
 960                                1);
 961           break;
 962
 963         case 'd':
 964
 965           this->dump_schedule_headers_ = 1;
 966           break;
 967
 968         case 's':
 969           update_data_ = 0;
 970           break;
 971
 972         case  'r':
 973           use_runtime_schedulers_ = 1;
 974           break;
 975
 976         case 'p':
 977           suppress_priority_ = 1;
 978           break;
 979
 980         default:
 981         case '?':
 982           ACE_DEBUG ((LM_DEBUG,
 983                       "Usage: %s %s\n",
 984                       argv[0], usage));
 985           ACE_OS::exit (0);
 986           break;
 987         }
 988     }
 989
 990   if (argc != get_opt.opt_ind ())
 991     ACE_ERROR_RETURN ((LM_ERROR,
 992                        "%s: too many arguments\n"
 993                        "Usage: %s %s\n",
 994                        argv[0],
 995                        argv[0],
 996                        usage),
 997                       1);
 998
 999   return 0;
1000 }
1001
1002 // function main
1003
1004 int
1005 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
1006 {
1007   // Enable FIFO scheduling
1008   int min_priority =
1009     ACE_Sched_Params::priority_min (ACE_SCHED_FIFO);
1010
1011   // Set all threads in the process into the RT scheduling class
1012   if (ACE_OS::sched_params (ACE_Sched_Params (ACE_SCHED_FIFO,
1013                                               min_priority,
1014                                               ACE_SCOPE_PROCESS)) != 0)
1015     {
1016       if (ACE_OS::last_error () == EPERM)
1017           ACE_DEBUG ((LM_DEBUG,
1018                       "%s: user is not superuser, "
1019                       "so remain in time-sharing class\n", argv[0]));
1020       else
1021           ACE_ERROR ((LM_ERROR,
1022                       "%s: ACE_OS::sched_params failed\n", argv[0]));
1023     }
1024
1025
1026   try
1027     {
1028       // Initialize ORB.
1029       TAO_ORB_Manager orb_Manager;
1030
1031       orb_Manager.init (argc,
1032                         argv);
1033
1034
1035       // Create the demo supplier.
1036       DualEC_Supplier *event_Supplier_ptr;
1037
1038       ACE_NEW_RETURN (event_Supplier_ptr,
1039                       DualEC_Supplier(argc, argv),
1040                       -1);
1041
1042       // Initialize everthing
1043       if (event_Supplier_ptr->init () == -1)
1044         ACE_OS::exit (1);
1045
1046       // now we can go ahead
1047       event_Supplier_ptr->start_generating_events ();
1048
1049       // when done, we clean up
1050       delete event_Supplier_ptr;
1051     }
1052   catch (const CORBA::Exception& ex)
1053     {
1054       ex._tao_print_exception ("SYS_EX");
1055     }
1056
1057   return 0;
1058 }
1059