]> code.delx.au - pulseaudio/blobdiff - src/modules/module-combine.c
bluetooth: remove left-over debug line
[pulseaudio] / src / modules / module-combine.c
index 665bf9dd4fcc8ee8bb32d4b9acc7150ffc0b6b0c..4a57ce2dc60120023ccfb9991ce57b7613d8e67e 100644 (file)
@@ -1,13 +1,11 @@
-/* $Id$ */
-
 /***
   This file is part of PulseAudio.
 
 /***
   This file is part of PulseAudio.
 
-  Copyright 2004-2006 Lennart Poettering
+  Copyright 2004-2008 Lennart Poettering
 
   PulseAudio is free software; you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published
 
   PulseAudio is free software; you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published
-  by the Free Software Foundation; either version 2 of the License,
+  by the Free Software Foundation; either version 2.1 of the License,
   or (at your option) any later version.
 
   PulseAudio is distributed in the hope that it will be useful, but
   or (at your option) any later version.
 
   PulseAudio is distributed in the hope that it will be useful, but
@@ -28,6 +26,7 @@
 #include <stdio.h>
 #include <errno.h>
 
 #include <stdio.h>
 #include <errno.h>
 
+#include <pulse/rtclock.h>
 #include <pulse/timeval.h>
 #include <pulse/xmalloc.h>
 
 #include <pulse/timeval.h>
 #include <pulse/xmalloc.h>
 
@@ -38,6 +37,7 @@
 #include <pulsecore/sink-input.h>
 #include <pulsecore/memblockq.h>
 #include <pulsecore/log.h>
 #include <pulsecore/sink-input.h>
 #include <pulsecore/memblockq.h>
 #include <pulsecore/log.h>
+#include <pulsecore/core-rtclock.h>
 #include <pulsecore/core-util.h>
 #include <pulsecore/modargs.h>
 #include <pulsecore/namereg.h>
 #include <pulsecore/core-util.h>
 #include <pulsecore/modargs.h>
 #include <pulsecore/namereg.h>
 #include <pulsecore/thread.h>
 #include <pulsecore/thread-mq.h>
 #include <pulsecore/rtpoll.h>
 #include <pulsecore/thread.h>
 #include <pulsecore/thread-mq.h>
 #include <pulsecore/rtpoll.h>
-#include <pulsecore/rtclock.h>
 #include <pulsecore/core-error.h>
 #include <pulsecore/core-error.h>
+#include <pulsecore/time-smoother.h>
 
 #include "module-combine-symdef.h"
 
 
 #include "module-combine-symdef.h"
 
-PA_MODULE_AUTHOR("Lennart Poettering")
-PA_MODULE_DESCRIPTION("Combine multiple sinks to one")
-PA_MODULE_VERSION(PACKAGE_VERSION)
+PA_MODULE_AUTHOR("Lennart Poettering");
+PA_MODULE_DESCRIPTION("Combine multiple sinks to one");
+PA_MODULE_VERSION(PACKAGE_VERSION);
+PA_MODULE_LOAD_ONCE(FALSE);
 PA_MODULE_USAGE(
         "sink_name=<name for the sink> "
 PA_MODULE_USAGE(
         "sink_name=<name for the sink> "
-        "master=<master sink> "
+        "sink_properties=<properties for the sink> "
         "slaves=<slave sinks> "
         "slaves=<slave sinks> "
-        "adjust_time=<seconds> "
+        "adjust_time=<how often to readjust rates in s> "
         "resample_method=<method> "
         "format=<sample format> "
         "resample_method=<method> "
         "format=<sample format> "
-        "channels=<number of channels> "
         "rate=<sample rate> "
         "rate=<sample rate> "
-        "channel_map=<channel map>")
+        "channels=<number of channels> "
+        "channel_map=<channel map>");
 
 #define DEFAULT_SINK_NAME "combined"
 
 #define DEFAULT_SINK_NAME "combined"
-#define MEMBLOCKQ_MAXLENGTH (1024*170)
 
 
-#define DEFAULT_ADJUST_TIME 10
+#define MEMBLOCKQ_MAXLENGTH (1024*1024*16)
+
+#define DEFAULT_ADJUST_TIME_USEC (10*PA_USEC_PER_SEC)
+
+#define BLOCK_USEC (PA_USEC_PER_MSEC * 200)
 
 static const char* const valid_modargs[] = {
     "sink_name",
 
 static const char* const valid_modargs[] = {
     "sink_name",
-    "master",
+    "sink_properties",
     "slaves",
     "adjust_time",
     "resample_method",
     "format",
     "slaves",
     "adjust_time",
     "resample_method",
     "format",
-    "channels",
     "rate",
     "rate",
+    "channels",
     "channel_map",
     NULL
 };
     "channel_map",
     NULL
 };
@@ -87,15 +91,22 @@ struct output {
 
     pa_sink *sink;
     pa_sink_input *sink_input;
 
     pa_sink *sink;
     pa_sink_input *sink_input;
+    pa_bool_t ignore_state_change;
 
     pa_asyncmsgq *inq,    /* Message queue from the sink thread to this sink input */
                  *outq;   /* Message queue from this sink input to the sink thread */
 
     pa_asyncmsgq *inq,    /* Message queue from the sink thread to this sink input */
                  *outq;   /* Message queue from this sink input to the sink thread */
-    pa_rtpoll_item *inq_rtpoll_item, *outq_rtpoll_item;
+    pa_rtpoll_item *inq_rtpoll_item_read, *inq_rtpoll_item_write;
+    pa_rtpoll_item *outq_rtpoll_item_read, *outq_rtpoll_item_write;
 
     pa_memblockq *memblockq;
 
 
     pa_memblockq *memblockq;
 
+    /* For communication of the stream latencies to the main thread */
     pa_usec_t total_latency;
 
     pa_usec_t total_latency;
 
+    /* For coomunication of the stream parameters to the sink thread */
+    pa_atomic_t max_request;
+    pa_atomic_t requested_latency;
+
     PA_LLIST_FIELDS(struct output);
 };
 
     PA_LLIST_FIELDS(struct output);
 };
 
@@ -109,48 +120,53 @@ struct userdata {
     pa_rtpoll *rtpoll;
 
     pa_time_event *time_event;
     pa_rtpoll *rtpoll;
 
     pa_time_event *time_event;
-    uint32_t adjust_time;
+    pa_usec_t adjust_time;
 
     pa_bool_t automatic;
 
     pa_bool_t automatic;
-    size_t block_size;
+    pa_bool_t auto_desc;
 
 
-    pa_hook_slot *sink_new_slot, *sink_unlink_slot, *sink_state_changed_slot;
+    pa_hook_slot *sink_put_slot, *sink_unlink_slot, *sink_state_changed_slot;
 
     pa_resample_method_t resample_method;
 
 
     pa_resample_method_t resample_method;
 
-    struct timeval adjust_timestamp;
+    pa_usec_t block_usec;
 
 
-    struct output *master;
     pa_idxset* outputs; /* managed in main context */
 
     struct {
         PA_LLIST_HEAD(struct output, active_outputs); /* managed in IO thread context */
         pa_atomic_t running;  /* we cache that value here, so that every thread can query it cheaply */
     pa_idxset* outputs; /* managed in main context */
 
     struct {
         PA_LLIST_HEAD(struct output, active_outputs); /* managed in IO thread context */
         pa_atomic_t running;  /* we cache that value here, so that every thread can query it cheaply */
-        struct timeval timestamp;
+        pa_usec_t timestamp;
         pa_bool_t in_null_mode;
         pa_bool_t in_null_mode;
+        pa_smoother *smoother;
+        uint64_t counter;
     } thread_info;
 };
 
 enum {
     SINK_MESSAGE_ADD_OUTPUT = PA_SINK_MESSAGE_MAX,
     SINK_MESSAGE_REMOVE_OUTPUT,
     } thread_info;
 };
 
 enum {
     SINK_MESSAGE_ADD_OUTPUT = PA_SINK_MESSAGE_MAX,
     SINK_MESSAGE_REMOVE_OUTPUT,
-    SINK_MESSAGE_NEED
+    SINK_MESSAGE_NEED,
+    SINK_MESSAGE_UPDATE_LATENCY,
+    SINK_MESSAGE_UPDATE_MAX_REQUEST,
+    SINK_MESSAGE_UPDATE_REQUESTED_LATENCY
 };
 
 enum {
 };
 
 enum {
-    SINK_INPUT_MESSAGE_POST = PA_SINK_INPUT_MESSAGE_MAX
+    SINK_INPUT_MESSAGE_POST = PA_SINK_INPUT_MESSAGE_MAX,
 };
 
 };
 
+static void output_disable(struct output *o);
+static void output_enable(struct output *o);
 static void output_free(struct output *o);
 static int output_create_sink_input(struct output *o);
 static void output_free(struct output *o);
 static int output_create_sink_input(struct output *o);
