native: rework handling of seeks that depend on variables the client does not know anything about

All seeks/flushes that depend on the playback buffer read pointer cannot
be accounted for properly in the client since it does not know the
actual read pointer. Due to that the clients do not account for it at
all. We need do the same on the server side. And we did, but a little
bit too extreme. While we properly have not applied the changes to the
"request" counter we still do have to apply it to the "missing" counter.
This patch fixes that.

7 files changed, 20 insertions, 9 deletions

diff --git a/src/modules/module-combine.c b/src/modules/module-combine.c
index 62e62679..cffb901b 100644
--- a/src/modules/module-combine.c
+++ b/src/modules/module-combine.c
@@ -562,7 +562,7 @@ static int sink_input_process_msg(pa_msgobject *obj, int code, void *data, int64
             if (PA_SINK_IS_OPENED(o->sink_input->sink->thread_info.state))
                 pa_memblockq_push_align(o->memblockq, chunk);
             else
-                pa_memblockq_flush_write(o->memblockq);
+                pa_memblockq_flush_write(o->memblockq, TRUE);
 
             return 0;
     }
@@ -982,7 +982,7 @@ static void output_disable(struct output *o) {
     o->sink_input = NULL;
 
     /* Finally, drop all queued data */
-    pa_memblockq_flush_write(o->memblockq);
+    pa_memblockq_flush_write(o->memblockq, TRUE);
     pa_asyncmsgq_flush(o->inq, FALSE);
     pa_asyncmsgq_flush(o->outq, FALSE);
 }
diff --git a/src/modules/module-loopback.c b/src/modules/module-loopback.c
index bb0182b0..15ef96ef 100644
--- a/src/modules/module-loopback.c
+++ b/src/modules/module-loopback.c
@@ -430,7 +430,7 @@ static int sink_input_process_msg_cb(pa_msgobject *obj, int code, void *data, in
             if (PA_SINK_IS_OPENED(u->sink_input->sink->thread_info.state))
                 pa_memblockq_push_align(u->memblockq, chunk);
             else
-                pa_memblockq_flush_write(u->memblockq);
+                pa_memblockq_flush_write(u->memblockq, TRUE);
 
             update_min_memblockq_length(u);
 
@@ -457,7 +457,7 @@ static int sink_input_process_msg_cb(pa_msgobject *obj, int code, void *data, in
             if (PA_SINK_IS_OPENED(u->sink_input->sink->thread_info.state))
                 pa_memblockq_seek(u->memblockq, -offset, PA_SEEK_RELATIVE, TRUE);
             else
-                pa_memblockq_flush_write(u->memblockq);
+                pa_memblockq_flush_write(u->memblockq, TRUE);
 
             u->recv_counter -= offset;
 
diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 5e42ab28..36514e0e 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -2154,6 +2154,11 @@ pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *use
          * index, but the read index might jump. */
         invalidate_indexes(s, TRUE, FALSE);
 
+    /* Note that we do not update requested_bytes here. This is
+     * because we cannot really know how data actually was dropped
+     * from the write index due to this. This 'error' will be applied
+     * by both client and server and hence we should be fine. */
+
     return o;
 }
 
diff --git a/src/pulsecore/memblockq.c b/src/pulsecore/memblockq.c
index 4641801d..2b063fac 100644
--- a/src/pulsecore/memblockq.c
+++ b/src/pulsecore/memblockq.c
@@ -254,6 +254,8 @@ static void write_index_changed(pa_memblockq *bq, int64_t old_write_index, pa_bo
 
     if (account)
         bq->requested -= delta;
+    else
+        bq->missing -= delta;
 
     /* pa_log("pushed/seeked %lli: requested counter at %lli, account=%i", (long long) delta, (long long) bq->requested, account); */
 }
@@ -642,7 +644,7 @@ void pa_memblockq_seek(pa_memblockq *bq, int64_t offset, pa_seek_mode_t seek, pa
     write_index_changed(bq, old, account);
 }
 
-void pa_memblockq_flush_write(pa_memblockq *bq) {
+void pa_memblockq_flush_write(pa_memblockq *bq, pa_bool_t account) {
     int64_t old;
     pa_assert(bq);
 
@@ -652,7 +654,7 @@ void pa_memblockq_flush_write(pa_memblockq *bq) {
     bq->write_index = bq->read_index;
 
     pa_memblockq_prebuf_force(bq);
-    write_index_changed(bq, old, TRUE);
+    write_index_changed(bq, old, account);
 }
 
 void pa_memblockq_flush_read(pa_memblockq *bq) {
diff --git a/src/pulsecore/memblockq.h b/src/pulsecore/memblockq.h
index b756549a..402c6cdb 100644
--- a/src/pulsecore/memblockq.h
+++ b/src/pulsecore/memblockq.h
@@ -118,7 +118,7 @@ size_t pa_memblockq_pop_missing(pa_memblockq *bq);
 int pa_memblockq_splice(pa_memblockq *bq, pa_memblockq *source);
 
 /* Set the queue to silence, set write index to read index */
-void pa_memblockq_flush_write(pa_memblockq *bq);
+void pa_memblockq_flush_write(pa_memblockq *bq, pa_bool_t account);
 
 /* Set the queue to silence, set write read index to write index*/
 void pa_memblockq_flush_read(pa_memblockq *bq);
diff --git a/src/pulsecore/protocol-native.c b/src/pulsecore/protocol-native.c
index 89f33d08..ff246120 100644
--- a/src/pulsecore/protocol-native.c
+++ b/src/pulsecore/protocol-native.c
@@ -1326,6 +1326,10 @@ static void handle_seek(playback_stream *s, int64_t indexw) {
     playback_stream_request_bytes(s);
 }
 
+static void flush_write_no_account(pa_memblockq *q) {
+    pa_memblockq_flush_write(q, FALSE);
+}
+
 /* Called from thread context */
 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
     pa_sink_input *i = PA_SINK_INPUT(o);
@@ -1387,7 +1391,7 @@ static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int
 
             switch  (code) {
                 case SINK_INPUT_MESSAGE_FLUSH:
-                    func = pa_memblockq_flush_write;
+                    func = flush_write_no_account;
                     break;
 
                 case SINK_INPUT_MESSAGE_PREBUF_FORCE:
diff --git a/src/pulsecore/sink-input.c b/src/pulsecore/sink-input.c
index e73999d9..395110b7 100644
--- a/src/pulsecore/sink-input.c
+++ b/src/pulsecore/sink-input.c
@@ -842,7 +842,7 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam
         /* We were asked to drop all buffered data, and rerequest new
          * data from implementor the next time push() is called */
 
-        pa_memblockq_flush_write(i->thread_info.render_memblockq);
+        pa_memblockq_flush_write(i->thread_info.render_memblockq, TRUE);
 
     } else if (i->thread_info.rewrite_nbytes > 0) {
         size_t max_rewrite, amount;