src/modules/module-combine.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of PulseAudio.
   5
   6   Copyright 2004-2006 Lennart Poettering
   7
   8   PulseAudio is free software; you can redistribute it and/or modify
   9   it under the terms of the GNU Lesser General Public License as published
  10   by the Free Software Foundation; either version 2 of the License,
  11   or (at your option) any later version.
  12
  13   PulseAudio is distributed in the hope that it will be useful, but
  14   WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16   General Public License for more details.
  17
  18   You should have received a copy of the GNU Lesser General Public License
  19   along with PulseAudio; if not, write to the Free Software
  20   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  21   USA.
  22 ***/
  23
  24 #ifdef HAVE_CONFIG_H
  25 #include <config.h>
  26 #endif
  27
  28 #include <stdio.h>
  29 #include <errno.h>
  30
  31 #include <pulse/timeval.h>
  32 #include <pulse/xmalloc.h>
  33
  34 #include <pulsecore/macro.h>
  35 #include <pulsecore/module.h>
  36 #include <pulsecore/llist.h>
  37 #include <pulsecore/sink.h>
  38 #include <pulsecore/sink-input.h>
  39 #include <pulsecore/memblockq.h>
  40 #include <pulsecore/log.h>
  41 #include <pulsecore/core-util.h>
  42 #include <pulsecore/modargs.h>
  43 #include <pulsecore/namereg.h>
  44 #include <pulsecore/mutex.h>
  45 #include <pulsecore/thread.h>
  46 #include <pulsecore/thread-mq.h>
  47 #include <pulsecore/rtpoll.h>
  48 #include <pulsecore/rtclock.h>
  49 #include <pulsecore/core-error.h>
  50
  51 #include "module-combine-symdef.h"
  52
  53 PA_MODULE_AUTHOR("Lennart Poettering")
  54 PA_MODULE_DESCRIPTION("Combine multiple sinks to one")
  55 PA_MODULE_VERSION(PACKAGE_VERSION)
  56 PA_MODULE_USAGE(
  57         "sink_name=<name for the sink> "
  58         "master=<master sink> "
  59         "slaves=<slave sinks> "
  60         "adjust_time=<seconds> "
  61         "resample_method=<method> "
  62         "format=<sample format> "
  63         "channels=<number of channels> "
  64         "rate=<sample rate> "
  65         "channel_map=<channel map>")
  66
  67 #define DEFAULT_SINK_NAME "combined"
  68 #define MEMBLOCKQ_MAXLENGTH (1024*170)
  69
  70 #define DEFAULT_ADJUST_TIME 10
  71
  72 static const char* const valid_modargs[] = {
  73     "sink_name",
  74     "master",
  75     "slaves",
  76     "adjust_time",
  77     "resample_method",
  78     "format",
  79     "channels",
  80     "rate",
  81     "channel_map",
  82     NULL
  83 };
  84
  85 struct output {
  86     struct userdata *userdata;
  87     pa_sink *sink;
  88     pa_sink_input *sink_input;
  89
  90     pa_asyncmsgq *inq,    /* Message queue from the master to this sink input */
  91                  *outq;   /* Message queue from this sink input to the master */
  92     pa_rtpoll_item *inq_rtpoll_item, *outq_rtpoll_item;
  93
  94     pa_memblockq *memblockq;
  95
  96     pa_usec_t total_latency;
  97
  98     PA_LLIST_FIELDS(struct output);
  99 };
 100
 101 struct userdata {
 102     pa_core *core;
 103     pa_module *module;
 104     pa_sink *sink;
 105
 106     pa_thread *thread;
 107     pa_thread_mq thread_mq;
 108     pa_rtpoll *rtpoll;
 109
 110     struct output *master;
 111
 112     pa_time_event *time_event;
 113     uint32_t adjust_time;
 114
 115     int automatic;
 116     size_t block_size;
 117
 118     struct timespec timestamp;
 119
 120     pa_hook_slot *sink_new_slot, *sink_unlink_slot, *sink_state_changed_slot;
 121
 122     pa_resample_method_t resample_method;
 123
 124     struct timespec adjust_timestamp;
 125
 126     pa_idxset* outputs; /* managed in main context */
 127
 128     struct {
 129         PA_LLIST_HEAD(struct output, outputs); /* managed in IO thread context */
 130         struct output *master;
 131     } thread_info;
 132 };
 133
 134 enum {
 135     SINK_MESSAGE_ADD_OUTPUT = PA_SINK_MESSAGE_MAX,
 136     SINK_MESSAGE_REMOVE_OUTPUT,
 137     SINK_MESSAGE_NEED,
 138     SINK_MESSAGE_SET_MASTER
 139 };
 140
 141 enum {
 142     SINK_INPUT_MESSAGE_POST = PA_SINK_INPUT_MESSAGE_MAX
 143 };
 144
 145 static void output_free(struct output *o);
 146 static int output_create_sink_input(struct userdata *u, struct output *o);
 147 static int update_master(struct userdata *u, struct output *o);
 148 static int pick_master(struct userdata *u, struct output *except);
 149
 150 static void adjust_rates(struct userdata *u) {
 151     struct output *o;
 152     pa_usec_t max_sink_latency = 0, min_total_latency = (pa_usec_t) -1, target_latency;
 153     uint32_t base_rate;
 154     uint32_t idx;
 155
 156     pa_assert(u);
 157     pa_sink_assert_ref(u->sink);
 158
 159     if (pa_idxset_size(u->outputs) <= 0)
 160         return;
 161
 162     if (!PA_SINK_OPENED(pa_sink_get_state(u->sink)))
 163         return;
 164
 165     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
 166         uint32_t sink_latency;
 167
 168         if (!o->sink_input || !PA_SINK_OPENED(pa_sink_get_state(o->sink)))
 169             continue;
 170
 171         sink_latency = o->sink_input->sink ? pa_sink_get_latency(o->sink_input->sink) : 0;
 172         o->total_latency = sink_latency + pa_sink_input_get_latency(o->sink_input);
 173
 174         if (sink_latency > max_sink_latency)
 175             max_sink_latency = sink_latency;
 176
 177         if (o->total_latency < min_total_latency)
 178             min_total_latency = o->total_latency;
 179     }
 180
 181     if (min_total_latency == (pa_usec_t) -1)
 182         return;
 183
 184     target_latency = max_sink_latency > min_total_latency ? max_sink_latency : min_total_latency;
 185
 186     pa_log_info("[%s] target latency is %0.0f usec.", u->sink->name, (float) target_latency);
 187     pa_log_info("[%s] master is %s", u->sink->name, u->master->sink->description);
 188
 189     base_rate = u->sink->sample_spec.rate;
 190
 191     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
 192         uint32_t r = base_rate;
 193
 194         if (!o->sink_input || !PA_SINK_OPENED(pa_sink_get_state(o->sink)))
 195             continue;
 196
 197         if (o->total_latency < target_latency)
 198             r -= (uint32_t) (((((double) target_latency - o->total_latency))/u->adjust_time)*r/ 1000000);
 199         else if (o->total_latency > target_latency)
 200             r += (uint32_t) (((((double) o->total_latency - target_latency))/u->adjust_time)*r/ 1000000);
 201
 202         if (r < (uint32_t) (base_rate*0.9) || r > (uint32_t) (base_rate*1.1)) {
 203             pa_log_warn("[%s] sample rates too different, not adjusting (%u vs. %u).", o->sink_input->name, base_rate, r);
 204             pa_sink_input_set_rate(o->sink_input, base_rate);
 205         } else {
 206             pa_log_info("[%s] new rate is %u Hz; ratio is %0.3f; latency is %0.0f usec.", o->sink_input->name, r, (double) r / base_rate, (float) o->total_latency);
 207             pa_sink_input_set_rate(o->sink_input, r);
 208         }
 209     }
 210 }
 211
 212 static void time_callback(pa_mainloop_api*a, pa_time_event* e, const struct timeval *tv, void *userdata) {
 213     struct userdata *u = userdata;
 214     struct timeval n;
 215
 216     pa_assert(u);
 217     pa_assert(a);
 218     pa_assert(u->time_event == e);
 219
 220     adjust_rates(u);
 221
 222     pa_gettimeofday(&n);
 223     n.tv_sec += u->adjust_time;
 224     u->sink->core->mainloop->time_restart(e, &n);
 225 }
 226
 227 static void thread_func(void *userdata) {
 228     struct userdata *u = userdata;
 229
 230     pa_assert(u);
 231
 232     pa_log_debug("Thread starting up");
 233
 234     pa_thread_mq_install(&u->thread_mq);
 235     pa_rtpoll_install(u->rtpoll);
 236
 237     pa_rtclock_get(&u->timestamp);
 238
 239     /* This is only run when we are in NULL mode, to make sure that
 240      * playback doesn't stop. In all other cases we hook our stuff
 241      * into the master sink. */
 242
 243     for (;;) {
 244         int ret;
 245
 246         /* Render some data and drop it immediately */
 247         if (u->sink->thread_info.state == PA_SINK_RUNNING) {
 248             struct timespec now;
 249
 250             pa_rtclock_get(&now);
 251
 252             if (pa_timespec_cmp(&u->timestamp, &now) <= 0) {
 253                 pa_sink_skip(u->sink, u->block_size);
 254                 pa_timespec_add(&u->timestamp, pa_bytes_to_usec(u->block_size, &u->sink->sample_spec));
 255             }
 256
 257             pa_rtpoll_set_timer_absolute(u->rtpoll, &u->timestamp);
 258         } else
 259             pa_rtpoll_set_timer_disabled(u->rtpoll);
 260
 261         /* Hmm, nothing to do. Let's sleep */
 262         if ((ret = pa_rtpoll_run(u->rtpoll, 1)) < 0)
 263             goto fail;
 264
 265         if (ret == 0)
 266             goto finish;
 267     }
 268
 269 fail:
 270     /* If this was no regular exit from the loop we have to continue
 271      * processing messages until we received PA_MESSAGE_SHUTDOWN */
 272     pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
 273     pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
 274
 275 finish:
 276     pa_log_debug("Thread shutting down");
 277 }
 278
 279 /* Called from I/O thread context */
 280 static void render_memblock(struct userdata *u, struct output *o, size_t length) {
 281     pa_assert(u);
 282     pa_assert(o);
 283
 284     if (!PA_SINK_OPENED(u->sink->thread_info.state))
 285         return;
 286
 287     /* We are run by the master output (u->master), possibly on behalf
 288      * of another output (o). The other output is waiting for us,
 289      * hence it is safe to access its mainblockq directly. */
 290
 291     /* Maybe there's some data in the requesting output's queue
 292      * now? */
 293     while (pa_asyncmsgq_process_one(o->inq) > 0)
 294         ;
 295
 296     /* Ok, now let's prepare some data if we really have to */
 297     while (!pa_memblockq_is_readable(o->memblockq)) {
 298         struct output *j;
 299         pa_memchunk chunk;
 300
 301         /* Render data! */
 302         pa_sink_render(u->sink, length, &chunk);
 303
 304         /* OK, let's send this data to the other threads */
 305         for (j = o->userdata->thread_info.outputs; j; j = j->next)
 306
 307             /* Send to other outputs, which are not the requesting
 308              * one, and not the master */
 309
 310             if (j != o && j != u->thread_info.master && j->sink_input)
 311                 pa_asyncmsgq_post(j->inq, PA_MSGOBJECT(j->sink_input), SINK_INPUT_MESSAGE_POST, NULL, 0, &chunk, NULL);
 312
 313         /* Now push it into the master queue */
 314         pa_memblockq_push_align(u->thread_info.master->memblockq, &chunk);
 315
 316         /* And into the requesting output's queue */
 317         if (o != u->thread_info.master)
 318             pa_memblockq_push_align(o->memblockq, &chunk);
 319
 320         pa_memblock_unref(chunk.memblock);
 321     }
 322 }
 323
 324 /* Called from I/O thread context */
 325 static void request_memblock(struct output *o, size_t length) {
 326     pa_assert(o);
 327     pa_sink_input_assert_ref(o->sink_input);
 328     pa_sink_assert_ref(o->userdata->sink);
 329
 330     /* If another thread already prepared some data we received
 331      * the data over the asyncmsgq, hence let's first process
 332      * it. */
 333     while (pa_asyncmsgq_process_one(o->inq) > 0)
 334         ;
 335
 336     /* Check whether we're now readable */
 337     if (pa_memblockq_is_readable(o->memblockq))
 338         return;
 339
 340     /* OK, we need to prepare new data */
 341
 342     if (o == o->userdata->thread_info.master)
 343         /* OK, we're the master, so let's render some data */
 344         render_memblock(o->userdata, o, length);
 345
 346     else
 347         /* We're not the master, we need to ask the master to do the
 348          * rendering for us */
 349
 350         pa_asyncmsgq_send(o->outq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_NEED, o, length, NULL);
 351 }
 352
 353 /* Called from I/O thread context */
 354 static int sink_input_peek_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
 355     struct output *o;
 356
 357     pa_sink_input_assert_ref(i);
 358     pa_assert_se(o = i->userdata);
 359
 360     /* If necessary, get some new data */
 361     request_memblock(o, length);
 362
 363     return  pa_memblockq_peek(o->memblockq, chunk);
 364 }
 365
 366 /* Called from I/O thread context */
 367 static void sink_input_drop_cb(pa_sink_input *i, size_t length) {
 368     struct output *o;
 369
 370     pa_sink_input_assert_ref(i);
 371     pa_assert(length > 0);
 372     pa_assert_se(o = i->userdata);
 373
 374     pa_memblockq_drop(o->memblockq, length);
 375 }
 376
 377 /* Called from I/O thread context for the master */
 378 static void create_master_rtpolls(struct userdata *u) {
 379     struct output *k;
 380
 381     pa_assert(u);
 382
 383     pa_assert(!u->master->outq_rtpoll_item);
 384
 385     /* Set up the queues from the outputs to the master */
 386     for (k = u->thread_info.outputs; k; k = k->next) {
 387
 388         pa_assert(!k->outq_rtpoll_item);
 389
 390         if (k == u->master)
 391             continue;
 392
 393         k->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
 394                 u->master->sink->rtpoll,
 395                 PA_RTPOLL_EARLY+1,  /* This one has a slightly lower priority than the normal message handling */
 396                 k->outq);
 397     }
 398 }
 399
 400 /* Called from I/O thread context for the master */
 401 static void free_master_rtpolls(struct userdata *u) {
 402     struct output *k;
 403
 404     pa_assert(!u->master->outq_rtpoll_item);
 405
 406     for (k = u->thread_info.outputs; k; k = k->next) {
 407
 408         if (k == u->master)
 409             continue;
 410
 411         if (k->outq_rtpoll_item) {
 412             pa_rtpoll_item_free(k->outq_rtpoll_item);
 413             k->outq_rtpoll_item = NULL;
 414         }
 415
 416         pa_assert(!k->outq_rtpoll_item);
 417     }
 418 }
 419
 420 /* Called from I/O thread context */
 421 static void sink_input_attach_cb(pa_sink_input *i) {
 422     struct output *o;
 423
 424     pa_sink_input_assert_ref(i);
 425     pa_assert_se(o = i->userdata);
 426
 427     if (o->userdata->thread_info.master == o) {
 428         create_master_rtpolls(o->userdata);
 429
 430         /* Calling these two functions here is safe, because both
 431          * threads that might access this sink are known to be
 432          * waiting for us. */
 433         pa_sink_set_asyncmsgq(o->userdata->sink, i->sink->asyncmsgq);
 434         pa_sink_set_rtpoll(o->userdata->sink, i->sink->rtpoll);
 435         pa_sink_attach_within_thread(o->userdata->sink);
 436     }
 437
 438     /* Set up the queues from the inputs to the master */
 439     pa_assert(!o->inq_rtpoll_item);
 440     o->inq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
 441             i->sink->rtpoll,
 442             PA_RTPOLL_NORMAL,  /* This one has a lower priority than the normal message handling */
 443             o->inq);
 444 }
 445
 446 /* Called from I/O thread context */
 447 static void sink_input_detach_cb(pa_sink_input *i) {
 448     struct output *o;
 449
 450     pa_sink_input_assert_ref(i);
 451     pa_assert_se(o = i->userdata);
 452
 453 /*     pa_log("detaching %s", i->sink->name); */
 454
 455     pa_assert(o->inq_rtpoll_item);
 456     pa_rtpoll_item_free(o->inq_rtpoll_item);
 457     o->inq_rtpoll_item = NULL;
 458
 459     if (o->userdata->thread_info.master == o) {
 460         pa_sink_detach_within_thread(o->userdata->sink);
 461         free_master_rtpolls(o->userdata);
 462     }
 463 }
 464
 465 /* Called from main context */
 466 static void sink_input_kill_cb(pa_sink_input *i) {
 467     struct output *o;
 468
 469     pa_sink_input_assert_ref(i);
 470     o = i->userdata;
 471     pa_assert(o);
 472
 473     pa_sink_input_unlink(o->sink_input);
 474     pa_sink_input_unref(o->sink_input);
 475     o->sink_input = NULL;
 476
 477     pa_module_unload_request(o->userdata->module);
 478 }
 479
 480 /* Called from thread context */
 481 static int sink_input_process_msg(pa_msgobject *obj, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 482     struct output *o = PA_SINK_INPUT(obj)->userdata;
 483
 484     switch (code) {
 485
 486         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
 487              pa_usec_t *r = data;
 488
 489             *r = pa_bytes_to_usec(pa_memblockq_get_length(o->memblockq), &o->sink_input->sample_spec);
 490
 491             /* Fall through, the default handler will add in the extra
 492              * latency added by the resampler */
 493             break;
 494         }
 495
 496         case SINK_INPUT_MESSAGE_POST: {
 497
 498             if (PA_SINK_OPENED(o->sink_input->sink->thread_info.state))
 499                 pa_memblockq_push_align(o->memblockq, chunk);
 500             else
 501                 pa_memblockq_flush(o->memblockq);
 502
 503             break;
 504         }
 505     }
 506
 507     return pa_sink_input_process_msg(obj, code, data, offset, chunk);
 508 }
 509
 510 /* Called from main context */
 511 static int suspend(struct userdata *u) {
 512     struct output *o;
 513     uint32_t idx;
 514
 515     pa_assert(u);
 516
 517     /* Let's suspend by unlinking all streams */
 518
 519     if (update_master(u, NULL) < 0)
 520         pa_module_unload_request(u->module);
 521
 522     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
 523
 524         if (o->sink_input) {
 525             pa_sink_input_unlink(o->sink_input);
 526             pa_sink_input_unref(o->sink_input);
 527             o->sink_input = NULL;
 528         }
 529     }
 530
 531     pa_log_info("Device suspended...");
 532
 533     return 0;
 534 }
 535
 536 /* Called from main context */
 537 static int unsuspend(struct userdata *u) {
 538     struct output *o;
 539     uint32_t idx;
 540
 541     pa_assert(u);
 542
 543     /* Let's resume */
 544
 545     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
 546
 547         pa_sink_suspend(o->sink, 0);
 548
 549         if (PA_SINK_OPENED(pa_sink_get_state(o->sink))) {
 550             if (output_create_sink_input(u, o) < 0)
 551                 output_free(o);
 552         }
 553     }
 554
 555     if (pick_master(u, NULL) < 0)
 556         pa_module_unload_request(u->module);
 557
 558     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
 559         if (o->sink_input && pa_sink_get_state(o->sink_input) == PA_SINK_INPUT_INIT)
 560              pa_sink_input_put(o->sink_input);
 561
 562     pa_log_info("Resumed successfully...");
 563     return 0;
 564 }
 565
 566 /* Called from main context */
 567 static int sink_set_state(pa_sink *sink, pa_sink_state_t state) {
 568     struct userdata *u;
 569
 570     pa_sink_assert_ref(sink);
 571     pa_assert_se(u = sink->userdata);
 572
 573     /* Please note that in contrast to the ALSA modules we call
 574      * suspend/unsuspend from main context here! */
 575
 576     switch (state) {
 577         case PA_SINK_SUSPENDED:
 578             pa_assert(PA_SINK_OPENED(pa_sink_get_state(u->sink)));
 579
 580             if (suspend(u) < 0)
 581                 return -1;
 582
 583             break;
 584
 585         case PA_SINK_IDLE:
 586         case PA_SINK_RUNNING:
 587
 588             if (pa_sink_get_state(u->sink) == PA_SINK_SUSPENDED) {
 589                 if (unsuspend(u) < 0)
 590                     return -1;
 591             }
 592
 593             break;
 594
 595         case PA_SINK_UNLINKED:
 596         case PA_SINK_INIT:
 597             ;
 598     }
 599
 600     return 0;
 601 }
 602
 603 /* Called from thread context of the master */
 604 static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
 605     struct userdata *u = PA_SINK(o)->userdata;
 606
 607     switch (code) {
 608
 609         case PA_SINK_MESSAGE_SET_STATE:
 610
 611             if ((pa_sink_state_t) PA_PTR_TO_UINT(data) == PA_SINK_RUNNING) {
 612                 /* Only useful when running in NULL mode, i.e. when no
 613                  * master sink is attached */
 614                 pa_rtclock_get(&u->timestamp);
 615             }
 616
 617             break;
 618
 619         case PA_SINK_MESSAGE_GET_LATENCY: {
 620             struct timespec now;
 621
 622             /* This code will only be called when running in NULL
 623              * mode, i.e. when no master sink is attached. See
 624              * sink_get_latency_cb() below */
 625             pa_rtclock_get(&now);
 626
 627             if (pa_timespec_cmp(&u->timestamp, &now) > 0)
 628                 *((pa_usec_t*) data) = 0;
 629             else
 630                 *((pa_usec_t*) data) = pa_timespec_diff(&u->timestamp, &now);
 631             break;
 632         }
 633
 634         case SINK_MESSAGE_SET_MASTER:
 635
 636             if (u->thread_info.master && data != u->thread_info.master) {
 637
 638                 if (u->thread_info.master->sink_input->thread_info.attached)
 639                     free_master_rtpolls(u);
 640
 641             }
 642
 643             if ((u->thread_info.master = data)) {
 644
 645                 /* There's now a master, and we're being executed in
 646                  * its thread, let's register the asyncmsgqs from other
 647                  * outputs to us */
 648
 649                 if (u->thread_info.master->sink_input->thread_info.attached)
 650                     create_master_rtpolls(u);
 651
 652             }
 653
 654             return 0;
 655
 656         case SINK_MESSAGE_ADD_OUTPUT: {
 657             struct output *op = data;
 658
 659             PA_LLIST_PREPEND(struct output, u->thread_info.outputs, op);
 660
 661             pa_assert(!op->outq_rtpoll_item);
 662
 663             if (op != u->thread_info.master) {
 664                 /* Create pa_asyncmsgq to master */
 665
 666                 op->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
 667                         u->thread_info.master->sink->rtpoll,
 668                         PA_RTPOLL_EARLY+1,  /* This one has a slightly lower priority than the normal message handling */
 669                         op->outq);
 670             }
 671
 672             return 0;
 673         }
 674
 675         case SINK_MESSAGE_REMOVE_OUTPUT: {
 676             struct output *op = data;
 677
 678             PA_LLIST_REMOVE(struct output, u->thread_info.outputs, op);
 679
 680             /* Remove the q that leads from this output to the master output */
 681
 682             if (op->outq_rtpoll_item) {
 683                 pa_rtpoll_item_free(op->outq_rtpoll_item);
 684                 op->outq_rtpoll_item = NULL;
 685             }
 686
 687             return 0;
 688         }
 689
 690         case SINK_MESSAGE_NEED:
 691             render_memblock(u, data, (size_t) offset);
 692             return 0;
 693     }
 694
 695     return pa_sink_process_msg(o, code, data, offset, chunk);
 696 }
 697
 698 /* Called from main context */
 699 static pa_usec_t sink_get_latency_cb(pa_sink *s) {
 700     struct userdata *u;
 701
 702     pa_sink_assert_ref(s);
 703     u = s->userdata;
 704     pa_assert(u);
 705
 706     if (u->master) {
 707         /* If we have a master sink, we just return the latency of it
 708          * and add our own buffering on top */
 709
 710         if (!u->master->sink_input)
 711             return 0;
 712
 713         return
 714             pa_sink_input_get_latency(u->master->sink_input) +
 715             pa_sink_get_latency(u->master->sink_input->sink);
 716
 717     } else {
 718         pa_usec_t usec;
 719
 720         /* We have no master, hence let's ask our own thread which
 721          * implements the NULL sink */
 722
 723         if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
 724             return 0;
 725
 726         return usec;
 727     }
 728 }
 729
 730 static void update_description(struct userdata *u) {
 731     int first = 1;
 732     char *t;
 733     struct output *o;
 734     uint32_t idx;
 735
 736     pa_assert(u);
 737
 738     if (pa_idxset_isempty(u->outputs)) {
 739         pa_sink_set_description(u->sink, "Simultaneous output");
 740         return;
 741     }
 742
 743     t = pa_xstrdup("Simultaneous output to");
 744
 745     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
 746         char *e;
 747
 748         if (first) {
 749             e = pa_sprintf_malloc("%s %s", t, o->sink->description);
 750             first = 0;
 751         } else
 752             e = pa_sprintf_malloc("%s, %s", t, o->sink->description);
 753
 754         pa_xfree(t);
 755         t = e;
 756     }
 757
 758     pa_sink_set_description(u->sink, t);
 759     pa_xfree(t);
 760 }
 761
 762 static int update_master(struct userdata *u, struct output *o) {
 763     pa_assert(u);
 764
 765     /* Make sure everything is detached from the old thread before we move our stuff to a new thread */
 766     if (u->sink && PA_SINK_LINKED(pa_sink_get_state(u->sink))) {
 767         pa_sink_detach(u->sink);
 768         pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_SET_MASTER, NULL, 0, NULL);
 769     } else
 770         u->thread_info.master = NULL;
 771
 772     if (o) {
 773         /* If we have a master sink we run our own sink in its thread */
 774
 775         pa_assert(o->sink_input);
 776         pa_assert(PA_SINK_OPENED(pa_sink_get_state(o->sink)));
 777
 778         if (u->thread) {
 779             /* If we previously were in NULL mode, let's kill the thread */
 780             pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
 781             pa_thread_free(u->thread);
 782             u->thread = NULL;
 783
 784             pa_assert(u->rtpoll);
 785             pa_rtpoll_free(u->rtpoll);
 786             u->rtpoll = NULL;
 787         }
 788
 789         pa_sink_set_asyncmsgq(u->sink, o->sink->asyncmsgq);
 790         pa_sink_set_rtpoll(u->sink, o->sink->rtpoll);
 791         u->master = o;
 792
 793         pa_log_info("Master sink is now '%s'", o->sink_input->sink->name);
 794
 795     } else {
 796
 797         /* We have no master sink, let's create our own thread */
 798
 799         pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
 800         u->master = NULL;
 801
 802         if (!u->thread) {
 803             pa_assert(!u->rtpoll);
 804
 805             u->rtpoll = pa_rtpoll_new();
 806             pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
 807
 808             pa_sink_set_rtpoll(u->sink, u->rtpoll);
 809
 810             if (!(u->thread = pa_thread_new(thread_func, u))) {
 811                 pa_log("Failed to create thread.");
 812                 return -1;
 813             }
 814         }
 815
 816         pa_log_info("No suitable master sink found, going to NULL mode\n");
 817     }
 818
 819     /* Now attach everything again */
 820     if (u->sink && PA_SINK_LINKED(pa_sink_get_state(u->sink))) {
 821         pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_SET_MASTER, u->master, 0, NULL);
 822         pa_sink_attach(u->sink);
 823     } else
 824         u->thread_info.master = u->master;
 825
 826     return 0;
 827 }
 828
 829 static int pick_master(struct userdata *u, struct output *except) {
 830     struct output *o;
 831     uint32_t idx;
 832     pa_assert(u);
 833
 834     if (u->master && u->master != except && u->master->sink_input && PA_SINK_OPENED(pa_sink_get_state(u->master->sink)))
 835         return update_master(u, u->master);
 836
 837     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
 838         if (o != except && o->sink_input && PA_SINK_OPENED(pa_sink_get_state(o->sink)))
 839             return update_master(u, o);
 840
 841     return update_master(u, NULL);
 842 }
 843
 844 static int output_create_sink_input(struct userdata *u, struct output *o) {
 845     pa_sink_input_new_data data;
 846     char *t;
 847
 848     pa_assert(u);
 849     pa_assert(!o->sink_input);
 850
 851     t = pa_sprintf_malloc("Simultaneous output on %s", o->sink->description);
 852
 853     pa_sink_input_new_data_init(&data);
 854     data.sink = o->sink;
 855     data.driver = __FILE__;
 856     data.name = t;
 857     pa_sink_input_new_data_set_sample_spec(&data, &u->sink->sample_spec);
 858     pa_sink_input_new_data_set_channel_map(&data, &u->sink->channel_map);
 859     data.module = u->module;
 860     data.resample_method = u->resample_method;
 861
 862     o->sink_input = pa_sink_input_new(u->core, &data, PA_SINK_INPUT_VARIABLE_RATE|PA_SINK_INPUT_DONT_MOVE);
 863
 864     pa_xfree(t);
 865
 866     if (!o->sink_input)
 867         return -1;
 868
 869     o->sink_input->parent.process_msg = sink_input_process_msg;
 870     o->sink_input->peek = sink_input_peek_cb;
 871     o->sink_input->drop = sink_input_drop_cb;
 872     o->sink_input->attach = sink_input_attach_cb;
 873     o->sink_input->detach = sink_input_detach_cb;
 874     o->sink_input->kill = sink_input_kill_cb;
 875     o->sink_input->userdata = o;
 876
 877     return 0;
 878 }
 879
 880 static struct output *output_new(struct userdata *u, pa_sink *sink) {
 881     struct output *o;
 882
 883     pa_assert(u);
 884     pa_assert(sink);
 885     pa_assert(u->sink);
 886
 887     o = pa_xnew(struct output, 1);
 888     o->userdata = u;
 889     o->inq = pa_asyncmsgq_new(0);
 890     o->outq = pa_asyncmsgq_new(0);
 891     o->inq_rtpoll_item = NULL;
 892     o->outq_rtpoll_item = NULL;
 893     o->sink = sink;
 894     o->sink_input = NULL;
 895     o->memblockq = pa_memblockq_new(
 896             0,
 897             MEMBLOCKQ_MAXLENGTH,
 898             MEMBLOCKQ_MAXLENGTH,
 899             pa_frame_size(&u->sink->sample_spec),
 900             1,
 901             0,
 902             NULL);
 903
 904
 905     pa_assert_se(pa_idxset_put(u->outputs, o, NULL) == 0);
 906
 907     update_description(u);
 908
 909     if (u->sink && PA_SINK_LINKED(pa_sink_get_state(u->sink)))
 910         pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_ADD_OUTPUT, o, 0, NULL);
 911     else
 912         PA_LLIST_PREPEND(struct output, u->thread_info.outputs, o);
 913
 914     if (PA_SINK_OPENED(pa_sink_get_state(u->sink)) || pa_sink_get_state(u->sink) == PA_SINK_INIT) {
 915         pa_sink_suspend(sink, 0);
 916
 917         if (PA_SINK_OPENED(pa_sink_get_state(sink)))
 918             if (output_create_sink_input(u, o) < 0)
 919                 goto fail;
 920     }
 921
 922     return o;
 923
 924 fail:
 925
 926     if (o) {
 927         if (o->sink_input) {
 928             pa_sink_input_unlink(o->sink_input);
 929             pa_sink_input_unref(o->sink_input);
 930         }
 931
 932         if (o->memblockq)
 933             pa_memblockq_free(o->memblockq);
 934
 935         if (o->inq)
 936             pa_asyncmsgq_unref(o->inq);
 937
 938         if (o->outq)
 939             pa_asyncmsgq_unref(o->outq);
 940
 941         pa_xfree(o);
 942     }
 943
 944     return NULL;
 945 }
 946
 947 static pa_hook_result_t sink_new_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
 948     struct output *o;
 949
 950     pa_core_assert_ref(c);
 951     pa_sink_assert_ref(s);
 952     pa_assert(u);
 953     pa_assert(u->automatic);
 954
 955     if (!(s->flags & PA_SINK_HARDWARE) || s == u->sink)
 956         return PA_HOOK_OK;
 957
 958     pa_log_info("Configuring new sink: %s", s->name);
 959
 960     if (!(o = output_new(u, s))) {
 961         pa_log("Failed to create sink input on sink '%s'.", s->name);
 962         return PA_HOOK_OK;
 963     }
 964
 965     if (pick_master(u, NULL) < 0)
 966         pa_module_unload_request(u->module);
 967
 968     if (o->sink_input)
 969         pa_sink_input_put(o->sink_input);
 970
 971     return PA_HOOK_OK;
 972 }
 973
 974 static pa_hook_result_t sink_unlink_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
 975     struct output *o;
 976     uint32_t idx;
 977
 978     pa_assert(c);
 979     pa_sink_assert_ref(s);
 980     pa_assert(u);
 981
 982     if (s == u->sink)
 983         return PA_HOOK_OK;
 984
 985     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
 986         if (o->sink == s)
 987             break;
 988
 989     if (!o)
 990         return PA_HOOK_OK;
 991
 992     pa_log_info("Unconfiguring sink: %s", s->name);
 993
 994     output_free(o);
 995
 996     return PA_HOOK_OK;
 997 }
 998
 999 static pa_hook_result_t sink_state_changed_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
