"sink_properties=<properties for the sink> "
"sink_master=<name of sink to filter> "
"adjust_time=<how often to readjust rates in s> "
+ "adjust_threshold=<how much drift to readjust after in ms> "
"format=<sample format> "
"rate=<sample rate> "
"channels=<number of channels> "
PA_ECHO_CANCELLER_INVALID = -1,
PA_ECHO_CANCELLER_SPEEX = 0,
PA_ECHO_CANCELLER_ADRIAN,
+#ifdef HAVE_WEBRTC
+ PA_ECHO_CANCELLER_WEBRTC,
+#endif
} pa_echo_canceller_method_t;
#define DEFAULT_ECHO_CANCELLER "speex"
.run = pa_adrian_ec_run,
.done = pa_adrian_ec_done,
},
+#ifdef HAVE_WEBRTC
+ {
+ /* WebRTC's audio processing engine */
+ .init = pa_webrtc_ec_init,
+ .run = pa_webrtc_ec_run,
+ .done = pa_webrtc_ec_done,
+ },
+#endif
};
#define DEFAULT_RATE 32000
#define DEFAULT_CHANNELS 1
#define DEFAULT_ADJUST_TIME_USEC (1*PA_USEC_PER_SEC)
+#define DEFAULT_ADJUST_TOLERANCE (5*PA_USEC_PER_MSEC)
#define DEFAULT_SAVE_AEC FALSE
#define DEFAULT_AUTOLOADED FALSE
#define MEMBLOCKQ_MAXLENGTH (16*1024*1024)
+/* Can only be used in main context */
+#define IS_ACTIVE(u) ((pa_source_get_state((u)->source) == PA_SOURCE_RUNNING) && \
+ (pa_sink_get_state((u)->sink) == PA_SINK_RUNNING))
+
/* This module creates a new (virtual) source and sink.
*
* The data sent to the new sink is kept in a memblockq before being
pa_bool_t need_realign;
/* to wakeup the source I/O thread */
- pa_bool_t in_push;
pa_asyncmsgq *asyncmsgq;
pa_rtpoll_item *rtpoll_item_read, *rtpoll_item_write;
pa_atomic_t request_resync;
- int active_mask;
pa_time_event *time_event;
pa_usec_t adjust_time;
+ int adjust_threshold;
FILE *captured_file;
FILE *played_file;
"sink_properties",
"sink_master",
"adjust_time",
+ "adjust_threshold",
"format",
"rate",
"channels",
pa_assert(u->time_event == e);
pa_assert_ctl_context();
- if (u->active_mask != 3)
+ if (!IS_ACTIVE(u))
return;
/* update our snapshots */
new_rate = base_rate;
}
else {
- if (diff_time > 1000) {
+ if (diff_time > u->adjust_threshold) {
/* diff too big, quickly adjust */
pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->source_output), SOURCE_OUTPUT_MESSAGE_APPLY_DIFF_TIME,
NULL, diff_time, NULL, NULL);
!PA_SOURCE_OUTPUT_IS_LINKED(pa_source_output_get_state(u->source_output)))
return 0;
- pa_log_debug("Source state %d %d", state, u->active_mask);
-
if (state == PA_SOURCE_RUNNING) {
/* restart timer when both sink and source are active */
- u->active_mask |= 1;
- if (u->active_mask == 3)
+ if (IS_ACTIVE(u) && u->adjust_time)
pa_core_rttime_restart(u->core, u->time_event, pa_rtclock_now() + u->adjust_time);
pa_atomic_store(&u->request_resync, 1);
pa_source_output_cork(u->source_output, FALSE);
} else if (state == PA_SOURCE_SUSPENDED) {
- u->active_mask &= ~1;
pa_source_output_cork(u->source_output, TRUE);
}
+
return 0;
}
!PA_SINK_INPUT_IS_LINKED(pa_sink_input_get_state(u->sink_input)))
return 0;
- pa_log_debug("Sink state %d %d", state, u->active_mask);
-
if (state == PA_SINK_RUNNING) {
/* restart timer when both sink and source are active */
- u->active_mask |= 2;
- if (u->active_mask == 3)
+ if (IS_ACTIVE(u) && u->adjust_time)
pa_core_rttime_restart(u->core, u->time_event, pa_rtclock_now() + u->adjust_time);
pa_atomic_store(&u->request_resync, 1);
pa_sink_input_cork(u->sink_input, FALSE);
} else if (state == PA_SINK_SUSPENDED) {
- u->active_mask &= ~2;
pa_sink_input_cork(u->sink_input, TRUE);
}
+
return 0;
}
/* Called from input thread context */
static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
struct userdata *u;
- size_t rlen, plen;
+ size_t rlen, plen, to_skip;
+ pa_memchunk rchunk, pchunk;
pa_source_output_assert_ref(o);
pa_source_output_assert_io_context(o);
return;
}
- /* handle queued messages */
- u->in_push = TRUE;
+ if (PA_UNLIKELY(u->source->thread_info.state != PA_SOURCE_RUNNING ||
+ u->sink->thread_info.state != PA_SINK_RUNNING)) {
+ pa_source_post(u->source, chunk);
+ return;
+ }
+
+ /* handle queued messages, do any message sending of our own */
while (pa_asyncmsgq_process_one(u->asyncmsgq) > 0)
;
- u->in_push = FALSE;
+ pa_memblockq_push_align(u->source_memblockq, chunk);
+
+ rlen = pa_memblockq_get_length(u->source_memblockq);
+ plen = pa_memblockq_get_length(u->sink_memblockq);
+
+ /* Let's not do anything else till we have enough data to process */
+ if (rlen < u->blocksize)
+ return;
+
+ /* See if we need to drop samples in order to sync */
if (pa_atomic_cmpxchg (&u->request_resync, 1, 0)) {
do_resync(u);
}
- pa_memblockq_push_align(u->source_memblockq, chunk);
+ /* Okay, skip cancellation for skipped source samples if needed. */
+ if (PA_UNLIKELY(u->source_skip)) {
+ /* The slightly tricky bit here is that we drop all but modulo
+ * blocksize bytes and then adjust for that last bit on the sink side.
+ * We do this because the source data is coming at a fixed rate, which
+ * means the only way to try to catch up is drop sink samples and let
+ * the canceller cope up with this. */
+ to_skip = rlen >= u->source_skip ? u->source_skip : rlen;
+ to_skip -= to_skip % u->blocksize;
+
+ if (to_skip) {
+ pa_memblockq_peek_fixed_size(u->source_memblockq, to_skip, &rchunk);
+ pa_source_post(u->source, &rchunk);
+
+ pa_memblock_unref(rchunk.memblock);
+ pa_memblockq_drop(u->source_memblockq, u->blocksize);
+
+ rlen -= to_skip;
+ u->source_skip -= to_skip;
+ }
- rlen = pa_memblockq_get_length(u->source_memblockq);
- plen = pa_memblockq_get_length(u->sink_memblockq);
+ if (rlen && u->source_skip % u->blocksize) {
+ u->sink_skip += u->blocksize - (u->source_skip % u->blocksize);
+ u->source_skip -= (u->source_skip % u->blocksize);
+ }
+ }
- while (rlen >= u->blocksize) {
- pa_memchunk rchunk, pchunk;
+ /* And for the sink, these samples have been played back already, so we can
+ * just drop them and get on with it. */
+ if (PA_UNLIKELY(u->sink_skip)) {
+ to_skip = plen >= u->sink_skip ? u->sink_skip : plen;
+
+ pa_memblockq_drop(u->sink_memblockq, to_skip);
+
+ plen -= to_skip;
+ u->sink_skip -= to_skip;
+ }
+ while (rlen >= u->blocksize) {
/* take fixed block from recorded samples */
pa_memblockq_peek_fixed_size(u->source_memblockq, u->blocksize, &rchunk);
- if (plen > u->blocksize && u->source_skip == 0) {
+ if (plen > u->blocksize) {
uint8_t *rdata, *pdata, *cdata;
pa_memchunk cchunk;
int unused;
- if (u->sink_skip) {
- size_t to_skip;
-
- if (u->sink_skip > plen)
- to_skip = plen;
- else
- to_skip = u->sink_skip;
-
- pa_memblockq_drop(u->sink_memblockq, to_skip);
- plen -= to_skip;
-
- u->sink_skip -= to_skip;
- }
-
- if (plen > u->blocksize && u->sink_skip == 0) {
+ if (plen > u->blocksize) {
/* take fixed block from played samples */
pa_memblockq_peek_fixed_size(u->sink_memblockq, u->blocksize, &pchunk);
pa_memblockq_drop(u->source_memblockq, u->blocksize);
rlen -= u->blocksize;
-
- if (u->source_skip) {
- if (u->source_skip > u->blocksize) {
- u->source_skip -= u->blocksize;
- }
- else {
- u->sink_skip += (u->blocksize - u->source_skip);
- u->source_skip = 0;
- }
- }
}
}
pa_source_output_assert_io_context(u->source_output);
- if (PA_SOURCE_IS_OPENED(u->source_output->source->thread_info.state))
+ if (u->source_output->source->thread_info.state == PA_SOURCE_RUNNING)
pa_memblockq_push_align(u->sink_memblockq, chunk);
else
pa_memblockq_flush_write(u->sink_memblockq, TRUE);
return PA_ECHO_CANCELLER_SPEEX;
else if (pa_streq(method, "adrian"))
return PA_ECHO_CANCELLER_ADRIAN;
+#ifdef HAVE_WEBRTC
+ else if (pa_streq(method, "webrtc"))
+ return PA_ECHO_CANCELLER_WEBRTC;
+#endif
else
return PA_ECHO_CANCELLER_INVALID;
}
+/* Common initialisation bits between module-echo-cancel and the standalone test program */
+static int init_common(pa_modargs *ma, struct userdata *u, pa_sample_spec *source_ss, pa_channel_map *source_map) {
+ pa_echo_canceller_method_t ec_method;
+
+ if (pa_modargs_get_sample_spec_and_channel_map(ma, source_ss, source_map, PA_CHANNEL_MAP_DEFAULT) < 0) {
+ pa_log("Invalid sample format specification or channel map");
+ goto fail;
+ }
+
+ u->ec = pa_xnew0(pa_echo_canceller, 1);
+ if (!u->ec) {
+ pa_log("Failed to alloc echo canceller");
+ goto fail;
+ }
+
+ if ((ec_method = get_ec_method_from_string(pa_modargs_get_value(ma, "aec_method", DEFAULT_ECHO_CANCELLER))) < 0) {
+ pa_log("Invalid echo canceller implementation");
+ goto fail;
+ }
+
+ u->ec->init = ec_table[ec_method].init;
+ u->ec->run = ec_table[ec_method].run;
+ u->ec->done = ec_table[ec_method].done;
+
+ return 0;
+
+fail:
+ return -1;
+}
+
+
int pa__init(pa_module*m) {
struct userdata *u;
pa_sample_spec source_ss, sink_ss;
pa_source_new_data source_data;
pa_sink_new_data sink_data;
pa_memchunk silence;
- pa_echo_canceller_method_t ec_method;
- uint32_t adjust_time_sec;
+ uint32_t temp;
pa_bool_t use_volume_sharing = TRUE;
pa_assert(m);
}
pa_assert(sink_master);
+ if (source_master->monitor_of == sink_master) {
+ pa_log("Can't cancel echo between a sink and its monitor");
+ goto fail;
+ }
+
source_ss = source_master->sample_spec;
source_ss.rate = DEFAULT_RATE;
source_ss.channels = DEFAULT_CHANNELS;
pa_channel_map_init_auto(&source_map, source_ss.channels, PA_CHANNEL_MAP_DEFAULT);
- if (pa_modargs_get_sample_spec_and_channel_map(ma, &source_ss, &source_map, PA_CHANNEL_MAP_DEFAULT) < 0) {
- pa_log("Invalid sample format specification or channel map");
- goto fail;
- }
sink_ss = sink_master->sample_spec;
sink_map = sink_master->channel_map;
m->userdata = u;
u->dead = FALSE;
- u->ec = pa_xnew0(pa_echo_canceller, 1);
- if (!u->ec) {
- pa_log("Failed to alloc echo canceller");
- goto fail;
- }
-
- if ((ec_method = get_ec_method_from_string(pa_modargs_get_value(ma, "aec_method", DEFAULT_ECHO_CANCELLER))) < 0) {
- pa_log("Invalid echo canceller implementation");
+ temp = DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC;
+ if (pa_modargs_get_value_u32(ma, "adjust_time", &temp) < 0) {
+ pa_log("Failed to parse adjust_time value");
goto fail;
}
- u->ec->init = ec_table[ec_method].init;
- u->ec->run = ec_table[ec_method].run;
- u->ec->done = ec_table[ec_method].done;
+ if (temp != DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC)
+ u->adjust_time = temp * PA_USEC_PER_SEC;
+ else
+ u->adjust_time = DEFAULT_ADJUST_TIME_USEC;
- adjust_time_sec = DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC;
- if (pa_modargs_get_value_u32(ma, "adjust_time", &adjust_time_sec) < 0) {
- pa_log("Failed to parse adjust_time value");
+ temp = DEFAULT_ADJUST_TOLERANCE / PA_USEC_PER_MSEC;
+ if (pa_modargs_get_value_u32(ma, "adjust_threshold", &temp) < 0) {
+ pa_log("Failed to parse adjust_threshold value");
goto fail;
}
- if (adjust_time_sec != DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC)
- u->adjust_time = adjust_time_sec * PA_USEC_PER_SEC;
+ if (temp != DEFAULT_ADJUST_TOLERANCE / PA_USEC_PER_MSEC)
+ u->adjust_threshold = temp * PA_USEC_PER_MSEC;
else
- u->adjust_time = DEFAULT_ADJUST_TIME_USEC;
+ u->adjust_threshold = DEFAULT_ADJUST_TOLERANCE;
u->save_aec = DEFAULT_SAVE_AEC;
if (pa_modargs_get_value_boolean(ma, "save_aec", &u->save_aec) < 0) {
goto fail;
}
+ if (init_common(ma, u, &source_ss, &source_map))
+ goto fail;
+
u->asyncmsgq = pa_asyncmsgq_new(0);
u->need_realign = TRUE;
+
if (u->ec->init) {
if (!u->ec->init(u->core, u->ec, &source_ss, &source_map, &sink_ss, &sink_map, &u->blocksize, pa_modargs_get_value(ma, "aec_args", NULL))) {
pa_log("Failed to init AEC engine");
pa_sink_input_get_silence(u->sink_input, &silence);
- u->source_memblockq = pa_memblockq_new(0, MEMBLOCKQ_MAXLENGTH, 0,
- pa_frame_size(&source_ss), 1, 1, 0, &silence);
- u->sink_memblockq = pa_memblockq_new(0, MEMBLOCKQ_MAXLENGTH, 0,
- pa_frame_size(&sink_ss), 1, 1, 0, &silence);
+ u->source_memblockq = pa_memblockq_new("module-echo-cancel source_memblockq", 0, MEMBLOCKQ_MAXLENGTH, 0,
+ &source_ss, 1, 1, 0, &silence);
+ u->sink_memblockq = pa_memblockq_new("module-echo-cancel sink_memblockq", 0, MEMBLOCKQ_MAXLENGTH, 0,
+ &sink_ss, 1, 1, 0, &silence);
pa_memblock_unref(silence.memblock);
goto fail;
}
- /* our source and sink are not suspended when we create them */
- u->active_mask = 3;
-
if (u->adjust_time > 0)
u->time_event = pa_core_rttime_new(m->core, pa_rtclock_now() + u->adjust_time, time_callback, u);
if (u->asyncmsgq)
pa_asyncmsgq_unref(u->asyncmsgq);
+ if (u->save_aec) {
+ if (u->played_file)
+ fclose(u->played_file);
+ if (u->captured_file)
+ fclose(u->captured_file);
+ if (u->canceled_file)
+ fclose(u->canceled_file);
+ }
+
pa_xfree(u);
}
+
+#ifdef ECHO_CANCEL_TEST
+/*
+ * Stand-alone test program for running in the canceller on pre-recorded files.
+ */
+int main(int argc, char* argv[]) {
+ struct userdata u;
+ pa_sample_spec source_ss, sink_ss;
+ pa_channel_map source_map, sink_map;
+ pa_modargs *ma = NULL;
+ uint8_t *rdata = NULL, *pdata = NULL, *cdata = NULL;
+ int ret = 0, unused;
+
+ pa_memzero(&u, sizeof(u));
+
+ if (argc < 4 || argc > 6) {
+ goto usage;
+ }
+
+ u.ec = pa_xnew0(pa_echo_canceller, 1);
+ if (!u.ec) {
+ pa_log("Failed to alloc echo canceller");
+ goto fail;
+ }
+
+ u.captured_file = fopen(argv[2], "r");
+ if (u.captured_file == NULL) {
+ perror ("fopen failed");
+ goto fail;
+ }
+ u.played_file = fopen(argv[1], "r");
+ if (u.played_file == NULL) {
+ perror ("fopen failed");
+ goto fail;
+ }
+ u.canceled_file = fopen(argv[3], "wb");
+ if (u.canceled_file == NULL) {
+ perror ("fopen failed");
+ goto fail;
+ }
+
+ u.core = pa_xnew0(pa_core, 1);
+ u.core->cpu_info.cpu_type = PA_CPU_X86;
+ u.core->cpu_info.flags.x86 |= PA_CPU_X86_SSE;
+
+ if (!(ma = pa_modargs_new(argc > 4 ? argv[4] : NULL, valid_modargs))) {
+ pa_log("Failed to parse module arguments.");
+ goto fail;
+ }
+
+ source_ss.format = PA_SAMPLE_S16LE;
+ source_ss.rate = DEFAULT_RATE;
+ source_ss.channels = DEFAULT_CHANNELS;
+ pa_channel_map_init_auto(&source_map, source_ss.channels, PA_CHANNEL_MAP_DEFAULT);
+
+ init_common(ma, &u, &source_ss, &source_map);
+
+ if (!u.ec->init(u.core, u.ec, &source_ss, &source_map, &sink_ss, &sink_map, &u.blocksize,
+ (argc > 4) ? argv[5] : NULL )) {
+ pa_log("Failed to init AEC engine");
+ goto fail;
+ }
+
+ rdata = pa_xmalloc(u.blocksize);
+ pdata = pa_xmalloc(u.blocksize);
+ cdata = pa_xmalloc(u.blocksize);
+
+ while (fread(rdata, u.blocksize, 1, u.captured_file) > 0) {
+ if (fread(pdata, u.blocksize, 1, u.played_file) == 0) {
+ perror("played file ended before captured file");
+ break;
+ }
+
+ u.ec->run(u.ec, rdata, pdata, cdata);
+
+ unused = fwrite(cdata, u.blocksize, 1, u.canceled_file);
+ }
+
+ u.ec->done(u.ec);
+
+ fclose(u.captured_file);
+ fclose(u.played_file);
+ fclose(u.canceled_file);
+
+out:
+ pa_xfree(rdata);
+ pa_xfree(pdata);
+ pa_xfree(cdata);
+
+ pa_xfree(u.ec);
+ pa_xfree(u.core);
+
+ if (ma)
+ pa_modargs_free(ma);
+
+ return ret;
+
+usage:
+ pa_log("Usage: %s play_file rec_file out_file [module args] [aec_args]",argv[0]);
+
+fail:
+ ret = -1;
+ goto out;
+}
+#endif /* ECHO_CANCEL_TEST */