diff options
Diffstat (limited to 'src/pulsecore/pstream.c')
| -rw-r--r-- | src/pulsecore/pstream.c | 484 |
1 files changed, 278 insertions, 206 deletions
diff --git a/src/pulsecore/pstream.c b/src/pulsecore/pstream.c index 33963796..e26ca473 100644 --- a/src/pulsecore/pstream.c +++ b/src/pulsecore/pstream.c @@ -1,18 +1,19 @@ -/* $Id$ */ - /*** This file is part of PulseAudio. - + + Copyright 2004-2006 Lennart Poettering + Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB + PulseAudio is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version. - + PulseAudio is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. - + You should have received a copy of the GNU Lesser General Public License along with PulseAudio; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -25,7 +26,6 @@ #include <stdio.h> #include <stdlib.h> -#include <assert.h> #include <unistd.h> #ifdef HAVE_SYS_SOCKET_H @@ -38,17 +38,17 @@ #include <netinet/in.h> #endif -#include "winsock.h" #include <pulse/xmalloc.h> +#include <pulsecore/winsock.h> #include <pulsecore/queue.h> #include <pulsecore/log.h> #include <pulsecore/core-scache.h> #include <pulsecore/creds.h> -#include <pulsecore/mutex.h> #include <pulsecore/refcnt.h> -#include <pulsecore/anotify.h> +#include <pulsecore/flist.h> +#include <pulsecore/macro.h> #include "pstream.h" @@ -82,7 +82,8 @@ typedef uint32_t pa_pstream_descriptor[PA_PSTREAM_DESCRIPTOR_MAX]; #define PA_PSTREAM_DESCRIPTOR_SIZE (PA_PSTREAM_DESCRIPTOR_MAX*sizeof(uint32_t)) #define FRAME_SIZE_MAX_ALLOW PA_SCACHE_ENTRY_SIZE_MAX /* allow uploading a single sample in one frame at max */ -#define FRAME_SIZE_MAX_USE (1024*64) + +PA_STATIC_FLIST_DECLARE(items, 0, pa_xfree); struct item_info { enum { @@ -92,11 +93,10 @@ struct item_info { PA_PSTREAM_ITEM_SHMREVOKE } type; - /* packet info */ pa_packet *packet; #ifdef HAVE_CREDS - int with_creds; + pa_bool_t with_creds; pa_creds creds; #endif @@ -112,15 +112,14 @@ struct item_info { struct pa_pstream { PA_REFCNT_DECLARE; - + pa_mainloop_api *mainloop; + pa_defer_event *defer_event; pa_iochannel *io; pa_queue *send_queue; - pa_mutex *mutex; /* only for access to the queue */ - pa_anotify *anotify; - int dead; + pa_bool_t dead; struct { pa_pstream_descriptor descriptor; @@ -140,7 +139,7 @@ struct pa_pstream { size_t index; } read; - int use_shm; + pa_bool_t use_shm; pa_memimport *import; pa_memexport *export; @@ -156,11 +155,17 @@ struct pa_pstream { pa_pstream_notify_cb_t die_callback; void *die_callback_userdata; + pa_pstream_block_id_cb_t revoke_callback; + void *revoke_callback_userdata; + + pa_pstream_block_id_cb_t release_callback; + void *release_callback_userdata; + pa_mempool *mempool; #ifdef HAVE_CREDS pa_creds read_creds, write_creds; - int read_creds_valid, send_creds_now; + pa_bool_t read_creds_valid, send_creds_now; #endif }; @@ -168,11 +173,13 @@ static int do_write(pa_pstream *p); static int do_read(pa_pstream *p); static void do_something(pa_pstream *p) { - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); pa_pstream_ref(p); + p->mainloop->defer_enable(p->defer_event, 0); + if (!p->dead && pa_iochannel_is_readable(p->io)) { if (do_read(p) < 0) goto fail; @@ -189,27 +196,31 @@ static void do_something(pa_pstream *p) { fail: - p->dead = 1; - if (p->die_callback) p->die_callback(p, p->die_callback_userdata); - + + pa_pstream_unlink(p); pa_pstream_unref(p); } static void io_callback(pa_iochannel*io, void *userdata) { pa_pstream *p = userdata; - - assert(p); - assert(p->io == io); - + + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + pa_assert(p->io == io); + do_something(p); } -static void anotify_callback(uint8_t event, void *userdata) { +static void defer_callback(pa_mainloop_api *m, pa_defer_event *e, void*userdata) { pa_pstream *p = userdata; - assert(p); + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + pa_assert(p->defer_event == e); + pa_assert(p->mainloop == m); + do_something(p); } @@ -217,24 +228,22 @@ static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userd pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_mempool *pool) { pa_pstream *p; - - assert(m); - assert(io); - assert(pool); + + pa_assert(m); + pa_assert(io); + pa_assert(pool); p = pa_xnew(pa_pstream, 1); PA_REFCNT_INIT(p); p->io = io; pa_iochannel_set_callback(io, io_callback, p); - p->dead = 0; - - p->mutex = pa_mutex_new(1); - p->anotify = pa_anotify_new(m, anotify_callback, p); + p->dead = FALSE; 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; @@ -251,64 +260,63 @@ pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_mempool *poo p->drain_callback_userdata = NULL; p->die_callback = NULL; p->die_callback_userdata = NULL; + p->revoke_callback = NULL; + p->revoke_callback_userdata = NULL; + p->release_callback = NULL; + p->release_callback_userdata = NULL; p->mempool = pool; - p->use_shm = 0; + p->use_shm = FALSE; p->export = NULL; /* We do importing unconditionally */ p->import = pa_memimport_new(p->mempool, memimport_release_cb, p); - pa_iochannel_socket_set_rcvbuf(io, 1024*8); - pa_iochannel_socket_set_sndbuf(io, 1024*8); + pa_iochannel_socket_set_rcvbuf(io, pa_mempool_block_size_max(p->mempool)); + pa_iochannel_socket_set_sndbuf(io, pa_mempool_block_size_max(p->mempool)); #ifdef HAVE_CREDS - p->send_creds_now = 0; - p->read_creds_valid = 0; + p->send_creds_now = FALSE; + p->read_creds_valid = FALSE; #endif return p; } -static void item_free(void *item, PA_GCC_UNUSED void *p) { +static void item_free(void *item, PA_GCC_UNUSED void *q) { struct item_info *i = item; - assert(i); + pa_assert(i); if (i->type == PA_PSTREAM_ITEM_MEMBLOCK) { - assert(i->chunk.memblock); + pa_assert(i->chunk.memblock); pa_memblock_unref(i->chunk.memblock); } else if (i->type == PA_PSTREAM_ITEM_PACKET) { - assert(i->packet); + pa_assert(i->packet); pa_packet_unref(i->packet); } - pa_xfree(i); + if (pa_flist_push(PA_STATIC_FLIST_GET(items), i) < 0) + pa_xfree(i); } static void pstream_free(pa_pstream *p) { - assert(p); + pa_assert(p); + + pa_pstream_unlink(p); - pa_pstream_close(p); - pa_queue_free(p->send_queue, item_free, NULL); if (p->write.current) item_free(p->write.current, NULL); - if (p->read.memblock) - pa_memblock_unref(p->read.memblock); - - if (p->read.packet) - pa_packet_unref(p->read.packet); - if (p->write.memchunk.memblock) pa_memblock_unref(p->write.memchunk.memblock); - if (p->mutex) - pa_mutex_free(p->mutex); + if (p->read.memblock) + pa_memblock_unref(p->read.memblock); - if (p->anotify) - pa_anotify_free(p->anotify); + if (p->read.packet) + pa_packet_unref(p->read.packet); pa_xfree(p); } @@ -316,123 +324,156 @@ static void pstream_free(pa_pstream *p) { void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet, const pa_creds *creds) { struct item_info *i; - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); - assert(packet); + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + pa_assert(packet); + + if (p->dead) + return; + + if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items)))) + i = pa_xnew(struct item_info, 1); - i = pa_xnew(struct item_info, 1); i->type = PA_PSTREAM_ITEM_PACKET; i->packet = pa_packet_ref(packet); - + #ifdef HAVE_CREDS if ((i->with_creds = !!creds)) i->creds = *creds; #endif - pa_mutex_lock(p->mutex); pa_queue_push(p->send_queue, i); - pa_mutex_unlock(p->mutex); - - pa_anotify_signal(p->anotify, 0); + + p->mainloop->defer_enable(p->defer_event, 1); } void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa_seek_mode_t seek_mode, const pa_memchunk *chunk) { size_t length, idx; - - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); - assert(channel != (uint32_t) -1); - assert(chunk); + size_t bsm; + + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + pa_assert(channel != (uint32_t) -1); + pa_assert(chunk); + + if (p->dead) + return; idx = 0; + length = chunk->length; + + bsm = pa_mempool_block_size_max(p->mempool); while (length > 0) { struct item_info *i; size_t n; - - i = pa_xnew(struct item_info, 1); + + if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items)))) + i = pa_xnew(struct item_info, 1); i->type = PA_PSTREAM_ITEM_MEMBLOCK; - n = length < FRAME_SIZE_MAX_USE ? length : FRAME_SIZE_MAX_USE; + n = PA_MIN(length, bsm); i->chunk.index = chunk->index + idx; i->chunk.length = n; i->chunk.memblock = pa_memblock_ref(chunk->memblock); - + i->channel = channel; i->offset = offset; i->seek_mode = seek_mode; #ifdef HAVE_CREDS - i->with_creds = 0; + i->with_creds = FALSE; #endif - - pa_mutex_lock(p->mutex); + pa_queue_push(p->send_queue, i); - pa_mutex_unlock(p->mutex); idx += n; length -= n; } - pa_anotify_signal(p->anotify, 0); + p->mainloop->defer_enable(p->defer_event, 1); } -static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userdata) { +void pa_pstream_send_release(pa_pstream *p, uint32_t block_id) { struct item_info *item; - pa_pstream *p = userdata; + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); + if (p->dead) + return; /* pa_log("Releasing block %u", block_id); */ - item = pa_xnew(struct item_info, 1); + if (!(item = pa_flist_pop(PA_STATIC_FLIST_GET(items)))) + item = pa_xnew(struct item_info, 1); item->type = PA_PSTREAM_ITEM_SHMRELEASE; item->block_id = block_id; #ifdef HAVE_CREDS - item->with_creds = 0; + item->with_creds = FALSE; #endif - pa_mutex_lock(p->mutex); pa_queue_push(p->send_queue, item); - pa_mutex_unlock(p->mutex); - - pa_anotify_signal(p->anotify, 0); + p->mainloop->defer_enable(p->defer_event, 1); } -static void memexport_revoke_cb(pa_memexport *e, uint32_t block_id, void *userdata) { - struct item_info *item; +/* might be called from thread context */ +static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userdata) { pa_pstream *p = userdata; - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + + if (p->dead) + return; + + if (p->release_callback) + p->release_callback(p, block_id, p->release_callback_userdata); + else + pa_pstream_send_release(p, block_id); +} + +void pa_pstream_send_revoke(pa_pstream *p, uint32_t block_id) { + struct item_info *item; + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + if (p->dead) + return; /* pa_log("Revoking block %u", block_id); */ - - item = pa_xnew(struct item_info, 1); + + if (!(item = pa_flist_pop(PA_STATIC_FLIST_GET(items)))) + item = pa_xnew(struct item_info, 1); item->type = PA_PSTREAM_ITEM_SHMREVOKE; item->block_id = block_id; #ifdef HAVE_CREDS - item->with_creds = 0; + item->with_creds = FALSE; #endif - pa_mutex_lock(p->mutex); pa_queue_push(p->send_queue, item); - pa_mutex_unlock(p->mutex); + p->mainloop->defer_enable(p->defer_event, 1); +} + +/* might be called from thread context */ +static void memexport_revoke_cb(pa_memexport *e, uint32_t block_id, void *userdata) { + pa_pstream *p = userdata; + + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); - pa_anotify_signal(p->anotify, 0); + if (p->revoke_callback) + p->revoke_callback(p, block_id, p->revoke_callback_userdata); + else + pa_pstream_send_revoke(p, block_id); } static void prepare_next_write_item(pa_pstream *p) { - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); - pa_mutex_lock(p->mutex); p->write.current = pa_queue_pop(p->send_queue); - pa_mutex_unlock(p->mutex); if (!p->write.current) return; - + p->write.index = 0; p->write.data = NULL; pa_memchunk_reset(&p->write.memchunk); @@ -442,10 +483,10 @@ static void prepare_next_write_item(pa_pstream *p) { p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = 0; p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0; p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = 0; - + if (p->write.current->type == PA_PSTREAM_ITEM_PACKET) { - - assert(p->write.current->packet); + + pa_assert(p->write.current->packet); p->write.data = p->write.current->packet->data; p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->packet->length); @@ -458,14 +499,14 @@ static void prepare_next_write_item(pa_pstream *p) { p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(PA_FLAG_SHMREVOKE); p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl(p->write.current->block_id); - + } else { uint32_t flags; - int send_payload = 1; - - assert(p->write.current->type == PA_PSTREAM_ITEM_MEMBLOCK); - assert(p->write.current->chunk.memblock); - + pa_bool_t send_payload = TRUE; + + pa_assert(p->write.current->type == PA_PSTREAM_ITEM_MEMBLOCK); + pa_assert(p->write.current->chunk.memblock); + p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl(p->write.current->channel); p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl((uint32_t) (((uint64_t) p->write.current->offset) >> 32)); p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = htonl((uint32_t) ((uint64_t) p->write.current->offset)); @@ -476,7 +517,7 @@ static void prepare_next_write_item(pa_pstream *p) { uint32_t block_id, shm_id; size_t offset, length; - assert(p->export); + pa_assert(p->export); if (pa_memexport_put(p->export, p->write.current->chunk.memblock, @@ -484,15 +525,15 @@ static void prepare_next_write_item(pa_pstream *p) { &shm_id, &offset, &length) >= 0) { - + flags |= PA_FLAG_SHMDATA; - send_payload = 0; - + send_payload = FALSE; + p->write.shm_info[PA_PSTREAM_SHM_BLOCKID] = htonl(block_id); p->write.shm_info[PA_PSTREAM_SHM_SHMID] = htonl(shm_id); p->write.shm_info[PA_PSTREAM_SHM_INDEX] = htonl((uint32_t) (offset + p->write.current->chunk.index)); p->write.shm_info[PA_PSTREAM_SHM_LENGTH] = htonl((uint32_t) p->write.current->chunk.length); - + p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(sizeof(p->write.shm_info)); p->write.data = p->write.shm_info; } @@ -506,7 +547,7 @@ static void prepare_next_write_item(pa_pstream *p) { pa_memblock_ref(p->write.memchunk.memblock); p->write.data = NULL; } - + p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(flags); } @@ -521,9 +562,9 @@ static int do_write(pa_pstream *p) { size_t l; ssize_t r; pa_memblock *release_memblock = NULL; - - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); + + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); if (!p->write.current) prepare_next_write_item(p); @@ -535,7 +576,7 @@ 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 || p->write.memchunk.memblock); + pa_assert(p->write.data || p->write.memchunk.memblock); if (p->write.data) d = p->write.data; @@ -543,20 +584,20 @@ static int do_write(pa_pstream *p) { d = (uint8_t*) pa_memblock_acquire(p->write.memchunk.memblock) + p->write.memchunk.index; release_memblock = p->write.memchunk.memblock; } - + 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); } - assert(l > 0); - + pa_assert(l > 0); + #ifdef HAVE_CREDS if (p->send_creds_now) { if ((r = pa_iochannel_write_with_creds(p->io, d, l, &p->write_creds)) < 0) goto fail; - p->send_creds_now = 0; + p->send_creds_now = FALSE; } else #endif @@ -569,10 +610,15 @@ static int do_write(pa_pstream *p) { p->write.index += r; if (p->write.index >= PA_PSTREAM_DESCRIPTOR_SIZE + ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH])) { - assert(p->write.current); - item_free(p->write.current, (void *) 1); + pa_assert(p->write.current); + item_free(p->write.current, NULL); p->write.current = NULL; + if (p->write.memchunk.memblock) + pa_memblock_unref(p->write.memchunk.memblock); + + pa_memchunk_reset(&p->write.memchunk); + if (p->drain_callback && !pa_pstream_is_pending(p)) p->drain_callback(p, p->drain_callback_userdata); } @@ -583,24 +629,23 @@ fail: if (release_memblock) pa_memblock_release(release_memblock); - + return -1; } static int do_read(pa_pstream *p) { void *d; - size_t l; + size_t l; ssize_t r; pa_memblock *release_memblock = NULL; - - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); - + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + if (p->read.index < PA_PSTREAM_DESCRIPTOR_SIZE) { d = (uint8_t*) p->read.descriptor + p->read.index; l = PA_PSTREAM_DESCRIPTOR_SIZE - p->read.index; } else { - assert(p->read.data || p->read.memblock); + pa_assert(p->read.data || p->read.memblock); if (p->read.data) d = p->read.data; @@ -608,15 +653,15 @@ static int do_read(pa_pstream *p) { 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); } #ifdef HAVE_CREDS { - int b = 0; - + pa_bool_t b = 0; + if ((r = pa_iochannel_read_with_creds(p->io, d, l, &p->read_creds, &b)) <= 0) goto fail; @@ -629,7 +674,7 @@ static int do_read(pa_pstream *p) { if (release_memblock) pa_memblock_release(release_memblock); - + p->read.index += r; if (p->read.index == PA_PSTREAM_DESCRIPTOR_SIZE) { @@ -638,97 +683,97 @@ static int do_read(pa_pstream *p) { flags = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]); - if (!p->import && (flags & PA_FLAG_SHMMASK) != 0) { + if (!p->use_shm && (flags & PA_FLAG_SHMMASK) != 0) { pa_log_warn("Recieved SHM frame on a socket where SHM is disabled."); return -1; } - + if (flags == PA_FLAG_SHMRELEASE) { /* This is a SHM memblock release frame with no payload */ /* pa_log("Got release frame for %u", ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])); */ - - assert(p->export); + + pa_assert(p->export); pa_memexport_process_release(p->export, ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])); goto frame_done; - + } else if (flags == PA_FLAG_SHMREVOKE) { /* This is a SHM memblock revoke frame with no payload */ /* pa_log("Got revoke frame for %u", ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])); */ - assert(p->import); + pa_assert(p->import); pa_memimport_process_revoke(p->import, ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])); goto frame_done; } length = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]); - - if (length > FRAME_SIZE_MAX_ALLOW) { - pa_log_warn("Recieved invalid frame size : %lu", (unsigned long) length); + + if (length > FRAME_SIZE_MAX_ALLOW || length <= 0) { + pa_log_warn("Recieved invalid frame size: %lu", (unsigned long) length); return -1; } - - assert(!p->read.packet && !p->read.memblock); + + pa_assert(!p->read.packet && !p->read.memblock); channel = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]); - + if (channel == (uint32_t) -1) { if (flags != 0) { pa_log_warn("Received packet frame with invalid flags value."); return -1; } - + /* Frame is a packet frame */ p->read.packet = pa_packet_new(length); p->read.data = p->read.packet->data; - + } else { if ((flags & PA_FLAG_SEEKMASK) > PA_SEEK_RELATIVE_END) { pa_log_warn("Received memblock frame with invalid seek mode."); return -1; } - + if ((flags & PA_FLAG_SHMMASK) == PA_FLAG_SHMDATA) { if (length != sizeof(p->read.shm_info)) { pa_log_warn("Recieved SHM memblock frame with Invalid frame length."); return -1; } - + /* Frame is a memblock frame referencing an SHM memblock */ p->read.data = p->read.shm_info; } else if ((flags & PA_FLAG_SHMMASK) == 0) { /* Frame is a memblock frame */ - + p->read.memblock = pa_memblock_new(p->mempool, length); p->read.data = NULL; } else { - + pa_log_warn("Recieved memblock frame with invalid flags value."); return -1; } } - + } else if (p->read.index > PA_PSTREAM_DESCRIPTOR_SIZE) { /* Frame payload available */ - + if (p->read.memblock && p->recieve_memblock_callback) { /* Is this memblock data? Than pass it to the user */ l = (p->read.index - r) < PA_PSTREAM_DESCRIPTOR_SIZE ? p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE : (size_t) r; - + if (l > 0) { pa_memchunk chunk; - + chunk.memblock = p->read.memblock; chunk.index = p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE - l; chunk.length = l; @@ -739,7 +784,7 @@ static int do_read(pa_pstream *p) { offset = (int64_t) ( (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) | (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO])))); - + p->recieve_memblock_callback( p, ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]), @@ -758,14 +803,14 @@ static int do_read(pa_pstream *p) { /* Frame complete */ if (p->read.index >= ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) + PA_PSTREAM_DESCRIPTOR_SIZE) { - + if (p->read.memblock) { /* This was a memblock frame. We can unref the memblock now */ pa_memblock_unref(p->read.memblock); } else if (p->read.packet) { - + if (p->recieve_packet_callback) #ifdef HAVE_CREDS p->recieve_packet_callback(p, p->read.packet, p->read_creds_valid ? &p->read_creds : NULL, p->recieve_packet_callback_userdata); @@ -776,10 +821,10 @@ static int do_read(pa_pstream *p) { pa_packet_unref(p->read.packet); } else { pa_memblock *b; - - assert((ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]) & PA_FLAG_SHMMASK) == PA_FLAG_SHMDATA); - assert(p->import); + pa_assert((ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]) & PA_FLAG_SHMMASK) == PA_FLAG_SHMDATA); + + pa_assert(p->import); if (!(b = pa_memimport_get(p->import, ntohl(p->read.shm_info[PA_PSTREAM_SHM_BLOCKID]), @@ -794,7 +839,7 @@ static int do_read(pa_pstream *p) { if (p->recieve_memblock_callback) { int64_t offset; pa_memchunk chunk; - + chunk.memblock = b; chunk.index = 0; chunk.length = pa_memblock_get_length(b); @@ -802,7 +847,7 @@ static int do_read(pa_pstream *p) { offset = (int64_t) ( (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) | (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO])))); - + p->recieve_memblock_callback( p, ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]), @@ -828,7 +873,7 @@ frame_done: p->read.data = NULL; #ifdef HAVE_CREDS - p->read_creds_valid = 0; + p->read_creds_valid = FALSE; #endif return 0; @@ -841,75 +886,90 @@ fail: } 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_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + p->die_callback = cb; p->die_callback_userdata = userdata; } 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_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); p->drain_callback = cb; p->drain_callback_userdata = userdata; } 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_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); p->recieve_packet_callback = cb; p->recieve_packet_callback_userdata = userdata; } 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_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); p->recieve_memblock_callback = cb; p->recieve_memblock_callback_userdata = userdata; } -int pa_pstream_is_pending(pa_pstream *p) { - int b; +void pa_pstream_set_release_callback(pa_pstream *p, pa_pstream_block_id_cb_t cb, void *userdata) { + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); + p->release_callback = cb; + p->release_callback_userdata = userdata; +} + +void pa_pstream_set_revoke_callback(pa_pstream *p, pa_pstream_block_id_cb_t cb, void *userdata) { + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + + p->release_callback = cb; + p->release_callback_userdata = userdata; +} - pa_mutex_lock(p->mutex); +pa_bool_t pa_pstream_is_pending(pa_pstream *p) { + pa_bool_t b; + + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); if (p->dead) - b = 0; + b = FALSE; else b = p->write.current || !pa_queue_is_empty(p->send_queue); - pa_mutex_unlock(p->mutex); - return b; } void pa_pstream_unref(pa_pstream*p) { - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); if (PA_REFCNT_DEC(p) <= 0) pstream_free(p); } pa_pstream* pa_pstream_ref(pa_pstream*p) { - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); - + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + PA_REFCNT_INC(p); return p; } -void pa_pstream_close(pa_pstream *p) { - assert(p); +void pa_pstream_unlink(pa_pstream *p) { + pa_assert(p); + + if (p->dead) + return; - p->dead = 1; + p->dead = TRUE; if (p->import) { pa_memimport_free(p->import); @@ -926,20 +986,25 @@ 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; } -void pa_pstream_use_shm(pa_pstream *p, int enable) { - assert(p); - assert(PA_REFCNT_VALUE(p) > 0); +void pa_pstream_enable_shm(pa_pstream *p, pa_bool_t enable) { + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); p->use_shm = enable; if (enable) { - + if (!p->export) p->export = pa_memexport_new(p->mempool, memexport_revoke_cb, p); @@ -951,3 +1016,10 @@ void pa_pstream_use_shm(pa_pstream *p, int enable) { } } } + +pa_bool_t pa_pstream_get_shm(pa_pstream *p) { + pa_assert(p); + pa_assert(PA_REFCNT_VALUE(p) > 0); + + return p->use_shm; +} |