[PATCH v10 2/3] pipe-source: respect frame boundaries

[rommer@xxxxxxxxxxx (Raman Shyshniou)]

Currently pipe-source doesn't track frame boundaries while posting
data. This can lead to assertion failure in source-output for any
sample specs except s8/u8 mono. The simplest way to reproduce is
to write s24le format to pipe:
$ pacat -r -d null --format=s24le > /tmp/music.input

E: [pipe-source] source-output.c: Assertion
'pa_frame_aligned(chunk->length, &o->source->sample_spec)' failed at
pulsecore/source-output.c:738, function pa_source_output_push().
---
 src/modules/module-pipe-source.c | 52 ++++++++++++++++++++++++++++++++++++----
 1 file changed, 47 insertions(+), 5 deletions(-)

diff --git a/src/modules/module-pipe-source.c b/src/modules/module-pipe-source.c
index c069dd4..4fd515b 100644
--- a/src/modules/module-pipe-source.c
+++ b/src/modules/module-pipe-source.c
@@ -202,6 +202,7 @@ static void thread_func(void *userdata) {
     struct userdata *u = userdata;
     struct pollfd *pollfd;
     int read_type = 0;
+    size_t fs;
 
     pa_assert(u);
 
@@ -209,6 +210,8 @@ static void thread_func(void *userdata) {
 
     pa_thread_mq_install(&u->thread_mq);
 
+    fs = pa_frame_size(&u->source->sample_spec);
+
     u->memchunk.memblock = pa_memblock_new(u->core->mempool, u->pipe_size);
     u->timestamp = pa_rtclock_now();
 
@@ -229,7 +232,7 @@ static void thread_func(void *userdata) {
             pa_assert(u->memchunk.length == 0);
 
             p = pa_memblock_acquire(u->memchunk.memblock);
-            l = pa_read(u->fd, p, pa_memblock_get_length(u->memchunk.memblock), &read_type);
+            l = pa_read(u->fd, (uint8_t*) p + u->memchunk.index, pa_memblock_get_length(u->memchunk.memblock) - u->memchunk.index, &read_type);
             pa_memblock_release(u->memchunk.memblock);
 
             if (l < 0) {
@@ -252,6 +255,7 @@ static void thread_func(void *userdata) {
                     }
 
                     u->writer_connected = false;
+                    u->memchunk.index = 0;
                 }
 
             } else {
@@ -269,21 +273,59 @@ static void thread_func(void *userdata) {
 
         /* Post data to source */
         if (PA_SOURCE_IS_OPENED(u->source->thread_info.state)) {
+            size_t total_len = u->memchunk.index + u->memchunk.length;
+            size_t frames = total_len / fs;
+            size_t tail = total_len % fs;
 
             if (u->writer_connected && u->corkfd >= 0) {
                 pa_assert_se(pa_close(u->corkfd) == 0);
                 u->corkfd = -1;
             }
 
-            if (u->memchunk.length > 0) {
+            if (frames > 0) {
+                pa_memblock *new_memblock;
+
+                new_memblock = pa_memblock_new(u->core->mempool, u->pipe_size);
+
+                u->memchunk.index = 0;
+                u->memchunk.length = frames * fs;
+
+                if (tail > 0) {
+                    void *src, *dst;
+
+                    dst = pa_memblock_acquire(new_memblock);
+                    src = pa_memblock_acquire(u->memchunk.memblock);
+
+                    memcpy(dst, (uint8_t *) src + u->memchunk.length, tail);
+
+                    pa_memblock_release(u->memchunk.memblock);
+                    pa_memblock_release(new_memblock);
+                }
+
                 pa_source_post(u->source, &u->memchunk);
                 pa_memblock_unref(u->memchunk.memblock);
-                u->memchunk.memblock = pa_memblock_new(u->core->mempool, u->pipe_size);
-                u->memchunk.length = 0;
+                u->memchunk.memblock = new_memblock;
+            }
+
+            u->memchunk.index = tail;
+            u->memchunk.length = 0;
+
+        } else if (u->suspended_by_user) {
+            size_t total_len = u->memchunk.index + u->memchunk.length;
+            size_t frames = total_len / fs;
+            size_t tail = total_len % fs;
+
+            if (frames > 0 && tail > 0) {
+                void *p;
+
+                p = pa_memblock_acquire(u->memchunk.memblock);
+                memmove(p, (uint8_t *) p + frames * fs, tail);
+                pa_memblock_release(u->memchunk.memblock);
             }
 
-        } else if (u->suspended_by_user)
+            u->memchunk.index = tail;
             u->memchunk.length = 0;
+        }
 
         pollfd->events = u->memchunk.length ? 0 : POLLIN;
 
-- 
1.8.3.1