qos_udp.cpp

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

// QoS_UDP.cpp,v 1.18 2003/06/24 20:24:35 yamuna Exp


#include "QoS_UDP.h"

#if defined (ACE_HAS_RAPI) || defined (ACE_HAS_WINSOCK2_GQOS)

#include "UDP.h"
#include "orbsvcs/AV/AVStreams_i.h"
#include "orbsvcs/AV/MCast.h"
#include "orbsvcs/AV/Fill_ACE_QoS.h"

#if !defined (__ACE_INLINE__)
#include "orbsvcs/AV/QoS_UDP.i"
#endif /* __ACE_INLINE__ */

//------------------------------------------------------------
// TAO_AV_UDP_Flow_Handler
//------------------------------------------------------------

static int resv_error = 0;
static int resv_confirm = 0;

int
FillQoSParams (ACE_QoS_Params &qos_params,
               iovec* iov,
               ACE_QoS* qos)
{
  qos_params.callee_data (iov);
  qos_params.caller_data (0);
  qos_params.socket_qos (qos);
  qos_params.group_socket_qos (0);
  qos_params.flags (ACE_JL_BOTH);

  return 0;
}

TAO_AV_UDP_QoS_Session_Helper::TAO_AV_UDP_QoS_Session_Helper (void)
{

}

TAO_AV_UDP_QoS_Session_Helper::~TAO_AV_UDP_QoS_Session_Helper (void)
{
}

int
TAO_AV_UDP_QoS_Session_Helper::set_qos (ACE_Flow_Spec &ace_flow_spec,
                                        TAO_AV_UDP_QoS_Flow_Handler *handler)
{
  ACE_QoS* ace_qos = 0;

  ACE_NEW_RETURN (ace_qos,
                  ACE_QoS,
                  -1);

  Fill_ACE_QoS fill_ace_qos;

  if (handler->flowspec_entry ()->role () == TAO_FlowSpec_Entry::TAO_AV_PRODUCER)
    {
      if (fill_ace_qos.fill_simplex_sender_qos (*ace_qos,
                                                &ace_flow_spec) !=0)
        ACE_ERROR_RETURN ((LM_ERROR,
                           "Unable to fill simplex sender qos (%N|%l)\n"),
                          -1);
      else
        if (TAO_debug_level > 0)
          ACE_DEBUG ((LM_DEBUG,
                      "Filled up the Sender QoS parameters\n"));
    }
  else if (handler->flowspec_entry ()->role () == TAO_FlowSpec_Entry::TAO_AV_CONSUMER)
    {
      if (fill_ace_qos.fill_simplex_receiver_qos (*ace_qos,
                                                  &ace_flow_spec) !=0)
        ACE_ERROR_RETURN ((LM_ERROR,
                           "Unable to fill simplex receiver qos (%N|%l)\n"),
                          -1);
      else
        if (TAO_debug_level > 0)
          ACE_DEBUG ((LM_DEBUG,
                      "Filled up the Receiver QoS parameters\n"));

    }

  ACE_QoS_Manager qos_manager = handler->get_socket ()->qos_manager ();

  // Set the QoS for the session. Replaces the ioctl () call that
  // was being made previously.
  if (handler->qos_session ()->qos (handler->get_socket (),
                                    &qos_manager,
                                    *ace_qos) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "Unable to set QoS (%N|%l)\n"),
                      -1);
  else
    ACE_DEBUG ((LM_DEBUG,
                "Setting QOS succeeds\n"));

  return 0;
}

ACE_QoS_Session *
TAO_AV_UDP_QoS_Session_Helper::open_qos_session (TAO_AV_UDP_QoS_Flow_Handler *handler,
                                                 ACE_INET_Addr &addr)
{
  ACE_QoS_Params qos_params;

  ACE_QoS* ace_qos = 0;

  FillQoSParams (qos_params,
                 0,
                 ace_qos);


  // Create a QoS Session Factory.
  ACE_QoS_Session_Factory session_factory;

  // Ask the factory to create a QoS session.
  ACE_QoS_Session *qos_session = session_factory.create_session ();

  // Create a destination address for the QoS session. The same
  // address should be used for the subscribe call later. A copy
  // is made below only to distinguish the two usages of the dest
  // address.
  ACE_INET_Addr dest_addr (addr);

  // A QoS session is defined by the 3-tuple [DestAddr, DestPort,
  // Protocol]. Initialize the QoS session.
  if (qos_session->open (dest_addr,
                         IPPROTO_UDP) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "Error in opening the QoS session\n"),
                      0);
  else
    ACE_DEBUG ((LM_DEBUG,
                "QoS session opened successfully\n"));

  if (handler->flowspec_entry ()->role () == TAO_FlowSpec_Entry::TAO_AV_PRODUCER)
    {
      // This is a sender
      qos_session->flags (ACE_QoS_Session::ACE_QOS_SENDER);
    }
  else if (handler->flowspec_entry ()->role () == TAO_FlowSpec_Entry::TAO_AV_CONSUMER)
    {
      // This is a receiver
      qos_session->flags (ACE_QoS_Session::ACE_QOS_RECEIVER);
    }

  return qos_session;
}

