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