rt_policy_validator.cpp

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

      for (CORBA::ULong band = 0;
           band < bands.length ();
           ++band)
        {
          int match = 0;
          for (CORBA::ULong lane = 0;
               lane != this->thread_pool_->number_of_lanes () && !match;
               ++lane)
            {
              CORBA::Short lane_priority =
                lanes[lane]->lane_priority ();

              if (lane_priority <= bands[band].high &&
                  lane_priority >= bands[band].low)
                match = 1;
            }
          if (!match)
            ACE_THROW (PortableServer::POA::InvalidPolicy ());
        }

      // Done with checks.
      return;
    }

  // If priority banded connections are not set, and the priority
  // model is SERVER_DECLARED, make sure we have at least one thread
  // lane that can provide service for the specified SERVER_DECLARED
  // priority.
  if (rt_priority_model == TAO_POA_Cached_Policies::SERVER_DECLARED)
    {
      // If this POA is using the default thread pool (which doesn't
      // have lanes) or a thread pool without lanes, we are done with
      // the checks.
      if (this->thread_pool_ == 0 ||
          !this->thread_pool_->with_lanes ())
        return;

      // If this POA is using a thread pool with lanes, make sure we
      // have at least one thread lane that can provide service for
      // the specified SERVER_DECLARED priority.
      TAO_Thread_Lane **lanes =
        this->thread_pool_->lanes ();

      int match = 0;
      for (CORBA::ULong lane = 0;
           lane != this->thread_pool_->number_of_lanes () && !match;
           ++lane)
        {
          CORBA::Short lane_priority =
            lanes[lane]->lane_priority ();

          if (lane_priority <= priority &&
              lane_priority >= priority)
            match = 1;
        }
      if (!match)
        ACE_THROW (PortableServer::POA::InvalidPolicy ());

      // Done with checks.
      return;
    }

}

