rt_poa.cpp

来自「这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用」· C++ 代码 · 共 952 行 · 第 1/3 页

// @(#) RT_POA.cpp,v 1.13 2003/08/17 07:26:27 ossama Exp

#include "RT_POA.h"

#if defined (TAO_HAS_CORBA_MESSAGING) && TAO_HAS_CORBA_MESSAGING != 0

#include "RT_Acceptor_Filters.h"

#include "tao/ORB_Core.h"
#include "tao/ORB.h"
#include "tao/Server_Strategy_Factory.h"
#include "tao/Environment.h"
#include "tao/Exception.h"
#include "tao/Stub.h"
#include "tao/Policy_Manager.h"
#include "tao/debug.h"
#include "tao/RTCORBA/Thread_Pool.h"
#include "tao/Thread_Lane_Resources.h"
#include "tao/Acceptor_Registry.h"

#include "tao/RTCORBA/RT_Policy_i.h"

#include "tao/PortableServer/Default_Acceptor_Filter.h"
#include "tao/RTPortableServer/RT_Policy_Validator.h"

#include "ace/Auto_Ptr.h"

#if !defined (__ACE_INLINE__)
# include "RT_POA.i"
#endif /* ! __ACE_INLINE__ */


ACE_RCSID (RTPortableServer,
           RT_POA,
           "RT_POA.cpp,v 1.13 2003/08/17 07:26:27 ossama Exp")


