]> code.delx.au - pulseaudio/blobdiff - src/pulsecore/protocol-native.c
sink: volume handling rework, new flat volume logic
[pulseaudio] / src / pulsecore / protocol-native.c
index 30d68f3531aa9b469695e05a999e1ed47d49bc37..b1285e1551f6458835f2611b978d88279bf20511 100644 (file)
@@ -29,6 +29,7 @@
 #include <stdlib.h>
 #include <unistd.h>
 
+#include <pulse/rtclock.h>
 #include <pulse/timeval.h>
 #include <pulse/version.h>
 #include <pulse/utf8.h>
@@ -61,7 +62,7 @@
 #include "protocol-native.h"
 
 /* Kick a client if it doesn't authenticate within this time */
-#define AUTH_TIMEOUT 60
+#define AUTH_TIMEOUT (60 * PA_USEC_PER_SEC)
 
 /* Don't accept more connection than this */
 #define MAX_CONNECTIONS 64
@@ -86,7 +87,15 @@ typedef struct record_stream {
     pa_bool_t early_requests:1;
 
     pa_buffer_attr buffer_attr;
-    pa_usec_t source_latency;
+
+    pa_atomic_t on_the_fly;
+    pa_usec_t configured_source_latency;
+    size_t drop_initial;
+
+    /* Only updated after SOURCE_OUTPUT_MESSAGE_UPDATE_LATENCY */
+    size_t on_the_fly_snapshot;
+    pa_usec_t current_monitor_latency;
+    pa_usec_t current_source_latency;
 } record_stream;
 
 PA_DECLARE_CLASS(record_stream);
@@ -119,12 +128,14 @@ typedef struct playback_stream {
     uint32_t syncid;
 
     pa_atomic_t missing;
-    pa_usec_t sink_latency;
+    pa_usec_t configured_sink_latency;
     pa_buffer_attr buffer_attr;
 
     /* Only updated after SINK_INPUT_MESSAGE_UPDATE_LATENCY */
     int64_t read_index, write_index;
     size_t render_memblockq_length;
+    pa_usec_t current_sink_latency;
+    uint64_t playing_for, underrun_for;
 } playback_stream;
 
 PA_DECLARE_CLASS(playback_stream);
@@ -181,6 +192,10 @@ struct pa_native_protocol {
     pa_hashmap *extensions;
 };
 