-static void update_master(struct userdata *u, struct output *o);
-static void pick_master(struct userdata *u, struct output *except);
 
 static void adjust_rates(struct userdata *u) {
     struct output *o;
 
 static void adjust_rates(struct userdata *u) {
     struct output *o;
-    pa_usec_t max_sink_latency = 0, min_total_latency = (pa_usec_t) -1, target_latency;
+    pa_usec_t max_sink_latency = 0, min_total_latency = (pa_usec_t) -1, target_latency, avg_total_latency = 0;
     uint32_t base_rate;
     uint32_t idx;
     uint32_t base_rate;
     uint32_t idx;
+    unsigned n = 0;
 
     pa_assert(u);
     pa_sink_assert_ref(u->sink);
 
     pa_assert(u);
     pa_sink_assert_ref(u->sink);
@@ -158,62 +174,70 @@ static void adjust_rates(struct userdata *u) {
     if (pa_idxset_size(u->outputs) <= 0)
         return;
 
     if (pa_idxset_size(u->outputs) <= 0)
         return;
 
-    if (!u->master)
-        return;
-
-    if (!PA_SINK_OPENED(pa_sink_get_state(u->sink)))
+    if (!PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)))
         return;
 
         return;
 
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
+    PA_IDXSET_FOREACH(o, u->outputs, idx) {
         pa_usec_t sink_latency;
 
         pa_usec_t sink_latency;
 
-        if (!o->sink_input || !PA_SINK_OPENED(pa_sink_get_state(o->sink)))
+        if (!o->sink_input || !PA_SINK_IS_OPENED(pa_sink_get_state(o->sink)))
             continue;
 
             continue;
 
-        sink_latency = pa_sink_get_latency(o->sink);
-        o->total_latency = sink_latency + pa_sink_input_get_latency(o->sink_input);
+        o->total_latency = pa_sink_input_get_latency(o->sink_input, &sink_latency);
+        o->total_latency += sink_latency;
 
         if (sink_latency > max_sink_latency)
             max_sink_latency = sink_latency;
 
         if (min_total_latency == (pa_usec_t) -1 || o->total_latency < min_total_latency)
             min_total_latency = o->total_latency;
 
         if (sink_latency > max_sink_latency)
             max_sink_latency = sink_latency;
 
         if (min_total_latency == (pa_usec_t) -1 || o->total_latency < min_total_latency)
             min_total_latency = o->total_latency;
+
+        avg_total_latency += o->total_latency;
+        n++;
+
+        pa_log_debug("[%s] total=%0.2fms sink=%0.2fms ", o->sink->name, (double) o->total_latency / PA_USEC_PER_MSEC, (double) sink_latency / PA_USEC_PER_MSEC);
+
+        if (o->total_latency > 10*PA_USEC_PER_SEC)
+            pa_log_warn("[%s] Total latency of output is very high (%0.2fms), most likely the audio timing in one of your drivers is broken.", o->sink->name, (double) o->total_latency / PA_USEC_PER_MSEC);
     }
 
     if (min_total_latency == (pa_usec_t) -1)
         return;
 
     }
 
     if (min_total_latency == (pa_usec_t) -1)
         return;
 
+    avg_total_latency /= n;
+
     target_latency = max_sink_latency > min_total_latency ? max_sink_latency : min_total_latency;
 
     target_latency = max_sink_latency > min_total_latency ? max_sink_latency : min_total_latency;
 
-    pa_log_info("[%s] target latency is %0.0f usec.", u->sink->name, (float) target_latency);
-    pa_log_info("[%s] master %s latency %0.0f usec.", u->sink->name, u->master->sink->name, (float) u->master->total_latency);
+    pa_log_info("[%s] avg total latency is %0.2f msec.", u->sink->name, (double) avg_total_latency / PA_USEC_PER_MSEC);
+    pa_log_info("[%s] target latency is %0.2f msec.", u->sink->name, (double) target_latency / PA_USEC_PER_MSEC);
 
     base_rate = u->sink->sample_spec.rate;
 
 
     base_rate = u->sink->sample_spec.rate;
 
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
+    PA_IDXSET_FOREACH(o, u->outputs, idx) {
         uint32_t r = base_rate;
 
         uint32_t r = base_rate;
 
-        if (!o->sink_input || !PA_SINK_OPENED(pa_sink_get_state(o->sink)))
+        if (!o->sink_input || !PA_SINK_IS_OPENED(pa_sink_get_state(o->sink)))
             continue;
 
         if (o->total_latency < target_latency)
             continue;
 
         if (o->total_latency < target_latency)
-            r -= (uint32_t) (((((double) target_latency - o->total_latency))/u->adjust_time)*r/PA_USEC_PER_SEC);
+            r -= (uint32_t) ((((double) (target_latency - o->total_latency))/(double)u->adjust_time)*(double)r);
         else if (o->total_latency > target_latency)
         else if (o->total_latency > target_latency)
-            r += (uint32_t) (((((double) o->total_latency - target_latency))/u->adjust_time)*r/PA_USEC_PER_SEC);
+            r += (uint32_t) ((((double) (o->total_latency - target_latency))/(double)u->adjust_time)*(double)r);
 
         if (r < (uint32_t) (base_rate*0.9) || r > (uint32_t) (base_rate*1.1)) {
 
         if (r < (uint32_t) (base_rate*0.9) || r > (uint32_t) (base_rate*1.1)) {
-            pa_log_warn("[%s] sample rates too different, not adjusting (%u vs. %u).", o->sink_input->name, base_rate, r);
+            pa_log_warn("[%s] sample rates too different, not adjusting (%u vs. %u).", o->sink_input->sink->name, base_rate, r);
             pa_sink_input_set_rate(o->sink_input, base_rate);
         } else {
             pa_sink_input_set_rate(o->sink_input, base_rate);
         } else {
-            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);
+            pa_log_info("[%s] new rate is %u Hz; ratio is %0.3f; latency is %0.0f usec.", o->sink_input->sink->name, r, (double) r / base_rate, (float) o->total_latency);
             pa_sink_input_set_rate(o->sink_input, r);
         }
     }
             pa_sink_input_set_rate(o->sink_input, r);
         }
     }
+
+    pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UPDATE_LATENCY, NULL, (int64_t) avg_total_latency, NULL);
 }
 
 }
 
