Using the new feature of pa_{source,sink}_get_latency_within_thread() leads
to improved end to end latency estimation and to correct handling of port
latency offsets.
---
src/modules/module-loopback.c | 67 ++++++++++++++++++++++++++++++-------------
1 file changed, 47 insertions(+), 20 deletions(-)
diff --git a/src/modules/module-loopback.c b/src/modules/module-loopback.c
index f505295..aee8658 100644
--- a/src/modules/module-loopback.c
+++ b/src/modules/module-loopback.c
@@ -110,12 +110,12 @@ struct userdata {
/* Used for sink input and source output snapshots */
struct {
int64_t send_counter;
- pa_usec_t source_latency;
+ int64_t source_latency;
pa_usec_t source_timestamp;
int64_t recv_counter;
size_t loopback_memblockq_length;
- pa_usec_t sink_latency;
+ int64_t sink_latency;
pa_usec_t sink_timestamp;
} latency_snapshot;
@@ -128,6 +128,7 @@ struct userdata {
/* Copied from main thread */
pa_usec_t effective_source_latency;
+ int64_t source_latency_offset;
pa_usec_t minimum_latency;
/* Various booleans */
@@ -171,7 +172,8 @@ enum {
SINK_INPUT_MESSAGE_SINK_CHANGED,
SINK_INPUT_MESSAGE_SOURCE_CHANGED,
SINK_INPUT_MESSAGE_SET_EFFECTIVE_SOURCE_LATENCY,
- SINK_INPUT_MESSAGE_UPDATE_MIN_LATENCY
+ SINK_INPUT_MESSAGE_UPDATE_MIN_LATENCY,
+ SINK_INPUT_MESSAGE_SOURCE_LATENCY_OFFSET_CHANGED
};
enum {
@@ -314,7 +316,8 @@ static void adjust_rates(struct userdata *u) {
size_t buffer;
uint32_t old_rate, base_rate, new_rate, run_hours;
int32_t latency_difference;
- pa_usec_t current_buffer_latency, snapshot_delay, current_source_sink_latency, current_latency, latency_at_optimum_rate;
+ pa_usec_t current_buffer_latency, snapshot_delay;
+ int64_t current_source_sink_latency, current_latency, latency_at_optimum_rate;
pa_usec_t final_latency;
pa_assert(u);
@@ -361,7 +364,7 @@ static void adjust_rates(struct userdata *u) {
latency_at_optimum_rate = current_source_sink_latency + current_buffer_latency * old_rate / base_rate;
final_latency = PA_MAX(u->latency, u->minimum_latency);
- latency_difference = (int32_t)((int64_t)latency_at_optimum_rate - final_latency);
+ latency_difference = (int32_t)(latency_at_optimum_rate - final_latency);
pa_log_debug("Loopback overall latency is %0.2f ms + %0.2f ms + %0.2f ms = %0.2f ms",
(double) u->latency_snapshot.sink_latency / PA_USEC_PER_MSEC,
@@ -474,12 +477,23 @@ static void update_latency_boundaries(struct userdata *u, pa_source *source, pa_
/* Called from output context
* Sets the memblockq to the configured latency corrected by latency_offset_usec */
-static void memblockq_adjust(struct userdata *u, pa_usec_t latency_offset_usec, bool allow_push) {
+static void memblockq_adjust(struct userdata *u, int64_t latency_offset_usec, bool allow_push) {
size_t current_memblockq_length, requested_memblockq_length, buffer_correction;
- pa_usec_t requested_buffer_latency, final_latency;
+ int64_t requested_buffer_latency;
+ pa_usec_t final_latency, requested_sink_latency;
final_latency = PA_MAX(u->latency, u->output_thread_info.minimum_latency);
- requested_buffer_latency = PA_CLIP_SUB(final_latency, latency_offset_usec);
+
+ /* If source or sink have some large negative latency offset, we might want to
+ * hold more than final_latency in the memblockq */
+ requested_buffer_latency = (int64_t)final_latency - latency_offset_usec;
+
+ /* Keep at least one sink latency in the queue to make sure that the sink
+ * never underruns initially */
+ requested_sink_latency = pa_sink_get_requested_latency_within_thread(u->sink_input->sink);
+ if (requested_buffer_latency < (int64_t)requested_sink_latency)
+ requested_buffer_latency = requested_sink_latency;
+
requested_memblockq_length = pa_usec_to_bytes(requested_buffer_latency, &u->sink_input->sample_spec);
current_memblockq_length = pa_memblockq_get_length(u->memblockq);
@@ -500,7 +514,8 @@ static void memblockq_adjust(struct userdata *u, pa_usec_t latency_offset_usec,
/* Called from input thread context */
static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
struct userdata *u;
- pa_usec_t push_time, current_source_latency;
+ pa_usec_t push_time;
+ int64_t current_source_latency;
pa_source_output_assert_ref(o);
pa_source_output_assert_io_context(o);
@@ -508,9 +523,9 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk)
/* Send current source latency and timestamp with the message */
push_time = pa_rtclock_now();
- current_source_latency = pa_source_get_latency_within_thread(u->source_output->source, false);
+ current_source_latency = pa_source_get_latency_within_thread(u->source_output->source, true);
- pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_POST, PA_UINT_TO_PTR(current_source_latency), push_time, chunk, NULL);
+ pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_POST, PA_INT_TO_PTR(current_source_latency), push_time, chunk, NULL);
u->send_counter += (int64_t) chunk->length;
}
@@ -539,7 +554,7 @@ static int source_output_process_msg_cb(pa_msgobject *obj, int code, void *data,
u->latency_snapshot.send_counter = u->send_counter;
/* Add content of delay memblockq to the source latency */
- u->latency_snapshot.source_latency = pa_source_get_latency_within_thread(u->source_output->source, false) +
+ u->latency_snapshot.source_latency = pa_source_get_latency_within_thread(u->source_output->source, true) +
pa_bytes_to_usec(length, &u->source_output->source->sample_spec);
u->latency_snapshot.source_timestamp = pa_rtclock_now();
@@ -773,7 +788,7 @@ static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk
* when it starts pushing. Adjust the memblockq accordingly and ensure that there is
* enough data in the queue to avoid underruns. */
if (!u->output_thread_info.push_called)
- memblockq_adjust(u, u->output_thread_info.effective_source_latency, true);
+ memblockq_adjust(u, (int64_t)u->output_thread_info.effective_source_latency + u->output_thread_info.source_latency_offset, true);
return 0;
}
@@ -815,19 +830,19 @@ static int sink_input_process_msg_cb(pa_msgobject *obj, int code, void *data, in
* are enabled. Disable them on first push and correct the memblockq. Do the
* same if the pop_cb() requested the adjustment */
if (!u->output_thread_info.push_called || u->output_thread_info.pop_adjust) {
- pa_usec_t time_delta;
+ int64_t time_delta;
- time_delta = PA_PTR_TO_UINT(data);
+ time_delta = PA_PTR_TO_INT(data);
time_delta += pa_rtclock_now() - offset;
- time_delta += pa_sink_get_latency_within_thread(u->sink_input->sink, false);
+ time_delta += pa_sink_get_latency_within_thread(u->sink_input->sink, true);
/* If the source has overrun, assume that the maximum it should have pushed is
* one full source latency. It may still be possible that the next push also
* contains too much data, then the resulting latency will be wrong. */
if (pa_bytes_to_usec(chunk->length, &u->sink_input->sample_spec) > u->output_thread_info.effective_source_latency)
- time_delta = PA_CLIP_SUB(time_delta, u->output_thread_info.effective_source_latency);
+ time_delta -= (int64_t)u->output_thread_info.effective_source_latency;
else
- time_delta = PA_CLIP_SUB(time_delta, pa_bytes_to_usec(chunk->length, &u->sink_input->sample_spec));
+ time_delta -= (int64_t)pa_bytes_to_usec(chunk->length, &u->sink_input->sample_spec);
memblockq_adjust(u, time_delta, true);
@@ -838,7 +853,7 @@ static int sink_input_process_msg_cb(pa_msgobject *obj, int code, void *data, in
/* If pop has not been called yet, make sure the latency does not grow too much.
* Don't push any silence here, because we already have new data in the queue */
if (!u->output_thread_info.pop_called)
- memblockq_adjust(u, pa_sink_get_latency_within_thread(u->sink_input->sink, false), false);
+ memblockq_adjust(u, pa_sink_get_latency_within_thread(u->sink_input->sink, true), false);
/* Is this the end of an underrun? Then let's start things
* right-away */
@@ -876,7 +891,7 @@ static int sink_input_process_msg_cb(pa_msgobject *obj, int code, void *data, in
u->latency_snapshot.recv_counter = u->output_thread_info.recv_counter;
u->latency_snapshot.loopback_memblockq_length = pa_memblockq_get_length(u->memblockq);
/* Add content of render memblockq to sink latency */
- u->latency_snapshot.sink_latency = pa_sink_get_latency_within_thread(u->sink_input->sink, false) +
+ u->latency_snapshot.sink_latency = pa_sink_get_latency_within_thread(u->sink_input->sink, true) +
pa_bytes_to_usec(length, &u->sink_input->sink->sample_spec);
u->latency_snapshot.sink_timestamp = pa_rtclock_now();
@@ -907,6 +922,12 @@ static int sink_input_process_msg_cb(pa_msgobject *obj, int code, void *data, in
u->output_thread_info.minimum_latency = (pa_usec_t)offset;
return 0;
+
+ case SINK_INPUT_MESSAGE_SOURCE_LATENCY_OFFSET_CHANGED:
+
+ u->output_thread_info.source_latency_offset = offset;
+
+ return 0;
}
return pa_sink_input_process_msg(obj, code, data, offset, chunk);
@@ -1185,6 +1206,12 @@ static pa_hook_result_t source_port_latency_offset_changed_cb(pa_core *core, pa_
u->source_latency_offset = source->port_latency_offset;
update_minimum_latency(u, u->sink_input->sink, true);
+ /* Update variable in output thread */
+ if (u->sink_input->sink)
+ pa_asyncmsgq_send(u->sink_input->sink->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_SOURCE_LATENCY_OFFSET_CHANGED, NULL, u->source_latency_offset, NULL);
+ else
+ u->output_thread_info.source_latency_offset = u->source_latency_offset;
+
return PA_HOOK_OK;
}
--
2.10.1
