From 2c2713a72cbb7388e034575b229cd3fd5700ff9f Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 7 Apr 2009 00:50:47 +0200 Subject: make use of SO_TIMESTAMP timestamp for accuracy and leave smoother paused until we have data --- src/modules/rtp/module-rtp-recv.c | 50 +++++++++++++++++++++++++++++++++------ 1 file changed, 43 insertions(+), 7 deletions(-) (limited to 'src/modules/rtp/module-rtp-recv.c') diff --git a/src/modules/rtp/module-rtp-recv.c b/src/modules/rtp/module-rtp-recv.c index 3b0fc532..c61d2d8b 100644 --- a/src/modules/rtp/module-rtp-recv.c +++ b/src/modules/rtp/module-rtp-recv.c @@ -52,6 +52,8 @@ #include #include #include +#include +#include #include "module-rtp-recv-symdef.h" @@ -165,7 +167,7 @@ static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes) { pa_memblockq_rewind(s->memblockq, nbytes); } -/* Called from thread context */ +/* Called from I/O thread context */ static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) { struct session *s; @@ -184,11 +186,24 @@ static void sink_input_kill(pa_sink_input* i) { session_free(s); } +/* Called from IO context */ +static void sink_input_suspend_within_thread(pa_sink_input* i, pa_bool_t b) { + struct session *s; + pa_sink_input_assert_ref(i); + pa_assert_se(s = i->userdata); + + if (b) { + pa_smoother_pause(s->smoother, pa_rtclock_usec()); + pa_memblockq_flush_read(s->memblockq); + } else + s->first_packet = FALSE; +} + /* Called from I/O thread context */ static int rtpoll_work_cb(pa_rtpoll_item *i) { pa_memchunk chunk; int64_t k, j, delta; - struct timeval now; + struct timeval now = { 0, 0 }; struct session *s; struct pollfd *p; @@ -206,10 +221,11 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) { p->revents = 0; - if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->module->core->mempool) < 0) + if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->module->core->mempool, &now) < 0) return 0; - if (s->sdp_info.payload != s->rtp_context.payload) { + if (s->sdp_info.payload != s->rtp_context.payload || + !PA_SINK_IS_OPENED(s->sink_input->sink->thread_info.state)) { pa_memblock_unref(chunk.memblock); return 0; } @@ -240,10 +256,19 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) { pa_memblockq_seek(s->memblockq, delta * (int64_t) s->rtp_context.frame_size, PA_SEEK_RELATIVE, TRUE); - pa_rtclock_get(&now); + if (now.tv_sec == 0) { + PA_ONCE_BEGIN { + pa_log_warn("Using artificial time instead of timestamp"); + } PA_ONCE_END; + pa_rtclock_get(&now); + } else + pa_rtclock_from_wallclock(&now); pa_smoother_put(s->smoother, pa_timeval_load(&now), pa_bytes_to_usec((uint64_t) pa_memblockq_get_write_index(s->memblockq), &s->sink_input->sample_spec)); + /* Tell the smoother that we are rolling now, in case it is still paused */ + pa_smoother_resume(s->smoother, pa_timeval_load(&now), TRUE); + if (pa_memblockq_push(s->memblockq, &chunk) < 0) { pa_log_warn("Queue overrun"); pa_memblockq_seek(s->memblockq, (int64_t) chunk.length, PA_SEEK_RELATIVE, TRUE); @@ -267,6 +292,8 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) { wi = pa_smoother_get(s->smoother, pa_timeval_load(&now)); ri = pa_bytes_to_usec((uint64_t) pa_memblockq_get_read_index(s->memblockq), &s->sink_input->sample_spec); + pa_log_debug("wi=%lu ri=%lu", (unsigned long) wi, (unsigned long) ri); + sink_delay = pa_sink_get_latency_within_thread(s->sink_input->sink); render_delay = pa_bytes_to_usec(pa_memblockq_get_length(s->sink_input->thread_info.render_memblockq), &s->sink_input->sink->sample_spec); @@ -292,7 +319,7 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) { fix_samples = (unsigned) (fix * (pa_usec_t) s->sink_input->thread_info.sample_spec.rate / (pa_usec_t) RATE_UPDATE_INTERVAL); /* Check if deviation is in bounds */ - if (fix_samples > s->sink_input->sample_spec.rate*.20) + if (fix_samples > s->sink_input->sample_spec.rate*.50) pa_log_debug("Hmmm, rate fix is too large (%lu Hz), not applying.", (unsigned long) fix_samples); else { /* Fix up rate */ @@ -366,6 +393,14 @@ static int mcast_socket(const struct sockaddr* sa, socklen_t salen) { goto fail; } + pa_make_udp_socket_low_delay(fd); + + one = 1; + if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one)) < 0) { + pa_log("SO_TIMESTAMP failed: %s", pa_cstrerror(errno)); + goto fail; + } + one = 1; if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) { pa_log("SO_REUSEADDR failed: %s", pa_cstrerror(errno)); @@ -441,7 +476,7 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in TRUE, 10, pa_timeval_load(&now), - FALSE); + TRUE); s->last_rate_update = pa_timeval_load(&now); pa_atomic_store(&s->timestamp, (int) now.tv_sec); @@ -482,6 +517,7 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in s->sink_input->kill = sink_input_kill; s->sink_input->attach = sink_input_attach; s->sink_input->detach = sink_input_detach; + s->sink_input->suspend_within_thread = sink_input_suspend_within_thread; pa_sink_input_get_silence(s->sink_input, &silence); -- cgit