-static void time_callback(pa_mainloop_api*a, pa_time_event* e, const struct timeval *tv, void *userdata) {
+static void time_callback(pa_mainloop_api *a, pa_time_event *e, const struct timeval *t, void *userdata) {
     struct userdata *u = userdata;
     struct userdata *u = userdata;
-    struct timeval n;
 
     pa_assert(u);
     pa_assert(a);
 
     pa_assert(u);
     pa_assert(a);
@@ -221,9 +245,37 @@ static void time_callback(pa_mainloop_api*a, pa_time_event* e, const struct time
 
     adjust_rates(u);
 
 
     adjust_rates(u);
 
-    pa_gettimeofday(&n);
-    n.tv_sec += u->adjust_time;
-    u->sink->core->mainloop->time_restart(e, &n);
+    pa_core_rttime_restart(u->core, e, pa_rtclock_now() + u->adjust_time);
+}
+
+static void process_render_null(struct userdata *u, pa_usec_t now) {
+    size_t ate = 0;
+    pa_assert(u);
+
+    if (u->thread_info.in_null_mode)
+        u->thread_info.timestamp = now;
+
+    while (u->thread_info.timestamp < now + u->block_usec) {
+        pa_memchunk chunk;
+
+        pa_sink_render(u->sink, u->sink->thread_info.max_request, &chunk);
+        pa_memblock_unref(chunk.memblock);
+
+        u->thread_info.counter += chunk.length;
+
+/*         pa_log_debug("Ate %lu bytes.", (unsigned long) chunk.length); */
+        u->thread_info.timestamp += pa_bytes_to_usec(chunk.length, &u->sink->sample_spec);
+
+        ate += chunk.length;
+
+        if (ate >= u->sink->thread_info.max_request)
+            break;
+    }
+
+/*     pa_log_debug("Ate in sum %lu bytes (of %lu)", (unsigned long) ate, (unsigned long) nbytes); */
+
+    pa_smoother_put(u->thread_info.smoother, now,
+                    pa_bytes_to_usec(u->thread_info.counter, &u->sink->sample_spec) - (u->thread_info.timestamp - now));
 }
 
 static void thread_func(void *userdata) {
 }
 
 static void thread_func(void *userdata) {
@@ -233,36 +285,32 @@ static void thread_func(void *userdata) {
 
     pa_log_debug("Thread starting up");
 
 
     pa_log_debug("Thread starting up");
 
-    if (u->core->high_priority)
-        pa_make_realtime();
+    if (u->core->realtime_scheduling)
+        pa_make_realtime(u->core->realtime_priority+1);
 
     pa_thread_mq_install(&u->thread_mq);
 
     pa_thread_mq_install(&u->thread_mq);
-    pa_rtpoll_install(u->rtpoll);
 
 
-    pa_rtclock_get(&u->thread_info.timestamp);
+    u->thread_info.timestamp = pa_rtclock_now();
     u->thread_info.in_null_mode = FALSE;
 
     for (;;) {
         int ret;
 
     u->thread_info.in_null_mode = FALSE;
 
     for (;;) {
         int ret;
 
-        /* If no outputs are connected, render some data and drop it immediately. */
-        if (u->sink->thread_info.state == PA_SINK_RUNNING && !u->thread_info.active_outputs) {
-            struct timeval now;
-
-            pa_rtclock_get(&now);
+        if (PA_SINK_IS_OPENED(u->sink->thread_info.state))
+            if (u->sink->thread_info.rewind_requested)
+                pa_sink_process_rewind(u->sink, 0);
 
 
-            if (!u->thread_info.in_null_mode || pa_timeval_cmp(&u->thread_info.timestamp, &now) <= 0) {
-                pa_sink_skip(u->sink, u->block_size);
+        /* If no outputs are connected, render some data and drop it immediately. */
+        if (PA_SINK_IS_OPENED(u->sink->thread_info.state) && !u->thread_info.active_outputs) {
+            pa_usec_t now;
 
 
-                if (!u->thread_info.in_null_mode)
-                    u->thread_info.timestamp = now;
+            now = pa_rtclock_now();
 
 
-                pa_timeval_add(&u->thread_info.timestamp, pa_bytes_to_usec(u->block_size, &u->sink->sample_spec));
-            }
+            if (!u->thread_info.in_null_mode || u->thread_info.timestamp <= now)
+                process_render_null(u, now);
 
 
-            pa_rtpoll_set_timer_absolute(u->rtpoll, &u->thread_info.timestamp);
+            pa_rtpoll_set_timer_absolute(u->rtpoll, u->thread_info.timestamp);
             u->thread_info.in_null_mode = TRUE;
             u->thread_info.in_null_mode = TRUE;
-
         } else {
             pa_rtpoll_set_timer_disabled(u->rtpoll);
             u->thread_info.in_null_mode = FALSE;
         } else {
             pa_rtpoll_set_timer_disabled(u->rtpoll);
             u->thread_info.in_null_mode = FALSE;
@@ -294,7 +342,7 @@ static void render_memblock(struct userdata *u, struct output *o, size_t length)
     pa_assert(o);
 
     /* We are run by the sink thread, on behalf of an output (o). The
     pa_assert(o);
 
     /* We are run by the sink thread, on behalf of an output (o). The
-     * other output is waiting for us, hence it is safe to access its
+     * output is waiting for us, hence it is safe to access its
      * mainblockq and asyncmsgq directly. */
 
     /* If we are not running, we cannot produce any data */
      * mainblockq and asyncmsgq directly. */
 
     /* If we are not running, we cannot produce any data */
@@ -314,19 +362,18 @@ static void render_memblock(struct userdata *u, struct output *o, size_t length)
         /* Render data! */
         pa_sink_render(u->sink, length, &chunk);
 
         /* Render data! */
         pa_sink_render(u->sink, length, &chunk);
 
-        /* OK, let's send this data to the other threads */
-        for (j = u->thread_info.active_outputs; j; j = j->next)
+        u->thread_info.counter += chunk.length;
 
 
-            /* Send to other outputs, which are not the requesting
-             * one */
+        /* OK, let's send this data to the other threads */
+        PA_LLIST_FOREACH(j, u->thread_info.active_outputs) {
+            if (j == o)
+                continue;
 
 
-            if (j != o)
-                pa_asyncmsgq_post(j->inq, PA_MSGOBJECT(j->sink_input), SINK_INPUT_MESSAGE_POST, NULL, 0, &chunk, NULL);
+            pa_asyncmsgq_post(j->inq, PA_MSGOBJECT(j->sink_input), SINK_INPUT_MESSAGE_POST, NULL, 0, &chunk, NULL);
+        }
 
         /* And place it directly into the requesting output's queue */
 
         /* And place it directly into the requesting output's queue */
-        if (o)
-            pa_memblockq_push_align(o->memblockq, &chunk);
-
+        pa_memblockq_push_align(o->memblockq, &chunk);
         pa_memblock_unref(chunk.memblock);
     }
 }
         pa_memblock_unref(chunk.memblock);
     }
 }
@@ -349,46 +396,120 @@ static void request_memblock(struct output *o, size_t length) {
 
     /* OK, we need to prepare new data, but only if the sink is actually running */
     if (pa_atomic_load(&o->userdata->thread_info.running))
 
     /* OK, we need to prepare new data, but only if the sink is actually running */
     if (pa_atomic_load(&o->userdata->thread_info.running))
-        pa_asyncmsgq_send(o->outq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_NEED, o, length, NULL);
+        pa_asyncmsgq_send(o->outq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_NEED, o, (int64_t) length, NULL);
 }
 
 /* Called from I/O thread context */
 }
 
 /* Called from I/O thread context */
-static int sink_input_peek_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
+static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk) {
     struct output *o;
 
     pa_sink_input_assert_ref(i);
     pa_assert_se(o = i->userdata);
 
     /* If necessary, get some new data */
     struct output *o;
 
     pa_sink_input_assert_ref(i);
     pa_assert_se(o = i->userdata);
 
     /* If necessary, get some new data */
-    request_memblock(o, length);
+    request_memblock(o, nbytes);
+
+    /* pa_log("%s q size is %u + %u (%u/%u)", */
+    /*        i->sink->name, */
+    /*        pa_memblockq_get_nblocks(o->memblockq), */
+    /*        pa_memblockq_get_nblocks(i->thread_info.render_memblockq), */
+    /*        pa_memblockq_get_maxrewind(o->memblockq), */
+    /*        pa_memblockq_get_maxrewind(i->thread_info.render_memblockq)); */
+
+    if (pa_memblockq_peek(o->memblockq, chunk) < 0)
+        return -1;
+
+    pa_memblockq_drop(o->memblockq, chunk->length);
+
+    return 0;
+}
+
+/* Called from I/O thread context */
+static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes) {
+    struct output *o;
+
+    pa_sink_input_assert_ref(i);
+    pa_assert_se(o = i->userdata);
 
 
-    return pa_memblockq_peek(o->memblockq, chunk);
+    pa_memblockq_rewind(o->memblockq, nbytes);
 }
 
 /* Called from I/O thread context */
 }
 
 /* Called from I/O thread context */
-static void sink_input_drop_cb(pa_sink_input *i, size_t length) {
+static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
     struct output *o;
 
     pa_sink_input_assert_ref(i);
     struct output *o;
 
     pa_sink_input_assert_ref(i);
-    pa_assert(length > 0);
     pa_assert_se(o = i->userdata);
 
     pa_assert_se(o = i->userdata);
 
-    pa_memblockq_drop(o->memblockq, length);
+    pa_memblockq_set_maxrewind(o->memblockq, nbytes);
+}
+
+/* Called from I/O thread context */
+static void sink_input_update_max_request_cb(pa_sink_input *i, size_t nbytes) {
+    struct output *o;
+
+    pa_sink_input_assert_ref(i);
+    pa_assert_se(o = i->userdata);
+
+    if (pa_atomic_load(&o->max_request) == (int) nbytes)
+        return;
+
+    pa_atomic_store(&o->max_request, (int) nbytes);
+    pa_asyncmsgq_post(o->outq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_UPDATE_MAX_REQUEST, NULL, 0, NULL, NULL);
+}
+
+/* Called from thread context */
+static void sink_input_update_sink_requested_latency_cb(pa_sink_input *i) {
+    struct output *o;
+    pa_usec_t c;
+
+    pa_assert(i);
+
+    pa_sink_input_assert_ref(i);
+    pa_assert_se(o = i->userdata);
+
+    c = pa_sink_get_requested_latency_within_thread(i->sink);
+
+    if (c == (pa_usec_t) -1)
+        c = i->sink->thread_info.max_latency;
+
+    if (pa_atomic_load(&o->requested_latency) == (int) c)
+        return;
+
+    pa_atomic_store(&o->requested_latency, (int) c);
+    pa_asyncmsgq_post(o->outq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_UPDATE_REQUESTED_LATENCY, NULL, 0, NULL, NULL);
 }
 
 /* Called from I/O thread context */
 static void sink_input_attach_cb(pa_sink_input *i) {
     struct output *o;
 }
 
 /* Called from I/O thread context */
 static void sink_input_attach_cb(pa_sink_input *i) {
     struct output *o;
+    pa_usec_t c;
 
     pa_sink_input_assert_ref(i);
     pa_assert_se(o = i->userdata);
 
     /* Set up the queue from the sink thread to us */
 
     pa_sink_input_assert_ref(i);
     pa_assert_se(o = i->userdata);
 
     /* Set up the queue from the sink thread to us */
-    pa_assert(!o->inq_rtpoll_item);
-    o->inq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
-            i->sink->rtpoll,
+    pa_assert(!o->inq_rtpoll_item_read && !o->outq_rtpoll_item_write);
+
+    o->inq_rtpoll_item_read = pa_rtpoll_item_new_asyncmsgq_read(
+            i->sink->thread_info.rtpoll,
             PA_RTPOLL_LATE,  /* This one is not that important, since we check for data in _peek() anyway. */
             o->inq);
             PA_RTPOLL_LATE,  /* This one is not that important, since we check for data in _peek() anyway. */
             o->inq);
+
+    o->outq_rtpoll_item_write = pa_rtpoll_item_new_asyncmsgq_write(
+            i->sink->thread_info.rtpoll,
+            PA_RTPOLL_EARLY,
+            o->outq);
+
+    pa_sink_input_request_rewind(i, 0, FALSE, TRUE, TRUE);
+
+    pa_atomic_store(&o->max_request, (int) pa_sink_input_get_max_request(i));
+
+    c = pa_sink_get_requested_latency_within_thread(i->sink);
+    pa_atomic_store(&o->requested_latency, (int) (c == (pa_usec_t) -1 ? 0 : c));
+
+    pa_asyncmsgq_post(o->outq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_UPDATE_MAX_REQUEST, NULL, 0, NULL, NULL);
+    pa_asyncmsgq_post(o->outq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_UPDATE_REQUESTED_LATENCY, NULL, 0, NULL, NULL);
 }
 
 /* Called from I/O thread context */
 }
 
 /* Called from I/O thread context */
