From ae230ebe444a1b11590de43a054a37fbde79812d Mon Sep 17 00:00:00 2001 From: Thomas Ries Date: Sun, 18 Jun 2006 19:12:35 +0000 Subject: [PATCH] - RTCP support - dejitter feature by Hans Carlos Hofmann () --- ChangeLog | 2 + doc/siproxd.conf.example | 7 + src/proxy.c | 27 +- src/readconf.c | 4 + src/rtpproxy.c | 33 +- src/rtpproxy.h | 24 +- src/rtpproxy_relay.c | 887 +++++++++++++++++++++++++++++++-------- src/siproxd.h | 12 +- 8 files changed, 812 insertions(+), 184 deletions(-) diff --git a/ChangeLog b/ChangeLog index 019d76d..6935c14 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,5 +1,7 @@ 0.5.13 ====== + 18-Jun-2006: - included RTCP support and a de-jitter feature + (submitted by by Hans Carlos Hofmann) 11-Jun-2006: - RTP timeout handling: allow unidirectional data w/o terminating connection after timeout. 20-May-2006: - Fixed compiling issue when building on MacOS (Georg Schwarz) diff --git a/doc/siproxd.conf.example b/doc/siproxd.conf.example index c1bdaf7..f70ff12 100644 --- a/doc/siproxd.conf.example +++ b/doc/siproxd.conf.example @@ -142,6 +142,13 @@ rtp_timeout = 300 # rtp_dscp = 46 +###################################################################### +# Dejitter value +# in useconds +# +rtp_input_dejitter = 300000 +rtp_output_dejitter = 300000 + ###################################################################### # Default Expiration timeout for Registrations # If a REGISTER request does not contain an Expires header diff --git a/src/proxy.c b/src/proxy.c index 5a7ee1c..7549ae7 100644 --- a/src/proxy.c +++ b/src/proxy.c @@ -21,6 +21,7 @@ #include "config.h" #include +#include #include #include #include @@ -791,6 +792,7 @@ int proxy_rewrite_invitation_body(sip_ticket_t *ticket, int direction){ sdp_media_t *sdp_med; int rtp_direction=0; int have_c_media=0; + int isrtp = 0 ; if (configuration.rtp_proxy_enable == 0) return STS_SUCCESS; @@ -1000,6 +1002,28 @@ if (configuration.debuglevel) msg_port=atoi(sdp_message_m_port_get(sdp, media_stream_no)); if (msg_port > 0) { + + /* is this an RTP stream ? */ + { + char *protocol = sdp_message_m_proto_get (sdp, media_stream_no); + if (protocol == NULL) { + DEBUGC(DBCLASS_PROXY, "no protocol definition found!"); + } else { + char *check; + char *cmp; + isrtp = 1; + check = protocol ; + cmp = "RTP/" ; + while (*cmp && (isrtp = isrtp && *check) && + (isrtp = isrtp && (*cmp++ == toupper(*check++))) ) {} ; + if (isrtp) { + DEBUGC(DBCLASS_PROXY, "found RTP protocol [%s]!", protocol); + } else { + DEBUGC(DBCLASS_PROXY, "found non RTP protocol [%s]!", protocol); + } + } + } + osip_uri_t *cont_url = NULL; char *client_id=NULL; /* try to get some additional UA specific unique ID. @@ -1051,7 +1075,8 @@ if (configuration.debuglevel) rtp_direction, media_stream_no, map_addr, &map_port, - addr_media, msg_port); + addr_media, msg_port, + isrtp); if (sts == STS_SUCCESS) { /* and rewrite the port */ diff --git a/src/readconf.c b/src/readconf.c index 92d1d5c..3de1960 100644 --- a/src/readconf.c +++ b/src/readconf.c @@ -157,6 +157,8 @@ static int parse_config (FILE *configfile) { { "rtp_timeout", TYP_INT4, &configuration.rtp_timeout }, { "rtp_proxy_enable", TYP_INT4, &configuration.rtp_proxy_enable }, { "rtp_dscp", TYP_INT4, &configuration.rtp_dscp }, + { "rtp_input_dejitter", TYP_INT4, &configuration.rtp_input_dejitter }, + { "rtp_output_dejitter", TYP_INT4, &configuration.rtp_output_dejitter }, { "user", TYP_STRING, &configuration.user }, { "chrootjail", TYP_STRING, &configuration.chrootjail }, { "hosts_allow_reg", TYP_STRING, &configuration.hosts_allow_reg }, @@ -312,6 +314,8 @@ int make_default_config(void){ memset (&configuration, 0, sizeof(configuration)); configuration.sip_listen_port=SIP_PORT; configuration.default_expires=DEFAULT_EXPIRES; + configuration.rtp_input_dejitter=DEFAULT_DEJITTER; + configuration.rtp_output_dejitter=DEFAULT_DEJITTER; return STS_SUCCESS; } diff --git a/src/rtpproxy.c b/src/rtpproxy.c index 67698a3..73a5f6b 100644 --- a/src/rtpproxy.c +++ b/src/rtpproxy.c @@ -2,17 +2,17 @@ Copyright (C) 2003-2005 Thomas Ries 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 @@ -47,6 +47,16 @@ int rtpproxy_init( void ) { sts = STS_SUCCESS; } else if (configuration.rtp_proxy_enable == 1) { // Relay sts = rtp_relay_init (); + if ((configuration.rtp_output_dejitter < 0) || + (configuration.rtp_output_dejitter > DEJITTERLIMIT)) { + ERROR("CONFIG: rtp_output_dejitter has invalid value %i [0 .. %i]", + configuration.rtp_output_dejitter, DEJITTERLIMIT) ; + } + if ((configuration.rtp_input_dejitter < 0) || + (configuration.rtp_input_dejitter > DEJITTERLIMIT)) { + ERROR("CONFIG: rtp_input_dejitter has invalid value %i [0 .. %i]", + configuration.rtp_input_dejitter, DEJITTERLIMIT) ; + } } else { ERROR("CONFIG: rtp_proxy_enable has invalid value: %d", configuration.rtp_proxy_enable); @@ -64,17 +74,26 @@ int rtpproxy_init( void ) { */ int rtp_start_fwd (osip_call_id_t *callid, char *client_id, int direction, int media_stream_no, - struct in_addr local_ipaddr, int *local_port, - struct in_addr remote_ipaddr, int remote_port) { - int sts=STS_FAILURE; + struct in_addr local_ipaddr, int *local_port, + struct in_addr remote_ipaddr, int remote_port, + int isrtp) { + int sts=STS_FAILURE; + int dejitter=0; if (configuration.rtp_proxy_enable == 0) { sts = STS_SUCCESS; } else if (configuration.rtp_proxy_enable == 1) { // Relay + if (isrtp) { + if (direction == DIR_OUTGOING) { + dejitter = configuration.rtp_output_dejitter; + } else { + dejitter = configuration.rtp_input_dejitter; + } + } sts = rtp_relay_start_fwd (callid, client_id, direction, media_stream_no, local_ipaddr, local_port, - remote_ipaddr, remote_port); + remote_ipaddr, remote_port, dejitter); } else { ERROR("CONFIG: rtp_proxy_enable has invalid value: %d", configuration.rtp_proxy_enable); diff --git a/src/rtpproxy.h b/src/rtpproxy.h index fea361f..7cd70fd 100644 --- a/src/rtpproxy.h +++ b/src/rtpproxy.h @@ -23,14 +23,33 @@ #define CALLIDNUM_SIZE 256 #define CALLIDHOST_SIZE 128 #define CLIENT_ID_SIZE 128 + +typedef struct { + struct timeval starttime ; + int calccount ; + int dejitter ; + struct timeval dejitter_tv ; + double dejitter_d ; + int time_code_a ; + double recived_a ; /* time in µsec sience epoch */ + int time_code_b ; + double recived_b ; /* time in µsec sience epoch */ + int time_code_c ; + double recived_c ; /* time in µsec sience epoch */ + +} timecontrol_t ; + typedef struct { int rtp_rx_sock; /* rx socket (0 -> free slot)*/ int rtp_tx_sock; /* tx socket */ + int rtp_con_rx_sock; /* rx socket rtcp */ + int rtp_con_tx_sock; /* tx socket rtcp */ char callid_number[CALLIDNUM_SIZE]; /* call ID */ char callid_host[CALLIDHOST_SIZE]; /* --"-- */ char client_id[CLIENT_ID_SIZE]; int direction; /* Direction of RTP stream */ int media_stream_no; + timecontrol_t tc; struct in_addr local_ipaddr; /* local IP */ int local_port; /* local allocated port */ struct in_addr remote_ipaddr; /* remote IP */ @@ -45,7 +64,8 @@ typedef struct { int rtp_relay_init(void); 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); + struct in_addr local_ipaddr, int *local_port, + struct in_addr remote_ipaddr, int remote_port, + int dejitter); int rtp_relay_stop_fwd (osip_call_id_t *callid, int rtp_direction, int media_stream_no, int nolock); diff --git a/src/rtpproxy_relay.c b/src/rtpproxy_relay.c index e17d248..d9b09f6 100644 --- a/src/rtpproxy_relay.c +++ b/src/rtpproxy_relay.c @@ -2,17 +2,17 @@ Copyright (C) 2003-2005 Thomas Ries 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 @@ -71,12 +71,66 @@ static pthread_t rtpproxy_tid=0; static fd_set master_fdset; static int master_fd_max; -/* forward declarations */ +/* + * RTP buffers for dejitter + */ +typedef char rtp_buff_t[RTP_BUFFER_SIZE]; +typedef struct { + void *next; /* next free or next que element */ + int socked; /* socket number */ + size_t message_len; /* length of message */ + int flags; /* flags */ + struct sockaddr_in dst_addr; /* where shall i send */ + struct timeval transm_time; /* when shall i send */ + rtp_proxytable_t *errret; /* deliver error status */ + rtp_buff_t rtp_buff; /* Data storage */ +} rtp_delayed_message; + + +/* + * table to buffer date for dejitter function + */ +#define NUMBER_OF_BUFFER (10*RTPPROXY_SIZE) +static rtp_delayed_message rtp_buffer_area[NUMBER_OF_BUFFER]; + +static rtp_delayed_message *free_memory; +static rtp_delayed_message *msg_que; + +static struct timeval current_tv; +static struct timeval minstep; + +/* + * forward declarations of internal functions + */ 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() */}; +/* dejitter */ +void rtp_buffer_init (); +void add_time_values(const struct timeval *a, const struct timeval *b, + struct timeval *r); +void sub_time_values(const struct timeval *a, const struct timeval *b, + struct timeval *r); +int cmp_time_values(const struct timeval *a, const struct timeval *b); +double make_double_time ( const struct timeval *tv); +void send_top_of_que (); +void delayedsendto(int s, const void *msg, size_t len, int flags, + const struct sockaddr_in *to, + const struct timeval *tv, rtp_proxytable_t *errret); +void cancelmessages (rtp_proxytable_t *dropentry); +void flushbuffers(); +int delay_of_next_transmission(struct timeval *tv); +void split_double_time ( double d, struct timeval *tv); +void init_calculate_transmit_time (timecontrol_t *tc, int dejitter); +int fetch_missalign_long_network_oder (char *where); +void calculate_transmit_time (rtp_buff_t *rtp_buff, timecontrol_t *tc, + const struct timeval *input_tv, + struct timeval *ttv); +void match_socket (int i); +void error_summary (int i, int k); + /* * initialize and create rtp_relay proxy thread @@ -89,6 +143,8 @@ int rtp_relay_init( void ) { int arg=0; struct sigaction sigact; + rtp_buffer_init(); + atexit(rtpproxy_kill); /* cancel RTP thread at exit */ /* clean proxy table */ @@ -159,26 +215,42 @@ int rtp_relay_init( void ) { * 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 i; int num_fd; - osip_call_id_t callid; - static char rtp_buff[RTP_BUFFER_SIZE]; + static rtp_buff_t rtp_buff; int count; + struct timeval last_tv ; + struct timeval sleep_tv ; + struct timeval input_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 (;;) { - tv.tv_sec = 5; - tv.tv_usec = 0; + if (!delay_of_next_transmission(&sleep_tv)) + { +// DEBUGC(DBCLASS_RTP, "rtp que empty") ; + sleep_tv.tv_sec = 5; + sleep_tv.tv_usec = 0; + }; - num_fd=select(fd_max+1, &fdset, NULL, NULL, &tv); + num_fd=select(fd_max+1, &fdset, NULL, NULL, &sleep_tv); + gettimeofday(&input_tv,&tz); + current_tv = input_tv; + + /* + * Send delayed Packates + */ + flushbuffers(); + + /* exit point for this thread in case of program terminaction */ pthread_testcancel(); if ((num_fd<0) && (errno==EINTR)) { /* @@ -192,7 +264,6 @@ static void *rtpproxy_main(void *arg) { continue; } - time(&t); /* * LOCK the MUTEX @@ -201,71 +272,64 @@ static void *rtpproxy_main(void *arg) { /* check for data available and send to destination */ for (i=0;(i0);i++) { - if ( (rtp_proxytable[i].rtp_rx_sock != 0) && + /* RTCP control socket */ + if ( (rtp_proxytable[i].rtp_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].sock*/ + count=read(rtp_proxytable[i].rtp_con_rx_sock, rtp_buff, RTP_BUFFER_SIZE); + + /* check if something went banana */ + if (count < 0) error_summary (i,1) ; + + /* + * 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; + struct timeval ttv ; + + add_time_values(&(rtp_proxytable[i].tc.dejitter_tv), + &input_tv,&ttv) ; + + /* 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); + delayedsendto(rtp_proxytable[i].rtp_con_tx_sock, rtp_buff, + count, 0, &dst_addr, &ttv, &rtp_proxytable[i]) ; + } + } /* count > 0 */ + /* update timestamp of last usage for both (RX and TX) entries. + * This allows silence (no data) on one stream without breaking + * the connection after the RTP timeout */ + rtp_proxytable[i].timestamp=current_tv.tv_sec; + if (rtp_proxytable[i].opposite_entry > 0) { + rtp_proxytable[rtp_proxytable[i].opposite_entry-1].timestamp= + current_tv.tv_sec; + } + } /* 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].sock*/ + /* 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 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 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) match_socket(i); if (rtp_proxytable[i].rtp_tx_sock != 0) { - /* write to dest via socket rtp_tx_sock */ struct sockaddr_in dst_addr; + struct timeval ttv; + + /* 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, + &rtp_proxytable[i].remote_ipaddr, sizeof(struct in_addr)); dst_addr.sin_port= htons(rtp_proxytable[i].remote_port); - - 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) { - 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"); - /* 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, 1); - } - } + delayedsendto(rtp_proxytable[i].rtp_tx_sock, rtp_buff, + count, 0, &dst_addr, &ttv, &rtp_proxytable[i]); } } /* count > 0 */ - /* update timestamp of last usage for both (RX and TX) entries. * This allows silence (no data) on one stream without breaking * the connection after the RTP timeout */ - rtp_proxytable[i].timestamp=t; + rtp_proxytable[i].timestamp=current_tv.tv_sec; if (rtp_proxytable[i].opposite_entry > 0) { - rtp_proxytable[rtp_proxytable[i].opposite_entry-1].timestamp=t; + rtp_proxytable[rtp_proxytable[i].opposite_entry-1].timestamp= + current_tv.tv_sec; } - } + } /* if */ } /* for i */ /* * age and clean rtp_proxytable (check every 10 seconds) */ - if (t > (last_t+10) ) { - last_t = t; - for (i=0;i last_tv.tv_sec) { + last_tv.tv_sec = current_tv.tv_sec + 10 ; + for (i=0;i...) */ /* copy master FD set */ @@ -405,16 +422,16 @@ 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 remote_port, int dejitter) { static int prev_used_port = 0; int num_ports; int i2, i, j; int sock, port; + int sock_con; int freeidx; int sts=STS_SUCCESS; int tos; osip_call_id_t cid; - if (callid == NULL) { ERROR("rtp_relay_start_fwd: callid is NULL!"); @@ -489,7 +506,7 @@ int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id, * (seen with KPhone during HOLD/unHOLD) * Also the destination IP may change during a re-Invite * (seen with Sipphone.com, re-Invites when using - * the SIP - POTS gateway [SIP Minutes] + * the SIP - POTS gateway [SIP Minutes] */ /* Port number */ if (rtp_proxytable[i].remote_port != remote_port) { @@ -505,6 +522,13 @@ int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id, memcpy (&rtp_proxytable[i].remote_ipaddr, &remote_ipaddr, sizeof(remote_ipaddr)); } + + /* + * set up timecrontrol for dejitter function + */ + init_calculate_transmit_time (&rtp_proxytable[i].tc,dejitter); + + /* return the already known local port number */ DEBUGC(DBCLASS_RTP,"RTP stream already active (remaddr=%s, " "remport=%i, lclport=%i, id=%s, #=%i)", @@ -513,8 +537,8 @@ int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id, rtp_proxytable[i].local_port, rtp_proxytable[i].callid_number, rtp_proxytable[i].media_stream_no); - *local_port=rtp_proxytable[i].local_port; - sts = STS_SUCCESS; + *local_port=rtp_proxytable[i].local_port; + sts = STS_SUCCESS; goto unlock_and_exit; } } @@ -527,7 +551,7 @@ int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id, for (j=0; j RTP port pool fully allocated */ - if ((port == 0) || (sock == 0)) { + if ((port == 0) || (sock == 0) || (sock_con == 0)) { ERROR("rtp_relay_start_fwd: no RTP port available or bind() failed"); sts = STS_FAILURE; goto unlock_and_exit; } + /*&&&: do RTP and RTCP both set DSCP value? */ /* set DSCP value, need to be ROOT */ if (configuration.rtp_dscp) { int uid,euid; @@ -619,6 +659,7 @@ int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id, /* write entry into rtp_proxytable slot (freeidx) */ rtp_proxytable[freeidx].rtp_rx_sock=sock; + rtp_proxytable[freeidx].rtp_con_rx_sock = sock_con; if (callid->number) { strcpy(rtp_proxytable[freeidx].callid_number, callid->number); @@ -648,14 +689,25 @@ int rtp_relay_start_fwd (osip_call_id_t *callid, char *client_id, rtp_proxytable[freeidx].remote_port=remote_port; time(&rtp_proxytable[freeidx].timestamp); + /* + * set up timecrontrol for dejitter function + */ + init_calculate_transmit_time (&rtp_proxytable[freeidx].tc,dejitter); + *local_port=port; - /* call to firewall API */ + /* call to firewall API: RTP port */ fwapi_start_rtp(rtp_proxytable[freeidx].direction, rtp_proxytable[freeidx].local_ipaddr, rtp_proxytable[freeidx].local_port, rtp_proxytable[freeidx].remote_ipaddr, rtp_proxytable[freeidx].remote_port); + /* call to firewall API: RTCP port */ + fwapi_start_rtp(rtp_proxytable[freeidx].direction, + rtp_proxytable[freeidx].local_ipaddr, + rtp_proxytable[freeidx].local_port + 1, + rtp_proxytable[freeidx].remote_ipaddr, + rtp_proxytable[freeidx].remote_port + 1); /* prepare FD set for next select operation */ rtp_recreate_fdset(); @@ -742,13 +794,14 @@ int rtp_relay_stop_fwd (osip_call_id_t *callid, (rtp_proxytable[i].direction == rtp_direction) && ((media_stream_no < 0) || (media_stream_no == rtp_proxytable[i].media_stream_no))) { + /* close RTP sockets */ sts = close(rtp_proxytable[i].rtp_rx_sock); - DEBUGC(DBCLASS_RTP,"closed socket %i for RTP stream " + DEBUGC(DBCLASS_RTP,"closed socket %i for RTP stream " "%s:%s == %s:%s (idx=%i) sts=%i", - rtp_proxytable[i].rtp_rx_sock, - rtp_proxytable[i].callid_number, - rtp_proxytable[i].callid_host, - callid->number, callid->host, i, sts); + rtp_proxytable[i].rtp_rx_sock, + rtp_proxytable[i].callid_number, + rtp_proxytable[i].callid_host, + callid->number, callid->host, i, sts); if (sts < 0) { ERROR("Error in close(%i): %s nolock=%i %s:%s\n", rtp_proxytable[i].rtp_rx_sock, @@ -761,11 +814,26 @@ int rtp_relay_stop_fwd (osip_call_id_t *callid, rtp_proxytable[i].local_port, rtp_proxytable[i].remote_ipaddr, rtp_proxytable[i].remote_port); + /* close RTCP socket */ + sts = close(rtp_proxytable[i].rtp_con_rx_sock); + DEBUGC(DBCLASS_RTP,"closed socket %i for RTCP stream sts=%i", + rtp_proxytable[i].rtp_con_rx_sock, sts); + if (sts < 0) { + ERROR("Error in close(%i): %s nolock=%i %s:%s\n", + rtp_proxytable[i].rtp_con_rx_sock, + strerror(errno), nolock, + callid->number, callid->host); + } + /* call to firewall API */ + fwapi_stop_rtp(rtp_proxytable[i].direction, + rtp_proxytable[i].local_ipaddr, + rtp_proxytable[i].local_port + 1, + rtp_proxytable[i].remote_ipaddr, + rtp_proxytable[i].remote_port + 1); /* clean up */ memset(&rtp_proxytable[i], 0, sizeof(rtp_proxytable[0])); got_match=1; } - } /* did not find an active stream... */ @@ -812,10 +880,16 @@ static int rtp_recreate_fdset(void) { master_fd_max=-1; for (i=0;i master_fd_max) { - master_fd_max=rtp_proxytable[i].rtp_rx_sock; - } + if (rtp_proxytable[i].rtp_rx_sock > master_fd_max) { + master_fd_max=rtp_proxytable[i].rtp_rx_sock; + } + /* RTPCP */ + FD_SET(rtp_proxytable[i].rtp_con_rx_sock, &master_fdset); + if (rtp_proxytable[i].rtp_con_rx_sock > master_fd_max) { + master_fd_max=rtp_proxytable[i].rtp_con_rx_sock; + } } } /* for i */ return STS_SUCCESS; @@ -855,3 +929,474 @@ void rtpproxy_kill( void ) { return; } + +/*********** + * De-Jitter + ***********/ + +/* + * Initialize RTP dejitter + */ +void rtp_buffer_init () { + int i; + rtp_delayed_message *m; + + memset (&rtp_buffer_area, 0, sizeof(rtp_buffer_area)); + free_memory = NULL; + msg_que = NULL; + for (i=0,m=&rtp_buffer_area[0];inext = free_memory; + free_memory = m; + } +} + +/* + * Add timeval times + */ +void add_time_values(const struct timeval *a, const struct timeval *b, + struct timeval *r) { + r->tv_sec = a->tv_sec + b->tv_sec; + r->tv_usec = a->tv_usec + b->tv_usec; + if (r->tv_usec >= 1000000) { + r->tv_usec -= 1000000; + r->tv_sec++; + } +} + +/* + * Subtract timeval values + */ +void sub_time_values(const struct timeval *a, const struct timeval *b, + struct timeval *r) { + if ((a->tv_sec < b->tv_sec) || + ((a->tv_sec == b->tv_sec) && (a->tv_usec < b->tv_usec))) { + r->tv_usec = 0; + r->tv_sec = 0; + return; + } + if (a->tv_usec < b->tv_usec) { + r->tv_sec = a->tv_sec - b->tv_sec - 1; + r->tv_usec = a->tv_usec + 1000000 - b->tv_usec; + } else { + r->tv_sec = a->tv_sec - b->tv_sec; + r->tv_usec = a->tv_usec - b->tv_usec; + } +} + +/* + * Compare timeval values + */ +int cmp_time_values(const struct timeval *a, const struct timeval *b) { + if (a->tv_sec < b->tv_sec) return -1; + if (a->tv_sec > b->tv_sec) return 1; + if (a->tv_usec < b->tv_usec) return -1; + if (a->tv_usec > b->tv_usec) return 1; + return 0; +} + +/* + * Convert TIMEVAL to DOUBLE + */ +double make_double_time ( const struct timeval *tv) { + return 1000000.0 * tv->tv_sec + tv->tv_usec; +} + +/* + * Send Top of queue + */ +void send_top_of_que () { + rtp_delayed_message *m; + int sts; + + if (msg_que) { + m = msg_que; + msg_que = m->next; + m->next = free_memory; + free_memory = m; + + if ((m->errret != NULL) && (m->errret->rtp_tx_sock)) { + sts = sendto(m->socked, &(m->rtp_buff), m->message_len, + m->flags, (const struct sockaddr *)&(m->dst_addr), + (socklen_t)sizeof(m->dst_addr)); + if ((sts == -1) && (m->errret != NULL) && (errno != ECONNREFUSED)) { + osip_call_id_t callid; + + ERROR("sendto() [%s:%i size=%i] delayed call failed: %s", + utils_inet_ntoa(m->errret->remote_ipaddr), + m->errret->remote_port, m->message_len, 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"); + /* don't lock the mutex, as we own the lock */ + + callid.number=m->errret->callid_number; + callid.host=m->errret->callid_host; + if (STS_SUCCESS != rtp_relay_stop_fwd(&callid, + m->errret->direction, + m->errret->media_stream_no, 0)) { + ERROR("fatal error in delayed error close! [%s:%i size=%i]", + utils_inet_ntoa(m->errret->remote_ipaddr), + m->errret->remote_port, m->message_len); + /* brute force protection agains looping errors */ + m->errret->rtp_rx_sock = 0; + rtp_recreate_fdset(); + } /* if stp_stop */ + } /* if sendto fails */ + } + } /* if (msg_que) */ +} + +/* + * Delayed send + */ +void delayedsendto(int s, const void *msg, size_t len, int flags, + const struct sockaddr_in *to, + const struct timeval *tv, rtp_proxytable_t *errret) { + rtp_delayed_message *m; + rtp_delayed_message **linkin; + + if (!free_memory) send_top_of_que(); + + m = free_memory; + + m->socked = s; + memcpy(&(m->rtp_buff), msg, m->message_len = len); + m->flags = flags; + m->dst_addr = *to; + m->transm_time = *tv; + m->errret = errret; + + free_memory = m->next; + + if (cmp_time_values(¤t_tv,tv) >= 0) { + m->next = msg_que; + msg_que = m; + send_top_of_que (); + } else { + linkin = &msg_que; + while ((*linkin != NULL) && + (cmp_time_values(&((*linkin)->transm_time),tv) < 0)) { + linkin = (rtp_delayed_message **)&((*linkin)->next); + } + m->next = *linkin; + *linkin = m; + } +} + +/* + * Cancel a message + */ +void cancelmessages (rtp_proxytable_t *dropentry) { + rtp_delayed_message **linkout; + rtp_delayed_message *m; + + linkout = &msg_que; + + while (*linkout != NULL) { + if ((*linkout)->errret == dropentry) { + m = *linkout; + *linkout = m->next; + m->next = free_memory; + free_memory = m; + } else { + linkout = (rtp_delayed_message **)&((*linkout)->next); + } + } +} + +/* + * Flush buffers + */ +void flushbuffers() { + struct timezone tz; + + while (msg_que && + (cmp_time_values(&(msg_que->transm_time),¤t_tv)<=0)) { + send_top_of_que (); + gettimeofday(¤t_tv,&tz); + } +} + +/* + * Delay of next transmission + */ +int delay_of_next_transmission(struct timeval *tv) { + struct timezone tz ; + + if (msg_que) { + gettimeofday(¤t_tv,&tz); + sub_time_values(&(msg_que->transm_time),¤t_tv,tv); + if (cmp_time_values(tv,&minstep)<=0) { + *tv = minstep ; + } + return -1; + } + return 0; +} + +/* + * Convert DOUBLE time into TIMEVAL + */ +void split_double_time ( double d, struct timeval *tv) { + tv->tv_sec = d / 1000000.0; + tv->tv_usec = d - 1000000.0 * tv->tv_sec; +} + +/* + * Initialize calculation of transmit the frame + */ +void init_calculate_transmit_time (timecontrol_t *tc, int dejitter) { + struct timezone tz; + + minstep.tv_sec = 0; + minstep.tv_usec = 6000; + memset(tc, 0, sizeof(*tc)); + if (dejitter>0) { + gettimeofday(&(tc->starttime),&tz); + + tc->dejitter = dejitter; + tc->dejitter_d = dejitter; + split_double_time (tc->dejitter_d,&(tc->dejitter_tv)); + } +} + +/* + * + */ +int fetch_missalign_long_network_oder (char *where) { + int i = 0; + int k; + int j; + + for (j=0;j<4;j++) { + k = *where; + i = (i<<8) | (0xFF & k); + where ++; + } + return i; +} + +/* + * Calculate transmit time + */ +void calculate_transmit_time (rtp_buff_t *rtp_buff, timecontrol_t *tc, + const struct timeval *input_tv, + struct timeval *ttv) { + int packet_time_code; + double currenttime; + double calculatedtime = 0; + double calculatedtime2 = 0; + struct timeval input_r_tv; + struct timeval output_r_tv; + + if (!tc || !tc->dejitter) { + *ttv = current_tv; + return; + } + + + /* I hate this computer language ... :-/ quite confuse ! look modula */ + packet_time_code = fetch_missalign_long_network_oder(&((*rtp_buff)[4])); + + if (tc->calccount == 0) { + DEBUGC(DBCLASS_RTP, "initialise time calculatin"); + tc->starttime = *input_tv; + tc->time_code_a = packet_time_code; + } + + sub_time_values(input_tv,&(tc->starttime),&input_r_tv); + + calculatedtime = currenttime = make_double_time (&input_r_tv); + if (tc->calccount < 10) { + DEBUGC(DBCLASS_RTP, "initial data stage 1 %f usec", currenttime); + tc->recived_a = currenttime / (packet_time_code - tc->time_code_a); + } else if (tc->calccount < 20) { + tc->recived_a = 0.95 * tc->recived_a + 0.05 * currenttime / + (packet_time_code - tc->time_code_a); + } else { + tc->recived_a = 0.99 * tc->recived_a + 0.01 * currenttime / + (packet_time_code - tc->time_code_a); + } + if (tc->calccount > 20) { + if (!tc->time_code_b) { + tc->time_code_b = packet_time_code; + tc->recived_b = currenttime; + } else if (tc->time_code_b < packet_time_code) { + calculatedtime = tc->recived_b = tc->recived_b + + (packet_time_code - tc->time_code_b) * tc->recived_a; + tc->time_code_b = packet_time_code; + if (tc->calccount < 28) { + tc->recived_b = 0.90 * tc->recived_b + 0.1 * currenttime; + } else if (tc->calccount < 300) { + tc->recived_b = 0.95 * tc->recived_b + 0.05 * currenttime; + } else { + tc->recived_b = 0.99 * tc->recived_b + 0.01 * currenttime; + } + } else { + calculatedtime = tc->recived_b + + (packet_time_code - tc->time_code_b) * tc->recived_a; + } + } + tc->recived_c = currenttime; + tc->time_code_c = packet_time_code; + + if (tc->calccount < 30) { + /* + * But in the start phase, + * we asume every packet as not delayed. + */ + calculatedtime = currenttime; + } + + /* + ** theoretical value for F1000 Phone + */ + //calculatedtime = (tc->recived_a = 125.) * packet_time_code; + + tc->calccount ++; + calculatedtime += tc->dejitter_d; + + if (calculatedtime < currenttime) { + calculatedtime = currenttime; + } else if (calculatedtime > currenttime + 2.* tc->dejitter_d) { + calculatedtime = currenttime + 2.* tc->dejitter_d; + } + + if (tc->calccount % 100 == 0) { + DEBUGC(DBCLASS_RTP, "currenttime = %f", currenttime); + DEBUGC(DBCLASS_RTP, "packetcode = %i", packet_time_code); + DEBUGC(DBCLASS_RTP, "timecodes %i, %i, %i", + tc->time_code_a, tc->time_code_b, tc->time_code_c); + DEBUGC(DBCLASS_RTP, "measuredtimes %f usec, %f usec, %f usec", + tc->recived_a, tc->recived_b, tc->recived_c); + DEBUGC(DBCLASS_RTP, "p2 - p1 = (%i,%f usec)", + tc->time_code_b - tc->time_code_a, + tc->recived_b - tc->recived_a); + if (tc->time_code_c) { + DEBUGC(DBCLASS_RTP, "p3 - p2 = (%i,%f usec)", + tc->time_code_c - tc->time_code_b, + tc->recived_c - tc->recived_b); + } + DEBUGC(DBCLASS_RTP, "calculatedtime = %f", calculatedtime); + if (calculatedtime2) { + DEBUGC(DBCLASS_RTP, "calculatedtime2 = %f", calculatedtime2); + } + DEBUGC(DBCLASS_RTP, "transmtime = %f (%f)", (calculatedtime) / + (160. * tc->recived_a) - packet_time_code / 160, + currenttime / (160. * tc->recived_a) - + packet_time_code / 160); + DEBUGC(DBCLASS_RTP, "synthetic latency = %f, %f, %f, %i, %i", + calculatedtime-currenttime, calculatedtime, + currenttime, packet_time_code, + packet_time_code / 160); + } + + split_double_time (calculatedtime,&output_r_tv); + add_time_values(&output_r_tv,&(tc->starttime),ttv); +} + +/* + * + */ +void match_socket (int i) { + int j; + int rtp_direction = rtp_proxytable[i].direction; + int media_stream_no = rtp_proxytable[i].media_stream_no; +/*chnnel int channel = rtp_proxytable[i].channel;*/ + osip_call_id_t callid; + + callid.number = rtp_proxytable[i].callid_number; + callid.host = rtp_proxytable[i].callid_host; + + for (j=0;(j entry %i (fd=%i)", + j, rtp_proxytable[j].rtp_rx_sock, + i, rtp_proxytable[i].rtp_rx_sock); + break; + } + } +} + +/* + * + */ +void error_summary (int i, int k) { + /* + * 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!", + k ? rtp_proxytable[i].rtp_rx_sock : rtp_proxytable[i].rtp_con_rx_sock, + utils_inet_ntoa(rtp_proxytable[i].local_ipaddr), + rtp_proxytable[i].local_port + k); + } + + /* + * 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