TAO/examples/Kokyu_dsrt_schedulers/FP_Scheduler.cpp

   1 #include "FP_Scheduler.h"
   2 #include "Kokyu_qosC.h"
   3 #include "utils.h"
   4 #include "tao/ORB_Constants.h"
   5 #include "tao/CodecFactory/CodecFactory.h"
   6 #include "tao/RTScheduling/Request_Interceptor.h"
   7
   8 FP_Segment_Sched_Param_Policy::FP_Segment_Sched_Param_Policy ()
   9 {
  10 }
  11
  12 FP_Segment_Sched_Param_Policy::FP_Segment_Sched_Param_Policy (
  13     const FP_Segment_Sched_Param_Policy &rhs)
  14   : CORBA::Object (),
  15   CORBA::Policy (),
  16   FP_Scheduling::SegmentSchedulingParameterPolicy (),
  17   CORBA::LocalObject (),
  18   value_ (rhs.value_)
  19 {
  20 }
  21
  22 FP_Scheduling::SegmentSchedulingParameter
  23 FP_Segment_Sched_Param_Policy::value (void)
  24 {
  25   return value_;
  26 }
  27
  28 void
  29 FP_Segment_Sched_Param_Policy::value (
  30         const FP_Scheduling::SegmentSchedulingParameter & value)
  31 {
  32   this->value_ = value;
  33 }
  34
  35 CORBA::Policy_ptr
  36 FP_Segment_Sched_Param_Policy::copy (void)
  37 {
  38   FP_Segment_Sched_Param_Policy* tmp = 0;
  39   ACE_NEW_THROW_EX (tmp, FP_Segment_Sched_Param_Policy (*this),
  40                     CORBA::NO_MEMORY (TAO::VMCID,
  41                                       CORBA::COMPLETED_NO));
  42
  43   return tmp;
  44 }
  45
  46 CORBA::PolicyType
  47 FP_Segment_Sched_Param_Policy::policy_type (void)
  48 {
  49   return 0;
  50 }
  51
  52 void
  53 FP_Segment_Sched_Param_Policy::destroy (void)
  54 {
  55 }
  56
  57 Fixed_Priority_Scheduler::Fixed_Priority_Scheduler (
  58   CORBA::ORB_ptr orb,
  59   Kokyu::DSRT_Dispatcher_Impl_t disp_impl_type,
  60   int ace_sched_policy,
  61   int ace_sched_scope)
  62   : orb_ (CORBA::ORB::_duplicate (orb)),
  63     disp_impl_type_ (disp_impl_type),
  64     ace_sched_policy_ (ace_sched_policy),
  65     ace_sched_scope_ (ace_sched_scope)
  66 {
  67
  68   Kokyu::DSRT_ConfigInfo config;
  69
  70   config.impl_type_ = this->disp_impl_type_;
  71   config.sched_policy_ = ace_sched_policy_;
  72   config.sched_scope_ = ace_sched_scope_;
  73
  74   Kokyu::DSRT_Dispatcher_Factory<FP_Scheduler_Traits>::DSRT_Dispatcher_Auto_Ptr
  75     tmp( Kokyu::DSRT_Dispatcher_Factory<FP_Scheduler_Traits>::
  76          create_DSRT_dispatcher (config) );
  77   kokyu_dispatcher_ = tmp;
  78
  79   CORBA::Object_var object =
  80     orb->resolve_initial_references ("RTScheduler_Current");
  81
  82   this->current_ =
  83     RTScheduling::Current::_narrow (object.in ());
  84
  85   IOP::CodecFactory_var codec_factory;
  86   CORBA::Object_var obj =
  87     orb->resolve_initial_references ("CodecFactory");
  88
  89   if (CORBA::is_nil(obj.in ()))
  90     {
  91       ACE_ERROR ((LM_ERROR, "Nil Codec factory\n"));
  92     }
  93   else
  94     {
  95       codec_factory = IOP::CodecFactory::_narrow (obj.in ());
  96     }
  97
  98   IOP::Encoding encoding;
  99   encoding.format = IOP::ENCODING_CDR_ENCAPS;
 100   encoding.major_version = 1;
 101   encoding.minor_version = 2;
 102
 103   codec_ = codec_factory->create_codec (encoding);
 104 }
 105
 106 Fixed_Priority_Scheduler::~Fixed_Priority_Scheduler (void)
 107 {
 108   //  delete kokyu_dispatcher_;
 109 }
 110
 111 void
 112 Fixed_Priority_Scheduler::shutdown (void)
 113 {
 114   kokyu_dispatcher_->shutdown ();
 115   ACE_DEBUG ((LM_DEBUG, "kokyu DSRT dispatcher shutdown\n"));
 116 }
 117
 118 FP_Scheduling::SegmentSchedulingParameterPolicy_ptr
 119 Fixed_Priority_Scheduler::create_segment_scheduling_parameter (
 120         const FP_Scheduling::SegmentSchedulingParameter & value
 121       )
 122 {
 123   FP_Scheduling::SegmentSchedulingParameterPolicy_ptr
 124     segment_sched_param_policy;
 125   ACE_NEW_THROW_EX (segment_sched_param_policy,
 126                     FP_Segment_Sched_Param_Policy,
 127                     CORBA::NO_MEMORY (
 128                                       CORBA::SystemException::_tao_minor_code (
 129                                        TAO::VMCID,
 130                                        ENOMEM),
 131                                       CORBA::COMPLETED_NO));
 132
 133   segment_sched_param_policy->value (value);
 134
 135   return segment_sched_param_policy;
 136 }
 137
 138
 139 void
 140 Fixed_Priority_Scheduler::begin_new_scheduling_segment (const RTScheduling::Current::IdType& guid,
 141                                              const char *,
 142                                              CORBA::Policy_ptr sched_policy,
 143                                              CORBA::Policy_ptr)
 144 {
 145 #ifdef KOKYU_DSRT_LOGGING
 146   ACE_DEBUG ((LM_DEBUG,
 147               "(%t|%T):FP_Scheduler::begin_new_scheduling_segment enter\n"));
 148 #endif
 149
 150 #ifdef KOKYU_DSRT_LOGGING
 151   int int_guid;
 152   ACE_OS::memcpy (&int_guid,
 153                   guid.get_buffer (),
 154                   guid.length ());
 155   ACE_DEBUG ((LM_DEBUG, "(%t|%T): guid is %d\n", int_guid));
 156 #endif
 157
 158   FP_Scheduler_Traits::QoSDescriptor_t qos;
 159   FP_Scheduling::SegmentSchedulingParameterPolicy_var sched_param_policy =
 160     FP_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_policy);
 161
 162   FP_Scheduling::SegmentSchedulingParameter sched_param = sched_param_policy->value ();
 163   RTCORBA::Priority desired_priority = sched_param.base_priority;
 164   qos.priority_ = desired_priority;
 165   kokyu_dispatcher_->schedule (guid, qos);
 166
 167 #ifdef KOKYU_DSRT_LOGGING
 168   ACE_DEBUG ((LM_DEBUG,
 169               "(%t|%T):FP_Scheduler::begin_new_scheduling_segment exit\n"));
 170 #endif
 171 }
 172
 173 void
 174 Fixed_Priority_Scheduler::begin_nested_scheduling_segment (const RTScheduling::Current::IdType &guid,
 175                                                            const char *name,
 176                                                            CORBA::Policy_ptr sched_param,
 177                                                            CORBA::Policy_ptr implicit_sched_param)
 178 {
 179   this->begin_new_scheduling_segment (guid,
 180                                       name,
 181                                       sched_param,
 182                                       implicit_sched_param);
 183 }
 184
 185 void
 186 Fixed_Priority_Scheduler::update_scheduling_segment (const RTScheduling::Current::IdType &guid,
 187                                           const char* name,
 188                                           CORBA::Policy_ptr sched_policy,
 189                                           CORBA::Policy_ptr implicit_sched_param)
 190 {
 191   ACE_UNUSED_ARG ((name));
 192   ACE_UNUSED_ARG ((implicit_sched_param));
 193
 194 #ifdef KOKYU_DSRT_LOGGING
 195   int int_guid ;
 196   ACE_OS::memcpy (&int_guid,
 197                   guid.get_buffer (),
 198                   guid.length ());
 199   ACE_DEBUG ((LM_DEBUG, "(%t|%T): update_sched_seg::guid is %d\n", int_guid));
 200 #endif
 201
 202   FP_Scheduling::SegmentSchedulingParameterPolicy_var sched_param_policy =
 203     FP_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_policy);
 204
 205   FP_Scheduling::SegmentSchedulingParameter sched_param =
 206     sched_param_policy->value ();
 207
 208   RTCORBA::Priority desired_priority = sched_param.base_priority;
 209
 210   FP_Scheduler_Traits::QoSDescriptor_t qos;
 211   qos.priority_ = desired_priority;
 212
 213   kokyu_dispatcher_->update_schedule (guid, qos);
 214 }
 215
 216 void
 217 Fixed_Priority_Scheduler::end_scheduling_segment (const RTScheduling::Current::IdType &guid,
 218                                        const char *)
 219 {
 220 #ifdef KOKYU_DSRT_LOGGING
 221   int int_guid;
 222   ACE_OS::memcpy (&int_guid,
 223                   guid.get_buffer (),
 224                   guid.length ());
 225   ACE_DEBUG ((LM_DEBUG, "(%t|%T) call to end_sched_segment for guid %d\n", int_guid));
 226 #endif
 227
 228   kokyu_dispatcher_->cancel_schedule (guid);
 229 }
 230
 231 void
 232 Fixed_Priority_Scheduler::end_nested_scheduling_segment (const RTScheduling::Current::IdType &,
 233                                                          const char *,
 234                                                          CORBA::Policy_ptr)
 235 {
 236
 237 }
 238
 239
 240 void
 241 Fixed_Priority_Scheduler::send_request (PortableInterceptor::ClientRequestInfo_ptr ri)
 242 {
 243   Kokyu::Svc_Ctxt_DSRT_QoS sc_qos;
 244
 245   CORBA::String_var operation = ri->operation ();
 246
 247 #ifdef KOKYU_DSRT_LOGGING
 248   ACE_DEBUG ((LM_DEBUG,
 249               "(%t|%T): send_request "
 250               "from \"%s\"\n",
 251               operation.in ()));
 252 #endif
 253   // Make the context to send the context to the target
 254   IOP::ServiceContext sc;
 255   sc.context_id = Client_Interceptor::SchedulingInfo;
 256
 257   CORBA::Policy_var sched_policy =
 258     this->current_->scheduling_parameter();
 259
 260   RTScheduling::Current::IdType_var guid = this->current_->id ();
 261   /*
 262   ACE_OS::memcpy (&guid,
 263                   guid->get_buffer (),
 264                   guid->length ());
 265   */
 266   RTCORBA::Priority desired_priority;
 267   if (CORBA::is_nil (sched_policy.in ()))
 268     {
 269       desired_priority = 0;
 270     }
 271   else
 272     {
 273       FP_Scheduling::SegmentSchedulingParameterPolicy_var sched_param_policy =
 274         FP_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_policy.in ());
 275
 276       FP_Scheduling::SegmentSchedulingParameter sched_param =
 277         sched_param_policy->value ();
 278
 279       desired_priority = sched_param.base_priority;
 280
 281 #ifdef KOKYU_DSRT_LOGGING
 282       int int_guid;
 283       ACE_OS::memcpy (&int_guid,
 284                       guid->get_buffer (),
 285                       guid->length ());
 286       ACE_DEBUG ((LM_DEBUG,
 287                   ACE_TEXT ("(%t): send_request desired_priority from current = %d, guid = %d\n"),
 288                   desired_priority, int_guid));
 289 #endif
 290
 291       //Fill the guid in the SC Qos struct
 292       sc_qos.guid.length (guid->length ());
 293       guid_copy (sc_qos.guid, guid.in ());
 294       sc_qos.desired_priority = desired_priority;
 295       CORBA::Any sc_qos_as_any;
 296       sc_qos_as_any <<= sc_qos;
 297
 298       CORBA::OctetSeq_var cdtmp = codec_->encode (sc_qos_as_any);
 299       sc.context_data = cdtmp.in ();
 300
 301 #ifdef KOKYU_DSRT_LOGGING
 302       ACE_DEBUG ((LM_DEBUG,
 303                   ACE_TEXT ("(%t|%T): send_request : about to add sched SC\n")));
 304 #endif
 305
 306       // Add this context to the service context list.
 307       ri->add_request_service_context (sc, 0);
 308     }
 309
 310
 311 #ifdef KOKYU_DSRT_LOGGING
 312   ACE_DEBUG ((LM_DEBUG,
 313               ACE_TEXT ("(%t|%T): send_request : ")
 314               ACE_TEXT ("about to call scheduler to inform block\n")));
 315 #endif
 316
 317   kokyu_dispatcher_->update_schedule (guid.in (),
 318                                       Kokyu::BLOCK);
 319
 320 #ifdef KOKYU_DSRT_LOGGING
 321   ACE_DEBUG ((LM_DEBUG,
 322               ACE_TEXT ("(%t|%T): send_request interceptor done\n")));
 323 #endif
 324 }
 325
 326 void
 327 Fixed_Priority_Scheduler::receive_request (PortableInterceptor::ServerRequestInfo_ptr ri,
 328                                 RTScheduling::Current::IdType_out guid_out,
 329                                 CORBA::String_out /*name*/,
 330                                 CORBA::Policy_out sched_param_out,
 331                                 CORBA::Policy_out /*implicit_sched_param_out*/)
 332 {
 333   const Kokyu::Svc_Ctxt_DSRT_QoS* sc_qos_ptr = 0;
 334
 335 #ifdef KOKYU_DSRT_LOGGING
 336   ACE_DEBUG ((LM_DEBUG, "(%t|%T):entered FP_Scheduler::receive_request\n"));
 337 #endif
 338
 339   RTScheduling::Current::IdType guid;
 340
 341   CORBA::String_var operation = ri->operation ();
 342
 343 #ifdef KOKYU_DSRT_LOGGING
 344   ACE_DEBUG ((LM_DEBUG,
 345               "(%t|%T): receive_request from "
 346               "\"%s\"\n",
 347               operation.in ()));
 348 #endif
 349
 350   // Ignore the "_is_a" operation since it may have been invoked
 351   // locally on the server side as a side effect of another call,
 352   // meaning that the client hasn't added the service context yet.
 353   if (ACE_OS::strcmp ("_is_a", operation.in ()) == 0)
 354     return;
 355
 356   IOP::ServiceContext_var sc =
 357     ri->get_request_service_context (Server_Interceptor::SchedulingInfo);
 358
 359   RTCORBA::Priority desired_priority;
 360
 361   if (sc.ptr () == 0)
 362     {
 363       desired_priority = 0;
 364     }
 365   else
 366     {
 367       CORBA::OctetSeq oc_seq = CORBA::OctetSeq (sc->context_data.length (),
 368                                                  sc->context_data.length (),
 369                                                  sc->context_data.get_buffer (),
 370                                                  0);
 371       CORBA::Any sc_qos_as_any;
 372       CORBA::Any_var scqostmp = codec_->decode (oc_seq);
 373       sc_qos_as_any = scqostmp.in ();
 374       //Don't store in a _var, since >>= returns a pointer to an
 375       //internal buffer and we are not supposed to free it.
 376       sc_qos_as_any >>= sc_qos_ptr;
 377
 378       desired_priority = sc_qos_ptr->desired_priority;
 379       guid.length (sc_qos_ptr->guid.length ());
 380       guid_copy (guid, sc_qos_ptr->guid);
 381
 382       ACE_NEW (guid_out.ptr (),
 383                RTScheduling::Current::IdType);
 384       guid_out.ptr ()->length (guid.length ());
 385       *(guid_out.ptr ()) = guid;
 386
 387 #ifdef KOKYU_DSRT_LOGGING
 388       int int_guid;
 389       ACE_OS::memcpy (&int_guid,
 390                       guid.get_buffer (),
 391                       guid.length ());
 392       ACE_DEBUG ((LM_DEBUG,
 393                   "(%t|%T): Desired_Priority = %d, guid = %d in recvd service context\n",
 394                   desired_priority,
 395                   int_guid));
 396 #endif
 397
 398       FP_Scheduling::SegmentSchedulingParameter sched_param;
 399       sched_param.base_priority = desired_priority;
 400       sched_param_out = this->create_segment_scheduling_parameter (sched_param);
 401     }
 402
 403   FP_Scheduler_Traits::QoSDescriptor_t qos;
 404   qos.priority_ = desired_priority;
 405   this->kokyu_dispatcher_->schedule (guid, qos);
 406
 407 #ifdef KOKYU_DSRT_LOGGING
 408   ACE_DEBUG ((LM_DEBUG, "(%t|%T): receive_request interceptor done\n"));
 409 #endif
 410
 411 }
 412
 413 void
 414 Fixed_Priority_Scheduler::send_poll (PortableInterceptor::ClientRequestInfo_ptr)
 415 {
 416 }
 417
 418 void
 419 Fixed_Priority_Scheduler::send_reply (PortableInterceptor::ServerRequestInfo_ptr ri)
 420 {
 421   RTCORBA::Priority desired_priority = 0;
 422   Kokyu::Svc_Ctxt_DSRT_QoS sc_qos;
 423
 424   CORBA::String_var operation = ri->operation ();
 425
 426 #ifdef KOKYU_DSRT_LOGGING
 427   ACE_DEBUG ((LM_DEBUG,
 428               "(%t|%T): send_reply from \"%s\"\n",
 429               ri->operation ()));
 430 #endif
 431
 432   // Make the context to send the context to the target
 433   IOP::ServiceContext sc;
 434   sc.context_id = Server_Interceptor::SchedulingInfo;
 435
 436   ACE_DEBUG ((LM_DEBUG, "in send_reply: before accessing current_->sched_param\n"));
 437   CORBA::Policy_var sched_policy =
 438     this->current_->scheduling_parameter();
 439
 440   RTScheduling::Current::IdType_var guid = this->current_->id ();
 441
 442   if (CORBA::is_nil (sched_policy.in ()))
 443   {
 444     ACE_DEBUG ((LM_DEBUG, "sched_policy nil. desired_priority not set in sched params\n"));
 445     desired_priority = 0;
 446   }
 447   else
 448     {
 449       ACE_DEBUG ((LM_DEBUG, "sched_policy not nil. desired_priority set in sched params\n"));
 450
 451       FP_Scheduling::SegmentSchedulingParameterPolicy_var sched_param_policy =
 452         FP_Scheduling::SegmentSchedulingParameterPolicy::_narrow (sched_policy.in ());
 453
 454       FP_Scheduling::SegmentSchedulingParameter sched_param =
 455         sched_param_policy->value ();
 456
 457       desired_priority = sched_param.base_priority;
 458
 459       //Fill the guid in the SC Qos struct
 460       sc_qos.guid.length (guid->length ());
 461       guid_copy (sc_qos.guid, guid.in ());
 462       sc_qos.desired_priority = desired_priority;
 463       CORBA::Any sc_qos_as_any;
 464       sc_qos_as_any <<= sc_qos;
 465
 466       CORBA::OctetSeq_var cdtmp = codec_->encode (sc_qos_as_any);
 467       sc.context_data = cdtmp.in ();
 468
 469       // Add this context to the service context list.
 470       ri->add_reply_service_context (sc, 1);
 471
 472 #ifdef KOKYU_DSRT_LOGGING
 473       ACE_DEBUG ((LM_DEBUG, "(%t|%T):reply sc added\n"));
 474 #endif
 475     }
 476
 477   kokyu_dispatcher_->update_schedule (guid.in (),
 478                                       Kokyu::BLOCK);
 479
 480 #ifdef KOKYU_DSRT_LOGGING
 481   ACE_DEBUG ((LM_DEBUG, "(%t|%T): send_reply interceptor done\n"));
 482 #endif
 483 }
 484
 485 void
 486 Fixed_Priority_Scheduler::send_exception (PortableInterceptor::ServerRequestInfo_ptr ri)
 487 {
 488   send_reply (ri);
 489 }
 490
 491 void
 492 Fixed_Priority_Scheduler::send_other (PortableInterceptor::ServerRequestInfo_ptr ri)
 493 {
 494   send_reply (ri);
 495 }
 496
 497 void
 498 Fixed_Priority_Scheduler::receive_reply (PortableInterceptor::ClientRequestInfo_ptr ri)
 499 {
 500   RTScheduling::Current::IdType guid;
 501   RTCORBA::Priority desired_priority=0;
 502
 503   CORBA::String_var operation = ri->operation ();
 504
 505   CORBA::Object_var target = ri->target ();
 506
 507   ACE_CString opname = operation.in ();
 508 #ifdef KOKYU_DSRT_LOGGING
 509   ACE_DEBUG ((LM_DEBUG,
 510               "(%t|%T):receive_reply from "
 511               "\"%s\"\n",
 512               opname.c_str ()));
 513 #endif
 514
 515   // Check that the reply service context was received as
 516   // expected.
 517   IOP::ServiceContext_var sc =
 518     ri->get_reply_service_context (Client_Interceptor::SchedulingInfo);
 519
 520   if (sc.ptr () == 0)
 521     {
 522       desired_priority = 0;
 523     }
 524   else
 525     {
 526       CORBA::OctetSeq oc_seq = CORBA::OctetSeq (sc->context_data.length (),
 527                                                 sc->context_data.length (),
 528                                                 sc->context_data.get_buffer (),
 529                                                 0);
 530
 531       //Don't store in a _var, since >>= returns a pointer to an internal buffer
 532       //and we are not supposed to free it.
 533       const Kokyu::Svc_Ctxt_DSRT_QoS* sc_qos_ptr = 0;
 534       CORBA::Any sc_qos_as_any;
 535       CORBA::Any_var scqostmp = codec_->decode (oc_seq);
 536       sc_qos_as_any = scqostmp.in ();
 537       sc_qos_as_any >>= sc_qos_ptr;
 538
 539       desired_priority = sc_qos_ptr->desired_priority;
 540       guid.length (sc_qos_ptr->guid.length ());
 541       guid_copy (guid, sc_qos_ptr->guid);
 542
 543 #ifdef KOKYU_DSRT_LOGGING
 544       ACE_DEBUG ((LM_DEBUG,
 545                   "(%t): Desired_Priority = %d in recvd service context\n",
 546                   desired_priority));
 547 #endif
 548     }
 549
 550   FP_Scheduler_Traits::QoSDescriptor_t qos;
 551   qos.priority_ = desired_priority;
 552   this->kokyu_dispatcher_->schedule (guid, qos);
 553 }
 554
 555 void
 556 Fixed_Priority_Scheduler::receive_exception (PortableInterceptor::ClientRequestInfo_ptr ri)
 557 {
 558   receive_reply (ri);
 559 }
 560
 561 void
 562 Fixed_Priority_Scheduler::receive_other (PortableInterceptor::ClientRequestInfo_ptr ri)
 563 {
 564   receive_reply (ri);
 565 }
 566
 567 void
 568 Fixed_Priority_Scheduler::cancel (const RTScheduling::Current::IdType &)
 569 {
 570   throw CORBA::NO_IMPLEMENT ();
 571 }
 572
 573 CORBA::PolicyList*
 574 Fixed_Priority_Scheduler::scheduling_policies (void)
 575 {
 576   throw CORBA::NO_IMPLEMENT ();
 577 }
 578
 579 void
 580 Fixed_Priority_Scheduler::scheduling_policies (const CORBA::PolicyList &)
 581 {
 582   throw CORBA::NO_IMPLEMENT ();
 583 }
 584
 585 CORBA::PolicyList*
 586 Fixed_Priority_Scheduler::poa_policies (void)
 587 {
 588   throw CORBA::NO_IMPLEMENT ();
 589 }
 590
 591 char *
 592 Fixed_Priority_Scheduler::scheduling_discipline_name (void)
 593 {
 594   throw CORBA::NO_IMPLEMENT ();
 595 }
 596
 597 RTScheduling::ResourceManager_ptr
 598 Fixed_Priority_Scheduler::create_resource_manager (const char *,
 599                                         CORBA::Policy_ptr)
 600 {
 601   throw CORBA::NO_IMPLEMENT ();
 602 }
 603
 604 void
 605 Fixed_Priority_Scheduler::set_scheduling_parameter (PortableServer::Servant &,
 606                                          const char *,
 607                                          CORBA::Policy_ptr)
 608 {
 609   throw CORBA::NO_IMPLEMENT ();
 610 }
 611