@@ -398,10 +519,15 @@ static void sink_input_detach_cb(pa_sink_input *i) {
     pa_sink_input_assert_ref(i);
     pa_assert_se(o = i->userdata);
 
     pa_sink_input_assert_ref(i);
     pa_assert_se(o = i->userdata);
 
-    /* Shut down the queue from the sink thread to us */
-    pa_assert(o->inq_rtpoll_item);
-    pa_rtpoll_item_free(o->inq_rtpoll_item);
-    o->inq_rtpoll_item = NULL;
+    if (o->inq_rtpoll_item_read) {
+        pa_rtpoll_item_free(o->inq_rtpoll_item_read);
+        o->inq_rtpoll_item_read = NULL;
+    }
+
+    if (o->outq_rtpoll_item_write) {
+        pa_rtpoll_item_free(o->outq_rtpoll_item_write);
+        o->outq_rtpoll_item_write = NULL;
+    }
 }
 
 /* Called from main context */
 }
 
 /* Called from main context */
@@ -409,9 +535,9 @@ static void sink_input_kill_cb(pa_sink_input *i) {
     struct output *o;
 
     pa_sink_input_assert_ref(i);
     struct output *o;
 
     pa_sink_input_assert_ref(i);
-    pa_assert(o = i->userdata);
+    pa_assert_se(o = i->userdata);
 
 
-    pa_module_unload_request(o->userdata->module);
+    pa_module_unload_request(o->userdata->module, TRUE);
     output_free(o);
 }
 
     output_free(o);
 }
 
@@ -433,49 +559,17 @@ static int sink_input_process_msg(pa_msgobject *obj, int code, void *data, int64
 
         case SINK_INPUT_MESSAGE_POST:
 
 
         case SINK_INPUT_MESSAGE_POST:
 
-            if (PA_SINK_OPENED(o->sink_input->sink->thread_info.state))
+            if (PA_SINK_IS_OPENED(o->sink_input->sink->thread_info.state))
                 pa_memblockq_push_align(o->memblockq, chunk);
             else
                 pa_memblockq_push_align(o->memblockq, chunk);
             else
-                pa_memblockq_flush(o->memblockq);
+                pa_memblockq_flush_write(o->memblockq);
 
 
-            break;
+            return 0;
     }
 
     return pa_sink_input_process_msg(obj, code, data, offset, chunk);
 }
 
     }
 
     return pa_sink_input_process_msg(obj, code, data, offset, chunk);
 }
 
-/* Called from main context */
-static void disable_output(struct output *o) {
-    pa_assert(o);
-
-    if (!o->sink_input)
-        return;
-
-    pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
-    pa_sink_input_unlink(o->sink_input);
-    pa_sink_input_unref(o->sink_input);
-    o->sink_input = NULL;
-
-}
-
-/* Called from main context */
-static void enable_output(struct output *o) {
-    pa_assert(o);
-
-    if (o->sink_input)
-        return;
-
-    if (output_create_sink_input(o) >= 0) {
-
-        pa_memblockq_flush(o->memblockq);
-
-        pa_sink_input_put(o->sink_input);
-
-        if (o->userdata->sink && PA_SINK_LINKED(pa_sink_get_state(o->userdata->sink)))
-            pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_ADD_OUTPUT, o, 0, NULL);
-    }
-}
-
 /* Called from main context */
 static void suspend(struct userdata *u) {
     struct output *o;
 /* Called from main context */
 static void suspend(struct userdata *u) {
     struct output *o;
@@ -484,10 +578,8 @@ static void suspend(struct userdata *u) {
     pa_assert(u);
 
     /* Let's suspend by unlinking all streams */
     pa_assert(u);
 
     /* Let's suspend by unlinking all streams */
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
-        disable_output(o);
-
-    pick_master(u, NULL);
+    PA_IDXSET_FOREACH(o, u->outputs, idx)
+        output_disable(o);
 
     pa_log_info("Device suspended...");
 }
 
     pa_log_info("Device suspended...");
 }
@@ -500,15 +592,8 @@ static void unsuspend(struct userdata *u) {
     pa_assert(u);
 
     /* Let's resume */
     pa_assert(u);
 
     /* Let's resume */
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
-
-        pa_sink_suspend(o->sink, FALSE);
-
-        if (PA_SINK_OPENED(pa_sink_get_state(o->sink)))
-            enable_output(o);
-    }
-
-    pick_master(u, NULL);
+    PA_IDXSET_FOREACH(o, u->outputs, idx)
+        output_enable(o);
 
     pa_log_info("Resumed successfully...");
 }
 
     pa_log_info("Resumed successfully...");
 }
@@ -525,7 +610,7 @@ static int sink_set_state(pa_sink *sink, pa_sink_state_t state) {
 
     switch (state) {
         case PA_SINK_SUSPENDED:
 
     switch (state) {
         case PA_SINK_SUSPENDED:
-            pa_assert(PA_SINK_OPENED(pa_sink_get_state(u->sink)));
+            pa_assert(PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)));
 
             suspend(u);
             break;
 
             suspend(u);
             break;
@@ -540,119 +625,184 @@ static int sink_set_state(pa_sink *sink, pa_sink_state_t state) {
 
         case PA_SINK_UNLINKED:
         case PA_SINK_INIT:
 
         case PA_SINK_UNLINKED:
         case PA_SINK_INIT:
+        case PA_SINK_INVALID_STATE:
             ;
     }
 
     return 0;
 }
 
             ;
     }
 
     return 0;
 }
 
-/* Called from thread context of the master */
-static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
-    struct userdata *u = PA_SINK(o)->userdata;
+/* Called from IO context */
+static void update_max_request(struct userdata *u) {
+    size_t max_request = 0;
+    struct output *o;
 
 
-    switch (code) {
+    pa_assert(u);
+    pa_sink_assert_io_context(u->sink);
 
 
-        case PA_SINK_MESSAGE_SET_STATE:
-            pa_atomic_store(&u->thread_info.running, PA_PTR_TO_UINT(data) == PA_SINK_RUNNING);
-            break;
+    /* Collects the max_request values of all streams and sets the
+     * largest one locally */
 
 
-        case PA_SINK_MESSAGE_GET_LATENCY:
+    PA_LLIST_FOREACH(o, u->thread_info.active_outputs) {
+        size_t mr = (size_t) pa_atomic_load(&o->max_request);
 
 
-            /* This code will only be called when running in NULL
-             * mode, i.e. when no output is attached. See
-             * sink_get_latency_cb() below */
+        if (mr > max_request)
+            max_request = mr;
+    }
 
 
-            if (u->thread_info.in_null_mode) {
-                struct timeval now;
+    if (max_request <= 0)
+        max_request = pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec);
 
 
-                if (pa_timeval_cmp(&u->thread_info.timestamp, pa_rtclock_get(&now)) > 0) {
-                    *((pa_usec_t*) data) = pa_timeval_diff(&u->thread_info.timestamp, &now);
-                    break;
-                }
-            }
+    pa_sink_set_max_request_within_thread(u->sink, max_request);
+}
 
 
-            *((pa_usec_t*) data) = 0;
+/* Called from IO context */
+static void update_fixed_latency(struct userdata *u) {
+    pa_usec_t fixed_latency = 0;
+    struct output *o;
 
 
-            break;
+    pa_assert(u);
+    pa_sink_assert_io_context(u->sink);
 
 
-        case SINK_MESSAGE_ADD_OUTPUT: {
-            struct output *op = data;
+    /* Collects the requested_latency values of all streams and sets
+     * the largest one as fixed_latency locally */
 
 
-            PA_LLIST_PREPEND(struct output, u->thread_info.active_outputs, op);
+    PA_LLIST_FOREACH(o, u->thread_info.active_outputs) {
+        pa_usec_t rl = (size_t) pa_atomic_load(&o->requested_latency);
 
 
-            pa_assert(!op->outq_rtpoll_item);
+        if (rl > fixed_latency)
+            fixed_latency = rl;
+    }
 
 
-            /* Create pa_asyncmsgq to the sink thread */
+    if (fixed_latency <= 0)
+        fixed_latency = u->block_usec;
 
 
-            op->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
-                    u->rtpoll,
-                    PA_RTPOLL_EARLY-1,  /* This item is very important */
-                    op->outq);
+    pa_sink_set_fixed_latency_within_thread(u->sink, fixed_latency);
+}
 
 
-            return 0;
-        }
+/* Called from thread context of the io thread */
+static void output_add_within_thread(struct output *o) {
+    pa_assert(o);
+    pa_sink_assert_io_context(o->sink);
 
 
-        case SINK_MESSAGE_REMOVE_OUTPUT: {
-            struct output *op = data;
+    PA_LLIST_PREPEND(struct output, o->userdata->thread_info.active_outputs, o);
 
 
-            PA_LLIST_REMOVE(struct output, u->thread_info.active_outputs, op);
+    pa_assert(!o->outq_rtpoll_item_read && !o->inq_rtpoll_item_write);
 
 
-            /* Remove the q that leads from this output to the sink thread */
+    o->outq_rtpoll_item_read = pa_rtpoll_item_new_asyncmsgq_read(
+            o->userdata->rtpoll,
+            PA_RTPOLL_EARLY-1,  /* This item is very important */
+            o->outq);
+    o->inq_rtpoll_item_write = pa_rtpoll_item_new_asyncmsgq_write(
+            o->userdata->rtpoll,
+            PA_RTPOLL_EARLY,
+            o->inq);
+}
 
 
-            pa_assert(op->outq_rtpoll_item);
-            pa_rtpoll_item_free(op->outq_rtpoll_item);
-            op->outq_rtpoll_item = NULL;
+/* Called from thread context of the io thread */
+static void output_remove_within_thread(struct output *o) {
+    pa_assert(o);
+    pa_sink_assert_io_context(o->sink);
 
 
-            return 0;
-        }
+    PA_LLIST_REMOVE(struct output, o->userdata->thread_info.active_outputs, o);
 
 
-        case SINK_MESSAGE_NEED:
-            render_memblock(u, data, (size_t) offset);
-            return 0;
+    if (o->outq_rtpoll_item_read) {
+        pa_rtpoll_item_free(o->outq_rtpoll_item_read);
+        o->outq_rtpoll_item_read = NULL;
     }
 
     }
 
