rtcp.cpp

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

/*
 * Copyright (c) 1994-1995 Regents of the University of California.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. All advertising materials mentioning features or use of this software
 *    must display the following acknowledgement:
 *      This product includes software developed by the University of
 *      California, Berkeley and the Network Research Group at
 *      Lawrence Berkeley Laboratory.
 * 4. Neither the name of the University nor of the Laboratory may be used
 *    to endorse or promote products derived from this software without
 *    specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */

// RTCP.cpp,v 5.26 2003/11/04 05:21:31 dhinton Exp
#include "ntp-time.h"
#include "RTCP.h"
#include "media_timer.h"
#include "tao/debug.h"
#include "global.h"
#include "md5.h"

#include "RTCP_Packet.h"
#include "ace/OS_NS_time.h"
#include "ace/OS_NS_strings.h"

int
TAO_AV_RTCP_Callback::receive_control_frame (ACE_Message_Block *data,
                                             const ACE_Addr &peer_address)
{
  int length = ACE_static_cast (int, data->length ());
  int more = length;
  char *buf_ptr = data->rd_ptr ();
  char first_rtcp_packet = 1;
  RTCP_Channel_In *c;

  // This code performs the RTCP Header validity checks detailed in RFC 1889
  // Appendix A.2

  while (more > 0)
    {
      // the second byte of the control packet is the type
      switch ((unsigned char)buf_ptr[length - more + 1])
      {
        case RTCP_PT_SR:
          {
            RTCP_SR_Packet sr(&buf_ptr[length-more],
                              &more);

            if (!sr.is_valid(first_rtcp_packet))
              ACE_DEBUG ((LM_DEBUG,
                          "TAO_AV_RTCP_Callback::receive_control_frame - "
                          "warning invalid rtcp packet\n"));

            if (this->inputs_.find (sr.ssrc (), c) == -1)
              {
                ACE_NEW_RETURN (c,
                                RTCP_Channel_In (sr.ssrc (),
                                                 &peer_address),
                                -1);
                this->inputs_.bind (sr.ssrc (), c);
              }
            c->updateStatistics (&sr);

            if (TAO_debug_level > 0)
              sr.dump ();
            break;
          }
        case RTCP_PT_RR:
          {
            RTCP_RR_Packet rr(&buf_ptr[length-more],
                              &more);

            if (!rr.is_valid(first_rtcp_packet))
              ACE_DEBUG ((LM_DEBUG,
                          "TAO_AV_RTCP_Callback::receive_control_frame - "
                          "warning invalid rtcp packet\n"));

            if (this->inputs_.find (rr.ssrc (), c) == -1)
              {
                ACE_NEW_RETURN (c,
                                RTCP_Channel_In (rr.ssrc (),
                                                 &peer_address),
                                -1);
                this->inputs_.bind (rr.ssrc (), c);
              }

            c->updateStatistics (&rr);

            if (TAO_debug_level > 0)
              rr.dump ();
            break;
          }
        case RTCP_PT_SDES:
          {
            RTCP_SDES_Packet sdes (&buf_ptr[length-more],
                                   &more);

            if (!sdes.is_valid(first_rtcp_packet))
              ACE_DEBUG ((LM_DEBUG,
                          "TAO_AV_RTCP_Callback::receive_control_frame - "
                          "warning invalid rtcp packet\n"));

            if (TAO_debug_level > 0)
              sdes.dump ();
            break;
          }
        case RTCP_PT_BYE:
          {
            RTCP_BYE_Packet bye (&buf_ptr[length-more],
                                 &more);

            if (!bye.is_valid(first_rtcp_packet))
              ACE_DEBUG ((LM_DEBUG,
                          "TAO_AV_RTCP_Callback::receive_control_frame - "
                          "warning invalid rtcp packet\n"));

            // Inform the listener that a source(s) has left the session
            ACE_UINT32 *ssrc_list;
            unsigned char length;

            bye.ssrc_list(&ssrc_list, length);

            for (int i=0; i<length; i++)
              {
                RTCP_Channel_In *c = 0;

                // remove the channel from the list
                this->inputs_.unbind(ssrc_list[i], c);

                if (c != 0)
                  delete c;
              }

            if (TAO_debug_level > 0)
              bye.dump ();

            break;
          }
        case RTCP_PT_APP:
          // If we receive one of these, ignore it.
          ACE_DEBUG ((LM_DEBUG,
                      "TAO_AV_RTCP_Callback::receive_control_frame - "
                      "APP packet - ignore\n"));
          more -= (4 + (ACE_UINT16)buf_ptr[length - more + 2]);
          break;
        default:
          ACE_DEBUG ((LM_DEBUG,
                      "TAO_AV_RTCP_Callback::receive_control_frame - "
                      "UNKNOWN packet type %u; ignore the rest\n",
                      (int)buf_ptr[length - more + 1]));
          more = 0;
      }

      first_rtcp_packet = 0;

    }

    if (more != 0)
      ACE_DEBUG ((LM_DEBUG,
                  "TAO_AV_RTCP_Callback::receive_control_frame - "
                  "Error in overall packet length\n"));
    return 0;
}

