TAO/examples/RTScheduling/MIF_Scheduler/MIF_Scheduler.cpp

   1 #include "MIF_Scheduler.h"
   2 #include "ace/Atomic_Op.h"
   3 #include "tao/RTScheduling/Request_Interceptor.h"
   4 #include "test.h"
   5
   6 ACE_Atomic_Op<TAO_SYNCH_MUTEX, long> server_guid_counter;
   7
   8 DT::DT (TAO_SYNCH_MUTEX &lock,
   9         int guid)
  10   : dt_cond_ (lock),
  11     guid_ (guid),
  12     eligible_ (0)
  13 {
  14 }
  15
  16 void
  17 DT::suspend (void)
  18 {
  19   eligible_ = 0;
  20   while (!eligible_)
  21     this->dt_cond_.wait ();
  22 }
  23
  24 void
  25 DT::resume (void)
  26 {
  27   eligible_ = 1;
  28   this->dt_cond_.signal ();
  29 }
  30
  31 CORBA::Short
  32 Segment_Sched_Param_Policy::importance (void)
  33 {
  34   return this->importance_;
  35 }
  36
  37 void
  38 Segment_Sched_Param_Policy::importance (CORBA::Short importance)
  39 {
  40   this->importance_ = importance;
  41 }
  42
  43 CORBA::Policy_ptr
  44 Segment_Sched_Param_Policy::copy (void)
  45 {
  46   Segment_Sched_Param_Policy *copy = 0;
  47   ACE_NEW_THROW_EX (copy,
  48                     Segment_Sched_Param_Policy,
  49                     CORBA::NO_MEMORY ());
  50
  51   copy->importance (this->importance_);
  52
  53   return copy;
  54 }
  55
  56 CORBA::PolicyType
  57 Segment_Sched_Param_Policy::policy_type (void)
  58 {
  59   return 0;
  60 }
  61
  62 void
  63 Segment_Sched_Param_Policy::destroy (void)
  64 {
  65 }
  66
  67 MIF_Scheduler::MIF_Scheduler (CORBA::ORB_ptr orb)
  68   : wait_cond_ (lock_),
  69     wait_ (0)
  70 {
  71   try
  72     {
  73       CORBA::Object_var object =
  74         orb->resolve_initial_references ("RTScheduler_Current");
  75
  76       this->current_ =
  77         RTScheduling::Current::_narrow (object.in ());
  78
  79       object =
  80         orb->resolve_initial_references ("PriorityMappingManager");
  81
  82       this->mapping_manager_ =
  83         RTCORBA::PriorityMappingManager::_narrow (object.in ());
  84     }
  85   catch (const CORBA::Exception& ex)
  86     {
  87       ex._tao_print_exception ("Caught exception:");
  88     }
  89 }
  90
  91 MIF_Scheduler::~MIF_Scheduler (void)
  92 {
  93   while (free_que_.message_count () > 0)
  94     {
  95       DT *dt = 0;
  96       ACE_Message_Block *msg = 0;
  97       free_que_.dequeue_head (msg);
  98       dt = dynamic_cast<DT*> (msg);
  99       delete dt;
 100     }
 101 }
 102
 103 void
 104 MIF_Scheduler::incr_thr_count (void)
 105 {
 106   lock_.acquire ();
 107   wait_++;
 108   lock_.release ();
 109 }
 110
 111 void
 112 MIF_Scheduler::wait (void)
 113 {
 114   lock_.acquire ();
 115   while (wait_ > 0)
 116     wait_cond_.wait ();
 117
 118   ACE_DEBUG ((LM_DEBUG,
 119               "After Wait %d\n",
 120               wait_));
 121
 122   lock_.release ();
 123 }
 124
 125 void
 126 MIF_Scheduler::resume_main (void)
 127 {
 128   wait_--;
 129   wait_cond_.signal ();
 130 }
 131
 132 MIF_Scheduling::SegmentSchedulingParameterPolicy_ptr
 133 MIF_Scheduler::create_segment_scheduling_parameter (CORBA::Short importance)
 134 {
 135   MIF_Scheduling::SegmentSchedulingParameterPolicy_ptr segment_policy;
 136   ACE_NEW_THROW_EX (segment_policy,
 137                     Segment_Sched_Param_Policy,
 138                     CORBA::NO_MEMORY (
 139                                       CORBA::SystemException::_tao_minor_code (
 140                TAO::VMCID,
 141                                        ENOMEM),
 142                                       CORBA::COMPLETED_NO));
 143
 144   segment_policy->importance (importance);
 145
 146   return segment_policy;
 147 }
 148
 149
 150 void
 151 MIF_Scheduler::begin_new_scheduling_segment (const RTScheduling::Current::IdType &/*guid*/,
 152                                              const char *,
 153                                              CORBA::Policy_ptr sched_policy,
 154                                              CORBA::Policy_ptr)
 155 {
 156   size_t count = 0;
 157   RTScheduling::Current::IdType_var guid = this->current_->id ();
 158
 159   ACE_OS::memcpy (&count,
 160                   guid->get_buffer (),
 161                   guid->length ());
 162
 163
 164   MIF_Scheduling::SegmentSchedulingParameterPolicy_var sched_param =
 165     MIF_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_policy);
 166
 167   CORBA::Short desired_priority = sched_param->importance ();
 168
 169   if (TAO_debug_level > 0)
 170     ACE_DEBUG ((LM_DEBUG,
 171                 "%t MIF_Scheduler::begin_scheduling_segment - Importance %d\n",
 172                 desired_priority));
 173
 174
 175   if (desired_priority != 100)
 176     {
 177       //NOT Main Thread
 178       DT* new_dt = 0;
 179       ACE_NEW (new_dt,
 180                DT (this->lock_,
 181                    count));
 182
 183       new_dt->msg_priority (desired_priority);
 184       lock_.acquire ();
 185       ready_que_.enqueue_prio (new_dt);
 186       resume_main ();
 187       new_dt->suspend ();
 188       lock_.release ();
 189     }
 190 }
 191
 192 void
 193 MIF_Scheduler::begin_nested_scheduling_segment (const RTScheduling::Current::IdType &guid,
 194                                                            const char *name,
 195                                                            CORBA::Policy_ptr sched_param,
 196                                                            CORBA::Policy_ptr implicit_sched_param)
 197 {
 198   this->begin_new_scheduling_segment (guid,
 199                                       name,
 200                                       sched_param,
 201                                       implicit_sched_param);
 202 }
 203
 204 void
 205 MIF_Scheduler::update_scheduling_segment (const RTScheduling::Current::IdType &/*guid*/,
 206                                           const char* /*name*/,
 207                                           CORBA::Policy_ptr sched_policy,
 208                                           CORBA::Policy_ptr /*implicit_sched_param*/)
 209 {
 210   size_t count = 0;
 211   RTScheduling::Current::IdType_var guid = this->current_->id ();
 212
 213   ACE_OS::memcpy (&count,
 214                   guid->get_buffer (),
 215                   guid->length ());
 216
 217   MIF_Scheduling::SegmentSchedulingParameterPolicy_var sched_param =
 218     MIF_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_policy);
 219
 220   CORBA::Short desired_priority = sched_param->importance ();
 221
 222   if (TAO_debug_level > 0)
 223     ACE_DEBUG ((LM_DEBUG,
 224                 "%t MIF_Scheduler::update_scheduling_segment - Importance %d\n",
 225                 desired_priority));
 226
 227   DT* new_dt = 0;
 228   ACE_NEW (new_dt,
 229            DT (this->lock_,
 230                count));
 231
 232   new_dt->msg_priority (desired_priority);
 233
 234   if (ready_que_.message_count () > 0)
 235     {
 236       DT* run_dt;
 237       ACE_Message_Block* msg = 0;
 238       ready_que_.dequeue_head (msg);
 239       run_dt = dynamic_cast<DT*> (msg);
 240       if ((desired_priority == 100) || run_dt->msg_priority () >= (unsigned int)desired_priority)
 241         {
 242           ready_que_.enqueue_prio (new_dt);
 243           lock_.acquire ();
 244           run_dt->resume ();
 245           new_dt->suspend ();
 246           lock_.release ();
 247           free_que_.enqueue_prio (run_dt);
 248         }
 249       else
 250         {
 251           ready_que_.enqueue_prio (run_dt);
 252           delete new_dt;
 253         }
 254     }
 255   else delete new_dt;
 256 }
 257
 258 void
 259 MIF_Scheduler::end_scheduling_segment (const RTScheduling::Current::IdType &guid,
 260                                        const char *)
 261 {
 262   size_t count = 0;
 263   ACE_OS::memcpy (&count,
 264                   guid.get_buffer (),
 265                   guid.length ());
 266
 267   ACE_DEBUG ((LM_DEBUG,
 268               "MIF_Scheduler::end_scheduling_segment %d\n",
 269                   count));
 270
 271   if (ready_que_.message_count () > 0)
 272     {
 273       DT* run_dt = 0;
 274       ACE_Message_Block* msg = 0;
 275       ready_que_.dequeue_head (msg);
 276       run_dt = dynamic_cast<DT*> (msg);
 277       lock_.acquire ();
 278       run_dt->resume ();
 279       lock_.release ();
 280       free_que_.enqueue_prio (run_dt);
 281     }
 282 }
 283
 284 void
 285 MIF_Scheduler::end_nested_scheduling_segment (const RTScheduling::Current::IdType &,
 286                                                          const char *,
 287                                                          CORBA::Policy_ptr)
 288 {
 289 }
 290
 291 void
 292 MIF_Scheduler::send_request (PortableInterceptor::ClientRequestInfo_ptr request_info)
 293 {
 294   CORBA::Policy_var sched_param = current_->scheduling_parameter ();
 295
 296   MIF_Scheduling::SegmentSchedulingParameterPolicy_var sched_param_var =
 297     MIF_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_param.in ());
 298
 299   IOP::ServiceContext srv_con;
 300   srv_con.context_id = Client_Interceptor::SchedulingInfo;
 301
 302   RTScheduling::Current::IdType_var guid = current_->id ();
 303
 304   int guid_length = guid->length ();
 305
 306   CORBA::OctetSeq seq_buf (guid_length);
 307   seq_buf.length (seq_buf.maximum ());
 308   ACE_OS::memcpy (seq_buf.get_buffer (),
 309                   guid->get_buffer (),
 310                   guid_length);
 311
 312   int cxt_data_length = sizeof (int) + guid_length;
 313   srv_con.context_data.length (cxt_data_length);
 314
 315   int i = 0;
 316   for (;i < guid_length;i++)
 317     {
 318       srv_con.context_data [i] = seq_buf [i];
 319     }
 320
 321   int importance = sched_param_var->importance ();
 322   CORBA::OctetSeq int_buf (sizeof (importance));
 323   int_buf.length (int_buf.maximum ());
 324   ACE_OS::memcpy (int_buf.get_buffer (),
 325                   &importance,
 326                   sizeof (importance));
 327
 328   int j = 0;
 329   for (;i < cxt_data_length;i++)
 330     {
 331       srv_con.context_data [i] = int_buf [j++];
 332     }
 333
 334   request_info->add_request_service_context (srv_con,
 335                                              0);
 336
 337   lock_.acquire ();
 338   if (ready_que_.message_count () > 0)
 339     {
 340       int priority;
 341       ACE_hthread_t current;
 342       ACE_Thread::self (current);
 343       if (ACE_Thread::getprio (current, priority) == -1)
 344         return;
 345
 346       ACE_DEBUG ((LM_DEBUG,
 347                   "Initial thread priority is %d %d\n",
 348                   priority,
 349                   ACE_DEFAULT_THREAD_PRIORITY));
 350
 351       RTCORBA::Priority rtpriority;
 352       RTCORBA::PriorityMapping* pm = this->mapping_manager_->mapping ();
 353       if (pm->to_CORBA(priority + 1, rtpriority))
 354         {
 355           current_->the_priority (rtpriority);
 356
 357           ACE_Thread::self (current);
 358           if (ACE_Thread::getprio (current, priority) == -1)
 359             return;
 360
 361           ACE_DEBUG ((LM_DEBUG,
 362                       "Bumped thread priority is %d\n",
 363                       priority));
 364         }
 365
 366       DT* run_dt = 0;
 367       ACE_Message_Block* msg = 0;
 368       ready_que_.dequeue_head (msg);
 369       run_dt = dynamic_cast<DT*> (msg);
 370       run_dt->resume ();
 371       free_que_.enqueue_prio (run_dt);
 372     }
 373   lock_.release ();
 374
 375 }
 376
 377 void
 378 MIF_Scheduler::receive_request (PortableInterceptor::ServerRequestInfo_ptr request_info,
 379                                 RTScheduling::Current::IdType_out guid_out,
 380                                 CORBA::String_out,
 381                                 CORBA::Policy_out sched_param_out,
 382                                 CORBA::Policy_out /*implicit_sched_param*/)
 383 {
 384
 385   if (TAO_debug_level > 0)
 386     ACE_DEBUG ((LM_DEBUG,
 387                 "MIF_Scheduler::receive_request\n"));
 388
 389   IOP::ServiceContext_var serv_cxt =
 390     request_info->get_request_service_context (Server_Interceptor::SchedulingInfo);
 391
 392   if (serv_cxt != 0)
 393     {
 394       ACE_DEBUG ((LM_DEBUG,
 395                   "Got scheduling info\n"));
 396
 397       RTScheduling::Current::IdType* guid;
 398       ACE_NEW (guid,
 399                RTScheduling::Current::IdType);
 400
 401       guid->length (sizeof(size_t));
 402       ACE_OS::memcpy (guid->get_buffer (),
 403                       serv_cxt->context_data.get_buffer (),
 404                       sizeof (size_t));
 405
 406       size_t gu_id;
 407       ACE_OS::memcpy (&gu_id,
 408                       guid->get_buffer (),
 409                       guid->length ());
 410
 411       ACE_DEBUG ((LM_DEBUG,
 412                   "MIF_Scheduler::receive_request %d\n",
 413                   gu_id));
 414
 415
 416       CORBA::OctetSeq int_buf (sizeof (long));
 417       int_buf.length (int_buf.maximum ());
 418       int i = sizeof (long);
 419       for (unsigned int j = 0;j < sizeof (int);j++)
 420         {
 421           int_buf [j] = serv_cxt->context_data [i++];
 422         }
 423
 424       int importance = 0;
 425       ACE_OS::memcpy (&importance,
 426                       int_buf.get_buffer (),
 427                       sizeof (importance));
 428
 429       guid_out = guid;
 430       sched_param_out = DT_TEST::instance ()->scheduler ()->create_segment_scheduling_parameter (importance);
 431
 432       if (TAO_debug_level > 0)
 433         ACE_DEBUG ((LM_DEBUG,
 434                     "%t The Guid is %d Importance is %d\n",
 435                     gu_id,
 436                     importance));
 437
 438       DT* new_dt;
 439       ACE_NEW (new_dt,
 440                DT (this->lock_,
 441                    gu_id));
 442
 443       new_dt->msg_priority (importance);
 444       lock_.acquire ();
 445       ready_que_.enqueue_prio (new_dt);
 446       new_dt->suspend ();
 447       lock_.release ();
 448     }
 449 }
 450
 451 void
 452 MIF_Scheduler::send_reply (PortableInterceptor::ServerRequestInfo_ptr)
 453 {
 454
 455   RTScheduling::Current::IdType_var guid = current_->id ();
 456
 457   size_t count;
 458   ACE_OS::memcpy (&count,
 459                   guid->get_buffer (),
 460                   guid->length ());
 461
 462   ACE_DEBUG ((LM_DEBUG,
 463               "MIF_Scheduler::send_reply %d\n",
 464               count));
 465
 466   if (ready_que_.message_count () > 0)
 467     {
 468       DT* run_dt = 0;
 469       ACE_Message_Block* msg = 0;
 470       ready_que_.dequeue_head (msg);
 471       run_dt = dynamic_cast<DT*> (msg);
 472       lock_.acquire ();
 473       run_dt->resume ();
 474       lock_.release ();
 475       free_que_.enqueue_prio (run_dt);
 476     }
 477 }
 478
 479 void
 480 MIF_Scheduler::send_exception (PortableInterceptor::ServerRequestInfo_ptr)
 481 {
 482   if (ready_que_.message_count () > 0)
 483     {
 484       DT* run_dt = 0;
 485       ACE_Message_Block* msg = 0;
 486       ready_que_.dequeue_head (msg);
 487       run_dt = dynamic_cast<DT*> (msg);
 488       lock_.acquire ();
 489       run_dt->resume ();
 490       lock_.release ();
 491       free_que_.enqueue_prio (run_dt);
 492     }
 493 }
 494
 495 void
 496 MIF_Scheduler::send_other (PortableInterceptor::ServerRequestInfo_ptr)
 497 {
 498   if (TAO_debug_level > 0)
 499     {
 500       RTScheduling::Current::IdType_var guid = current_->id ();
 501
 502       size_t count;
 503       ACE_OS::memcpy (&count,
 504                       guid->get_buffer (),
 505                       guid->length ());
 506
 507
 508       ACE_DEBUG ((LM_DEBUG,
 509                   "MIF_Scheduler::send_other %d\n",
 510                   count));
 511     }
 512
 513   if (ready_que_.message_count () > 0)
 514     {
 515       DT* run_dt;
 516       ACE_Message_Block* msg;
 517       ready_que_.dequeue_head (msg);
 518       run_dt = dynamic_cast<DT*> (msg);
 519       lock_.acquire ();
 520       run_dt->resume ();
 521       lock_.release ();
 522       free_que_.enqueue_prio (run_dt);
 523     }
 524 }
 525
 526 void
 527 MIF_Scheduler::send_poll (PortableInterceptor::ClientRequestInfo_ptr)
 528 {
 529 }
 530
 531 void
 532 MIF_Scheduler::receive_reply (PortableInterceptor::ClientRequestInfo_ptr)
 533 {
 534   CORBA::Policy_var sched_param = current_->scheduling_parameter ();
 535
 536   MIF_Scheduling::SegmentSchedulingParameterPolicy_var sched_param_var =
 537     MIF_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_param.in ());
 538
 539   int importance = sched_param_var->importance ();
 540
 541   RTScheduling::Current::IdType_var guid = current_->id ();
 542
 543   size_t gu_id;
 544   ACE_OS::memcpy (&gu_id,
 545                   guid->get_buffer (),
 546                   guid->length ());
 547
 548   if (TAO_debug_level > 0)
 549     ACE_DEBUG ((LM_DEBUG,
 550                 "MIF_Scheduler::receive_reply Guid = %d Imp = %d\n",
 551                 gu_id,
 552                 importance));
 553
 554
 555   DT* new_dt;
 556   ACE_NEW (new_dt,
 557            DT (this->lock_,
 558                gu_id));
 559
 560   new_dt->msg_priority (importance);
 561
 562   lock_.acquire ();
 563   ready_que_.enqueue_prio (new_dt);
 564   int priority;
 565   ACE_hthread_t current;
 566   ACE_Thread::self (current);
 567   if (ACE_Thread::getprio (current, priority) == -1)
 568     return;
 569
 570   RTCORBA::Priority rtpriority;
 571   RTCORBA::PriorityMapping* pm = this->mapping_manager_->mapping ();
 572   if (pm->to_CORBA(priority - 1, rtpriority))
 573     {
 574       current_->the_priority (rtpriority);
 575     }
 576
 577   new_dt->suspend ();
 578   lock_.release ();
 579 }
 580
 581 void
 582 MIF_Scheduler::receive_exception (PortableInterceptor::ClientRequestInfo_ptr)
 583 {
 584   CORBA::Policy_var sched_param = current_->scheduling_parameter ();
 585
 586   MIF_Scheduling::SegmentSchedulingParameterPolicy_var sched_param_var =
 587     MIF_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_param.in ());
 588
 589   int importance = sched_param_var->importance ();
 590
 591   RTScheduling::Current::IdType_var guid = current_->id ();
 592
 593   size_t gu_id;
 594   ACE_OS::memcpy (&gu_id,
 595                   guid->get_buffer (),
 596                   guid->length ());
 597
 598   DT* new_dt;
 599   ACE_NEW (new_dt,
 600            DT (this->lock_,
 601                gu_id));
 602
 603   new_dt->msg_priority (importance);
 604
 605   lock_.acquire ();
 606   ready_que_.enqueue_prio (new_dt);
 607
 608   int priority;
 609   ACE_hthread_t current;
 610   ACE_Thread::self (current);
 611   if (ACE_Thread::getprio (current, priority) == -1)
 612     return;
 613
 614   RTCORBA::Priority rtpriority;
 615   RTCORBA::PriorityMapping* pm = this->mapping_manager_->mapping ();
 616   if (pm->to_CORBA(priority - 1, rtpriority))
 617     {
 618       current_->the_priority (rtpriority);
 619     }
 620
 621   new_dt->suspend ();
 622   lock_.release ();
 623 }
 624
 625 void
 626 MIF_Scheduler::receive_other (PortableInterceptor::ClientRequestInfo_ptr)
 627 {
 628   CORBA::Policy_var sched_param = current_->scheduling_parameter ();
 629
 630   MIF_Scheduling::SegmentSchedulingParameterPolicy_var sched_param_var =
 631     MIF_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_param.in ());
 632
 633   int importance = sched_param_var->importance ();
 634
 635   RTScheduling::Current::IdType_var guid = current_->id ();
 636
 637   size_t gu_id;
 638   ACE_OS::memcpy (&gu_id,
 639                   guid->get_buffer (),
 640                   guid->length ());
 641
 642   DT* new_dt = 0;
 643   ACE_NEW (new_dt,
 644            DT (this->lock_,
 645                gu_id));
 646
 647   new_dt->msg_priority (importance);
 648
 649   lock_.acquire ();
 650   ready_que_.enqueue_prio (new_dt);
 651
 652   int priority;
 653   ACE_hthread_t current;
 654   ACE_Thread::self (current);
 655   if (ACE_Thread::getprio (current, priority) == -1)
 656     return;
 657
 658   RTCORBA::Priority rtpriority;
 659   RTCORBA::PriorityMapping* pm = this->mapping_manager_->mapping ();
 660   if (pm->to_CORBA(priority - 1, rtpriority))
 661     {
 662       current_->the_priority (rtpriority);
 663     }
 664
 665   new_dt->suspend ();
 666   lock_.release ();
 667 }
 668
 669 void
 670 MIF_Scheduler::cancel (const RTScheduling::Current::IdType &)
 671 {
 672 }
 673
 674 CORBA::PolicyList*
 675 MIF_Scheduler::scheduling_policies (void)
 676 {
 677   return 0;
 678 }
 679
 680 void
 681 MIF_Scheduler::scheduling_policies (const CORBA::PolicyList &)
 682 {
 683 }
 684
 685 CORBA::PolicyList*
 686 MIF_Scheduler::poa_policies (void)
 687 {
 688         return 0;
 689 }
 690
 691 char *
 692 MIF_Scheduler::scheduling_discipline_name (void)
 693 {
 694         return 0;
 695 }
 696
 697 RTScheduling::ResourceManager_ptr
 698 MIF_Scheduler::create_resource_manager (const char *,
 699                                         CORBA::Policy_ptr)
 700 {
 701         return 0;
 702 }
 703
 704 void
 705 MIF_Scheduler::set_scheduling_parameter (PortableServer::Servant &,
 706                                          const char *,
 707                                          CORBA::Policy_ptr)
 708 {
 709 }