-    return pa_sink_process_msg(o, code, data, offset, chunk);
+    if (o->inq_rtpoll_item_write) {
+        pa_rtpoll_item_free(o->inq_rtpoll_item_write);
+        o->inq_rtpoll_item_write = NULL;
+    }
 }
 
 }
 
-/* Called from main context */
-static pa_usec_t sink_get_latency_cb(pa_sink *s) {
-    struct userdata *u;
+/* Called from thread context of the io thread */
+static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
+    struct userdata *u = PA_SINK(o)->userdata;
 
 
-    pa_sink_assert_ref(s);
-    pa_assert_se(u = s->userdata);
+    switch (code) {
+
+        case PA_SINK_MESSAGE_SET_STATE:
+            pa_atomic_store(&u->thread_info.running, PA_PTR_TO_UINT(data) == PA_SINK_RUNNING);
+
+            if (PA_PTR_TO_UINT(data) == PA_SINK_SUSPENDED)
+                pa_smoother_pause(u->thread_info.smoother, pa_rtclock_now());
+            else
+                pa_smoother_resume(u->thread_info.smoother, pa_rtclock_now(), TRUE);
+
+            break;
 
 
-    if (u->master) {
-        /* If we have a master sink, we just return the latency of it
-         * and add our own buffering on top */
+        case PA_SINK_MESSAGE_GET_LATENCY: {
+            pa_usec_t x, y, c, *delay = data;
 
 
-        if (!u->master->sink_input)
+            x = pa_rtclock_now();
+            y = pa_smoother_get(u->thread_info.smoother, x);
+
+            c = pa_bytes_to_usec(u->thread_info.counter, &u->sink->sample_spec);
+
+            if (y < c)
+                *delay = c - y;
+            else
+                *delay = 0;
+
+            return 0;
+        }
+
+        case SINK_MESSAGE_ADD_OUTPUT:
+            output_add_within_thread(data);
+            update_max_request(u);
+            update_fixed_latency(u);
             return 0;
 
             return 0;
 
-        return
-            pa_sink_input_get_latency(u->master->sink_input) +
-            pa_sink_get_latency(u->master->sink);
+        case SINK_MESSAGE_REMOVE_OUTPUT:
+            output_remove_within_thread(data);
+            update_max_request(u);
+            update_fixed_latency(u);
+            return 0;
 
 
-    } else {
-        pa_usec_t usec = 0;
+        case SINK_MESSAGE_NEED:
+            render_memblock(u, (struct output*) data, (size_t) offset);
+            return 0;
+
+        case SINK_MESSAGE_UPDATE_LATENCY: {
+            pa_usec_t x, y, latency = (pa_usec_t) offset;
 
 
-        /* We have no master, hence let's ask our own thread which
-         * implements the NULL sink */
+            x = pa_rtclock_now();
+            y = pa_bytes_to_usec(u->thread_info.counter, &u->sink->sample_spec);
 
 
-        if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
+            if (y > latency)
+                y -= latency;
+            else
+                y = 0;
+
+            pa_smoother_put(u->thread_info.smoother, x, y);
             return 0;
             return 0;
+        }
 
 
-        return usec;
-    }
+        case SINK_MESSAGE_UPDATE_MAX_REQUEST:
+            update_max_request(u);
+            break;
+
+        case SINK_MESSAGE_UPDATE_REQUESTED_LATENCY:
+            update_fixed_latency(u);
+            break;
+}
+
+    return pa_sink_process_msg(o, code, data, offset, chunk);
 }
 
 static void update_description(struct userdata *u) {
 }
 
 static void update_description(struct userdata *u) {
-    int first = 1;
+    pa_bool_t first = TRUE;
     char *t;
     struct output *o;
     uint32_t idx;
 
     pa_assert(u);
 
     char *t;
     struct output *o;
     uint32_t idx;
 
     pa_assert(u);
 
+    if (!u->auto_desc)
+        return;
+
     if (pa_idxset_isempty(u->outputs)) {
         pa_sink_set_description(u->sink, "Simultaneous output");
         return;
     if (pa_idxset_isempty(u->outputs)) {
         pa_sink_set_description(u->sink, "Simultaneous output");
         return;
@@ -660,14 +810,14 @@ static void update_description(struct userdata *u) {
 
     t = pa_xstrdup("Simultaneous output to");
 
 
     t = pa_xstrdup("Simultaneous output to");
 
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) {
+    PA_IDXSET_FOREACH(o, u->outputs, idx) {
         char *e;
 
         if (first) {
         char *e;
 
         if (first) {
-            e = pa_sprintf_malloc("%s %s", t, o->sink->description);
-            first = 0;
+            e = pa_sprintf_malloc("%s %s", t, pa_strnull(pa_proplist_gets(o->sink->proplist, PA_PROP_DEVICE_DESCRIPTION)));
+            first = FALSE;
         } else
         } else
-            e = pa_sprintf_malloc("%s, %s", t, o->sink->description);
+            e = pa_sprintf_malloc("%s, %s", t, pa_strnull(pa_proplist_gets(o->sink->proplist, PA_PROP_DEVICE_DESCRIPTION)));
 
         pa_xfree(t);
         t = e;
 
         pa_xfree(t);
         t = e;
@@ -677,81 +827,48 @@ static void update_description(struct userdata *u) {
     pa_xfree(t);
 }
 
     pa_xfree(t);
 }
 
-static void update_master(struct userdata *u, struct output *o) {
-    pa_assert(u);
-
-    if (u->master == o)
-        return;
-
-    if ((u->master = o))
-        pa_log_info("Master sink is now '%s'", o->sink_input->sink->name);
-    else
-        pa_log_info("No master selected, lacking suitable outputs.");
-}
-
-static void pick_master(struct userdata *u, struct output *except) {
-    struct output *o;
-    uint32_t idx;
-    pa_assert(u);
-
-    if (u->master &&
-        u->master != except &&
-        u->master->sink_input &&
-        PA_SINK_OPENED(pa_sink_get_state(u->master->sink))) {
-        update_master(u, u->master);
-        return;
-    }
-
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
-        if (o != except &&
-            o->sink_input &&
-            PA_SINK_OPENED(pa_sink_get_state(o->sink))) {
-            update_master(u, o);
-            return;
-        }
-
-    update_master(u, NULL);
-}
-
 static int output_create_sink_input(struct output *o) {
     pa_sink_input_new_data data;
 static int output_create_sink_input(struct output *o) {
     pa_sink_input_new_data data;
-    char *t;
 
     pa_assert(o);
 
     if (o->sink_input)
         return 0;
 
 
     pa_assert(o);
 
     if (o->sink_input)
         return 0;
 
-    t = pa_sprintf_malloc("Simultaneous output on %s", o->sink->description);
-
     pa_sink_input_new_data_init(&data);
     data.sink = o->sink;
     data.driver = __FILE__;
     pa_sink_input_new_data_init(&data);
     data.sink = o->sink;
     data.driver = __FILE__;
-    data.name = t;
+    pa_proplist_setf(data.proplist, PA_PROP_MEDIA_NAME, "Simultaneous output on %s", pa_strnull(pa_proplist_gets(o->sink->proplist, PA_PROP_DEVICE_DESCRIPTION)));
+    pa_proplist_sets(data.proplist, PA_PROP_MEDIA_ROLE, "filter");
     pa_sink_input_new_data_set_sample_spec(&data, &o->userdata->sink->sample_spec);
     pa_sink_input_new_data_set_channel_map(&data, &o->userdata->sink->channel_map);
     data.module = o->userdata->module;
     data.resample_method = o->userdata->resample_method;
 
     pa_sink_input_new_data_set_sample_spec(&data, &o->userdata->sink->sample_spec);
     pa_sink_input_new_data_set_channel_map(&data, &o->userdata->sink->channel_map);
     data.module = o->userdata->module;
     data.resample_method = o->userdata->resample_method;
 
-    o->sink_input = pa_sink_input_new(o->userdata->core, &data, PA_SINK_INPUT_VARIABLE_RATE|PA_SINK_INPUT_DONT_MOVE);
+    pa_sink_input_new(&o->sink_input, o->userdata->core, &data, PA_SINK_INPUT_VARIABLE_RATE|PA_SINK_INPUT_DONT_MOVE|PA_SINK_INPUT_NO_CREATE_ON_SUSPEND);
 
 
-    pa_xfree(t);
+    pa_sink_input_new_data_done(&data);
 
     if (!o->sink_input)
         return -1;
 
     o->sink_input->parent.process_msg = sink_input_process_msg;
 
     if (!o->sink_input)
         return -1;
 
     o->sink_input->parent.process_msg = sink_input_process_msg;
-    o->sink_input->peek = sink_input_peek_cb;
-    o->sink_input->drop = sink_input_drop_cb;
+    o->sink_input->pop = sink_input_pop_cb;
+    o->sink_input->process_rewind = sink_input_process_rewind_cb;
+    o->sink_input->update_max_rewind = sink_input_update_max_rewind_cb;
+    o->sink_input->update_max_request = sink_input_update_max_request_cb;
+    o->sink_input->update_sink_requested_latency = sink_input_update_sink_requested_latency_cb;
     o->sink_input->attach = sink_input_attach_cb;
     o->sink_input->detach = sink_input_detach_cb;
     o->sink_input->kill = sink_input_kill_cb;
     o->sink_input->userdata = o;
 
     o->sink_input->attach = sink_input_attach_cb;
     o->sink_input->detach = sink_input_detach_cb;
     o->sink_input->kill = sink_input_kill_cb;
     o->sink_input->userdata = o;
 
+    pa_sink_input_set_requested_latency(o->sink_input, BLOCK_USEC);
 
     return 0;
 }
 
 
     return 0;
 }
 
+/* Called from main context */
 static struct output *output_new(struct userdata *u, pa_sink *sink) {
     struct output *o;
 
 static struct output *output_new(struct userdata *u, pa_sink *sink) {
     struct output *o;
 
@@ -759,14 +876,11 @@ static struct output *output_new(struct userdata *u, pa_sink *sink) {
     pa_assert(sink);
     pa_assert(u->sink);
 
     pa_assert(sink);
     pa_assert(u->sink);
 
-    o = pa_xnew(struct output, 1);
+    o = pa_xnew0(struct output, 1);
     o->userdata = u;
     o->inq = pa_asyncmsgq_new(0);
     o->outq = pa_asyncmsgq_new(0);
     o->userdata = u;
     o->inq = pa_asyncmsgq_new(0);
     o->outq = pa_asyncmsgq_new(0);
-    o->inq_rtpoll_item = NULL;
-    o->outq_rtpoll_item = NULL;
     o->sink = sink;
     o->sink = sink;
-    o->sink_input = NULL;
     o->memblockq = pa_memblockq_new(
             0,
             MEMBLOCKQ_MAXLENGTH,
     o->memblockq = pa_memblockq_new(
             0,
             MEMBLOCKQ_MAXLENGTH,
@@ -774,61 +888,138 @@ static struct output *output_new(struct userdata *u, pa_sink *sink) {
             pa_frame_size(&u->sink->sample_spec),
             1,
             0,
             pa_frame_size(&u->sink->sample_spec),
             1,
             0,
+            0,
             NULL);
 
     pa_assert_se(pa_idxset_put(u->outputs, o, NULL) == 0);
             NULL);
 
     pa_assert_se(pa_idxset_put(u->outputs, o, NULL) == 0);
+    update_description(u);
 
 
-    if (u->sink && PA_SINK_LINKED(pa_sink_get_state(u->sink)))
-        pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_ADD_OUTPUT, o, 0, NULL);
-    else {
-        /* If the sink is not yet started, we need to do the activation ourselves */
-        PA_LLIST_PREPEND(struct output, u->thread_info.active_outputs, o);
+    return o;
+}
 
 
-        o->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
-                u->rtpoll,
-                PA_RTPOLL_EARLY-1,  /* This item is very important */
-                o->outq);
-    }
+/* Called from main context */
+static void output_free(struct output *o) {
+    pa_assert(o);
 
 
-    if (PA_SINK_OPENED(pa_sink_get_state(u->sink)) || pa_sink_get_state(u->sink) == PA_SINK_INIT) {
-        pa_sink_suspend(sink, FALSE);
+    output_disable(o);
 
 
-        if (PA_SINK_OPENED(pa_sink_get_state(sink)))
-            if (output_create_sink_input(o) < 0)
-                goto fail;
-    }
+    pa_assert_se(pa_idxset_remove_by_data(o->userdata->outputs, o, NULL));
+    update_description(o->userdata);
 
 
+    if (o->inq_rtpoll_item_read)
+        pa_rtpoll_item_free(o->inq_rtpoll_item_read);
+    if (o->inq_rtpoll_item_write)
+        pa_rtpoll_item_free(o->inq_rtpoll_item_write);
 
 
-    update_description(u);
+    if (o->outq_rtpoll_item_read)
+        pa_rtpoll_item_free(o->outq_rtpoll_item_read);
+    if (o->outq_rtpoll_item_write)
+        pa_rtpoll_item_free(o->outq_rtpoll_item_write);
 
 
-    return o;
+    if (o->inq)
+        pa_asyncmsgq_unref(o->inq);
 
 
-fail:
+    if (o->outq)
+        pa_asyncmsgq_unref(o->outq);
 
 
-    if (o) {
-        pa_idxset_remove_by_data(u->outputs, o, NULL);
+    if (o->memblockq)
+        pa_memblockq_free(o->memblockq);
 
 
-        if (o->sink_input) {
-            pa_sink_input_unlink(o->sink_input);
-            pa_sink_input_unref(o->sink_input);
-        }
+    pa_xfree(o);
+}
+
+/* Called from main context */
+static void output_enable(struct output *o) {
+    pa_assert(o);
+
+    if (o->sink_input)
+        return;
+
+    /* This might cause the sink to be resumed. The state change hook
+     * of the sink might hence be called from here, which might then
+     * cause us to be called in a loop. Make sure that state changes
+     * for this output don't cause this loop by setting a flag here */
+    o->ignore_state_change = TRUE;
+
+    if (output_create_sink_input(o) >= 0) {
 
 
-        if (o->memblockq)
-            pa_memblockq_free(o->memblockq);
+        if (pa_sink_get_state(o->sink) != PA_SINK_INIT) {
 
 
-        if (o->inq)
-            pa_asyncmsgq_unref(o->inq);
+            /* First we register the output. That means that the sink
+             * will start to pass data to this output. */
+            pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_ADD_OUTPUT, o, 0, NULL);
 
 
-        if (o->outq)
-            pa_asyncmsgq_unref(o->outq);
+            /* Then we enable the sink input. That means that the sink
+             * is now asked for new data. */
+            pa_sink_input_put(o->sink_input);
 
 
-        pa_xfree(o);
+        } else
+            /* Hmm the sink is not yet started, do things right here */
+            output_add_within_thread(o);
     }
 
     }
 
-    return NULL;
+    o->ignore_state_change = FALSE;
 }
 
 }
 
-static pa_hook_result_t sink_new_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
+/* Called from main context */
+static void output_disable(struct output *o) {
+    pa_assert(o);
+
+    if (!o->sink_input)
+        return;
+
+    /* First we disable the sink input. That means that the sink is
+     * not asked for new data anymore  */
+    pa_sink_input_unlink(o->sink_input);
+
+    /* Then we unregister the output. That means that the sink doesn't
+     * pass any further data to this output */
+    pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
+
+    /* Now dellocate the stream */
+    pa_sink_input_unref(o->sink_input);
+    o->sink_input = NULL;
+
+    /* Finally, drop all queued data */
+    pa_memblockq_flush_write(o->memblockq);
+    pa_asyncmsgq_flush(o->inq, FALSE);
+    pa_asyncmsgq_flush(o->outq, FALSE);
+}
+
+/* Called from main context */
+static void output_verify(struct output *o) {
+    pa_assert(o);
+
+    if (PA_SINK_IS_OPENED(pa_sink_get_state(o->userdata->sink)))
+        output_enable(o);
+    else
+        output_disable(o);
+}
+
+/* Called from main context */
+static pa_bool_t is_suitable_sink(struct userdata *u, pa_sink *s) {
+    const char *t;
+
+    pa_sink_assert_ref(s);
+
+    if (s == u->sink)
+        return FALSE;
+
+    if (!(s->flags & PA_SINK_HARDWARE))
+        return FALSE;
+
+    if (!(s->flags & PA_SINK_LATENCY))
+        return FALSE;
+
+    if ((t = pa_proplist_gets(s->proplist, PA_PROP_DEVICE_CLASS)))
+        if (!pa_streq(t, "sound"))
+            return FALSE;
+
+    return TRUE;
+}
+
+/* Called from main context */
+static pa_hook_result_t sink_put_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
     struct output *o;
 
     pa_core_assert_ref(c);
     struct output *o;
 
     pa_core_assert_ref(c);
@@ -836,75 +1027,69 @@ static pa_hook_result_t sink_new_hook_cb(pa_core *c, pa_sink *s, struct userdata
     pa_assert(u);
     pa_assert(u->automatic);
 
     pa_assert(u);
     pa_assert(u->automatic);
 
-    if (!(s->flags & PA_SINK_HARDWARE) || s == u->sink)
+    if (!is_suitable_sink(u, s))
         return PA_HOOK_OK;
 
     pa_log_info("Configuring new sink: %s", s->name);
         return PA_HOOK_OK;
 
     pa_log_info("Configuring new sink: %s", s->name);
-
     if (!(o = output_new(u, s))) {
         pa_log("Failed to create sink input on sink '%s'.", s->name);
         return PA_HOOK_OK;
     }
 
     if (!(o = output_new(u, s))) {
         pa_log("Failed to create sink input on sink '%s'.", s->name);
         return PA_HOOK_OK;
     }
 
-    if (o->sink_input)
-        pa_sink_input_put(o->sink_input);
-
-    pick_master(u, NULL);
+    output_verify(o);
 
     return PA_HOOK_OK;
 }
 
 
     return PA_HOOK_OK;
 }
 
-static pa_hook_result_t sink_unlink_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
+/* Called from main context */
+static struct output* find_output(struct userdata *u, pa_sink *s) {
     struct output *o;
     uint32_t idx;
 
     struct output *o;
     uint32_t idx;
 
-    pa_assert(c);
-    pa_sink_assert_ref(s);
     pa_assert(u);
     pa_assert(u);
+    pa_assert(s);
 
 
-    if (s == u->sink)
-        return PA_HOOK_OK;
+    if (u->sink == s)
+        return NULL;
 
 
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
+    PA_IDXSET_FOREACH(o, u->outputs, idx)
         if (o->sink == s)
         if (o->sink == s)
-            break;
+            return o;
+
+    return NULL;
+}
+
+/* Called from main context */
+static pa_hook_result_t sink_unlink_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
+    struct output *o;
+
+    pa_assert(c);
+    pa_sink_assert_ref(s);
+    pa_assert(u);
 
 
-    if (!o)
+    if (!(o = find_output(u, s)))
         return PA_HOOK_OK;
 
     pa_log_info("Unconfiguring sink: %s", s->name);
         return PA_HOOK_OK;
 
     pa_log_info("Unconfiguring sink: %s", s->name);
-
     output_free(o);
 
     return PA_HOOK_OK;
 }
 
     output_free(o);
 
     return PA_HOOK_OK;
 }
 
