1 files changed, 63 insertions, 95 deletions

diff --git a/src/pulsecore/pstream.c b/src/pulsecore/pstream.c
index fdb1a66a..f4aab1c2 100644
--- a/src/pulsecore/pstream.c
+++ b/src/pulsecore/pstream.c
@@ -49,7 +49,6 @@
 #include <pulsecore/log.h>
 #include <pulsecore/core-scache.h>
 #include <pulsecore/creds.h>
-#include <pulsecore/mutex.h>
 #include <pulsecore/refcnt.h>
 
 #include "pstream.h"
@@ -118,8 +117,8 @@ struct pa_pstream {
     pa_mainloop_api *mainloop;
     pa_defer_event *defer_event;
     pa_iochannel *io;
+
     pa_queue *send_queue;
-    pa_mutex *mutex;
 
     int dead;
 
@@ -129,6 +128,7 @@ struct pa_pstream {
         uint32_t shm_info[PA_PSTREAM_SHM_MAX];
         void *data;
         size_t index;
+        pa_memchunk memchunk;
     } write;
 
     struct {
@@ -173,8 +173,6 @@ static void do_something(pa_pstream *p) {
 
     pa_pstream_ref(p);
 
-    pa_mutex_lock(p->mutex);
-
     p->mainloop->defer_enable(p->defer_event, 0);
 
     if (!p->dead && pa_iochannel_is_readable(p->io)) {
@@ -188,8 +186,6 @@ static void do_something(pa_pstream *p) {
             goto fail;
     }
 
-    pa_mutex_unlock(p->mutex);
-
     pa_pstream_unref(p);
     return;
 
@@ -200,8 +196,6 @@ fail:
     if (p->die_callback)
         p->die_callback(p, p->die_callback_userdata);
 
-    pa_mutex_unlock(p->mutex);
-
     pa_pstream_unref(p);
 }
 
@@ -214,16 +208,6 @@ static void io_callback(pa_iochannel*io, void *userdata) {
     do_something(p);
 }
 
-static void defer_callback(pa_mainloop_api *m, pa_defer_event *e, void*userdata) {
-    pa_pstream *p = userdata;
-
-    assert(p);
-    assert(p->defer_event == e);
-    assert(p->mainloop == m);
-
-    do_something(p);
-}
-
 static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userdata);
 
 pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_mempool *pool) {
@@ -239,17 +223,14 @@ pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_mempool *poo
     pa_iochannel_set_callback(io, io_callback, p);
     p->dead = 0;
 
-    p->mutex = pa_mutex_new(1);
-
     p->mainloop = m;
-    p->defer_event = m->defer_new(m, defer_callback, p);
-    m->defer_enable(p->defer_event, 0);
 
     p->send_queue = pa_queue_new();
     assert(p->send_queue);
 
     p->write.current = NULL;
     p->write.index = 0;
+    pa_memchunk_reset(&p->write.memchunk);
     p->read.memblock = NULL;
     p->read.packet = NULL;
     p->read.index = 0;
@@ -312,8 +293,8 @@ static void pstream_free(pa_pstream *p) {
     if (p->read.packet)
         pa_packet_unref(p->read.packet);
 
-    if (p->mutex)
-        pa_mutex_free(p->mutex);
+    if (p->write.memchunk.memblock)
+        pa_memblock_unref(p->write.memchunk.memblock);
 
     pa_xfree(p);
 }
@@ -325,10 +306,8 @@ void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet, const pa_creds *cre
     assert(PA_REFCNT_VALUE(p) > 0);
     assert(packet);
 
-    pa_mutex_lock(p->mutex);
-
     if (p->dead)
-        goto finish;
+        return;
 
     i = pa_xnew(struct item_info, 1);
     i->type = PA_PSTREAM_ITEM_PACKET;
@@ -340,11 +319,6 @@ void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet, const pa_creds *cre
 #endif
 
     pa_queue_push(p->send_queue, i);
-    p->mainloop->defer_enable(p->defer_event, 1);
-
-finish:
-
-    pa_mutex_unlock(p->mutex);
 }
 
 void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa_seek_mode_t seek_mode, const pa_memchunk *chunk) {
@@ -355,12 +329,9 @@ void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa
     assert(channel != (uint32_t) -1);
     assert(chunk);
 
-    pa_mutex_lock(p->mutex);
-
     if (p->dead)
-        goto finish;
+        return;
 
-    length = chunk->length;
     idx = 0;
 
     while (length > 0) {
@@ -389,10 +360,6 @@ void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa
     }
 
     p->mainloop->defer_enable(p->defer_event, 1);
-
-finish:
-
-    pa_mutex_unlock(p->mutex);
 }
 
 static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userdata) {
@@ -402,10 +369,8 @@ static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userd
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    pa_mutex_lock(p->mutex);
-
     if (p->dead)
-        goto finish;
+        return;
 
 /*     pa_log("Releasing block %u", block_id); */
 
@@ -417,11 +382,6 @@ static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userd
 #endif
 
     pa_queue_push(p->send_queue, item);
-    p->mainloop->defer_enable(p->defer_event, 1);
-
-finish:
-
-    pa_mutex_unlock(p->mutex);
 }
 
 static void memexport_revoke_cb(pa_memexport *e, uint32_t block_id, void *userdata) {
@@ -431,11 +391,8 @@ static void memexport_revoke_cb(pa_memexport *e, uint32_t block_id, void *userda
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    pa_mutex_lock(p->mutex);
-
     if (p->dead)
-        goto finish;
-
+        return;
 /*     pa_log("Revoking block %u", block_id); */
 
     item = pa_xnew(struct item_info, 1);
@@ -446,22 +403,20 @@ static void memexport_revoke_cb(pa_memexport *e, uint32_t block_id, void *userda
 #endif
 
     pa_queue_push(p->send_queue, item);
-    p->mainloop->defer_enable(p->defer_event, 1);
-
-finish:
-
-    pa_mutex_unlock(p->mutex);
 }
 
 static void prepare_next_write_item(pa_pstream *p) {
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    if (!(p->write.current = pa_queue_pop(p->send_queue)))
+    p->write.current = pa_queue_pop(p->send_queue);
+
+    if (!p->write.current)
         return;
 
     p->write.index = 0;
     p->write.data = NULL;
+    pa_memchunk_reset(&p->write.memchunk);
 
     p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = 0;
     p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl((uint32_t) -1);
@@ -528,7 +483,9 @@ static void prepare_next_write_item(pa_pstream *p) {
 
         if (send_payload) {
             p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->chunk.length);
-            p->write.data = (uint8_t*) p->write.current->chunk.memblock->data + p->write.current->chunk.index;
+            p->write.memchunk = p->write.current->chunk;
+            pa_memblock_ref(p->write.memchunk.memblock);
+            p->write.data = NULL;
         }
 
         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(flags);
@@ -544,6 +501,7 @@ static int do_write(pa_pstream *p) {
     void *d;
     size_t l;
     ssize_t r;
+    pa_memblock *release_memblock = NULL;
 
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
@@ -558,9 +516,16 @@ static int do_write(pa_pstream *p) {
         d = (uint8_t*) p->write.descriptor + p->write.index;
         l = PA_PSTREAM_DESCRIPTOR_SIZE - p->write.index;
     } else {
-        assert(p->write.data);
+        assert(p->write.data || p->write.memchunk.memblock);
+
+        if (p->write.data)
+            d = p->write.data;
+        else {
+            d = (uint8_t*) pa_memblock_acquire(p->write.memchunk.memblock) + p->write.memchunk.index;
+            release_memblock = p->write.memchunk.memblock;
+        }
 
-        d = (uint8_t*) p->write.data + p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE;
+        d = (uint8_t*) d + p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE;
         l = ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE);
     }
 
@@ -570,14 +535,17 @@ static int do_write(pa_pstream *p) {
     if (p->send_creds_now) {
 
         if ((r = pa_iochannel_write_with_creds(p->io, d, l, &p->write_creds)) < 0)
-            return -1;
+            goto fail;
 
         p->send_creds_now = 0;
     } else
 #endif
 
     if ((r = pa_iochannel_write(p->io, d, l)) < 0)
-        return -1;
+        goto fail;
+
+    if (release_memblock)
+        pa_memblock_release(release_memblock);
 
     p->write.index += r;
 
@@ -591,13 +559,20 @@ static int do_write(pa_pstream *p) {
     }
 
     return 0;
+
+fail:
+
+    if (release_memblock)
+        pa_memblock_release(release_memblock);
+
+    return -1;
 }
 
 static int do_read(pa_pstream *p) {
     void *d;
     size_t l;
     ssize_t r;
-
+    pa_memblock *release_memblock = NULL;
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
@@ -605,8 +580,16 @@ static int do_read(pa_pstream *p) {
         d = (uint8_t*) p->read.descriptor + p->read.index;
         l = PA_PSTREAM_DESCRIPTOR_SIZE - p->read.index;
     } else {
-        assert(p->read.data);
-        d = (uint8_t*) p->read.data + p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE;
+        assert(p->read.data || p->read.memblock);
+
+        if (p->read.data)
+            d = p->read.data;
+        else {
+            d = pa_memblock_acquire(p->read.memblock);
+            release_memblock = p->read.memblock;
+        }
+
+        d = (uint8_t*) d + p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE;
         l = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE);
     }
 
@@ -615,15 +598,18 @@ static int do_read(pa_pstream *p) {
         int b = 0;
 
         if ((r = pa_iochannel_read_with_creds(p->io, d, l, &p->read_creds, &b)) <= 0)
-            return -1;
+            goto fail;
 
         p->read_creds_valid = p->read_creds_valid || b;
     }
 #else
     if ((r = pa_iochannel_read(p->io, d, l)) <= 0)
-        return -1;
+        goto fail;
 #endif
 
+    if (release_memblock)
+        pa_memblock_release(release_memblock);
+
     p->read.index += r;
 
     if (p->read.index == PA_PSTREAM_DESCRIPTOR_SIZE) {
@@ -704,7 +690,7 @@ static int do_read(pa_pstream *p) {
                 /* Frame is a memblock frame */
 
                 p->read.memblock = pa_memblock_new(p->mempool, length);
-                p->read.data = p->read.memblock->data;
+                p->read.data = NULL;
             } else {
 
                 pa_log_warn("Recieved memblock frame with invalid flags value.");
@@ -791,7 +777,7 @@ static int do_read(pa_pstream *p) {
 
                     chunk.memblock = b;
                     chunk.index = 0;
-                    chunk.length = b->length;
+                    chunk.length = pa_memblock_get_length(b);
 
                     offset = (int64_t) (
                             (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) |
@@ -819,52 +805,51 @@ frame_done:
     p->read.memblock = NULL;
     p->read.packet = NULL;
     p->read.index = 0;
+    p->read.data = NULL;
 
 #ifdef HAVE_CREDS
     p->read_creds_valid = 0;
 #endif
 
     return 0;
+
+fail:
+    if (release_memblock)
+        pa_memblock_release(release_memblock);
+
+    return -1;
 }
 
 void pa_pstream_set_die_callback(pa_pstream *p, pa_pstream_notify_cb_t cb, void *userdata) {
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    pa_mutex_lock(p->mutex);
     p->die_callback = cb;
     p->die_callback_userdata = userdata;
-    pa_mutex_unlock(p->mutex);
 }
 
 void pa_pstream_set_drain_callback(pa_pstream *p, pa_pstream_notify_cb_t cb, void *userdata) {
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    pa_mutex_lock(p->mutex);
     p->drain_callback = cb;
     p->drain_callback_userdata = userdata;
-    pa_mutex_unlock(p->mutex);
 }
 
 void pa_pstream_set_recieve_packet_callback(pa_pstream *p, pa_pstream_packet_cb_t cb, void *userdata) {
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    pa_mutex_lock(p->mutex);
     p->recieve_packet_callback = cb;
     p->recieve_packet_callback_userdata = userdata;
-    pa_mutex_unlock(p->mutex);
 }
 
 void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, pa_pstream_memblock_cb_t cb, void *userdata) {
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    pa_mutex_lock(p->mutex);
     p->recieve_memblock_callback = cb;
     p->recieve_memblock_callback_userdata = userdata;
-    pa_mutex_unlock(p->mutex);
 }
 
 int pa_pstream_is_pending(pa_pstream *p) {
@@ -873,15 +858,11 @@ int pa_pstream_is_pending(pa_pstream *p) {
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    pa_mutex_lock(p->mutex);
-
     if (p->dead)
         b = 0;
     else
         b = p->write.current || !pa_queue_is_empty(p->send_queue);
 
-    pa_mutex_unlock(p->mutex);
-
     return b;
 }
 
@@ -904,8 +885,6 @@ pa_pstream* pa_pstream_ref(pa_pstream*p) {
 void pa_pstream_close(pa_pstream *p) {
     assert(p);
 
-    pa_mutex_lock(p->mutex);
-
     p->dead = 1;
 
     if (p->import) {
@@ -923,25 +902,16 @@ void pa_pstream_close(pa_pstream *p) {
         p->io = NULL;
     }
 
-    if (p->defer_event) {
-        p->mainloop->defer_free(p->defer_event);
-        p->defer_event = NULL;
-    }
-
     p->die_callback = NULL;
     p->drain_callback = NULL;
     p->recieve_packet_callback = NULL;
     p->recieve_memblock_callback = NULL;
-
-    pa_mutex_unlock(p->mutex);
 }
 
 void pa_pstream_use_shm(pa_pstream *p, int enable) {
     assert(p);
     assert(PA_REFCNT_VALUE(p) > 0);
 
-    pa_mutex_lock(p->mutex);
-
     p->use_shm = enable;
 
     if (enable) {
@@ -956,6 +926,4 @@ void pa_pstream_use_shm(pa_pstream *p, int enable) {
             p->export = NULL;
         }
     }
-
-    pa_mutex_unlock(p->mutex);
 }