src/modules/echo-cancel/module-echo-cancel.c

   1 /***
   2     This file is part of PulseAudio.
   3
   4     Copyright 2010 Wim Taymans <wim.taymans@gmail.com>
   5
   6     Based on module-virtual-sink.c
   7              module-virtual-source.c
   8              module-loopback.c
   9
  10         Copyright 2010 Intel Corporation
  11         Contributor: Pierre-Louis Bossart <pierre-louis.bossart@intel.com>
  12
  13     PulseAudio is free software; you can redistribute it and/or modify
  14     it under the terms of the GNU Lesser General Public License as published
  15     by the Free Software Foundation; either version 2.1 of the License,
  16     or (at your option) any later version.
  17
  18     PulseAudio is distributed in the hope that it will be useful, but
  19     WITHOUT ANY WARRANTY; without even the implied warranty of
  20     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  21     General Public License for more details.
  22
  23     You should have received a copy of the GNU Lesser General Public License
  24     along with PulseAudio; if not, write to the Free Software
  25     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  26     USA.
  27 ***/
  28
  29 #ifdef HAVE_CONFIG_H
  30 #include <config.h>
  31 #endif
  32
  33 #include <stdio.h>
  34
  35 #include "echo-cancel.h"
  36
  37 #include <pulse/xmalloc.h>
  38 #include <pulse/timeval.h>
  39 #include <pulse/rtclock.h>
  40
  41 #include <pulsecore/i18n.h>
  42 #include <pulsecore/atomic.h>
  43 #include <pulsecore/macro.h>
  44 #include <pulsecore/namereg.h>
  45 #include <pulsecore/sink.h>
  46 #include <pulsecore/module.h>
  47 #include <pulsecore/core-rtclock.h>
  48 #include <pulsecore/core-util.h>
  49 #include <pulsecore/modargs.h>
  50 #include <pulsecore/log.h>
  51 #include <pulsecore/rtpoll.h>
  52 #include <pulsecore/sample-util.h>
  53 #include <pulsecore/ltdl-helper.h>
  54
  55 #include "module-echo-cancel-symdef.h"
  56
  57 PA_MODULE_AUTHOR("Wim Taymans");
  58 PA_MODULE_DESCRIPTION("Echo Cancellation");
  59 PA_MODULE_VERSION(PACKAGE_VERSION);
  60 PA_MODULE_LOAD_ONCE(FALSE);
  61 PA_MODULE_USAGE(
  62         _("source_name=<name for the source> "
  63           "source_properties=<properties for the source> "
  64           "source_master=<name of source to filter> "
  65           "sink_name=<name for the sink> "
  66           "sink_properties=<properties for the sink> "
  67           "sink_master=<name of sink to filter> "
  68           "adjust_time=<how often to readjust rates in s> "
  69           "adjust_threshold=<how much drift to readjust after in ms> "
  70           "format=<sample format> "
  71           "rate=<sample rate> "
  72           "channels=<number of channels> "
  73           "channel_map=<channel map> "
  74           "aec_method=<implementation to use> "
  75           "aec_args=<parameters for the AEC engine> "
  76           "save_aec=<save AEC data in /tmp> "
  77           "autoloaded=<set if this module is being loaded automatically> "
  78           "use_volume_sharing=<yes or no> "
  79         ));
  80
  81 /* NOTE: Make sure the enum and ec_table are maintained in the correct order */
  82 typedef enum {
  83     PA_ECHO_CANCELLER_INVALID = -1,
  84     PA_ECHO_CANCELLER_SPEEX = 0,
  85     PA_ECHO_CANCELLER_ADRIAN,
  86 #ifdef HAVE_WEBRTC
  87     PA_ECHO_CANCELLER_WEBRTC,
  88 #endif
  89 } pa_echo_canceller_method_t;
  90
  91 #define DEFAULT_ECHO_CANCELLER "speex"
  92
  93 static const pa_echo_canceller ec_table[] = {
  94     {
  95         /* Speex */
  96         .init                   = pa_speex_ec_init,
  97         .run                    = pa_speex_ec_run,
  98         .done                   = pa_speex_ec_done,
  99     },
 100     {
 101         /* Adrian Andre's NLMS implementation */
 102         .init                   = pa_adrian_ec_init,
 103         .run                    = pa_adrian_ec_run,
 104         .done                   = pa_adrian_ec_done,
 105     },
 106 #ifdef HAVE_WEBRTC
 107     {
 108         /* WebRTC's audio processing engine */
 109         .init                   = pa_webrtc_ec_init,
 110         .run                    = pa_webrtc_ec_run,
 111         .done                   = pa_webrtc_ec_done,
 112     },
 113 #endif
 114 };
 115
 116 #define DEFAULT_RATE 32000
 117 #define DEFAULT_CHANNELS 1
 118 #define DEFAULT_ADJUST_TIME_USEC (1*PA_USEC_PER_SEC)
 119 #define DEFAULT_ADJUST_TOLERANCE (5*PA_USEC_PER_MSEC)
 120 #define DEFAULT_SAVE_AEC FALSE
 121 #define DEFAULT_AUTOLOADED FALSE
 122
 123 #define MEMBLOCKQ_MAXLENGTH (16*1024*1024)
 124
 125 /* Can only be used in main context */
 126 #define IS_ACTIVE(u) ((pa_source_get_state((u)->source) == PA_SOURCE_RUNNING) && \
 127                       (pa_sink_get_state((u)->sink) == PA_SINK_RUNNING))
 128
 129 /* This module creates a new (virtual) source and sink.
 130  *
 131  * The data sent to the new sink is kept in a memblockq before being
 132  * forwarded to the real sink_master.
 133  *
 134  * Data read from source_master is matched against the saved sink data and
 135  * echo canceled data is then pushed onto the new source.
 136  *
 137  * Both source and sink masters have their own threads to push/pull data
 138  * respectively. We however perform all our actions in the source IO thread.
 139  * To do this we send all played samples to the source IO thread where they
 140  * are then pushed into the memblockq.
 141  *
 142  * Alignment is performed in two steps:
 143  *
 144  * 1) when something happens that requires quick adjustment of the alignment of
 145  *    capture and playback samples, we perform a resync. This adjusts the
 146  *    position in the playback memblock to the requested sample. Quick
 147  *    adjustments include moving the playback samples before the capture
 148  *    samples (because else the echo canceler does not work) or when the
 149  *    playback pointer drifts too far away.
 150  *
 151  * 2) periodically check the difference between capture and playback. we use a
 152  *    low and high watermark for adjusting the alignment. playback should always
 153  *    be before capture and the difference should not be bigger than one frame
 154  *    size. We would ideally like to resample the sink_input but most driver
 155  *    don't give enough accuracy to be able to do that right now.
 156  */
 157
 158 struct snapshot {
 159     pa_usec_t sink_now;
 160     pa_usec_t sink_latency;
 161     size_t sink_delay;
 162     int64_t send_counter;
 163
 164     pa_usec_t source_now;
 165     pa_usec_t source_latency;
 166     size_t source_delay;
 167     int64_t recv_counter;
 168     size_t rlen;
 169     size_t plen;
 170 };
 171
 172 struct userdata {
 173     pa_core *core;
 174     pa_module *module;
 175
 176     pa_bool_t autoloaded;
 177     pa_bool_t dead;
 178     pa_bool_t save_aec;
 179
 180     pa_echo_canceller *ec;
 181     uint32_t blocksize;
 182
 183     pa_bool_t need_realign;
 184
 185     /* to wakeup the source I/O thread */
 186     pa_asyncmsgq *asyncmsgq;
 187     pa_rtpoll_item *rtpoll_item_read, *rtpoll_item_write;
 188
 189     pa_source *source;
 190     pa_bool_t source_auto_desc;
 191     pa_source_output *source_output;
 192     pa_memblockq *source_memblockq; /* echo canceler needs fixed sized chunks */
 193     size_t source_skip;
 194
 195     pa_sink *sink;
 196     pa_bool_t sink_auto_desc;
 197     pa_sink_input *sink_input;
 198     pa_memblockq *sink_memblockq;
 199     int64_t send_counter;          /* updated in sink IO thread */
 200     int64_t recv_counter;
 201     size_t sink_skip;
 202
 203     pa_atomic_t request_resync;
 204
 205     pa_time_event *time_event;
 206     pa_usec_t adjust_time;
 207     int adjust_threshold;
 208
 209     FILE *captured_file;
 210     FILE *played_file;
 211     FILE *canceled_file;
 212 };
 213
 214 static void source_output_snapshot_within_thread(struct userdata *u, struct snapshot *snapshot);
 215
 216 static const char* const valid_modargs[] = {
 217     "source_name",
 218     "source_properties",
 219     "source_master",
 220     "sink_name",
 221     "sink_properties",
 222     "sink_master",
 223     "adjust_time",
 224     "adjust_threshold",
 225     "format",
 226     "rate",
 227     "channels",
 228     "channel_map",
 229     "aec_method",
 230     "aec_args",
 231     "save_aec",
 232     "autoloaded",
 233     "use_volume_sharing",
 234     NULL
 235 };
 236
 237 enum {
 238     SOURCE_OUTPUT_MESSAGE_POST = PA_SOURCE_OUTPUT_MESSAGE_MAX,
 239     SOURCE_OUTPUT_MESSAGE_REWIND,
 240     SOURCE_OUTPUT_MESSAGE_LATENCY_SNAPSHOT,
 241     SOURCE_OUTPUT_MESSAGE_APPLY_DIFF_TIME
 242 };
 243
 244 enum {
 245     SINK_INPUT_MESSAGE_LATENCY_SNAPSHOT
 246 };
 247
 248 static int64_t calc_diff(struct userdata *u, struct snapshot *snapshot) {
 249     int64_t buffer, diff_time, buffer_latency;
 250
 251     /* get the number of samples between capture and playback */
 252     if (snapshot->plen > snapshot->rlen)
 253         buffer = snapshot->plen - snapshot->rlen;
 254     else
 255         buffer = 0;
 256
 257     buffer += snapshot->source_delay + snapshot->sink_delay;
 258
 259     /* add the amount of samples not yet transferred to the source context */
 260     if (snapshot->recv_counter <= snapshot->send_counter)
 261         buffer += (int64_t) (snapshot->send_counter - snapshot->recv_counter);
 262     else
 263         buffer += PA_CLIP_SUB(buffer, (int64_t) (snapshot->recv_counter - snapshot->send_counter));
 264
 265     /* convert to time */
 266     buffer_latency = pa_bytes_to_usec(buffer, &u->source_output->sample_spec);
 267
 268     /* capture and playback samples are perfectly aligned when diff_time is 0 */
 269     diff_time = (snapshot->sink_now + snapshot->sink_latency - buffer_latency) -
 270           (snapshot->source_now - snapshot->source_latency);
 271
 272     pa_log_debug("diff %lld (%lld - %lld + %lld) %lld %lld %lld %lld", (long long) diff_time,
 273         (long long) snapshot->sink_latency,
 274         (long long) buffer_latency, (long long) snapshot->source_latency,
 275         (long long) snapshot->source_delay, (long long) snapshot->sink_delay,
 276         (long long) (snapshot->send_counter - snapshot->recv_counter),
 277         (long long) (snapshot->sink_now - snapshot->source_now));
 278
 279     return diff_time;
 280 }
 281
 282 /* Called from main context */
 283 static void time_callback(pa_mainloop_api *a, pa_time_event *e, const struct timeval *t, void *userdata) {
 284     struct userdata *u = userdata;
 285     uint32_t old_rate, base_rate, new_rate;
 286     int64_t diff_time;
 287     /*size_t fs*/
 288     struct snapshot latency_snapshot;
 289
 290     pa_assert(u);
 291     pa_assert(a);
 292     pa_assert(u->time_event == e);
 293     pa_assert_ctl_context();
 294
 295     if (!IS_ACTIVE(u))
 296         return;
 297
 298     /* update our snapshots */
 299     pa_asyncmsgq_send(u->source_output->source->asyncmsgq, PA_MSGOBJECT(u->source_output), SOURCE_OUTPUT_MESSAGE_LATENCY_SNAPSHOT, &latency_snapshot, 0, NULL);
 300     pa_asyncmsgq_send(u->sink_input->sink->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_LATENCY_SNAPSHOT, &latency_snapshot, 0, NULL);
 301
 302     /* calculate drift between capture and playback */
 303     diff_time = calc_diff(u, &latency_snapshot);
 304
 305     /*fs = pa_frame_size(&u->source_output->sample_spec);*/
 306     old_rate = u->sink_input->sample_spec.rate;
 307     base_rate = u->source_output->sample_spec.rate;
 308
 309     if (diff_time < 0) {
 310         /* recording before playback, we need to adjust quickly. The echo
 311          * canceler does not work in this case. */
 312         pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->source_output), SOURCE_OUTPUT_MESSAGE_APPLY_DIFF_TIME,
 313             NULL, diff_time, NULL, NULL);
 314         /*new_rate = base_rate - ((pa_usec_to_bytes(-diff_time, &u->source_output->sample_spec) / fs) * PA_USEC_PER_SEC) / u->adjust_time;*/
 315         new_rate = base_rate;
 316     }
 317     else {
 318         if (diff_time > u->adjust_threshold) {
 319             /* diff too big, quickly adjust */
 320             pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->source_output), SOURCE_OUTPUT_MESSAGE_APPLY_DIFF_TIME,
 321                 NULL, diff_time, NULL, NULL);
 322         }
 323
 324         /* recording behind playback, we need to slowly adjust the rate to match */
 325         /*new_rate = base_rate + ((pa_usec_to_bytes(diff_time, &u->source_output->sample_spec) / fs) * PA_USEC_PER_SEC) / u->adjust_time;*/
 326
 327         /* assume equal samplerates for now */
 328         new_rate = base_rate;
 329     }
 330
 331     /* make sure we don't make too big adjustments because that sounds horrible */
 332     if (new_rate > base_rate * 1.1 || new_rate < base_rate * 0.9)
 333         new_rate = base_rate;
 334
 335     if (new_rate != old_rate) {
 336         pa_log_info("Old rate %lu Hz, new rate %lu Hz", (unsigned long) old_rate, (unsigned long) new_rate);
 337
 338         pa_sink_input_set_rate(u->sink_input, new_rate);
 339     }
 340
 341     pa_core_rttime_restart(u->core, u->time_event, pa_rtclock_now() + u->adjust_time);
 342 }
 343
 344 /* Called from source I/O thread context */
 345 static int source_process_msg_cb(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 346     struct userdata *u = PA_SOURCE(o)->userdata;
 347
 348     switch (code) {
 349
 350         case PA_SOURCE_MESSAGE_GET_LATENCY:
 351
 352             /* The source is _put() before the source output is, so let's
 353              * make sure we don't access it in that time. Also, the
 354              * source output is first shut down, the source second. */
 355             if (!PA_SOURCE_IS_LINKED(u->source->thread_info.state) ||
 356                 !PA_SOURCE_OUTPUT_IS_LINKED(u->source_output->thread_info.state)) {
 357                 *((pa_usec_t*) data) = 0;
 358                 return 0;
 359             }
 360
 361             *((pa_usec_t*) data) =
 362
 363                 /* Get the latency of the master source */
 364                 pa_source_get_latency_within_thread(u->source_output->source) +
 365                 /* Add the latency internal to our source output on top */
 366                 pa_bytes_to_usec(pa_memblockq_get_length(u->source_output->thread_info.delay_memblockq), &u->source_output->source->sample_spec) +
 367                 /* and the buffering we do on the source */
 368                 pa_bytes_to_usec(u->blocksize, &u->source_output->source->sample_spec);
 369
 370             return 0;
 371
 372     }
 373
 374     return pa_source_process_msg(o, code, data, offset, chunk);
 375 }
 376
 377 /* Called from sink I/O thread context */
 378 static int sink_process_msg_cb(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 379     struct userdata *u = PA_SINK(o)->userdata;
 380
 381     switch (code) {
 382
 383         case PA_SINK_MESSAGE_GET_LATENCY:
 384
 385             /* The sink is _put() before the sink input is, so let's
 386              * make sure we don't access it in that time. Also, the
 387              * sink input is first shut down, the sink second. */
 388             if (!PA_SINK_IS_LINKED(u->sink->thread_info.state) ||
 389                 !PA_SINK_INPUT_IS_LINKED(u->sink_input->thread_info.state)) {
 390                 *((pa_usec_t*) data) = 0;
 391                 return 0;
 392             }
 393
 394             *((pa_usec_t*) data) =
 395
 396                 /* Get the latency of the master sink */
 397                 pa_sink_get_latency_within_thread(u->sink_input->sink) +
 398
 399                 /* Add the latency internal to our sink input on top */
 400                 pa_bytes_to_usec(pa_memblockq_get_length(u->sink_input->thread_info.render_memblockq), &u->sink_input->sink->sample_spec);
 401
 402             return 0;
 403     }
 404
 405     return pa_sink_process_msg(o, code, data, offset, chunk);
 406 }
 407
 408
 409 /* Called from main context */
 410 static int source_set_state_cb(pa_source *s, pa_source_state_t state) {
 411     struct userdata *u;
 412
 413     pa_source_assert_ref(s);
 414     pa_assert_se(u = s->userdata);
 415
 416     if (!PA_SOURCE_IS_LINKED(state) ||
 417         !PA_SOURCE_OUTPUT_IS_LINKED(pa_source_output_get_state(u->source_output)))
 418         return 0;
 419
 420     if (state == PA_SOURCE_RUNNING) {
 421         /* restart timer when both sink and source are active */
 422         if (IS_ACTIVE(u) && u->adjust_time)
 423             pa_core_rttime_restart(u->core, u->time_event, pa_rtclock_now() + u->adjust_time);
 424
 425         pa_atomic_store(&u->request_resync, 1);
 426         pa_source_output_cork(u->source_output, FALSE);
 427     } else if (state == PA_SOURCE_SUSPENDED) {
 428         pa_source_output_cork(u->source_output, TRUE);
 429     }
 430
 431     return 0;
 432 }
 433
 434 /* Called from main context */
 435 static int sink_set_state_cb(pa_sink *s, pa_sink_state_t state) {
 436     struct userdata *u;
 437
 438     pa_sink_assert_ref(s);
 439     pa_assert_se(u = s->userdata);
 440
 441     if (!PA_SINK_IS_LINKED(state) ||
 442         !PA_SINK_INPUT_IS_LINKED(pa_sink_input_get_state(u->sink_input)))
 443         return 0;
 444
 445     if (state == PA_SINK_RUNNING) {
 446         /* restart timer when both sink and source are active */
 447         if (IS_ACTIVE(u) && u->adjust_time)
 448             pa_core_rttime_restart(u->core, u->time_event, pa_rtclock_now() + u->adjust_time);
 449
 450         pa_atomic_store(&u->request_resync, 1);
 451         pa_sink_input_cork(u->sink_input, FALSE);
 452     } else if (state == PA_SINK_SUSPENDED) {
 453         pa_sink_input_cork(u->sink_input, TRUE);
 454     }
 455
 456     return 0;
 457 }
 458
 459 /* Called from I/O thread context */
 460 static void source_update_requested_latency_cb(pa_source *s) {
 461     struct userdata *u;
 462
 463     pa_source_assert_ref(s);
 464     pa_assert_se(u = s->userdata);
 465
 466     if (!PA_SOURCE_IS_LINKED(u->source->thread_info.state) ||
 467         !PA_SOURCE_OUTPUT_IS_LINKED(u->source_output->thread_info.state))
 468         return;
 469
 470     pa_log_debug("Source update requested latency");
 471
 472     /* Just hand this one over to the master source */
 473     pa_source_output_set_requested_latency_within_thread(
 474             u->source_output,
 475             pa_source_get_requested_latency_within_thread(s));
 476 }
 477
 478 /* Called from I/O thread context */
 479 static void sink_update_requested_latency_cb(pa_sink *s) {
 480     struct userdata *u;
 481
 482     pa_sink_assert_ref(s);
 483     pa_assert_se(u = s->userdata);
 484
 485     if (!PA_SINK_IS_LINKED(u->sink->thread_info.state) ||
 486         !PA_SINK_INPUT_IS_LINKED(u->sink_input->thread_info.state))
 487         return;
 488
 489     pa_log_debug("Sink update requested latency");
 490
 491     /* Just hand this one over to the master sink */
 492     pa_sink_input_set_requested_latency_within_thread(
 493             u->sink_input,
 494             pa_sink_get_requested_latency_within_thread(s));
 495 }
 496
 497 /* Called from I/O thread context */
 498 static void sink_request_rewind_cb(pa_sink *s) {
 499     struct userdata *u;
 500
 501     pa_sink_assert_ref(s);
 502     pa_assert_se(u = s->userdata);
 503
 504     if (!PA_SINK_IS_LINKED(u->sink->thread_info.state) ||
 505         !PA_SINK_INPUT_IS_LINKED(u->sink_input->thread_info.state))
 506         return;
 507
 508     pa_log_debug("Sink request rewind %lld", (long long) s->thread_info.rewind_nbytes);
 509
 510     /* Just hand this one over to the master sink */
 511     pa_sink_input_request_rewind(u->sink_input,
 512                                  s->thread_info.rewind_nbytes, TRUE, FALSE, FALSE);
 513 }
 514
 515 /* Called from main context */
 516 static void source_set_volume_cb(pa_source *s) {
 517     struct userdata *u;
 518
 519     pa_source_assert_ref(s);
 520     pa_assert_se(u = s->userdata);
 521
 522     if (!PA_SOURCE_IS_LINKED(pa_source_get_state(s)) ||
 523         !PA_SOURCE_OUTPUT_IS_LINKED(pa_source_output_get_state(u->source_output)))
 524         return;
 525
 526     pa_source_output_set_volume(u->source_output, &s->real_volume, s->save_volume, TRUE);
 527 }
 528
 529 /* Called from main context */
 530 static void sink_set_volume_cb(pa_sink *s) {
 531     struct userdata *u;
 532
 533     pa_sink_assert_ref(s);
 534     pa_assert_se(u = s->userdata);
 535
 536     if (!PA_SINK_IS_LINKED(pa_sink_get_state(s)) ||
 537         !PA_SINK_INPUT_IS_LINKED(pa_sink_input_get_state(u->sink_input)))
 538         return;
 539
 540     pa_sink_input_set_volume(u->sink_input, &s->real_volume, s->save_volume, TRUE);
 541 }
 542
 543 static void source_get_volume_cb(pa_source *s) {
 544     struct userdata *u;
 545     pa_cvolume v;
 546
 547     pa_source_assert_ref(s);
 548     pa_assert_se(u = s->userdata);
 549
 550     if (!PA_SOURCE_IS_LINKED(pa_source_get_state(s)) ||
 551         !PA_SOURCE_OUTPUT_IS_LINKED(pa_source_output_get_state(u->source_output)))
 552         return;
 553
 554     pa_source_output_get_volume(u->source_output, &v, TRUE);
 555
 556     if (pa_cvolume_equal(&s->real_volume, &v))
 557         /* no change */
 558         return;
 559
 560     s->real_volume = v;
 561     pa_source_set_soft_volume(s, NULL);
 562 }
 563
 564 /* Called from main context */
 565 static void source_set_mute_cb(pa_source *s) {
 566     struct userdata *u;
 567
 568     pa_source_assert_ref(s);
 569     pa_assert_se(u = s->userdata);
 570
 571     if (!PA_SOURCE_IS_LINKED(pa_source_get_state(s)) ||
 572         !PA_SOURCE_OUTPUT_IS_LINKED(pa_source_output_get_state(u->source_output)))
 573         return;
 574
 575     pa_source_output_set_mute(u->source_output, s->muted, s->save_muted);
 576 }
 577
 578 /* Called from main context */
 579 static void sink_set_mute_cb(pa_sink *s) {
 580     struct userdata *u;
 581
 582     pa_sink_assert_ref(s);
 583     pa_assert_se(u = s->userdata);
 584
 585     if (!PA_SINK_IS_LINKED(pa_sink_get_state(s)) ||
 586         !PA_SINK_INPUT_IS_LINKED(pa_sink_input_get_state(u->sink_input)))
 587         return;
 588
 589     pa_sink_input_set_mute(u->sink_input, s->muted, s->save_muted);
 590 }
 591
 592 /* Called from main context */
 593 static void source_get_mute_cb(pa_source *s) {
 594     struct userdata *u;
 595
 596     pa_source_assert_ref(s);
 597     pa_assert_se(u = s->userdata);
 598
 599     if (!PA_SOURCE_IS_LINKED(pa_source_get_state(s)) ||
 600         !PA_SOURCE_OUTPUT_IS_LINKED(pa_source_output_get_state(u->source_output)))
 601         return;
 602
 603     pa_source_output_get_mute(u->source_output);
 604 }
 605
 606 /* must be called from the input thread context */
 607 static void apply_diff_time(struct userdata *u, int64_t diff_time) {
 608     int64_t diff;
 609
 610     if (diff_time < 0) {
 611         diff = pa_usec_to_bytes(-diff_time, &u->source_output->sample_spec);
 612
 613         if (diff > 0) {
 614             /* add some extra safety samples to compensate for jitter in the
 615              * timings */
 616             diff += 10 * pa_frame_size (&u->source_output->sample_spec);
 617
 618             pa_log("Playback after capture (%lld), drop sink %lld", (long long) diff_time, (long long) diff);
 619
 620             u->sink_skip = diff;
 621             u->source_skip = 0;
 622         }
 623     } else if (diff_time > 0) {
 624         diff = pa_usec_to_bytes(diff_time, &u->source_output->sample_spec);
 625
 626         if (diff > 0) {
 627             pa_log("playback too far ahead (%lld), drop source %lld", (long long) diff_time, (long long) diff);
 628
 629             u->source_skip = diff;
 630             u->sink_skip = 0;
 631         }
 632     }
 633 }
 634
 635 /* must be called from the input thread */
 636 static void do_resync(struct userdata *u) {
 637     int64_t diff_time;
 638     struct snapshot latency_snapshot;
 639
 640     pa_log("Doing resync");
 641
 642     /* update our snapshot */
 643     source_output_snapshot_within_thread(u, &latency_snapshot);
 644     pa_asyncmsgq_send(u->sink_input->sink->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_LATENCY_SNAPSHOT, &latency_snapshot, 0, NULL);
 645
 646     /* calculate drift between capture and playback */
 647     diff_time = calc_diff(u, &latency_snapshot);
 648
 649     /* and adjust for the drift */
 650     apply_diff_time(u, diff_time);
 651 }
 652
 653 /* Called from input thread context */
 654 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
 655     struct userdata *u;
 656     size_t rlen, plen, to_skip;
 657     pa_memchunk rchunk, pchunk;
 658
 659     pa_source_output_assert_ref(o);
 660     pa_source_output_assert_io_context(o);
 661     pa_assert_se(u = o->userdata);
 662
 663     if (!PA_SOURCE_OUTPUT_IS_LINKED(pa_source_output_get_state(u->source_output))) {
 664         pa_log("push when no link?");
 665         return;
 666     }
 667
 668     if (PA_UNLIKELY(u->source->thread_info.state != PA_SOURCE_RUNNING ||
 669                     u->sink->thread_info.state != PA_SINK_RUNNING)) {
 670         pa_source_post(u->source, chunk);
 671         return;
 672     }
 673
 674     /* handle queued messages, do any message sending of our own */
 675     while (pa_asyncmsgq_process_one(u->asyncmsgq) > 0)
 676         ;
 677
 678     pa_memblockq_push_align(u->source_memblockq, chunk);
 679
 680     rlen = pa_memblockq_get_length(u->source_memblockq);
 681     plen = pa_memblockq_get_length(u->sink_memblockq);
 682
 683     /* Let's not do anything else till we have enough data to process */
 684     if (rlen < u->blocksize)
 685         return;
 686
 687     /* See if we need to drop samples in order to sync */
 688     if (pa_atomic_cmpxchg (&u->request_resync, 1, 0)) {
 689         do_resync(u);
 690     }
 691
 692     /* Okay, skip cancellation for skipped source samples if needed. */
 693     if (PA_UNLIKELY(u->source_skip)) {
 694         /* The slightly tricky bit here is that we drop all but modulo
 695          * blocksize bytes and then adjust for that last bit on the sink side.
 696          * We do this because the source data is coming at a fixed rate, which
 697          * means the only way to try to catch up is drop sink samples and let
 698          * the canceller cope up with this. */
 699         to_skip = rlen >= u->source_skip ? u->source_skip : rlen;
 700         to_skip -= to_skip % u->blocksize;
 701
 702         if (to_skip) {
 703             pa_memblockq_peek_fixed_size(u->source_memblockq, to_skip, &rchunk);
 704             pa_source_post(u->source, &rchunk);
 705
 706             pa_memblock_unref(rchunk.memblock);
 707             pa_memblockq_drop(u->source_memblockq, u->blocksize);
 708
 709             rlen -= to_skip;
 710             u->source_skip -= to_skip;
 711         }
 712
 713         if (rlen && u->source_skip % u->blocksize) {
 714             u->sink_skip += u->blocksize - (u->source_skip % u->blocksize);
 715             u->source_skip -= (u->source_skip % u->blocksize);
 716         }
 717     }
 718
 719     /* And for the sink, these samples have been played back already, so we can
 720      * just drop them and get on with it. */
 721     if (PA_UNLIKELY(u->sink_skip)) {
 722         to_skip = plen >= u->sink_skip ? u->sink_skip : plen;
 723
 724         pa_memblockq_drop(u->sink_memblockq, to_skip);
 725
 726         plen -= to_skip;
 727         u->sink_skip -= to_skip;
 728     }
 729
 730     while (rlen >= u->blocksize) {
 731         /* take fixed block from recorded samples */
 732         pa_memblockq_peek_fixed_size(u->source_memblockq, u->blocksize, &rchunk);
 733
 734         if (plen > u->blocksize) {
 735             uint8_t *rdata, *pdata, *cdata;
 736             pa_memchunk cchunk;
 737             int unused;
 738
 739             if (plen > u->blocksize) {
 740                 /* take fixed block from played samples */
 741                 pa_memblockq_peek_fixed_size(u->sink_memblockq, u->blocksize, &pchunk);
 742
 743                 rdata = pa_memblock_acquire(rchunk.memblock);
 744                 rdata += rchunk.index;
 745                 pdata = pa_memblock_acquire(pchunk.memblock);
 746                 pdata += pchunk.index;
 747
 748                 cchunk.index = 0;
 749                 cchunk.length = u->blocksize;
 750                 cchunk.memblock = pa_memblock_new(u->source->core->mempool, cchunk.length);
 751                 cdata = pa_memblock_acquire(cchunk.memblock);
 752
 753                 if (u->save_aec) {
 754                     if (u->captured_file)
 755                         unused = fwrite(rdata, 1, u->blocksize, u->captured_file);
 756                     if (u->played_file)
 757                         unused = fwrite(pdata, 1, u->blocksize, u->played_file);
 758                 }
 759
 760                 /* perform echo cancellation */
 761                 u->ec->run(u->ec, rdata, pdata, cdata);
 762
 763                 if (u->save_aec) {
 764                     if (u->canceled_file)
 765                         unused = fwrite(cdata, 1, u->blocksize, u->canceled_file);
 766                 }
 767
 768                 pa_memblock_release(cchunk.memblock);
 769                 pa_memblock_release(pchunk.memblock);
 770                 pa_memblock_release(rchunk.memblock);
 771
 772                 /* drop consumed sink samples */
 773                 pa_memblockq_drop(u->sink_memblockq, u->blocksize);
 774                 pa_memblock_unref(pchunk.memblock);
 775
 776                 pa_memblock_unref(rchunk.memblock);
 777                 /* the filtered samples now become the samples from our
 778                  * source */
 779                 rchunk = cchunk;
 780
 781                 plen -= u->blocksize;
 782             }
 783         }
 784
 785         /* forward the (echo-canceled) data to the virtual source */
 786         pa_source_post(u->source, &rchunk);
 787         pa_memblock_unref(rchunk.memblock);
 788
 789         pa_memblockq_drop(u->source_memblockq, u->blocksize);
 790         rlen -= u->blocksize;
 791     }
 792 }
 793
 794 /* Called from I/O thread context */
 795 static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk) {
 796     struct userdata *u;
 797
 798     pa_sink_input_assert_ref(i);
 799     pa_assert(chunk);
 800     pa_assert_se(u = i->userdata);
 801
 802     if (u->sink->thread_info.rewind_requested)
 803         pa_sink_process_rewind(u->sink, 0);
 804
 805     pa_sink_render_full(u->sink, nbytes, chunk);
 806
 807     if (i->thread_info.underrun_for > 0) {
 808         pa_log_debug("Handling end of underrun.");
 809         pa_atomic_store(&u->request_resync, 1);
 810     }
 811
 812     /* let source thread handle the chunk. pass the sample count as well so that
 813      * the source IO thread can update the right variables. */
 814     pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->source_output), SOURCE_OUTPUT_MESSAGE_POST,
 815         NULL, 0, chunk, NULL);
 816     u->send_counter += chunk->length;
 817
 818     return 0;
 819 }
 820
 821 /* Called from input thread context */
 822 static void source_output_process_rewind_cb(pa_source_output *o, size_t nbytes) {
 823     struct userdata *u;
 824
 825     pa_source_output_assert_ref(o);
 826     pa_source_output_assert_io_context(o);
 827     pa_assert_se(u = o->userdata);
 828
 829     pa_source_process_rewind(u->source, nbytes);
 830
 831     /* go back on read side, we need to use older sink data for this */
 832     pa_memblockq_rewind(u->sink_memblockq, nbytes);
 833
 834     /* manipulate write index */
 835     pa_memblockq_seek(u->source_memblockq, -nbytes, PA_SEEK_RELATIVE, TRUE);
 836
 837     pa_log_debug("Source rewind (%lld) %lld", (long long) nbytes,
 838         (long long) pa_memblockq_get_length (u->source_memblockq));
 839 }
 840
 841 /* Called from I/O thread context */
 842 static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes) {
 843     struct userdata *u;
 844
 845     pa_sink_input_assert_ref(i);
 846     pa_assert_se(u = i->userdata);
 847
 848     pa_log_debug("Sink process rewind %lld", (long long) nbytes);
 849
 850     pa_sink_process_rewind(u->sink, nbytes);
 851
 852     pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->source_output), SOURCE_OUTPUT_MESSAGE_REWIND, NULL, (int64_t) nbytes, NULL, NULL);
 853     u->send_counter -= nbytes;
 854 }
 855
 856 static void source_output_snapshot_within_thread(struct userdata *u, struct snapshot *snapshot) {
 857     size_t delay, rlen, plen;
 858     pa_usec_t now, latency;
 859
 860     now = pa_rtclock_now();
 861     latency = pa_source_get_latency_within_thread(u->source_output->source);
 862     delay = pa_memblockq_get_length(u->source_output->thread_info.delay_memblockq);
 863
 864     delay = (u->source_output->thread_info.resampler ? pa_resampler_request(u->source_output->thread_info.resampler, delay) : delay);
 865     rlen = pa_memblockq_get_length(u->source_memblockq);
 866     plen = pa_memblockq_get_length(u->sink_memblockq);
 867
 868     snapshot->source_now = now;
 869     snapshot->source_latency = latency;
 870     snapshot->source_delay = delay;
 871     snapshot->recv_counter = u->recv_counter;
 872     snapshot->rlen = rlen + u->sink_skip;
 873     snapshot->plen = plen + u->source_skip;
 874 }
 875
 876
 877 /* Called from output thread context */
 878 static int source_output_process_msg_cb(pa_msgobject *obj, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 879     struct userdata *u = PA_SOURCE_OUTPUT(obj)->userdata;
 880
 881     switch (code) {
 882
 883         case SOURCE_OUTPUT_MESSAGE_POST:
 884
 885             pa_source_output_assert_io_context(u->source_output);
 886
 887             if (u->source_output->source->thread_info.state == PA_SOURCE_RUNNING)
 888                 pa_memblockq_push_align(u->sink_memblockq, chunk);
 889             else
 890                 pa_memblockq_flush_write(u->sink_memblockq, TRUE);
 891
 892             u->recv_counter += (int64_t) chunk->length;
 893
 894             return 0;
 895
 896         case SOURCE_OUTPUT_MESSAGE_REWIND:
 897             pa_source_output_assert_io_context(u->source_output);
 898
 899             /* manipulate write index, never go past what we have */
 900             if (PA_SOURCE_IS_OPENED(u->source_output->source->thread_info.state))
 901                 pa_memblockq_seek(u->sink_memblockq, -offset, PA_SEEK_RELATIVE, TRUE);
 902             else
 903                 pa_memblockq_flush_write(u->sink_memblockq, TRUE);
 904
 905             pa_log_debug("Sink rewind (%lld)", (long long) offset);
 906
 907             u->recv_counter -= offset;
 908
 909             return 0;
 910
 911         case SOURCE_OUTPUT_MESSAGE_LATENCY_SNAPSHOT: {
 912             struct snapshot *snapshot = (struct snapshot *) data;
 913
 914             source_output_snapshot_within_thread(u, snapshot);
 915             return 0;
 916         }
 917
 918         case SOURCE_OUTPUT_MESSAGE_APPLY_DIFF_TIME:
 919             apply_diff_time(u, offset);
 920             return 0;
 921
 922     }
 923
 924     return pa_source_output_process_msg(obj, code, data, offset, chunk);
 925 }
 926
 927 static int sink_input_process_msg_cb(pa_msgobject *obj, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 928     struct userdata *u = PA_SINK_INPUT(obj)->userdata;
 929
 930     switch (code) {
 931
 932         case SINK_INPUT_MESSAGE_LATENCY_SNAPSHOT: {
 933             size_t delay;
 934             pa_usec_t now, latency;
 935             struct snapshot *snapshot = (struct snapshot *) data;
 936
 937             pa_sink_input_assert_io_context(u->sink_input);
 938
 939             now = pa_rtclock_now();
 940             latency = pa_sink_get_latency_within_thread(u->sink_input->sink);
 941             delay = pa_memblockq_get_length(u->sink_input->thread_info.render_memblockq);
 942
 943             delay = (u->sink_input->thread_info.resampler ? pa_resampler_request(u->sink_input->thread_info.resampler, delay) : delay);
 944
 945             snapshot->sink_now = now;
 946             snapshot->sink_latency = latency;
 947             snapshot->sink_delay = delay;
 948             snapshot->send_counter = u->send_counter;
 949             return 0;
 950         }
 951     }
 952
 953     return pa_sink_input_process_msg(obj, code, data, offset, chunk);
 954 }
 955
 956 /* Called from I/O thread context */
 957 static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
 958     struct userdata *u;
 959
 960     pa_sink_input_assert_ref(i);
 961     pa_assert_se(u = i->userdata);
 962
 963     pa_log_debug("Sink input update max rewind %lld", (long long) nbytes);
 964
 965     pa_memblockq_set_maxrewind(u->sink_memblockq, nbytes);
 966     pa_sink_set_max_rewind_within_thread(u->sink, nbytes);
 967 }
 968
 969 /* Called from I/O thread context */
 970 static void source_output_update_max_rewind_cb(pa_source_output *o, size_t nbytes) {
 971     struct userdata *u;
 972
 973     pa_source_output_assert_ref(o);
 974     pa_assert_se(u = o->userdata);
 975
 976     pa_log_debug("Source output update max rewind %lld", (long long) nbytes);
 977
 978     pa_source_set_max_rewind_within_thread(u->source, nbytes);
 979 }
 980
 981 /* Called from I/O thread context */
 982 static void sink_input_update_max_request_cb(pa_sink_input *i, size_t nbytes) {
 983     struct userdata *u;
 984
 985     pa_sink_input_assert_ref(i);
 986     pa_assert_se(u = i->userdata);
 987
 988     pa_log_debug("Sink input update max request %lld", (long long) nbytes);
 989
 990     pa_sink_set_max_request_within_thread(u->sink, nbytes);
 991 }
 992
 993 /* Called from I/O thread context */
 994 static void sink_input_update_sink_requested_latency_cb(pa_sink_input *i) {
 995     struct userdata *u;
 996     pa_usec_t latency;
 997
 998     pa_sink_input_assert_ref(i);
 999     pa_assert_se(u = i->userdata);
1000
1001     latency = pa_sink_get_requested_latency_within_thread(i->sink);
1002
1003     pa_log_debug("Sink input update requested latency %lld", (long long) latency);
1004 }
1005
1006 /* Called from I/O thread context */
1007 static void source_output_update_source_requested_latency_cb(pa_source_output *o) {
1008     struct userdata *u;
1009     pa_usec_t latency;
1010
1011     pa_source_output_assert_ref(o);
1012     pa_assert_se(u = o->userdata);
1013
1014     latency = pa_source_get_requested_latency_within_thread(o->source);
1015
1016     pa_log_debug("source output update requested latency %lld", (long long) latency);
1017 }
1018
1019 /* Called from I/O thread context */
1020 static void sink_input_update_sink_latency_range_cb(pa_sink_input *i) {
1021     struct userdata *u;
1022
1023     pa_sink_input_assert_ref(i);
1024     pa_assert_se(u = i->userdata);
1025
1026     pa_log_debug("Sink input update latency range %lld %lld",
1027         (long long) i->sink->thread_info.min_latency,
1028         (long long) i->sink->thread_info.max_latency);
1029
1030     pa_sink_set_latency_range_within_thread(u->sink, i->sink->thread_info.min_latency, i->sink->thread_info.max_latency);
1031 }
1032
1033 /* Called from I/O thread context */
1034 static void source_output_update_source_latency_range_cb(pa_source_output *o) {
1035     struct userdata *u;
1036
1037     pa_source_output_assert_ref(o);
1038     pa_assert_se(u = o->userdata);
1039
1040     pa_log_debug("Source output update latency range %lld %lld",
1041         (long long) o->source->thread_info.min_latency,
1042         (long long) o->source->thread_info.max_latency);
1043
1044     pa_source_set_latency_range_within_thread(u->source, o->source->thread_info.min_latency, o->source->thread_info.max_latency);
1045 }
1046
1047 /* Called from I/O thread context */
1048 static void sink_input_update_sink_fixed_latency_cb(pa_sink_input *i) {
1049     struct userdata *u;
1050
1051     pa_sink_input_assert_ref(i);
1052     pa_assert_se(u = i->userdata);
1053
1054     pa_log_debug("Sink input update fixed latency %lld",
1055         (long long) i->sink->thread_info.fixed_latency);
1056
1057     pa_sink_set_fixed_latency_within_thread(u->sink, i->sink->thread_info.fixed_latency);
1058 }
1059
1060 /* Called from I/O thread context */
1061 static void source_output_update_source_fixed_latency_cb(pa_source_output *o) {
1062     struct userdata *u;
1063
1064     pa_source_output_assert_ref(o);
1065     pa_assert_se(u = o->userdata);
1066
1067     pa_log_debug("Source output update fixed latency %lld",
1068         (long long) o->source->thread_info.fixed_latency);
1069
1070     pa_source_set_fixed_latency_within_thread(u->source, o->source->thread_info.fixed_latency);
1071 }
1072
1073 /* Called from output thread context */
1074 static void source_output_attach_cb(pa_source_output *o) {
1075     struct userdata *u;
1076
1077     pa_source_output_assert_ref(o);
1078     pa_source_output_assert_io_context(o);
1079     pa_assert_se(u = o->userdata);
1080
1081     pa_source_set_rtpoll(u->source, o->source->thread_info.rtpoll);
1082     pa_source_set_latency_range_within_thread(u->source, o->source->thread_info.min_latency, o->source->thread_info.max_latency);
1083     pa_source_set_fixed_latency_within_thread(u->source, o->source->thread_info.fixed_latency);
1084     pa_source_set_max_rewind_within_thread(u->source, pa_source_output_get_max_rewind(o));
1085
1086     pa_log_debug("Source output %d attach", o->index);
1087
1088     pa_source_attach_within_thread(u->source);
1089
1090     u->rtpoll_item_read = pa_rtpoll_item_new_asyncmsgq_read(
1091             o->source->thread_info.rtpoll,
1092             PA_RTPOLL_LATE,
1093             u->asyncmsgq);
1094 }
1095
1096 /* Called from I/O thread context */
1097 static void sink_input_attach_cb(pa_sink_input *i) {
1098     struct userdata *u;
1099
1100     pa_sink_input_assert_ref(i);
1101     pa_assert_se(u = i->userdata);
1102
1103     pa_sink_set_rtpoll(u->sink, i->sink->thread_info.rtpoll);
1104     pa_sink_set_latency_range_within_thread(u->sink, i->sink->thread_info.min_latency, i->sink->thread_info.max_latency);
1105
1106     /* (8.1) IF YOU NEED A FIXED BLOCK SIZE ADD THE LATENCY FOR ONE
1107      * BLOCK MINUS ONE SAMPLE HERE. SEE (7) */
1108     pa_sink_set_fixed_latency_within_thread(u->sink, i->sink->thread_info.fixed_latency);
1109
1110     /* (8.2) IF YOU NEED A FIXED BLOCK SIZE ROUND
1111      * pa_sink_input_get_max_request(i) UP TO MULTIPLES OF IT
1112      * HERE. SEE (6) */
1113     pa_sink_set_max_request_within_thread(u->sink, pa_sink_input_get_max_request(i));
1114     pa_sink_set_max_rewind_within_thread(u->sink, pa_sink_input_get_max_rewind(i));
1115
1116     pa_log_debug("Sink input %d attach", i->index);
1117
1118     u->rtpoll_item_write = pa_rtpoll_item_new_asyncmsgq_write(
1119             i->sink->thread_info.rtpoll,
1120             PA_RTPOLL_LATE,
1121             u->asyncmsgq);
1122
1123     pa_sink_attach_within_thread(u->sink);
1124 }
1125
1126
1127 /* Called from output thread context */
1128 static void source_output_detach_cb(pa_source_output *o) {
1129     struct userdata *u;
1130
1131     pa_source_output_assert_ref(o);
1132     pa_source_output_assert_io_context(o);
1133     pa_assert_se(u = o->userdata);
1134
1135     pa_source_detach_within_thread(u->source);
1136     pa_source_set_rtpoll(u->source, NULL);
1137
1138     pa_log_debug("Source output %d detach", o->index);
1139
1140     if (u->rtpoll_item_read) {
1141         pa_rtpoll_item_free(u->rtpoll_item_read);
1142         u->rtpoll_item_read = NULL;
1143     }
1144 }
1145
1146 /* Called from I/O thread context */
1147 static void sink_input_detach_cb(pa_sink_input *i) {
1148     struct userdata *u;
1149
1150     pa_sink_input_assert_ref(i);
1151     pa_assert_se(u = i->userdata);
1152
1153     pa_sink_detach_within_thread(u->sink);
1154
1155     pa_sink_set_rtpoll(u->sink, NULL);
1156
1157     pa_log_debug("Sink input %d detach", i->index);
1158
1159     if (u->rtpoll_item_write) {
1160         pa_rtpoll_item_free(u->rtpoll_item_write);
1161         u->rtpoll_item_write = NULL;
1162     }
1163 }
1164
1165 /* Called from output thread context */
1166 static void source_output_state_change_cb(pa_source_output *o, pa_source_output_state_t state) {
1167     struct userdata *u;
1168
1169     pa_source_output_assert_ref(o);
1170     pa_source_output_assert_io_context(o);
1171     pa_assert_se(u = o->userdata);
1172
1173     pa_log_debug("Source output %d state %d", o->index, state);
1174 }
1175
1176 /* Called from IO thread context */
1177 static void sink_input_state_change_cb(pa_sink_input *i, pa_sink_input_state_t state) {
1178     struct userdata *u;
1179
1180     pa_sink_input_assert_ref(i);
1181     pa_assert_se(u = i->userdata);
1182
1183     pa_log_debug("Sink input %d state %d", i->index, state);
1184
1185     /* If we are added for the first time, ask for a rewinding so that
1186      * we are heard right-away. */
1187     if (PA_SINK_INPUT_IS_LINKED(state) &&
1188         i->thread_info.state == PA_SINK_INPUT_INIT) {
1189         pa_log_debug("Requesting rewind due to state change.");
1190         pa_sink_input_request_rewind(i, 0, FALSE, TRUE, TRUE);
1191     }
1192 }
1193
1194 /* Called from main thread */
1195 static void source_output_kill_cb(pa_source_output *o) {
1196     struct userdata *u;
1197
1198     pa_source_output_assert_ref(o);
1199     pa_assert_ctl_context();
1200     pa_assert_se(u = o->userdata);
1201
1202     u->dead = TRUE;
1203
1204     /* The order here matters! We first kill the source output, followed
1205      * by the source. That means the source callbacks must be protected
1206      * against an unconnected source output! */
1207     pa_source_output_unlink(u->source_output);
1208     pa_source_unlink(u->source);
1209
1210     pa_source_output_unref(u->source_output);
1211     u->source_output = NULL;
1212
1213     pa_source_unref(u->source);
1214     u->source = NULL;
1215
1216     pa_log_debug("Source output kill %d", o->index);
1217
1218     pa_module_unload_request(u->module, TRUE);
1219 }
1220
1221 /* Called from main context */
1222 static void sink_input_kill_cb(pa_sink_input *i) {
1223     struct userdata *u;
1224
1225     pa_sink_input_assert_ref(i);
1226     pa_assert_se(u = i->userdata);
1227
1228     u->dead = TRUE;
1229
1230     /* The order here matters! We first kill the sink input, followed
1231      * by the sink. That means the sink callbacks must be protected
1232      * against an unconnected sink input! */
1233     pa_sink_input_unlink(u->sink_input);
1234     pa_sink_unlink(u->sink);
1235
1236     pa_sink_input_unref(u->sink_input);
1237     u->sink_input = NULL;
1238
1239     pa_sink_unref(u->sink);
1240     u->sink = NULL;
1241
1242     pa_log_debug("Sink input kill %d", i->index);
1243
1244     pa_module_unload_request(u->module, TRUE);
1245 }
1246
1247 /* Called from main thread */
1248 static pa_bool_t source_output_may_move_to_cb(pa_source_output *o, pa_source *dest) {
1249     struct userdata *u;
1250
1251     pa_source_output_assert_ref(o);
1252     pa_assert_ctl_context();
1253     pa_assert_se(u = o->userdata);
1254
1255     if (u->dead)
1256         return FALSE;
1257
1258     return (u->source != dest) && (u->sink != dest->monitor_of);
1259 }
1260
1261 /* Called from main context */
1262 static pa_bool_t sink_input_may_move_to_cb(pa_sink_input *i, pa_sink *dest) {
1263     struct userdata *u;
1264
1265     pa_sink_input_assert_ref(i);
1266     pa_assert_se(u = i->userdata);
1267
1268     if (u->dead)
1269         return FALSE;
1270
1271     return u->sink != dest;
1272 }
1273
1274 /* Called from main thread */
1275 static void source_output_moving_cb(pa_source_output *o, pa_source *dest) {
1276     struct userdata *u;
1277
1278     pa_source_output_assert_ref(o);
1279     pa_assert_ctl_context();
1280     pa_assert_se(u = o->userdata);
1281
1282     if (dest) {
1283         pa_source_set_asyncmsgq(u->source, dest->asyncmsgq);
1284         pa_source_update_flags(u->source, PA_SOURCE_LATENCY|PA_SOURCE_DYNAMIC_LATENCY, dest->flags);
1285     } else
1286         pa_source_set_asyncmsgq(u->source, NULL);
1287
1288     if (u->source_auto_desc && dest) {
1289         const char *z;
1290         pa_proplist *pl;
1291
1292         pl = pa_proplist_new();
1293         z = pa_proplist_gets(dest->proplist, PA_PROP_DEVICE_DESCRIPTION);
1294         pa_proplist_setf(pl, PA_PROP_DEVICE_DESCRIPTION, "Echo-Cancel Source %s on %s",
1295                          pa_proplist_gets(u->source->proplist, "device.echo-cancel.name"), z ? z : dest->name);
1296
1297         pa_source_update_proplist(u->source, PA_UPDATE_REPLACE, pl);
1298         pa_proplist_free(pl);
1299     }
1300 }
1301
1302 /* Called from main context */
1303 static void sink_input_moving_cb(pa_sink_input *i, pa_sink *dest) {
1304     struct userdata *u;
1305
1306     pa_sink_input_assert_ref(i);
1307     pa_assert_se(u = i->userdata);
1308
1309     if (dest) {
1310         pa_sink_set_asyncmsgq(u->sink, dest->asyncmsgq);
1311         pa_sink_update_flags(u->sink, PA_SINK_LATENCY|PA_SINK_DYNAMIC_LATENCY, dest->flags);
1312     } else
1313         pa_sink_set_asyncmsgq(u->sink, NULL);
1314
1315     if (u->sink_auto_desc && dest) {
1316         const char *z;
1317         pa_proplist *pl;
1318
1319         pl = pa_proplist_new();
1320         z = pa_proplist_gets(dest->proplist, PA_PROP_DEVICE_DESCRIPTION);
1321         pa_proplist_setf(pl, PA_PROP_DEVICE_DESCRIPTION, "Echo-Cancel Sink %s on %s",
1322                          pa_proplist_gets(u->sink->proplist, "device.echo-cancel.name"), z ? z : dest->name);
1323
1324         pa_sink_update_proplist(u->sink, PA_UPDATE_REPLACE, pl);
1325         pa_proplist_free(pl);
1326     }
1327 }
1328
1329 /* Called from main context */
1330 static void sink_input_volume_changed_cb(pa_sink_input *i) {
1331     struct userdata *u;
1332
1333     pa_sink_input_assert_ref(i);
1334     pa_assert_se(u = i->userdata);
1335
1336     pa_sink_volume_changed(u->sink, &i->volume);
1337 }
1338
1339 /* Called from main context */
1340 static void sink_input_mute_changed_cb(pa_sink_input *i) {
1341     struct userdata *u;
1342
1343     pa_sink_input_assert_ref(i);
1344     pa_assert_se(u = i->userdata);
1345
1346     pa_sink_mute_changed(u->sink, i->muted);
1347 }
1348
1349 static pa_echo_canceller_method_t get_ec_method_from_string(const char *method) {
1350     if (pa_streq(method, "speex"))
1351         return PA_ECHO_CANCELLER_SPEEX;
1352     else if (pa_streq(method, "adrian"))
1353         return PA_ECHO_CANCELLER_ADRIAN;
1354 #ifdef HAVE_WEBRTC
1355     else if (pa_streq(method, "webrtc"))
1356         return PA_ECHO_CANCELLER_WEBRTC;
1357 #endif
1358     else
1359         return PA_ECHO_CANCELLER_INVALID;
1360 }
1361
1362 /* Common initialisation bits between module-echo-cancel and the standalone test program */
1363 static int init_common(pa_modargs *ma, struct userdata *u, pa_sample_spec *source_ss, pa_channel_map *source_map) {
1364     pa_echo_canceller_method_t ec_method;
1365
1366     if (pa_modargs_get_sample_spec_and_channel_map(ma, source_ss, source_map, PA_CHANNEL_MAP_DEFAULT) < 0) {
1367         pa_log("Invalid sample format specification or channel map");
1368         goto fail;
1369     }
1370
1371     u->ec = pa_xnew0(pa_echo_canceller, 1);
1372     if (!u->ec) {
1373         pa_log("Failed to alloc echo canceller");
1374         goto fail;
1375     }
1376
1377     if ((ec_method = get_ec_method_from_string(pa_modargs_get_value(ma, "aec_method", DEFAULT_ECHO_CANCELLER))) < 0) {
1378         pa_log("Invalid echo canceller implementation");
1379         goto fail;
1380     }
1381
1382     u->ec->init = ec_table[ec_method].init;
1383     u->ec->run = ec_table[ec_method].run;
1384     u->ec->done = ec_table[ec_method].done;
1385
1386     return 0;
1387
1388 fail:
1389     return -1;
1390 }
1391
1392
1393 int pa__init(pa_module*m) {
1394     struct userdata *u;
1395     pa_sample_spec source_ss, sink_ss;
1396     pa_channel_map source_map, sink_map;
1397     pa_modargs *ma;
1398     pa_source *source_master=NULL;
1399     pa_sink *sink_master=NULL;
1400     pa_source_output_new_data source_output_data;
1401     pa_sink_input_new_data sink_input_data;
1402     pa_source_new_data source_data;
1403     pa_sink_new_data sink_data;
1404     pa_memchunk silence;
1405     uint32_t temp;
1406     pa_bool_t use_volume_sharing = TRUE;
1407
1408     pa_assert(m);
1409
1410     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
1411         pa_log("Failed to parse module arguments.");
1412         goto fail;
1413     }
1414
1415     if (!(source_master = pa_namereg_get(m->core, pa_modargs_get_value(ma, "source_master", NULL), PA_NAMEREG_SOURCE))) {
1416         pa_log("Master source not found");
1417         goto fail;
1418     }
1419     pa_assert(source_master);
1420
1421     if (!(sink_master = pa_namereg_get(m->core, pa_modargs_get_value(ma, "sink_master", NULL), PA_NAMEREG_SINK))) {
1422         pa_log("Master sink not found");
1423         goto fail;
1424     }
1425     pa_assert(sink_master);
1426
1427     if (source_master->monitor_of == sink_master) {
1428         pa_log("Can't cancel echo between a sink and its monitor");
1429         goto fail;
1430     }
1431
1432     source_ss = source_master->sample_spec;
1433     source_ss.rate = DEFAULT_RATE;
1434     source_ss.channels = DEFAULT_CHANNELS;
1435     pa_channel_map_init_auto(&source_map, source_ss.channels, PA_CHANNEL_MAP_DEFAULT);
1436
1437     sink_ss = sink_master->sample_spec;
1438     sink_map = sink_master->channel_map;
1439
1440     if (pa_modargs_get_value_boolean(ma, "use_volume_sharing", &use_volume_sharing) < 0) {
1441         pa_log("use_volume_sharing= expects a boolean argument");
1442         goto fail;
1443     }
1444
1445     u = pa_xnew0(struct userdata, 1);
1446     if (!u) {
1447         pa_log("Failed to alloc userdata");
1448         goto fail;
1449     }
1450     u->core = m->core;
1451     u->module = m;
1452     m->userdata = u;
1453     u->dead = FALSE;
1454
1455     temp = DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC;
1456     if (pa_modargs_get_value_u32(ma, "adjust_time", &temp) < 0) {
1457         pa_log("Failed to parse adjust_time value");
1458         goto fail;
1459     }
1460
1461     if (temp != DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC)
1462         u->adjust_time = temp * PA_USEC_PER_SEC;
1463     else
1464         u->adjust_time = DEFAULT_ADJUST_TIME_USEC;
1465
1466     temp = DEFAULT_ADJUST_TOLERANCE / PA_USEC_PER_MSEC;
1467     if (pa_modargs_get_value_u32(ma, "adjust_threshold", &temp) < 0) {
1468         pa_log("Failed to parse adjust_threshold value");
1469         goto fail;
1470     }
1471
1472     if (temp != DEFAULT_ADJUST_TOLERANCE / PA_USEC_PER_MSEC)
1473         u->adjust_threshold = temp * PA_USEC_PER_MSEC;
1474     else
1475         u->adjust_threshold = DEFAULT_ADJUST_TOLERANCE;
1476
1477     u->save_aec = DEFAULT_SAVE_AEC;
1478     if (pa_modargs_get_value_boolean(ma, "save_aec", &u->save_aec) < 0) {
1479         pa_log("Failed to parse save_aec value");
1480         goto fail;
1481     }
1482
1483     u->autoloaded = DEFAULT_AUTOLOADED;
1484     if (pa_modargs_get_value_boolean(ma, "autoloaded", &u->autoloaded) < 0) {
1485         pa_log("Failed to parse autoloaded value");
1486         goto fail;
1487     }
1488
1489     if (init_common(ma, u, &source_ss, &source_map))
1490         goto fail;
1491
1492     u->asyncmsgq = pa_asyncmsgq_new(0);
1493     u->need_realign = TRUE;
1494
1495     if (u->ec->init) {
1496         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))) {
1497             pa_log("Failed to init AEC engine");
1498             goto fail;
1499         }
1500     }
1501
1502     /* Create source */
1503     pa_source_new_data_init(&source_data);
1504     source_data.driver = __FILE__;
1505     source_data.module = m;
1506     if (!(source_data.name = pa_xstrdup(pa_modargs_get_value(ma, "source_name", NULL))))
1507         source_data.name = pa_sprintf_malloc("%s.echo-cancel", source_master->name);
1508     pa_source_new_data_set_sample_spec(&source_data, &source_ss);
1509     pa_source_new_data_set_channel_map(&source_data, &source_map);
1510     pa_proplist_sets(source_data.proplist, PA_PROP_DEVICE_MASTER_DEVICE, source_master->name);
1511     pa_proplist_sets(source_data.proplist, PA_PROP_DEVICE_CLASS, "filter");
1512     if (!u->autoloaded)
1513         pa_proplist_sets(source_data.proplist, PA_PROP_DEVICE_INTENDED_ROLES, "phone");
1514     pa_proplist_sets(source_data.proplist, "device.echo-cancel.name", source_data.name);
1515
1516     if (pa_modargs_get_proplist(ma, "source_properties", source_data.proplist, PA_UPDATE_REPLACE) < 0) {
1517         pa_log("Invalid properties");
1518         pa_source_new_data_done(&source_data);
1519         goto fail;
1520     }
1521
1522     if ((u->source_auto_desc = !pa_proplist_contains(source_data.proplist, PA_PROP_DEVICE_DESCRIPTION))) {
1523         const char *z;
1524
1525         z = pa_proplist_gets(source_master->proplist, PA_PROP_DEVICE_DESCRIPTION);
1526         pa_proplist_setf(source_data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Echo-Cancel Source %s on %s", source_data.name, z ? z : source_master->name);
1527     }
1528
1529     u->source = pa_source_new(m->core, &source_data, (source_master->flags & (PA_SOURCE_LATENCY | PA_SOURCE_DYNAMIC_LATENCY))
1530                                                      | (use_volume_sharing ? PA_SOURCE_SHARE_VOLUME_WITH_MASTER : 0));
1531     pa_source_new_data_done(&source_data);
1532
1533     if (!u->source) {
1534         pa_log("Failed to create source.");
1535         goto fail;
1536     }
1537
1538     u->source->parent.process_msg = source_process_msg_cb;
1539     u->source->set_state = source_set_state_cb;
1540     u->source->update_requested_latency = source_update_requested_latency_cb;
1541     pa_source_set_get_mute_callback(u->source, source_get_mute_cb);
1542     pa_source_set_set_mute_callback(u->source, source_set_mute_cb);
1543     if (!use_volume_sharing) {
1544         pa_source_set_get_volume_callback(u->source, source_get_volume_cb);
1545         pa_source_set_set_volume_callback(u->source, source_set_volume_cb);
1546         pa_source_enable_decibel_volume(u->source, TRUE);
1547     }
1548     u->source->userdata = u;
1549
1550     pa_source_set_asyncmsgq(u->source, source_master->asyncmsgq);
1551
1552     /* Create sink */
1553     pa_sink_new_data_init(&sink_data);
1554     sink_data.driver = __FILE__;
1555     sink_data.module = m;
1556     if (!(sink_data.name = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", NULL))))
1557         sink_data.name = pa_sprintf_malloc("%s.echo-cancel", sink_master->name);
1558     pa_sink_new_data_set_sample_spec(&sink_data, &sink_ss);
1559     pa_sink_new_data_set_channel_map(&sink_data, &sink_map);
1560     pa_proplist_sets(sink_data.proplist, PA_PROP_DEVICE_MASTER_DEVICE, sink_master->name);
1561     pa_proplist_sets(sink_data.proplist, PA_PROP_DEVICE_CLASS, "filter");
1562     if (!u->autoloaded)
1563         pa_proplist_sets(sink_data.proplist, PA_PROP_DEVICE_INTENDED_ROLES, "phone");
1564     pa_proplist_sets(sink_data.proplist, "device.echo-cancel.name", sink_data.name);
1565
1566     if (pa_modargs_get_proplist(ma, "sink_properties", sink_data.proplist, PA_UPDATE_REPLACE) < 0) {
1567         pa_log("Invalid properties");
1568         pa_sink_new_data_done(&sink_data);
1569         goto fail;
1570     }
1571
1572     if ((u->sink_auto_desc = !pa_proplist_contains(sink_data.proplist, PA_PROP_DEVICE_DESCRIPTION))) {
1573         const char *z;
1574
1575         z = pa_proplist_gets(sink_master->proplist, PA_PROP_DEVICE_DESCRIPTION);
1576         pa_proplist_setf(sink_data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Echo-Cancel Sink %s on %s", sink_data.name, z ? z : sink_master->name);
1577     }
1578
1579     u->sink = pa_sink_new(m->core, &sink_data, (sink_master->flags & (PA_SINK_LATENCY | PA_SINK_DYNAMIC_LATENCY))
1580                                                | (use_volume_sharing ? PA_SINK_SHARE_VOLUME_WITH_MASTER : 0));
1581     pa_sink_new_data_done(&sink_data);
1582
1583     if (!u->sink) {
1584         pa_log("Failed to create sink.");
1585         goto fail;
1586     }
1587
1588     u->sink->parent.process_msg = sink_process_msg_cb;
1589     u->sink->set_state = sink_set_state_cb;
1590     u->sink->update_requested_latency = sink_update_requested_latency_cb;
1591     u->sink->request_rewind = sink_request_rewind_cb;
1592     pa_sink_set_set_mute_callback(u->sink, sink_set_mute_cb);
1593     if (!use_volume_sharing) {
1594         pa_sink_set_set_volume_callback(u->sink, sink_set_volume_cb);
1595         pa_sink_enable_decibel_volume(u->sink, TRUE);
1596     }
1597     u->sink->userdata = u;
1598
1599     pa_sink_set_asyncmsgq(u->sink, sink_master->asyncmsgq);
1600
1601     /* Create source output */
1602     pa_source_output_new_data_init(&source_output_data);
1603     source_output_data.driver = __FILE__;
1604     source_output_data.module = m;
1605     pa_source_output_new_data_set_source(&source_output_data, source_master, FALSE);
1606     source_output_data.destination_source = u->source;
1607     /* FIXME
1608        source_output_data.flags = PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND; */
1609
1610     pa_proplist_sets(source_output_data.proplist, PA_PROP_MEDIA_NAME, "Echo-Cancel Source Stream");
1611     pa_proplist_sets(source_output_data.proplist, PA_PROP_MEDIA_ROLE, "filter");
1612     pa_source_output_new_data_set_sample_spec(&source_output_data, &source_ss);
1613     pa_source_output_new_data_set_channel_map(&source_output_data, &source_map);
1614
1615     pa_source_output_new(&u->source_output, m->core, &source_output_data);
1616     pa_source_output_new_data_done(&source_output_data);
1617
1618     if (!u->source_output)
1619         goto fail;
1620
1621     u->source_output->parent.process_msg = source_output_process_msg_cb;
1622     u->source_output->push = source_output_push_cb;
1623     u->source_output->process_rewind = source_output_process_rewind_cb;
1624     u->source_output->update_max_rewind = source_output_update_max_rewind_cb;
1625     u->source_output->update_source_requested_latency = source_output_update_source_requested_latency_cb;
1626     u->source_output->update_source_latency_range = source_output_update_source_latency_range_cb;
1627     u->source_output->update_source_fixed_latency = source_output_update_source_fixed_latency_cb;
1628     u->source_output->kill = source_output_kill_cb;
1629     u->source_output->attach = source_output_attach_cb;
1630     u->source_output->detach = source_output_detach_cb;
1631     u->source_output->state_change = source_output_state_change_cb;
1632     u->source_output->may_move_to = source_output_may_move_to_cb;
1633     u->source_output->moving = source_output_moving_cb;
1634     u->source_output->userdata = u;
1635
1636     u->source->output_from_master = u->source_output;
1637
1638     /* Create sink input */
1639     pa_sink_input_new_data_init(&sink_input_data);
1640     sink_input_data.driver = __FILE__;
1641     sink_input_data.module = m;
1642     pa_sink_input_new_data_set_sink(&sink_input_data, sink_master, FALSE);
1643     sink_input_data.origin_sink = u->sink;
1644     pa_proplist_sets(sink_input_data.proplist, PA_PROP_MEDIA_NAME, "Echo-Cancel Sink Stream");
1645     pa_proplist_sets(sink_input_data.proplist, PA_PROP_MEDIA_ROLE, "filter");
1646     pa_sink_input_new_data_set_sample_spec(&sink_input_data, &sink_ss);
1647     pa_sink_input_new_data_set_channel_map(&sink_input_data, &sink_map);
1648     sink_input_data.flags = PA_SINK_INPUT_VARIABLE_RATE;
1649
1650     pa_sink_input_new(&u->sink_input, m->core, &sink_input_data);
1651     pa_sink_input_new_data_done(&sink_input_data);
1652
1653     if (!u->sink_input)
1654         goto fail;
1655
1656     u->sink_input->parent.process_msg = sink_input_process_msg_cb;
1657     u->sink_input->pop = sink_input_pop_cb;
1658     u->sink_input->process_rewind = sink_input_process_rewind_cb;
1659     u->sink_input->update_max_rewind = sink_input_update_max_rewind_cb;
1660     u->sink_input->update_max_request = sink_input_update_max_request_cb;
1661     u->sink_input->update_sink_requested_latency = sink_input_update_sink_requested_latency_cb;
1662     u->sink_input->update_sink_latency_range = sink_input_update_sink_latency_range_cb;
1663     u->sink_input->update_sink_fixed_latency = sink_input_update_sink_fixed_latency_cb;
1664     u->sink_input->kill = sink_input_kill_cb;
1665     u->sink_input->attach = sink_input_attach_cb;
1666     u->sink_input->detach = sink_input_detach_cb;
1667     u->sink_input->state_change = sink_input_state_change_cb;
1668     u->sink_input->may_move_to = sink_input_may_move_to_cb;
1669     u->sink_input->moving = sink_input_moving_cb;
1670     if (!use_volume_sharing)
1671         u->sink_input->volume_changed = sink_input_volume_changed_cb;
1672     u->sink_input->mute_changed = sink_input_mute_changed_cb;
1673     u->sink_input->userdata = u;
1674
1675     u->sink->input_to_master = u->sink_input;
1676
1677     pa_sink_input_get_silence(u->sink_input, &silence);
1678
1679     u->source_memblockq = pa_memblockq_new("module-echo-cancel source_memblockq", 0, MEMBLOCKQ_MAXLENGTH, 0,
1680         &source_ss, 1, 1, 0, &silence);
1681     u->sink_memblockq = pa_memblockq_new("module-echo-cancel sink_memblockq", 0, MEMBLOCKQ_MAXLENGTH, 0,
1682         &sink_ss, 1, 1, 0, &silence);
1683
1684     pa_memblock_unref(silence.memblock);
1685
1686     if (!u->source_memblockq || !u->sink_memblockq) {
1687         pa_log("Failed to create memblockq.");
1688         goto fail;
1689     }
1690
1691     if (u->adjust_time > 0)
1692         u->time_event = pa_core_rttime_new(m->core, pa_rtclock_now() + u->adjust_time, time_callback, u);
1693
1694     if (u->save_aec) {
1695         pa_log("Creating AEC files in /tmp");
1696         u->captured_file = fopen("/tmp/aec_rec.sw", "wb");
1697         if (u->captured_file == NULL)
1698             perror ("fopen failed");
1699         u->played_file = fopen("/tmp/aec_play.sw", "wb");
1700         if (u->played_file == NULL)
1701             perror ("fopen failed");
1702         u->canceled_file = fopen("/tmp/aec_out.sw", "wb");
1703         if (u->canceled_file == NULL)
1704             perror ("fopen failed");
1705     }
1706
1707     pa_sink_put(u->sink);
1708     pa_source_put(u->source);
1709
1710     pa_sink_input_put(u->sink_input);
1711     pa_source_output_put(u->source_output);
1712
1713     pa_modargs_free(ma);
1714
1715     return 0;
1716
1717 fail:
1718     if (ma)
1719         pa_modargs_free(ma);
1720
1721     pa__done(m);
1722
1723     return -1;
1724 }
1725
1726 int pa__get_n_used(pa_module *m) {
1727     struct userdata *u;
1728
1729     pa_assert(m);
1730     pa_assert_se(u = m->userdata);
1731
1732     return pa_sink_linked_by(u->sink) + pa_source_linked_by(u->source);
1733 }
1734
1735 void pa__done(pa_module*m) {
1736     struct userdata *u;
1737
1738     pa_assert(m);
1739
1740     if (!(u = m->userdata))
1741         return;
1742
1743     u->dead = TRUE;
1744
1745     /* See comments in source_output_kill_cb() above regarding
1746      * destruction order! */
1747
1748     if (u->time_event)
1749         u->core->mainloop->time_free(u->time_event);
1750
1751     if (u->source_output)
1752         pa_source_output_unlink(u->source_output);
1753     if (u->sink_input)
1754         pa_sink_input_unlink(u->sink_input);
1755
1756     if (u->source)
1757         pa_source_unlink(u->source);
1758     if (u->sink)
1759         pa_sink_unlink(u->sink);
1760
1761     if (u->source_output)
1762         pa_source_output_unref(u->source_output);
1763     if (u->sink_input)
1764         pa_sink_input_unref(u->sink_input);
1765
1766     if (u->source)
1767         pa_source_unref(u->source);
1768     if (u->sink)
1769         pa_sink_unref(u->sink);
1770
1771     if (u->source_memblockq)
1772         pa_memblockq_free(u->source_memblockq);
1773     if (u->sink_memblockq)
1774         pa_memblockq_free(u->sink_memblockq);
1775
1776     if (u->ec) {
1777         if (u->ec->done)
1778             u->ec->done(u->ec);
1779
1780         pa_xfree(u->ec);
1781     }
1782
1783     if (u->asyncmsgq)
1784         pa_asyncmsgq_unref(u->asyncmsgq);
1785
1786     if (u->save_aec) {
1787         if (u->played_file)
1788             fclose(u->played_file);
1789         if (u->captured_file)
1790             fclose(u->captured_file);
1791         if (u->canceled_file)
1792             fclose(u->canceled_file);
1793     }
1794
1795     pa_xfree(u);
1796 }
1797
1798 #ifdef ECHO_CANCEL_TEST
1799 /*
1800  * Stand-alone test program for running in the canceller on pre-recorded files.
1801  */
1802 int main(int argc, char* argv[]) {
1803     struct userdata u;
1804     pa_sample_spec source_ss, sink_ss;
1805     pa_channel_map source_map, sink_map;
1806     pa_modargs *ma = NULL;
1807     uint8_t *rdata = NULL, *pdata = NULL, *cdata = NULL;
1808     int ret = 0, unused;
1809
1810     pa_memzero(&u, sizeof(u));
1811
1812     if (argc < 4 || argc > 6) {
1813         goto usage;
1814     }
1815
1816     u.ec = pa_xnew0(pa_echo_canceller, 1);
1817     if (!u.ec) {
1818         pa_log("Failed to alloc echo canceller");
1819         goto fail;
1820     }
1821
1822     u.captured_file = fopen(argv[2], "r");
1823     if (u.captured_file == NULL) {
1824         perror ("fopen failed");
1825         goto fail;
1826     }
1827     u.played_file = fopen(argv[1], "r");
1828     if (u.played_file == NULL) {
1829         perror ("fopen failed");
1830         goto fail;
1831     }
1832     u.canceled_file = fopen(argv[3], "wb");
1833     if (u.canceled_file == NULL) {
1834         perror ("fopen failed");
1835         goto fail;
1836     }
1837
1838     u.core = pa_xnew0(pa_core, 1);
1839     u.core->cpu_info.cpu_type = PA_CPU_X86;
1840     u.core->cpu_info.flags.x86 |= PA_CPU_X86_SSE;
1841
1842     if (!(ma = pa_modargs_new(argc > 4 ? argv[4] : NULL, valid_modargs))) {
1843         pa_log("Failed to parse module arguments.");
1844         goto fail;
1845     }
1846
1847     source_ss.format = PA_SAMPLE_S16LE;
1848     source_ss.rate = DEFAULT_RATE;
1849     source_ss.channels = DEFAULT_CHANNELS;
1850     pa_channel_map_init_auto(&source_map, source_ss.channels, PA_CHANNEL_MAP_DEFAULT);
1851
1852     init_common(ma, &u, &source_ss, &source_map);
1853
1854     if (!u.ec->init(u.core, u.ec, &source_ss, &source_map, &sink_ss, &sink_map, &u.blocksize,
1855                      (argc > 4) ? argv[5] : NULL )) {
1856         pa_log("Failed to init AEC engine");
1857         goto fail;
1858     }
1859
1860     rdata = pa_xmalloc(u.blocksize);
1861     pdata = pa_xmalloc(u.blocksize);
1862     cdata = pa_xmalloc(u.blocksize);
1863
1864     while (fread(rdata, u.blocksize, 1, u.captured_file) > 0) {
1865         if (fread(pdata, u.blocksize, 1, u.played_file) == 0) {
1866             perror("played file ended before captured file");
1867             break;
1868         }
1869
1870         u.ec->run(u.ec, rdata, pdata, cdata);
1871
1872         unused = fwrite(cdata, u.blocksize, 1, u.canceled_file);
1873     }
1874
1875     u.ec->done(u.ec);
1876
1877     fclose(u.captured_file);
1878     fclose(u.played_file);
1879     fclose(u.canceled_file);
1880
1881 out:
1882     pa_xfree(rdata);
1883     pa_xfree(pdata);
1884     pa_xfree(cdata);
1885
1886     pa_xfree(u.ec);
1887     pa_xfree(u.core);
1888
1889     if (ma)
1890         pa_modargs_free(ma);
1891
1892     return ret;
1893
1894 usage:
1895     pa_log("Usage: %s play_file rec_file out_file [module args] [aec_args]",argv[0]);
1896
1897 fail:
1898     ret = -1;
1899     goto out;
1900 }
1901 #endif /* ECHO_CANCEL_TEST */