void
TAO_POA_RT_Policy_Validator::validate_thread_pool (TAO_Policy_Set &policies
                                                   ACE_ENV_ARG_DECL)
{
  this->thread_pool_ =
    TAO_POA_RT_Policy_Validator::extract_thread_pool (this->orb_core_,
                                                      policies
                                                      ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
TAO_POA_RT_Policy_Validator::validate_lifespan (TAO_Policy_Set &policies
                                                ACE_ENV_ARG_DECL)
{
  // If this POA is using a RTCORBA thread pool, make sure the
  // lifespan policy is not persistent since we cannot support it
  // right now.
  if (this->thread_pool_ != 0)
    {
      CORBA::Policy_var policy =
        policies.get_cached_policy (TAO_CACHED_POLICY_LIFESPAN);
      PortableServer::LifespanPolicy_var lifespan_policy =
        PortableServer::LifespanPolicy::_narrow (policy.in ()
                                                 ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      PortableServer::LifespanPolicyValue lifespan =
        lifespan_policy->value (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;

      if (lifespan == PortableServer::PERSISTENT)
        {
          ACE_THROW (PortableServer::POA::InvalidPolicy ());
        }
    }
}

void
TAO_POA_RT_Policy_Validator::merge_policies_impl (TAO_Policy_Set &policies
                                                  ACE_ENV_ARG_DECL)
{
  // Check if the user has specified the priority model policy.
  CORBA::Policy_var priority_model =
    policies.get_cached_policy (TAO_CACHED_POLICY_PRIORITY_MODEL);

  if (CORBA::is_nil (priority_model.in ()))
    {
      // If not, check if the priority model policy has been specified
      // at the ORB level.
      priority_model =
        this->orb_core_.get_cached_policy (TAO_CACHED_POLICY_PRIORITY_MODEL);
      if (!CORBA::is_nil (priority_model.in ()))
        {
          // If so, we'll use that policy.
          policies.set_policy (priority_model.in () ACE_ENV_ARG_PARAMETER);
          ACE_CHECK;
        }
    }

  // Check if the user has specified the server protocol policy.
  CORBA::Policy_var server_protocol =
    policies.get_cached_policy (TAO_CACHED_POLICY_RT_SERVER_PROTOCOL);

  if (CORBA::is_nil (server_protocol.in ()))
    {
      // If not, check if the server protocol policy has been
      // specified at the ORB level.
      server_protocol =
        this->orb_core_.get_cached_policy (TAO_CACHED_POLICY_RT_SERVER_PROTOCOL);
      if (!CORBA::is_nil (server_protocol.in ()))
        {
          // If so, we'll use that policy.
          policies.set_policy (server_protocol.in () ACE_ENV_ARG_PARAMETER);
          ACE_CHECK;
        }
    }

  // Check if the user has specified the thread pool policy.
  CORBA::Policy_var thread_pool =
    policies.get_cached_policy (TAO_CACHED_POLICY_THREADPOOL);

  if (CORBA::is_nil (thread_pool.in ()))
    {
      // If not, check if the thread pool policy has been specified at
      // the ORB level.
      thread_pool =
        this->orb_core_.get_cached_policy (TAO_CACHED_POLICY_THREADPOOL);
      if (!CORBA::is_nil (thread_pool.in ()))
        {
          // If so, we'll use that policy.
          policies.set_policy (thread_pool.in () ACE_ENV_ARG_PARAMETER);
          ACE_CHECK;
        }
    }
}

/* static */
TAO_Thread_Pool *
TAO_POA_RT_Policy_Validator::extract_thread_pool (TAO_ORB_Core &orb_core,
                                                  TAO_Policy_Set &policies
                                                  ACE_ENV_ARG_DECL)
{
  CORBA::Policy_var policy =
    policies.get_cached_policy (TAO_CACHED_POLICY_THREADPOOL);

  RTCORBA::ThreadpoolPolicy_var thread_pool_policy =
    RTCORBA::ThreadpoolPolicy::_narrow (policy.in ()
                                        ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (0);

  if (CORBA::is_nil (thread_pool_policy.in ()))
    return 0;

  RTCORBA::ThreadpoolId thread_pool_id =
    thread_pool_policy->threadpool (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK_RETURN (0);

  // Get the RTORB.
  CORBA::Object_var object =
    orb_core.resolve_rt_orb (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK_RETURN (0);

  RTCORBA::RTORB_var rt_orb =
    RTCORBA::RTORB::_narrow (object.in ()
                             ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (0);

  TAO_RT_ORB *tao_rt_orb =
    ACE_dynamic_cast (TAO_RT_ORB *,
                      rt_orb.in ());

  TAO_Thread_Pool_Manager &tp_manager =
    tao_rt_orb->tp_manager ();

  TAO_Thread_Pool_Manager::THREAD_POOLS &thread_pools =
    tp_manager.thread_pools ();

  // Check that the thread pool id is valid.
  TAO_Thread_Pool *thread_pool = 0;
  int result =
    thread_pools.find (thread_pool_id,
                       thread_pool);

  if (result != 0)
    ACE_THROW_RETURN (PortableServer::POA::InvalidPolicy (),
                      0);

  return thread_pool;
}

/* static */
TAO_Acceptor_Registry *
TAO_POA_RT_Policy_Validator::extract_acceptor_registry (TAO_ORB_Core &orb_core,
                                                        TAO_Thread_Pool *thread_pool)
{
  TAO_Acceptor_Registry *acceptor_registry = 0;

  // If <thread_pool_> != 0, it means that we have a RT thread pool.
  if (thread_pool)
    {
      TAO_Thread_Lane **lanes =
        thread_pool->lanes ();

      // All the lanes have similar acceptor registries.  Therefore,
      // looking at the first lane should suffice.
      TAO_Thread_Lane_Resources &resources =
        lanes[0]->resources ();

      acceptor_registry =
        &resources.acceptor_registry ();
    }
  else
    // We are dealing with the default thread pool.
    {
      TAO_Thread_Lane_Resources_Manager &thread_lane_resources_manager =
        orb_core.thread_lane_resources_manager ();

      TAO_Thread_Lane_Resources &resources =
        thread_lane_resources_manager.default_lane_resources ();

      acceptor_registry =
        &resources.acceptor_registry ();
    }

  return acceptor_registry;
}

#endif /* TAO_HAS_CORBA_MESSAGING) && TAO_HAS_CORBA_MESSAGING != 0 */