rtcosscheduling_serverscheduler_i.cpp

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

        RTPortableServer::POA::_narrow(poa.in()
                                       ACE_ENV_ARG_PARAMETER);

      /// return the reference to the RT POA
      return rt_poa.in();
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION(ACE_ANY_EXCEPTION,
                "ERROR: Could not create a Scheduling Service POA\n");
      return parent;
    }
  ACE_ENDTRY;
  return parent;
}

void
RTCosScheduling_ServerScheduler_i::schedule_object (
    CORBA::Object_ptr obj,
    const char * name
    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException
    , RTCosScheduling::UnknownName))
{
  /// Check and see if the object name is in the resource map,
  /// if it is, then there is a ceiling for it and we can add it to the
  /// object name map.  If there is no name in the resource map, then there
  /// is no valid ceiling for it, throw an UnknownName exception
  if (this->resource_map_.find(name) == -1)
    {
      ACE_THROW (RTCosScheduling::UnknownName());
    }
  else
    {
     this->object_name_map_.rebind(obj, name);
     ACE_CHECK;
    }
}



#if TAO_HAS_INTERCEPTORS
RTCosScheduling_ServerScheduler_Interceptor::RTCosScheduling_ServerScheduler_Interceptor(
  CORBA::ORB_ptr orb,
  char *shared_file,
  CosSchedulingObjectMap *ObjectMap,
  CosSchedulingResourceMap *resourceMap)
