native: fix request counter miscalculations

Do not subtract bytes the client sends us beyond what we requested from
our missing bytes counter.

This was mostly a thinko that caused servers asking for too little data
when the client initially sent more data than requested, because that
data sent too much was accounted for twice.

This commit fixes this miscalculation.

http://bugzilla.redhat.com/show_bug.cgi?id=534130

3 files changed, 55 insertions, 57 deletions

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 40812566..0f9e50ab 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -741,6 +741,8 @@ void pa_command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tag
 
     s->requested_bytes += bytes;
 
+    /* pa_log("got request for %lli, now at %lli", (long long) bytes, (long long) s->requested_bytes); */
+
     if (s->requested_bytes > 0 && s->write_callback)
         s->write_callback(s, (size_t) s->requested_bytes, s->write_userdata);
 
@@ -1362,6 +1364,8 @@ int pa_stream_write(
      * that's OK, the server side applies the same error */
     s->requested_bytes -= (seek == PA_SEEK_RELATIVE ? offset : 0) + (int64_t) length;
 
+    /* pa_log("wrote %lli, now at %lli", (long long) length, (long long) s->requested_bytes); */
+
     if (s->direction == PA_STREAM_PLAYBACK) {
 
         /* Update latency request correction */
diff --git a/src/pulsecore/memblockq.c b/src/pulsecore/memblockq.c
index 32758be3..4641801d 100644
--- a/src/pulsecore/memblockq.c
+++ b/src/pulsecore/memblockq.c
@@ -55,8 +55,7 @@ struct pa_memblockq {
     pa_bool_t in_prebuf;
     pa_memchunk silence;
     pa_mcalign *mcalign;
-    int64_t missing;
-    size_t requested;
+    int64_t missing, requested;
 };
 
 pa_memblockq* pa_memblockq_new(
@@ -84,8 +83,8 @@ pa_memblockq* pa_memblockq_new(
     pa_log_debug("memblockq requested: maxlength=%lu, tlength=%lu, base=%lu, prebuf=%lu, minreq=%lu maxrewind=%lu",
                  (unsigned long) maxlength, (unsigned long) tlength, (unsigned long) base, (unsigned long) prebuf, (unsigned long) minreq, (unsigned long) maxrewind);
 
-    bq->missing = 0;
-    bq->requested = bq->maxlength = bq->tlength = bq->prebuf = bq->minreq = bq->maxrewind = 0;
+    bq->missing = bq->requested = 0;
+    bq->maxlength = bq->tlength = bq->prebuf = bq->minreq = bq->maxrewind = 0;
     bq->in_prebuf = TRUE;
 
     pa_memblockq_set_maxlength(bq, maxlength);
@@ -246,10 +245,34 @@ static pa_bool_t can_push(pa_memblockq *bq, size_t l) {
     return TRUE;
 }
 
+static void write_index_changed(pa_memblockq *bq, int64_t old_write_index, pa_bool_t account) {
+    int64_t delta;
+
+    pa_assert(bq);
+
+    delta = bq->write_index - old_write_index;
+
+    if (account)
+        bq->requested -= delta;
+
+    /* pa_log("pushed/seeked %lli: requested counter at %lli, account=%i", (long long) delta, (long long) bq->requested, account); */
+}
+
+static void read_index_changed(pa_memblockq *bq, int64_t old_read_index) {
+    int64_t delta;
+
+    pa_assert(bq);
+
+    delta = bq->read_index - old_read_index;
+    bq->missing += delta;
+
+    /* pa_log("popped %lli: missing counter at %lli", (long long) delta, (long long) bq->missing); */
+}
+
 int pa_memblockq_push(pa_memblockq* bq, const pa_memchunk *uchunk) {
     struct list_item *q, *n;
     pa_memchunk chunk;
-    int64_t old, delta;
+    int64_t old;
 
     pa_assert(bq);
     pa_assert(uchunk);
@@ -409,18 +432,7 @@ int pa_memblockq_push(pa_memblockq* bq, const pa_memchunk *uchunk) {
 
 finish:
 
-    delta = bq->write_index - old;
-
-    if (delta >= (int64_t) bq->requested) {
-        delta -= (int64_t) bq->requested;
-        bq->requested = 0;
-    } else {
-        bq->requested -= (size_t) delta;
-        delta = 0;
-    }
-
-    bq->missing -= delta;
-
+    write_index_changed(bq, old, TRUE);
     return 0;
 }
 
@@ -514,7 +526,7 @@ int pa_memblockq_peek(pa_memblockq* bq, pa_memchunk *chunk) {
 }
 
 void pa_memblockq_drop(pa_memblockq *bq, size_t length) {
-    int64_t old, delta;
+    int64_t old;
     pa_assert(bq);
     pa_assert(length % bq->base == 0);
 
@@ -553,19 +565,21 @@ void pa_memblockq_drop(pa_memblockq *bq, size_t length) {
     }
 
     drop_backlog(bq);
-
-    delta = bq->read_index - old;
-    bq->missing += delta;
+    read_index_changed(bq, old);
 }
 
 void pa_memblockq_rewind(pa_memblockq *bq, size_t length) {
+    int64_t old;
     pa_assert(bq);
     pa_assert(length % bq->base == 0);
 
+    old = bq->read_index;
+
     /* This is kind of the inverse of pa_memblockq_drop() */
 
     bq->read_index -= (int64_t) length;
-    bq->missing -= (int64_t) length;
+
+    read_index_changed(bq, old);
 }
 
 pa_bool_t pa_memblockq_is_readable(pa_memblockq *bq) {
@@ -602,7 +616,7 @@ size_t pa_memblockq_missing(pa_memblockq *bq) {
 }
 
 void pa_memblockq_seek(pa_memblockq *bq, int64_t offset, pa_seek_mode_t seek, pa_bool_t account) {
-    int64_t old, delta;
+    int64_t old;
     pa_assert(bq);
 
     old = bq->write_index;
@@ -625,24 +639,11 @@ void pa_memblockq_seek(pa_memblockq *bq, int64_t offset, pa_seek_mode_t seek, pa
     }
 
     drop_backlog(bq);
-
-    delta = bq->write_index - old;
-
-    if (account) {
-        if (delta >= (int64_t) bq->requested) {
-            delta -= (int64_t) bq->requested;
-            bq->requested = 0;
-        } else if (delta >= 0) {
-            bq->requested -= (size_t) delta;
-            delta = 0;
-        }
-    }
-
-    bq->missing -= delta;
+    write_index_changed(bq, old, account);
 }
 
 void pa_memblockq_flush_write(pa_memblockq *bq) {
-    int64_t old, delta;
+    int64_t old;
     pa_assert(bq);
 
     pa_memblockq_silence(bq);
@@ -651,22 +652,11 @@ void pa_memblockq_flush_write(pa_memblockq *bq) {
     bq->write_index = bq->read_index;
 
     pa_memblockq_prebuf_force(bq);
-
-    delta = bq->write_index - old;
-
-    if (delta >= (int64_t) bq->requested) {
-        delta -= (int64_t) bq->requested;
-        bq->requested = 0;
-    } else if (delta >= 0) {
-        bq->requested -= (size_t) delta;
-        delta = 0;
-    }
-
-    bq->missing -= delta;
+    write_index_changed(bq, old, TRUE);
 }
 
 void pa_memblockq_flush_read(pa_memblockq *bq) {
-    int64_t old, delta;
+    int64_t old;
     pa_assert(bq);
 
     pa_memblockq_silence(bq);
@@ -675,9 +665,7 @@ void pa_memblockq_flush_read(pa_memblockq *bq) {
     bq->read_index = bq->write_index;
 
     pa_memblockq_prebuf_force(bq);
-
-    delta = bq->read_index - old;
-    bq->missing += delta;
+    read_index_changed(bq, old);
 }
 
 size_t pa_memblockq_get_tlength(pa_memblockq *bq) {
@@ -774,8 +762,11 @@ size_t pa_memblockq_pop_missing(pa_memblockq *bq) {
         return 0;
 
     l = (size_t) bq->missing;
+
+    bq->requested += bq->missing;
     bq->missing = 0;
-    bq->requested += l;
+
+    /* pa_log("sent %lli: request counter is at %lli", (long long) l, (long long) bq->requested); */
 
     return l;
 }
diff --git a/src/pulsecore/protocol-native.c b/src/pulsecore/protocol-native.c
index d49a78e5..89f33d08 100644
--- a/src/pulsecore/protocol-native.c
+++ b/src/pulsecore/protocol-native.c
@@ -1114,6 +1114,8 @@ static playback_stream* playback_stream_new(
 
     *missing = (uint32_t) pa_memblockq_pop_missing(s->memblockq);
 
+    /* pa_log("missing original: %li", (long int) *missing); */
+
     *ss = s->sink_input->sample_spec;
     *map = s->sink_input->channel_map;
 
@@ -1138,11 +1140,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)", */
+    /* pa_log("request_bytes(%lu) (tlength=%lu minreq=%lu length=%lu really missing=%lli)", */
     /*        (unsigned long) m, */
     /*        pa_memblockq_get_tlength(s->memblockq), */
     /*        pa_memblockq_get_minreq(s->memblockq), */
-    /*        pa_memblockq_get_length(s->memblockq)); */
+    /*        pa_memblockq_get_length(s->memblockq), */
+    /*        (long long) pa_memblockq_get_tlength(s->memblockq) - (long long) pa_memblockq_get_length(s->memblockq)); */
 
     if (m <= 0)
         return;