TAO/tao/RTPortableServer/RT_Servant_Dispatcher.cpp

   1 // -*- C++ -*-
   2 #include "tao/RTPortableServer/RT_Servant_Dispatcher.h"
   3
   4 #if defined (TAO_HAS_CORBA_MESSAGING) && TAO_HAS_CORBA_MESSAGING != 0
   5
   6 #include "tao/RTPortableServer/RT_POA.h"
   7
   8 #include "tao/ORB_Core.h"
   9 #include "tao/ORB_Core_TSS_Resources.h"
  10 #include "tao/TAO_Server_Request.h"
  11 #include "tao/Transport.h"
  12 #include "tao/Connection_Handler.h"
  13 #include "tao/Service_Context.h"
  14 #include "tao/Protocols_Hooks.h"
  15 #include "tao/Network_Priority_Protocols_Hooks.h"
  16 #include "tao/PortableServer/Network_Priority_Hook.h"
  17 #include "tao/debug.h"
  18 #include "tao/CDR.h"
  19
  20 #include "tao/RTCORBA/Thread_Pool.h"
  21
  22 #include "ace/OS_NS_stdio.h"
  23 #include "ace/OS_NS_string.h"
  24
  25 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  26
  27 void
  28 TAO_RT_Servant_Dispatcher::pre_invoke_remote_request (
  29   TAO_Root_POA &poa,
  30   CORBA::Short servant_priority,
  31   TAO_ServerRequest &req,
  32   TAO::Portable_Server::Servant_Upcall::Pre_Invoke_State &pre_invoke_state
  33   )
  34 {
  35   TAO_Service_Context &request_service_context = req.request_service_context ();
  36   TAO_Service_Context &reply_service_context = req.reply_service_context ();
  37
  38   TAO_Thread_Pool *thread_pool =
  39     static_cast <TAO_Thread_Pool *> (poa.thread_pool ());
  40
  41   if (thread_pool != 0 &&
  42       thread_pool->with_lanes ())
  43     {
  44       //
  45       // We don't mess with the priority of threads in lanes.
  46       //
  47
  48       if (TAO_debug_level > 0)
  49         {
  50           // Get the ORB_Core's TSS resources.
  51           TAO_ORB_Core_TSS_Resources *tss =
  52             poa.orb_core ().get_tss_resources ();
  53
  54           /// Get the lane attribute in TSS.
  55           TAO_Thread_Lane *lane =
  56             static_cast<TAO_Thread_Lane *> (tss->lane_);
  57
  58           ACE_ASSERT (lane->pool ().id () ==
  59                       thread_pool->id ());
  60
  61           TAOLIB_DEBUG ((LM_DEBUG,
  62                       ACE_TEXT ("Using thread pool with lane ")
  63                       ACE_TEXT ("(%P|%t|%d|%d): original thread ")
  64                       ACE_TEXT ("CORBA/native priority %d/%d not changed\n"),
  65                       lane->pool ().id (),
  66                       lane->id (),
  67                       lane->lane_priority (),
  68                       lane->native_priority ()));
  69         }
  70
  71       return;
  72     }
  73
  74   // Remember current thread's priority.
  75   TAO_Protocols_Hooks *tph = poa.orb_core ().get_protocols_hooks ();
  76
  77   if (tph != 0)
  78     {
  79       const char *priority_model = 0;
  80       RTCORBA::Priority target_priority = TAO_INVALID_PRIORITY;
  81
  82       // NOT_SPECIFIED PriorityModel processing.
  83       if (poa.priority_model () ==
  84           TAO::Portable_Server::Cached_Policies::NOT_SPECIFIED)
  85         {
  86           priority_model = "RTCORBA::NOT_SPECIFIED";
  87         }
  88
  89       // CLIENT_PROPAGATED PriorityModel processing.
  90       else if (poa.priority_model () ==
  91           TAO::Portable_Server::Cached_Policies::CLIENT_PROPAGATED)
  92         {
  93           priority_model = "RTCORBA::CLIENT_PROPAGATED";
  94
  95           // Attempt to extract client-propagated priority from the
  96           // ServiceContextList of the request.
  97           const IOP::ServiceContext *context = 0;
  98
  99           if (request_service_context.get_context (IOP::RTCorbaPriority,
 100                                                    &context) == 1)
 101             {
 102               // Extract the target priority
 103               TAO_InputCDR cdr (reinterpret_cast
 104                                 <const char*>
 105                                  (context->context_data.get_buffer ()),
 106                                 context->context_data.length ());
 107               CORBA::Boolean byte_order;
 108               if (!(cdr >> ACE_InputCDR::to_boolean (byte_order)))
 109                 throw ::CORBA::MARSHAL ();
 110               cdr.reset_byte_order (static_cast<int> (byte_order));
 111
 112               if (!(cdr >> target_priority))
 113                 throw ::CORBA::MARSHAL ();
 114
 115               // Save the target priority in the response service
 116               // context to propagate back to the client as specified
 117               // by the RTCORBA specification.
 118               reply_service_context.set_context (*context);
 119             }
 120           else
 121             {
 122               // Use default priority if none came in the request.
 123               // (Request must have come from a non-RT ORB.)
 124               target_priority = poa.server_priority ();
 125             }
 126         }
 127       else
 128         // SERVER_DECLARED PriorityModel processing.
 129         {
 130           priority_model = "RTCORBA::SERVER_DECLARED";
 131
 132           // Use the request associated with the servant.
 133           target_priority = servant_priority;
 134         }
 135
 136       char thread_pool_id[BUFSIZ];
 137       if (TAO_debug_level > 0)
 138         {
 139           if (thread_pool == 0)
 140             ACE_OS::strcpy (thread_pool_id,
 141                             "default thread pool");
 142           else
 143             ACE_OS::sprintf (thread_pool_id,
 144                              "thread pool %d",
 145                              thread_pool->id ());
 146         }
 147
 148       // Target priority is invalid.
 149       if (target_priority == TAO_INVALID_PRIORITY)
 150         {
 151           if (TAO_debug_level > 0)
 152             {
 153     // If we are in a multi-threaded configuration, print out the current
 154     // thread priority.
 155     #if defined (ACE_HAS_THREADS)
 156
 157               if (tph->get_thread_CORBA_and_native_priority (
 158                     pre_invoke_state.original_CORBA_priority_,
 159                     pre_invoke_state.original_native_priority_) == -1)
 160                 throw ::CORBA::DATA_CONVERSION (CORBA::OMGVMCID | 2,
 161                                                 CORBA::COMPLETED_NO);
 162
 163               TAOLIB_DEBUG ((LM_DEBUG,
 164                           ACE_TEXT ("(%P|%t): %C processing using %C ")
 165                           ACE_TEXT ("original thread CORBA/native priority %d/%d ")
 166                           ACE_TEXT ("not changed\n"),
 167                           priority_model,
 168                           thread_pool_id,
 169                           pre_invoke_state.original_CORBA_priority_,
 170                           pre_invoke_state.original_native_priority_));
 171
 172     // If we are in a single-threaded configuration, we cannot get the
 173     // current thread priority.  Therefore, print out a simpler message.
 174     #else /* ACE_HAS_THREADS */
 175
 176               TAOLIB_DEBUG ((LM_DEBUG,
 177                           ACE_TEXT ("(%P|%t): %C processing using %C ")
 178                           ACE_TEXT ("original thread CORBA/native priority ")
 179                           ACE_TEXT ("not changed\n"),
 180                           priority_model,
 181                           thread_pool_id));
 182
 183     #endif /* ACE_HAS_THREADS */
 184
 185             }
 186         }
 187       else
 188         {
 189           // Get the current thread's priority.
 190
 191           if (tph->get_thread_CORBA_and_native_priority (
 192                 pre_invoke_state.original_CORBA_priority_,
 193                 pre_invoke_state.original_native_priority_) == -1)
 194             throw ::CORBA::DATA_CONVERSION (CORBA::OMGVMCID | 2,
 195                                             CORBA::COMPLETED_NO);
 196
 197           // Priority needs to be temporarily changed for the
 198           // duration of request.
 199           if (target_priority != pre_invoke_state.original_CORBA_priority_)
 200             {
 201               if (tph->set_thread_CORBA_priority (target_priority) == -1)
 202                 throw ::CORBA::DATA_CONVERSION (CORBA::OMGVMCID | 2,
 203                                                 CORBA::COMPLETED_NO);
 204
 205               pre_invoke_state.state_ =
 206                 TAO::Portable_Server::Servant_Upcall::Pre_Invoke_State::PRIORITY_RESET_REQUIRED;
 207
 208               if (TAO_debug_level > 0)
 209                 {
 210                   CORBA::Short native_priority = 0;
 211                   tph->get_thread_native_priority (native_priority);
 212
 213                   TAOLIB_DEBUG ((LM_DEBUG,
 214                               ACE_TEXT ("%C processing using %C ")
 215                               ACE_TEXT ("(%P|%t): original thread CORBA/native priority %d/%d ")
 216                               ACE_TEXT ("temporarily changed to CORBA/native priority %d/%d\n"),
 217                               priority_model,
 218                               thread_pool_id,
 219                               pre_invoke_state.original_CORBA_priority_,
 220                               pre_invoke_state.original_native_priority_,
 221                               target_priority,
 222                               native_priority));
 223                 }
 224             }
 225           // No change in priority required.
 226           else
 227             {
 228               if (TAO_debug_level > 0)
 229                 {
 230                   TAOLIB_DEBUG ((LM_DEBUG,
 231                               ACE_TEXT ("%C processing using %C ")
 232                               ACE_TEXT ("(%P|%t): original thread CORBA/native priority %d/%d ")
 233                               ACE_TEXT ("is the same as the target priority\n"),
 234                               priority_model,
 235                               thread_pool_id,
 236                               pre_invoke_state.original_CORBA_priority_,
 237                               pre_invoke_state.original_native_priority_));
 238                 }
 239             }
 240         }
 241     }
 242
 243   TAO_Network_Priority_Protocols_Hooks *nph =
 244     poa.orb_core ().get_network_priority_protocols_hooks ();
 245
 246   if (nph != 0)
 247     {
 248       poa.network_priority_hook ()-> set_dscp_codepoint (req, poa);
 249     }
 250   else if (tph != 0)
 251     {
 252       CORBA::Policy_var policy =
 253         poa.policies ().get_cached_policy (
 254           TAO_CACHED_POLICY_RT_SERVER_PROTOCOL);
 255       CORBA::Boolean set_server_network_priority =
 256         tph->set_server_network_priority (
 257           req.transport ()->tag (), policy.in ());
 258       TAO_Connection_Handler *connection_handler =
 259         req.transport ()->connection_handler ();
 260       connection_handler->set_dscp_codepoint (set_server_network_priority);
 261     }
 262 }
 263
 264 void
 265 TAO_RT_Servant_Dispatcher::pre_invoke_collocated_request (TAO_Root_POA &poa,
 266                                                           CORBA::Short servant_priority,
 267                                                           TAO::Portable_Server::Servant_Upcall::Pre_Invoke_State &pre_invoke_state)
 268 {
 269   TAO_Thread_Pool *thread_pool =
 270     static_cast <TAO_Thread_Pool *> (poa.thread_pool ());
 271
 272   if (thread_pool == 0 || thread_pool->with_lanes ())
 273     {
 274       //
 275       // We don't mess with the priority of threads in lanes or for
 276       // the default thread pool.
 277       //
 278       return;
 279     }
 280
 281   if (poa.priority_model () !=
 282       TAO::Portable_Server::Cached_Policies::SERVER_DECLARED ||
 283       servant_priority == TAO_INVALID_PRIORITY)
 284     {
 285       //
 286       // We either don't have server declared model or servant
 287       // priority is invalid.
 288       //
 289       return;
 290     }
 291
 292   //
 293   // SERVER_DECLARED PriorityModel processing.
 294   //
 295
 296   // Remember current thread's priority.
 297   TAO_Protocols_Hooks *tph = poa.orb_core ().get_protocols_hooks ();
 298
 299   if (tph != 0)
 300     {
 301       if (tph->get_thread_CORBA_and_native_priority (
 302             pre_invoke_state.original_CORBA_priority_,
 303             pre_invoke_state.original_native_priority_) == -1)
 304         throw ::CORBA::DATA_CONVERSION (CORBA::OMGVMCID | 2, CORBA::COMPLETED_NO);
 305
 306       // Change the priority of the current thread for the duration of
 307       // request.
 308       if (servant_priority != pre_invoke_state.original_CORBA_priority_)
 309         {
 310           if (tph->set_thread_CORBA_priority (servant_priority) == -1)
 311             throw ::CORBA::DATA_CONVERSION (CORBA::OMGVMCID | 2, CORBA::COMPLETED_NO);
 312
 313           pre_invoke_state.state_ =
 314             TAO::Portable_Server::Servant_Upcall::Pre_Invoke_State::PRIORITY_RESET_REQUIRED;
 315         }
 316     }
 317 }
 318
 319 void
 320 TAO_RT_Servant_Dispatcher::post_invoke (TAO_Root_POA &poa,
 321                                         TAO::Portable_Server::Servant_Upcall::Pre_Invoke_State &pre_invoke_state)
 322
 323 {
 324   if (pre_invoke_state.state_ ==
 325       TAO::Portable_Server::Servant_Upcall::Pre_Invoke_State::PRIORITY_RESET_REQUIRED)
 326     {
 327       pre_invoke_state.state_ =
 328         TAO::Portable_Server::Servant_Upcall::Pre_Invoke_State::NO_ACTION_REQUIRED;
 329
 330       try
 331         {
 332           // Reset the priority of the current thread back to its original
 333           // value.
 334           TAO_Protocols_Hooks *tph = poa.orb_core ().get_protocols_hooks ();
 335
 336           if (tph != 0)
 337             {
 338               if (tph->restore_thread_CORBA_and_native_priority (
 339                        pre_invoke_state.original_CORBA_priority_,
 340                        pre_invoke_state.original_native_priority_) == -1)
 341                 throw ::CORBA::DATA_CONVERSION (CORBA::OMGVMCID | 2,
 342                                                 CORBA::COMPLETED_NO);
 343             }
 344         }
 345       catch (const ::CORBA::Exception& ex)
 346         {
 347           // Eat up the exception.
 348           ex._tao_print_exception (
 349             "Exception caught: TAO - "
 350             "Priority_Model_Processing::"
 351             "~Priority_Model_Processing");
 352         }
 353     }
 354 }
 355
 356 TAO_Root_POA *
 357 TAO_RT_Servant_Dispatcher::create_Root_POA (const ACE_CString &name,
 358                                             PortableServer::POAManager_ptr poa_manager,
 359                                             const TAO_POA_Policy_Set &policies,
 360                                             ACE_Lock &lock,
 361                                             TAO_SYNCH_MUTEX &thread_lock,
 362                                             TAO_ORB_Core &orb_core,
 363                                             TAO_Object_Adapter *object_adapter)
 364 {
 365   TAO_RT_POA *poa = 0;
 366
 367   ACE_NEW_THROW_EX (poa,
 368                     TAO_RT_POA (name,
 369                                 poa_manager,
 370                                 policies,
 371                                 0,
 372                                 lock,
 373                                 thread_lock,
 374                                 orb_core,
 375                                 object_adapter),
 376                     CORBA::NO_MEMORY ());
 377
 378   return poa;
 379 }
 380
 381 TAO_END_VERSIONED_NAMESPACE_DECL
 382
 383 #endif /* TAO_HAS_CORBA_MESSAGING && TAO_HAS_CORBA_MESSAGING != 0 */