int
TAO_AV_UDP_QoS_Session_Helper::activate_qos_handler (ACE_QoS_Session *qos_session,
                                                     TAO_AV_UDP_QoS_Flow_Handler *handler)
{
  ACE_QoS_Decorator* qos_decorator;

  // Decorate the above handler with QoS functionality.
  ACE_NEW_RETURN (qos_decorator,
                  ACE_QoS_Decorator (handler,
                                     qos_session,
                                     handler->av_core ()->reactor ()),
                  -1);

  // Initialize the Decorator.
  if (qos_decorator->init () != 0)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "QoS Decorator init () failed (%N|%l)\n"),
                      -1);

  // Register the decorated Event Handler with the Reactor.
  int result = handler->av_core ()->reactor ()->register_handler (qos_decorator,
                                                                  ACE_Event_Handler::QOS_MASK |
                                                                  ACE_Event_Handler::READ_MASK);
  if (result == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "Error in registering the Decorator with the Reactor (%N|%l)\n"),
                      -1);

  return 0;

}

TAO_AV_UDP_QoS_Flow_Handler::TAO_AV_UDP_QoS_Flow_Handler (void)
{
  ACE_NEW (this->transport_,
           TAO_AV_UDP_QoS_Transport (this));
}

TAO_AV_UDP_QoS_Flow_Handler::~TAO_AV_UDP_QoS_Flow_Handler (void)
{
  delete this->transport_;
}

TAO_AV_Transport *
TAO_AV_UDP_QoS_Flow_Handler::transport (void)
{
  return this->transport_;
}

int
TAO_AV_UDP_QoS_Flow_Handler::handle_input (ACE_HANDLE /*fd*/)
{
  this->protocol_object_->handle_input ();
  return 0;
}

int
TAO_AV_UDP_QoS_Flow_Handler::translate (CosPropertyService::Properties &qos_params,
                                        ACE_Flow_Spec *ace_flow_spec)
{
  for (unsigned int i = 0;
       i < qos_params.length ();
       i++)
    {
      if (ACE_OS::strcmp (qos_params [i].property_name, "Service_Type") == 0)
        {
          CORBA::Short type;
          qos_params [i].property_value >>= type;
          ace_flow_spec->service_type (type);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "Token_Rate") == 0)
        {
          CORBA::ULong tok_rate;
          qos_params [i].property_value >>= tok_rate;
          ace_flow_spec->token_rate (tok_rate);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "Token_Bucket_Size") == 0)
        {
          CORBA::ULong tok_buck_size;
          qos_params [i].property_value >>= tok_buck_size;
          ace_flow_spec->token_bucket_size (tok_buck_size);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "Peak_Bandwidth") == 0)
        {
          CORBA::ULong peak_bw;
          qos_params [i].property_value >>= peak_bw;
          ace_flow_spec->peak_bandwidth (peak_bw);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "Latency") == 0)
        {
          CORBA::ULong lat;
          qos_params [i].property_value >>= lat;
          ace_flow_spec->latency (lat);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "Delay_Variation") == 0)
        {
          CORBA::ULong delay_var;
          qos_params [i].property_value >>= delay_var;
          ace_flow_spec->delay_variation (delay_var);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "Max_SDU_Size") == 0)
        {
          CORBA::ULong max_sdu;
          qos_params [i].property_value >>= max_sdu;
          ace_flow_spec->max_sdu_size (max_sdu);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "Minimum_Policed_Size") == 0)
        {
          CORBA::ULong min_pol_size;
          qos_params [i].property_value >>= min_pol_size;
          ace_flow_spec->minimum_policed_size (min_pol_size);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "TTL") == 0)
        {
          CORBA::ULong ttl;
          qos_params [i].property_value >>= ttl;
          ace_flow_spec->ttl (ttl);
        }
      else if (ACE_OS::strcmp (qos_params [i].property_name, "Priority") == 0)
        {
          CORBA::ULong priority;
          qos_params [i].property_value >>= priority;
          ace_flow_spec->priority (priority);
        }
    }

  return 0;
}

