rtpproxy_relay.c

siproxd is a proxy/masquerading for the SIP protocal.

/*
    Copyright (C) 2003  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 warranty 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>

#include <osipparser2/osip_parser.h>

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

static char const ident[]="$Id: " __FILE__ ": " PACKAGE "-" VERSION "-"
			  BUILDSTR " $";


/* 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 */
static void *rtpproxy_main(void *i);
static int rtp_recreate_fdset(void);
void rtpproxy_kill( void );
static void sighdl_alm(int sig) {/* just wake up from select() */};


/*
 * 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;

   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);

   return STS_SUCCESS;
}


/*
 * main() of rtpproxy
 */
static void *rtpproxy_main(void *arg) {
   struct timeval tv;
   fd_set fdset;
   int fd_max;
   time_t t, last_t=0;
   int i, sts;
   int num_fd;
   osip_call_id_t callid;
   static char rtp_buff[RTP_BUFFER_SIZE];
   int count;

   memcpy(&fdset, &master_fdset, sizeof(fdset));
   fd_max=master_fd_max;

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

      tv.tv_sec = 5;
      tv.tv_usec = 0;

      num_fd=select(fd_max+1, &fdset, NULL, NULL, &tv);
      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;
      }

      time(&t);

      /*
       * 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++) {
         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].sock*/
            count=read(rtp_proxytable[i].rtp_rx_sock, rtp_buff, RTP_BUFFER_SIZE);

            /* check if something went banana */
            if (count < 0) {
               /*
                * It has been seen on linux 2.2.x systems that for some
                * reason (ICMP issue? -> below) inside the RTP relay, select()
                * claims that a certain file descriptor has data available to
                * read, a subsequent call to read() or recv() then does block!!
                * So lets make the FD's we are going to use non-blocking, so
                * we will at least survive and not run into a deadlock.
                * 
                * We catch this here with this workaround (pronounce "HACK")
                * and hope that next time we pass by it will be ok again.
                */
               if (errno == EAGAIN) {
                  /*&&&& I may want to remove this WARNing */
                  WARN("read() [fd=%i, %s:%i] would block, but select() "
                       "claimed to be readable!",
                       rtp_proxytable[i].rtp_rx_sock,
                       utils_inet_ntoa(rtp_proxytable[i].local_ipaddr),
                       rtp_proxytable[i].local_port);
                  continue;
               }

               /*
                * I *MAY* receive ICMP destination unreachable messages when I
                * try to send RTP traffic to a destination that is in HOLD
                * (better: is not listening on the UDP port where I send
                * my RTP data to).
                * So I should *not* do this - or ignore errors originating
                * by this -> ECONNREFUSED
                *
                * Note: This error is originating from a previous send() on the
                *       same socket and has nothing to do with the read() we have
                *       done above!
                */
               if (errno != ECONNREFUSED) {
                  /* some other error that I probably want to know about */
                  int j;
                  WARN("read() [fd=%i, %s:%i] returned error [%i:%s]",
                  rtp_proxytable[i].rtp_rx_sock,
                  utils_inet_ntoa(rtp_proxytable[i].local_ipaddr),
                  rtp_proxytable[i].local_port, errno, strerror(errno));
                  for (j=0; j<RTPPROXY_SIZE;j++) {
                     DEBUGC(DBCLASS_RTP, "%i - rx:%i tx:%i %s@%s dir:%i "
                            "lp:%i, rp:%i rip:%s",
                            j,
                            rtp_proxytable[j].rtp_rx_sock,
                            rtp_proxytable[j].rtp_tx_sock,
                            rtp_proxytable[j].callid_number,
                            rtp_proxytable[j].callid_host,
                            rtp_proxytable[j].direction,
                            rtp_proxytable[j].local_port,
                            rtp_proxytable[j].remote_port,
                            utils_inet_ntoa(rtp_proxytable[j].remote_ipaddr));
                  } /* for j */
              } /* if errno != ECONNREFUSED */
            } /* count < 0 */

            /*
             * 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) {
                  int j;
                  int rtp_direction = rtp_proxytable[i].direction;
                  int media_stream_no = rtp_proxytable[i].media_stream_no;

                  callid.number = rtp_proxytable[i].callid_number;
                  callid.host = rtp_proxytable[i].callid_host;

                  for (j=0;(j<RTPPROXY_SIZE);j++) {
                     char *client_id = rtp_proxytable[i].client_id;
                     osip_call_id_t cid;
                     cid.number = rtp_proxytable[j].callid_number;
                     cid.host = rtp_proxytable[j].callid_host;

                     /* match on:
                      * - same call ID
                      * - same media stream
                      * - opposite direction
                      * - different client ID
                      */
                     if ( (rtp_proxytable[j].rtp_rx_sock != 0) &&
                          (compare_callid(&callid, &cid) == STS_SUCCESS) &&
                          (media_stream_no == rtp_proxytable[j].media_stream_no) &&
                          (rtp_direction != rtp_proxytable[j].direction) &&
                          (strcmp(rtp_proxytable[j].client_id, client_id) != 0) ) {
                        rtp_proxytable[i].rtp_tx_sock = rtp_proxytable[j].rtp_rx_sock;
                        DEBUGC(DBCLASS_RTP, "connected entry %i (fd=%i) <-> entry %i (fd=%i)",
                               j, rtp_proxytable[j].rtp_rx_sock,
                               i, rtp_proxytable[i].rtp_rx_sock);
                        break;
                     }
                  }
               } /* rtp_tx_sock == 0 */

               if (rtp_proxytable[i].rtp_tx_sock != 0) {
                   /* write to dest via socket rtp_tx_sock */
                   sts = sipsock_send_udp(&rtp_proxytable[i].rtp_tx_sock,
                                          rtp_proxytable[i].remote_ipaddr,
                                          rtp_proxytable[i].remote_port,
                                          rtp_buff, count, 0); /* don't dump it */

                  if (sts != STS_SUCCESS) {
                     /* 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");
                     /* don't lock the mutex, as we own the lock */
                     callid.number=rtp_proxytable[i].callid_number;
                     callid.host=rtp_proxytable[i].callid_host;
                     rtp_relay_stop_fwd(&callid,
                                        rtp_proxytable[i].direction, 1);
                  }
               }
            } /* count > 0 */

            /* update timestamp of last usage */
            rtp_proxytable[i].timestamp=t;
         }
      } /* for i */

      /*
       * age and clean rtp_proxytable (check every 10 seconds)
       */
      if (t > (last_t+10) ) {
         last_t = t;
	 for (i=0;i<RTPPROXY_SIZE; i++) {
            if ( (rtp_proxytable[i].rtp_rx_sock != 0) &&
		 ((rtp_proxytable[i].timestamp+configuration.rtp_timeout)<t)) {
               /* this one has expired, clean it up */
               callid.number=rtp_proxytable[i].callid_number;
               callid.host=rtp_proxytable[i].callid_host;
               DEBUGC(DBCLASS_RTP,"RTP stream rx_sock=%i tx_sock=%i "
                      "%s@%s (idx=%i) has expired",
                      rtp_proxytable[i].rtp_rx_sock,
                      rtp_proxytable[i].rtp_tx_sock,
                      callid.number, callid.host, i);
               /* don't lock the mutex, as we own the lock already here */
               rtp_relay_stop_fwd(&callid, rtp_proxytable[i].direction, 1);
	    }
	 }
      } /* if (t>...) */

      /* copy master FD set */
      memcpy(&fdset, &master_fdset, sizeof(fdset));
      fd_max=master_fd_max;

      /*
       * UNLOCK the MUTEX
       */
      pthread_mutex_unlock(&rtp_proxytable_mutex);
   } /* for(;;) */

   return NULL;
}


/*
 * start an rtp stream on the proxy
 *
 * RETURNS
 *	STS_SUCCESS on success
 *	STS_FAILURE on error
 */
int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id,
                         int rtp_direction,
                         int media_stream_no, struct in_addr local_ipaddr,
                         int *local_port, struct in_addr remote_ipaddr,
                         int remote_port) {
   int i, j;
   int sock, port;
   int freeidx;
   int sts=STS_SUCCESS;
   osip_call_id_t cid;
   

   if (callid == NULL) {
      ERROR("rtp_relay_start_fwd: callid is NULL!");