-/* $Id$ */
-
/***
This file is part of PulseAudio.
#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/refcnt.h>
+#include <pulsecore/flist.h>
+#include <pulsecore/macro.h>
#include "pstream.h"
#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 {
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
pa_queue *send_queue;
- int dead;
+ pa_bool_t dead;
struct {
pa_pstream_descriptor descriptor;
size_t index;
} read;
- int use_shm;
+ pa_bool_t use_shm;
pa_memimport *import;
pa_memexport *export;
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
};
static void defer_callback(pa_mainloop_api *m, pa_defer_event *e, void*userdata) {
pa_pstream *p = userdata;
-
+
pa_assert(p);
pa_assert(PA_REFCNT_VALUE(p) > 0);
pa_assert(p->defer_event == e);
pa_assert(p->mainloop == m);
-
+
do_something(p);
}
PA_REFCNT_INIT(p);
p->io = io;
pa_iochannel_set_callback(io, io_callback, p);
- p->dead = 0;
+ 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();
- pa_assert(p->send_queue);
p->write.current = NULL;
p->write.index = 0;
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, void *q) {
struct item_info *i = item;
pa_assert(i);
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) {
if (p->write.current)
item_free(p->write.current, NULL);
+ if (p->write.memchunk.memblock)
+ pa_memblock_unref(p->write.memchunk.memblock);
+
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);
-
pa_xfree(p);
}
if (p->dead)
return;
- 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_PACKET;
i->packet = pa_packet_ref(packet);
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;
+ size_t bsm;
pa_assert(p);
pa_assert(PA_REFCNT_VALUE(p) > 0);
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->offset = offset;
i->seek_mode = seek_mode;
#ifdef HAVE_CREDS
- i->with_creds = 0;
+ i->with_creds = FALSE;
#endif
pa_queue_push(p->send_queue, i);
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);
/* 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_queue_push(p->send_queue, item);
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;
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_queue_push(p->send_queue, item);
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);
+
+ 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) {
pa_assert(p);
pa_assert(PA_REFCNT_VALUE(p) > 0);
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);
+ p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl((uint32_t) p->write.current->packet->length);
} else if (p->write.current->type == PA_PSTREAM_ITEM_SHMRELEASE) {
} else {
uint32_t flags;
- int send_payload = 1;
+ 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_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));
- flags = p->write.current->seek_mode & PA_FLAG_SEEKMASK;
+ flags = (uint32_t) (p->write.current->seek_mode & PA_FLAG_SEEKMASK);
if (p->use_shm) {
uint32_t block_id, shm_id;
&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);
}
if (send_payload) {
- p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->chunk.length);
+ p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl((uint32_t) p->write.current->chunk.length);
p->write.memchunk = p->write.current->chunk;
pa_memblock_ref(p->write.memchunk.memblock);
p->write.data = NULL;
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
if (release_memblock)
pa_memblock_release(release_memblock);
- p->write.index += r;
+ p->write.index += (size_t) r;
if (p->write.index >= PA_PSTREAM_DESCRIPTOR_SIZE + ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH])) {
pa_assert(p->write.current);
- item_free(p->write.current, (void *) 1);
+ 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);
}
#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;
if (release_memblock)
pa_memblock_release(release_memblock);
- p->read.index += r;
+ p->read.index += (size_t) r;
if (p->read.index == PA_PSTREAM_DESCRIPTOR_SIZE) {
uint32_t flags, length, channel;
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;
+ l = (p->read.index - (size_t) r) < PA_PSTREAM_DESCRIPTOR_SIZE ? (size_t) (p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE) : (size_t) r;
if (l > 0) {
pa_memchunk chunk;
ntohl(p->read.shm_info[PA_PSTREAM_SHM_INDEX]),
ntohl(p->read.shm_info[PA_PSTREAM_SHM_LENGTH])))) {
- pa_log_warn("Failed to import memory block.");
- return -1;
+ if (pa_log_ratelimit())
+ pa_log_debug("Failed to import memory block.");
}
if (p->recieve_memblock_callback) {
chunk.memblock = b;
chunk.index = 0;
- chunk.length = pa_memblock_get_length(b);
+ chunk.length = b ? pa_memblock_get_length(b) : ntohl(p->read.shm_info[PA_PSTREAM_SHM_LENGTH]);
offset = (int64_t) (
(((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) |
p->recieve_memblock_callback_userdata);
}
- pa_memblock_unref(b);
+ if (b)
+ pa_memblock_unref(b);
}
goto frame_done;
p->read.data = NULL;
#ifdef HAVE_CREDS
- p->read_creds_valid = 0;
+ p->read_creds_valid = FALSE;
#endif
return 0;
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);
+
+ 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_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);
+ b = p->write.current || !pa_queue_isempty(p->send_queue);
return b;
}
if (p->dead)
return;
-
- p->dead = 1;
+
+ p->dead = TRUE;
if (p->import) {
pa_memimport_free(p->import);
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_memblock_callback = NULL;
}
-void pa_pstream_use_shm(pa_pstream *p, int enable) {
+void pa_pstream_enable_shm(pa_pstream *p, pa_bool_t enable) {
pa_assert(p);
pa_assert(PA_REFCNT_VALUE(p) > 0);
}
}
}
+
+pa_bool_t pa_pstream_get_shm(pa_pstream *p) {
+ pa_assert(p);
+ pa_assert(PA_REFCNT_VALUE(p) > 0);
+
+ return p->use_shm;
+}