rtpproxy_relay.c

这是一个C程序

/*
    Copyright (C) 2003-2007  Thomas Ries <[e-mail]>

    This file is part of Siproxd.

    Siproxd is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    Siproxd is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warrantry of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with Siproxd; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA 
*/

#include "config.h"

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/time.h>

#include <sys/socket.h>
#include <netinet/in.h>
#include <signal.h>

#ifdef HAVE_PTHREAD_SETSCHEDPARAM
   #include <sched.h>
#endif

#include <osipparser2/osip_parser.h>

#include "siproxd.h"
#include "rtpproxy.h"
#include "log.h"

#ifdef GPL
   #include "dejitter.h"
#endif

#if !defined(SOL_IP)
#define SOL_IP IPPROTO_IP
#endif

static char const ident[]="$Id: rtpproxy_relay.c,v 1.51 2007/06/08 19:39:03 hb9xar Exp $";

/* configuration storage */
extern struct siproxd_config configuration;

/*
 * table to remember all active rtp proxy streams
 */
rtp_proxytable_t rtp_proxytable[RTPPROXY_SIZE];

/*
 * Mutex for thread synchronization (locking when accessing common 
 * data structures -> rtp_proxytable[]).
 *
 * use a 'fast' mutex for synchronizing - as these are portable... 
 */
static pthread_mutex_t rtp_proxytable_mutex = PTHREAD_MUTEX_INITIALIZER;

/* thread id of RTP proxy */
static pthread_t rtpproxy_tid=0;

/* master fd_set */
static fd_set master_fdset;
static int    master_fd_max;

/*
 * forward declarations of internal functions
 */
static void sighdl_alm(int sig) {/* just wake up from select() */};
static void *rtpproxy_main(void *i);
static void rtpproxy_kill( void );
static int  rtp_recreate_fdset(void);
static void match_socket (int rtp_proxytable_idx);
static void error_handler (int rtp_proxytable_idx, int socket_type);


/*
 * initialize and create rtp_relay proxy thread
 *
 * RETURNS
 *	STS_SUCCESS on success
 */
int rtp_relay_init( void ) {
   int sts;
   int arg=0;
   struct sigaction sigact;

#ifdef USE_DEJITTER
   dejitter_init();
#endif

   atexit(rtpproxy_kill);  /* cancel RTP thread at exit */

   /* clean proxy table */
   memset (rtp_proxytable, 0, sizeof(rtp_proxytable));

   /* initialize fd set for RTP proxy thread */
   FD_ZERO(&master_fdset); /* start with an empty fdset */
   master_fd_max=-1;

   /* install signal handler for SIGALRM - used to wake up
      the rtpproxy thread from select() hibernation */
   sigact.sa_handler = sighdl_alm;
   sigemptyset(&sigact.sa_mask);
   sigact.sa_flags=0;
   sigaction(SIGALRM, &sigact, NULL);

   DEBUGC(DBCLASS_RTP,"create thread");
   sts=pthread_create(&rtpproxy_tid, NULL, rtpproxy_main, (void *)&arg);
   DEBUGC(DBCLASS_RTP,"created, sts=%i", sts);

   /* set realtime scheduling - if started by root */
#ifdef HAVE_PTHREAD_SETSCHEDPARAM
   {
      int uid,euid;
      struct sched_param schedparam;

#ifndef _CYGWIN
      uid=getuid();
      euid=geteuid();
      DEBUGC(DBCLASS_RTP,"uid=%i, euid=%i", uid, euid);
      if (uid != euid) seteuid(0);

      if (geteuid()==0) {
#endif

#if defined(HAVE_SCHED_GET_PRIORITY_MAX) && defined(HAVE_SCHED_GET_PRIORITY_MIN)
         int pmin, pmax;
         /* place ourself at 1/3 of the available priority space */
         pmin=sched_get_priority_min(SCHED_RR);
         pmax=sched_get_priority_max(SCHED_RR);
         schedparam.sched_priority=pmin+(pmax-pmin)/3;
         DEBUGC(DBCLASS_RTP,"pmin=%i, pmax=%i, using p=%i", pmin, pmax,
                schedparam.sched_priority);
#else
         /* just taken a number out of thin air */
         schedparam.sched_priority=10;
         DEBUGC(DBCLASS_RTP,"using p=%i", schedparam.sched_priority);
#endif
         sts=pthread_setschedparam(rtpproxy_tid, SCHED_RR, &schedparam);
         if (sts != 0) {
            ERROR("pthread_setschedparam failed: %s", strerror(errno));
         }
#ifndef _CYGWIN
      } else {
         INFO("Unable to use realtime scheduling for RTP proxy");
         INFO("You may want to start siproxd as root and switch UID afterwards");
      }
      if (uid != euid)  seteuid(euid);
#endif
   }
#endif

   return STS_SUCCESS;
}