+/* Called from main context */
 static pa_hook_result_t sink_state_changed_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
     struct output *o;
 static pa_hook_result_t sink_state_changed_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
     struct output *o;
-    uint32_t idx;
-    pa_sink_state_t state;
 
 
-    if (s == u->sink)
+    if (!(o = find_output(u, s)))
         return PA_HOOK_OK;
 
         return PA_HOOK_OK;
 
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
-        if (o->sink == s)
-            break;
-
-    if (!o)
+    /* This state change might be triggered because we are creating a
+     * stream here, in that case we don't want to create it a second
+     * time here and enter a loop */
+    if (o->ignore_state_change)
         return PA_HOOK_OK;
 
         return PA_HOOK_OK;
 
-    state = pa_sink_get_state(s);
-
-    if (PA_SINK_OPENED(state) && PA_SINK_OPENED(pa_sink_get_state(u->sink)) && !o->sink_input) {
-        enable_output(o);
-        pick_master(u, NULL);
-    }
-
-    if (state == PA_SINK_SUSPENDED && o->sink_input) {
-        disable_output(o);
-        pick_master(u, o);
-    }
+    output_verify(o);
 
     return PA_HOOK_OK;
 }
 
     return PA_HOOK_OK;
 }
@@ -912,13 +1097,14 @@ static pa_hook_result_t sink_state_changed_hook_cb(pa_core *c, pa_sink *s, struc
 int pa__init(pa_module*m) {
     struct userdata *u;
     pa_modargs *ma = NULL;
 int pa__init(pa_module*m) {
     struct userdata *u;
     pa_modargs *ma = NULL;
-    const char *master_name, *slaves, *rm;
-    pa_sink *master_sink = NULL;
+    const char *slaves, *rm;
     int resample_method = PA_RESAMPLER_TRIVIAL;
     pa_sample_spec ss;
     pa_channel_map map;
     struct output *o;
     uint32_t idx;
     int resample_method = PA_RESAMPLER_TRIVIAL;
     pa_sample_spec ss;
     pa_channel_map map;
     struct output *o;
     uint32_t idx;
+    pa_sink_new_data data;
+    uint32_t adjust_time_sec;
 
     pa_assert(m);
 
 
     pa_assert(m);
 
@@ -934,109 +1120,149 @@ int pa__init(pa_module*m) {
         }
     }
 
         }
     }
 
