rtcosscheduling_serverscheduler_i.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

/* -*- C++ -*- */

//=============================================================================
/**
 *  @file    RTCosScheduling_ServerScheduler_i.cpp
 *
 *  RTCosScheduling_ServerScheduler_i.cpp,v 1.6 2003/11/01 20:54:57 bala Exp
 *
 *  @author Matt Murphy <[e-mail]>
 */
//=============================================================================

#include "RTCosScheduling_ServerScheduler_i.h"
#include "ace/OS_NS_errno.h"

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



#include "tao/ORB_Core.h"
#include "tao/PortableServer/POA.h"
#include "tao/RTCORBA/Linear_Priority_Mapping.h"
#include "tao/RTCORBA/Priority_Mapping_Manager.h"
#include "tao/RTPortableServer/RTPortableServer.h"


namespace TAO {

RTCosScheduling_ServerScheduler_i::
  RTCosScheduling_ServerScheduler_i (char *node,
    char *file,
    char *shared_file,
    int numthreads
  )
: num_threads_(numthreads),
  shared_file_(shared_file)
{
  /// Read the Resources and ceilings from the config file
  /// and put them into the resource_map_
  if ( !ACE_OS::strcmp(file,"") || file == NULL || !TAO_HAS_INTERCEPTORS)
    {
      ACE_DEBUG((LM_DEBUG,
                 "Server will not schedule object execution "
                 "(ServerScheduler interceptor not installed)"));
    }
  else if (!resources(node, file, &this->resource_map_))
    {
      ACE_OS::exit(1);
    }
}


// Implementation skeleton destructor
RTCosScheduling_ServerScheduler_i::~RTCosScheduling_ServerScheduler_i (void)
{
  delete this->pm_;
}


int
RTCosScheduling_ServerScheduler_i::configure_ORB(TAO_ORB_Core *orb_core)
{
  ACE_TRY_NEW_ENV
    {

     // Get an object reference to orb from the orb core
     CORBA::ORB_ptr orb = orb_core->orb();

#if TAO_HAS_INTERCEPTORS
      /// Construct the interceptor that the Scheduling service uses
      RTCosScheduling_ServerScheduler_Interceptor *server_interceptor = 0;;
      ACE_NEW_THROW_EX(server_interceptor,
        RTCosScheduling_ServerScheduler_Interceptor(orb,
          this->shared_file_,
          &this->object_name_map_,
          &this->resource_map_),
          CORBA::NO_MEMORY());
      ACE_TRY_CHECK;

      /// First, get a list of all interceptors currently registered
      TAO_ServerRequestInterceptor_List::TYPE &interceptors =
        orb_core->server_request_interceptors();


      /// Now check to see if the ServerScheduler Interceptor has already
      /// been registered
      u_int i;
      u_int unregistered = 1;
      for (i = 0; i < interceptors.size() && unregistered; ++i)
         {
          if (ACE_OS::strncmp(
                interceptors[i]->_interface_repository_id (),
                server_interceptor->_interface_repository_id(),
                ACE_OS::strlen(
                  server_interceptor->_interface_repository_id())-2)
            == 0)
            {
              /// The interceptor is already registered,
              /// don't try to register it again
              unregistered = 0;
            }
        }

      if (unregistered)
        {
          orb_core->add_interceptor(server_interceptor
                                    ACE_ENV_ARG_PARAMETER);
        }
#endif /*  TAO_HAS_INTERCEPTORS */

      /// Resolve a reference to the Linear Priority Mapping Manager
     CORBA::Object_var rt_obj =
        orb->resolve_initial_references("PriorityMappingManager"
                                        ACE_ENV_ARG_PARAMETER);
      if (CORBA::is_nil(rt_obj.in()))
        {
          ACE_DEBUG((LM_DEBUG,
                     "Priority Mapping Manager not available, "
                    "RT not used on server!\n"));
          return 0;
        }
      RTCORBA::PriorityMappingManager_var mapping_manager =
        RTCORBA::PriorityMappingManager::_narrow(rt_obj.in()
                                                 ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      /// Create the Linear Priority Mapping Manager
      ACE_NEW_THROW_EX(this->pm_,
                       TAO_Linear_Priority_Mapping(
                         ACE_SCHED_FIFO),
                      CORBA::NO_MEMORY());

      mapping_manager->mapping(this->pm_);
      ACE_TRY_CHECK;

   }
 ACE_CATCHANY
   {
        ACE_ERROR((LM_ERROR, "Could not configure the orb"));
        ACE_OS::exit(1);
    }
 ACE_ENDTRY;
 return 1;
}


::PortableServer::POA_ptr RTCosScheduling_ServerScheduler_i::create_POA (
    PortableServer::POA_ptr parent,
    const char * adapter_name,
    PortableServer::POAManager_ptr a_POAManager,
    const CORBA::PolicyList & policies
    ACE_ENV_ARG_DECL_NOT_USED)
  ACE_THROW_SPEC (( CORBA::SystemException
    , PortableServer::POA::AdapterAlreadyExists
    , PortableServer::POA::InvalidPolicy))
{
  ACE_TRY_NEW_ENV
    {
        // We should hopefully be using more than one thread
#if defined (ACE_HAS_THREADS)
      u_int has_threads = 2;
#else
      u_int has_threads = 1;
#endif /* ACE_HAS_THREADS */

      /// Get the ORB core from the POA Manager
      TAO_POA *tao_poa = parent->_tao_poa_downcast();
      TAO_ORB_Core &orb_core = tao_poa->orb_core();

      /// configure the orb (linear mapping, register interceptor, etc.)
      configure_ORB(&orb_core);

      /// Get an object reference to orb from the orb core
      CORBA::ORB_ptr orb = orb_core.orb();

      /// Now resolve a reference to the Real Time ORB
      CORBA::Object_var rt_obj =
        orb->resolve_initial_references("RTORB"
                                        ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      if (CORBA::is_nil(rt_obj.in()))
        {
          ACE_DEBUG((LM_DEBUG,
                     "RTORB not available, "
                     "not using RT on the server!\n"));
          return 0;
        }

      /// Get the reference to the RT ORB
      RTCORBA::RTORB_var rt_orb =
         RTCORBA::RTORB::_narrow (rt_obj.in ()
                                  ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      if (CORBA::is_nil(rt_orb.in ()))
        {
          ACE_DEBUG((LM_DEBUG,
                     "Could not initialize orb for the server interceptor, "
                      "RT will not be used!\n"));
          return 0;
        }


      // Copy the non realtime policy list
      CORBA::PolicyList poa_policy_list = policies;
      poa_policy_list.length (policies.length()+has_threads);

      /// Set the server to run at max priority so it will immediately
      /// process intercepts and place new requests into the approprate queue
      poa_policy_list[policies.length()] =
        rt_orb->create_priority_model_policy (
          RTCORBA::SERVER_DECLARED,
          RTCORBA::maxPriority
          ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      // Set up the threadpool
#if defined (ACE_HAS_THREADS)
      CORBA::ULong stacksize                 = 0;

      /// ideally 1, but not implemented in TAO.  This is fine as long
      /// the system is not overloaded and can accept all requests
      CORBA::Boolean allow_request_buffering = 0;

      // ideally max_concurrent - num_threads_, but no request buffer in TAO
      CORBA::ULong max_buffered_requests     = 0;

      /// TAO does not have, nor plan to implement, request buffering
      CORBA::ULong max_request_buffer_size   = 0;

      /// This comes from the model of the system
      CORBA::ULong static_threads            = this->num_threads_;

      CORBA::ULong dynamic_threads           = 0;

      /// Set it to max so there will be no priority inversions
      /// while the request is accepted
      CORBA::ULong default_thread_priority   = RTCORBA::maxPriority;

      /// Create the threadpool the server uses to execute requests
      RTCORBA::ThreadpoolId threadpool =
        rt_orb->create_threadpool (stacksize,
                                          static_threads,
                                          dynamic_threads,
                                          default_thread_priority,
                                          allow_request_buffering,
                                          max_buffered_requests,
                                          max_request_buffer_size
                                          ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      poa_policy_list[policies.length()+1] =
        rt_orb->create_threadpool_policy (threadpool
                                          ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
#endif /* ACE_HAS_THREADS */


      /// Create the RT POA
      PortableServer::POA_var poa =
        parent->create_POA (adapter_name,
                            a_POAManager,
                            poa_policy_list
                            ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      RTPortableServer::POA_var rt_poa =