/*
 * main() of rtpproxy
 */
static void *rtpproxy_main(void *arg) {
   fd_set fdset;
   int fd_max;
   int i, sts;
   int num_fd;
   static rtp_buff_t rtp_buff;
   int count;
   struct timeval last_tv ;
   struct timeval sleep_tv ;
   struct timeval current_tv ;
   struct timezone tz ;

   memcpy(&fdset, &master_fdset, sizeof(fdset));
   fd_max=master_fd_max;
   last_tv.tv_sec = 0;
   last_tv.tv_usec = 0;

   /* loop forever... */
   for (;;) {

#ifdef USE_DEJITTER
      /* calculate time until next packet to send from dejitter buffer */
      if (!dejitter_delay_of_next_tx(&sleep_tv, &current_tv)) {
         sleep_tv.tv_sec = 5;
         sleep_tv.tv_usec = 0;
      };
#else
      sleep_tv.tv_sec = 5;
      sleep_tv.tv_usec = 0;
#endif

      num_fd=select(fd_max+1, &fdset, NULL, NULL, &sleep_tv);
      gettimeofday(&current_tv, &tz);

#ifdef USE_DEJITTER
      /* Send delayed Packets that are timed to be send */
      dejitter_flush(&current_tv, LOCK_FDSET);
#endif

      /* exit point for this thread in case of program terminaction */
      pthread_testcancel();
      if ((num_fd<0) && (errno==EINTR)) {
         /*
          * wakeup due to a change in the proxy table:
          * lock mutex, copy master FD set and unlock
          */
         pthread_mutex_lock(&rtp_proxytable_mutex);
         memcpy(&fdset, &master_fdset, sizeof(fdset));
         fd_max=master_fd_max;
         pthread_mutex_unlock(&rtp_proxytable_mutex);
         continue;
      }


      /*
       * LOCK the MUTEX
       */
      pthread_mutex_lock(&rtp_proxytable_mutex);

      /* check for data available and send to destination */
      for (i=0;(i<RTPPROXY_SIZE) && (num_fd>0);i++) {
         /*
          * RTCP control socket
          */
         if ( (rtp_proxytable[i].rtp_con_rx_sock != 0) &&
            FD_ISSET(rtp_proxytable[i].rtp_con_rx_sock, &fdset) ) {
            /* yup, have some data to send */
            num_fd--;

            /* read from sock rtp_proxytable[i].rtp_con_rx_sock */
            count=read(rtp_proxytable[i].rtp_con_rx_sock, rtp_buff, RTP_BUFFER_SIZE);

            /* check if something went banana */
            if (count < 0) error_handler(i,1) ;

            /* Buffer really full? This may indicate a too small buffer! */
            if (count == RTP_BUFFER_SIZE) {
               LIMIT_LOG_RATE(30) {
                  WARN("received an RTCP datagram bigger than buffer size");
               }
            }

            /*
             * forwarding an RTCP packet only makes sense if we really
             * have got some data in it (count > 0)
             */
            if (count > 0) {
               /* find the corresponding TX socket */
               if (rtp_proxytable[i].rtp_con_tx_sock == 0) match_socket(i);

               if (rtp_proxytable[i].rtp_con_tx_sock != 0) {
                  struct sockaddr_in dst_addr;

                  /* write to dest via socket rtp_con_tx_sock */
                  dst_addr.sin_family = AF_INET;
                  memcpy(&dst_addr.sin_addr.s_addr,
                         &rtp_proxytable[i].remote_ipaddr,
                         sizeof(struct in_addr));
                  dst_addr.sin_port= htons(rtp_proxytable[i].remote_port+1);

                  /* Don't dejitter RTCP packets */
                  sts = sendto(rtp_proxytable[i].rtp_con_tx_sock, rtp_buff,
                               count, 0, (const struct sockaddr *)&dst_addr,
                               (socklen_t)sizeof(dst_addr));
                  /* ignore errors here. We don't know if the remote
                     site does receive RTCP messages at all (or reject
                     them with ICMP-whatever). If it fails, it is lost.
                     Basta, end of story. */
               }
            } /* count > 0 */
            /* RTCP does not wind up the keepalive timestamp. */
         } /* if */

         /*
          * RTP data stream
          */
         if ( (rtp_proxytable[i].rtp_rx_sock != 0) &&
            FD_ISSET(rtp_proxytable[i].rtp_rx_sock, &fdset) ) {
            /* yup, have some data to send */
            num_fd--;

            /* read from sock rtp_proxytable[i].rtp_rx_sock */
            count=read(rtp_proxytable[i].rtp_rx_sock, rtp_buff, RTP_BUFFER_SIZE);

            /* check if something went banana */
            if (count < 0) error_handler (i,0);

            /* Buffer really full? This may indicate a too small buffer! */
            if (count == RTP_BUFFER_SIZE) {
               LIMIT_LOG_RATE(30) {
                  WARN("received an RTP datagram bigger than buffer size");
               }
            }

            /*
             * forwarding an RTP packet only makes sense if we really
             * have got some data in it (count > 0)
             */
            if (count > 0) {
               /* find the corresponding TX socket */
               if (rtp_proxytable[i].rtp_tx_sock == 0) match_socket(i);

               if (rtp_proxytable[i].rtp_tx_sock != 0) {
                  struct sockaddr_in dst_addr;
#ifdef USE_DEJITTER
                  struct timeval ttv;
#endif

                  /* write to dest via socket rtp_tx_sock */
                  dst_addr.sin_family = AF_INET;
                  memcpy(&dst_addr.sin_addr.s_addr,
                         &rtp_proxytable[i].remote_ipaddr,
                         sizeof(struct in_addr));
                  dst_addr.sin_port= htons(rtp_proxytable[i].remote_port);

#ifdef USE_DEJITTER
                  dejitter_calc_tx_time(&rtp_buff, &(rtp_proxytable[i].tc),
                                          &current_tv, &ttv);
                  dejitter_delayedsendto(rtp_proxytable[i].rtp_tx_sock,
                                         rtp_buff, count, 0, &dst_addr,
                                         &ttv, &current_tv,
                                         &rtp_proxytable[i], NOLOCK_FDSET);
#else
                  sts = sendto(rtp_proxytable[i].rtp_tx_sock, rtp_buff,
                               count, 0, (const struct sockaddr *)&dst_addr,
                               (socklen_t)sizeof(dst_addr));
                  if (sts == -1) {
                     if (errno != ECONNREFUSED) {
                        osip_call_id_t callid;

                        ERROR("sendto() [%s:%i size=%i] call failed: %s",
                        utils_inet_ntoa(rtp_proxytable[i].remote_ipaddr),
                        rtp_proxytable[i].remote_port, count, strerror(errno));

                        /* if sendto() fails with bad filedescriptor,
                         * this means that the opposite stream has been
                         * canceled or timed out.
                         * we should then cancel this stream as well.*/

                        WARN("stopping opposite stream");