1000     struct output *o;
1001     uint32_t idx;
1002     pa_sink_state_t state;
1003
1004     if (s == u->sink)
1005         return PA_HOOK_OK;
1006
1007     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
1008         if (o->sink == s)
1009             break;
1010
1011     if (!o)
1012         return PA_HOOK_OK;
1013
1014     state = pa_sink_get_state(s);
1015
1016     if (PA_SINK_OPENED(state) && PA_SINK_OPENED(pa_sink_get_state(u->sink)) && !o->sink_input) {
1017         output_create_sink_input(u, o);
1018
1019         if (pick_master(u, NULL) < 0)
1020             pa_module_unload_request(u->module);
1021
1022         if (o->sink_input)
1023             pa_sink_input_put(o->sink_input);
1024     }
1025
1026     if (state == PA_SINK_SUSPENDED && o->sink_input) {
1027         pa_sink_input_unlink(o->sink_input);
1028         pa_sink_input_unref(o->sink_input);
1029         o->sink_input = NULL;
1030
1031         pa_memblockq_flush(o->memblockq);
1032
1033         if (pick_master(u, o) < 0)
1034             pa_module_unload_request(u->module);
1035     }
1036
1037     return PA_HOOK_OK;
1038 }
1039
1040 int pa__init(pa_module*m) {
1041     struct userdata *u;
1042     pa_modargs *ma = NULL;
1043     const char *master_name, *slaves, *rm;
1044     pa_sink *master_sink = NULL;
1045     int resample_method = PA_RESAMPLER_TRIVIAL;
1046     pa_sample_spec ss;
1047     pa_channel_map map;
1048     struct output *o;
1049     uint32_t idx;
1050
1051     pa_assert(m);
1052
1053     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
1054         pa_log("failed to parse module arguments");
1055         goto fail;
1056     }
1057
1058     if ((rm = pa_modargs_get_value(ma, "resample_method", NULL))) {
1059         if ((resample_method = pa_parse_resample_method(rm)) < 0) {
1060             pa_log("invalid resample method '%s'", rm);
1061             goto fail;
1062         }
1063     }
1064
1065     u = pa_xnew(struct userdata, 1);
1066     u->core = m->core;
1067     u->module = m;
1068     m->userdata = u;
1069     u->sink = NULL;
1070     u->thread_info.master = u->master = NULL;
1071     u->time_event = NULL;
1072     u->adjust_time = DEFAULT_ADJUST_TIME;
1073     pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
1074     u->rtpoll = NULL;
1075     u->thread = NULL;
1076     PA_LLIST_HEAD_INIT(struct output, u->thread_info.outputs);
1077     u->resample_method = resample_method;
1078     u->outputs = pa_idxset_new(NULL, NULL);
1079     pa_timespec_reset(&u->adjust_timestamp);
1080
1081     if (pa_modargs_get_value_u32(ma, "adjust_time", &u->adjust_time) < 0) {
1082         pa_log("Failed to parse adjust_time value");
1083         goto fail;
1084     }
1085
1086     master_name = pa_modargs_get_value(ma, "master", NULL);
1087     slaves = pa_modargs_get_value(ma, "slaves", NULL);
1088     if (!master_name != !slaves) {
1089         pa_log("No master or slave sinks specified");
1090         goto fail;
1091     }
1092
1093     if (master_name) {
1094         if (!(master_sink = pa_namereg_get(m->core, master_name, PA_NAMEREG_SINK, 1))) {
1095             pa_log("Invalid master sink '%s'", master_name);
1096             goto fail;
1097         }
1098
1099         ss = master_sink->sample_spec;
1100         u->automatic = 0;
1101     } else {
1102         master_sink = NULL;
1103         ss = m->core->default_sample_spec;
1104         u->automatic = 1;
1105     }
1106
1107     if ((pa_modargs_get_sample_spec(ma, &ss) < 0)) {
1108         pa_log("Invalid sample specification.");
1109         goto fail;
1110     }
1111
1112     if (master_sink && ss.channels == master_sink->sample_spec.channels)
1113         map = master_sink->channel_map;
1114     else
1115         pa_channel_map_init_auto(&map, ss.channels, PA_CHANNEL_MAP_DEFAULT);
1116
1117     if ((pa_modargs_get_channel_map(ma, NULL, &map) < 0)) {
1118         pa_log("Invalid channel map.");
1119         goto fail;
1120     }
1121
1122     if (ss.channels != map.channels) {
1123         pa_log("Channel map and sample specification don't match.");
1124         goto fail;
1125     }
1126
1127     if (!(u->sink = pa_sink_new(m->core, __FILE__, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME), 0, &ss, &map))) {
1128         pa_log("Failed to create sink");
1129         goto fail;
1130     }
1131
1132     u->sink->parent.process_msg = sink_process_msg;
1133     u->sink->get_latency = sink_get_latency_cb;
1134     u->sink->set_state = sink_set_state;
1135     u->sink->userdata = u;
1136
1137     u->sink->flags = PA_SINK_LATENCY;
1138     pa_sink_set_module(u->sink, m);
1139     pa_sink_set_description(u->sink, "Simultaneous output");
1140
1141     u->block_size = pa_bytes_per_second(&ss) / 20; /* 50 ms */
1142     if (u->block_size <= 0)
1143         u->block_size = pa_frame_size(&ss);
1144
1145     if (!u->automatic) {
1146         const char*split_state;
1147         char *n = NULL;
1148         pa_assert(slaves);
1149
1150         /* The master and slaves have been specified manually */
1151
1152         if (!(u->master = output_new(u, master_sink))) {
1153             pa_log("Failed to create master sink input on sink '%s'.", master_sink->name);
1154             goto fail;
1155         }
1156
1157         split_state = NULL;
1158         while ((n = pa_split(slaves, ",", &split_state))) {
1159             pa_sink *slave_sink;
1160
1161             if (!(slave_sink = pa_namereg_get(m->core, n, PA_NAMEREG_SINK, 1)) || slave_sink == u->sink) {
1162                 pa_log("Invalid slave sink '%s'", n);
1163                 pa_xfree(n);
1164                 goto fail;
1165             }
1166
1167             pa_xfree(n);
1168
1169             if (!output_new(u, slave_sink)) {
1170                 pa_log("Failed to create slave sink input on sink '%s'.", slave_sink->name);
1171                 goto fail;
1172             }
1173         }
1174
1175         if (pa_idxset_size(u->outputs) <= 1)
1176             pa_log_warn("No slave sinks specified.");
1177
1178         u->sink_new_slot = NULL;
1179
1180     } else {
1181         pa_sink *s;
1182
1183         /* We're in automatic mode, we elect one hw sink to the master
1184          * and attach all other hw sinks as slaves to it */
1185
1186         for (s = pa_idxset_first(m->core->sinks, &idx); s; s = pa_idxset_next(m->core->sinks, &idx)) {
1187
1188             if (!(s->flags & PA_SINK_HARDWARE) || s == u->sink)
1189                 continue;
1190
1191             if (!output_new(u, s)) {
1192                 pa_log("Failed to create sink input on sink '%s'.", s->name);
1193                 goto fail;
1194             }
1195         }
1196
1197         u->sink_new_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_NEW_POST], (pa_hook_cb_t) sink_new_hook_cb, u);
1198     }
1199
1200     u->sink_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_UNLINK], (pa_hook_cb_t) sink_unlink_hook_cb, u);
1201     u->sink_state_changed_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_STATE_CHANGED], (pa_hook_cb_t) sink_state_changed_hook_cb, u);
1202
1203     if (pick_master(u, NULL) < 0)
1204         goto fail;
1205
1206     /* Activate the sink and the sink inputs */
1207     pa_sink_put(u->sink);
1208
1209     for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
1210         if (o->sink_input)
1211             pa_sink_input_put(o->sink_input);
1212
1213     if (u->adjust_time > 0) {
1214         struct timeval tv;
1215         pa_gettimeofday(&tv);
1216         tv.tv_sec += u->adjust_time;
1217         u->time_event = m->core->mainloop->time_new(m->core->mainloop, &tv, time_callback, u);
1218     }
1219
1220     pa_modargs_free(ma);
1221
1222     return 0;
1223
1224 fail:
1225
1226     if (ma)
1227         pa_modargs_free(ma);
1228
1229     pa__done(m);
1230
1231     return -1;
1232 }
1233
1234 static void output_free(struct output *o) {
1235     pa_assert(o);
1236
1237     /* Make sure the master points to a different output */
1238     if (pick_master(o->userdata, o) < 0)
1239         pa_module_unload_request(o->userdata->module);
1240
1241     if (o->userdata->sink && PA_SINK_LINKED(pa_sink_get_state(o->userdata->sink)))
1242         pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
1243     else
1244         PA_LLIST_REMOVE(struct output, o->userdata->thread_info.outputs, o);
1245
1246     pa_assert_se(pa_idxset_remove_by_data(o->userdata->outputs, o, NULL));
1247
1248     if (o->sink_input) {
1249         pa_sink_input_unlink(o->sink_input);
1250         pa_sink_input_unref(o->sink_input);
1251     }
1252
1253     update_description(o->userdata);
1254
1255     if (o->inq_rtpoll_item)
1256         pa_rtpoll_item_free(o->inq_rtpoll_item);
1257
1258     if (o->outq_rtpoll_item)
1259         pa_rtpoll_item_free(o->outq_rtpoll_item);
1260
1261     if (o->inq)
1262         pa_asyncmsgq_unref(o->inq);
1263
1264     if (o->outq)
1265         pa_asyncmsgq_unref(o->outq);
1266
1267     if (o->memblockq)
1268         pa_memblockq_free(o->memblockq);
1269
1270     pa_xfree(o);
1271 }
1272
1273 void pa__done(pa_module*m) {
1274     struct userdata *u;
1275     struct output *o;
1276
1277     pa_assert(m);
1278
1279     if (!(u = m->userdata))
1280         return;
1281
1282     if (u->sink_new_slot)
1283         pa_hook_slot_free(u->sink_new_slot);
1284
1285     if (u->sink_unlink_slot)
1286         pa_hook_slot_free(u->sink_unlink_slot);
1287
1288     if (u->sink_state_changed_slot)
1289         pa_hook_slot_free(u->sink_state_changed_slot);
1290
1291     if (u->sink)
1292         pa_sink_unlink(u->sink);
1293
1294     if (u->outputs) {
1295         while ((o = pa_idxset_first(u->outputs, NULL)))
1296             output_free(o);
1297
1298         pa_idxset_free(u->outputs, NULL, NULL);
1299     }
1300
1301     if (u->thread) {
1302         pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
1303         pa_thread_free(u->thread);
1304     }
1305
1306     pa_thread_mq_done(&u->thread_mq);
1307
1308     if (u->sink)
1309         pa_sink_unref(u->sink);
1310
1311     if (u->rtpoll)
1312         pa_rtpoll_free(u->rtpoll);
1313
1314     if (u->time_event)
1315         u->core->mainloop->time_free(u->time_event);
1316
1317     pa_xfree(u);
1318 }