ACE_INT32 random32 (int);

ACE_UINT32
TAO_AV_RTCP::alloc_srcid (ACE_UINT32 addr)
{
  md5_string s;

  s.type = addr;
  s.tv = ACE_OS::gettimeofday ();
  s.pid = ACE_OS::getpid();
  s.pgid = ACE_OS::getpgid(s.pid);
  s.ppid = ACE_OS::getppid();
  s.uid = ACE_OS::getuid();
  s.gid = ACE_OS::getgid();

  unsigned char *string_val = (unsigned char *) &s;
  int length = sizeof(s);

  MD5_CTX context;
  union
    {
      char c[16];
      u_long x[4];
    } digest;
  ACE_UINT32 r;
  int i;

  MD5Init (&context);
  MD5Update (&context, string_val, length);
  MD5Final ((unsigned char*)&digest, &context);
  r=0;
  for (i=0; i<3; i++)
    r ^= digest.x[i];

  return r;

/* used to be this
  ACE_Time_Value tv = ACE_OS::gettimeofday ();
  ACE_UINT32 srcid = ACE_UINT32 (tv.sec () + tv.usec ());
  srcid += (ACE_UINT32)ACE_OS::getuid();
  srcid += (ACE_UINT32)ACE_OS::getpid();
  srcid += addr;
  return (srcid);
*/
}


double
TAO_AV_RTCP::rtcp_interval (int members,
                            int senders,
                            double rtcp_bw,
                            int we_sent,
                            int packet_size,
                            int *avg_rtcp_size,
                            int initial)
{
  // Minimum time between RTCP packets from this site (in sec.).
  // This time prevents the reports from 'clumping' when sessions
  // are small and the law of large numbers isn't helping to smooth
  // out the traffic.  It also keeps the report interval from
  // becoming ridiculously small during transient outages like a
  // network partition.
//  double const RTCP_MIN_TIME = 5.0;   (from RTP.h)

  // Fraction of the RTCP bandwidth to be shared among active
  // senders.  (This fraction was chosen so that in a typical
  // session with one or two active senders, the computed report
  // time would be roughly equal to the minimum report time so that
  // we don't unnecessarily slow down receiver reports.) The
  // receiver fraction must be 1 - the sender fraction.
//  double const RTCP_SENDER_BW_FRACTION = 0.25;   (from RTP.h)
//  double const RTCP_RCVR_BW_FRACTION = (1-RTCP_SENDER_BW_FRACTION); (from RTP.h)

  // Gain (smoothing constant) for the low-pass filter that
  // estimates the average RTCP packet size
//  double const RTCP_SIZE_GAIN = (1.0/16.0);   (from RTP.h)

  double t;
  double rtcp_min_time = RTCP_MIN_RPT_TIME;
  int n;   // number of members for computation

  // Very first call at application start-up uses half the min
  // delay for quicker notification while still allowing some time
  // before reporting for randomization and to learn about other
  // sources so the report interval will converge to the correct
  // interval more quickly.  The average RTCP size is initialized
  // to 128 octets which is conservative (it assumes everyone else
  // is generating SRs instead of RRs: 20 IP + 8 UDP + 52 SR + 48
  // SDES CNAME).
  if (initial)
    {
      // initialize the random number generator
      ACE_OS::srand(ACE_OS::time(0L));

      rtcp_min_time /= 2;
      *avg_rtcp_size = 128;
    }

  // If there were active senders, give them at least a minimum
  // share of the RTCP bandwidth.  Otherwise all participants share
  // the RTCP bandwidth equally.
  n = members;
  if ((senders > 0) && (senders < members*RTCP_SENDER_BW_FRACTION))
    {
      if (we_sent)
        {
          rtcp_bw *= RTCP_SENDER_BW_FRACTION;
          n = senders;
        }
      else
        {
          rtcp_bw *= RTCP_RECEIVER_BW_FRACTION;
          n -= senders;
        }
    }

  // Update the average size estimate by the size of the report
  // packet we just sent.
  *avg_rtcp_size += (int)((packet_size - *avg_rtcp_size)*RTCP_SIZE_GAIN);

  // The effective number of sites times the average packet size is
  // the total number of octets sent when each site sends a report.
  // Dividing this by the effective bandwidth gives the time
  // interval over which those packets must be sent in order to
  // meet the bandwidth target, with a minimum enforced.  In that
  // time interval we send one report so this time is also our
  // average time between reports.
  t = (*avg_rtcp_size) * n / rtcp_bw;
  if (t < rtcp_min_time)
    t = rtcp_min_time;

  // To avoid traffic bursts from unintended synchronization with
  // other sites, we then pick our actual next report interval as a
  // random number uniformly distributed between 0.5*t and 1.5*t.

  // TODO: this may not be right.  need a random number between 0 and 1
  int max_rand = 32768;

  return t * ((double)ACE_OS::rand()/max_rand + 0.5);
//  return t * (drand48() + 0.5);
}