int
TAO_AV_UDP_QoS_Flow_Handler::translate (ACE_Flow_Spec *ace_flow_spec,
                                        CosPropertyService::Properties &qos_params)
{
  qos_params.length (9);

  qos_params [0].property_name = CORBA::string_dup ("Service_Type");
  qos_params [0].property_value <<= (CORBA::Short) ace_flow_spec->service_type ();

  qos_params [1].property_name = CORBA::string_dup ("Token_Rate");
  qos_params [1].property_value <<= (CORBA::ULong) ace_flow_spec->token_rate ();

  qos_params [2].property_name = CORBA::string_dup ("Token_Bucket_Size");
  qos_params [2].property_value <<= (CORBA::ULong) ace_flow_spec->token_bucket_size ();

  qos_params [3].property_name = CORBA::string_dup ("Peak_Bandwidth");
  qos_params [3].property_value <<= (CORBA::ULong) ace_flow_spec->peak_bandwidth ();

  qos_params [4].property_name = CORBA::string_dup ("Latency");
  qos_params [4].property_value <<= (CORBA::ULong) ace_flow_spec->latency ();

  qos_params [5].property_name = CORBA::string_dup ("Delay_Variation");
  qos_params [5].property_value <<= (CORBA::ULong) ace_flow_spec->delay_variation ();

  qos_params [6].property_name = CORBA::string_dup ("Max_SDU_Size");
  qos_params [6].property_value <<= (CORBA::ULong) ace_flow_spec->max_sdu_size ();

  qos_params [7].property_name = CORBA::string_dup ("Minimum_Policed_Size");
  qos_params [7].property_value <<= (CORBA::ULong) ace_flow_spec->minimum_policed_size ();

  qos_params [8].property_name = CORBA::string_dup ("TTL");
  qos_params [8].property_value <<= (CORBA::ULong) ace_flow_spec->ttl ();

  return 0;
}

int
TAO_AV_UDP_QoS_Flow_Handler::handle_qos (ACE_HANDLE /*fd*/)
{

  ACE_DECLARE_NEW_CORBA_ENV;
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "(%N,%l) TAO_AV_UDP_QoS_Flow_Handler::handle_qos\n"));

  if (this->qos_session_->update_qos () == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "Error in updating QoS\n"),
                      -1);
  else
  {
    if(TAO_debug_level > 0)
      ACE_DEBUG ((LM_DEBUG,
                  "(%N,%l) Updating QOS succeeds.\n"));
  }

  if (this->qos_session_->rsvp_event_type () == ACE_QoS_Session::RSVP_RESV_ERROR)
    {
      resv_error = 1;
    }

  if (this->qos_session_->rsvp_event_type () == ACE_QoS_Session::RSVP_RESV_CONFIRM)
    {
      resv_confirm = 1;
    }

  if (this->qos_session_->flags () == ACE_QoS_Session::ACE_QOS_SENDER)
    {
      if (this->qos_session_->rsvp_event_type () == ACE_QoS_Session::RSVP_RESV_EVENT)
            {
              if( TAO_debug_level > 0 )
              {
                 ACE_DEBUG ((LM_DEBUG,
                             "(%N,%l) Resv Event Received\n"));
              }
              if (!CORBA::is_nil (this->negotiator_))
                {
                  if( TAO_debug_level > 0 )
                  {
                     ACE_DEBUG ((LM_DEBUG,
                                 "(%N,%l) Negotiator Specified\n"));
                  }

                  AVStreams::streamQoS new_qos;
                  ACE_Flow_Spec *ace_flow_spec =
                    this->qos_session_->qos ().sending_flowspec ();

                  if (ace_flow_spec != 0)
                    {
                      new_qos.length (1);
                      this->translate (ace_flow_spec,
                                       new_qos [0].QoSParams);
                    }

                  AVStreams::Negotiator_var remote_negotiator;
                  this->negotiator_->negotiate (remote_negotiator.in (),
                                                new_qos
                                                ACE_ENV_ARG_PARAMETER);
                }
            }
    }
  else if (this->qos_session_->flags () == ACE_QoS_Session::ACE_QOS_RECEIVER)
    {
      if (this->qos_session_->rsvp_event_type () == ACE_QoS_Session::RSVP_PATH_EVENT)