Author: anthm
Date: Fri Jan 18 13:14:53 2008
New Revision: 7285
Modified:
freeswitch/trunk/src/switch_rtp.c
Log:
tighten up rtp
Modified: freeswitch/trunk/src/switch_rtp.c
==============================================================================
--- freeswitch/trunk/src/switch_rtp.c (original)
+++ freeswitch/trunk/src/switch_rtp.c Fri Jan 18 13:14:53 2008
@@ -44,6 +44,10 @@
#include <datatypes.h>
#include <srtp.h>
+#define READ_INC(rtp_session) switch_mutex_lock(rtp_session->read_mutex); rtp_session->reading++
+#define READ_DEC(rtp_session) switch_mutex_unlock(rtp_session->read_mutex); rtp_session->reading--
+#define WRITE_INC(rtp_session) switch_mutex_lock(rtp_session->write_mutex); rtp_session->writing++
+#define WRITE_DEC(rtp_session) switch_mutex_unlock(rtp_session->write_mutex); rtp_session->writing--
#include "stfu.h"
@@ -167,6 +171,8 @@
switch_payload_t te;
switch_payload_t cng_pt;
switch_mutex_t *flag_mutex;
+ switch_mutex_t *read_mutex;
+ switch_mutex_t *write_mutex;
switch_timer_t timer;
uint8_t ready;
uint8_t cn;
@@ -176,6 +182,8 @@
uint32_t missed_count;
rtp_msg_t write_msg;
switch_rtp_crypto_key_t *crypto_keys[SWITCH_RTP_CRYPTO_MAX];
+ int reading;
+ int writing;
};
static int global_init = 0;
@@ -194,13 +202,16 @@
switch_stun_packet_t *packet;
unsigned int elapsed;
switch_size_t bytes;
+ switch_status_t status = SWITCH_STATUS_SUCCESS;
+ WRITE_INC(rtp_session);
+
switch_assert(rtp_session != NULL);
switch_assert(rtp_session->ice_user != NULL);
if (rtp_session->stuncount != 0) {
rtp_session->stuncount--;
- return SWITCH_STATUS_SUCCESS;
+ goto end;
}
if (rtp_session->last_stun) {
@@ -208,7 +219,8 @@
if (elapsed > 30000) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "No stun for a long time (PUNT!)\n");
- return SWITCH_STATUS_FALSE;
+ status = SWITCH_STATUS_FALSE;
+ goto end;
}
}
@@ -218,6 +230,10 @@
switch_socket_sendto(rtp_session->sock, rtp_session->remote_addr, 0, (void *) packet, &bytes);
rtp_session->stuncount = 25;
+ end:
+ WRITE_DEC(rtp_session);
+
+
return SWITCH_STATUS_SUCCESS;
}
@@ -228,6 +244,10 @@
char username[33] = { 0 };
unsigned char buf[512] = { 0 };
switch_size_t cpylen = len;
+
+
+ READ_INC(rtp_session);
+ WRITE_INC(rtp_session);
if (cpylen > 512) {
cpylen = 512;
@@ -271,6 +291,9 @@
bytes = switch_stun_packet_length(rpacket);
switch_socket_sendto(rtp_session->sock, rtp_session->from_addr, 0, (void *) rpacket, &bytes);
}
+
+ READ_DEC(rtp_session);
+ WRITE_DEC(rtp_session);
}
@@ -383,6 +406,9 @@
int x;
#endif
+ WRITE_INC(rtp_session);
+ READ_INC(rtp_session);
+
*err = NULL;
if (switch_sockaddr_info_get(&rtp_session->local_addr, host, SWITCH_UNSPEC, port, 0, rtp_session->pool) != SWITCH_STATUS_SUCCESS) {
@@ -460,6 +486,9 @@
if (old_sock) {
switch_socket_close(old_sock);
}
+
+ WRITE_DEC(rtp_session);
+ READ_DEC(rtp_session);
return status;
}
@@ -609,8 +638,11 @@
rtp_session->pool = pool;
rtp_session->te = 101;
+ rtp_session->ready = 1;
switch_mutex_init(&rtp_session->flag_mutex, SWITCH_MUTEX_NESTED, pool);
+ switch_mutex_init(&rtp_session->read_mutex, SWITCH_MUTEX_NESTED, pool);
+ switch_mutex_init(&rtp_session->write_mutex, SWITCH_MUTEX_NESTED, pool);
switch_mutex_init(&rtp_session->dtmf_data.dtmf_mutex, SWITCH_MUTEX_NESTED, pool);
switch_queue_create(&rtp_session->dtmf_data.dtmf_queue, 100, rtp_session->pool);
switch_queue_create(&rtp_session->dtmf_data.dtmf_inqueue, 100, rtp_session->pool);
@@ -703,7 +735,7 @@
end:
if (rtp_session) {
- rtp_session->ready = 1;
+ rtp_session->ready = 2;
rtp_session->rx_host = switch_core_strdup(rtp_session->pool, rx_host);
rtp_session->rx_port = rx_port;
} else {
@@ -755,27 +787,42 @@
switch_assert(rtp_session != NULL);
switch_mutex_lock(rtp_session->flag_mutex);
if (switch_test_flag(rtp_session, SWITCH_RTP_FLAG_IO)) {
+ switch_clear_flag(rtp_session, SWITCH_RTP_FLAG_IO);
switch_assert(rtp_session->sock != NULL);
switch_socket_shutdown(rtp_session->sock, SWITCH_SHUTDOWN_READWRITE);
- switch_clear_flag(rtp_session, SWITCH_RTP_FLAG_IO);
}
switch_mutex_unlock(rtp_session->flag_mutex);
}
SWITCH_DECLARE(uint8_t) switch_rtp_ready(switch_rtp_t *rtp_session)
{
- return (rtp_session != NULL && rtp_session->sock && rtp_session->ready) ? 1 : 0;
+ return (rtp_session != NULL && switch_test_flag(rtp_session, SWITCH_RTP_FLAG_IO) && rtp_session->sock && rtp_session->ready == 2) ? 1 : 0;
}
SWITCH_DECLARE(void) switch_rtp_destroy(switch_rtp_t **rtp_session)
{
void *pop;
switch_socket_t *sock;
+ int sanity = 0;
- if (!switch_rtp_ready(*rtp_session)) {
+ switch_mutex_lock((*rtp_session)->flag_mutex);
+
+ if (!rtp_session || !*rtp_session || !(*rtp_session)->ready) {
return;
}
+ (*rtp_session)->ready = 0;
+
+
+ while((*rtp_session)->reading || (*rtp_session)->writing) {
+ switch_yield(10000);
+ if (++sanity > 1000) {
+ break;
+ }
+ }
+
+ switch_rtp_kill_socket(*rtp_session);
+
while(switch_queue_trypop((*rtp_session)->dtmf_data.dtmf_inqueue, &pop) == SWITCH_STATUS_SUCCESS) {
free(pop);
}
@@ -783,22 +830,16 @@
while(switch_queue_trypop((*rtp_session)->dtmf_data.dtmf_queue, &pop) == SWITCH_STATUS_SUCCESS) {
free(pop);
}
-
- (*rtp_session)->ready = 0;
-
- switch_mutex_lock((*rtp_session)->flag_mutex);
if ((*rtp_session)->jb) {
stfu_n_destroy(&(*rtp_session)->jb);
}
- switch_rtp_kill_socket(*rtp_session);
sock = (*rtp_session)->sock;
(*rtp_session)->sock = NULL;
switch_socket_close(sock);
-
if (switch_test_flag((*rtp_session), SWITCH_RTP_FLAG_VAD)) {
switch_rtp_disable_vad(*rtp_session);
}
@@ -824,8 +865,8 @@
}
switch_rtp_release_port((*rtp_session)->rx_host, (*rtp_session)->rx_port);
-
switch_mutex_unlock((*rtp_session)->flag_mutex);
+
return;
}
@@ -993,11 +1034,14 @@
switch_status_t status;
uint8_t check = 1;
stfu_frame_t *jb_frame;
+ int ret = -1;
if (!rtp_session->timer.interval) {
rtp_session->last_time = switch_time_now();
}
+ READ_INC(rtp_session);
+
while (switch_rtp_ready(rtp_session)) {
bytes = sizeof(rtp_msg_t);
status = switch_socket_recvfrom(rtp_session->from_addr, rtp_session->sock, 0, (void *) &rtp_session->recv_msg, &bytes);
@@ -1006,10 +1050,6 @@
switch_core_timer_step(&rtp_session->timer);
}
- if (!switch_test_flag(rtp_session, SWITCH_RTP_FLAG_IO)) {
- return -1;
- }
-
if (bytes && switch_test_flag(rtp_session, SWITCH_RTP_FLAG_SECURE_RECV)) {
int sbytes = (int) bytes;
err_status_t stat = 0;
@@ -1020,7 +1060,8 @@
rtp_session->recv_ctx = NULL;
if ((stat = srtp_create(&rtp_session->recv_ctx, &rtp_session->recv_policy))) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error! RE-Activating Secure RTP RECV\n");
- return -1;
+ ret = -1;
+ goto end;
} else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "RE-Activating Secure RTP RECV\n");
rtp_session->srtp_errs = 0;
@@ -1035,7 +1076,8 @@
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,
"error: srtp unprotection failed with code %d%s\n", stat,
stat == err_status_replay_fail ? " (replay check failed)" : stat == err_status_auth_fail ? " (auth check failed)" : "");
- return -1;
+ ret = -1;
+ goto end;
} else {
sbytes = 0;
}
@@ -1075,11 +1117,13 @@
*flags |= SFF_CNG;
/* Return a CNG frame */
*payload_type = SWITCH_RTP_CNG_PAYLOAD;
- return 2 + rtp_header_len;
+ ret = 2 + rtp_header_len;
+ goto end;
}
if (bytes < 0) {
- return (int) bytes;
+ ret = (int) bytes;
+ goto end;
}
if (rtp_session->timer.interval) {
@@ -1101,7 +1145,8 @@
do_2833(rtp_session);
if (!bytes && rtp_session->max_missed_packets) {
if (++rtp_session->missed_count >= rtp_session->max_missed_packets) {
- return -2;
+ ret = -2;
+ goto end;
}
}
@@ -1126,7 +1171,8 @@
rtp_session->recv_msg.header.pt = (uint32_t) rtp_session->cng_pt ? rtp_session->cng_pt : SWITCH_RTP_CNG_PAYLOAD;
*flags |= SFF_CNG;
*payload_type = (switch_payload_t)rtp_session->recv_msg.header.pt;
- return 2 + rtp_header_len;
+ ret = 2 + rtp_header_len;
+ goto end;
}
}
@@ -1146,7 +1192,8 @@
rtp_session->recv_msg.header.pt = (uint32_t) rtp_session->cng_pt ? rtp_session->cng_pt : SWITCH_RTP_CNG_PAYLOAD;
*flags |= SFF_CNG;
*payload_type = (switch_payload_t)rtp_session->recv_msg.header.pt;
- return 2 + rtp_header_len;
+ ret = 2 + rtp_header_len;
+ goto end;
}
if (bytes > 0) {
@@ -1157,7 +1204,8 @@
if (switch_test_flag(rtp_session, SWITCH_RTP_FLAG_DATAWAIT)) {
goto do_continue;
}
- return 0;
+ ret = 0;
+ goto end;
}
if (bytes && switch_test_flag(rtp_session, SWITCH_RTP_FLAG_AUTOADJ) && switch_sockaddr_get_port(rtp_session->from_addr)) {
@@ -1255,7 +1303,12 @@
do_continue:
- switch_yield(1000);
+ if (rtp_session->ms_per_packet) {
+ switch_yield((rtp_session->ms_per_packet / 1000) * 750);
+ } else {
+ switch_yield(1000);
+ }
+
}
*payload_type = (switch_payload_t) rtp_session->recv_msg.header.pt;
@@ -1268,7 +1321,13 @@
do_2833(rtp_session);
}
- return (int) bytes;
+ ret = (int) bytes;
+
+ end:
+
+ READ_DEC(rtp_session);
+
+ return ret;
}
SWITCH_DECLARE(switch_size_t) switch_rtp_has_dtmf(switch_rtp_t *rtp_session)
@@ -1453,11 +1512,14 @@
switch_size_t bytes;
uint8_t send = 1;
uint32_t this_ts = 0;
+ int ret;
if (!switch_rtp_ready(rtp_session)) {
return SWITCH_STATUS_FALSE;
}
+ WRITE_INC(rtp_session);
+
if (send_msg) {
bytes = datalen;
if (flags && *flags & SFF_RFC2833) {
@@ -1600,7 +1662,8 @@
}
}
} else {
- return SWITCH_STATUS_GENERR;
+ ret = -1;
+ goto end;
}
}
@@ -1625,7 +1688,8 @@
rtp_session->send_ctx = NULL;
if ((stat = srtp_create(&rtp_session->send_ctx, &rtp_session->send_policy))) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error! RE-Activating Secure RTP SEND\n");
- return -1;
+ ret = -1;
+ goto end;
} else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "RE-Activating Secure RTP SEND\n");
}
@@ -1642,7 +1706,8 @@
if (switch_socket_sendto(rtp_session->sock, rtp_session->remote_addr, 0, (void *) send_msg, &bytes) != SWITCH_STATUS_SUCCESS) {
rtp_session->seq--;
- return -1;
+ ret = -1;
+ goto end;
}
if (rtp_session->timer.interval) {
@@ -1654,11 +1719,19 @@
if (rtp_session->ice_user) {
if (ice_out(rtp_session) != SWITCH_STATUS_SUCCESS) {
- return -1;
+ ret = -1;
+ goto end;
}
}
- return (int) bytes;
+ ret = (int) bytes;
+
+ end:
+
+ WRITE_DEC(rtp_session);
+
+ return ret;
+
}
@@ -1729,16 +1802,12 @@
switch_payload_t payload;
rtp_msg_t *send_msg = NULL;
- if (!switch_rtp_ready(rtp_session)) {
+ if (!switch_rtp_ready(rtp_session) || !rtp_session->remote_addr) {
return -1;
}
fwd = (switch_test_flag(rtp_session, SWITCH_RTP_FLAG_RAW_WRITE) && switch_test_flag(frame, SFF_RAW_RTP)) ? 1 : 0;
- if (!switch_test_flag(rtp_session, SWITCH_RTP_FLAG_IO) || !rtp_session->remote_addr) {
- return -1;
- }
-
switch_assert(frame != NULL);
if (switch_test_flag(frame, SFF_CNG)) {
@@ -1771,18 +1840,13 @@
uint8_t m, switch_payload_t payload, uint32_t ts, switch_frame_flag_t *flags)
{
switch_size_t bytes;
+ int ret = -1;
- if (!switch_rtp_ready(rtp_session)) {
+ if (!switch_rtp_ready(rtp_session) || !rtp_session->remote_addr || datalen > SWITCH_RTP_MAX_BUF_LEN) {
return -1;
}
- if (!switch_test_flag(rtp_session, SWITCH_RTP_FLAG_IO) || !rtp_session->remote_addr) {
- return -1;
- }
-
- if (datalen > SWITCH_RTP_MAX_BUF_LEN) {
- return -1;
- }
+ WRITE_INC(rtp_session);
rtp_session->write_msg = rtp_session->send_msg;
rtp_session->write_msg.header.seq = htons(++rtp_session->seq);
@@ -1803,7 +1867,8 @@
rtp_session->send_ctx = NULL;
if ((stat = srtp_create(&rtp_session->send_ctx, &rtp_session->send_policy))) {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error! RE-Activating Secure RTP SEND\n");
- return -1;
+ ret = -1;
+ goto end;
} else {
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "RE-Activating Secure RTP SEND\n");
}
@@ -1818,12 +1883,20 @@
if (switch_socket_sendto(rtp_session->sock, rtp_session->remote_addr, 0, (void *) &rtp_session->write_msg, &bytes) != SWITCH_STATUS_SUCCESS) {
rtp_session->seq--;
- return -1;
+ ret = -1;
+ goto end;
}
rtp_session->last_write_ts = ts;
- return (int) bytes;
+ ret = (int) bytes;
+
+ end:
+
+ WRITE_DEC(rtp_session);
+
+ return ret;
+
}
SWITCH_DECLARE(uint32_t) switch_rtp_get_ssrc(switch_rtp_t *rtp_session)