TAO_RT_POA::TAO_RT_POA (const TAO_POA::String &name,
                        TAO_POA_Manager &poa_manager,
                        const TAO_POA_Policy_Set &policies,
                        TAO_POA *parent,
                        ACE_Lock &lock,
                        TAO_SYNCH_MUTEX &thread_lock,
                        TAO_ORB_Core &orb_core,
                        TAO_Object_Adapter *object_adapter
                        ACE_ENV_ARG_DECL)
  : TAO_POA (name,
             poa_manager,
             policies,
             parent,
             lock,
             thread_lock,
             orb_core,
             object_adapter
             ACE_ENV_ARG_PARAMETER),
  thread_pool_ (0)
{
  ACE_CHECK;

  // Parse the RT policies and update our policy cache.
  this->parse_rt_policies (this->policies ()
                           ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

TAO_RT_POA::~TAO_RT_POA (void)
{
}

TAO_POA *
TAO_RT_POA::new_POA (const String &name,
                     TAO_POA_Manager &poa_manager,
                     const TAO_POA_Policy_Set &policies,
                     TAO_POA *parent,
                     ACE_Lock &lock,
                     TAO_SYNCH_MUTEX &thread_lock,
                     TAO_ORB_Core &orb_core,
                     TAO_Object_Adapter *object_adapter
                     ACE_ENV_ARG_DECL)
{
  TAO_RT_POA *poa;

  ACE_NEW_THROW_EX (poa,
                    TAO_RT_POA (name,
                                poa_manager,
                                policies,
                                parent,
                                lock,
                                thread_lock,
                                orb_core,
                                object_adapter
                                ACE_ENV_ARG_PARAMETER),
                    CORBA::NO_MEMORY ());
  ACE_CHECK_RETURN (0);

  return poa;
}

void
TAO_RT_POA::parse_rt_policies (TAO_POA_Policy_Set &policies
                               ACE_ENV_ARG_DECL)
{
  {
    CORBA::Policy_var policy =
      policies.get_cached_policy (TAO_CACHED_POLICY_PRIORITY_MODEL);

    RTCORBA::PriorityModelPolicy_var priority_model =
      RTCORBA::PriorityModelPolicy::_narrow (policy.in ()
                                             ACE_ENV_ARG_PARAMETER);
    ACE_CHECK;

    if (!CORBA::is_nil (priority_model.in ()))
      {
        RTCORBA::PriorityModel rt_priority_model =
          priority_model->priority_model (ACE_ENV_SINGLE_ARG_PARAMETER);
        ACE_CHECK;

        this->cached_policies ().priority_model (
          TAO_POA_Cached_Policies::PriorityModel (rt_priority_model));

        RTCORBA::Priority priority =
          priority_model->server_priority (ACE_ENV_SINGLE_ARG_PARAMETER);
        ACE_CHECK;

        this->cached_policies ().server_priority (priority);
      }
  }

  this->thread_pool_ =
    TAO_POA_RT_Policy_Validator::extract_thread_pool (this->orb_core_,
                                                      policies.policies ()
                                                      ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
TAO_RT_POA::validate_priority (RTCORBA::Priority priority
                               ACE_ENV_ARG_DECL)
{
  if (priority < RTCORBA::minPriority
           // The line below will always be false unless the value of
           // RTCORBA::maxPriority, which is now assigned the value of
           // 32767, is changed in RTCORBA.pidl.
//      || priority > RTCORBA::maxPriority
     )
    {
      ACE_THROW (CORBA::BAD_PARAM ());
    }

  // If this POA is using a thread pool with lanes, make sure the
  // priority matches one of the thread lanes.  Note that in this
  // case, bands do not matter since matching the lanes priority is a
  // stricter condition than meeting the band ranges.  In addition,
  // when the POA was created, the bands had to match the lanes.
  if (this->thread_pool_ != 0 &&
      this->thread_pool_->with_lanes ())
    {
      TAO_Thread_Lane **lanes =
        this->thread_pool_->lanes ();

      for (CORBA::ULong i = 0;
           i != this->thread_pool_->number_of_lanes ();
           ++i)
        {
          if (lanes[i]->lane_priority () == priority)
            return;
        }

      ACE_THROW (CORBA::BAD_PARAM ());
    }
  else
    // Else we are dealing with a thread pool without lanes.
    {
      // Check if we have bands.
      CORBA::Policy_var bands =
        this->policies ().get_cached_policy (TAO_CACHED_POLICY_RT_PRIORITY_BANDED_CONNECTION);

      RTCORBA::PriorityBandedConnectionPolicy_var priority_bands
        = RTCORBA::PriorityBandedConnectionPolicy::_narrow (bands.in ()
                                                            ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      TAO_PriorityBandedConnectionPolicy *priority_bands_i =
        ACE_dynamic_cast (TAO_PriorityBandedConnectionPolicy *,
                          priority_bands.in ());

      if (priority_bands_i)
        {
          // If we do have bands, make sure that the priority is
          // matching one of the bands.
          RTCORBA::PriorityBands &bands =
            priority_bands_i->priority_bands_rep ();

          for (CORBA::ULong i = 0;
               i < bands.length ();
               ++i)
            {
              if (bands[i].low <= priority &&
                  bands[i].high >= priority)
                return;
            }

          ACE_THROW (CORBA::BAD_PARAM ());
        }
    }
}

void
TAO_RT_POA::validate_policies (ACE_ENV_SINGLE_ARG_DECL)
{
  // For each of the above operations, if the POA supports the
  // IMPLICIT_ACTIVATION option for the ImplicitActivationPolicy then
  // the ORB shall raise a WrongPolicy user exception. This relieves
  // an ORB implementation of the need to retrieve the target object's
  // priority from "somewhere" when a request arrives for an inactive
  // object.
  if (this->cached_policies_.implicit_activation () ==
      PortableServer::IMPLICIT_ACTIVATION)
    {
      ACE_THROW (PortableServer::POA::WrongPolicy ());
    }

  // For each of the above operations, if the POA does not support the
  // SERVER_DECLARED option for the PriorityModelPolicy then the ORB
  // shall raise a WrongPolicy user exception.
  if (this->cached_policies_.priority_model () !=
      TAO_POA_Cached_Policies::SERVER_DECLARED)
    {
      ACE_THROW (PortableServer::POA::WrongPolicy ());
    }

  // In all other respects the semantics of the corresponding
  // (i.e. without the name extensions "_with_priority" and
  // "_and_priority") PortableServer::POA operations shall be
  // observed.
}

TAO_Stub *
TAO_RT_POA::key_to_stub_i (const TAO::ObjectKey &object_key,
                           const char *type_id,
                           CORBA::Short priority
                           ACE_ENV_ARG_DECL)
{
  // Client exposed policies.
  CORBA::PolicyList_var client_exposed_policies =
    this->client_exposed_policies (priority
                                   ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (0);

  // Server protocol policy.
  CORBA::Policy_var protocol =
    this->policies ().get_cached_policy (TAO_CACHED_POLICY_RT_SERVER_PROTOCOL);

  RTCORBA::ServerProtocolPolicy_var server_protocol_policy =
    RTCORBA::ServerProtocolPolicy::_narrow (protocol.in ()
                                            ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (0);

  TAO_ServerProtocolPolicy *server_protocol =
    ACE_dynamic_cast (TAO_ServerProtocolPolicy *,
                      server_protocol_policy.in ());

  // Filter for server protocol.
  TAO_Server_Protocol_Acceptor_Filter filter (server_protocol->protocols_rep ());

  // If this POA is using the default thread pool or a thread pool
  // without lanes, create the IOR with the acceptors in the thread
  // pool.
  if (this->thread_pool_ == 0 ||
      !this->thread_pool_->with_lanes ())
    {
      TAO_Acceptor_Registry *acceptor_registry =
        TAO_POA_RT_Policy_Validator::extract_acceptor_registry (this->orb_core_,
                                                                this->thread_pool_);

      return
        this->TAO_POA::create_stub_object (object_key,
                                           type_id,
                                           client_exposed_policies._retn (),
                                           &filter,
                                           *acceptor_registry
                                           ACE_ENV_ARG_PARAMETER);
    }

  // If this POA has the SERVER_DECLARED policy, create the IOR with
  // the acceptors in the only thread lane that matches the priority
  // of the object.
  if (this->cached_policies_.priority_model () ==
      TAO_POA_Cached_Policies::SERVER_DECLARED)
    {
      TAO_Thread_Lane **lanes =
        this->thread_pool_->lanes ();

      for (CORBA::ULong i = 0;
           i != this->thread_pool_->number_of_lanes ();
           ++i)
        {
          if (lanes[i]->lane_priority () == priority)
            return this->TAO_POA::create_stub_object (object_key,
                                                      type_id,
                                                      client_exposed_policies._retn (),
                                                      &filter,
                                                      lanes[i]->resources ().acceptor_registry ()
                                                      ACE_ENV_ARG_PARAMETER);
        }

      ACE_ASSERT (0);
    }

  // If this POA has the CLIENT_PROPAGATED policy, create the IOR with
  // the acceptors in the thread lanes that matches the bands in this
  // POA.  If there are no bands, all the thread lanes are used.
  CORBA::Policy_var bands =
    this->policies ().get_cached_policy (TAO_CACHED_POLICY_RT_PRIORITY_BANDED_CONNECTION);

  RTCORBA::PriorityBandedConnectionPolicy_var priority_bands
    = RTCORBA::PriorityBandedConnectionPolicy::_narrow (bands.in ()
                                                        ACE_ENV_ARG_PARAMETER);