Author: anthm
Date: Tue Mar 31 14:10:43 2009
New Revision: 12854
Log:
add rtp-autoflush profile param and rtp_autoflush var to skip timer sleeps when the socket has data ready and refactor the jitter buffer code out into a different function and fix it so it works in blocking mode too
Modified:
freeswitch/trunk/src/include/switch_types.h
freeswitch/trunk/src/mod/endpoints/mod_sofia/mod_sofia.h
freeswitch/trunk/src/mod/endpoints/mod_sofia/sofia.c
freeswitch/trunk/src/mod/endpoints/mod_sofia/sofia_glue.c
freeswitch/trunk/src/switch_rtp.c
Modified: freeswitch/trunk/src/include/switch_types.h
==============================================================================
--- freeswitch/trunk/src/include/switch_types.h (original)
+++ freeswitch/trunk/src/include/switch_types.h Tue Mar 31 14:10:43 2009
@@ -470,7 +470,8 @@
SWITCH_RTP_FLAG_SECURE_RECV_RESET = (1 << 17),
SWITCH_RTP_FLAG_PROXY_MEDIA = (1 << 18),
SWITCH_RTP_FLAG_SHUTDOWN = (1 << 19),
- SWITCH_RTP_FLAG_FLUSH = (1 << 20)
+ SWITCH_RTP_FLAG_FLUSH = (1 << 20),
+ SWITCH_RTP_FLAG_AUTOFLUSH = (1 << 21)
} switch_rtp_flag_enum_t;
typedef uint32_t switch_rtp_flag_t;
Modified: freeswitch/trunk/src/mod/endpoints/mod_sofia/mod_sofia.h
==============================================================================
--- freeswitch/trunk/src/mod/endpoints/mod_sofia/mod_sofia.h (original)
+++ freeswitch/trunk/src/mod/endpoints/mod_sofia/mod_sofia.h Tue Mar 31 14:10:43 2009
@@ -181,6 +181,7 @@
PFLAG_MANAGE_SHARED_APPEARANCE,
PFLAG_DISABLE_SRV,
PFLAG_DISABLE_NAPTR,
+ PFLAG_AUTOFLUSH,
/* No new flags below this line */
PFLAG_MAX
Modified: freeswitch/trunk/src/mod/endpoints/mod_sofia/sofia.c
==============================================================================
--- freeswitch/trunk/src/mod/endpoints/mod_sofia/sofia.c (original)
+++ freeswitch/trunk/src/mod/endpoints/mod_sofia/sofia.c Tue Mar 31 14:10:43 2009
@@ -1655,6 +1655,12 @@
} else {
sofia_clear_pflag(profile, PFLAG_PASS_RFC2833);
}
+ } else if (!strcasecmp(var, "rtp-autoflush")) {
+ if (switch_true(val)) {
+ sofia_set_pflag(profile, PFLAG_AUTOFLUSH);
+ } else {
+ sofia_clear_pflag(profile, PFLAG_AUTOFLUSH);
+ }
} else if (!strcasecmp(var, "inbound-codec-negotiation")) {
if (!strcasecmp(val, "greedy")) {
sofia_set_pflag(profile, PFLAG_GREEDY);
@@ -2181,6 +2187,12 @@
if (switch_true(val)) {
sofia_set_pflag(profile, PFLAG_PASS_RFC2833);
}
+ } else if (!strcasecmp(var, "rtp-autoflush")) {
+ if (switch_true(val)) {
+ sofia_set_pflag(profile, PFLAG_AUTOFLUSH);
+ } else {
+ sofia_clear_pflag(profile, PFLAG_AUTOFLUSH);
+ }
} else if (!strcasecmp(var, "inbound-codec-negotiation")) {
if (!strcasecmp(val, "greedy")) {
sofia_set_pflag(profile, PFLAG_GREEDY);
Modified: freeswitch/trunk/src/mod/endpoints/mod_sofia/sofia_glue.c
==============================================================================
--- freeswitch/trunk/src/mod/endpoints/mod_sofia/sofia_glue.c (original)
+++ freeswitch/trunk/src/mod/endpoints/mod_sofia/sofia_glue.c Tue Mar 31 14:10:43 2009
@@ -2054,6 +2054,12 @@
flags |= SWITCH_RTP_FLAG_PASS_RFC2833;
}
+
+ if (sofia_test_pflag(tech_pvt->profile, PFLAG_AUTOFLUSH)
+ || ((val = switch_channel_get_variable(tech_pvt->channel, "rtp_autoflush")) && switch_true(val))) {
+ flags |= SWITCH_RTP_FLAG_AUTOFLUSH;
+ }
+
if (!(sofia_test_pflag(tech_pvt->profile, PFLAG_REWRITE_TIMESTAMPS) ||
((val = switch_channel_get_variable(tech_pvt->channel, "rtp_rewrite_timestamps")) && !switch_true(val)))) {
flags |= SWITCH_RTP_FLAG_RAW_WRITE;
Modified: freeswitch/trunk/src/switch_rtp.c
==============================================================================
--- freeswitch/trunk/src/switch_rtp.c (original)
+++ freeswitch/trunk/src/switch_rtp.c Tue Mar 31 14:10:43 2009
@@ -599,10 +599,10 @@
if (switch_test_flag(rtp_session, SWITCH_RTP_FLAG_USE_TIMER) || switch_test_flag(rtp_session, SWITCH_RTP_FLAG_NOBLOCK)) {
switch_socket_opt_set(rtp_session->sock_input, SWITCH_SO_NONBLOCK, TRUE);
switch_set_flag_locked(rtp_session, SWITCH_RTP_FLAG_NOBLOCK);
- } else {
- switch_socket_create_pollfd(&rtp_session->read_pollfd, rtp_session->sock_input, SWITCH_POLLIN | SWITCH_POLLERR, rtp_session->pool);
}
+ switch_socket_create_pollfd(&rtp_session->read_pollfd, rtp_session->sock_input, SWITCH_POLLIN | SWITCH_POLLERR, rtp_session->pool);
+
status = SWITCH_STATUS_SUCCESS;
*err = "Success";
switch_set_flag_locked(rtp_session, SWITCH_RTP_FLAG_IO);
@@ -1016,13 +1016,8 @@
SWITCH_DECLARE(switch_status_t) switch_rtp_activate_jitter_buffer(switch_rtp_t *rtp_session, uint32_t queue_frames)
{
- if (rtp_session->read_pollfd) {
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Can't use jitterbuffer without a timer.\n");
- return SWITCH_STATUS_FALSE;
- }
rtp_session->jb = stfu_n_init(queue_frames);
- switch_socket_create_pollfd(&rtp_session->jb_pollfd, rtp_session->sock_input, SWITCH_POLLIN | SWITCH_POLLERR, rtp_session->pool);
return SWITCH_STATUS_SUCCESS;
}
@@ -1371,18 +1366,53 @@
#define return_cng_frame() do_cng = 1; goto timer_check
+static switch_status_t read_rtp_packet(switch_rtp_t *rtp_session, switch_size_t *bytes, switch_frame_flag_t *flags)
+{
+ switch_status_t status = SWITCH_STATUS_FALSE;
+ stfu_frame_t *jb_frame;
+
+ switch_assert(bytes);
+
+ *bytes = sizeof(rtp_msg_t);
+ status = switch_socket_recvfrom(rtp_session->from_addr, rtp_session->sock_input, 0, (void *) &rtp_session->recv_msg, bytes);
+
+ if (rtp_session->jb && rtp_session->recv_msg.header.version == 2 && *bytes) {
+ if (rtp_session->recv_msg.header.m && rtp_session->recv_msg.header.pt != rtp_session->te) {
+ stfu_n_reset(rtp_session->jb);
+ }
+
+ stfu_n_eat(rtp_session->jb, ntohl(rtp_session->recv_msg.header.ts), rtp_session->recv_msg.body, *bytes - rtp_header_len);
+ *bytes = 0;
+ status = SWITCH_STATUS_FALSE;
+ }
+
+ if (rtp_session->jb) {
+ if ((jb_frame = stfu_n_read_a_frame(rtp_session->jb))) {
+ memcpy(rtp_session->recv_msg.body, jb_frame->data, jb_frame->dlen);
+ if (jb_frame->plc) {
+ *flags |= SFF_PLC;
+ }
+ *bytes = jb_frame->dlen + rtp_header_len;
+ rtp_session->recv_msg.header.ts = htonl(jb_frame->ts);
+ rtp_session->recv_msg.header.pt = rtp_session->payload;
+ status = SWITCH_STATUS_SUCCESS;
+ }
+ }
+
+ return status;
+}
+
static int rtp_common_read(switch_rtp_t *rtp_session, switch_payload_t *payload_type, switch_frame_flag_t *flags, switch_io_flag_t io_flags)
{
switch_size_t bytes = 0;
switch_status_t status = SWITCH_STATUS_SUCCESS, poll_status = SWITCH_STATUS_SUCCESS;
int check = 0;
- stfu_frame_t *jb_frame;
int ret = -1;
int sleep_mss = 1000;
int poll_sec = 5;
int poll_loop = 0;
int fdr = 0;
- int from_jb = 0;
+ int hot_socket = 0;
if (!switch_rtp_ready(rtp_session)) {
return -1;
@@ -1398,42 +1428,24 @@
int do_cng = 0;
if (rtp_session->timer.interval) {
- int do_sleep = 1;
- if (rtp_session->jb) {
- if (switch_poll(rtp_session->jb_pollfd, 1, &fdr, 1) == SWITCH_STATUS_SUCCESS) {
- do_sleep = 0;
+ if (switch_test_flag(rtp_session, SWITCH_RTP_FLAG_AUTOFLUSH)) {
+ if (switch_poll(rtp_session->read_pollfd, 1, &fdr, 1) == SWITCH_STATUS_SUCCESS) {
+ hot_socket = 1;
}
}
- if (do_sleep) {
+ if (!hot_socket) {
switch_core_timer_next(&rtp_session->timer);
}
}
recvfrom:
- if (rtp_session->jb) {
- if ((jb_frame = stfu_n_read_a_frame(rtp_session->jb))) {
- memcpy(rtp_session->recv_msg.body, jb_frame->data, jb_frame->dlen);
- if (jb_frame->plc) {
- *flags |= SFF_PLC;
- }
- bytes = jb_frame->dlen + rtp_header_len;
- rtp_session->recv_msg.header.ts = htonl(jb_frame->ts);
- rtp_session->recv_msg.header.pt = rtp_session->payload;
- from_jb = 1;
- goto post_read;
- }
- }
-
-
- if (rtp_session->read_pollfd) {
+ if (!rtp_session->timer.interval) {
poll_status = switch_poll(rtp_session->read_pollfd, 1, &fdr, poll_sec * 1000000);
}
-
-
+
if (poll_status == SWITCH_STATUS_SUCCESS) {
- bytes = sizeof(rtp_msg_t);
- status = switch_socket_recvfrom(rtp_session->from_addr, rtp_session->sock_input, 0, (void *) &rtp_session->recv_msg, &bytes);
+ status = read_rtp_packet(rtp_session, &bytes, flags);
} else {
if (!SWITCH_STATUS_IS_BREAK(poll_status) && poll_status != SWITCH_STATUS_TIMEOUT) {
ret = -1;
@@ -1445,8 +1457,6 @@
bytes = 0;
}
- post_read:
-
if (bytes < 0) {
ret = (int) bytes;
goto end;
@@ -1574,20 +1584,6 @@
goto end;
}
- if (rtp_session->jb && !from_jb && ((bytes && rtp_session->recv_msg.header.pt == rtp_session->payload) || check)) {
- if (bytes) {
-
- if (rtp_session->recv_msg.header.m) {
- stfu_n_reset(rtp_session->jb);
- }
-
- stfu_n_eat(rtp_session->jb, ntohl(rtp_session->recv_msg.header.ts), rtp_session->recv_msg.body, bytes - rtp_header_len);
- bytes = 0;
- goto recvfrom;
- }
- }
-
-
if (bytes && switch_test_flag(rtp_session, SWITCH_RTP_FLAG_SECURE_RECV)) {
int sbytes = (int) bytes;
err_status_t stat = 0;
@@ -1749,14 +1745,7 @@
}
if (check || (bytes && !rtp_session->timer.interval)) {
- if (rtp_session->jb && (jb_frame = stfu_n_read_a_frame(rtp_session->jb))) {
- memcpy(rtp_session->recv_msg.body, jb_frame->data, jb_frame->dlen);
- if (jb_frame->plc) {
- *flags |= SFF_PLC;
- }
- bytes = jb_frame->dlen + rtp_header_len;
- rtp_session->recv_msg.header.ts = htonl(jb_frame->ts);
- } else if (!bytes && switch_test_flag(rtp_session, SWITCH_RTP_FLAG_USE_TIMER)) { /* We're late! We're Late! */
+ if (!bytes && switch_test_flag(rtp_session, SWITCH_RTP_FLAG_USE_TIMER)) { /* We're late! We're Late! */
if (!switch_test_flag(rtp_session, SWITCH_RTP_FLAG_NOBLOCK) && status == SWITCH_STATUS_BREAK) {
switch_cond_next();
continue;