-    u = pa_xnew(struct userdata, 1);
+    m->userdata = u = pa_xnew0(struct userdata, 1);
     u->core = m->core;
     u->module = m;
     u->core = m->core;
     u->module = m;
-    m->userdata = u;
-    u->sink = NULL;
-    u->master = NULL;
-    u->time_event = NULL;
-    u->adjust_time = DEFAULT_ADJUST_TIME;
-    pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
     u->rtpoll = pa_rtpoll_new();
     u->rtpoll = pa_rtpoll_new();
-    u->thread = NULL;
+    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
     u->resample_method = resample_method;
     u->outputs = pa_idxset_new(NULL, NULL);
     u->resample_method = resample_method;
     u->outputs = pa_idxset_new(NULL, NULL);
-    memset(&u->adjust_timestamp, 0, sizeof(u->adjust_timestamp));
-    u->sink_new_slot = u->sink_unlink_slot = u->sink_state_changed_slot = NULL;
-    PA_LLIST_HEAD_INIT(struct output, u->thread_info.active_outputs);
-    pa_atomic_store(&u->thread_info.running, FALSE);
-    u->thread_info.in_null_mode = FALSE;
-    pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
+    u->thread_info.smoother = pa_smoother_new(
+            PA_USEC_PER_SEC,
+            PA_USEC_PER_SEC*2,
+            TRUE,
+            TRUE,
+            10,
+            0,
+            FALSE);
 
 
-    if (pa_modargs_get_value_u32(ma, "adjust_time", &u->adjust_time) < 0) {
+    adjust_time_sec = DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC;
+    if (pa_modargs_get_value_u32(ma, "adjust_time", &adjust_time_sec) < 0) {
         pa_log("Failed to parse adjust_time value");
         goto fail;
     }
 
         pa_log("Failed to parse adjust_time value");
         goto fail;
     }
 
-    master_name = pa_modargs_get_value(ma, "master", NULL);
+    if (adjust_time_sec != DEFAULT_ADJUST_TIME_USEC / PA_USEC_PER_SEC)
+        u->adjust_time = adjust_time_sec * PA_USEC_PER_SEC;
+    else
+        u->adjust_time = DEFAULT_ADJUST_TIME_USEC;
+
     slaves = pa_modargs_get_value(ma, "slaves", NULL);
     slaves = pa_modargs_get_value(ma, "slaves", NULL);
-    if (!master_name != !slaves) {
-        pa_log("No master or slave sinks specified");
-        goto fail;
-    }
+    u->automatic = !slaves;
 
 
-    if (master_name) {
-        if (!(master_sink = pa_namereg_get(m->core, master_name, PA_NAMEREG_SINK, 1))) {
-            pa_log("Invalid master sink '%s'", master_name);
-            goto fail;
+    ss = m->core->default_sample_spec;
+    map = m->core->default_channel_map;
+
+    /* Check the specified slave sinks for sample_spec and channel_map to use for the combined sink */
+    if (!u->automatic) {
+        const char*split_state = NULL;
+        char *n = NULL;
+        pa_sample_spec slaves_spec;
+        pa_channel_map slaves_map;
+        pa_bool_t is_first_slave = TRUE;
+
+        pa_sample_spec_init(&slaves_spec);
+
+        while ((n = pa_split(slaves, ",", &split_state))) {
+            pa_sink *slave_sink;
+
+            if (!(slave_sink = pa_namereg_get(m->core, n, PA_NAMEREG_SINK))) {
+                pa_log("Invalid slave sink '%s'", n);
+                pa_xfree(n);
+                goto fail;
+            }
+
+            pa_xfree(n);
+
+            if (is_first_slave) {
+                slaves_spec = slave_sink->sample_spec;
+                slaves_map = slave_sink->channel_map;
+                is_first_slave = FALSE;
+            } else {
+                if (slaves_spec.format != slave_sink->sample_spec.format)
+                    slaves_spec.format = PA_SAMPLE_INVALID;
+
+                if (slaves_spec.rate < slave_sink->sample_spec.rate)
+                    slaves_spec.rate = slave_sink->sample_spec.rate;
+
+                if (!pa_channel_map_equal(&slaves_map, &slave_sink->channel_map))
+                    slaves_spec.channels = 0;
+            }
         }
 
         }
 
-        ss = master_sink->sample_spec;
-        u->automatic = FALSE;
-    } else {
-        master_sink = NULL;
-        ss = m->core->default_sample_spec;
-        u->automatic = TRUE;
+        if (!is_first_slave) {
+            if (slaves_spec.format != PA_SAMPLE_INVALID)
+                ss.format = slaves_spec.format;
+
+            ss.rate = slaves_spec.rate;
+
+            if (slaves_spec.channels > 0) {
+                map = slaves_map;
+                ss.channels = slaves_map.channels;
+            }
+        }
     }
 
     }
 
-    if ((pa_modargs_get_sample_spec(ma, &ss) < 0)) {
+    if ((pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0)) {
         pa_log("Invalid sample specification.");
         goto fail;
     }
 
         pa_log("Invalid sample specification.");
         goto fail;
     }
 
