]> code.delx.au - pulseaudio/blobdiff - src/modules/echo-cancel/module-echo-cancel.c
echo-cancel: Add the WebRTC echo canceller
[pulseaudio] / src / modules / echo-cancel / module-echo-cancel.c
index 62905bdd8ed12edbe32df2721ea677ea08cba843..7360b270d7609f51181c6d0092b77dfe1f7844fe 100644 (file)
@@ -66,6 +66,7 @@ PA_MODULE_USAGE(
           "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> "
@@ -74,6 +75,7 @@ PA_MODULE_USAGE(
           "aec_args=<parameters for the AEC engine> "
           "save_aec=<save AEC data in /tmp> "
           "autoloaded=<set if this module is being loaded automatically> "
+          "use_volume_sharing=<yes or no> "
         ));
 
 /* NOTE: Make sure the enum and ec_table are maintained in the correct order */
@@ -81,6 +83,9 @@ typedef enum {
     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"
@@ -98,16 +103,29 @@ static const pa_echo_canceller ec_table[] = {
         .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_SAVE_AEC 0
+#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
@@ -156,7 +174,8 @@ struct userdata {
     pa_module *module;
 
     pa_bool_t autoloaded;
-    uint32_t save_aec;
+    pa_bool_t dead;
+    pa_bool_t save_aec;
 
     pa_echo_canceller *ec;
     uint32_t blocksize;
@@ -164,7 +183,6 @@ struct userdata {
     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;
 
@@ -184,9 +202,9 @@ struct userdata {
 
     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;
@@ -203,6 +221,7 @@ static const char* const valid_modargs[] = {
     "sink_properties",
     "sink_master",
     "adjust_time",
+    "adjust_threshold",
     "format",
     "rate",
     "channels",
@@ -211,6 +230,7 @@ static const char* const valid_modargs[] = {
     "aec_args",
     "save_aec",
     "autoloaded",
+    "use_volume_sharing",
     NULL
 };
 
@@ -272,7 +292,7 @@ static void time_callback(pa_mainloop_api *a, pa_time_event *e, const struct tim
     pa_assert(u->time_event == e);
     pa_assert_ctl_context();
 
-    if (u->active_mask != 3)
+    if (!IS_ACTIVE(u))
         return;
 
     /* update our snapshots */
@@ -295,7 +315,7 @@ static void time_callback(pa_mainloop_api *a, pa_time_event *e, const struct tim
         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);
@@ -397,20 +417,17 @@ static int source_set_state_cb(pa_source *s, pa_source_state_t state) {
         !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;
 }
 
@@ -425,20 +442,17 @@ static int sink_set_state_cb(pa_sink *s, pa_sink_state_t state) {
         !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;
 }
 
@@ -639,7 +653,8 @@ static void do_resync(struct userdata *u) {
 /* 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);
@@ -650,46 +665,78 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk)
         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);
 
@@ -705,9 +752,9 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk)
 
                 if (u->save_aec) {
                     if (u->captured_file)
-                        fwrite(rdata, 1, u->blocksize, u->captured_file);
+                        unused = fwrite(rdata, 1, u->blocksize, u->captured_file);
                     if (u->played_file)
-                        fwrite(pdata, 1, u->blocksize, u->played_file);
+                        unused = fwrite(pdata, 1, u->blocksize, u->played_file);
                 }
 
                 /* perform echo cancellation */
@@ -715,7 +762,7 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk)
 
                 if (u->save_aec) {
                     if (u->canceled_file)
-                        fwrite(cdata, 1, u->blocksize, u->canceled_file);
+                        unused = fwrite(cdata, 1, u->blocksize, u->canceled_file);
                 }
 
                 pa_memblock_release(cchunk.memblock);
@@ -741,16 +788,6 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk)
 
         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;
-            }
-        }
     }
 }
 
@@ -847,7 +884,7 @@ static int source_output_process_msg_cb(pa_msgobject *obj, int code, void *data,
 
             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);
@@ -1162,6 +1199,8 @@ static void source_output_kill_cb(pa_source_output *o) {
     pa_assert_ctl_context();
     pa_assert_se(u = o->userdata);
 
+    u->dead = TRUE;
+
     /* The order here matters! We first kill the source output, followed
      * by the source. That means the source callbacks must be protected
      * against an unconnected source output! */
@@ -1186,6 +1225,8 @@ static void sink_input_kill_cb(pa_sink_input *i) {
     pa_sink_input_assert_ref(i);
     pa_assert_se(u = i->userdata);
 
+    u->dead = TRUE;
+
     /* The order here matters! We first kill the sink input, followed
      * by the sink. That means the sink callbacks must be protected
      * against an unconnected sink input! */
@@ -1211,6 +1252,9 @@ static pa_bool_t source_output_may_move_to_cb(pa_source_output *o, pa_source *de
     pa_assert_ctl_context();
     pa_assert_se(u = o->userdata);
 
+    if (u->dead)
+        return FALSE;
+
     return (u->source != dest) && (u->sink != dest->monitor_of);
 }
 
@@ -1221,6 +1265,9 @@ static pa_bool_t sink_input_may_move_to_cb(pa_sink_input *i, pa_sink *dest) {
     pa_sink_input_assert_ref(i);
     pa_assert_se(u = i->userdata);
 
+    if (u->dead)
+        return FALSE;
+
     return u->sink != dest;
 }
 
@@ -1304,10 +1351,45 @@ static pa_echo_canceller_method_t get_ec_method_from_string(const char *method)
         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;
@@ -1320,8 +1402,8 @@ int pa__init(pa_module*m) {
     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);
 
@@ -1342,18 +1424,24 @@ int pa__init(pa_module*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;
 
+    if (pa_modargs_get_value_boolean(ma, "use_volume_sharing", &use_volume_sharing) < 0) {
+        pa_log("use_volume_sharing= expects a boolean argument");
+        goto fail;
+    }
+
     u = pa_xnew0(struct userdata, 1);
     if (!u) {
         pa_log("Failed to alloc userdata");
@@ -1362,35 +1450,32 @@ int pa__init(pa_module*m) {
     u->core = m->core;
     u->module = m;
     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_u32(ma, "save_aec", &u->save_aec) < 0) {
+    if (pa_modargs_get_value_boolean(ma, "save_aec", &u->save_aec) < 0) {
         pa_log("Failed to parse save_aec value");
         goto fail;
     }
@@ -1401,8 +1486,12 @@ int pa__init(pa_module*m) {
         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");
@@ -1437,8 +1526,8 @@ int pa__init(pa_module*m) {
         pa_proplist_setf(source_data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Echo-Cancel Source %s on %s", source_data.name, z ? z : source_master->name);
     }
 
-    u->source = pa_source_new(m->core, &source_data,
-                          (source_master->flags & (PA_SOURCE_LATENCY|PA_SOURCE_DYNAMIC_LATENCY)));
+    u->source = pa_source_new(m->core, &source_data, (source_master->flags & (PA_SOURCE_LATENCY | PA_SOURCE_DYNAMIC_LATENCY))
+                                                     | (use_volume_sharing ? PA_SOURCE_SHARE_VOLUME_WITH_MASTER : 0));
     pa_source_new_data_done(&source_data);
 
     if (!u->source) {
@@ -1449,11 +1538,13 @@ int pa__init(pa_module*m) {
     u->source->parent.process_msg = source_process_msg_cb;
     u->source->set_state = source_set_state_cb;
     u->source->update_requested_latency = source_update_requested_latency_cb;
-    pa_source_enable_decibel_volume(u->source, TRUE);
-    pa_source_set_get_volume_callback(u->source, source_get_volume_cb);
-    pa_source_set_set_volume_callback(u->source, source_set_volume_cb);
     pa_source_set_get_mute_callback(u->source, source_get_mute_cb);
     pa_source_set_set_mute_callback(u->source, source_set_mute_cb);
+    if (!use_volume_sharing) {
+        pa_source_set_get_volume_callback(u->source, source_get_volume_cb);
+        pa_source_set_set_volume_callback(u->source, source_set_volume_cb);
+        pa_source_enable_decibel_volume(u->source, TRUE);
+    }
     u->source->userdata = u;
 
     pa_source_set_asyncmsgq(u->source, source_master->asyncmsgq);
@@ -1485,8 +1576,8 @@ int pa__init(pa_module*m) {
         pa_proplist_setf(sink_data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Echo-Cancel Sink %s on %s", sink_data.name, z ? z : sink_master->name);
     }
 
-    u->sink = pa_sink_new(m->core, &sink_data,
-                          (sink_master->flags & (PA_SINK_LATENCY|PA_SINK_DYNAMIC_LATENCY)));
+    u->sink = pa_sink_new(m->core, &sink_data, (sink_master->flags & (PA_SINK_LATENCY | PA_SINK_DYNAMIC_LATENCY))
+                                               | (use_volume_sharing ? PA_SINK_SHARE_VOLUME_WITH_MASTER : 0));
     pa_sink_new_data_done(&sink_data);
 
     if (!u->sink) {
@@ -1498,9 +1589,11 @@ int pa__init(pa_module*m) {
     u->sink->set_state = sink_set_state_cb;
     u->sink->update_requested_latency = sink_update_requested_latency_cb;
     u->sink->request_rewind = sink_request_rewind_cb;
-    pa_sink_enable_decibel_volume(u->sink, TRUE);
-    pa_sink_set_set_volume_callback(u->sink, sink_set_volume_cb);
     pa_sink_set_set_mute_callback(u->sink, sink_set_mute_cb);
+    if (!use_volume_sharing) {
+        pa_sink_set_set_volume_callback(u->sink, sink_set_volume_cb);
+        pa_sink_enable_decibel_volume(u->sink, TRUE);
+    }
     u->sink->userdata = u;
 
     pa_sink_set_asyncmsgq(u->sink, sink_master->asyncmsgq);
@@ -1574,7 +1667,8 @@ int pa__init(pa_module*m) {
     u->sink_input->state_change = sink_input_state_change_cb;
     u->sink_input->may_move_to = sink_input_may_move_to_cb;
     u->sink_input->moving = sink_input_moving_cb;
-    u->sink_input->volume_changed = sink_input_volume_changed_cb;
+    if (!use_volume_sharing)
+        u->sink_input->volume_changed = sink_input_volume_changed_cb;
     u->sink_input->mute_changed = sink_input_mute_changed_cb;
     u->sink_input->userdata = u;
 
@@ -1582,10 +1676,10 @@ int pa__init(pa_module*m) {
 
     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);
 
@@ -1594,9 +1688,6 @@ int pa__init(pa_module*m) {
         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);
 
@@ -1649,6 +1740,8 @@ void pa__done(pa_module*m) {
     if (!(u = m->userdata))
         return;
 
+    u->dead = TRUE;
+
     /* See comments in source_output_kill_cb() above regarding
      * destruction order! */
 
@@ -1690,5 +1783,119 @@ void pa__done(pa_module*m) {
     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 */