// TAO_AV_RTCP_Flow_Factory
TAO_AV_RTCP_Flow_Factory::TAO_AV_RTCP_Flow_Factory (void)
{
}

TAO_AV_RTCP_Flow_Factory::~TAO_AV_RTCP_Flow_Factory (void)
{
}

int
TAO_AV_RTCP_Flow_Factory::match_protocol (const char *flow_string)
{
  if (ACE_OS::strncasecmp (flow_string,"RTCP",4) == 0)
      return 1;

  return 0;
}

int
TAO_AV_RTCP_Flow_Factory::init (int /* argc */,
                                char * /* argv */ [])
{
  return 0;
}

TAO_AV_Protocol_Object*
TAO_AV_RTCP_Flow_Factory::make_protocol_object (TAO_FlowSpec_Entry * /*entry*/,
                                                TAO_Base_StreamEndPoint * /*endpoint*/,
                                                TAO_AV_Flow_Handler *handler,
                                                TAO_AV_Transport *transport)
{
  TAO_AV_Callback *client_cb = 0;
  TAO_AV_RTCP_Callback *rtcp_cb = 0;

  // TODO: need to handle a client callback at some point
//  endpoint->get_control_callback (entry->flowname (),
//                                  client_cb);

  TAO_AV_Protocol_Object *object = 0;
  ACE_NEW_RETURN (object,
                  TAO_AV_RTCP_Object (client_cb,
                                      rtcp_cb,
                                      transport),
                  0);

  rtcp_cb->open (object, handler);

  return object;
}

// TAO_AV_RTCP_Object
int
TAO_AV_RTCP_Object::handle_input (void)
{
  size_t bufsiz = 2*this->transport_->mtu ();
  ACE_Message_Block data (bufsiz);

  int n = this->transport_->recv (data.rd_ptr (),bufsiz);
  if (n == 0)
    {
      if (TAO_debug_level > 0)
        ACE_DEBUG ((LM_ERROR, "TAO_AV_RTCP::handle_input:connection closed\n"));
      return -1;
    }
  if (n < 0)
    {
      if (TAO_debug_level > 0)
        ACE_DEBUG ((LM_ERROR,"TAO_AV_RTCP::handle_input:recv error\n"));
      return -1;
    }
  data.wr_ptr (n);
  ACE_Addr *peer_addr = this->transport_->get_peer_addr ();
  this->callback_->receive_control_frame (&data,*peer_addr);
  return 0;
}

int
TAO_AV_RTCP_Object::send_frame (ACE_Message_Block *frame,
                                TAO_AV_frame_info * /*frame_info*/)
{
  return this->transport_->send (frame);
}

int
TAO_AV_RTCP_Object::send_frame (const iovec *iov,
                               int iovcnt,
                                TAO_AV_frame_info * /*frame_info*/)
{
  return this->transport_->send (iov,
                                 iovcnt);
}

int
TAO_AV_RTCP_Object::send_frame (const char*,
                                size_t)
{
  return 0;