- Fix moving of sink inputs between sinks

- Don't write more than a single buffer size in the ALSA driver at a time, to give the clients time to fill up the memblockq again
- Add API for querying the requested latency of a sink input/source output
- Drop get_letancy() from vtable of sinks/sources


git-svn-id: file:///home/lennart/svn/public/pulseaudio/branches/glitch-free@2392 fefdeb5f-60dc-0310-8127-8f9354f1896f

18 files changed, 528 insertions, 445 deletions

diff --git a/src/modules/module-alsa-sink.c b/src/modules/module-alsa-sink.c
index 4a997cd1..97e1e59f 100644
--- a/src/modules/module-alsa-sink.c
+++ b/src/modules/module-alsa-sink.c
@@ -131,6 +131,7 @@ struct userdata {
 
     pa_smoother *smoother;
     int64_t frame_index;
+    uint64_t since_start;
 
     snd_pcm_sframes_t hwbuf_unused_frames;
 };
@@ -162,6 +163,32 @@ static void fix_tsched_watermark(struct userdata *u) {
         u->tsched_watermark = min_wakeup;
 }
 
+static void hw_sleep_time(struct userdata *u, pa_usec_t *sleep_usec, pa_usec_t*process_usec) {
+    pa_usec_t usec, wm;
+
+    pa_assert(sleep_usec);
+    pa_assert(process_usec);
+
+    pa_assert(u);
+
+    usec = pa_sink_get_requested_latency_within_thread(u->sink);
+
+    if (usec == (pa_usec_t) -1)
+        usec = pa_bytes_to_usec(u->hwbuf_size, &u->sink->sample_spec);
+
+/*     pa_log_debug("hw buffer time: %u ms", (unsigned) (usec / PA_USEC_PER_MSEC)); */
+
+    wm = pa_bytes_to_usec(u->tsched_watermark, &u->sink->sample_spec);
+
+    if (usec >= wm) {
+        *sleep_usec = usec - wm;
+        *process_usec = wm;
+    } else
+        *process_usec = *sleep_usec = usec / 2;
+
+/*     pa_log_debug("after watermark: %u ms", (unsigned) (*sleep_usec / PA_USEC_PER_MSEC)); */
+}
+
 static int try_recover(struct userdata *u, const char *call, int err) {
     pa_assert(u);
     pa_assert(call);
@@ -169,16 +196,14 @@ static int try_recover(struct userdata *u, const char *call, int err) {
 
     pa_log_debug("%s: %s", call, snd_strerror(err));
 
-    if (err == -EAGAIN) {
-        pa_log_debug("%s: EAGAIN", call);
-        return 1;
-    }
+    pa_assert(err != -EAGAIN);
 
     if (err == -EPIPE)
         pa_log_debug("%s: Buffer underrun!", call);
 
     if ((err = snd_pcm_recover(u->pcm_handle, err, 1)) == 0) {
         u->first = TRUE;
+        u->since_start = 0;
         return 0;
     }
 
@@ -186,20 +211,17 @@ static int try_recover(struct userdata *u, const char *call, int err) {
     return -1;
 }
 
-static void check_left_to_play(struct userdata *u, snd_pcm_sframes_t n) {
+static size_t check_left_to_play(struct userdata *u, snd_pcm_sframes_t n) {
     size_t left_to_play;
 
-    if (u->first || u->after_rewind)
-        return;
-
     if (n*u->frame_size < u->hwbuf_size)
         left_to_play = u->hwbuf_size - (n*u->frame_size);
     else
         left_to_play = 0;
 
-    if (left_to_play > 0)
-        pa_log_debug("%0.2f ms left to play", (double) pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) / PA_USEC_PER_MSEC);
-    else {
+    if (left_to_play > 0) {
+/*         pa_log_debug("%0.2f ms left to play", (double) pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) / PA_USEC_PER_MSEC); */
+    } else if (!u->first && !u->after_rewind) {
         pa_log_info("Underrun!");
 
         if (u->use_tsched) {
@@ -213,22 +235,24 @@ static void check_left_to_play(struct userdata *u, snd_pcm_sframes_t n) {
                               (double) pa_bytes_to_usec(u->tsched_watermark, &u->sink->sample_spec) / PA_USEC_PER_MSEC);
         }
     }
+
+    return left_to_play;
 }
 
-static int mmap_write(struct userdata *u) {
+static int mmap_write(struct userdata *u, pa_usec_t *sleep_usec) {
     int work_done = 0;
-    pa_bool_t checked_left_to_play = FALSE;
+    pa_usec_t max_sleep_usec, process_usec;
+    size_t left_to_play;
 
     pa_assert(u);
     pa_sink_assert_ref(u->sink);
 
+    if (u->use_tsched)
+        hw_sleep_time(u, &max_sleep_usec, &process_usec);
+
     for (;;) {
-        pa_memchunk chunk;
-        void *p;
         snd_pcm_sframes_t n;
-        int err, r;
-        const snd_pcm_channel_area_t *areas;
-        snd_pcm_uframes_t offset, frames;
+        int r;
 
         snd_pcm_hwsync(u->pcm_handle);
 
@@ -239,92 +263,110 @@ static int mmap_write(struct userdata *u) {
 
             if ((r = try_recover(u, "snd_pcm_avail_update", n)) == 0)
                 continue;
-            else if (r > 0)
-                return work_done;
 
             return r;
         }
 
-        if (!checked_left_to_play) {
-            check_left_to_play(u, n);
-            checked_left_to_play = TRUE;
-        }
+        left_to_play = check_left_to_play(u, n);
+
+        if (u->use_tsched)
+
+            /* We won't fill up the playback buffer before at least
+            * half the sleep time is over because otherwise we might
+            * ask for more data from the clients then they expect. We
+            * need to guarantee that clients only have to keep around
+            * a single hw buffer length. */
 
-        /* We only use part of the buffer that matches our
-         * dynamically requested latency */
+            if (pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) > max_sleep_usec/2)
+                break;
 
         if (PA_UNLIKELY(n <= u->hwbuf_unused_frames))
-            return work_done;
+            break;
 
-        frames = n = n - u->hwbuf_unused_frames;
+        n -= u->hwbuf_unused_frames;
 
-/*        pa_log_debug("%lu frames to write", (unsigned long) frames);*/
+/*         pa_log_debug("Filling up"); */
 
-        if (PA_UNLIKELY((err = snd_pcm_mmap_begin(u->pcm_handle, &areas, &offset, &frames)) < 0)) {
+        for (;;) {
+            pa_memchunk chunk;
+            void *p;
+            int err;
+            const snd_pcm_channel_area_t *areas;
+            snd_pcm_uframes_t offset, frames = (snd_pcm_uframes_t) n;
 
-            if ((r = try_recover(u, "snd_pcm_mmap_begin", err)) == 0)
-                continue;
-            else if (r > 0)
-                return work_done;
+/*             pa_log_debug("%lu frames to write", (unsigned long) frames); */
 
-            return r;
-        }
+            if (PA_UNLIKELY((err = snd_pcm_mmap_begin(u->pcm_handle, &areas, &offset, &frames)) < 0)) {
 
-        /* Make sure that if these memblocks need to be copied they will fit into one slot */
-        if (frames > pa_mempool_block_size_max(u->sink->core->mempool)/u->frame_size)
-            frames = pa_mempool_block_size_max(u->sink->core->mempool)/u->frame_size;
+                if ((r = try_recover(u, "snd_pcm_mmap_begin", err)) == 0)
+                    continue;
 
-        /* Check these are multiples of 8 bit */
-        pa_assert((areas[0].first & 7) == 0);
-        pa_assert((areas[0].step & 7)== 0);
+                return r;
+            }
 
-        /* We assume a single interleaved memory buffer */
-        pa_assert((areas[0].first >> 3) == 0);
-        pa_assert((areas[0].step >> 3) == u->frame_size);
+            /* Make sure that if these memblocks need to be copied they will fit into one slot */
+            if (frames > pa_mempool_block_size_max(u->sink->core->mempool)/u->frame_size)
+                frames = pa_mempool_block_size_max(u->sink->core->mempool)/u->frame_size;
 
-        p = (uint8_t*) areas[0].addr + (offset * u->frame_size);
+            /* Check these are multiples of 8 bit */
+            pa_assert((areas[0].first & 7) == 0);
+            pa_assert((areas[0].step & 7)== 0);
 
-        chunk.memblock = pa_memblock_new_fixed(u->core->mempool, p, frames * u->frame_size, TRUE);
-        chunk.length = pa_memblock_get_length(chunk.memblock);
-        chunk.index = 0;
+            /* We assume a single interleaved memory buffer */
+            pa_assert((areas[0].first >> 3) == 0);
+            pa_assert((areas[0].step >> 3) == u->frame_size);
 
-        pa_sink_render_into_full(u->sink, &chunk);
+            p = (uint8_t*) areas[0].addr + (offset * u->frame_size);
 
-        /* FIXME: Maybe we can do something to keep this memory block
-         * a little bit longer around? */
-        pa_memblock_unref_fixed(chunk.memblock);
+            chunk.memblock = pa_memblock_new_fixed(u->core->mempool, p, frames * u->frame_size, TRUE);
+            chunk.length = pa_memblock_get_length(chunk.memblock);
+            chunk.index = 0;
 
-        if (PA_UNLIKELY((err = snd_pcm_mmap_commit(u->pcm_handle, offset, frames)) < 0)) {
+            pa_sink_render_into_full(u->sink, &chunk);
 
-            if ((r = try_recover(u, "snd_pcm_mmap_commit", err)) == 0)
-                continue;
-            else if (r > 0)
-                return work_done;
+            /* FIXME: Maybe we can do something to keep this memory block
+             * a little bit longer around? */
+            pa_memblock_unref_fixed(chunk.memblock);
 
-            return r;
-        }
+            if (PA_UNLIKELY((err = snd_pcm_mmap_commit(u->pcm_handle, offset, frames)) < 0)) {
 
-        work_done = 1;
+                if ((r = try_recover(u, "snd_pcm_mmap_commit", err)) == 0)
+                    continue;
 
-        u->frame_index += frames;
+                return r;
+            }
 
-/*         pa_log_debug("wrote %lu frames", (unsigned long) frames); */
+            work_done = 1;
+
+            u->frame_index += frames;
+            u->since_start += frames * u->frame_size;
+
+/*             pa_log_debug("wrote %lu frames", (unsigned long) frames); */
 
-        if (PA_LIKELY(frames >= (snd_pcm_uframes_t) n))
-            return work_done;
+            if (frames >= (snd_pcm_uframes_t) n)
+                break;
+
+            n -= frames;
+        }
     }
+
+    *sleep_usec = pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) - process_usec;
+    return work_done;
 }
 
-static int unix_write(struct userdata *u) {
+static int unix_write(struct userdata *u, pa_usec_t *sleep_usec) {
     int work_done = 0;
-    pa_bool_t checked_left_to_play = FALSE;
+    pa_usec_t max_sleep_usec, process_usec;
+    size_t left_to_play;
 
     pa_assert(u);
     pa_sink_assert_ref(u->sink);
 
+    if (u->use_tsched)
+        hw_sleep_time(u, &max_sleep_usec, &process_usec);
+
     for (;;) {
-        void *p;
-        snd_pcm_sframes_t n, frames;
+        snd_pcm_sframes_t n;
         int r;
 
         snd_pcm_hwsync(u->pcm_handle);
@@ -333,67 +375,82 @@ static int unix_write(struct userdata *u) {
 
             if ((r = try_recover(u, "snd_pcm_avail_update", n)) == 0)
                 continue;
-            else if (r > 0)
-                return work_done;
 
             return r;
         }
 
-        if (!checked_left_to_play) {
-            check_left_to_play(u, n);
-            checked_left_to_play = TRUE;
-        }
+        left_to_play = check_left_to_play(u, n);
+
+        if (u->use_tsched)
+
+            /* We won't fill up the playback buffer before at least
+            * half the sleep time is over because otherwise we might
+            * ask for more data from the clients then they expect. We
+            * need to guarantee that clients only have to keep around
+            * a single hw buffer length. */
+
+            if (pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) > max_sleep_usec/2)
+                break;
 
         if (PA_UNLIKELY(n <= u->hwbuf_unused_frames))
-            return work_done;
+            break;
 
         n -= u->hwbuf_unused_frames;
 
+        for (;;) {
+            snd_pcm_sframes_t frames;
+            void *p;
+
 /*         pa_log_debug("%lu frames to write", (unsigned long) frames); */
 
-        if (u->memchunk.length <= 0)
-            pa_sink_render(u->sink, n * u->frame_size, &u->memchunk);
+            if (u->memchunk.length <= 0)
+                pa_sink_render(u->sink, n * u->frame_size, &u->memchunk);
 
-        pa_assert(u->memchunk.length > 0);
+            pa_assert(u->memchunk.length > 0);
 
-        frames = u->memchunk.length / u->frame_size;
+            frames = u->memchunk.length / u->frame_size;
 
-        if (frames > n)
-            frames = n;
+            if (frames > n)
+                frames = n;
 
-        p = pa_memblock_acquire(u->memchunk.memblock);
-        frames = snd_pcm_writei(u->pcm_handle, (const uint8_t*) p + u->memchunk.index, frames);
-        pa_memblock_release(u->memchunk.memblock);
+            p = pa_memblock_acquire(u->memchunk.memblock);
+            frames = snd_pcm_writei(u->pcm_handle, (const uint8_t*) p + u->memchunk.index, frames);
+            pa_memblock_release(u->memchunk.memblock);
 
-        pa_assert(frames != 0);
+            pa_assert(frames != 0);
 
-        if (PA_UNLIKELY(frames < 0)) {
+            if (PA_UNLIKELY(frames < 0)) {
 
-            if ((r = try_recover(u, "snd_pcm_writei", n)) == 0)
-                continue;
-            else if (r > 0)
-                return work_done;
+                if ((r = try_recover(u, "snd_pcm_writei", n)) == 0)
+                    continue;
 
-            return r;
-        }
+                return r;
+            }
 
-        u->memchunk.index += frames * u->frame_size;
-        u->memchunk.length -= frames * u->frame_size;
+            u->memchunk.index += frames * u->frame_size;
+            u->memchunk.length -= frames * u->frame_size;
 
-        if (u->memchunk.length <= 0) {
-            pa_memblock_unref(u->memchunk.memblock);
-            pa_memchunk_reset(&u->memchunk);
-        }
+            if (u->memchunk.length <= 0) {
+                pa_memblock_unref(u->memchunk.memblock);
+                pa_memchunk_reset(&u->memchunk);
+            }
 
-        work_done = 1;
+            work_done = 1;
 
-        u->frame_index += frames;
+            u->frame_index += frames;
+            u->since_start += frames * u->frame_size;
 
 /*         pa_log_debug("wrote %lu frames", (unsigned long) frames); */
 
-        if (PA_LIKELY(frames >= n))
-            return work_done;
+            if (frames >= n)
+                break;
+
+            n -= frames;
+        }
     }
+
+    *sleep_usec = pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) - process_usec;
+    return work_done;
 }
 
 static void update_smoother(struct userdata *u) {
@@ -494,35 +551,6 @@ static int suspend(struct userdata *u) {
     return 0;
 }
 
-static pa_usec_t hw_sleep_time(struct userdata *u) {
-    pa_usec_t usec, wm;
-
-    pa_assert(u);
-
-    usec = pa_sink_get_requested_latency_within_thread(u->sink);
-
-    if (usec == (pa_usec_t) -1)
-        usec = pa_bytes_to_usec(u->hwbuf_size, &u->sink->sample_spec);
-
-    pa_log_debug("hw buffer time: %u ms", (unsigned) (usec / PA_USEC_PER_MSEC));
-
-    wm = pa_bytes_to_usec(u->tsched_watermark, &u->sink->sample_spec);
-
-    if (usec >= wm)
-        usec -= wm;
-    else
-        usec /= 2;
-
-    if (u->first) {
-        pa_log_debug("Decreasing wakeup time for the first iteration by half.");
-        usec /= 2;
-    }
-
-    pa_log_debug("after watermark: %u ms", (unsigned) (usec / PA_USEC_PER_MSEC));
-
-    return usec;
-}
-
 static int update_sw_params(struct userdata *u) {
     snd_pcm_uframes_t avail_min;
     int err;
@@ -561,10 +589,10 @@ static int update_sw_params(struct userdata *u) {
     avail_min = u->hwbuf_unused_frames + 1;
 
     if (u->use_tsched) {
-        pa_usec_t usec;
+        pa_usec_t sleep_usec, process_usec;
 
-        usec = hw_sleep_time(u);
-        avail_min += pa_usec_to_bytes(usec, &u->sink->sample_spec);
+        hw_sleep_time(u, &sleep_usec, &process_usec);
+        avail_min += pa_usec_to_bytes(sleep_usec, &u->sink->sample_spec);
     }
 
     pa_log_debug("setting avail_min=%lu", (unsigned long) avail_min);
@@ -630,6 +658,7 @@ static int unsuspend(struct userdata *u) {
     /* FIXME: We need to reload the volume somehow */
 
     u->first = TRUE;
+    u->since_start = 0;
 
     pa_log_info("Resumed successfully...");
 
@@ -932,16 +961,17 @@ static void thread_func(void *userdata) {
 
         /* Render some data and write it to the dsp */
         if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
-            int work_done = 0;
+            int work_done;
+            pa_usec_t sleep_usec;
 
             if (u->sink->thread_info.rewind_nbytes > 0)
                 if (process_rewind(u) < 0)
                     goto fail;
 
             if (u->use_mmap)
-                work_done = mmap_write(u);
+                work_done = mmap_write(u, &sleep_usec);
             else
-                work_done = unix_write(u);
+                work_done = unix_write(u, &sleep_usec);
 
             if (work_done < 0)
                 goto fail;
@@ -961,23 +991,34 @@ static void thread_func(void *userdata) {
             }
 
             if (u->use_tsched) {
-                pa_usec_t usec, cusec;
+                pa_usec_t cusec;
 
-                /* OK, the playback buffer is now full, let's
-                 * calculate when to wake up next */
+                if (u->since_start <= u->hwbuf_size) {
+
+                    /* USB devices on ALSA seem to hit a buffer
+                     * underrun during the first iterations much
+                     * quicker then we calculate here, probably due to
+                     * the transport latency. To accomodate for that
+                     * we artificially decrease the sleep time until
+                     * we have filled the buffer at least once
+                     * completely.*/
 
-                usec = hw_sleep_time(u);
+                    pa_log_debug("Cutting sleep time for the initial iterations by half.");
+                    sleep_usec /= 2;
+                }
 
-/*                 pa_log_debug("Waking up in %0.2fms (sound card clock).", (double) usec / PA_USEC_PER_MSEC); */
+                /* OK, the playback buffer is now full, let's
+                 * calculate when to wake up next */
+/*                 pa_log_debug("Waking up in %0.2fms (sound card clock).", (double) sleep_usec / PA_USEC_PER_MSEC); */
 
                 /* Convert from the sound card time domain to the
                  * system time domain */
-                cusec = pa_smoother_translate(u->smoother, pa_rtclock_usec(), usec);
+                cusec = pa_smoother_translate(u->smoother, pa_rtclock_usec(), sleep_usec);
 
 /*                 pa_log_debug("Waking up in %0.2fms (system clock).", (double) cusec / PA_USEC_PER_MSEC); */
 
                 /* We don't trust the conversion, so we wake up whatever comes first */
-                pa_rtpoll_set_timer_relative(u->rtpoll, PA_MIN(usec, cusec));
+                pa_rtpoll_set_timer_relative(u->rtpoll, PA_MIN(sleep_usec, cusec));
             }
 
             u->first = FALSE;
@@ -1014,6 +1055,7 @@ static void thread_func(void *userdata) {
                     goto fail;
 
                 u->first = TRUE;
+                u->since_start = 0;
             }
 
             if (revents)
@@ -1115,12 +1157,13 @@ int pa__init(pa_module*m) {
     u->use_mmap = use_mmap;
     u->use_tsched = use_tsched;
     u->first = TRUE;
+    u->since_start = 0;
     u->after_rewind = FALSE;
     u->rtpoll = pa_rtpoll_new();
     pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
     u->alsa_rtpoll_item = NULL;
 
-    u->smoother = pa_smoother_new(DEFAULT_TSCHED_BUFFER_USEC*2, DEFAULT_TSCHED_BUFFER_USEC*2, TRUE);
+    u->smoother = pa_smoother_new(DEFAULT_TSCHED_BUFFER_USEC*2, DEFAULT_TSCHED_BUFFER_USEC*2, TRUE, 5);
     usec = pa_rtclock_usec();
     pa_smoother_set_time_offset(u->smoother, usec);
     pa_smoother_pause(u->smoother, usec);
diff --git a/src/modules/module-alsa-source.c b/src/modules/module-alsa-source.c
index 9eb6f06b..4838ad27 100644
--- a/src/modules/module-alsa-source.c
+++ b/src/modules/module-alsa-source.c
@@ -988,7 +988,7 @@ int pa__init(pa_module*m) {
     pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
     u->alsa_rtpoll_item = NULL;
 
-    u->smoother = pa_smoother_new(DEFAULT_TSCHED_WATERMARK_USEC, DEFAULT_TSCHED_WATERMARK_USEC, TRUE);
+    u->smoother = pa_smoother_new(DEFAULT_TSCHED_WATERMARK_USEC, DEFAULT_TSCHED_WATERMARK_USEC, TRUE, 5);
     pa_smoother_set_time_offset(u->smoother, pa_rtclock_usec());
 
     snd_config_update_free_global();
diff --git a/src/modules/module-combine.c b/src/modules/module-combine.c
index 2409ef8f..fc8be18d 100644
--- a/src/modules/module-combine.c
+++ b/src/modules/module-combine.c
@@ -618,35 +618,35 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
 }
 
 /* Called from main context */
-static pa_usec_t sink_get_latency_cb(pa_sink *s) {
-    struct userdata *u;
+/* static pa_usec_t sink_get_latency_cb(pa_sink *s) { */
+/*     struct userdata *u; */
 
-    pa_sink_assert_ref(s);
-    pa_assert_se(u = s->userdata);
+/*     pa_sink_assert_ref(s); */
+/*     pa_assert_se(u = s->userdata); */
 
-    if (u->master) {
-        /* If we have a master sink, we just return the latency of it
-         * and add our own buffering on top */
+/*     if (u->master) { */
+/*         /\* If we have a master sink, we just return the latency of it */
+/*          * and add our own buffering on top *\/ */
 
-        if (!u->master->sink_input)
-            return 0;
+/*         if (!u->master->sink_input) */
+/*             return 0; */
 
-        return
-            pa_sink_input_get_latency(u->master->sink_input) +
-            pa_sink_get_latency(u->master->sink);
+/*         return */
+/*             pa_sink_input_get_latency(u->master->sink_input) + */
+/*             pa_sink_get_latency(u->master->sink); */
 
-    } else {
-        pa_usec_t usec = 0;
+/*     } else { */
+/*         pa_usec_t usec = 0; */
 
-        /* We have no master, hence let's ask our own thread which
-         * implements the NULL sink */
+/*         /\* We have no master, hence let's ask our own thread which */
+/*          * implements the NULL sink *\/ */
 
-        if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
-            return 0;
+/*         if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0) */
+/*             return 0; */
 
-        return usec;
-    }
-}
+/*         return usec; */
+/*     } */
+/* } */
 
 static void update_description(struct userdata *u) {
     int first = 1;
@@ -1025,7 +1025,7 @@ int pa__init(pa_module*m) {
     }
 
     u->sink->parent.process_msg = sink_process_msg;
-    u->sink->get_latency = sink_get_latency_cb;
+/*     u->sink->get_latency = sink_get_latency_cb; */
     u->sink->set_state = sink_set_state;
     u->sink->userdata = u;
 
diff --git a/src/modules/module-esound-sink.c b/src/modules/module-esound-sink.c
index 2206e2bc..4aa7d674 100644
--- a/src/modules/module-esound-sink.c
+++ b/src/modules/module-esound-sink.c
@@ -534,7 +534,7 @@ int pa__init(pa_module*m) {
     u->module = m;
     m->userdata = u;
     u->fd = -1;
-    u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE);
+    u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE, 10);
     pa_memchunk_reset(&u->memchunk);
     u->offset = 0;
 
diff --git a/src/modules/module-rescue-streams.c b/src/modules/module-rescue-streams.c
index dda54735..7241a99f 100644
--- a/src/modules/module-rescue-streams.c
+++ b/src/modules/module-rescue-streams.c
@@ -75,7 +75,7 @@ static pa_hook_result_t sink_hook_callback(pa_core *c, pa_sink *sink, void* user
     }
 
     while ((i = pa_idxset_first(sink->inputs, NULL))) {
-        if (pa_sink_input_move_to(i, target, 1) < 0) {
+        if (pa_sink_input_move_to(i, target) < 0) {
             pa_log_warn("Failed to move sink input %u \"%s\" to %s.", i->index, pa_proplist_gets(i->proplist, PA_PROP_APPLICATION_NAME), target->name);
             return PA_HOOK_OK;
         }
diff --git a/src/modules/module-tunnel.c b/src/modules/module-tunnel.c
index e3ae5e1f..7a87fd8c 100644
--- a/src/modules/module-tunnel.c
+++ b/src/modules/module-tunnel.c
@@ -577,29 +577,29 @@ static void timeout_callback(pa_mainloop_api *m, pa_time_event*e, PA_GCC_UNUSED
 }
 
 #ifdef TUNNEL_SINK
-static pa_usec_t sink_get_latency(pa_sink *s) {
-    pa_usec_t t, c;
-    struct userdata *u = s->userdata;
+/* static pa_usec_t sink_get_latency(pa_sink *s) { */
+/*     pa_usec_t t, c; */
+/*     struct userdata *u = s->userdata; */
 
-    pa_sink_assert_ref(s);
+/*     pa_sink_assert_ref(s); */
 
-    c = pa_bytes_to_usec(u->counter, &s->sample_spec);
-    t = pa_smoother_get(u->smoother, pa_rtclock_usec());
+/*     c = pa_bytes_to_usec(u->counter, &s->sample_spec); */
+/*     t = pa_smoother_get(u->smoother, pa_rtclock_usec()); */
 
-    return c > t ? c - t : 0;
-}
+/*     return c > t ? c - t : 0; */
+/* } */
 #else
-static pa_usec_t source_get_latency(pa_source *s) {
-    pa_usec_t t, c;
-    struct userdata *u = s->userdata;
+/* static pa_usec_t source_get_latency(pa_source *s) { */
+/*     pa_usec_t t, c; */
+/*     struct userdata *u = s->userdata; */
 
-    pa_source_assert_ref(s);
+/*     pa_source_assert_ref(s); */
 
-    c = pa_bytes_to_usec(u->counter, &s->sample_spec);
-    t = pa_smoother_get(u->smoother, pa_rtclock_usec());
+/*     c = pa_bytes_to_usec(u->counter, &s->sample_spec); */
+/*     t = pa_smoother_get(u->smoother, pa_rtclock_usec()); */
 
-    return t > c ? t - c : 0;
-}
+/*     return t > c ? t - c : 0; */
+/* } */
 #endif
 
 static void update_description(struct userdata *u) {
@@ -1323,7 +1323,7 @@ int pa__init(pa_module*m) {
     u->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));;
     u->source = NULL;
 #endif
-    u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE);
+    u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE, 10);
     u->ctag = 1;
     u->device_index = u->channel = PA_INVALID_INDEX;
     u->auth_cookie_in_property = FALSE;
@@ -1377,7 +1377,7 @@ int pa__init(pa_module*m) {
     u->sink->parent.process_msg = sink_process_msg;
     u->sink->userdata = u;
     u->sink->set_state = sink_set_state;
-    u->sink->get_latency = sink_get_latency;
+/*     u->sink->get_latency = sink_get_latency; */
     u->sink->get_volume = sink_get_volume;
     u->sink->get_mute = sink_get_mute;
     u->sink->set_volume = sink_set_volume;
@@ -1412,7 +1412,7 @@ int pa__init(pa_module*m) {
     u->source->parent.process_msg = source_process_msg;
     u->source->userdata = u;
     u->source->set_state = source_set_state;
-    u->source->get_latency = source_get_latency;
+/*     u->source->get_latency = source_get_latency; */
 
     pa_source_set_asyncmsgq(u->source, u->thread_mq.inq);
     pa_source_set_rtpoll(u->source, u->rtpoll);
diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 3b1975fa..bd633cf0 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -42,10 +42,11 @@
 
 #include "internal.h"
 
-#define LATENCY_IPOL_INTERVAL_USEC (500*PA_USEC_PER_MSEC)
+#define LATENCY_IPOL_INTERVAL_USEC (333*PA_USEC_PER_MSEC)
 
 #define SMOOTHER_ADJUST_TIME (1000*PA_USEC_PER_MSEC)
 #define SMOOTHER_HISTORY_TIME (5000*PA_USEC_PER_MSEC)
+#define SMOOTHER_MIN_HISTORY (4)
 
 pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
     return pa_stream_new_with_proplist(c, name, ss, map, NULL);
@@ -344,6 +345,7 @@ void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED u
     pa_bool_t suspended;
     uint32_t di;
     pa_usec_t usec;
+    uint32_t maxlength = 0, fragsize = 0, minreq = 0, tlength = 0, prebuf = 0;
 
     pa_assert(pd);
     pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_MOVED || command == PA_COMMAND_RECORD_STREAM_MOVED);
@@ -367,14 +369,28 @@ void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED u
     }
 
     if (c->version >= 13) {
-        if (pa_tagstruct_get_usec(t, &usec) < 0) {
-            pa_context_fail(s->context, PA_ERR_PROTOCOL);
-            goto finish;
+
+        if (s->direction == PA_STREAM_RECORD) {
+            if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
+                pa_tagstruct_getu32(t, &fragsize) < 0 ||
+                pa_tagstruct_get_usec(t, &usec) < 0) {
+                pa_context_fail(c, PA_ERR_PROTOCOL);
+                goto finish;
+            }
+        } else {
+            if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
+                pa_tagstruct_getu32(t, &tlength) < 0 ||
+                pa_tagstruct_getu32(t, &prebuf) < 0 ||
+                pa_tagstruct_getu32(t, &minreq) < 0 ||
+                pa_tagstruct_get_usec(t, &usec) < 0) {
+                pa_context_fail(c, PA_ERR_PROTOCOL);
+                goto finish;
+            }
         }
     }
 
     if (!pa_tagstruct_eof(t)) {
-        pa_context_fail(s->context, PA_ERR_PROTOCOL);
+        pa_context_fail(c, PA_ERR_PROTOCOL);
         goto finish;
     }
 
@@ -394,6 +410,12 @@ void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED u
             s->timing_info.configured_source_usec = usec;
         else
             s->timing_info.configured_sink_usec = usec;
+
+        s->buffer_attr.maxlength = maxlength;
+        s->buffer_attr.fragsize = fragsize;
+        s->buffer_attr.tlength = tlength;
+        s->buffer_attr.prebuf = prebuf;
+        s->buffer_attr.minreq = minreq;
     }
 
     pa_xfree(s->device_name);
@@ -861,7 +883,7 @@ static int create_stream(
         if (s->smoother)
             pa_smoother_free(s->smoother);
 
-        s->smoother = pa_smoother_new(SMOOTHER_ADJUST_TIME, SMOOTHER_HISTORY_TIME, !(flags & PA_STREAM_NOT_MONOTONOUS));
+        s->smoother = pa_smoother_new(SMOOTHER_ADJUST_TIME, SMOOTHER_HISTORY_TIME, !(flags & PA_STREAM_NOT_MONOTONOUS), SMOOTHER_MIN_HISTORY);
 
         x = pa_rtclock_usec();
         pa_smoother_set_time_offset(s->smoother, x);
diff --git a/src/pulsecore/cli-command.c b/src/pulsecore/cli-command.c
index b6194f84..925a2e8c 100644
--- a/src/pulsecore/cli-command.c
+++ b/src/pulsecore/cli-command.c
@@ -1055,7 +1055,7 @@ static int pa_cli_command_move_sink_input(pa_core *c, pa_tokenizer *t, pa_strbuf
         return -1;
     }
 
-    if (pa_sink_input_move_to(si, sink, 0) < 0) {
+    if (pa_sink_input_move_to(si, sink) < 0) {
         pa_strbuf_puts(buf, "Moved failed.\n");
         return -1;
     }
diff --git a/src/pulsecore/cli-text.c b/src/pulsecore/cli-text.c
index f251a146..029a7089 100644
--- a/src/pulsecore/cli-text.c
+++ b/src/pulsecore/cli-text.c
@@ -253,7 +253,13 @@ char *pa_source_output_list_to_string(pa_core *c) {
     pa_strbuf_printf(s, "%u source outputs(s) available.\n", pa_idxset_size(c->source_outputs));
 
     for (o = pa_idxset_first(c->source_outputs, &idx); o; o = pa_idxset_next(c->source_outputs, &idx)) {
-        char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t;
+        char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t, clt[28];
+        pa_usec_t cl;
+
+        if ((cl = pa_source_output_get_requested_latency(o)) == (pa_usec_t) -1)
+            pa_snprintf(clt, sizeof(clt), "n/a");
+        else
+            pa_snprintf(clt, sizeof(clt), "%0.2f ms", (double) cl / PA_USEC_PER_MSEC);
 
         pa_assert(o->source);
 
@@ -264,7 +270,8 @@ char *pa_source_output_list_to_string(pa_core *c) {
             "\tflags: %s%s%s%s%s%s%s%s\n"
             "\tstate: %s\n"
             "\tsource: %u <%s>\n"
-            "\tlatency: %0.2f ms\n"
+            "\tcurrent latency: %0.2f ms\n"
+            "\trequested latency: %s\n"
             "\tsample spec: %s\n"
             "\tchannel map: %s\n"
             "\tresample method: %s\n",
@@ -281,6 +288,7 @@ char *pa_source_output_list_to_string(pa_core *c) {
             state_table[pa_source_output_get_state(o)],
             o->source->index, o->source->name,
             (double) pa_source_output_get_latency(o) / PA_USEC_PER_MSEC,
+            clt,
             pa_sample_spec_snprint(ss, sizeof(ss), &o->sample_spec),
             pa_channel_map_snprint(cm, sizeof(cm), &o->channel_map),
             pa_resample_method_to_string(pa_source_output_get_resample_method(o)));
@@ -315,7 +323,13 @@ char *pa_sink_input_list_to_string(pa_core *c) {
     pa_strbuf_printf(s, "%u sink input(s) available.\n", pa_idxset_size(c->sink_inputs));
 
     for (i = pa_idxset_first(c->sink_inputs, &idx); i; i = pa_idxset_next(c->sink_inputs, &idx)) {
-        char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cv[PA_CVOLUME_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t;
+        char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cv[PA_CVOLUME_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t, clt[28];
+        pa_usec_t cl;
+
+        if ((cl = pa_sink_input_get_requested_latency(i)) == (pa_usec_t) -1)
+            pa_snprintf(clt, sizeof(clt), "n/a");
+        else
+            pa_snprintf(clt, sizeof(clt), "%0.2f ms", (double) cl / PA_USEC_PER_MSEC);
 
         pa_assert(i->sink);
 
@@ -328,7 +342,8 @@ char *pa_sink_input_list_to_string(pa_core *c) {
             "\tsink: %u <%s>\n"
             "\tvolume: %s\n"
             "\tmuted: %s\n"
-            "\tlatency: %0.2f ms\n"
+            "\tcurrent latency: %0.2f ms\n"
+            "\trequested latency: %s\n"
             "\tsample spec: %s\n"
             "\tchannel map: %s\n"
             "\tresample method: %s\n",
@@ -347,6 +362,7 @@ char *pa_sink_input_list_to_string(pa_core *c) {
             pa_cvolume_snprint(cv, sizeof(cv), pa_sink_input_get_volume(i)),
             pa_yes_no(pa_sink_input_get_mute(i)),
             (double) pa_sink_input_get_latency(i) / PA_USEC_PER_MSEC,
+            clt,
             pa_sample_spec_snprint(ss, sizeof(ss), &i->sample_spec),
             pa_channel_map_snprint(cm, sizeof(cm), &i->channel_map),
             pa_resample_method_to_string(pa_sink_input_get_resample_method(i)));
diff --git a/src/pulsecore/protocol-native.c b/src/pulsecore/protocol-native.c
index 7eeefb84..a824b4da 100644
--- a/src/pulsecore/protocol-native.c
+++ b/src/pulsecore/protocol-native.c
@@ -818,6 +818,8 @@ static void fix_playback_buffer_attr_post(playback_stream *s, uint32_t *maxlengt
     *tlength = (uint32_t) pa_memblockq_get_tlength(s->memblockq);
     *prebuf = (uint32_t) pa_memblockq_get_prebuf(s->memblockq);
     *minreq = (uint32_t) pa_memblockq_get_minreq(s->memblockq);
+
+    s->minreq = *minreq;
 }
 
 static playback_stream* playback_stream_new(
@@ -933,7 +935,6 @@ static playback_stream* playback_stream_new(
     *ss = s->sink_input->sample_spec;
     *map = s->sink_input->channel_map;
 
-    s->minreq = *minreq;
     pa_atomic_store(&s->missing, 0);
     s->drain_request = FALSE;
 
@@ -1290,14 +1291,14 @@ static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk
         } else if (i->thread_info.playing_for > 0)
             pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_UNDERFLOW, NULL, 0, NULL, NULL);
 
-/*         pa_log("adding %llu bytes, total is %llu", (unsigned long long) nbytes, (unsigned long long) i->thread_info.underrun_for); */
+/*         pa_log("adding %llu bytes", (unsigned long long) nbytes); */
 
         request_bytes(s);
 
         return -1;
     }
 
-/*     pa_log("NOTUNDERRUN"); */
+/*     pa_log("NOTUNDERRUN %lu", (unsigned long) chunk->length); */
 
     if (i->thread_info.underrun_for > 0)
         pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_STARTED, NULL, 0, NULL, NULL);
@@ -1368,11 +1369,24 @@ static void sink_input_suspend_cb(pa_sink_input *i, pa_bool_t suspend) {
 static void sink_input_moved_cb(pa_sink_input *i) {
     playback_stream *s;
     pa_tagstruct *t;
+    uint32_t maxlength, tlength, prebuf, minreq;
 
     pa_sink_input_assert_ref(i);
     s = PLAYBACK_STREAM(i->userdata);
     playback_stream_assert_ref(s);
 
+    maxlength = (uint32_t) pa_memblockq_get_maxlength(s->memblockq);
+    tlength = (uint32_t) pa_memblockq_get_tlength(s->memblockq);
+    prebuf = (uint32_t) pa_memblockq_get_prebuf(s->memblockq);
+    minreq = (uint32_t) pa_memblockq_get_minreq(s->memblockq);
+
+    fix_playback_buffer_attr_pre(s, TRUE, &maxlength, &tlength, &prebuf, &minreq);
+    pa_memblockq_set_maxlength(s->memblockq, maxlength);
+    pa_memblockq_set_tlength(s->memblockq, tlength);
+    pa_memblockq_set_prebuf(s->memblockq, prebuf);
+    pa_memblockq_set_minreq(s->memblockq, minreq);
+    fix_playback_buffer_attr_post(s, &maxlength, &tlength, &prebuf, &minreq);
+
     if (s->connection->version < 12)
       return;
 
@@ -1383,6 +1397,15 @@ static void sink_input_moved_cb(pa_sink_input *i) {
     pa_tagstruct_putu32(t, i->sink->index);
     pa_tagstruct_puts(t, i->sink->name);
     pa_tagstruct_put_boolean(t, pa_sink_get_state(i->sink) == PA_SINK_SUSPENDED);
+
+    if (s->connection->version >= 13) {
+        pa_tagstruct_putu32(t, maxlength);
+        pa_tagstruct_putu32(t, tlength);
+        pa_tagstruct_putu32(t, prebuf);
+        pa_tagstruct_putu32(t, minreq);
+        pa_tagstruct_put_usec(t, s->sink_latency);
+    }
+
     pa_pstream_send_tagstruct(s->connection->pstream, t);
 }
 
@@ -1447,11 +1470,19 @@ static void source_output_suspend_cb(pa_source_output *o, pa_bool_t suspend) {
 static void source_output_moved_cb(pa_source_output *o) {
     record_stream *s;
     pa_tagstruct *t;
+    uint32_t maxlength, fragsize;
 
     pa_source_output_assert_ref(o);
     s = RECORD_STREAM(o->userdata);
     record_stream_assert_ref(s);
 
+    fragsize = (uint32_t) s->fragment_size;
+    maxlength = (uint32_t) pa_memblockq_get_length(s->memblockq);
+
+    fix_record_buffer_attr_pre(s, TRUE, &maxlength, &fragsize);
+    pa_memblockq_set_maxlength(s->memblockq, maxlength);
+    fix_record_buffer_attr_post(s, &maxlength, &fragsize);
+
     if (s->connection->version < 12)
       return;
 
@@ -1462,6 +1493,13 @@ static void source_output_moved_cb(pa_source_output *o) {
     pa_tagstruct_putu32(t, o->source->index);
     pa_tagstruct_puts(t, o->source->name);
     pa_tagstruct_put_boolean(t, pa_source_get_state(o->source) == PA_SOURCE_SUSPENDED);
+
+    if (s->connection->version >= 13) {
+        pa_tagstruct_putu32(t, maxlength);
+        pa_tagstruct_putu32(t, fragsize);
+        pa_tagstruct_put_usec(t, s->source_latency);
+    }
+
     pa_pstream_send_tagstruct(s->connection->pstream, t);
 }
 
@@ -1900,27 +1938,28 @@ static void command_auth(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t
     pa_proplist_sets(c->client->proplist, "native-protocol.version", tmp);
 
     if (!c->authorized) {
-        int success = 0;
+        pa_bool_t success = FALSE;
 
 #ifdef HAVE_CREDS
         const pa_creds *creds;
 
         if ((creds = pa_pdispatch_creds(pd))) {
             if (creds->uid == getuid())
-                success = 1;
+                success = TRUE;
             else if (c->protocol->auth_group) {
                 int r;
                 gid_t gid;
 
                 if ((gid = pa_get_gid_of_group(c->protocol->auth_group)) == (gid_t) -1)
-                    pa_log_warn("failed to get GID of group '%s'", c->protocol->auth_group);
+                    pa_log_warn("Failed to get GID of group '%s'", c->protocol->auth_group);
                 else if (gid == creds->gid)
-                    success = 1;
+                    success = TRUE;
+
                 if (!success) {
                     if ((r = pa_uid_in_group(creds->uid, c->protocol->auth_group)) < 0)
-                        pa_log_warn("failed to check group membership.");
+                        pa_log_warn("Failed to check group membership.");
                     else if (r > 0)
-                        success = 1;
+                        success = TRUE;
                 }
             }
 
@@ -1941,7 +1980,7 @@ static void command_auth(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t
 #endif
 
         if (!success && memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) == 0)
-            success = 1;
+            success = TRUE;
 
         if (!success) {
             pa_log_warn("Denied access to client with invalid authorization data.");
@@ -3037,6 +3076,9 @@ static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, u
         pa_tagstruct_putu32(reply, prebuf);
         pa_tagstruct_putu32(reply, minreq);
 
+        if (c->version >= 13)
+            pa_tagstruct_put_usec(reply, s->sink_latency);
+
     } else {
         record_stream *s;
         pa_bool_t adjust_latency = FALSE;
@@ -3063,6 +3105,9 @@ static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, u
         reply = reply_new(tag);
         pa_tagstruct_putu32(reply, maxlength);
         pa_tagstruct_putu32(reply, fragsize);
+
+        if (c->version >= 13)
+            pa_tagstruct_put_usec(reply, s->source_latency);
     }
 
     pa_pstream_send_tagstruct(c->pstream, reply);
@@ -3600,7 +3645,7 @@ static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag
 
         CHECK_VALIDITY(c->pstream, si && sink, tag, PA_ERR_NOENTITY);
 
-        if (pa_sink_input_move_to(si, sink, 0) < 0) {
+        if (pa_sink_input_move_to(si, sink) < 0) {
             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
             return;
         }
diff --git a/src/pulsecore/sink-input.c b/src/pulsecore/sink-input.c
index 0be1cc97..d51ff810 100644
--- a/src/pulsecore/sink-input.c
+++ b/src/pulsecore/sink-input.c
@@ -243,7 +243,6 @@ pa_sink_input* pa_sink_input_new(
     i->thread_info.state = i->state;
     i->thread_info.attached = FALSE;
     pa_atomic_store(&i->thread_info.drained, 1);
-    pa_atomic_store(&i->thread_info.render_memblockq_is_empty, 0);
     i->thread_info.sample_spec = i->sample_spec;
     i->thread_info.resampler = resampler;
     i->thread_info.volume = i->volume;
@@ -410,10 +409,6 @@ void pa_sink_input_put(pa_sink_input *i) {
 
     pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_NEW, i->index);
     pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT], i);
-
-    /* Please note that if you change something here, you have to
-       change something in pa_sink_input_move() with the ghost stream
-       registration too. */
 }
 
 void pa_sink_input_kill(pa_sink_input*i) {
@@ -511,10 +506,10 @@ int pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink frames */, pa
              * data, so let's just hand out silence */
             pa_atomic_store(&i->thread_info.drained, 1);
 
-            pa_memblockq_seek(i->thread_info.render_memblockq, slength, PA_SEEK_RELATIVE_ON_READ);
+            pa_memblockq_seek(i->thread_info.render_memblockq, slength, PA_SEEK_RELATIVE);
             i->thread_info.playing_for = 0;
             if (i->thread_info.underrun_for != (uint64_t) -1)
-                i->thread_info.underrun_for += slength;
+                i->thread_info.underrun_for += ilength;
             break;
         }
 
@@ -551,6 +546,8 @@ int pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink frames */, pa
                 pa_memchunk rchunk;
                 pa_resampler_run(i->thread_info.resampler, &wchunk, &rchunk);
 
+/*                 pa_log_debug("pushing %lu", (unsigned long) rchunk.length); */
+
                 if (rchunk.memblock) {
                     pa_memblockq_push_align(i->thread_info.render_memblockq, &rchunk);
                     pa_memblock_unref(rchunk.memblock);
@@ -571,6 +568,8 @@ int pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink frames */, pa
     pa_assert(chunk->length > 0);
     pa_assert(chunk->memblock);
 
+/*     pa_log_debug("peeking %lu", (unsigned long) chunk->length); */
+
     if (chunk->length > block_size_max_sink)
         chunk->length = block_size_max_sink;
 
@@ -586,8 +585,6 @@ int pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink frames */, pa
     else
         *volume = i->thread_info.volume;
 
-    pa_atomic_store(&i->thread_info.render_memblockq_is_empty, pa_memblockq_is_empty(i->thread_info.render_memblockq));
-
     return 0;
 }
 
@@ -599,6 +596,8 @@ void pa_sink_input_drop(pa_sink_input *i, size_t nbytes /* in sink sample spec *
     pa_assert(pa_frame_aligned(nbytes, &i->sink->sample_spec));
     pa_assert(nbytes > 0);
 
+/*     pa_log_debug("dropping %lu", (unsigned long) nbytes); */
+
     /* If there's still some rewrite request the handle, but the sink
     didn't do this for us, we do it here. However, since the sink
     apparently doesn't support rewinding, we pass 0 here. This still
@@ -606,12 +605,11 @@ void pa_sink_input_drop(pa_sink_input *i, size_t nbytes /* in sink sample spec *
     pa_sink_input_process_rewind(i, 0);
 
     pa_memblockq_drop(i->thread_info.render_memblockq, nbytes);
-
-    pa_atomic_store(&i->thread_info.render_memblockq_is_empty, pa_memblockq_is_empty(i->thread_info.render_memblockq));
 }
 
 /* Called from thread context */
 void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sample spec */) {
+    size_t lbq;
     pa_sink_input_assert_ref(i);
 
     pa_assert(PA_SINK_INPUT_IS_LINKED(i->thread_info.state));
@@ -619,6 +617,8 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam
 
 /*     pa_log_debug("rewind(%lu, %lu)", (unsigned long) nbytes, (unsigned long) i->thread_info.rewrite_nbytes); */
 
+    lbq = pa_memblockq_get_length(i->thread_info.render_memblockq);
+
     if (nbytes > 0) {
         pa_log_debug("Have to rewind %lu bytes on render memblockq.", (unsigned long) nbytes);
         pa_memblockq_rewind(i->thread_info.render_memblockq, nbytes);
@@ -635,7 +635,7 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam
         size_t max_rewrite, amount;
 
         /* Calculate how much make sense to rewrite at most */
-        max_rewrite = nbytes + pa_memblockq_get_length(i->thread_info.render_memblockq);
+        max_rewrite = nbytes + lbq;
 
         /* Transform into local domain */
         if (i->thread_info.resampler)
@@ -651,18 +651,16 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam
             if (i->process_rewind)
                 i->process_rewind(i, amount);
 
-            if (i->thread_info.rewrite_flush)
-                pa_memblockq_silence(i->thread_info.render_memblockq);
-            else {
+            /* Convert back to to sink domain */
+            if (i->thread_info.resampler)
+                amount = pa_resampler_result(i->thread_info.resampler, amount);
 
-                /* Convert back to to sink domain */
-                if (i->thread_info.resampler)
-                    amount = pa_resampler_result(i->thread_info.resampler, amount);
+            if (amount > 0)
+                /* Ok, now update the write pointer */
+                pa_memblockq_seek(i->thread_info.render_memblockq, - ((int64_t) amount), PA_SEEK_RELATIVE);
 
-                if (amount > 0)
-                    /* Ok, now update the write pointer */
-                    pa_memblockq_seek(i->thread_info.render_memblockq, - ((int64_t) amount), PA_SEEK_RELATIVE);
-            }
+            if (i->thread_info.rewrite_flush)
+                pa_memblockq_silence(i->thread_info.render_memblockq);
 
             /* And reset the resampler */
             if (i->thread_info.resampler)
@@ -702,6 +700,7 @@ static pa_usec_t fixup_latency(pa_sink *s, pa_usec_t usec) {
 }
 
 pa_usec_t pa_sink_input_set_requested_latency_within_thread(pa_sink_input *i, pa_usec_t usec) {
+    pa_sink_input_assert_ref(i);
 
     usec = fixup_latency(i->sink, usec);
 
@@ -728,6 +727,21 @@ pa_usec_t pa_sink_input_set_requested_latency(pa_sink_input *i, pa_usec_t usec)
     return usec;
 }
 
+pa_usec_t pa_sink_input_get_requested_latency(pa_sink_input *i) {
+    pa_usec_t usec = 0;
+
+    pa_sink_input_assert_ref(i);
+
+    if (PA_SINK_INPUT_IS_LINKED(i->state))
+        pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL);
+    else
+        /* If this sink input is not realized yet, we have to touch
+         * the thread info data directly */
+        usec = i->thread_info.requested_sink_latency;
+
+    return usec;
+}
+
 void pa_sink_input_set_volume(pa_sink_input *i, const pa_cvolume *volume) {
     pa_sink_input_assert_ref(i);
     pa_assert(PA_SINK_INPUT_IS_LINKED(i->state));
@@ -821,11 +835,9 @@ pa_resample_method_t pa_sink_input_get_resample_method(pa_sink_input *i) {
     return i->resample_method;
 }
 
-int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately) {
+int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest) {
     pa_resampler *new_resampler;
     pa_sink *origin;
-    pa_usec_t silence_usec = 0;
-    pa_sink_input_move_info info;
     pa_sink_input_move_hook_data hook_data;
 
     pa_sink_input_assert_ref(i);
@@ -881,80 +893,7 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately
     hook_data.destination = dest;
     pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_MOVE], &hook_data);
 
-    memset(&info, 0, sizeof(info));
-    info.sink_input = i;
-
-    if (!immediately) {
-        pa_usec_t old_latency, new_latency;
-
-        /* Let's do a little bit of Voodoo for compensating latency
-         * differences. We assume that the accuracy for our
-         * estimations is still good enough, even though we do these
-         * operations non-atomic. */
-
-        old_latency = pa_sink_get_latency(origin);
-        new_latency = pa_sink_get_latency(dest);
-
-        /* The already resampled data should go to the old sink */
-
-        if (old_latency >= new_latency) {
-
-            /* The latency of the old sink is larger than the latency
-             * of the new sink. Therefore to compensate for the
-             * difference we to play silence on the new one for a
-             * while */
-
-            silence_usec = old_latency - new_latency;
-
-        } else {
-
-            /* The latency of new sink is larger than the latency of
-             * the old sink. Therefore we have to precompute a little
-             * and make sure that this is still played on the old
-             * sink, until we can play the first sample on the new
-             * sink.*/
-
-            info.buffer_bytes = pa_usec_to_bytes(new_latency - old_latency, &origin->sample_spec);
-        }
-
-        /* Okey, let's move it */
-
-        if (info.buffer_bytes > 0) {
-            pa_proplist *p;
-
-            p = pa_proplist_new();
-            pa_proplist_sets(p, PA_PROP_MEDIA_NAME, "Ghost For Moved Stream");
-            pa_proplist_sets(p, PA_PROP_MEDIA_ROLE, "routing");
-
-            info.ghost_sink_input = pa_memblockq_sink_input_new(
-                    origin,
-                    &origin->sample_spec,
-                    &origin->channel_map,
-                    NULL,
-                    NULL,
-                    p);
-
-            pa_proplist_free(p);
-
-            if (info.ghost_sink_input) {
-                info.ghost_sink_input->thread_info.state = info.ghost_sink_input->state = PA_SINK_INPUT_RUNNING;
-                info.ghost_sink_input->thread_info.volume = info.ghost_sink_input->volume;
-                info.ghost_sink_input->thread_info.muted = info.ghost_sink_input->muted;
-
-                info.buffer = pa_memblockq_new(0, MOVE_BUFFER_LENGTH, 0, pa_frame_size(&origin->sample_spec), 0, 0, 0, NULL);
-            }
-        }
-    }
-
-    pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER, &info, 0, NULL);
-
-    if (info.ghost_sink_input) {
-        /* Basically, do what pa_sink_input_put() does ...*/
-
-        pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_NEW, info.ghost_sink_input->index);
-        pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT], info.ghost_sink_input);
-        pa_sink_input_unref(info.ghost_sink_input);
-    }
+    pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_START_MOVE, i, 0, NULL);
 
     pa_idxset_remove_by_data(origin->inputs, i, NULL);
     pa_idxset_put(dest->inputs, i, NULL);
@@ -965,40 +904,31 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately
         dest->n_corked++;
     }
 
-    /* Replace resampler */
+    /* Replace resampler and render queue */
     if (new_resampler != i->thread_info.resampler) {
-        pa_memchunk silence;
 
         if (i->thread_info.resampler)
             pa_resampler_free(i->thread_info.resampler);
         i->thread_info.resampler = new_resampler;
 
-        /* if the resampler changed, the silence memblock is
-         * probably invalid now, too */
-
-        pa_silence_memchunk_get(
-                &i->sink->core->silence_cache,
-                i->sink->core->mempool,
-                &silence,
-                &dest->sample_spec,
-                0);
-
-        pa_memblockq_set_silence(i->thread_info.render_memblockq, &silence);
-        pa_memblock_unref(silence.memblock);
+        pa_memblockq_free(i->thread_info.render_memblockq);
 
+        i->thread_info.render_memblockq = pa_memblockq_new(
+                0,
+                MEMBLOCKQ_MAXLENGTH,
+                0,
+                pa_frame_size(&i->sink->sample_spec),
+                0,
+                1,
+                0,
+                &i->sink->silence);
     }
 
-    pa_memblockq_flush(i->thread_info.render_memblockq);
-
-    /* Calculate the new sleeping time */
-    if (!immediately)
-        pa_memblockq_seek(i->thread_info.render_memblockq, pa_usec_to_bytes(silence_usec, &dest->sample_spec), PA_SEEK_RELATIVE);
-
-    pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_ADD_INPUT, i, 0, NULL);
-
     pa_sink_update_status(origin);
     pa_sink_update_status(dest);
 
+    pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_FINISH_MOVE, i, 0, NULL);
+
     if (i->moved)
         i->moved(i);
 
@@ -1015,6 +945,9 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately
 void pa_sink_input_set_state_within_thread(pa_sink_input *i, pa_sink_input_state_t state) {
     pa_sink_input_assert_ref(i);
 
+    if (state == i->thread_info.state)
+        return;
+
     if ((state == PA_SINK_INPUT_DRAINED || state == PA_SINK_INPUT_RUNNING) &&
         !(i->thread_info.state == PA_SINK_INPUT_DRAINED || i->thread_info.state != PA_SINK_INPUT_RUNNING))
         pa_atomic_store(&i->thread_info.drained, 1);
@@ -1060,7 +993,6 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t
             pa_usec_t *r = userdata;
 
             *r += pa_bytes_to_usec(pa_memblockq_get_length(i->thread_info.render_memblockq), &i->sink->sample_spec);
-
             return 0;
         }
 
@@ -1089,6 +1021,13 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t
 
             pa_sink_input_set_requested_latency_within_thread(i, (pa_usec_t) offset);
             return 0;
+
+        case PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY: {
+            pa_usec_t *r = userdata;
+
+            *r = i->thread_info.requested_sink_latency;
+            return 0;
+        }
     }
 
     return -1;
@@ -1103,11 +1042,12 @@ pa_sink_input_state_t pa_sink_input_get_state(pa_sink_input *i) {
     return i->state;
 }
 
+/* Called from IO context */
 pa_bool_t pa_sink_input_safe_to_remove(pa_sink_input *i) {
     pa_sink_input_assert_ref(i);
 
-    if (i->state == PA_SINK_INPUT_RUNNING || i->state == PA_SINK_INPUT_DRAINED || i->state == PA_SINK_INPUT_CORKED)
-        return pa_atomic_load(&i->thread_info.render_memblockq_is_empty);
+    if (PA_SINK_INPUT_IS_LINKED(i->thread_info.state))
+        return pa_memblockq_is_empty(i->thread_info.render_memblockq);
 
     return TRUE;
 }
@@ -1129,7 +1069,7 @@ void pa_sink_input_request_rewind(pa_sink_input *i, size_t nbytes  /* in our sam
     pa_assert(i->thread_info.rewrite_nbytes == 0);
 
     /* We don't take rewind requests while we are corked */
-    if (i->state == PA_SINK_INPUT_CORKED)
+    if (i->thread_info.state == PA_SINK_INPUT_CORKED)
         return;
 
     pa_assert(rewrite || flush);
diff --git a/src/pulsecore/sink-input.h b/src/pulsecore/sink-input.h
index 8edd7ecb..5f146122 100644
--- a/src/pulsecore/sink-input.h
+++ b/src/pulsecore/sink-input.h
@@ -144,7 +144,7 @@ struct pa_sink_input {
 
     struct {
         pa_sink_input_state_t state;
-        pa_atomic_t drained, render_memblockq_is_empty;
+        pa_atomic_t drained;
 
         pa_bool_t attached; /* True only between ->attach() and ->detach() calls */
 
@@ -181,6 +181,7 @@ enum {
     PA_SINK_INPUT_MESSAGE_SET_RATE,
     PA_SINK_INPUT_MESSAGE_SET_STATE,
     PA_SINK_INPUT_MESSAGE_SET_REQUESTED_LATENCY,
+    PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY,
     PA_SINK_INPUT_MESSAGE_MAX
 };
 
@@ -243,6 +244,10 @@ could be rewound in the HW device. This functionality is required for
 implementing the "zero latency" write-through functionality. */
 void pa_sink_input_request_rewind(pa_sink_input *i, size_t nbytes, pa_bool_t rewrite, pa_bool_t flush);
 
+void pa_sink_input_cork(pa_sink_input *i, pa_bool_t b);
+
+int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate);
+
 /* Callable by everyone from main thread*/
 
 /* External code may request disconnection with this function */
@@ -255,17 +260,14 @@ const pa_cvolume *pa_sink_input_get_volume(pa_sink_input *i);
 void pa_sink_input_set_mute(pa_sink_input *i, pa_bool_t mute);
 int pa_sink_input_get_mute(pa_sink_input *i);
 
-void pa_sink_input_cork(pa_sink_input *i, pa_bool_t b);
-
-int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate);
-
 pa_resample_method_t pa_sink_input_get_resample_method(pa_sink_input *i);
 
-int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately);
+int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest);
 
 pa_sink_input_state_t pa_sink_input_get_state(pa_sink_input *i);
 
-pa_bool_t pa_sink_input_safe_to_remove(pa_sink_input *i);
+pa_usec_t pa_sink_input_get_requested_latency(pa_sink_input *i);
+
 /* To be used exclusively by the sink driver IO thread */
 
 int pa_sink_input_peek(pa_sink_input *i, size_t length, pa_memchunk *chunk, pa_cvolume *volume);
@@ -279,12 +281,7 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t
 
 pa_usec_t pa_sink_input_set_requested_latency_within_thread(pa_sink_input *i, pa_usec_t usec);
 
-typedef struct pa_sink_input_move_info {
-    pa_sink_input *sink_input;
-    pa_sink_input *ghost_sink_input;
-    pa_memblockq *buffer;
-    size_t buffer_bytes;
-} pa_sink_input_move_info;
+pa_bool_t pa_sink_input_safe_to_remove(pa_sink_input *i);
 
 pa_memchunk* pa_sink_input_get_silence(pa_sink_input *i, pa_memchunk *ret);
 
diff --git a/src/pulsecore/sink.c b/src/pulsecore/sink.c
index d3bacbfd..31c3cfc8 100644
--- a/src/pulsecore/sink.c
+++ b/src/pulsecore/sink.c
@@ -113,7 +113,6 @@ static void reset_callbacks(pa_sink *s) {
     s->set_volume = NULL;
     s->get_mute = NULL;
     s->set_mute = NULL;
-    s->get_latency = NULL;
     s->request_rewind = NULL;
     s->update_requested_latency = NULL;
 }
@@ -769,9 +768,6 @@ pa_usec_t pa_sink_get_latency(pa_sink *s) {
     if (!PA_SINK_IS_OPENED(s->state))
         return 0;
 
-    if (s->get_latency)
-        return s->get_latency(s);
-
     if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
         return 0;
 
@@ -930,6 +926,10 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
         case PA_SINK_MESSAGE_ADD_INPUT: {
             pa_sink_input *i = PA_SINK_INPUT(userdata);
 
+            /* If you change anything here, make sure to change the
+             * sink input handling a few lines down at
+             * PA_SINK_MESSAGE_FINISH_MOVE, too. */
+
             pa_hashmap_put(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index), pa_sink_input_ref(i));
 
             /* Since the caller sleeps in pa_sink_input_put(), we can
@@ -965,10 +965,6 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
              * slow start, i.e. need some time to buffer client
              * samples before beginning streaming. */
 
-            /* If you change anything here, make sure to change the
-             * ghost sink input handling a few lines down at
-             * PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER, too. */
-
             return 0;
         }
 
@@ -977,7 +973,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
 
             /* If you change anything here, make sure to change the
              * sink input handling a few lines down at
-             * PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER, too. */
+             * PA_SINK_MESSAGE_PREPAPRE_MOVE, too. */
 
             pa_sink_input_set_state_within_thread(i, i->state);
 
@@ -1013,85 +1009,88 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
             return 0;
         }
 
-        case PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER: {
-            pa_sink_input_move_info *info = userdata;
-            int volume_is_norm;
+        case PA_SINK_MESSAGE_START_MOVE: {
+            pa_sink_input *i = PA_SINK_INPUT(userdata);
 
             /* We don't support moving synchronized streams. */
-            pa_assert(!info->sink_input->sync_prev);
-            pa_assert(!info->sink_input->sync_next);
-            pa_assert(!info->sink_input->thread_info.sync_next);
-            pa_assert(!info->sink_input->thread_info.sync_prev);
+            pa_assert(!i->sync_prev);
+            pa_assert(!i->sync_next);
+            pa_assert(!i->thread_info.sync_next);
+            pa_assert(!i->thread_info.sync_prev);
 
-            if (info->sink_input->detach)
-                info->sink_input->detach(info->sink_input);
+            if (i->thread_info.state != PA_SINK_INPUT_CORKED) {
+                pa_usec_t usec = 0;
+                size_t sink_nbytes, total_nbytes;
 
-            pa_assert(info->sink_input->thread_info.attached);
-            info->sink_input->thread_info.attached = FALSE;
-            pa_sink_invalidate_requested_latency(info->sink_input->sink);
+                /* Get the latency of the sink */
+                if (PA_MSGOBJECT(s)->process_msg(PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
+                    usec = 0;
 
-            if (info->ghost_sink_input) {
-                pa_assert(info->buffer_bytes > 0);
-                pa_assert(info->buffer);
+                sink_nbytes = pa_usec_to_bytes(usec, &s->sample_spec);
+                total_nbytes = sink_nbytes + pa_memblockq_get_length(i->thread_info.render_memblockq);
 
-                volume_is_norm = pa_cvolume_is_norm(&info->sink_input->thread_info.volume);
+                if (total_nbytes > 0) {
+                    i->thread_info.rewrite_nbytes = i->thread_info.resampler ? pa_resampler_request(i->thread_info.resampler, total_nbytes) : total_nbytes;
+                    i->thread_info.rewrite_flush = TRUE;
+                    pa_sink_input_process_rewind(i, sink_nbytes);
+                }
+            }
 
-                pa_log_debug("Buffering %lu bytes ...", (unsigned long) info->buffer_bytes);
+            if (i->detach)
+                i->detach(i);
 
-                while (info->buffer_bytes > 0) {
-                    pa_memchunk memchunk;
-                    pa_cvolume volume;
-                    size_t n;
+            pa_assert(i->thread_info.attached);
+            i->thread_info.attached = FALSE;
 
-                    if (pa_sink_input_peek(info->sink_input, info->buffer_bytes, &memchunk, &volume) < 0)
-                        break;
+            /* Let's remove the sink input ...*/
+            if (pa_hashmap_remove(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index)))
+                pa_sink_input_unref(i);
 
-                    n = memchunk.length > info->buffer_bytes ? info->buffer_bytes : memchunk.length;
-                    pa_sink_input_drop(info->sink_input, n);
-                    memchunk.length = n;
+            pa_sink_invalidate_requested_latency(s);
 
-                    if (!volume_is_norm) {
-                        pa_memchunk_make_writable(&memchunk, 0);
-                        pa_volume_memchunk(&memchunk, &s->sample_spec, &volume);
-                    }
+            pa_log_debug("Requesting rewind due to started move");
+            pa_sink_request_rewind(s, 0);
 
-                    if (pa_memblockq_push(info->buffer, &memchunk) < 0) {
-                        pa_memblock_unref(memchunk.memblock);
-                        break;
-                    }
+            return 0;
+        }
 
-                    pa_memblock_unref(memchunk.memblock);
-                    info->buffer_bytes -= n;
-                }
+        case PA_SINK_MESSAGE_FINISH_MOVE: {
+            pa_sink_input *i = PA_SINK_INPUT(userdata);
 
-                /* Add the remaining already resampled chunks to the buffer */
-                pa_memblockq_splice(info->buffer, info->sink_input->thread_info.render_memblockq);
+            /* We don't support moving synchronized streams. */
+            pa_assert(!i->sync_prev);
+            pa_assert(!i->sync_next);
+            pa_assert(!i->thread_info.sync_next);
+            pa_assert(!i->thread_info.sync_prev);
 
-                pa_memblockq_sink_input_set_queue(info->ghost_sink_input, info->buffer);
+            pa_hashmap_put(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index), pa_sink_input_ref(i));
 
-                pa_log_debug("Buffered %lu bytes ...", (unsigned long) pa_memblockq_get_length(info->buffer));
-            }
+            pa_assert(!i->thread_info.attached);
+            i->thread_info.attached = TRUE;
 
-            /* Let's remove the sink input ...*/
-            if (pa_hashmap_remove(s->thread_info.inputs, PA_UINT32_TO_PTR(info->sink_input->index)))
-                pa_sink_input_unref(info->sink_input);
+            if (i->attach)
+                i->attach(i);
 
-            /* .. and add the ghost sink input instead */
-            if (info->ghost_sink_input) {
-                pa_hashmap_put(s->thread_info.inputs, PA_UINT32_TO_PTR(info->ghost_sink_input->index), pa_sink_input_ref(info->ghost_sink_input));
-                info->ghost_sink_input->thread_info.sync_prev = info->ghost_sink_input->thread_info.sync_next = NULL;
+            pa_sink_input_update_max_rewind(i, s->thread_info.max_rewind);
 
-                pa_sink_input_update_max_rewind(info->ghost_sink_input, s->thread_info.max_rewind);
+            pa_sink_input_set_requested_latency_within_thread(i, i->thread_info.requested_sink_latency);
 
-                pa_assert(!info->ghost_sink_input->thread_info.attached);
-                info->ghost_sink_input->thread_info.attached = TRUE;
+            if (i->thread_info.state != PA_SINK_INPUT_CORKED) {
+                pa_usec_t usec = 0;
+                size_t nbytes;
 
-                if (info->ghost_sink_input->attach)
-                    info->ghost_sink_input->attach(info->ghost_sink_input);
-            }
+                /* Get the latency of the sink */
+                if (PA_MSGOBJECT(s)->process_msg(PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
+                    usec = 0;
 
-            pa_sink_invalidate_requested_latency(s);
-            pa_sink_request_rewind(s, 0);
+                nbytes = pa_usec_to_bytes(usec, &s->sample_spec);
+
+                if (nbytes > 0)
+                    pa_sink_input_drop(i, nbytes);
+
+                pa_log_debug("Requesting rewind due to finished move");
+                pa_sink_request_rewind(s, nbytes);
+            }
 
             return 0;
         }
diff --git a/src/pulsecore/sink.h b/src/pulsecore/sink.h
index f25f48cf..f297c8f1 100644
--- a/src/pulsecore/sink.h
+++ b/src/pulsecore/sink.h
@@ -118,11 +118,6 @@ struct pa_sink {
      * message will be sent to the IO thread instead. */
     int (*set_mute)(pa_sink *s);               /* dito */
 
-    /* Called when the latency is queried. Called from main loop
-    context. If this is NULL a PA_SINK_MESSAGE_GET_LATENCY message
-    will be sent to the IO thread instead. */
-    pa_usec_t (*get_latency)(pa_sink *s); /* dito */
-
     /* Called when a rewind request is issued. Called from IO thread
      * context. */
     void (*request_rewind)(pa_sink *s);        /* dito */
@@ -166,7 +161,8 @@ typedef enum pa_sink_message {
     PA_SINK_MESSAGE_GET_LATENCY,
     PA_SINK_MESSAGE_GET_REQUESTED_LATENCY,
     PA_SINK_MESSAGE_SET_STATE,
-    PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER,
+    PA_SINK_MESSAGE_START_MOVE,
+    PA_SINK_MESSAGE_FINISH_MOVE,
     PA_SINK_MESSAGE_ATTACH,
     PA_SINK_MESSAGE_DETACH,
     PA_SINK_MESSAGE_MAX
diff --git a/src/pulsecore/source-output.c b/src/pulsecore/source-output.c
index 836e30ed..3940d768 100644
--- a/src/pulsecore/source-output.c
+++ b/src/pulsecore/source-output.c
@@ -294,6 +294,7 @@ void pa_source_output_unlink(pa_source_output*o) {
 static void source_output_free(pa_object* mo) {
     pa_source_output *o = PA_SOURCE_OUTPUT(mo);
 
+    pa_assert(o);
     pa_assert(pa_source_output_refcnt(o) == 0);
 
     if (PA_SOURCE_OUTPUT_IS_LINKED(o->state))
@@ -326,7 +327,7 @@ void pa_source_output_put(pa_source_output *o) {
     state = o->flags & PA_SOURCE_OUTPUT_START_CORKED ? PA_SOURCE_OUTPUT_CORKED : PA_SOURCE_OUTPUT_RUNNING;
 
     update_n_corked(o, state);
-    o->thread_info.state = o->state = state;
+    o->state = state;
 
     pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, o, 0, NULL);
 
@@ -470,6 +471,7 @@ static pa_usec_t fixup_latency(pa_source *s, pa_usec_t usec) {
 }
 
 pa_usec_t pa_source_output_set_requested_latency_within_thread(pa_source_output *o, pa_usec_t usec) {
+    pa_source_output_assert_ref(o);
 
     usec = fixup_latency(o->source, usec);
 
@@ -496,6 +498,21 @@ pa_usec_t pa_source_output_set_requested_latency(pa_source_output *o, pa_usec_t
     return usec;
 }
 
+pa_usec_t pa_source_output_get_requested_latency(pa_source_output *o) {
+    pa_usec_t usec = 0;
+
+    pa_source_output_assert_ref(o);
+
+    if (PA_SOURCE_OUTPUT_IS_LINKED(o->state))
+        pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL);
+    else
+        /* If this sink input is not realized yet, we have to touch
+         * the thread info data directly */
+        usec = o->thread_info.requested_source_latency;
+
+    return usec;
+}
+
 void pa_source_output_cork(pa_source_output *o, pa_bool_t b) {
     pa_source_output_assert_ref(o);
     pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(o->state));
@@ -523,11 +540,11 @@ void pa_source_output_set_name(pa_source_output *o, const char *name) {
     const char *old;
     pa_source_output_assert_ref(o);
 
-    old = pa_proplist_gets(o->proplist, PA_PROP_MEDIA_NAME);
-
-    if (!old && !name)
+    if (!name && !pa_proplist_contains(o->proplist, PA_PROP_MEDIA_NAME))
         return;
 
+    old = pa_proplist_gets(o->proplist, PA_PROP_MEDIA_NAME);
+
     if (old && name && !strcmp(old, name))
         return;
 
@@ -550,7 +567,7 @@ pa_resample_method_t pa_source_output_get_resample_method(pa_source_output *o) {
 
 int pa_source_output_move_to(pa_source_output *o, pa_source *dest) {
     pa_source *origin;
-    pa_resampler *new_resampler = NULL;
+    pa_resampler *new_resampler;
     pa_source_output_move_hook_data hook_data;
 
     pa_source_output_assert_ref(o);
@@ -594,7 +611,8 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) {
             pa_log_warn("Unsupported resampling operation.");
             return -1;
         }
-    }
+    } else
+        new_resampler = NULL;
 
     hook_data.source_output = o;
     hook_data.destination = dest;
@@ -640,6 +658,9 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) {
 void pa_source_output_set_state_within_thread(pa_source_output *o, pa_source_output_state_t state) {
     pa_source_output_assert_ref(o);
 
+    if (state == o->thread_info.state)
+        return;
+
     if (o->state_change)
         o->state_change(o, state);
 
@@ -659,7 +680,6 @@ int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int
             pa_usec_t *r = userdata;
 
             *r += pa_bytes_to_usec(pa_memblockq_get_length(o->thread_info.delay_memblockq), &o->source->sample_spec);
-
             return 0;
         }
 
@@ -678,6 +698,13 @@ int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int
 
             pa_source_output_set_requested_latency_within_thread(o, (pa_usec_t) offset);
             return 0;
+
+        case PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY: {
+            pa_usec_t *r = userdata;
+
+            *r = o->thread_info.requested_source_latency;
+            return 0;
+        }
     }
 
     return -1;
diff --git a/src/pulsecore/source-output.h b/src/pulsecore/source-output.h
index 67cb3761..2dadb5c4 100644
--- a/src/pulsecore/source-output.h
+++ b/src/pulsecore/source-output.h
@@ -148,6 +148,7 @@ enum {
     PA_SOURCE_OUTPUT_MESSAGE_SET_RATE,
     PA_SOURCE_OUTPUT_MESSAGE_SET_STATE,
     PA_SOURCE_OUTPUT_MESSAGE_SET_REQUESTED_LATENCY,
+    PA_SOURCE_OUTPUT_MESSAGE_GET_REQUESTED_LATENCY,
     PA_SOURCE_OUTPUT_MESSAGE_MAX
 };
 
@@ -168,16 +169,16 @@ typedef struct pa_source_output_new_data {
     pa_resample_method_t resample_method;
 } pa_source_output_new_data;
 
-typedef struct pa_source_output_move_hook_data {
-    pa_source_output *source_output;
-    pa_source *destination;
-} pa_source_output_move_hook_data;
-
 pa_source_output_new_data* pa_source_output_new_data_init(pa_source_output_new_data *data);
 void pa_source_output_new_data_set_sample_spec(pa_source_output_new_data *data, const pa_sample_spec *spec);
 void pa_source_output_new_data_set_channel_map(pa_source_output_new_data *data, const pa_channel_map *map);
 void pa_source_output_new_data_done(pa_source_output_new_data *data);
 
+typedef struct pa_source_output_move_hook_data {
+    pa_source_output *source_output;
+    pa_source *destination;
+} pa_source_output_move_hook_data;
+
 /* To be called by the implementing module only */
 
 pa_source_output* pa_source_output_new(
@@ -192,6 +193,10 @@ void pa_source_output_set_name(pa_source_output *i, const char *name);
 
 pa_usec_t pa_source_output_set_requested_latency(pa_source_output *i, pa_usec_t usec);
 
+void pa_source_output_cork(pa_source_output *i, pa_bool_t b);
+
+int pa_source_output_set_rate(pa_source_output *o, uint32_t rate);
+
 /* Callable by everyone */
 
 /* External code may request disconnection with this funcion */
@@ -199,26 +204,24 @@ void pa_source_output_kill(pa_source_output*o);
 
 pa_usec_t pa_source_output_get_latency(pa_source_output *i);
 
-void pa_source_output_cork(pa_source_output *i, pa_bool_t b);
-
-int pa_source_output_set_rate(pa_source_output *o, uint32_t rate);
-
 pa_resample_method_t pa_source_output_get_resample_method(pa_source_output *o);
 
 int pa_source_output_move_to(pa_source_output *o, pa_source *dest);
 
 #define pa_source_output_get_state(o) ((o)->state)
 
+pa_usec_t pa_source_output_get_requested_latency(pa_source_output *o);
+
 /* To be used exclusively by the source driver thread */
 
 void pa_source_output_push(pa_source_output *o, const pa_memchunk *chunk);
 void pa_source_output_process_rewind(pa_source_output *o, size_t nbytes);
 void pa_source_output_update_max_rewind(pa_source_output *o, size_t nbytes);
 
-int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
-
 void pa_source_output_set_state_within_thread(pa_source_output *o, pa_source_output_state_t state);
 
+int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
+
 pa_usec_t pa_source_output_set_requested_latency_within_thread(pa_source_output *o, pa_usec_t usec);
 
 #endif
diff --git a/src/pulsecore/source.c b/src/pulsecore/source.c
index fc4734fd..426906eb 100644
--- a/src/pulsecore/source.c
+++ b/src/pulsecore/source.c
@@ -107,7 +107,6 @@ static void reset_callbacks(pa_source *s) {
     s->set_volume = NULL;
     s->get_mute = NULL;
     s->set_mute = NULL;
-    s->get_latency = NULL;
     s->update_requested_latency = NULL;
 }
 
@@ -440,9 +439,6 @@ pa_usec_t pa_source_get_latency(pa_source *s) {
     if (!PA_SOURCE_IS_OPENED(s->state))
         return 0;
 
-    if (s->get_latency)
-        return s->get_latency(s);
-
     if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
         return 0;
 
diff --git a/src/pulsecore/source.h b/src/pulsecore/source.h
index cce54620..f9c9cbf9 100644
--- a/src/pulsecore/source.h
+++ b/src/pulsecore/source.h
@@ -100,7 +100,6 @@ struct pa_source {
     int (*get_volume)(pa_source *s);         /* dito */
     int (*set_mute)(pa_source *s);           /* dito */
     int (*get_mute)(pa_source *s);           /* dito */
-    pa_usec_t (*get_latency)(pa_source *s);  /* dito */
     void (*update_requested_latency)(pa_source *s); /* dito */
 
     /* Contains copies of the above data so that the real-time worker