TAO/tao/RTScheduling/Request_Interceptor.cpp

   1 #include "tao/AnyTypeCode/Any.h"
   2 #include "tao/AnyTypeCode/TypeCode.h"
   3
   4 #include "tao/RTScheduling/Request_Interceptor.h"
   5 #include "tao/RTScheduling/Current.h"
   6 #include "tao/RTScheduling/Distributable_Thread.h"
   7
   8 #include "tao/TSS_Resources.h"
   9 #include "tao/debug.h"
  10 #include "tao/ORB_Constants.h"
  11 #include "ace/OS_NS_string.h"
  12
  13 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  14
  15 const IOP::ServiceId
  16 Client_Interceptor::SchedulingInfo = 30;
  17
  18 void
  19 Client_Interceptor::send_request (PortableInterceptor::ClientRequestInfo_ptr ri)
  20 {
  21   if (TAO_debug_level > 0)
  22     TAOLIB_DEBUG ((LM_DEBUG,
  23                 "Client_Interceptor::send_request\n"));
  24
  25   // Temporary current.
  26   TAO_RTScheduler_Current_i *new_current = 0;
  27   TAO_RTScheduler_Current_i *current = 0;
  28
  29   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
  30
  31   current = static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_current_impl_);
  32
  33   if (current != 0)
  34     {
  35       // If this is a one way request
  36       if (!ri->response_expected ())
  37         {
  38           // Generate GUID.
  39           RTScheduling::Current::IdType guid;
  40           guid.length (sizeof(size_t));
  41
  42           size_t temp = ++TAO_RTScheduler_Current::guid_counter;
  43           ACE_OS::memcpy (guid.get_buffer (),
  44                           &temp,
  45                           sizeof(size_t));
  46
  47           size_t id;
  48           ACE_OS::memcpy (&id,
  49                           guid.get_buffer (),
  50                           guid.length ());
  51
  52           if (TAO_debug_level > 0)
  53             TAOLIB_DEBUG ((LM_DEBUG,
  54                         "The Guid is %d %d\n",
  55                         id,
  56                         TAO_RTScheduler_Current::guid_counter.load ()));
  57
  58           // Create new DT.
  59           RTScheduling::DistributableThread_var dt =
  60             TAO_DistributableThread_Factory::create_DT ();
  61
  62           // Add new DT to map.
  63           int result = current->dt_hash ()->bind (guid, dt);
  64           if (result != 0)
  65             {
  66               TAOLIB_DEBUG ((LM_DEBUG,
  67                           "No Scheduling Segment Context\n"));
  68               throw ::CORBA::INTERNAL ();
  69             }
  70
  71           // @@ Store implicit_sched_param in a var
  72
  73           // Create new temporary current. Note that
  74           // the new <sched_param> is the current
  75           // <implicit_sched_param> and there is no
  76           // segment name.
  77           CORBA::Policy_var implicit_sched_param =
  78             current->implicit_scheduling_parameter ();
  79           ACE_NEW (new_current,
  80                    TAO_RTScheduler_Current_i (current->orb (),
  81                                               current->dt_hash (),
  82                                               guid,
  83                                               0,
  84                                               implicit_sched_param.in (),
  85                                               0,
  86                                               dt.in (),
  87                                               current));
  88
  89           // Install new current in the ORB.
  90           //current->implementation (new_current);
  91           tss->rtscheduler_current_impl_ = new_current;
  92         }
  93
  94       // Scheduler populates the service context with
  95       // scheduling parameters.
  96       RTScheduling::Scheduler_var scheduler = current->scheduler ();
  97       scheduler->send_request (ri);
  98
  99       // If this is a one way request
 100       if (!ri->response_expected ())
 101         {
 102           // Cleanup temporary DT.
 103           new_current->cleanup_DT ();
 104
 105           //Restore old current
 106           new_current->cleanup_current ();
 107         }
 108     }
 109 }
 110
 111 void
 112 Client_Interceptor::send_poll (PortableInterceptor::ClientRequestInfo_ptr ri)
 113 {
 114   if (TAO_debug_level > 0)
 115     {
 116       TAOLIB_DEBUG ((LM_DEBUG,
 117                   "Client_Interceptor::send_poll\n"));
 118     }
 119
 120   TAO_RTScheduler_Current_i *current = 0;
 121
 122   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
 123
 124   current = static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_current_impl_);
 125   if (current != 0)
 126     {
 127       RTScheduling::Scheduler_var scheduler = current->scheduler ();
 128       scheduler->send_poll (ri);
 129     }
 130 }
 131
 132 void
 133 Client_Interceptor::receive_reply (PortableInterceptor::ClientRequestInfo_ptr ri)
 134 {
 135   if (TAO_debug_level > 0)
 136     {
 137       TAOLIB_DEBUG ((LM_DEBUG,
 138                   "Client_Interceptor::receive_reply\n"));
 139     }
 140
 141   TAO_RTScheduler_Current_i *current = 0;
 142
 143   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
 144
 145   current = static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_current_impl_);
 146   if (current != 0)
 147     {
 148       RTScheduling::Scheduler_var scheduler = current->scheduler ();
 149       scheduler->receive_reply (ri);
 150     }
 151 }
 152
 153 void
 154 Client_Interceptor::receive_exception (PortableInterceptor::ClientRequestInfo_ptr ri)
 155 {
 156   if (TAO_debug_level > 0)
 157     {
 158       TAOLIB_DEBUG ((LM_DEBUG,
 159                   "Client_Interceptor::receive_exception\n"));
 160     }
 161
 162   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
 163
 164   TAO_RTScheduler_Current_i *current =
 165     static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_current_impl_);
 166
 167   if (current != 0)
 168     {
 169       if (ri == 0)
 170         {
 171           TAOLIB_ERROR ((LM_ERROR,
 172                       "Client_Interceptor::receive_exception ri = 0\n"));
 173           return;
 174         }
 175
 176       CORBA::Any_var ex = ri->received_exception ();
 177       CORBA::TypeCode_var type = ex->type ();
 178
 179       if (CORBA::is_nil (type.in ()))
 180         {
 181           TAOLIB_ERROR ((LM_ERROR,
 182                       "type = 0\n"));
 183           return;
 184         }
 185       const char * id = type->id ();
 186
 187       if (TAO_debug_level > 0)
 188         TAOLIB_DEBUG ((LM_DEBUG,
 189                     "Received Exception %C\n",
 190                     id));
 191
 192       // If the remote host threw a THREAD_CANCELLED
 193       // exception, make sure to take the appropriate
 194       // local action.
 195       if (ACE_OS::strstr (id, "CORBA::THREAD_CANCELLED") == 0)
 196         {
 197           // Perform the necessary cleanup as the
 198           // thread was cancelled.
 199           current->cancel_thread ();
 200         }
 201       else
 202         {
 203           // Inform scheduler that exception was
 204           // received.
 205           RTScheduling::Scheduler_var scheduler = current->scheduler ();
 206           scheduler->receive_exception (ri);
 207         }
 208     }
 209 }
 210
 211 void
 212 Client_Interceptor::receive_other (PortableInterceptor::ClientRequestInfo_ptr ri)
 213 {
 214   if (TAO_debug_level > 0)
 215     {
 216       TAOLIB_DEBUG ((LM_DEBUG,
 217                   "Client_Interceptor::receive_other\n"));
 218     }
 219
 220   TAO_RTScheduler_Current_i *current = 0;
 221
 222   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
 223
 224   current = static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_current_impl_);
 225   if (current != 0)
 226     {
 227       RTScheduling::Scheduler_var scheduler = current->scheduler ();
 228       scheduler->receive_other (ri);
 229     }
 230 }
 231
 232 char*
 233 Client_Interceptor::name ()
 234 {
 235   return CORBA::string_dup ("RTSchdeuler_Client_Interceptor");
 236 }
 237
 238 void
 239 Client_Interceptor::destroy ()
 240 {
 241 }
 242
 243 const IOP::ServiceId
 244 Server_Interceptor::SchedulingInfo = 30;
 245
 246 Server_Interceptor::Server_Interceptor (TAO_RTScheduler_Current_ptr current)
 247 {
 248   this->current_ = TAO_RTScheduler_Current::_duplicate (current);
 249 }
 250
 251 void
 252 Server_Interceptor::receive_request_service_contexts (
 253   PortableInterceptor::ServerRequestInfo_ptr)
 254 {
 255   if (TAO_debug_level > 0)
 256     {
 257       TAOLIB_DEBUG ((LM_DEBUG,
 258                   "Server_Interceptor::receive_request_service_contexts\n"));
 259     }
 260 }
 261
 262 void
 263 Server_Interceptor::receive_request (PortableInterceptor::ServerRequestInfo_ptr ri)
 264 {
 265   if (TAO_debug_level > 0)
 266     {
 267       TAOLIB_DEBUG ((LM_DEBUG,
 268                   "Server_Interceptor::receive_request\n"));
 269     }
 270
 271   IOP::ServiceContext_var serv_cxt;
 272
 273   try
 274     {
 275       serv_cxt =
 276         ri->get_request_service_context (Server_Interceptor::SchedulingInfo);
 277     }
 278   catch (const ::CORBA::Exception&)
 279     {
 280       if (TAO_debug_level > 0)
 281         TAOLIB_DEBUG ((LM_DEBUG,
 282                     "Invalid Service Request\n"));
 283       return;
 284     }
 285
 286   if (TAO_debug_level > 0)
 287     TAOLIB_DEBUG ((LM_DEBUG,
 288                 "Request from Distributable Thread\n"));
 289
 290   RTScheduling::Current::IdType_var guid_var;
 291   char* name = 0;
 292   CORBA::Policy_var sched_param = 0;
 293   CORBA::Policy_var implicit_sched_param = 0;
 294
 295   TAO_RTScheduler_Current_i* new_current = 0;
 296   ACE_NEW_THROW_EX (new_current,
 297                     TAO_RTScheduler_Current_i (this->current_->orb (),
 298                                                this->current_->dt_hash ()),
 299                     CORBA::NO_MEMORY (
 300                            CORBA::SystemException::_tao_minor_code (
 301                              TAO::VMCID,
 302                              ENOMEM),
 303                            CORBA::COMPLETED_NO));
 304
 305   // Scheduler retrieves scheduling parameters
 306   // from request and populates the out
 307   // parameters.
 308   RTScheduling::Scheduler_var scheduler = new_current->scheduler();
 309   scheduler->receive_request (ri,
 310                               guid_var.out (),
 311                               name,
 312                               sched_param.out (),
 313                               implicit_sched_param.out ());
 314
 315   if (guid_var->length () == 0)
 316     {
 317       TAOLIB_ERROR ((LM_ERROR,
 318                   "The scheduler MUST retrieve and return the "
 319                   "GUID from the service context\n"));
 320       return;
 321     }
 322   RTScheduling::Current::IdType guid;
 323   guid.length (sizeof (size_t));
 324   ACE_OS::memcpy (guid.get_buffer (),
 325                   guid_var->get_buffer (),
 326                   sizeof (size_t));
 327
 328   size_t id;
 329   ACE_OS::memcpy (&id,
 330                   guid.get_buffer (),
 331                   guid.length ());
 332
 333   if (TAO_debug_level > 0)
 334     TAOLIB_DEBUG ((LM_DEBUG,
 335                 "The Guid is %d\n",
 336                 id));
 337
 338   // Create new DT.
 339   RTScheduling::DistributableThread_var dt = TAO_DistributableThread_Factory::create_DT ();
 340
 341   // Add new DT to map.
 342   int result = new_current->dt_hash ()->bind (guid, dt);
 343
 344   if (result != 0)
 345     {
 346       throw ::CORBA::INTERNAL ();
 347     }
 348
 349   // Create new temporary current. Note that the new <sched_param> is
 350   // the current <implicit_sched_param> and there is no segment name.
 351   new_current->id (guid);
 352   new_current->name (name);
 353   new_current->scheduling_parameter (sched_param.in ());
 354   new_current->implicit_scheduling_parameter (implicit_sched_param.in ());
 355   new_current->DT (dt.in ());
 356
 357   // Install new current in the ORB and store the previous current
 358   // implementation
 359   // current->implementation (new_current)
 360   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
 361
 362   tss->rtscheduler_previous_current_impl_ = this->current_->implementation (new_current);
 363 }
 364
 365 void
 366 Server_Interceptor::send_reply (PortableInterceptor::ServerRequestInfo_ptr ri)
 367 {
 368   if (TAO_debug_level > 0)
 369     {
 370       TAOLIB_DEBUG ((LM_DEBUG,
 371                   "Server_Interceptor::send_reply\n"));
 372     }
 373
 374   TAO_RTScheduler_Current_i *current = 0;
 375   TAO_RTScheduler_Current_i *prev_current = 0;
 376
 377   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
 378
 379   current = static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_current_impl_);
 380   if (current != 0)
 381     {
 382       RTScheduling::DistributableThread_var dt = current->DT ();
 383       if (dt->state () == RTScheduling::DistributableThread::CANCELLED)
 384         {
 385           current->cancel_thread ();
 386
 387           return;
 388         }
 389       else TAOLIB_DEBUG ((LM_DEBUG,
 390                        "Thread Not Cancelled\n"));
 391
 392
 393       // Inform scheduler that upcall is complete.
 394       RTScheduling::Scheduler_var scheduler = current->scheduler ();
 395       scheduler->send_reply (ri);
 396
 397       current->cleanup_DT ();
 398       current->cleanup_current ();
 399
 400       // Get the previous current if any.
 401       prev_current = static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_previous_current_impl_);
 402
 403       // Restore the previous current.
 404       tss->rtscheduler_current_impl_ = prev_current;
 405
 406       // Reset the previous current pointer.
 407       tss->rtscheduler_previous_current_impl_ = 0;
 408     }
 409   else TAOLIB_DEBUG ((LM_DEBUG,
 410                    "Send Reply Current is 0\n"));
 411 }
 412
 413 void
 414 Server_Interceptor::send_exception (PortableInterceptor::ServerRequestInfo_ptr ri)
 415 {
 416   if (TAO_debug_level > 0)
 417     {
 418       TAOLIB_DEBUG ((LM_DEBUG,
 419                   "Server_Interceptor::send_exception\n"));
 420     }
 421
 422   TAO_RTScheduler_Current_i *current = 0;
 423
 424   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
 425
 426   current = static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_current_impl_);
 427   if (current != 0)
 428     {
 429       // Inform scheduler that upcall is complete.
 430       RTScheduling::Scheduler_var scheduler = current->scheduler ();
 431       scheduler->send_exception (ri);
 432
 433       current->cleanup_DT ();
 434       current->cleanup_current ();
 435     }
 436 }
 437
 438 void
 439 Server_Interceptor::send_other (PortableInterceptor::ServerRequestInfo_ptr ri)
 440 {
 441   if (TAO_debug_level > 0)
 442     {
 443       TAOLIB_DEBUG ((LM_DEBUG,
 444                   "Server_Interceptor::send_other\n"));
 445     }
 446
 447   TAO_RTScheduler_Current_i *current = 0;
 448
 449   TAO_TSS_Resources *tss = TAO_TSS_Resources::instance ();
 450
 451   current = static_cast<TAO_RTScheduler_Current_i *> (tss->rtscheduler_current_impl_);
 452   if (current != 0)
 453     {
 454       // Inform scheduler that upcall is complete.
 455       RTScheduling::Scheduler_var scheduler = current->scheduler ();
 456       scheduler->send_other (ri);
 457
 458       current->cleanup_DT ();
 459       current->cleanup_current ();
 460     }
 461 }
 462
 463 char*
 464 Server_Interceptor::name ()
 465 {
 466     return  CORBA::string_dup ("RTSchdeuler_Server_Interceptor");
 467 }
 468
 469 void
 470 Server_Interceptor::destroy ()
 471 {
 472 }
 473
 474 TAO_END_VERSIONED_NAMESPACE_DECL