+enum {
+    SOURCE_OUTPUT_MESSAGE_UPDATE_LATENCY = PA_SOURCE_OUTPUT_MESSAGE_MAX
+};
+
 enum {
     SINK_INPUT_MESSAGE_POST_DATA = PA_SINK_INPUT_MESSAGE_MAX, /* data from main loop to sink input */
     SINK_INPUT_MESSAGE_DRAIN, /* disabled prebuf, get playback started. */
@@ -230,6 +245,7 @@ static pa_usec_t source_output_get_latency_cb(pa_source_output *o);
 static void source_output_send_event_cb(pa_source_output *o, const char *event, pa_proplist *pl);
 
 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
+static int source_output_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
 
 static void command_exit(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 static void command_create_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
@@ -269,6 +285,7 @@ static void command_update_proplist(pa_pdispatch *pd, uint32_t command, uint32_t
 static void command_remove_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 static void command_extension(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 static void command_set_card_profile(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
+static void command_set_sink_or_source_port(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 
 static const pa_pdispatch_cb_t command_table[PA_COMMAND_MAX] = {
     [PA_COMMAND_ERROR] = NULL,
@@ -365,6 +382,9 @@ static const pa_pdispatch_cb_t command_table[PA_COMMAND_MAX] = {
 
     [PA_COMMAND_SET_CARD_PROFILE] = command_set_card_profile,
 
+    [PA_COMMAND_SET_SINK_PORT] = command_set_sink_or_source_port,
+    [PA_COMMAND_SET_SOURCE_PORT] = command_set_sink_or_source_port,
+
     [PA_COMMAND_EXTENSION] = command_extension
 };
 
@@ -474,6 +494,10 @@ static int record_stream_process_msg(pa_msgobject *o, int code, void*userdata, i
 
         case RECORD_STREAM_MESSAGE_POST_DATA:
 
+            /* We try to keep up to date with how many bytes are
+             * currently on the fly */
+            pa_atomic_sub(&s->on_the_fly, chunk->length);
+
             if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
 /*                 pa_log_warn("Failed to push data into output queue."); */
                 return -1;
@@ -537,29 +561,29 @@ static void fix_record_buffer_attr_pre(record_stream *s) {
         /* Ok, the user didn't ask us to adjust the latency, hence we
          * don't */
 
-        source_usec = 0;
+        source_usec = (pa_usec_t) -1;
     }
 
-    if (source_usec > 0)
-        s->source_latency = pa_source_output_set_requested_latency(s->source_output, source_usec);
+    if (source_usec != (pa_usec_t) -1)
+        s->configured_source_latency = pa_source_output_set_requested_latency(s->source_output, source_usec);
     else
-        s->source_latency = 0;
+        s->configured_source_latency = 0;
 
     if (s->early_requests) {
 
         /* Ok, we didn't necessarily get what we were asking for, so
          * let's tell the user */
 
-        fragsize_usec = s->source_latency;
+        fragsize_usec = s->configured_source_latency;
 
     } else if (s->adjust_latency) {
 
         /* Now subtract what we actually got */
 
-        if (fragsize_usec >= s->source_latency*2)
-            fragsize_usec -= s->source_latency;
+        if (fragsize_usec >= s->configured_source_latency*2)
+            fragsize_usec -= s->configured_source_latency;
         else
-            fragsize_usec = s->source_latency;
+            fragsize_usec = s->configured_source_latency;
     }
 
     if (pa_usec_to_bytes(orig_fragsize_usec, &s->source_output->sample_spec) !=
@@ -645,7 +669,9 @@ static record_stream* record_stream_new(
     s->buffer_attr = *attr;
     s->adjust_latency = adjust_latency;
     s->early_requests = early_requests;
+    pa_atomic_store(&s->on_the_fly, 0);
 
+    s->source_output->parent.process_msg = source_output_process_msg;
     s->source_output->push = source_output_push_cb;
     s->source_output->kill = source_output_kill_cb;
     s->source_output->get_latency = source_output_get_latency_cb;
@@ -675,9 +701,9 @@ static record_stream* record_stream_new(
     pa_idxset_put(c->record_streams, s, &s->index);
 
     pa_log_info("Final latency %0.2f ms = %0.2f ms + %0.2f ms",
-                ((double) pa_bytes_to_usec(s->buffer_attr.fragsize, &source_output->sample_spec) + (double) s->source_latency) / PA_USEC_PER_MSEC,
+                ((double) pa_bytes_to_usec(s->buffer_attr.fragsize, &source_output->sample_spec) + (double) s->configured_source_latency) / PA_USEC_PER_MSEC,
                 (double) pa_bytes_to_usec(s->buffer_attr.fragsize, &source_output->sample_spec) / PA_USEC_PER_MSEC,
-                (double) s->source_latency / PA_USEC_PER_MSEC);
+                (double) s->configured_source_latency / PA_USEC_PER_MSEC);
 
     pa_source_output_put(s->source_output);
     return s;
@@ -736,6 +762,7 @@ static int playback_stream_process_msg(pa_msgobject *o, int code, void*userdata,
         return -1;
 
     switch (code) {
+
         case PLAYBACK_STREAM_MESSAGE_REQUEST_DATA: {
             pa_tagstruct *t;
             int l = 0;
@@ -817,7 +844,7 @@ static int playback_stream_process_msg(pa_msgobject *o, int code, void*userdata,
             pa_tagstruct_putu32(t, s->buffer_attr.tlength);
             pa_tagstruct_putu32(t, s->buffer_attr.prebuf);
             pa_tagstruct_putu32(t, s->buffer_attr.minreq);
-            pa_tagstruct_put_usec(t, s->sink_latency);
+            pa_tagstruct_put_usec(t, s->configured_sink_latency);
             pa_pstream_send_tagstruct(s->connection->pstream, t);
 
             break;
@@ -829,7 +856,7 @@ static int playback_stream_process_msg(pa_msgobject *o, int code, void*userdata,
 
 /* Called from main context */
 static void fix_playback_buffer_attr(playback_stream *s) {
-    size_t frame_size;
+    size_t frame_size, max_prebuf;
     pa_usec_t orig_tlength_usec, tlength_usec, orig_minreq_usec, minreq_usec, sink_usec;
 
     pa_assert(s);
@@ -915,14 +942,14 @@ static void fix_playback_buffer_attr(playback_stream *s) {
         pa_log_debug("Traditional mode enabled, modifying sink usec only for compat with minreq.");
     }
 
-    s->sink_latency = pa_sink_input_set_requested_latency(s->sink_input, sink_usec);
+    s->configured_sink_latency = pa_sink_input_set_requested_latency(s->sink_input, sink_usec);
 
     if (s->early_requests) {
 
         /* Ok, we didn't necessarily get what we were asking for, so
          * let's tell the user */
 
-        minreq_usec = s->sink_latency;
+        minreq_usec = s->configured_sink_latency;
 
     } else if (s->adjust_latency) {
 
@@ -930,14 +957,14 @@ static void fix_playback_buffer_attr(playback_stream *s) {
          * let's subtract from what we asked for for the remaining
          * buffer space */
 
-        if (tlength_usec >= s->sink_latency)
-            tlength_usec -= s->sink_latency;
+        if (tlength_usec >= s->configured_sink_latency)
+            tlength_usec -= s->configured_sink_latency;
     }
 
     /* FIXME: This is actually larger than necessary, since not all of
      * the sink latency is actually rewritable. */
-    if (tlength_usec < s->sink_latency + 2*minreq_usec)
-        tlength_usec = s->sink_latency + 2*minreq_usec;
+    if (tlength_usec < s->configured_sink_latency + 2*minreq_usec)
+        tlength_usec = s->configured_sink_latency + 2*minreq_usec;
 
     if (pa_usec_to_bytes_round_up(orig_tlength_usec, &s->sink_input->sample_spec) !=
         pa_usec_to_bytes_round_up(tlength_usec, &s->sink_input->sample_spec))
@@ -955,8 +982,11 @@ static void fix_playback_buffer_attr(playback_stream *s) {
     if (s->buffer_attr.tlength <= s->buffer_attr.minreq)
         s->buffer_attr.tlength = s->buffer_attr.minreq*2 + (uint32_t) frame_size;
 
-    if (s->buffer_attr.prebuf == (uint32_t) -1 || s->buffer_attr.prebuf > s->buffer_attr.tlength)
-        s->buffer_attr.prebuf = s->buffer_attr.tlength;
+    max_prebuf = s->buffer_attr.tlength + (uint32_t)frame_size - s->buffer_attr.minreq;
+
+    if (s->buffer_attr.prebuf == (uint32_t) -1 ||
+        s->buffer_attr.prebuf > max_prebuf)
+        s->buffer_attr.prebuf = max_prebuf;
 }
 
 /* Called from main context */
@@ -1083,10 +1113,10 @@ static playback_stream* playback_stream_new(
     pa_idxset_put(c->output_streams, s, &s->index);
 
     pa_log_info("Final latency %0.2f ms = %0.2f ms + 2*%0.2f ms + %0.2f ms",
-                ((double) pa_bytes_to_usec(s->buffer_attr.tlength, &sink_input->sample_spec) + (double) s->sink_latency) / PA_USEC_PER_MSEC,
+                ((double) pa_bytes_to_usec(s->buffer_attr.tlength, &sink_input->sample_spec) + (double) s->configured_sink_latency) / PA_USEC_PER_MSEC,
                 (double) pa_bytes_to_usec(s->buffer_attr.tlength-s->buffer_attr.minreq*2, &sink_input->sample_spec) / PA_USEC_PER_MSEC,
                 (double) pa_bytes_to_usec(s->buffer_attr.minreq, &sink_input->sample_spec) / PA_USEC_PER_MSEC,
-                (double) s->sink_latency / PA_USEC_PER_MSEC);
+                (double) s->configured_sink_latency / PA_USEC_PER_MSEC);
 
     pa_sink_input_put(s->sink_input);
     return s;
@@ -1101,6 +1131,12 @@ static void playback_stream_request_bytes(playback_stream *s) {
 
     m = pa_memblockq_pop_missing(s->memblockq);
 
+    /* pa_log("request_bytes(%lu) (tlength=%lu minreq=%lu length=%lu)", */
+    /*        (unsigned long) m, */
+    /*        pa_memblockq_get_tlength(s->memblockq), */
+    /*        pa_memblockq_get_minreq(s->memblockq), */
+    /*        pa_memblockq_get_length(s->memblockq)); */
+
     if (m <= 0)
         return;
 
@@ -1114,7 +1150,6 @@ static void playback_stream_request_bytes(playback_stream *s) {
         pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_REQUEST_DATA, NULL, 0, NULL, NULL);
 }
 
-
 /* Called from main context */
 static void playback_stream_send_killed(playback_stream *p) {
     pa_tagstruct *t;
@@ -1316,7 +1351,9 @@ static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int
 /*             pa_log("sink input post: %lu %lli", (unsigned long) chunk->length, (long long) windex); */
 
             if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
-                pa_log_warn("Failed to push data into queue");
+
+                if (pa_log_ratelimit())
+                    pa_log_warn("Failed to push data into queue");
                 pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_OVERFLOW, NULL, 0, NULL, NULL);
                 pa_memblockq_seek(s->memblockq, (int64_t) chunk->length, PA_SEEK_RELATIVE, TRUE);
             }
@@ -1387,10 +1424,14 @@ static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int
         }
 
         case SINK_INPUT_MESSAGE_UPDATE_LATENCY:
-
+            /* Atomically get a snapshot of all timing parameters... */
             s->read_index = pa_memblockq_get_read_index(s->memblockq);
             s->write_index = pa_memblockq_get_write_index(s->memblockq);
             s->render_memblockq_length = pa_memblockq_get_length(s->sink_input->thread_info.render_memblockq);
+            s->current_sink_latency = pa_sink_get_latency_within_thread(s->sink_input->sink);
+            s->underrun_for = s->sink_input->thread_info.underrun_for;
+            s->playing_for = s->sink_input->thread_info.playing_for;
+
             return 0;
 
         case PA_SINK_INPUT_MESSAGE_SET_STATE: {
@@ -1440,7 +1481,8 @@ static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk
     if (pa_memblockq_is_readable(s->memblockq))
         s->is_underrun = FALSE;
     else {
-        pa_log_debug("Underrun on '%s', %lu bytes in queue.", pa_strnull(pa_proplist_gets(i->proplist, PA_PROP_MEDIA_NAME)), (unsigned long) pa_memblockq_get_length(s->memblockq));
+        if (!s->is_underrun)
+            pa_log_debug("Underrun on '%s', %lu bytes in queue.", pa_strnull(pa_proplist_gets(i->proplist, PA_PROP_MEDIA_NAME)), (unsigned long) pa_memblockq_get_length(s->memblockq));
 
         if (s->drain_request && pa_sink_input_safe_to_remove(i)) {
             s->drain_request = FALSE;
@@ -1583,6 +1625,9 @@ static void sink_input_moving_cb(pa_sink_input *i, pa_sink *dest) {
     s = PLAYBACK_STREAM(i->userdata);
     playback_stream_assert_ref(s);
 
+    if (!dest)
+        return;
+
     fix_playback_buffer_attr(s);
     pa_memblockq_apply_attr(s->memblockq, &s->buffer_attr);
     pa_memblockq_get_attr(s->memblockq, &s->buffer_attr);
@@ -1603,7 +1648,7 @@ static void sink_input_moving_cb(pa_sink_input *i, pa_sink *dest) {
         pa_tagstruct_putu32(t, s->buffer_attr.tlength);
         pa_tagstruct_putu32(t, s->buffer_attr.prebuf);
         pa_tagstruct_putu32(t, s->buffer_attr.minreq);
-        pa_tagstruct_put_usec(t, s->sink_latency);
+        pa_tagstruct_put_usec(t, s->configured_sink_latency);
     }
 
     pa_pstream_send_tagstruct(s->connection->pstream, t);
@@ -1611,6 +1656,27 @@ static void sink_input_moving_cb(pa_sink_input *i, pa_sink *dest) {
 
 /*** source_output callbacks ***/
 
+/* Called from thread context */
+static int source_output_process_msg(pa_msgobject *_o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
+    pa_source_output *o = PA_SOURCE_OUTPUT(_o);
+    record_stream *s;
+
+    pa_source_output_assert_ref(o);
+    s = RECORD_STREAM(o->userdata);
+    record_stream_assert_ref(s);
+
+    switch (code) {
+        case SOURCE_OUTPUT_MESSAGE_UPDATE_LATENCY:
+            /* Atomically get a snapshot of all timing parameters... */
+            s->current_monitor_latency = o->source->monitor_of ? pa_sink_get_latency_within_thread(o->source->monitor_of) : 0;
+            s->current_source_latency = pa_source_get_latency_within_thread(o->source);
+            s->on_the_fly_snapshot = pa_atomic_load(&s->on_the_fly);
+            return 0;
+    }
+
+    return pa_source_output_process_msg(_o, code, userdata, offset, chunk);
+}
+
 /* Called from thread context */
 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
     record_stream *s;
@@ -1620,6 +1686,7 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk)
     record_stream_assert_ref(s);
     pa_assert(chunk);
 
+    pa_atomic_add(&s->on_the_fly, chunk->length);
     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), RECORD_STREAM_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
 }
 
@@ -1696,6 +1763,9 @@ static void source_output_moving_cb(pa_source_output *o, pa_source *dest) {
     s = RECORD_STREAM(o->userdata);
     record_stream_assert_ref(s);
 
+    if (!dest)
+        return;
+
     fix_record_buffer_attr_pre(s);
     pa_memblockq_set_maxlength(s->memblockq, s->buffer_attr.maxlength);
     pa_memblockq_get_attr(s->memblockq, &s->buffer_attr);
@@ -1715,7 +1785,7 @@ static void source_output_moving_cb(pa_source_output *o, pa_source *dest) {
     if (s->connection->version >= 13) {
         pa_tagstruct_putu32(t, s->buffer_attr.maxlength);
         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
-        pa_tagstruct_put_usec(t, s->source_latency);
+        pa_tagstruct_put_usec(t, s->configured_source_latency);
     }
 
     pa_pstream_send_tagstruct(s->connection->pstream, t);
@@ -1895,7 +1965,7 @@ static void command_create_playback_stream(pa_pdispatch *pd, uint32_t command, u
         (no_move ?  PA_SINK_INPUT_DONT_MOVE : 0) |
         (variable_rate ?  PA_SINK_INPUT_VARIABLE_RATE : 0) |
         (dont_inhibit_auto_suspend ? PA_SINK_INPUT_DONT_INHIBIT_AUTO_SUSPEND : 0) |
-        (fail_on_suspend ? PA_SINK_INPUT_FAIL_ON_SUSPEND : 0);
+        (fail_on_suspend ? PA_SINK_INPUT_NO_CREATE_ON_SUSPEND|PA_SINK_INPUT_KILL_ON_SUSPEND : 0);
 
     /* Only since protocol version 15 there's a seperate muted_set
      * flag. For older versions we synthesize it here */
@@ -1938,7 +2008,7 @@ static void command_create_playback_stream(pa_pdispatch *pd, uint32_t command, u
     }
 
     if (c->version >= 13)
-        pa_tagstruct_put_usec(reply, s->sink_latency);
+        pa_tagstruct_put_usec(reply, s->configured_sink_latency);
 
     pa_pstream_send_tagstruct(c->pstream, reply);
 }
@@ -2151,7 +2221,7 @@ static void command_create_record_stream(pa_pdispatch *pd, uint32_t command, uin
         (no_move ?  PA_SOURCE_OUTPUT_DONT_MOVE : 0) |
         (variable_rate ?  PA_SOURCE_OUTPUT_VARIABLE_RATE : 0) |
         (dont_inhibit_auto_suspend ? PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND : 0) |
-        (fail_on_suspend ? PA_SOURCE_OUTPUT_FAIL_ON_SUSPEND : 0);
+        (fail_on_suspend ? PA_SOURCE_OUTPUT_NO_CREATE_ON_SUSPEND|PA_SOURCE_OUTPUT_KILL_ON_SUSPEND : 0);
 
     s = record_stream_new(c, source, &ss, &map, peak_detect, &attr, flags, p, adjust_latency, direct_on_input, early_requests, &ret);
     pa_proplist_free(p);
@@ -2185,7 +2255,7 @@ static void command_create_record_stream(pa_pdispatch *pd, uint32_t command, uin
     }
 
     if (c->version >= 13)
-        pa_tagstruct_put_usec(reply, s->source_latency);
+        pa_tagstruct_put_usec(reply, s->configured_source_latency);
 
     pa_pstream_send_tagstruct(c->pstream, reply);
 }
@@ -2472,7 +2542,6 @@ static void command_get_playback_latency(pa_pdispatch *pd, uint32_t command, uin
     playback_stream *s;
     struct timeval tv, now;
     uint32_t idx;
-    pa_usec_t latency;
 
     pa_native_connection_assert_ref(c);
     pa_assert(t);
@@ -2488,25 +2557,27 @@ static void command_get_playback_latency(pa_pdispatch *pd, uint32_t command, uin
     s = pa_idxset_get_by_index(c->output_streams, idx);
     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
-    CHECK_VALIDITY(c->pstream, pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_UPDATE_LATENCY, s, 0, NULL) == 0, tag, PA_ERR_NOENTITY)
-
-    reply = reply_new(tag);
-
-    latency = pa_sink_get_latency(s->sink_input->sink);
-    latency += pa_bytes_to_usec(s->render_memblockq_length, &s->sink_input->sample_spec);
 
-    pa_tagstruct_put_usec(reply, latency);
+    /* Get an atomic snapshot of all timing parameters */
+    pa_assert_se(pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_UPDATE_LATENCY, s, 0, NULL) == 0);
 
+    reply = reply_new(tag);
+    pa_tagstruct_put_usec(reply,
+                          s->current_sink_latency +
+                          pa_bytes_to_usec(s->render_memblockq_length, &s->sink_input->sink->sample_spec));
     pa_tagstruct_put_usec(reply, 0);
-    pa_tagstruct_put_boolean(reply, s->sink_input->thread_info.playing_for > 0);
+    pa_tagstruct_put_boolean(reply,
+                             s->playing_for > 0 &&
+                             pa_sink_get_state(s->sink_input->sink) == PA_SINK_RUNNING &&
+                             pa_sink_input_get_state(s->sink_input) == PA_SINK_INPUT_RUNNING);
     pa_tagstruct_put_timeval(reply, &tv);
     pa_tagstruct_put_timeval(reply, pa_gettimeofday(&now));
     pa_tagstruct_puts64(reply, s->write_index);
     pa_tagstruct_puts64(reply, s->read_index);
 
     if (c->version >= 13) {
-        pa_tagstruct_putu64(reply, s->sink_input->thread_info.underrun_for);
-        pa_tagstruct_putu64(reply, s->sink_input->thread_info.playing_for);
+        pa_tagstruct_putu64(reply, s->underrun_for);
+        pa_tagstruct_putu64(reply, s->playing_for);
     }
 
     pa_pstream_send_tagstruct(c->pstream, reply);
@@ -2533,10 +2604,17 @@ static void command_get_record_latency(pa_pdispatch *pd, uint32_t command, uint3
     s = pa_idxset_get_by_index(c->record_streams, idx);
     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
 
+    /* Get an atomic snapshot of all timing parameters */
+    pa_assert_se(pa_asyncmsgq_send(s->source_output->source->asyncmsgq, PA_MSGOBJECT(s->source_output), SOURCE_OUTPUT_MESSAGE_UPDATE_LATENCY, s, 0, NULL) == 0);
+
     reply = reply_new(tag);
-    pa_tagstruct_put_usec(reply, s->source_output->source->monitor_of ? pa_sink_get_latency(s->source_output->source->monitor_of) : 0);
-    pa_tagstruct_put_usec(reply, pa_source_get_latency(s->source_output->source));
-    pa_tagstruct_put_boolean(reply, pa_source_get_state(s->source_output->source) == PA_SOURCE_RUNNING);
+    pa_tagstruct_put_usec(reply, s->current_monitor_latency);
+    pa_tagstruct_put_usec(reply,
+                          s->current_source_latency +
+                          pa_bytes_to_usec(s->on_the_fly_snapshot, &s->source_output->sample_spec));
+    pa_tagstruct_put_boolean(reply,
+                             pa_source_get_state(s->source_output->source) == PA_SOURCE_RUNNING &&
+                             pa_source_output_get_state(s->source_output) == PA_SOURCE_OUTPUT_RUNNING);
     pa_tagstruct_put_timeval(reply, &tv);
     pa_tagstruct_put_timeval(reply, pa_gettimeofday(&now));
     pa_tagstruct_puts64(reply, pa_memblockq_get_write_index(s->memblockq));
@@ -2625,7 +2703,9 @@ static void command_finish_upload_stream(pa_pdispatch *pd, uint32_t command, uin
     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
     CHECK_VALIDITY(c->pstream, upload_stream_isinstance(s), tag, PA_ERR_NOENTITY);
 
-    if (pa_scache_add_item(c->protocol->core, s->name, &s->sample_spec, &s->channel_map, &s->memchunk, s->proplist, &idx) < 0)
+    if (!s->memchunk.memblock)
+        pa_pstream_send_error(c->pstream, tag, PA_ERR_TOOLARGE);
+    else if (pa_scache_add_item(c->protocol->core, s->name, &s->sample_spec, &s->channel_map, &s->memchunk, s->proplist, &idx) < 0)
         pa_pstream_send_error(c->pstream, tag, PA_ERR_INTERNAL);
     else
         pa_pstream_send_simple_ack(c->pstream, tag);
@@ -2782,6 +2862,23 @@ static void sink_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_sin
         pa_tagstruct_putu32(t, sink->n_volume_steps);
         pa_tagstruct_putu32(t, sink->card ? sink->card->index : PA_INVALID_INDEX);
     }
+
+    if (c->version >= 16) {
+        pa_tagstruct_putu32(t, sink->ports ? pa_hashmap_size(sink->ports) : 0);
+
+        if (sink->ports) {
+            void *state;
+            pa_device_port *p;
+
+            PA_HASHMAP_FOREACH(p, sink->ports, state) {
+                pa_tagstruct_puts(t, p->name);
+                pa_tagstruct_puts(t, p->description);
+                pa_tagstruct_putu32(t, p->priority);
+            }
+        }
+
+        pa_tagstruct_puts(t, sink->active_port ? sink->active_port->name : NULL);
+    }
 }
 
 static void source_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_source *source) {
@@ -2822,6 +2919,24 @@ static void source_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_s
         pa_tagstruct_putu32(t, source->n_volume_steps);
         pa_tagstruct_putu32(t, source->card ? source->card->index : PA_INVALID_INDEX);
     }
+
+    if (c->version >= 16) {
+
+        pa_tagstruct_putu32(t, source->ports ? pa_hashmap_size(source->ports) : 0);
+
+        if (source->ports) {
+            void *state;
+            pa_device_port *p;
+
+            PA_HASHMAP_FOREACH(p, source->ports, state) {
+                pa_tagstruct_puts(t, p->name);
+                pa_tagstruct_puts(t, p->description);
+                pa_tagstruct_putu32(t, p->priority);
+            }
+        }
+
+        pa_tagstruct_puts(t, source->active_port ? source->active_port->name : NULL);
+    }
 }
 
 static void client_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_client *client) {
@@ -2884,6 +2999,7 @@ static void module_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_m
 static void sink_input_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_sink_input *s) {
     pa_sample_spec fixed_ss;
     pa_usec_t sink_latency;
+    pa_cvolume v;
 
     pa_assert(t);
     pa_sink_input_assert_ref(s);
@@ -2897,7 +3013,7 @@ static void sink_input_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t,
     pa_tagstruct_putu32(t, s->sink->index);
     pa_tagstruct_put_sample_spec(t, &fixed_ss);
     pa_tagstruct_put_channel_map(t, &s->channel_map);
-    pa_tagstruct_put_cvolume(t, pa_sink_input_get_volume(s));
+    pa_tagstruct_put_cvolume(t, pa_sink_input_get_volume(s, &v, TRUE));
     pa_tagstruct_put_usec(t, pa_sink_input_get_latency(s, &sink_latency));
     pa_tagstruct_put_usec(t, sink_latency);
     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_sink_input_get_resample_method(s)));
@@ -3122,10 +3238,10 @@ static void command_get_info_list(pa_pdispatch *pd, uint32_t command, uint32_t t
 static void command_get_server_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
     pa_tagstruct *reply;
-    char txt[256];
     pa_sink *def_sink;
     pa_source *def_source;
     pa_sample_spec fixed_ss;
+    char *h, *u;
 
     pa_native_connection_assert_ref(c);
     pa_assert(t);
@@ -3140,8 +3256,14 @@ static void command_get_server_info(pa_pdispatch *pd, uint32_t command, uint32_t
     reply = reply_new(tag);
     pa_tagstruct_puts(reply, PACKAGE_NAME);
     pa_tagstruct_puts(reply, PACKAGE_VERSION);
-    pa_tagstruct_puts(reply, pa_get_user_name(txt, sizeof(txt)));
-    pa_tagstruct_puts(reply, pa_get_host_name(txt, sizeof(txt)));
+
+    u = pa_get_user_name_malloc();
+    pa_tagstruct_puts(reply, u);
+    pa_xfree(u);
+
+    h = pa_get_host_name_malloc();
+    pa_tagstruct_puts(reply, h);
+    pa_xfree(h);
 
     fixup_sample_spec(c, &fixed_ss, &c->protocol->core->default_sample_spec);
     pa_tagstruct_put_sample_spec(reply, &fixed_ss);
@@ -3215,6 +3337,7 @@ static void command_set_volume(
     pa_source *source = NULL;
     pa_sink_input *si = NULL;
     const char *name = NULL;
+    const char *client_name;
 
     pa_native_connection_assert_ref(c);
     pa_assert(t);
@@ -3261,12 +3384,20 @@ static void command_set_volume(
 
     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
 
-    if (sink)
+    client_name = pa_strnull(pa_proplist_gets(c->client->proplist, PA_PROP_APPLICATION_PROCESS_BINARY));
+
+    if (sink) {
+        pa_log_debug("Client %s changes volume of sink %s.", client_name, sink->name);
         pa_sink_set_volume(sink, &volume, TRUE, TRUE);
-    else if (source)
-        pa_source_set_volume(source, &volume);
-    else if (si)
-        pa_sink_input_set_volume(si, &volume, TRUE);
+    } else if (source) {
+        pa_log_debug("Client %s changes volume of sink %s.", client_name, source->name);
+        pa_source_set_volume(source, &volume, TRUE);
+    } else if (si) {
+        pa_log_debug("Client %s changes volume of sink %s.",
+                     client_name,
+                     pa_strnull(pa_proplist_gets(si->proplist, PA_PROP_MEDIA_NAME)));
+        pa_sink_input_set_volume(si, &volume, TRUE, TRUE);
+    }
 
     pa_pstream_send_simple_ack(c->pstream, tag);
 }
@@ -3334,9 +3465,9 @@ static void command_set_mute(
     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
 
     if (sink)
-        pa_sink_set_mute(sink, mute);
+        pa_sink_set_mute(sink, mute, TRUE);
     else if (source)
-        pa_source_set_mute(source, mute);
+        pa_source_set_mute(source, mute, TRUE);
     else if (si)
         pa_sink_input_set_mute(si, mute, TRUE);
 
@@ -3514,7 +3645,7 @@ static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, u
         pa_tagstruct_putu32(reply, s->buffer_attr.minreq);
 
         if (c->version >= 13)
-            pa_tagstruct_put_usec(reply, s->sink_latency);
+            pa_tagstruct_put_usec(reply, s->configured_sink_latency);
 
     } else {
         record_stream *s;
@@ -3550,7 +3681,7 @@ static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, u
         pa_tagstruct_putu32(reply, s->buffer_attr.fragsize);
 
         if (c->version >= 13)
-            pa_tagstruct_put_usec(reply, s->source_latency);
+            pa_tagstruct_put_usec(reply, s->configured_source_latency);
     }
 
     pa_pstream_send_tagstruct(c->pstream, reply);
@@ -4032,7 +4163,7 @@ static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa
 
             pa_log_debug("%s all sinks", b ? "Suspending" : "Resuming");
 
-            if (pa_sink_suspend_all(c->protocol->core, b) < 0) {
+            if (pa_sink_suspend_all(c->protocol->core, b, PA_SUSPEND_USER) < 0) {
                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
                 return;
             }
@@ -4046,7 +4177,7 @@ static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa
 
             CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
 
-            if (pa_sink_suspend(sink, b) < 0) {
+            if (pa_sink_suspend(sink, b, PA_SUSPEND_USER) < 0) {
                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
                 return;
             }
@@ -4059,7 +4190,7 @@ static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa
 
             pa_log_debug("%s all sources", b ? "Suspending" : "Resuming");
 
-            if (pa_source_suspend_all(c->protocol->core, b) < 0) {
+            if (pa_source_suspend_all(c->protocol->core, b, PA_SUSPEND_USER) < 0) {
                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
                 return;
             }
@@ -4074,7 +4205,7 @@ static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa
 
             CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
 
-            if (pa_source_suspend(source, b) < 0) {
+            if (pa_source_suspend(source, b, PA_SUSPEND_USER) < 0) {
                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
                 return;
             }
@@ -4129,6 +4260,7 @@ static void command_set_card_profile(pa_pdispatch *pd, uint32_t command, uint32_
     uint32_t idx = PA_INVALID_INDEX;
     const char *name = NULL, *profile = NULL;
     pa_card *card = NULL;
+    int ret;
 
     pa_native_connection_assert_ref(c);
     pa_assert(t);
@@ -4154,14 +4286,72 @@ static void command_set_card_profile(pa_pdispatch *pd, uint32_t command, uint32_
 
     CHECK_VALIDITY(c->pstream, card, tag, PA_ERR_NOENTITY);
 
-    if (pa_card_set_profile(card, profile, TRUE) < 0) {
-        pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
+    if ((ret = pa_card_set_profile(card, profile, TRUE)) < 0) {
+        pa_pstream_send_error(c->pstream, tag, -ret);
         return;
     }
 
     pa_pstream_send_simple_ack(c->pstream, tag);
 }
 
+static void command_set_sink_or_source_port(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
+    uint32_t idx = PA_INVALID_INDEX;
+    const char *name = NULL, *port = NULL;
+    int ret;
+
+    pa_native_connection_assert_ref(c);
+    pa_assert(t);
+
+    if (pa_tagstruct_getu32(t, &idx) < 0 ||
+        pa_tagstruct_gets(t, &name) < 0 ||
+        pa_tagstruct_gets(t, &port) < 0 ||
+        !pa_tagstruct_eof(t)) {
+        protocol_error(c);
+        return;
+    }
+
+    CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
+    CHECK_VALIDITY(c->pstream, !name || pa_namereg_is_valid_name(name), tag, PA_ERR_INVALID);
+    CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
+    CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
+    CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
+
+    if (command == PA_COMMAND_SET_SINK_PORT) {
+        pa_sink *sink;
+
+        if (idx != PA_INVALID_INDEX)
+            sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
+        else
+            sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
+
+        CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
+
+        if ((ret = pa_sink_set_port(sink, port, TRUE)) < 0) {
+            pa_pstream_send_error(c->pstream, tag, -ret);
+            return;
+        }
+    } else {
+        pa_source *source;
+
+        pa_assert(command = PA_COMMAND_SET_SOURCE_PORT);
+
+        if (idx != PA_INVALID_INDEX)
+            source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
+        else
+            source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
+
+        CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
+
+        if ((ret = pa_source_set_port(source, port, TRUE)) < 0) {
+            pa_pstream_send_error(c->pstream, tag, -ret);
+            return;
+        }
+    }
+
+    pa_pstream_send_simple_ack(c->pstream, tag);
+}
+
 /*** pstream callbacks ***/
 
 static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
@@ -4186,7 +4376,7 @@ static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t o
     pa_native_connection_assert_ref(c);
 
     if (!(stream = OUTPUT_STREAM(pa_idxset_get_by_index(c->output_streams, channel)))) {
-        pa_log("client sent block for invalid stream.");
+        pa_log_debug("Client sent block for invalid stream.");
         /* Ignoring */
         return;
     }
@@ -4315,11 +4505,10 @@ static void client_send_event_cb(pa_client *client, const char*event, pa_proplis
 
 /*** module entry points ***/
 
-static void auth_timeout(pa_mainloop_api*m, pa_time_event *e, const struct timeval *tv, void *userdata) {
+static void auth_timeout(pa_mainloop_api*m, pa_time_event *e, const struct timeval *t, void *userdata) {
     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
 
     pa_assert(m);
-    pa_assert(tv);
     pa_native_connection_assert_ref(c);
     pa_assert(c->auth_timeout_event == e);
 
@@ -4377,12 +4566,9 @@ void pa_native_protocol_connect(pa_native_protocol *p, pa_iochannel *io, pa_nati
         c->authorized = TRUE;
     }
 
-    if (!c->authorized) {
-        struct timeval tv;
-        pa_gettimeofday(&tv);
-        tv.tv_sec += AUTH_TIMEOUT;
-        c->auth_timeout_event = p->core->mainloop->time_new(p->core->mainloop, &tv, auth_timeout, c);
-    } else
+    if (!c->authorized)
+        c->auth_timeout_event = pa_core_rttime_new(p->core, pa_rtclock_now() + AUTH_TIMEOUT, auth_timeout, c);
+    else
         c->auth_timeout_event = NULL;
 
     c->is_local = pa_iochannel_socket_is_local(io);
@@ -4401,7 +4587,7 @@ void pa_native_protocol_connect(pa_native_protocol *p, pa_iochannel *io, pa_nati
     pa_pstream_set_revoke_callback(c->pstream, pstream_revoke_callback, c);
     pa_pstream_set_release_callback(c->pstream, pstream_release_callback, c);
 
-    c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
+    c->pdispatch = pa_pdispatch_new(p->core->mainloop, TRUE, command_table, PA_COMMAND_MAX);
 
     c->record_streams = pa_idxset_new(NULL, NULL);
     c->output_streams = pa_idxset_new(NULL, NULL);