-    if (master_sink && ss.channels == master_sink->sample_spec.channels)
-        map = master_sink->channel_map;
-    else
-        pa_channel_map_init_auto(&map, ss.channels, PA_CHANNEL_MAP_DEFAULT);
-
-    if ((pa_modargs_get_channel_map(ma, NULL, &map) < 0)) {
-        pa_log("Invalid channel map.");
+    pa_sink_new_data_init(&data);
+    data.namereg_fail = FALSE;
+    data.driver = __FILE__;
+    data.module = m;
+    pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
+    pa_sink_new_data_set_sample_spec(&data, &ss);
+    pa_sink_new_data_set_channel_map(&data, &map);
+    pa_proplist_sets(data.proplist, PA_PROP_DEVICE_CLASS, "filter");
+
+    if (slaves)
+        pa_proplist_sets(data.proplist, "combine.slaves", slaves);
+
+    if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
+        pa_log("Invalid properties");
+        pa_sink_new_data_done(&data);
         goto fail;
     }
 
         goto fail;
     }
 
-    if (ss.channels != map.channels) {
-        pa_log("Channel map and sample specification don't match.");
-        goto fail;
+    /* Check proplist for a description & fill in a default value if not */
+    u->auto_desc = FALSE;
+    if (NULL == pa_proplist_gets(data.proplist, PA_PROP_DEVICE_DESCRIPTION)) {
+        u->auto_desc = TRUE;
+        pa_proplist_sets(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Simultaneous Output");
     }
 
     }
 
-    if (!(u->sink = pa_sink_new(m->core, __FILE__, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME), 0, &ss, &map))) {
+    u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY);
+    pa_sink_new_data_done(&data);
+
+    if (!u->sink) {
         pa_log("Failed to create sink");
         goto fail;
     }
 
     u->sink->parent.process_msg = sink_process_msg;
         pa_log("Failed to create sink");
         goto fail;
     }
 
     u->sink->parent.process_msg = sink_process_msg;
-    u->sink->get_latency = sink_get_latency_cb;
     u->sink->set_state = sink_set_state;
     u->sink->userdata = u;
 
     u->sink->set_state = sink_set_state;
     u->sink->userdata = u;
 
-    u->sink->flags = PA_SINK_LATENCY;
-    pa_sink_set_module(u->sink, m);
-    pa_sink_set_description(u->sink, "Simultaneous output");
     pa_sink_set_rtpoll(u->sink, u->rtpoll);
     pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
 
     pa_sink_set_rtpoll(u->sink, u->rtpoll);
     pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
 
-    u->block_size = pa_bytes_per_second(&ss) / 20; /* 50 ms */
-    if (u->block_size <= 0)
-        u->block_size = pa_frame_size(&ss);
+    u->block_usec = BLOCK_USEC;
+    pa_sink_set_max_request(u->sink, pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec));
 
     if (!u->automatic) {
         const char*split_state;
         char *n = NULL;
         pa_assert(slaves);
 
 
     if (!u->automatic) {
         const char*split_state;
         char *n = NULL;
         pa_assert(slaves);
 
-        /* The master and slaves have been specified manually */
-
-        if (!(u->master = output_new(u, master_sink))) {
-            pa_log("Failed to create master sink input on sink '%s'.", master_sink->name);
-            goto fail;
-        }
+        /* The slaves have been specified manually */
 
         split_state = NULL;
         while ((n = pa_split(slaves, ",", &split_state))) {
             pa_sink *slave_sink;
 
 
         split_state = NULL;
         while ((n = pa_split(slaves, ",", &split_state))) {
             pa_sink *slave_sink;
 
-            if (!(slave_sink = pa_namereg_get(m->core, n, PA_NAMEREG_SINK, 1)) || slave_sink == u->sink) {
+            if (!(slave_sink = pa_namereg_get(m->core, n, PA_NAMEREG_SINK)) || slave_sink == u->sink) {
                 pa_log("Invalid slave sink '%s'", n);
                 pa_xfree(n);
                 goto fail;
                 pa_log("Invalid slave sink '%s'", n);
                 pa_xfree(n);
                 goto fail;
@@ -1053,17 +1279,16 @@ int pa__init(pa_module*m) {
         if (pa_idxset_size(u->outputs) <= 1)
             pa_log_warn("No slave sinks specified.");
 
         if (pa_idxset_size(u->outputs) <= 1)
             pa_log_warn("No slave sinks specified.");
 
-        u->sink_new_slot = NULL;
+        u->sink_put_slot = NULL;
 
     } else {
         pa_sink *s;
 
 
     } else {
         pa_sink *s;
 
-        /* We're in automatic mode, we elect one hw sink to the master
-         * and attach all other hw sinks as slaves to it */
+        /* We're in automatic mode, we add every sink that matches our needs  */
 
 
-        for (s = pa_idxset_first(m->core->sinks, &idx); s; s = pa_idxset_next(m->core->sinks, &idx)) {
+        PA_IDXSET_FOREACH(s, m->core->sinks, idx) {
 
 
-            if (!(s->flags & PA_SINK_HARDWARE) || s == u->sink)
+            if (!is_suitable_sink(u, s))
                 continue;
 
             if (!output_new(u, s)) {
                 continue;
 
             if (!output_new(u, s)) {
@@ -1072,13 +1297,11 @@ int pa__init(pa_module*m) {
             }
         }
 
             }
         }
 
-        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);
+        u->sink_put_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_PUT], PA_HOOK_LATE, (pa_hook_cb_t) sink_put_hook_cb, u);
     }
 
     }
 
-    u->sink_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_UNLINK], (pa_hook_cb_t) sink_unlink_hook_cb, u);
-    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);
-
-    pick_master(u, NULL);
+    u->sink_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_UNLINK], PA_HOOK_EARLY, (pa_hook_cb_t) sink_unlink_hook_cb, u);
+    u->sink_state_changed_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_STATE_CHANGED], PA_HOOK_NORMAL, (pa_hook_cb_t) sink_state_changed_hook_cb, u);
 
     if (!(u->thread = pa_thread_new(thread_func, u))) {
         pa_log("Failed to create thread.");
 
     if (!(u->thread = pa_thread_new(thread_func, u))) {
         pa_log("Failed to create thread.");
@@ -1088,16 +1311,11 @@ int pa__init(pa_module*m) {
     /* Activate the sink and the sink inputs */
     pa_sink_put(u->sink);
 
     /* Activate the sink and the sink inputs */
     pa_sink_put(u->sink);
 
-    for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
-        if (o->sink_input)
-            pa_sink_input_put(o->sink_input);
+    PA_IDXSET_FOREACH(o, u->outputs, idx)
+        output_verify(o);
 
 
-    if (u->adjust_time > 0) {
-        struct timeval tv;
-        pa_gettimeofday(&tv);
-        tv.tv_sec += u->adjust_time;
-        u->time_event = m->core->mainloop->time_new(m->core->mainloop, &tv, time_callback, u);
-    }
+    if (u->adjust_time > 0)
+        u->time_event = pa_core_rttime_new(m->core, pa_rtclock_now() + u->adjust_time, time_callback, u);
 
     pa_modargs_free(ma);
 
 
     pa_modargs_free(ma);
 
@@ -1113,35 +1331,6 @@ fail:
     return -1;
 }
 
     return -1;
 }
 
-static void output_free(struct output *o) {
-    pa_assert(o);
-
-    pick_master(o->userdata, o);
-
-    disable_output(o);
-
-    pa_assert_se(pa_idxset_remove_by_data(o->userdata->outputs, o, NULL));
-
-    update_description(o->userdata);
-
-    if (o->inq_rtpoll_item)
-        pa_rtpoll_item_free(o->inq_rtpoll_item);
-
-    if (o->outq_rtpoll_item)
-        pa_rtpoll_item_free(o->outq_rtpoll_item);
-
-    if (o->inq)
-        pa_asyncmsgq_unref(o->inq);
-
-    if (o->outq)
-        pa_asyncmsgq_unref(o->outq);
-
-    if (o->memblockq)
-        pa_memblockq_free(o->memblockq);
-
-    pa_xfree(o);
-}
-
 void pa__done(pa_module*m) {
     struct userdata *u;
     struct output *o;
 void pa__done(pa_module*m) {
     struct userdata *u;
     struct output *o;
@@ -1151,8 +1340,8 @@ void pa__done(pa_module*m) {
     if (!(u = m->userdata))
         return;
 
     if (!(u = m->userdata))
         return;
 
-    if (u->sink_new_slot)
-        pa_hook_slot_free(u->sink_new_slot);
+    if (u->sink_put_slot)
+        pa_hook_slot_free(u->sink_put_slot);
 
     if (u->sink_unlink_slot)
         pa_hook_slot_free(u->sink_unlink_slot);
 
     if (u->sink_unlink_slot)
         pa_hook_slot_free(u->sink_unlink_slot);
@@ -1186,5 +1375,8 @@ void pa__done(pa_module*m) {
     if (u->time_event)
         u->core->mainloop->time_free(u->time_event);
 
     if (u->time_event)
         u->core->mainloop->time_free(u->time_event);
 
+    if (u->thread_info.smoother)
+        pa_smoother_free(u->thread_info.smoother);
+
     pa_xfree(u);
 }
     pa_xfree(u);
 }