: name_("RTCosScheduling_ServerScheduler_Interceptor"),
  orb_(orb),
  object_name_map_(ObjectMap),
  resource_map_(resourceMap)
{
  ACE_TRY_NEW_ENV
    {
      // Create a new Priority Ceiling protocol manager factory
      ACE_NEW_THROW_EX(this->PCP_factory_,
                       PCP_Manager_Factory(shared_file),
                       CORBA::NO_MEMORY());
      ACE_CHECK;


      /// Now resolve a reference to the Real Time ORB
      CORBA::Object_var obj =
        this->orb_->resolve_initial_references("RTORB"
                                       ACE_ENV_ARG_PARAMETER);
      RTCORBA::RTORB_var rt_orb;
      ACE_TRY_CHECK;
      if (CORBA::is_nil(obj.in ()))
        {
          ACE_DEBUG((LM_DEBUG,
                     "RTORB not available, "
                     "RT ServerScheduler not used!\n"));
          return;
        }
      else
        {
          rt_orb =RTCORBA::RTORB::_narrow (obj.in ()
                                           ACE_ENV_ARG_PARAMETER);
        }

      // Now get a reference to the RTCurrent
      // for the PCP manager to control
      obj = this->orb_->resolve_initial_references ("RTCurrent"
                                                    ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      if (CORBA::is_nil(obj.in()))
        {
          ACE_DEBUG((LM_DEBUG,
                     "RTCurrent not available, "
                     "RT ServerScheduler not used!\n"));
          return;
        }
      else
        {
          this->current_ =
            RTCORBA::Current::_narrow (obj.in ()
                                       ACE_ENV_ARG_PARAMETER);
        }

      // Now get a reference to the codec factory to create a codec that will
      // decode the client priority sent in the service context
      obj =
        this->orb_->resolve_initial_references ("CodecFactory"
                                                ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      if (CORBA::is_nil(obj.in()))
        {
          ACE_DEBUG((LM_DEBUG,
                     "Could not initalize the server interceptor Codec, "
                     "RT ServerScheduler not used!\n"));
          return;
        }
      else
        {
          this->codec_factory_ = IOP::CodecFactory::_narrow(obj.in());
        }
      ACE_TRY_CHECK;


      // Set up the codec
      IOP::Encoding encoding;
      encoding.format = IOP::ENCODING_CDR_ENCAPS;
      encoding.major_version = 1;
      encoding.minor_version = 2;

      this->codec_ = this->codec_factory_->create_codec(encoding);
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION(ACE_ANY_EXCEPTION,
                          "Error in installing the interceptor for the ServerScheduler\n");
    }
  ACE_ENDTRY;
}


// Delete the objects we have made in the interceptor
RTCosScheduling_ServerScheduler_Interceptor::~RTCosScheduling_ServerScheduler_Interceptor()
{
  this->object_name_map_ = 0;
  this->resource_map_ = 0;
  delete this->PCP_factory_;
}

void
RTCosScheduling_ServerScheduler_Interceptor::receive_request(
  PortableInterceptor::ServerRequestInfo_ptr ri
  ACE_ENV_ARG_DECL_NOT_USED)
ACE_THROW_SPEC((
  CORBA::SystemException,
  PortableInterceptor::ForwardRequest))
{
  ACE_TRY_NEW_ENV
    {

      ACE_CString name = "";
      COS_SCHEDULING_RESOURCE_VALUE ceiling, base_priority = 0;

      // Now get a reference to the POA, this is used to get a reference
      // to the target object
      PortableInterceptor::AdapterName *adapter_seq = ri->adapter_name();
      ACE_TRY_CHECK;
      PortableServer::POA_var poa;
      const char *adapter_name =
        (*adapter_seq)[adapter_seq->length() - 1];

      CORBA::Object_var obj =
        this->orb_->resolve_initial_references("RootPOA"
                                               ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      if (CORBA::is_nil(obj.in()))
        {
          ACE_DEBUG((LM_DEBUG,
                     "Could not get root POA, "
                     "RT scheduling not used on server!\n"));
          return;
        }
      else
        {
          PortableServer::POA_var root_poa =
            PortableServer::POA::_narrow(obj.in());
          if (CORBA::is_nil(root_poa.in()))
            {
              ACE_ERROR((LM_ERROR,
                         "No POA found in Interceptor prio not set\n"));
              return;
            }
          poa = root_poa->find_POA(adapter_name, 0);
          ACE_TRY_CHECK;
        }

      // decode the Client priority sent in the service context
      IOP::ServiceId id = IOP::RTCorbaPriority;
      IOP::ServiceContext_var sc;
      ACE_TRY_EX(svc_req)
        {
          sc = ri->get_request_service_context(id);
          ACE_TRY_CHECK_EX(svc_req);
        }
      ACE_CATCHANY
        {
          // The RTCorbaPriority was not sent, do not use real time
          // Here we cannot let the server continue to run at
          // RTCORBA::maxPriority, so change to RTCORBA::minPriority
          // and things will run best effort
          this->current_->the_priority(RTCORBA::minPriority);
          return;
        }
      ACE_ENDTRY;

      CORBA::OctetSeq ocSeq = CORBA::OctetSeq(
        sc->context_data.length(),
        sc->context_data.length(),
        sc->context_data.get_buffer(),
        0);
      ACE_TRY_CHECK;

      CORBA::Any the_priority_as_any;
      the_priority_as_any = *this->codec_->decode(ocSeq);
      ACE_TRY_CHECK;

      RTCORBA::Priority the_client_priority;
      the_priority_as_any >>= the_client_priority;
      ACE_TRY_CHECK;

      // get the object from the object ID that is passed
      CORBA::OctetSeq_var oseq = ri->object_id();
      PortableServer::ObjectId oid(oseq -> length(),
                                   oseq -> length(),
                                   oseq -> get_buffer(),
                                    0);
      ACE_TRY_CHECK;

      CORBA::Object_var target_object = poa->id_to_reference(oid);
      ACE_TRY_CHECK;

      // Check to make sure we have the object as scheduled by the
      // ServerScheduler
      if (this->object_name_map_->find(target_object, name) == -1 )
        {
          ACE_DEBUG((LM_DEBUG,
                     "Object not found in config file, "
                     "RT ServerScheduler not used!\n"));
          // Here we cannot let the server continue to run at
          // RTCORBA:: maxPriority, so it will run at minPriority
          // and things will run best effort
          this->current_->the_priority(RTCORBA::minPriority);
          return;
        }

      /// If the object name if found, get its priority ceiling
      if (this->resource_map_->find(name, ceiling) == -1 )
        {
          /// We could not find the Object's priority ceiling
          /// given its name
          ACE_DEBUG((LM_DEBUG,
                     "Object Resource Ceiling not found in config file, "
                     "RT ServerScheduler not used!\n"));
          // Here we cannot let the server continue to run at
          // RTCORBA:: maxPriority, so it will run at minPriority
          // and things will run best effort
          this->current_->the_priority(RTCORBA::minPriority);
          return;
        }

      /// Get the base priority of the server
      if (this->resource_map_->find("BP", base_priority) == -1 )
        {