diff options
| author | Lennart Poettering <lennart@poettering.net> | 2006-02-20 04:05:16 +0000 |
|---|---|---|
| committer | Lennart Poettering <lennart@poettering.net> | 2006-02-20 04:05:16 +0000 |
| commit | 304449002cbc84fdcf235b5dfaec891278dd7085 (patch) | |
| tree | 2a2d00e34d5c620835b76a0d6f7890a1d3e9fb97 | |
| parent | 0876b1ba82ea9c988df90ca98d202765ac697313 (diff) | |
1) Add flexible seeking support (including absolute) for memory block queues and playback streams
2) Add support to synchronize multiple playback streams
3) add two tests for 1) and 2)
4) s/PA_ERROR/PA_ERR/
5) s/PA_ERROR_OK/PA_OK/
6) update simple API to deal properly with new peek/drop recording API
7) add beginnings of proper validity checking on API calls in client libs (needs to be extended)
8) report playback buffer overflows/underflows to the client
9) move client side recording mcalign stuff into the memblockq
10) create typedefs for a bunch of API callback prototypes
11) simplify handling of HUP poll() events
Yes, i know, it's usually better to commit a lot of small patches instead of a
single big one. In this case however, this would have contradicted the other
rule: never commit broken or incomplete stuff.
*** This stuff needs a lot of additional testing! ***
git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@511 fefdeb5f-60dc-0310-8127-8f9354f1896f
38 files changed, 1939 insertions, 752 deletions
diff --git a/src/Makefile.am b/src/Makefile.am index f973dc46..4cf61f27 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -180,7 +180,9 @@ noinst_PROGRAMS = \ pacat-simple \ parec-simple \ strlist-test \ - voltest + voltest \ + memblockq-test \ + sync-playback if HAVE_SIGXCPU noinst_PROGRAMS += \ @@ -248,6 +250,24 @@ mainloop_test_glib12_CFLAGS = $(mainloop_test_CFLAGS) $(GLIB12_CFLAGS) -DGLIB_MA mainloop_test_glib12_LDADD = $(mainloop_test_LDADD) $(GLIB12_LIBS) libpolyp-mainloop-glib12-@PA_MAJORMINOR@.la mainloop_test_glib12_LDFLAGS = $(AM_LDFLAGS) $(BINLDFLAGS) +memblockq_test_SOURCES = \ + tests/memblockq-test.c \ + polypcore/memblockq.c \ + polypcore/log.c \ + polypcore/memblock.c \ + polypcore/xmalloc.c \ + polypcore/util.c \ + polypcore/mcalign.c \ + polypcore/memchunk.c +memblockq_test_CFLAGS = $(AM_CFLAGS) +memblockq_test_LDADD = $(AM_LDADD) +memblockq_test_LDFLAGS = $(AM_LDFLAGS) $(BINLDFLAGS) + +sync_playback_SOURCES = tests/sync-playback.c +sync_playback_LDADD = $(AM_LDADD) libpolyp-@PA_MAJORMINOR@.la libpolyp-mainloop-@PA_MAJORMINOR@.la +sync_playback_CFLAGS = $(AM_CFLAGS) +sync_playback_LDFLAGS = $(AM_LDFLAGS) $(BINLDFLAGS) + ################################### # Client library # ################################### diff --git a/src/modules/module-combine.c b/src/modules/module-combine.c index aabb8f28..750eca67 100644 --- a/src/modules/module-combine.c +++ b/src/modules/module-combine.c @@ -144,7 +144,7 @@ static void request_memblock(struct userdata *u) { return; for (o = u->outputs; o; o = o->next) - pa_memblockq_push_align(o->memblockq, &chunk, 0); + pa_memblockq_push_align(o->memblockq, &chunk); pa_memblock_unref(chunk.memblock); } @@ -212,7 +212,15 @@ static struct output *output_new(struct userdata *u, pa_sink *sink, int resample o->userdata = u; o->counter = 0; - o->memblockq = pa_memblockq_new(MEMBLOCKQ_MAXLENGTH, MEMBLOCKQ_MAXLENGTH, pa_frame_size(&u->sink->sample_spec), 0, 0, sink->core->memblock_stat); + o->memblockq = pa_memblockq_new( + 0, + MEMBLOCKQ_MAXLENGTH, + MEMBLOCKQ_MAXLENGTH, + pa_frame_size(&u->sink->sample_spec), + 1, + 0, + NULL, + sink->core->memblock_stat); snprintf(t, sizeof(t), "%s: output #%u", u->sink->name, u->n_outputs+1); if (!(o->sink_input = pa_sink_input_new(sink, __FILE__, t, &u->sink->sample_spec, &u->sink->channel_map, 1, resample_method))) diff --git a/src/modules/module-tunnel.c b/src/modules/module-tunnel.c index 61b9bb3b..67236588 100644 --- a/src/modules/module-tunnel.c +++ b/src/modules/module-tunnel.c @@ -214,7 +214,7 @@ static void send_bytes(struct userdata *u) { return; } - pa_pstream_send_memblock(u->pstream, u->channel, 0, &chunk); + pa_pstream_send_memblock(u->pstream, u->channel, 0, PA_SEEK_RELATIVE, &chunk); pa_memblock_unref(chunk.memblock); if (chunk.length > u->requested_bytes) @@ -442,7 +442,7 @@ static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, void *user } #ifndef TUNNEL_SINK -static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk, void *userdata) { +static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata) { struct userdata *u = userdata; assert(p && chunk && u); diff --git a/src/polyp/context.c b/src/polyp/context.c index 6b778562..eac5dd23 100644 --- a/src/polyp/context.c +++ b/src/polyp/context.c @@ -74,6 +74,8 @@ static const pa_pdispatch_callback command_table[PA_COMMAND_MAX] = { [PA_COMMAND_REQUEST] = pa_command_request, + [PA_COMMAND_OVERFLOW] = pa_command_overflow_or_underflow, + [PA_COMMAND_UNDERFLOW] = pa_command_overflow_or_underflow, [PA_COMMAND_PLAYBACK_STREAM_KILLED] = pa_command_stream_killed, [PA_COMMAND_RECORD_STREAM_KILLED] = pa_command_stream_killed, [PA_COMMAND_SUBSCRIBE_EVENT] = pa_command_subscribe_event @@ -109,9 +111,10 @@ pa_context *pa_context_new(pa_mainloop_api *mainloop, const char *name) { PA_LLIST_HEAD_INIT(pa_stream, c->streams); PA_LLIST_HEAD_INIT(pa_operation, c->operations); - c->error = PA_ERROR_OK; + c->error = PA_OK; c->state = PA_CONTEXT_UNCONNECTED; c->ctag = 0; + c->csyncid = 0; c->state_callback = NULL; c->state_userdata = NULL; @@ -234,14 +237,24 @@ void pa_context_set_state(pa_context *c, pa_context_state_t st) { void pa_context_fail(pa_context *c, int error) { assert(c); - c->error = error; + + pa_context_set_error(c, error); pa_context_set_state(c, PA_CONTEXT_FAILED); } +int pa_context_set_error(pa_context *c, int error) { + assert(error >= 0 && error < PA_ERR_MAX); + + if (c) + c->error = error; + + return error; +} + static void pstream_die_callback(pa_pstream *p, void *userdata) { pa_context *c = userdata; assert(p && c); - pa_context_fail(c, PA_ERROR_CONNECTIONTERMINATED); + pa_context_fail(c, PA_ERR_CONNECTIONTERMINATED); } static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, void *userdata) { @@ -252,34 +265,34 @@ static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, void *user if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) { pa_log(__FILE__": invalid packet.\n"); - pa_context_fail(c, PA_ERROR_PROTOCOL); + pa_context_fail(c, PA_ERR_PROTOCOL); } pa_context_unref(c); } -static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, PA_GCC_UNUSED uint32_t delta, const pa_memchunk *chunk, void *userdata) { +static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata) { pa_context *c = userdata; pa_stream *s; - assert(p && chunk && c && chunk->memblock && chunk->memblock->data); + + assert(p); + assert(chunk); + assert(chunk->memblock); + assert(chunk->length); + assert(c); pa_context_ref(c); if ((s = pa_dynarray_get(c->record_streams, channel))) { - pa_mcalign_push(s->mcalign, chunk); - for (;;) { - pa_memchunk t; - - if (pa_mcalign_pop(s->mcalign, &t) < 0) - break; + pa_memblockq_seek(s->record_memblockq, offset, seek); + pa_memblockq_push_align(s->record_memblockq, chunk); - assert(s->record_memblockq); - pa_memblockq_push(s->record_memblockq, &t, 0); - if (s->read_callback) - s->read_callback(s, pa_stream_readable_size(s), s->read_userdata); + if (s->read_callback) { + size_t l; - pa_memblock_unref(t.memblock); + if ((l = pa_memblockq_get_length(s->record_memblockq)) > 0) + s->read_callback(s, l, s->read_userdata); } } @@ -293,14 +306,14 @@ int pa_context_handle_error(pa_context *c, uint32_t command, pa_tagstruct *t) { assert(t); if (pa_tagstruct_getu32(t, &c->error) < 0) { - pa_context_fail(c, PA_ERROR_PROTOCOL); + pa_context_fail(c, PA_ERR_PROTOCOL); return -1; } } else if (command == PA_COMMAND_TIMEOUT) - c->error = PA_ERROR_TIMEOUT; + c->error = PA_ERR_TIMEOUT; else { - pa_context_fail(c, PA_ERROR_PROTOCOL); + pa_context_fail(c, PA_ERR_PROTOCOL); return -1; } @@ -316,7 +329,7 @@ static void setup_complete_callback(pa_pdispatch *pd, uint32_t command, uint32_t if (command != PA_COMMAND_REPLY) { if (pa_context_handle_error(c, command, t) < 0) - pa_context_fail(c, PA_ERROR_PROTOCOL); + pa_context_fail(c, PA_ERR_PROTOCOL); pa_context_fail(c, c->error); goto finish; @@ -368,7 +381,7 @@ static void setup_context(pa_context *c, pa_iochannel *io) { assert(c->pdispatch); if (!c->conf->cookie_valid) { - pa_context_fail(c, PA_ERROR_AUTHKEY); + pa_context_fail(c, PA_ERR_AUTHKEY); goto finish; } @@ -401,7 +414,7 @@ static int context_connect_spawn(pa_context *c) { if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) < 0) { pa_log(__FILE__": socketpair() failed: %s\n", strerror(errno)); - pa_context_fail(c, PA_ERROR_INTERNAL); + pa_context_fail(c, PA_ERR_INTERNAL); goto fail; } @@ -415,7 +428,7 @@ static int context_connect_spawn(pa_context *c) { if ((pid = fork()) < 0) { pa_log(__FILE__": fork() failed: %s\n", strerror(errno)); - pa_context_fail(c, PA_ERROR_INTERNAL); + pa_context_fail(c, PA_ERR_INTERNAL); if (c->spawn_api.postfork) c->spawn_api.postfork(); @@ -471,10 +484,10 @@ static int context_connect_spawn(pa_context *c) { if (r < 0) { pa_log(__FILE__": waitpid() failed: %s\n", strerror(errno)); - pa_context_fail(c, PA_ERROR_INTERNAL); + pa_context_fail(c, PA_ERR_INTERNAL); goto fail; } else if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) { - pa_context_fail(c, PA_ERROR_CONNECTIONREFUSED); + pa_context_fail(c, PA_ERR_CONNECTIONREFUSED); goto fail; } @@ -527,7 +540,7 @@ static int try_next_connection(pa_context *c) { } #endif - pa_context_fail(c, PA_ERROR_CONNECTIONREFUSED); + pa_context_fail(c, PA_ERR_CONNECTIONREFUSED); goto finish; } @@ -569,7 +582,7 @@ static void on_connection(pa_socket_client *client, pa_iochannel*io, void *userd goto finish; } - pa_context_fail(c, PA_ERROR_CONNECTIONREFUSED); + pa_context_fail(c, PA_ERR_CONNECTIONREFUSED); goto finish; } @@ -593,7 +606,7 @@ int pa_context_connect(pa_context *c, const char *server, int spawn, const pa_sp if (server) { if (!(c->server_list = pa_strlist_parse(server))) { - pa_context_fail(c, PA_ERROR_INVALIDSERVER); + pa_context_fail(c, PA_ERR_INVALIDSERVER); goto finish; } } else { @@ -759,7 +772,7 @@ void pa_context_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_U success = 0; } else if (!pa_tagstruct_eof(t)) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } diff --git a/src/polyp/def.h b/src/polyp/def.h index ba35b31e..f8e0bed4 100644 --- a/src/polyp/def.h +++ b/src/polyp/def.h @@ -47,7 +47,7 @@ typedef enum pa_context_state { /** The state of a stream */ typedef enum pa_stream_state { - PA_STREAM_DISCONNECTED, /**< The stream is not yet connected to any sink or source */ + PA_STREAM_UNCONNECTED, /**< The stream is not yet connected to any sink or source */ PA_STREAM_CREATING, /**< The stream is being created */ PA_STREAM_READY, /**< The stream is established, you may pass audio data to it now */ PA_STREAM_FAILED, /**< An error occured that made the stream invalid */ @@ -103,22 +103,24 @@ typedef struct pa_buffer_attr { /** Error values as used by pa_context_errno(). Use pa_strerror() to convert these values to human readable strings */ enum { - PA_ERROR_OK, /**< No error */ - PA_ERROR_ACCESS, /**< Access failure */ - PA_ERROR_COMMAND, /**< Unknown command */ - PA_ERROR_INVALID, /**< Invalid argument */ - PA_ERROR_EXIST, /**< Entity exists */ - PA_ERROR_NOENTITY, /**< No such entity */ - PA_ERROR_CONNECTIONREFUSED, /**< Connection refused */ - PA_ERROR_PROTOCOL, /**< Protocol error */ - PA_ERROR_TIMEOUT, /**< Timeout */ - PA_ERROR_AUTHKEY, /**< No authorization key */ - PA_ERROR_INTERNAL, /**< Internal error */ - PA_ERROR_CONNECTIONTERMINATED, /**< Connection terminated */ - PA_ERROR_KILLED, /**< Entity killed */ - PA_ERROR_INVALIDSERVER, /**< Invalid server */ - PA_ERROR_INITFAILED, /**< Module initialization failed */ - PA_ERROR_MAX /**< Not really an error but the first invalid error code */ + PA_OK = 0, /**< No error */ + PA_ERR_ACCESS, /**< Access failure */ + PA_ERR_COMMAND, /**< Unknown command */ + PA_ERR_INVALID, /**< Invalid argument */ + PA_ERR_EXIST, /**< Entity exists */ + PA_ERR_NOENTITY, /**< No such entity */ + PA_ERR_CONNECTIONREFUSED, /**< Connection refused */ + PA_ERR_PROTOCOL, /**< Protocol error */ + PA_ERR_TIMEOUT, /**< Timeout */ + PA_ERR_AUTHKEY, /**< No authorization key */ + PA_ERR_INTERNAL, /**< Internal error */ + PA_ERR_CONNECTIONTERMINATED, /**< Connection terminated */ + PA_ERR_KILLED, /**< Entity killed */ + PA_ERR_INVALIDSERVER, /**< Invalid server */ + PA_ERR_MODINITFAILED, /**< Module initialization failed */ + PA_ERR_BADSTATE, /**< Bad state */ + PA_ERR_NODATA, /**< No data */ + PA_ERR_MAX /**< Not really an error but the first invalid error code */ }; /** Subscription event mask, as used by pa_context_subscribe() */ @@ -208,6 +210,15 @@ typedef struct pa_spawn_api { * passed to the new process. */ } pa_spawn_api; +/** Seek type \since 0.8*/ +typedef enum pa_seek_mode { + PA_SEEK_RELATIVE = 0, /**< Seek relatively to the write index */ + PA_SEEK_ABSOLUTE = 1, /**< Seek relatively to the start of the buffer queue */ + PA_SEEK_RELATIVE_ON_READ = 2, /**< Seek relatively to the read index */ + PA_SEEK_RELATIVE_END = 3, /**< Seek relatively to the current end of the buffer queue */ +} pa_seek_mode_t; + + PA_C_DECL_END #endif diff --git a/src/polyp/error.c b/src/polyp/error.c index ece77bf2..eff37cc8 100644 --- a/src/polyp/error.c +++ b/src/polyp/error.c @@ -30,25 +30,28 @@ #include "error.h" -static const char* const errortab[PA_ERROR_MAX] = { - [PA_ERROR_OK] = "OK", - [PA_ERROR_ACCESS] = "Access denied", - [PA_ERROR_COMMAND] = "Unknown command", - [PA_ERROR_INVALID] = "Invalid argument", - [PA_ERROR_EXIST] = "Entity exists", - [PA_ERROR_NOENTITY] = "No such entity", - [PA_ERROR_CONNECTIONREFUSED] = "Connection refused", - [PA_ERROR_PROTOCOL] = "Protocol error", - [PA_ERROR_TIMEOUT] = "Timeout", - [PA_ERROR_AUTHKEY] = "No authorization key", - [PA_ERROR_INTERNAL] = "Internal error", - [PA_ERROR_CONNECTIONTERMINATED] = "Connection terminated", - [PA_ERROR_KILLED] = "Entity killed", - [PA_ERROR_INVALIDSERVER] = "Invalid server", +static const char* const errortab[PA_ERR_MAX] = { + [PA_OK] = "OK", + [PA_ERR_ACCESS] = "Access denied", + [PA_ERR_COMMAND] = "Unknown command", + [PA_ERR_INVALID] = "Invalid argument", + [PA_ERR_EXIST] = "Entity exists", + [PA_ERR_NOENTITY] = "No such entity", + [PA_ERR_CONNECTIONREFUSED] = "Connection refused", + [PA_ERR_PROTOCOL] = "Protocol error", + [PA_ERR_TIMEOUT] = "Timeout", + [PA_ERR_AUTHKEY] = "No authorization key", + [PA_ERR_INTERNAL] = "Internal error", + [PA_ERR_CONNECTIONTERMINATED] = "Connection terminated", + [PA_ERR_KILLED] = "Entity killed", + [PA_ERR_INVALIDSERVER] = "Invalid server", + [PA_ERR_MODINITFAILED] = "Module initalization failed", + [PA_ERR_BADSTATE] = "Bad state", + [PA_ERR_NODATA] = "No data", }; -const char*pa_strerror(uint32_t error) { - if (error >= PA_ERROR_MAX) +const char*pa_strerror(int error) { + if (error < 0 || error >= PA_ERR_MAX) return NULL; return errortab[error]; diff --git a/src/polyp/error.h b/src/polyp/error.h index ff2507b2..9856c1af 100644 --- a/src/polyp/error.h +++ b/src/polyp/error.h @@ -31,7 +31,7 @@ PA_C_DECL_BEGIN /** Return a human readable error message for the specified numeric error code */ -const char* pa_strerror(uint32_t error); +const char* pa_strerror(int error); PA_C_DECL_END diff --git a/src/polyp/internal.h b/src/polyp/internal.h index 7f4d38ac..578969ee 100644 --- a/src/polyp/internal.h +++ b/src/polyp/internal.h @@ -54,8 +54,9 @@ struct pa_context { pa_dynarray *record_streams, *playback_streams; PA_LLIST_HEAD(pa_stream, streams); PA_LLIST_HEAD(pa_operation, operations); - + uint32_t ctag; + uint32_t csyncid; uint32_t error; pa_context_state_t state; @@ -90,6 +91,7 @@ struct pa_stream { pa_sample_spec sample_spec; pa_channel_map channel_map; uint32_t channel; + uint32_t syncid; int channel_valid; uint32_t device_index; pa_stream_direction_t direction; @@ -98,7 +100,6 @@ struct pa_stream { pa_usec_t previous_time; pa_usec_t previous_ipol_time; pa_stream_state_t state; - pa_mcalign *mcalign; pa_memchunk peek_memchunk; pa_memblockq *record_memblockq; @@ -110,14 +111,20 @@ struct pa_stream { pa_time_event *ipol_event; int ipol_requested; - void (*state_callback)(pa_stream*c, void *userdata); + pa_stream_notify_cb_t state_callback; void *state_userdata; - void (*read_callback)(pa_stream *p, size_t length, void *userdata); + pa_stream_request_cb_t read_callback; void *read_userdata; - void (*write_callback)(pa_stream *p, size_t length, void *userdata); + pa_stream_request_cb_t write_callback; void *write_userdata; + + pa_stream_notify_cb_t overflow_callback; + void *overflow_userdata; + + pa_stream_notify_cb_t underflow_callback; + void *underflow_userdata; }; typedef void (*pa_operation_callback)(void); @@ -136,6 +143,7 @@ struct pa_operation { void pa_command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); void pa_command_subscribe_event(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); +void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); pa_operation *pa_operation_new(pa_context *c, pa_stream *s); void pa_operation_done(pa_operation *o); @@ -146,6 +154,7 @@ void pa_context_simple_ack_callback(pa_pdispatch *pd, uint32_t command, uint32_t void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); void pa_context_fail(pa_context *c, int error); +int pa_context_set_error(pa_context *c, int error); void pa_context_set_state(pa_context *c, pa_context_state_t st); int pa_context_handle_error(pa_context *c, uint32_t command, pa_tagstruct *t); pa_operation* pa_context_send_simple_command(pa_context *c, uint32_t command, void (*internal_callback)(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata), void (*cb)(void), void *userdata); @@ -154,5 +163,23 @@ void pa_stream_set_state(pa_stream *s, pa_stream_state_t st); void pa_stream_trash_ipol(pa_stream *s); +#define PA_CHECK_VALIDITY(context, expression, error) do { \ + if (!(expression)) \ + return -pa_context_set_error((context), (error)); \ +} while(0) + +#define PA_CHECK_VALIDITY_RETURN_NULL(context, expression, error) do { \ + if (!(expression)) { \ + pa_context_set_error((context), (error)); \ + return NULL; \ + } \ +} while(0) + +#define PA_CHECK_VALIDITY_RETURN_ANY(context, expression, error, value) do { \ + if (!(expression)) { \ + pa_context_set_error((context), (error)); \ + return value; \ + } \ +} while(0) #endif diff --git a/src/polyp/introspect.c b/src/polyp/introspect.c index 4af724b4..75ce3ff9 100644 --- a/src/polyp/introspect.c +++ b/src/polyp/introspect.c @@ -52,7 +52,7 @@ static void context_stat_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNU pa_tagstruct_getu32(t, &i.memblock_allocated_size) < 0 || pa_tagstruct_getu32(t, &i.scache_size) < 0 || !pa_tagstruct_eof(t)) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -92,7 +92,7 @@ static void context_get_server_info_callback(pa_pdispatch *pd, uint32_t command, pa_tagstruct_getu32(t, &i.cookie) < 0 || !pa_tagstruct_eof(t)) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -139,7 +139,7 @@ static void context_get_sink_info_callback(pa_pdispatch *pd, uint32_t command, P pa_tagstruct_get_usec(t, &i.latency) < 0 || pa_tagstruct_gets(t, &i.driver) < 0) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -234,7 +234,7 @@ static void context_get_source_info_callback(pa_pdispatch *pd, uint32_t command, pa_tagstruct_get_usec(t, &i.latency) < 0 || pa_tagstruct_gets(t, &i.driver) < 0) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -322,7 +322,7 @@ static void context_get_client_info_callback(pa_pdispatch *pd, uint32_t command, pa_tagstruct_gets(t, &i.name) < 0 || pa_tagstruct_getu32(t, &i.owner_module) < 0 || pa_tagstruct_gets(t, &i.driver) < 0 ) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -389,7 +389,7 @@ static void context_get_module_info_callback(pa_pdispatch *pd, uint32_t command, pa_tagstruct_gets(t, &i.argument) < 0 || pa_tagstruct_getu32(t, &i.n_used) < 0 || pa_tagstruct_get_boolean(t, &i.auto_unload) < 0) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -464,7 +464,7 @@ static void context_get_sink_input_info_callback(pa_pdispatch *pd, uint32_t comm pa_tagstruct_gets(t, &i.resample_method) < 0 || pa_tagstruct_gets(t, &i.driver) < 0) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -538,7 +538,7 @@ static void context_get_source_output_info_callback(pa_pdispatch *pd, uint32_t c pa_tagstruct_gets(t, &i.resample_method) < 0 || pa_tagstruct_gets(t, &i.driver) < 0) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -677,7 +677,7 @@ static void context_get_sample_info_callback(pa_pdispatch *pd, uint32_t command, pa_tagstruct_get_boolean(t, &i.lazy) < 0 || pa_tagstruct_gets(t, &i.filename) < 0) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -787,7 +787,7 @@ static void load_module_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUS } else if (pa_tagstruct_getu32(t, &idx) < 0 || !pa_tagstruct_eof(t)) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -848,7 +848,7 @@ static void context_get_autoload_info_callback(pa_pdispatch *pd, uint32_t comman pa_tagstruct_getu32(t, &i.type) < 0 || pa_tagstruct_gets(t, &i.module) < 0 || pa_tagstruct_gets(t, &i.argument) < 0) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } @@ -926,7 +926,7 @@ static void context_add_autoload_callback(pa_pdispatch *pd, uint32_t command, PA idx = PA_INVALID_INDEX; } else if (pa_tagstruct_getu32(t, &idx) || !pa_tagstruct_eof(t)) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } diff --git a/src/polyp/simple.c b/src/polyp/simple.c index e14cab2e..8a20c223 100644 --- a/src/polyp/simple.c +++ b/src/polyp/simple.c @@ -45,13 +45,11 @@ struct pa_simple { int dead; - void *read_data; + const void *read_data; size_t read_index, read_length; pa_usec_t latency; }; -static void read_callback(pa_stream *s, const void*data, size_t length, void *userdata); - static int check_error(pa_simple *p, int *rerror) { pa_context_state_t cst; pa_stream_state_t sst; @@ -92,7 +90,7 @@ static int iterate(pa_simple *p, int block, int *rerror) { do { if (pa_mainloop_iterate(p->mainloop, 1, NULL) < 0) { if (rerror) - *rerror = PA_ERROR_INTERNAL; + *rerror = PA_ERR_INTERNAL; return -1; } @@ -106,7 +104,7 @@ static int iterate(pa_simple *p, int block, int *rerror) { if (pa_mainloop_iterate(p->mainloop, 0, NULL) < 0) { if (rerror) - *rerror = PA_ERROR_INTERNAL; + *rerror = PA_ERR_INTERNAL; return -1; } @@ -128,7 +126,7 @@ pa_simple* pa_simple_new( int *rerror) { pa_simple *p; - int error = PA_ERROR_INTERNAL; + int error = PA_ERR_INTERNAL; assert(ss && (dir == PA_STREAM_PLAYBACK || dir == PA_STREAM_RECORD)); p = pa_xmalloc(sizeof(pa_simple)); @@ -157,7 +155,7 @@ pa_simple* pa_simple_new( goto fail; if (dir == PA_STREAM_PLAYBACK) - pa_stream_connect_playback(p->stream, dev, attr, 0, NULL); + pa_stream_connect_playback(p->stream, dev, attr, 0, NULL, NULL); else pa_stream_connect_record(p->stream, dev, attr, 0); @@ -167,8 +165,6 @@ pa_simple* pa_simple_new( goto fail; } - pa_stream_set_read_callback(p->stream, read_callback, p); - return p; fail: @@ -181,8 +177,6 @@ fail: void pa_simple_free(pa_simple *s) { assert(s); - pa_xfree(s->read_data); - if (s->stream) pa_stream_unref(s->stream); @@ -215,7 +209,7 @@ int pa_simple_write(pa_simple *p, const void*data, size_t length, int *rerror) { if (l > length) l = length; - pa_stream_write(p->stream, data, l, NULL, 0); + pa_stream_write(p->stream, data, l, NULL, 0, PA_SEEK_RELATIVE); data = (const uint8_t*) data + l; length -= l; } @@ -227,19 +221,6 @@ int pa_simple_write(pa_simple *p, const void*data, size_t length, int *rerror) { return 0; } -static void read_callback(pa_stream *s, const void*data, size_t length, void *userdata) { - pa_simple *p = userdata; - assert(s && data && length && p); - - if (p->read_data) { - pa_log(__FILE__": Buffer overflow, dropping incoming memory blocks.\n"); - pa_xfree(p->read_data); - } - - p->read_data = pa_xmemdup(data, p->read_length = length); - p->read_index = 0; -} - int pa_simple_read(pa_simple *p, void*data, size_t length, int *rerror) { assert(p && data && p->direction == PA_STREAM_RECORD); @@ -251,13 +232,18 @@ int pa_simple_read(pa_simple *p, void*data, size_t length, int *rerror) { } while (length > 0) { + + if (!p->read_data) + if (pa_stream_peek(p->stream, &p->read_data, &p->read_length) >= 0) + p->read_index = 0; + if (p->read_data) { size_t l = length; if (p->read_length <= l) l = p->read_length; - memcpy(data, (uint8_t*) p->read_data+p->read_index, l); + memcpy(data, (const uint8_t*) p->read_data+p->read_index, l); data = (uint8_t*) data + l; length -= l; @@ -266,8 +252,9 @@ int pa_simple_read(pa_simple *p, void*data, size_t length, int *rerror) { p->read_length -= l; if (!p->read_length) { - pa_xfree(p->read_data); + pa_stream_drop(p->stream); p->read_data = NULL; + p->read_length = 0; p->read_index = 0; } diff --git a/src/polyp/stream.c b/src/polyp/stream.c index 35de2d01..5497f0c4 100644 --- a/src/polyp/stream.c +++ b/src/polyp/stream.c @@ -28,6 +28,7 @@ #include <stdio.h> #include <string.h> +#include <polyp/def.h> #include <polypcore/xmalloc.h> #include <polypcore/pstream-util.h> #include <polypcore/util.h> @@ -39,14 +40,11 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) { pa_stream *s; + assert(c); - assert(ss); - - if (!pa_sample_spec_valid(ss)) - return NULL; - if (map && !pa_channel_map_valid(map)) - return NULL; + PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID); + PA_CHECK_VALIDITY_RETURN_NULL(c, !map || pa_channel_map_valid(map), PA_ERR_INVALID); s = pa_xnew(pa_stream, 1); s->ref = 1; @@ -59,6 +57,10 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec * s->write_userdata = NULL; s->state_callback = NULL; s->state_userdata = NULL; + s->overflow_callback = NULL; + s->overflow_userdata = NULL; + s->underflow_callback = NULL; + s->underflow_userdata = NULL; s->direction = PA_STREAM_NODIRECTION; s->name = pa_xstrdup(name); @@ -71,13 +73,13 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec * s->channel = 0; s->channel_valid = 0; + s->syncid = c->csyncid++; s->device_index = PA_INVALID_INDEX; s->requested_bytes = 0; - s->state = PA_STREAM_DISCONNECTED; + s->state = PA_STREAM_UNCONNECTED; memset(&s->buffer_attr, 0, sizeof(s->buffer_attr)); - s->mcalign = pa_mcalign_new(pa_frame_size(ss), c->memblock_stat); - + s->peek_memchunk.index = 0; s->peek_memchunk.length = 0; s->peek_memchunk.memblock = NULL; @@ -114,42 +116,52 @@ static void stream_free(pa_stream *s) { if (s->record_memblockq) pa_memblockq_free(s->record_memblockq); - pa_mcalign_free(s->mcalign); - pa_xfree(s->name); pa_xfree(s); } void pa_stream_unref(pa_stream *s) { - assert(s && s->ref >= 1); + assert(s); + assert(s->ref >= 1); if (--(s->ref) == 0) stream_free(s); } pa_stream* pa_stream_ref(pa_stream *s) { - assert(s && s->ref >= 1); + assert(s); + assert(s->ref >= 1); + s->ref++; return s; } pa_stream_state_t pa_stream_get_state(pa_stream *s) { - assert(s && s->ref >= 1); + assert(s); + assert(s->ref >= 1); + return s->state; } pa_context* pa_stream_get_context(pa_stream *s) { - assert(s && s->ref >= 1); + assert(s); + assert(s->ref >= 1); + return s->context; } uint32_t pa_stream_get_index(pa_stream *s) { - assert(s && s->ref >= 1); + assert(s); + assert(s->ref >= 1); + + PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX); + return s->device_index; } void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) { - assert(s && s->ref >= 1); + assert(s); + assert(s->ref >= 1); if (s->state == st) return; @@ -159,6 +171,8 @@ void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) { s->state = st; if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED) && s->context) { + /* Detach from context */ + if (s->channel_valid) pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL); @@ -182,14 +196,14 @@ void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED if (pa_tagstruct_getu32(t, &channel) < 0 || !pa_tagstruct_eof(t)) { - pa_context_fail(c, PA_ERROR_PROTOCOL); + pa_context_fail(c, PA_ERR_PROTOCOL); goto finish; } if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel))) goto finish; - c->error = PA_ERROR_KILLED; + c->error = PA_ERR_KILLED; pa_stream_set_state(s, PA_STREAM_FAILED); finish: @@ -207,24 +221,55 @@ void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32 if (pa_tagstruct_getu32(t, &channel) < 0 || pa_tagstruct_getu32(t, &bytes) < 0 || !pa_tagstruct_eof(t)) { - pa_context_fail(c, PA_ERROR_PROTOCOL); + pa_context_fail(c, PA_ERR_PROTOCOL); goto finish; } if (!(s = pa_dynarray_get(c->playback_streams, channel))) goto finish; - if (s->state != PA_STREAM_READY) - goto finish; + if (s->state == PA_STREAM_READY) { + s->requested_bytes += bytes; + + if (s->requested_bytes > 0 && s->write_callback) + s->write_callback(s, s->requested_bytes, s->write_userdata); + } - pa_stream_ref(s); +finish: + pa_context_unref(c); +} + +void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) { + pa_stream *s; + pa_context *c = userdata; + uint32_t channel; + + assert(pd); + assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW); + assert(t); + assert(c); + + pa_context_ref(c); + + if (pa_tagstruct_getu32(t, &channel) < 0 || + !pa_tagstruct_eof(t)) { + pa_context_fail(c, PA_ERR_PROTOCOL); + goto finish; + } - s->requested_bytes += bytes; + if (!(s = pa_dynarray_get(c->playback_streams, channel))) + goto finish; - if (s->requested_bytes && s->write_callback) - s->write_callback(s, s->requested_bytes, s->write_userdata); + if (s->state == PA_STREAM_READY) { - pa_stream_unref(s); + if (command == PA_COMMAND_OVERFLOW) { + if (s->overflow_callback) + s->overflow_callback(s, s->overflow_userdata); + } else if (command == PA_COMMAND_UNDERFLOW) { + if (s->underflow_callback) + s->underflow_callback(s, s->underflow_userdata); + } + } finish: pa_context_unref(c); @@ -270,14 +315,21 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->device_index) < 0) || ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &s->requested_bytes) < 0) || !pa_tagstruct_eof(t)) { - pa_context_fail(s->context, PA_ERROR_PROTOCOL); + pa_context_fail(s->context, PA_ERR_PROTOCOL); goto finish; } if (s->direction == PA_STREAM_RECORD) { assert(!s->record_memblockq); - s->record_memblockq = pa_memblockq_new(s->buffer_attr.maxlength, 0, - pa_frame_size(&s->sample_spec), 0, 0, s->context->memblock_stat); + s->record_memblockq = pa_memblockq_new( + 0, + s->buffer_attr.maxlength, + 0, + pa_frame_size(&s->sample_spec), + 1, + 0, + NULL, + s->context->memblock_stat); assert(s->record_memblockq); } @@ -303,13 +355,32 @@ finish: pa_stream_unref(s); } -static void create_stream(pa_stream *s, const char *dev, const pa_buffer_attr *attr, pa_stream_flags_t flags, const pa_cvolume *volume) { +static int create_stream( + pa_stream_direction_t direction, + pa_stream *s, + const char *dev, + const pa_buffer_attr *attr, + pa_stream_flags_t flags, + const pa_cvolume *volume, + pa_stream *sync_stream) { + pa_tagstruct *t; uint32_t tag; - assert(s && s->ref >= 1 && s->state == PA_STREAM_DISCONNECTED); + + assert(s); + assert(s->ref >= 1); + + PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY(s->context, (flags & ~(PA_STREAM_START_CORKED|PA_STREAM_INTERPOLATE_LATENCY)) == 0, PA_ERR_INVALID); + PA_CHECK_VALIDITY(s->context, direction == PA_STREAM_PLAYBACK || flags == 0, PA_ERR_INVALID); pa_stream_ref(s); + s->direction = direction; + + if (sync_stream) + s->syncid = sync_stream->syncid; + s->interpolate = !!(flags & PA_STREAM_INTERPOLATE_LATENCY); pa_stream_trash_ipol(s); @@ -336,25 +407,28 @@ static void create_stream(pa_stream *s, const char *dev, const pa_buffer_attr *a dev = s->context->conf->default_source; } - pa_tagstruct_put(t, - PA_TAG_U32, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM, - PA_TAG_U32, tag = s->context->ctag++, - PA_TAG_STRING, s->name, - PA_TAG_SAMPLE_SPEC, &s->sample_spec, - PA_TAG_CHANNEL_MAP, &s->channel_map, - PA_TAG_U32, PA_INVALID_INDEX, - PA_TAG_STRING, dev, - PA_TAG_U32, s->buffer_attr.maxlength, - PA_TAG_BOOLEAN, !!(flags & PA_STREAM_START_CORKED), - PA_TAG_INVALID); + pa_tagstruct_put( + t, + PA_TAG_U32, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM, + PA_TAG_U32, tag = s->context->ctag++, + PA_TAG_STRING, s->name, + PA_TAG_SAMPLE_SPEC, &s->sample_spec, + PA_TAG_CHANNEL_MAP, &s->channel_map, + PA_TAG_U32, PA_INVALID_INDEX, + PA_TAG_STRING, dev, + PA_TAG_U32, s->buffer_attr.maxlength, + PA_TAG_BOOLEAN, !!(flags & PA_STREAM_START_CORKED), + PA_TAG_INVALID); if (s->direction == PA_STREAM_PLAYBACK) { pa_cvolume cv; - pa_tagstruct_put(t, - PA_TAG_U32, s->buffer_attr.tlength, - PA_TAG_U32, s->buffer_attr.prebuf, - PA_TAG_U32, s->buffer_attr.minreq, - PA_TAG_INVALID); + pa_tagstruct_put( + t, + PA_TAG_U32, s->buffer_attr.tlength, + PA_TAG_U32, s->buffer_attr.prebuf, + PA_TAG_U32, s->buffer_attr.minreq, + PA_TAG_U32, s->syncid, + PA_TAG_INVALID); if (!volume) { pa_cvolume_reset(&cv, s->sample_spec.channels); @@ -369,23 +443,57 @@ static void create_stream(pa_stream *s, const char *dev, const pa_buffer_attr *a pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s); pa_stream_unref(s); + return 0; } -void pa_stream_connect_playback(pa_stream *s, const char *dev, const pa_buffer_attr *attr, pa_stream_flags_t flags, pa_cvolume *volume) { - assert(s && s->context->state == PA_CONTEXT_READY && s->ref >= 1); - s->direction = PA_STREAM_PLAYBACK; - create_stream(s, dev, attr, flags, volume); +int pa_stream_connect_playback( + pa_stream *s, + const char *dev, + const pa_buffer_attr *attr, + pa_stream_flags_t flags, + pa_cvolume *volume, + pa_stream *sync_stream) { + + assert(s); + assert(s->ref >= 1); + + return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream); } -void pa_stream_connect_record(pa_stream *s, const char *dev, const pa_buffer_attr *attr, pa_stream_flags_t flags) { - assert(s && s->context->state == PA_CONTEXT_READY && s->ref >= 1); - s->direction = PA_STREAM_RECORD; - create_stream(s, dev, attr, flags, 0); +int pa_stream_connect_record( + pa_stream *s, + const char *dev, + const pa_buffer_attr *attr, + pa_stream_flags_t flags) { + + assert(s); + assert(s->ref >= 1); + + return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL); } -void pa_stream_write(pa_stream *s, const void *data, size_t length, void (*free_cb)(void *p), size_t delta) { +int pa_stream_write( + pa_stream *s, + const void *data, + size_t length, + void (*free_cb)(void *p), + int64_t offset, + pa_seek_mode_t seek) { + pa_memchunk chunk; - assert(s && s->context && data && length && s->state == PA_STREAM_READY && s->ref >= 1); + + assert(s); + assert(s->ref >= 1); + assert(s->context); + assert(data); + + PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID); + PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID); + + if (length <= 0) + return 0; if (free_cb) { chunk.memblock = pa_memblock_new_user((void*) data, length, free_cb, 1, s->context->memblock_stat); @@ -398,7 +506,7 @@ void pa_stream_write(pa_stream *s, const void *data, size_t length, void (*free_ chunk.index = 0; chunk.length = length; - pa_pstream_send_memblock(s->context->pstream, s->channel, delta, &chunk); + pa_pstream_send_memblock(s->context->pstream, s->channel, offset, seek, &chunk); pa_memblock_unref(chunk.memblock); if (length < s->requested_bytes) @@ -407,72 +515,87 @@ void pa_stream_write(pa_stream *s, const void *data, size_t length, void (*free_ s->requested_bytes = 0; s->counter += length; + return 0; } -void pa_stream_peek(pa_stream *s, void **data, size_t *length) { - assert(s && s->record_memblockq && data && length && s->state == PA_STREAM_READY && s->ref >= 1); +int pa_stream_peek(pa_stream *s, const void **data, size_t *length) { + assert(s); + assert(s->ref >= 1); + assert(data); + assert(length); + PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE); + if (!s->peek_memchunk.memblock) { - *data = NULL; - *length = 0; - - if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) - return; - pa_memblockq_drop(s->record_memblockq, &s->peek_memchunk, s->peek_memchunk.length); + if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) { + *data = NULL; + *length = 0; + return 0; + } } - *data = (char*)s->peek_memchunk.memblock->data + s->peek_memchunk.index; + *data = (const char*) s->peek_memchunk.memblock->data + s->peek_memchunk.index; *length = s->peek_memchunk.length; + return 0; } -void pa_stream_drop(pa_stream *s) { - assert(s && s->peek_memchunk.memblock && s->state == PA_STREAM_READY && s->ref >= 1); - - s->counter += s->peek_memchunk.length; +int pa_stream_drop(pa_stream *s) { + assert(s); + assert(s->ref >= 1); + PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE); + + pa_memblockq_drop(s->record_memblockq, &s->peek_memchunk, s->peek_memchunk.length); + pa_memblock_unref(s->peek_memchunk.memblock); - s->peek_memchunk.length = 0; + s->peek_memchunk.index = 0; s->peek_memchunk.memblock = NULL; + + s->counter += s->peek_memchunk.length; + return 0; } size_t pa_stream_writable_size(pa_stream *s) { - assert(s && s->ref >= 1); - return s->state == PA_STREAM_READY ? s->requested_bytes : 0; + assert(s); + assert(s->ref >= 1); + + PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1); + PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE, (size_t) -1); + + return s->requested_bytes; } size_t pa_stream_readable_size(pa_stream *s) { - size_t sz; - - assert(s && s->ref >= 1); - - if (s->state != PA_STREAM_READY) - return 0; - - assert(s->record_memblockq); - - sz = (size_t)pa_memblockq_get_length(s->record_memblockq); + assert(s); + assert(s->ref >= 1); - if (s->peek_memchunk.memblock) - sz += s->peek_memchunk.length; + PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1); + PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1); - return sz; + return pa_memblockq_get_length(s->record_memblockq); } pa_operation * pa_stream_drain(pa_stream *s, void (*cb) (pa_stream*s, int success, void *userdata), void *userdata) { pa_operation *o; pa_tagstruct *t; uint32_t tag; - assert(s && s->ref >= 1 && s->state == PA_STREAM_READY); + + assert(s); + assert(s->ref >= 1); + + PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE); o = pa_operation_new(s->context, s); - assert(o); o->callback = (pa_operation_callback) cb; o->userdata = userdata; t = pa_tagstruct_new(NULL, 0); - assert(t); pa_tagstruct_putu32(t, PA_COMMAND_DRAIN_PLAYBACK_STREAM); pa_tagstruct_putu32(t, tag = s->context->ctag++); pa_tagstruct_putu32(t, s->channel); @@ -501,7 +624,7 @@ static void stream_get_latency_info_callback(pa_pdispatch *pd, uint32_t command, pa_tagstruct_get_timeval(t, &remote) < 0 || pa_tagstruct_getu64(t, &i.counter) < 0 || !pa_tagstruct_eof(t)) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } else { pa_gettimeofday(&now); @@ -549,15 +672,18 @@ pa_operation* pa_stream_get_latency_info(pa_stream *s, void (*cb)(pa_stream *p, pa_operation *o; pa_tagstruct *t; struct timeval now; - assert(s && s->direction != PA_STREAM_UPLOAD); + + assert(s); + assert(s->ref >= 1); + PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE); + o = pa_operation_new(s->context, s); - assert(o); o->callback = (pa_operation_callback) cb; o->userdata = userdata; t = pa_tagstruct_new(NULL, 0); - assert(t); pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY); pa_tagstruct_putu32(t, tag = s->context->ctag++); pa_tagstruct_putu32(t, s->channel); @@ -585,7 +711,7 @@ void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UN pa_stream_set_state(s, PA_STREAM_FAILED); goto finish; } else if (!pa_tagstruct_eof(t)) { - pa_context_fail(s->context, PA_ERROR_PROTOCOL); + pa_context_fail(s->context, PA_ERR_PROTOCOL); goto finish; } @@ -595,18 +721,19 @@ finish: pa_stream_unref(s); } -void pa_stream_disconnect(pa_stream *s) { +int pa_stream_disconnect(pa_stream *s) { pa_tagstruct *t; uint32_t tag; - assert(s && s->ref >= 1); - if (!s->channel_valid || !s->context->state == PA_CONTEXT_READY) - return; + assert(s); + assert(s->ref >= 1); + + PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE); + PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE); pa_stream_ref(s); t = pa_tagstruct_new(NULL, 0); - assert(t); pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM : (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM)); @@ -616,26 +743,49 @@ void pa_stream_disconnect(pa_stream *s) { pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s); pa_stream_unref(s); + return 0; } -void pa_stream_set_read_callback(pa_stream *s, void (*cb)(pa_stream *p, size_t length, void *userdata), void *userdata) { - assert(s && s->ref >= 1); +void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) { + assert(s); + assert(s->ref >= 1); + s->read_callback = cb; s->read_userdata = userdata; } -void pa_stream_set_write_callback(pa_stream *s, void (*cb)(pa_stream *p, size_t length, void *userdata), void *userdata) { - assert(s && s->ref >= 1); +void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) { + assert(s); + assert(s->ref >= 1); + s->write_callback = cb; s->write_userdata = userdata; } -void pa_stream_set_state_callback(pa_stream *s, void (*cb)(pa_stream *s, void *userdata), void *userdata) { - assert(s && s->ref >= 1); +void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) { + assert(s); + assert(s->ref >= 1); + s->state_callback = cb; s->state_userdata = userdata; } +void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) { + assert(s); + assert(s->ref >= 1); + + s->overflow_callback = cb; + s->overflow_userdata = userdata; +} + +void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) { + assert(s); + assert(s->ref >= 1); + + s->underflow_callback = cb; + s->underflow_userdata = userdata; +} + void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) { pa_operation *o = userdata; int success = 1; @@ -647,7 +797,7 @@ void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UN success = 0; } else if (!pa_tagstruct_eof(t)) { - pa_context_fail(o->context, PA_ERROR_PROTOCOL); + pa_context_fail(o->context, PA_ERR_PROTOCOL); goto finish; } diff --git a/src/polyp/stream.h b/src/polyp/stream.h index 9bbda436..ce535963 100644 --- a/src/polyp/stream.h +++ b/src/polyp/stream.h @@ -40,8 +40,27 @@ PA_C_DECL_BEGIN * An opaque stream for playback or recording */ typedef struct pa_stream pa_stream; +/** A generic callback for operation completion */ +typedef void (*pa_stream_success_cb_t) (pa_stream*s, int success, void *userdata); + +/** A generic free callback */ +typedef void (*pa_free_cb_t)(void *p); + +/** A generic request callback */ +typedef void (*pa_stream_request_cb_t)(pa_stream *p, size_t length, void *userdata); + +/** A generic notification callback */ +typedef void (*pa_stream_notify_cb_t)(pa_stream *p, void *userdata); + +/** Callback prototype for pa_stream_get_latency_info() */ +typedef void (*pa_stream_get_latency_info_cb_t)(pa_stream *p, const pa_latency_info *i, void *userdata); + /** Create a new, unconnected stream with the specified name and sample type */ -pa_stream* pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map); +pa_stream* pa_stream_new( + pa_context *c, + const char *name, + const pa_sample_spec *ss, + const pa_channel_map *map); /** Decrease the reference counter by one */ void pa_stream_unref(pa_stream *s); @@ -59,108 +78,101 @@ pa_context* pa_stream_get_context(pa_stream *p); uint32_t pa_stream_get_index(pa_stream *s); /** Connect the stream to a sink */ -void pa_stream_connect_playback( - pa_stream *s, - const char *dev, - const pa_buffer_attr *attr, - pa_stream_flags_t flags, - pa_cvolume *volume); +int pa_stream_connect_playback( + pa_stream *s /**< The stream to connect to a sink */, + const char *dev /**< Name of the sink to connect to, or NULL for default */ , + const pa_buffer_attr *attr /**< Buffering attributes, or NULL for default */, + pa_stream_flags_t flags /**< Additional flags, or 0 for default */, + pa_cvolume *volume /**< Initial volume, or NULL for default */, + pa_stream *sync_stream /**< Synchronize this stream with the specified one, or NULL for a standalone stream*/); /** Connect the stream to a source */ -void pa_stream_connect_record( - pa_stream *s, - const char *dev, - const pa_buffer_attr *attr, - pa_stream_flags_t flags); +int pa_stream_connect_record( + pa_stream *s, + const char *dev, + const pa_buffer_attr *attr, + pa_stream_flags_t flags); /** Disconnect a stream from a source/sink */ -void pa_stream_disconnect(pa_stream *s); +int pa_stream_disconnect(pa_stream *s); /** Write some data to the server (for playback sinks), if free_cb is * non-NULL this routine is called when all data has been written out * and an internal reference to the specified data is kept, the data * is not copied. If NULL, the data is copied into an internal - * buffer. */ -void pa_stream_write(pa_stream *p /**< The stream to use */, - const void *data /**< The data to write */, - size_t length /**< The length of the data to write */, - void (*free_cb)(void *p) /**< A cleanup routine for the data or NULL to request an internal copy */, - size_t delta /**< Drop this many - bytes in the playback - buffer before writing - this data. Use - (size_t) -1 for - clearing the whole - playback - buffer. Normally you - will specify 0 here, - i.e. append to the - playback buffer. If - the value given here - is greater than the - buffered data length - the buffer is cleared - and the data is - written to the - buffer's start. This - value is ignored on - upload streams. */); - -/** Read the next fragment from the buffer (for capture sources). + * buffer. The client my freely seek around in the output buffer. For + * most applications passing 0 and PA_SEEK_RELATIVE as arguments for + * offset and seek should be useful.*/ +int pa_stream_write( + pa_stream *p /**< The stream to use */, + const void *data /**< The data to write */, + size_t length /**< The length of the data to write */, + pa_free_cb_t free_cb /**< A cleanup routine for the data or NULL to request an internal copy */, + int64_t offset, /**< Offset for seeking, must be 0 for upload streams */ + pa_seek_mode_t seek /**< Seek mode, must be PA_SEEK_RELATIVE for upload streams */); + +/** Read the next fragment from the buffer (for recording). * data will point to the actual data and length will contain the size * of the data in bytes (which can be less than a complete framgnet). - * Use pa_stream_drop() to actually remove the data from the buffer. - * \since 0.8 */ -void pa_stream_peek(pa_stream *p /**< The stream to use */, - void **data /**< Pointer to pointer that will point to data */, - size_t *length /**< The length of the data read */); + * Use pa_stream_drop() to actually remove the data from the + * buffer. If no data is available will return a NULL pointer \since 0.8 */ +int pa_stream_peek( + pa_stream *p /**< The stream to use */, + const void **data /**< Pointer to pointer that will point to data */, + size_t *length /**< The length of the data read */); /** Remove the current fragment. It is invalid to do this without first * calling pa_stream_peek(). \since 0.8 */ -void pa_stream_drop(pa_stream *p); +int pa_stream_drop(pa_stream *p); -/** Return the amount of bytes that may be written using pa_stream_write() */ +/** Return the nember of bytes that may be written using pa_stream_write() */ size_t pa_stream_writable_size(pa_stream *p); -/** Return the ammount of bytes that may be read using pa_stream_read() \since 0.8 */ +/** Return the number of bytes that may be read using pa_stream_read() \since 0.8 */ size_t pa_stream_readable_size(pa_stream *p); -/** Drain a playback stream */ -pa_operation* pa_stream_drain(pa_stream *s, void (*cb) (pa_stream*s, int success, void *userdata), void *userdata); +/** Drain a playback stream. Use this for notification when the buffer is empty */ +pa_operation* pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata); /** Get the playback latency of a stream */ -pa_operation* pa_stream_get_latency_info(pa_stream *p, void (*cb)(pa_stream *p, const pa_latency_info *i, void *userdata), void *userdata); +pa_operation* pa_stream_get_latency_info(pa_stream *p, pa_stream_get_latency_info_cb_t cby, void *userdata); /** Set the callback function that is called whenever the state of the stream changes */ -void pa_stream_set_state_callback(pa_stream *s, void (*cb)(pa_stream *s, void *userdata), void *userdata); +void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata); /** Set the callback function that is called when new data may be * written to the stream. */ -void pa_stream_set_write_callback(pa_stream *p, void (*cb)(pa_stream *p, size_t length, void *userdata), void *userdata); +void pa_stream_set_write_callback(pa_stream *p, pa_stream_request_cb_t cb, void *userdata); /** Set the callback function that is called when new data is available from the stream. - * Return the number of bytes read. \since 0.8 - */ -void pa_stream_set_read_callback(pa_stream *p, void (*cb)(pa_stream *p, size_t length, void *userdata), void *userdata); + * Return the number of bytes read. \since 0.8 */ +void pa_stream_set_read_callback(pa_stream *p, pa_stream_request_cb_t cb, void *userdata); + +/** Set the callback function that is called when a buffer overflow happens. (Only for playback streams) \since 0.8 */ +void pa_stream_set_overflow_callback(pa_stream *p, pa_stream_notify_cb_t cb, void *userdata); + +/** Set the callback function that is called when a buffer underflow happens. (Only for playback streams) \since 0.8 */ +void pa_stream_set_underflow_callback(pa_stream *p, pa_stream_notify_cb_t cb, void *userdata); /** Pause (or resume) playback of this stream temporarily. Available on both playback and recording streams. \since 0.3 */ -pa_operation* pa_stream_cork(pa_stream *s, int b, void (*cb) (pa_stream*s, int success, void *userdata), void *userdata); +pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, void *userdata); /** Flush the playback buffer of this stream. Most of the time you're * better off using the parameter delta of pa_stream_write() instead of this * function. Available on both playback and recording streams. \since 0.3 */ -pa_operation* pa_stream_flush(pa_stream *s, void (*cb)(pa_stream *s, int success, void *userdata), void *userdata); +pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata); -/** Reenable prebuffering. Available for playback streams only. \since 0.6 */ -pa_operation* pa_stream_prebuf(pa_stream *s, void (*cb)(pa_stream *s, int success, void *userdata), void *userdata); +/** Reenable prebuffering as specified in the pa_buffer_attr + * structure. Available for playback streams only. \since 0.6 */ +pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata); /** Request immediate start of playback on this stream. This disables - * prebuffering as specified in the pa_buffer_attr structure. Available for playback streams only. \since - * 0.3 */ -pa_operation* pa_stream_trigger(pa_stream *s, void (*cb)(pa_stream *s, int success, void *userdata), void *userdata); + * prebuffering as specified in the pa_buffer_attr + * structure, temporarily. Available for playback streams only. \since 0.3 */ +pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata); /** Rename the stream. \since 0.5 */ -pa_operation* pa_stream_set_name(pa_stream *s, const char *name, void(*cb)(pa_stream*c, int success, void *userdata), void *userdata); +pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, void *userdata); /** Return the total number of bytes written to/read from the * stream. This counter is not reset on pa_stream_flush(), you may do diff --git a/src/polyp/subscribe.c b/src/polyp/subscribe.c index b90e0bf1..4e00997a 100644 --- a/src/polyp/subscribe.c +++ b/src/polyp/subscribe.c @@ -44,7 +44,7 @@ void pa_command_subscribe_event(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSE if (pa_tagstruct_getu32(t, &e) < 0 || pa_tagstruct_getu32(t, &index) < 0 || !pa_tagstruct_eof(t)) { - pa_context_fail(c, PA_ERROR_PROTOCOL); + pa_context_fail(c, PA_ERR_PROTOCOL); goto finish; } diff --git a/src/polypcore/iochannel.c b/src/polypcore/iochannel.c index 7fd09152..c33f593e 100644 --- a/src/polypcore/iochannel.c +++ b/src/polypcore/iochannel.c @@ -59,17 +59,17 @@ static void enable_mainloop_sources(pa_iochannel *io) { pa_io_event_flags_t f = PA_IO_EVENT_NULL; assert(io->input_event); - if (!io->readable) + if (!pa_iochannel_is_readable(io)) f |= PA_IO_EVENT_INPUT; - if (!io->writable) + if (!pa_iochannel_is_writable(io)) f |= PA_IO_EVENT_OUTPUT; io->mainloop->io_enable(io->input_event, f); } else { if (io->input_event) - io->mainloop->io_enable(io->input_event, io->readable ? PA_IO_EVENT_NULL : PA_IO_EVENT_INPUT); + io->mainloop->io_enable(io->input_event, pa_iochannel_is_readable(io) ? PA_IO_EVENT_NULL : PA_IO_EVENT_INPUT); if (io->output_event) - io->mainloop->io_enable(io->output_event, io->writable ? PA_IO_EVENT_NULL : PA_IO_EVENT_OUTPUT); + io->mainloop->io_enable(io->output_event, pa_iochannel_is_writable(io) ? PA_IO_EVENT_NULL : PA_IO_EVENT_OUTPUT); } } @@ -82,33 +82,21 @@ static void callback(pa_mainloop_api* m, pa_io_event *e, int fd, pa_io_event_fla assert(fd >= 0); assert(userdata); - if ((f & (PA_IO_EVENT_HANGUP|PA_IO_EVENT_ERROR)) && !io->hungup) { + if ((f & (PA_IO_EVENT_HANGUP|PA_IO_EVENT_ERROR)) & !io->hungup) { io->hungup = 1; changed = 1; + } - if (e == io->input_event) { - io->mainloop->io_free(io->input_event); - io->input_event = NULL; - - if (io->output_event == e) - io->output_event = NULL; - } else if (e == io->output_event) { - io->mainloop->io_free(io->output_event); - io->output_event = NULL; - } - } else { - - if ((f & PA_IO_EVENT_INPUT) && !io->readable) { - io->readable = 1; - changed = 1; - assert(e == io->input_event); - } - - if ((f & PA_IO_EVENT_OUTPUT) && !io->writable) { - io->writable = 1; - changed = 1; - assert(e == io->output_event); - } + if ((f & PA_IO_EVENT_INPUT) && !io->readable) { + io->readable = 1; + changed = 1; + assert(e == io->input_event); + } + + if ((f & PA_IO_EVENT_OUTPUT) && !io->writable) { + io->writable = 1; + changed = 1; + assert(e == io->output_event); } if (changed) { @@ -217,6 +205,7 @@ ssize_t pa_iochannel_write(pa_iochannel*io, const void*data, size_t l) { if (r < 0) #endif r = write(io->ofd, data, l); + if (r >= 0) { io->writable = 0; enable_mainloop_sources(io); diff --git a/src/polypcore/llist.h b/src/polypcore/llist.h index eb8cd017..c54742d3 100644 --- a/src/polypcore/llist.h +++ b/src/polypcore/llist.h @@ -66,4 +66,14 @@ _item->next = _item->prev = NULL; \ } while(0) +#define PA_LLIST_FIND_HEAD(t,item,head) \ +do { \ + t **_head = (head), *_item = (item); \ + *_head = _item; \ + assert(_head); \ + while ((*_head)->prev) \ + *_head = (*_head)->prev; \ +} while (0) \ + + #endif diff --git a/src/polypcore/mcalign.c b/src/polypcore/mcalign.c index 0f229e28..f90fd7e8 100644 --- a/src/polypcore/mcalign.c +++ b/src/polypcore/mcalign.c @@ -43,6 +43,7 @@ pa_mcalign *pa_mcalign_new(size_t base, pa_memblock_stat *s) { assert(base); m = pa_xnew(pa_mcalign, 1); + m->base = base; pa_memchunk_reset(&m->leftover); pa_memchunk_reset(&m->current); @@ -64,11 +65,16 @@ void pa_mcalign_free(pa_mcalign *m) { } void pa_mcalign_push(pa_mcalign *m, const pa_memchunk *c) { - assert(m && c && c->memblock && c->length); + assert(m); + assert(c); + + assert(c->memblock); + assert(c->length > 0); + + assert(!m->current.memblock); /* Append to the leftover memory block */ if (m->leftover.memblock) { - assert(!m->current.memblock); /* Try to merge */ if (m->leftover.memblock == c->memblock && @@ -110,8 +116,6 @@ void pa_mcalign_push(pa_mcalign *m, const pa_memchunk *c) { } } } else { - assert(!m->leftover.memblock && !m->current.memblock); - /* Nothing to merge or copy, just store it */ if (c->length >= m->base) @@ -124,7 +128,8 @@ void pa_mcalign_push(pa_mcalign *m, const pa_memchunk *c) { } int pa_mcalign_pop(pa_mcalign *m, pa_memchunk *c) { - assert(m && c); + assert(m); + assert(c); /* First test if there's a leftover memory block available */ if (m->leftover.memblock) { @@ -187,3 +192,15 @@ int pa_mcalign_pop(pa_mcalign *m, pa_memchunk *c) { return -1; } + +size_t pa_mcalign_csize(pa_mcalign *m, size_t l) { + assert(m); + assert(l > 0); + + assert(!m->current.memblock); + + if (m->leftover.memblock) + l += m->leftover.length; + + return (l/m->base)*m->base; +} diff --git a/src/polypcore/mcalign.h b/src/polypcore/mcalign.h index a9107e0e..58019462 100644 --- a/src/polypcore/mcalign.h +++ b/src/polypcore/mcalign.h @@ -74,4 +74,7 @@ void pa_mcalign_push(pa_mcalign *m, const pa_memchunk *c); * nonzero otherwise. */ int pa_mcalign_pop(pa_mcalign *m, pa_memchunk *c); +/* If we pass l bytes in now, how many bytes would we get out? */ +size_t pa_mcalign_csize(pa_mcalign *m, size_t l); + #endif diff --git a/src/polypcore/memblock.c b/src/polypcore/memblock.c index 2c0bef57..04e8436f 100644 --- a/src/polypcore/memblock.c +++ b/src/polypcore/memblock.c @@ -111,13 +111,16 @@ pa_memblock *pa_memblock_new_user(void *d, size_t length, void (*free_cb)(void * } pa_memblock* pa_memblock_ref(pa_memblock*b) { - assert(b && b->ref >= 1); + assert(b); + assert(b->ref >= 1); + b->ref++; return b; } void pa_memblock_unref(pa_memblock*b) { - assert(b && b->ref >= 1); + assert(b); + assert(b->ref >= 1); if ((--(b->ref)) == 0) { stat_remove(b); diff --git a/src/polypcore/memblock.h b/src/polypcore/memblock.h index c5751406..9471278a 100644 --- a/src/polypcore/memblock.h +++ b/src/polypcore/memblock.h @@ -79,7 +79,6 @@ references to the memory. This causes the memory to be copied and converted into a PA_MEMBLOCK_DYNAMIC type memory block */ void pa_memblock_unref_fixed(pa_memblock*b); - pa_memblock_stat* pa_memblock_stat_new(void); void pa_memblock_stat_unref(pa_memblock_stat *s); pa_memblock_stat * pa_memblock_stat_ref(pa_memblock_stat *s); diff --git a/src/polypcore/memblockq.c b/src/polypcore/memblockq.c index 4a0225e5..05c810bd 100644 --- a/src/polypcore/memblockq.c +++ b/src/polypcore/memblockq.c @@ -38,30 +38,45 @@ struct memblock_list { struct memblock_list *next, *prev; + int64_t index; pa_memchunk chunk; }; struct pa_memblockq { struct memblock_list *blocks, *blocks_tail; unsigned n_blocks; - size_t current_length, maxlength, tlength, base, prebuf, orig_prebuf, minreq; - pa_mcalign *mcalign; + size_t maxlength, tlength, base, prebuf, minreq; + int64_t read_index, write_index; + enum { PREBUF, RUNNING } state; pa_memblock_stat *memblock_stat; + pa_memblock *silence; + pa_mcalign *mcalign; }; -pa_memblockq* pa_memblockq_new(size_t maxlength, size_t tlength, size_t base, size_t prebuf, size_t minreq, pa_memblock_stat *s) { +pa_memblockq* pa_memblockq_new( + int64_t idx, + size_t maxlength, + size_t tlength, + size_t base, + size_t prebuf, + size_t minreq, + pa_memblock *silence, + pa_memblock_stat *s) { + pa_memblockq* bq; - assert(maxlength && base && maxlength); - bq = pa_xmalloc(sizeof(pa_memblockq)); - bq->blocks = bq->blocks_tail = 0; + assert(base > 0); + assert(maxlength >= base); + + bq = pa_xnew(pa_memblockq, 1); + bq->blocks = bq->blocks_tail = NULL; bq->n_blocks = 0; - bq->current_length = 0; + bq->base = base; + bq->read_index = bq->write_index = idx; + bq->memblock_stat = s; pa_log_debug(__FILE__": memblockq requested: maxlength=%u, tlength=%u, base=%u, prebuf=%u, minreq=%u\n", maxlength, tlength, base, prebuf, minreq); - - bq->base = base; bq->maxlength = ((maxlength+base-1)/base)*base; assert(bq->maxlength >= base); @@ -70,26 +85,25 @@ pa_memblockq* pa_memblockq_new(size_t maxlength, size_t tlength, size_t base, si if (!bq->tlength || bq->tlength >= bq->maxlength) bq->tlength = bq->maxlength; - bq->minreq = (minreq/base)*base; - if (bq->minreq == 0) - bq->minreq = 1; - - bq->prebuf = (prebuf == (size_t) -1) ? bq->maxlength/2 : prebuf; - bq->prebuf = (bq->prebuf/base)*base; + bq->prebuf = (prebuf == (size_t) -1) ? bq->tlength/2 : prebuf; + bq->prebuf = ((bq->prebuf+base-1)/base)*base; if (bq->prebuf > bq->maxlength) bq->prebuf = bq->maxlength; - if (bq->prebuf > bq->tlength - bq->minreq) - bq->prebuf = bq->tlength - bq->minreq; + bq->minreq = (minreq/base)*base; + + if (bq->minreq > bq->tlength - bq->prebuf) + bq->minreq = bq->tlength - bq->prebuf; - bq->orig_prebuf = bq->prebuf; + if (!bq->minreq) + bq->minreq = 1; pa_log_debug(__FILE__": memblockq sanitized: maxlength=%u, tlength=%u, base=%u, prebuf=%u, minreq=%u\n", bq->maxlength, bq->tlength, bq->base, bq->prebuf, bq->minreq); - - bq->mcalign = NULL; - - bq->memblock_stat = s; + bq->state = bq->prebuf ? PREBUF : RUNNING; + bq->silence = silence ? pa_memblock_ref(silence) : NULL; + bq->mcalign = NULL; + return bq; } @@ -97,248 +111,510 @@ void pa_memblockq_free(pa_memblockq* bq) { assert(bq); pa_memblockq_flush(bq); - + + if (bq->silence) + pa_memblock_unref(bq->silence); + if (bq->mcalign) pa_mcalign_free(bq->mcalign); - + pa_xfree(bq); } -void pa_memblockq_push(pa_memblockq* bq, const pa_memchunk *chunk, size_t delta) { - struct memblock_list *q; - assert(bq && chunk && chunk->memblock && chunk->length && (chunk->length % bq->base) == 0); - - pa_memblockq_seek(bq, delta); - - if (bq->blocks_tail && bq->blocks_tail->chunk.memblock == chunk->memblock) { - /* Try to merge memory chunks */ +static void drop_block(pa_memblockq *bq, struct memblock_list *q) { + assert(bq); + assert(q); - if (bq->blocks_tail->chunk.index+bq->blocks_tail->chunk.length == chunk->index) { - bq->blocks_tail->chunk.length += chunk->length; - bq->current_length += chunk->length; - return; - } - } + assert(bq->n_blocks >= 1); - q = pa_xmalloc(sizeof(struct memblock_list)); - - q->chunk = *chunk; - pa_memblock_ref(q->chunk.memblock); - assert(q->chunk.index+q->chunk.length <= q->chunk.memblock->length); - q->next = NULL; - if ((q->prev = bq->blocks_tail)) - bq->blocks_tail->next = q; + if (q->prev) + q->prev->next = q->next; else - bq->blocks = q; + bq->blocks = q->next; - bq->blocks_tail = q; + if (q->next) + q->next->prev = q->prev; + else + bq->blocks_tail = q->prev; - bq->n_blocks++; - bq->current_length += chunk->length; + pa_memblock_unref(q->chunk.memblock); + pa_xfree(q); - pa_memblockq_shorten(bq, bq->maxlength); + bq->n_blocks--; } -int pa_memblockq_peek(pa_memblockq* bq, pa_memchunk *chunk) { - assert(bq && chunk); +static int can_push(pa_memblockq *bq, size_t l) { + int64_t end; - if (!bq->blocks || bq->current_length < bq->prebuf) - return -1; + assert(bq); - bq->prebuf = 0; + if (bq->read_index > bq->write_index) { + int64_t d = bq->read_index - bq->write_index; - *chunk = bq->blocks->chunk; - pa_memblock_ref(chunk->memblock); + if (l > d) + l -= d; + else + return 1; + } - return 0; + end = bq->blocks_tail ? bq->blocks_tail->index + bq->blocks_tail->chunk.length : 0; + + /* Make sure that the list doesn't get too long */ + if (bq->write_index + l > end) + if (bq->write_index + l - bq->read_index > bq->maxlength) + return 0; + + return 1; } -void pa_memblockq_drop(pa_memblockq *bq, const pa_memchunk *chunk, size_t length) { - assert(bq && chunk && length); +int pa_memblockq_push(pa_memblockq* bq, const pa_memchunk *uchunk) { + + struct memblock_list *q, *n; + pa_memchunk chunk; + + assert(bq); + assert(uchunk); + assert(uchunk->memblock); + assert(uchunk->length > 0); + assert(uchunk->index + uchunk->length <= uchunk->memblock->length); + + if (uchunk->length % bq->base) + return -1; + + if (!can_push(bq, uchunk->length)) + return -1; - if (!bq->blocks || memcmp(&bq->blocks->chunk, chunk, sizeof(pa_memchunk))) - return; + chunk = *uchunk; - assert(length <= bq->blocks->chunk.length); - pa_memblockq_skip(bq, length); -} + if (bq->read_index > bq->write_index) { -static void remove_block(pa_memblockq *bq, struct memblock_list *q) { - assert(bq && q); + /* We currently have a buffer underflow, we need to drop some + * incoming data */ - if (q->prev) - q->prev->next = q->next; - else { - assert(bq->blocks == q); - bq->blocks = q->next; + int64_t d = bq->read_index - bq->write_index; + + if (chunk.length > d) { + chunk.index += d; + chunk.length -= d; + bq->write_index = bq->read_index; + } else { + /* We drop the incoming data completely */ + bq->write_index += chunk.length; + return 0; + } } - if (q->next) - q->next->prev = q->prev; - else { - assert(bq->blocks_tail == q); - bq->blocks_tail = q->prev; + /* We go from back to front to look for the right place to add + * this new entry. Drop data we will overwrite on the way */ + + q = bq->blocks_tail; + while (q) { + + if (bq->write_index >= q->index + q->chunk.length) + /* We found the entry where we need to place the new entry immediately after */ + break; + else if (bq->write_index + chunk.length <= q->index) { + /* This entry isn't touched at all, let's skip it */ + q = q->prev; + } else if (bq->write_index <= q->index && + bq->write_index + chunk.length >= q->index + q->chunk.length) { + + /* This entry is fully replaced by the new entry, so let's drop it */ + + struct memblock_list *p; + p = q; + q = q->prev; + drop_block(bq, p); + } else if (bq->write_index >= q->index) { + /* The write index points into this memblock, so let's + * truncate or split it */ + + if (bq->write_index + chunk.length < q->index + q->chunk.length) { + + /* We need to save the end of this memchunk */ + struct memblock_list *p; + size_t d; + + /* Create a new list entry for the end of thie memchunk */ + p = pa_xnew(struct memblock_list, 1); + p->chunk = q->chunk; + pa_memblock_ref(p->chunk.memblock); + + /* Calculate offset */ + d = bq->write_index + chunk.length - q->index; + assert(d > 0); + + /* Drop it from the new entry */ + p->index = q->index + d; + p->chunk.length -= d; + + /* Add it to the list */ + p->prev = q; + if ((p->next = q->next)) + q->next->prev = p; + else + bq->blocks_tail = p; + q->next = p; + + bq->n_blocks++; + } + + /* Truncate the chunk */ + if (!(q->chunk.length = bq->write_index - q->index)) { + struct memblock_list *p; + p = q; + q = q->prev; + drop_block(bq, p); + } + + /* We had to truncate this block, hence we're now at the right position */ + break; + } else { + size_t d; + + assert(bq->write_index + chunk.length > q->index && + bq->write_index + chunk.length < q->index + q->chunk.length && + bq->write_index < q->index); + + /* The job overwrites the current entry at the end, so let's drop the beginning of this entry */ + + d = bq->write_index + chunk.length - q->index; + q->index += d; + q->chunk.index += d; + q->chunk.length -= d; + + q = q->prev; + } + } + + if (q) { + assert(bq->write_index >= q->index + q->chunk.length); + assert(!q->next || (bq->write_index+chunk.length <= q->next->index)); + + /* Try to merge memory blocks */ + + if (q->chunk.memblock == chunk.memblock && + q->chunk.index + q->chunk.length == chunk.index && + bq->write_index == q->index + q->chunk.length) { + + q->chunk.length += chunk.length; + bq->write_index += chunk.length; + return 0; + } + } else + assert(!bq->blocks || (bq->write_index+chunk.length <= bq->blocks->index)); + + + n = pa_xnew(struct memblock_list, 1); + n->chunk = chunk; + pa_memblock_ref(n->chunk.memblock); + n->index = bq->write_index; + bq->write_index += n->chunk.length; + + n->next = q ? q->next : bq->blocks; + n->prev = q; + + if (n->next) + n->next->prev = n; + else + bq->blocks_tail = n; + + if (n->prev) + n->prev->next = n; + else + bq->blocks = n; - pa_memblock_unref(q->chunk.memblock); - pa_xfree(q); - - bq->n_blocks--; + bq->n_blocks++; + return 0; } -void pa_memblockq_skip(pa_memblockq *bq, size_t length) { - assert(bq && length && (length % bq->base) == 0); +int pa_memblockq_peek(pa_memblockq* bq, pa_memchunk *chunk) { + assert(bq); + assert(chunk); - while (length > 0) { - size_t l = length; - assert(bq->blocks && bq->current_length >= length); - - if (l > bq->blocks->chunk.length) - l = bq->blocks->chunk.length; + if (bq->state == PREBUF) { - bq->blocks->chunk.index += l; - bq->blocks->chunk.length -= l; - bq->current_length -= l; - - if (!bq->blocks->chunk.length) - remove_block(bq, bq->blocks); + /* We need to pre-buffer */ + if (pa_memblockq_get_length(bq) < bq->prebuf) + return -1; + + bq->state = RUNNING; - length -= l; + } else if (bq->prebuf > 0 && bq->read_index >= bq->write_index) { + + /* Buffer underflow protection */ + bq->state = PREBUF; + return -1; } -} + + /* Do we need to spit out silence? */ + if (!bq->blocks || bq->blocks->index > bq->read_index) { -void pa_memblockq_shorten(pa_memblockq *bq, size_t length) { - size_t l; - assert(bq); + size_t length; + + /* How much silence shall we return? */ + length = bq->blocks ? bq->blocks->index - bq->read_index : 0; + + /* We need to return silence, since no data is yet available */ + if (bq->silence) { + chunk->memblock = pa_memblock_ref(bq->silence); - if (bq->current_length <= length) - return; + if (!length || length > chunk->memblock->length) + length = chunk->memblock->length; + + chunk->length = length; + } else { + chunk->memblock = NULL; + chunk->length = length; + } + + chunk->index = 0; + return 0; + } - /*pa_log(__FILE__": Warning! pa_memblockq_shorten()\n");*/ + /* Ok, let's pass real data to the caller */ + assert(bq->blocks->index == bq->read_index); - l = bq->current_length - length; - l /= bq->base; - l *= bq->base; + *chunk = bq->blocks->chunk; + pa_memblock_ref(chunk->memblock); - pa_memblockq_skip(bq, l); + return 0; } - -void pa_memblockq_empty(pa_memblockq *bq) { +void pa_memblockq_drop(pa_memblockq *bq, const pa_memchunk *chunk, size_t length) { assert(bq); - pa_memblockq_shorten(bq, 0); + assert(length % bq->base == 0); + + assert(!chunk || length <= chunk->length); + + if (chunk) { + + if (bq->blocks && bq->blocks->index == bq->read_index) { + /* The first item in queue is valid */ + + /* Does the chunk match with what the user supplied us? */ + if (memcmp(chunk, &bq->blocks->chunk, sizeof(pa_memchunk)) != 0) + return; + + } else { + size_t l; + + /* The first item in the queue is not yet relevant */ + + assert(!bq->blocks || bq->blocks->index > bq->read_index); + l = bq->blocks ? bq->blocks->index - bq->read_index : 0; + + if (bq->silence) { + + if (!l || l > bq->silence->length) + l = bq->silence->length; + + } + + /* Do the entries still match? */ + if (chunk->index != 0 || chunk->length != l || chunk->memblock != bq->silence) + return; + } + } + + while (length > 0) { + + if (bq->blocks) { + size_t d; + + assert(bq->blocks->index >= bq->read_index); + + d = (size_t) (bq->blocks->index - bq->read_index); + + if (d >= length) { + /* The first block is too far in the future */ + + bq->read_index += length; + break; + } else { + + length -= d; + bq->read_index += d; + } + + assert(bq->blocks->index == bq->read_index); + + if (bq->blocks->chunk.length <= length) { + /* We need to drop the full block */ + + length -= bq->blocks->chunk.length; + bq->read_index += bq->blocks->chunk.length; + + drop_block(bq, bq->blocks); + } else { + /* Only the start of this block needs to be dropped */ + + bq->blocks->chunk.index += length; + bq->blocks->chunk.length -= length; + bq->blocks->index += length; + bq->read_index += length; + break; + } + + } else { + + /* The list is empty, there's nothing we could drop */ + bq->read_index += length; + break; + } + } } int pa_memblockq_is_readable(pa_memblockq *bq) { assert(bq); - return bq->current_length && (bq->current_length >= bq->prebuf); + if (bq->prebuf > 0) { + size_t l = pa_memblockq_get_length(bq); + + if (bq->state == PREBUF && l < bq->prebuf) + return 0; + + if (l <= 0) + return 0; + } + + return 1; } int pa_memblockq_is_writable(pa_memblockq *bq, size_t length) { assert(bq); - return bq->current_length + length <= bq->tlength; + if (length % bq->base) + return 0; + + return pa_memblockq_get_length(bq) + length <= bq->tlength; } -uint32_t pa_memblockq_get_length(pa_memblockq *bq) { +size_t pa_memblockq_get_length(pa_memblockq *bq) { assert(bq); - return bq->current_length; + + if (bq->write_index <= bq->read_index) + return 0; + + return (size_t) (bq->write_index - bq->read_index); } -uint32_t pa_memblockq_missing(pa_memblockq *bq) { +size_t pa_memblockq_missing(pa_memblockq *bq) { size_t l; assert(bq); - if (bq->current_length >= bq->tlength) + if ((l = pa_memblockq_get_length(bq)) >= bq->tlength) return 0; - l = bq->tlength - bq->current_length; - assert(l); - + l = bq->tlength - l; return (l >= bq->minreq) ? l : 0; } -void pa_memblockq_push_align(pa_memblockq* bq, const pa_memchunk *chunk, size_t delta) { - pa_memchunk rchunk; - assert(bq && chunk && bq->base); +size_t pa_memblockq_get_minreq(pa_memblockq *bq) { + assert(bq); - if (bq->base == 1) { - pa_memblockq_push(bq, chunk, delta); - return; - } + return bq->minreq; +} - if (!bq->mcalign) { - bq->mcalign = pa_mcalign_new(bq->base, bq->memblock_stat); - assert(bq->mcalign); +void pa_memblockq_seek(pa_memblockq *bq, int64_t offset, pa_seek_mode_t seek) { + assert(bq); + + switch (seek) { + case PA_SEEK_RELATIVE: + bq->write_index += offset; + return; + case PA_SEEK_ABSOLUTE: + bq->write_index = offset; + return; + case PA_SEEK_RELATIVE_ON_READ: + bq->write_index = bq->read_index + offset; + return; + case PA_SEEK_RELATIVE_END: + bq->write_index = (bq->blocks_tail ? bq->blocks_tail->index + bq->blocks_tail->chunk.length : bq->read_index) + offset; + return; } + + assert(0); +} + +void pa_memblockq_flush(pa_memblockq *bq) { + assert(bq); - pa_mcalign_push(bq->mcalign, chunk); + while (bq->blocks) + drop_block(bq, bq->blocks); - while (pa_mcalign_pop(bq->mcalign, &rchunk) >= 0) { - pa_memblockq_push(bq, &rchunk, delta); - pa_memblock_unref(rchunk.memblock); - delta = 0; - } + assert(bq->n_blocks == 0); + bq->write_index = bq->read_index; + + pa_memblockq_prebuf_force(bq); } -uint32_t pa_memblockq_get_minreq(pa_memblockq *bq) { +size_t pa_memblockq_get_tlength(pa_memblockq *bq) { assert(bq); - return bq->minreq; + + return bq->tlength; } -void pa_memblockq_prebuf_disable(pa_memblockq *bq) { +int64_t pa_memblockq_get_read_index(pa_memblockq *bq) { assert(bq); - bq->prebuf = 0; + return bq->read_index; } -void pa_memblockq_prebuf_reenable(pa_memblockq *bq) { +int64_t pa_memblockq_get_write_index(pa_memblockq *bq) { assert(bq); - bq->prebuf = bq->orig_prebuf; + return bq->write_index; } -void pa_memblockq_seek(pa_memblockq *bq, size_t length) { +int pa_memblockq_push_align(pa_memblockq* bq, const pa_memchunk *chunk) { + pa_memchunk rchunk; + assert(bq); + assert(chunk && bq->base); + + if (bq->base == 1) + return pa_memblockq_push(bq, chunk); + + if (!bq->mcalign) + bq->mcalign = pa_mcalign_new(bq->base, bq->memblock_stat); - if (!length) - return; + if (!can_push(bq, pa_mcalign_csize(bq->mcalign, chunk->length))) + return -1; + + pa_mcalign_push(bq->mcalign, chunk); + + while (pa_mcalign_pop(bq->mcalign, &rchunk) >= 0) { + int r; + r = pa_memblockq_push(bq, &rchunk); + pa_memblock_unref(rchunk.memblock); - while (length >= bq->base) { - size_t l = length; - if (!bq->current_length) - return; + if (r < 0) + return -1; + } - assert(bq->blocks_tail); - - if (l > bq->blocks_tail->chunk.length) - l = bq->blocks_tail->chunk.length; + return 0; +} - bq->blocks_tail->chunk.length -= l; - bq->current_length -= l; - - if (bq->blocks_tail->chunk.length == 0) - remove_block(bq, bq->blocks); +void pa_memblockq_shorten(pa_memblockq *bq, size_t length) { + size_t l; + assert(bq); - length -= l; - } + l = pa_memblockq_get_length(bq); + + if (l > length) + pa_memblockq_drop(bq, NULL, l - length); } -void pa_memblockq_flush(pa_memblockq *bq) { - struct memblock_list *l; +void pa_memblockq_prebuf_disable(pa_memblockq *bq) { assert(bq); - - while ((l = bq->blocks)) { - bq->blocks = l->next; - pa_memblock_unref(l->chunk.memblock); - pa_xfree(l); - } - bq->blocks_tail = NULL; - bq->n_blocks = 0; - bq->current_length = 0; + if (bq->state == PREBUF) + bq->state = RUNNING; } -uint32_t pa_memblockq_get_tlength(pa_memblockq *bq) { +void pa_memblockq_prebuf_force(pa_memblockq *bq) { assert(bq); - return bq->tlength; + + if (bq->state == RUNNING && bq->prebuf > 0) + bq->state = PREBUF; } diff --git a/src/polypcore/memblockq.h b/src/polypcore/memblockq.h index 7bb25f90..210f1a07 100644 --- a/src/polypcore/memblockq.h +++ b/src/polypcore/memblockq.h @@ -23,9 +23,11 @@ ***/ #include <sys/types.h> +#include <inttypes.h> #include <polypcore/memblock.h> #include <polypcore/memchunk.h> +#include <polyp/def.h> /* A memblockq is a queue of pa_memchunks (yepp, the name is not * perfect). It is similar to the ring buffers used by most other @@ -35,42 +37,59 @@ typedef struct pa_memblockq pa_memblockq; + /* Parameters: - - maxlength: maximum length of queue. If more data is pushed into the queue, data from the front is dropped - - length: the target length of the queue. - - base: a base value for all metrics. Only multiples of this value are popped from the queue - - prebuf: before passing the first byte out, make sure that enough bytes are in the queue - - minreq: pa_memblockq_missing() will only return values greater than this value + + - idx: start value for both read and write index + + - maxlength: maximum length of queue. If more data is pushed into + the queue, the operation will fail. Must not be 0. + + - tlength: the target length of the queue. Pass 0 for the default. + + - base: a base value for all metrics. Only multiples of this value + are popped from the queue or should be pushed into + it. Must not be 0. + + - prebuf: If the queue runs empty wait until this many bytes are in + queue again before passing the first byte out. If set + to 0 pa_memblockq_pop() will return a silence memblock + if no data is in the queue and will never fail. Pass + (size_t) -1 for the default. + + - minreq: pa_memblockq_missing() will only return values greater + than this value. Pass 0 for the default. + + - silence: return this memblock whzen reading unitialized data */ -pa_memblockq* pa_memblockq_new(size_t maxlength, - size_t tlength, - size_t base, - size_t prebuf, - size_t minreq, - pa_memblock_stat *s); +pa_memblockq* pa_memblockq_new( + int64_t idx, + size_t maxlength, + size_t tlength, + size_t base, + size_t prebuf, + size_t minreq, + pa_memblock *silence, + pa_memblock_stat *s); + void pa_memblockq_free(pa_memblockq*bq); -/* Push a new memory chunk into the queue. Optionally specify a value for future cancellation. */ -void pa_memblockq_push(pa_memblockq* bq, const pa_memchunk *chunk, size_t delta); +/* Push a new memory chunk into the queue. */ +int pa_memblockq_push(pa_memblockq* bq, const pa_memchunk *chunk); -/* Same as pa_memblockq_push(), however chunks are filtered through a mcalign object, and thus aligned to multiples of base */ -void pa_memblockq_push_align(pa_memblockq* bq, const pa_memchunk *chunk, size_t delta); +/* Push a new memory chunk into the queue, but filter it through a + * pa_mcalign object. Don't mix this with pa_memblockq_seek() unless + * you know what you do. */ +int pa_memblockq_push_align(pa_memblockq* bq, const pa_memchunk *chunk); /* Return a copy of the next memory chunk in the queue. It is not removed from the queue */ int pa_memblockq_peek(pa_memblockq* bq, pa_memchunk *chunk); -/* Drop the specified bytes from the queue, only valid aufter pa_memblockq_peek() */ +/* Drop the specified bytes from the queue, but only if the first + * chunk in the queue matches the one passed here. If NULL is passed, + * this check isn't done. */ void pa_memblockq_drop(pa_memblockq *bq, const pa_memchunk *chunk, size_t length); -/* Drop the specified bytes from the queue */ -void pa_memblockq_skip(pa_memblockq *bq, size_t length); - -/* Shorten the pa_memblockq to the specified length by dropping data at the end of the queue */ -void pa_memblockq_shorten(pa_memblockq *bq, size_t length); - -/* Empty the pa_memblockq */ -void pa_memblockq_empty(pa_memblockq *bq); - /* Test if the pa_memblockq is currently readable, that is, more data than base */ int pa_memblockq_is_readable(pa_memblockq *bq); @@ -78,27 +97,38 @@ int pa_memblockq_is_readable(pa_memblockq *bq); int pa_memblockq_is_writable(pa_memblockq *bq, size_t length); /* Return the length of the queue in bytes */ -uint32_t pa_memblockq_get_length(pa_memblockq *bq); +size_t pa_memblockq_get_length(pa_memblockq *bq); /* Return how many bytes are missing in queue to the specified fill amount */ -uint32_t pa_memblockq_missing(pa_memblockq *bq); +size_t pa_memblockq_missing(pa_memblockq *bq); /* Returns the minimal request value */ -uint32_t pa_memblockq_get_minreq(pa_memblockq *bq); - -/* Force disabling of pre-buf even when the pre-buffer is not yet filled */ -void pa_memblockq_prebuf_disable(pa_memblockq *bq); - -/* Reenable pre-buf to the initial level */ -void pa_memblockq_prebuf_reenable(pa_memblockq *bq); +size_t pa_memblockq_get_minreq(pa_memblockq *bq); /* Manipulate the write pointer */ -void pa_memblockq_seek(pa_memblockq *bq, size_t delta); +void pa_memblockq_seek(pa_memblockq *bq, int64_t offset, pa_seek_mode_t seek); -/* Flush the queue */ +/* Set the queue to silence, set write index to read index */ void pa_memblockq_flush(pa_memblockq *bq); /* Get Target length */ uint32_t pa_memblockq_get_tlength(pa_memblockq *bq); +/* Return the current read index */ +int64_t pa_memblockq_get_read_index(pa_memblockq *bq); + +/* Return the current write index */ +int64_t pa_memblockq_get_write_index(pa_memblockq *bq); + +/* Shorten the pa_memblockq to the specified length by dropping data + * at the read end of the queue. The read index is increased until the + * queue has the specified length */ +void pa_memblockq_shorten(pa_memblockq *bq, size_t length); + +/* Ignore prebuf for now */ +void pa_memblockq_prebuf_disable(pa_memblockq *bq); + +/* Force prebuf */ +void pa_memblockq_prebuf_force(pa_memblockq *bq); + #endif diff --git a/src/polypcore/native-common.h b/src/polypcore/native-common.h index ac3ea823..0d17b022 100644 --- a/src/polypcore/native-common.h +++ b/src/polypcore/native-common.h @@ -28,22 +28,22 @@ PA_C_DECL_BEGIN enum { + /* Generic commands */ PA_COMMAND_ERROR, PA_COMMAND_TIMEOUT, /* pseudo command */ PA_COMMAND_REPLY, + + /* Commands from client to server */ PA_COMMAND_CREATE_PLAYBACK_STREAM, PA_COMMAND_DELETE_PLAYBACK_STREAM, PA_COMMAND_CREATE_RECORD_STREAM, PA_COMMAND_DELETE_RECORD_STREAM, PA_COMMAND_EXIT, - PA_COMMAND_REQUEST, PA_COMMAND_AUTH, PA_COMMAND_SET_CLIENT_NAME, PA_COMMAND_LOOKUP_SINK, PA_COMMAND_LOOKUP_SOURCE, PA_COMMAND_DRAIN_PLAYBACK_STREAM, - PA_COMMAND_PLAYBACK_STREAM_KILLED, - PA_COMMAND_RECORD_STREAM_KILLED, PA_COMMAND_STAT, PA_COMMAND_GET_PLAYBACK_LATENCY, PA_COMMAND_CREATE_UPLOAD_STREAM, @@ -68,7 +68,6 @@ enum { PA_COMMAND_GET_SAMPLE_INFO, PA_COMMAND_GET_SAMPLE_INFO_LIST, PA_COMMAND_SUBSCRIBE, - PA_COMMAND_SUBSCRIBE_EVENT, PA_COMMAND_SET_SINK_VOLUME, PA_COMMAND_SET_SINK_INPUT_VOLUME, @@ -95,6 +94,15 @@ enum { PA_COMMAND_CORK_RECORD_STREAM, PA_COMMAND_FLUSH_RECORD_STREAM, PA_COMMAND_PREBUF_PLAYBACK_STREAM, + + /* Commands from server to client */ + PA_COMMAND_REQUEST, + PA_COMMAND_OVERFLOW, + PA_COMMAND_UNDERFLOW, + PA_COMMAND_PLAYBACK_STREAM_KILLED, + PA_COMMAND_RECORD_STREAM_KILLED, + PA_COMMAND_SUBSCRIBE_EVENT, + PA_COMMAND_MAX }; diff --git a/src/polypcore/packet.c b/src/polypcore/packet.c index 41803cf9..31ddad95 100644 --- a/src/polypcore/packet.c +++ b/src/polypcore/packet.c @@ -32,37 +32,46 @@ pa_packet* pa_packet_new(size_t length) { pa_packet *p; + assert(length); + p = pa_xmalloc(sizeof(pa_packet)+length); p->ref = 1; p->length = length; p->data = (uint8_t*) (p+1); p->type = PA_PACKET_APPENDED; + return p; } -pa_packet* pa_packet_new_dynamic(uint8_t* data, size_t length) { +pa_packet* pa_packet_new_dynamic(void* data, size_t length) { pa_packet *p; - assert(data && length); - p = pa_xmalloc(sizeof(pa_packet)); + + assert(data); + assert(length); + + p = pa_xnew(pa_packet, 1); p->ref = 1; p->length = length; p->data = data; p->type = PA_PACKET_DYNAMIC; + return p; } pa_packet* pa_packet_ref(pa_packet *p) { - assert(p && p->ref >= 1); + assert(p); + assert(p->ref >= 1); + p->ref++; return p; } void pa_packet_unref(pa_packet *p) { - assert(p && p->ref >= 1); - p->ref--; - - if (p->ref == 0) { + assert(p); + assert(p->ref >= 1); + + if (--p->ref == 0) { if (p->type == PA_PACKET_DYNAMIC) pa_xfree(p->data); pa_xfree(p); diff --git a/src/polypcore/packet.h b/src/polypcore/packet.h index 0ac47485..fbc58232 100644 --- a/src/polypcore/packet.h +++ b/src/polypcore/packet.h @@ -33,7 +33,7 @@ typedef struct pa_packet { } pa_packet; pa_packet* pa_packet_new(size_t length); -pa_packet* pa_packet_new_dynamic(uint8_t* data, size_t length); +pa_packet* pa_packet_new_dynamic(void* data, size_t length); pa_packet* pa_packet_ref(pa_packet *p); void pa_packet_unref(pa_packet *p); diff --git a/src/polypcore/protocol-esound.c b/src/polypcore/protocol-esound.c index a16ac280..5adff57a 100644 --- a/src/polypcore/protocol-esound.c +++ b/src/polypcore/protocol-esound.c @@ -186,6 +186,7 @@ static void connection_free(struct connection *c) { if (c->sink_input) { pa_sink_input_disconnect(c->sink_input); + pa_log("disconnect\n"); pa_sink_input_unref(c->sink_input); } @@ -333,7 +334,15 @@ static int esd_proto_stream_play(struct connection *c, PA_GCC_UNUSED esd_proto_t } l = (size_t) (pa_bytes_per_second(&ss)*PLAYBACK_BUFFER_SECONDS); - c->input_memblockq = pa_memblockq_new(l, 0, pa_frame_size(&ss), l/2, l/PLAYBACK_BUFFER_FRAGMENTS, c->protocol->core->memblock_stat); + c->input_memblockq = pa_memblockq_new( + 0, + l, + 0, + pa_frame_size(&ss), + (size_t) -1, + l/PLAYBACK_BUFFER_FRAGMENTS, + NULL, + c->protocol->core->memblock_stat); pa_iochannel_socket_set_rcvbuf(c->io, l/PLAYBACK_BUFFER_FRAGMENTS*2); c->playback.fragment_size = l/10; @@ -405,7 +414,15 @@ static int esd_proto_stream_record(struct connection *c, esd_proto_t request, co } l = (size_t) (pa_bytes_per_second(&ss)*RECORD_BUFFER_SECONDS); - c->output_memblockq = pa_memblockq_new(l, 0, pa_frame_size(&ss), 0, 0, c->protocol->core->memblock_stat); + c->output_memblockq = pa_memblockq_new( + 0, + l, + 0, + pa_frame_size(&ss), + 1, + 0, + NULL, + c->protocol->core->memblock_stat); pa_iochannel_socket_set_sndbuf(c->io, l/RECORD_BUFFER_FRAGMENTS*2); c->source_output->owner = c->protocol->module; @@ -724,8 +741,7 @@ static int do_read(struct connection *c) { assert(c->read_data_length < sizeof(c->request)); if ((r = pa_iochannel_read(c->io, ((uint8_t*) &c->request) + c->read_data_length, sizeof(c->request) - c->read_data_length)) <= 0) { - if (r != 0) - pa_log_warn(__FILE__": read() failed: %s\n", strerror(errno)); + pa_log_debug(__FILE__": read() failed: %s\n", r < 0 ? strerror(errno) : "EOF"); return -1; } @@ -773,8 +789,7 @@ static int do_read(struct connection *c) { assert(c->read_data && c->read_data_length < handler->data_length); if ((r = pa_iochannel_read(c->io, (uint8_t*) c->read_data + c->read_data_length, handler->data_length - c->read_data_length)) <= 0) { - if (r != 0) - pa_log_warn(__FILE__": read() failed: %s\n", strerror(errno)); + pa_log_debug(__FILE__": read() failed: %s\n", r < 0 ? strerror(errno) : "EOF"); return -1; } @@ -794,8 +809,7 @@ static int do_read(struct connection *c) { assert(c->scache.memchunk.memblock && c->scache.name && c->scache.memchunk.index < c->scache.memchunk.length); if ((r = pa_iochannel_read(c->io, (uint8_t*) c->scache.memchunk.memblock->data+c->scache.memchunk.index, c->scache.memchunk.length-c->scache.memchunk.index)) <= 0) { - if (r!= 0) - pa_log_warn(__FILE__": read() failed: %s\n", strerror(errno)); + pa_log_debug(__FILE__": read() failed: %s\n", r < 0 ? strerror(errno) : "EOF"); return -1; } @@ -852,13 +866,10 @@ static int do_read(struct connection *c) { } if ((r = pa_iochannel_read(c->io, (uint8_t*) c->playback.current_memblock->data+c->playback.memblock_index, l)) <= 0) { - if (r != 0) - pa_log(__FILE__": read() failed: %s\n", strerror(errno)); + pa_log_debug(__FILE__": read() failed: %s\n", r < 0 ? strerror(errno) : "EOF"); return -1; } -/* pa_log(__FILE__": read %u\n", r); */ - chunk.memblock = c->playback.current_memblock; chunk.index = c->playback.memblock_index; chunk.length = r; @@ -867,7 +878,7 @@ static int do_read(struct connection *c) { c->playback.memblock_index += r; assert(c->input_memblockq); - pa_memblockq_push_align(c->input_memblockq, &chunk, 0); + pa_memblockq_push_align(c->input_memblockq, &chunk); assert(c->sink_input); pa_sink_notify(c->sink_input->sink); } @@ -910,6 +921,8 @@ static int do_write(struct connection *c) { pa_memblockq_drop(c->output_memblockq, &chunk, r); pa_memblock_unref(chunk.memblock); + + pa_source_notify(c->source_output->source); } return 0; @@ -921,21 +934,18 @@ static void do_work(struct connection *c) { assert(c->protocol && c->protocol->core && c->protocol->core->mainloop && c->protocol->core->mainloop->defer_enable); c->protocol->core->mainloop->defer_enable(c->defer_event, 0); -/* pa_log("DOWORK %i\n", pa_iochannel_is_hungup(c->io)); */ + if (c->dead) + return; - if (!c->dead && pa_iochannel_is_readable(c->io)) + if (pa_iochannel_is_readable(c->io)) { if (do_read(c) < 0) goto fail; + } else if (pa_iochannel_is_hungup(c->io)) + goto fail; - if (!c->dead && pa_iochannel_is_writable(c->io)) + if (pa_iochannel_is_writable(c->io)) if (do_write(c) < 0) goto fail; - - /* In case the line was hungup, make sure to rerun this function - as soon as possible, until all data has been read. */ - - if (!c->dead && pa_iochannel_is_hungup(c->io)) - c->protocol->core->mainloop->defer_enable(c->defer_event, 1); return; @@ -943,15 +953,17 @@ fail: if (c->state == ESD_STREAMING_DATA && c->sink_input) { c->dead = 1; - pa_memblockq_prebuf_disable(c->input_memblockq); pa_iochannel_free(c->io); c->io = NULL; - + + pa_memblockq_prebuf_disable(c->input_memblockq); + pa_sink_notify(c->sink_input->sink); } else connection_free(c); } + static void io_callback(pa_iochannel*io, void *userdata) { struct connection *c = userdata; assert(io && c && c->io == io); @@ -1024,7 +1036,7 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) struct connection *c = o->userdata; assert(o && c && chunk); - pa_memblockq_push(c->output_memblockq, chunk, 0); + pa_memblockq_push(c->output_memblockq, chunk); /* do something */ assert(c->protocol && c->protocol->core && c->protocol->core->mainloop && c->protocol->core->mainloop->defer_enable); diff --git a/src/polypcore/protocol-native.c b/src/polypcore/protocol-native.c index 1362fdf2..aaa4fc48 100644 --- a/src/polypcore/protocol-native.c +++ b/src/polypcore/protocol-native.c @@ -48,6 +48,8 @@ #include <polypcore/authkey-prop.h> #include <polypcore/strlist.h> #include <polypcore/props.h> +#include <polypcore/sample-util.h> +#include <polypcore/llist.h> #include "protocol-native.h" @@ -77,6 +79,11 @@ struct playback_stream { size_t requested_bytes; int drain_request; uint32_t drain_tag; + uint32_t syncid; + int underrun; + + /* Sync group members */ + PA_LLIST_FIELDS(struct playback_stream); }; struct upload_stream { @@ -153,7 +160,8 @@ static void command_get_server_info(pa_pdispatch *pd, uint32_t command, uint32_t static void command_subscribe(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); static void command_set_volume(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); static void command_cork_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); -static void command_flush_or_trigger_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); +static void command_flush_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); +static void command_trigger_or_prebuf_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); static void command_set_default_sink_or_source(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); static void command_set_stream_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); static void command_kill(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); @@ -210,9 +218,9 @@ static const pa_pdispatch_callback command_table[PA_COMMAND_MAX] = { [PA_COMMAND_SET_SINK_INPUT_VOLUME] = command_set_volume, [PA_COMMAND_CORK_PLAYBACK_STREAM] = command_cork_playback_stream, - [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = command_flush_or_trigger_playback_stream, - [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = command_flush_or_trigger_playback_stream, - [PA_COMMAND_PREBUF_PLAYBACK_STREAM] = command_flush_or_trigger_playback_stream, + [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = command_flush_playback_stream, + [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = command_trigger_or_prebuf_playback_stream, + [PA_COMMAND_PREBUF_PLAYBACK_STREAM] = command_trigger_or_prebuf_playback_stream, [PA_COMMAND_CORK_RECORD_STREAM] = command_cork_record_stream, [PA_COMMAND_FLUSH_RECORD_STREAM] = command_flush_record_stream, @@ -244,7 +252,7 @@ static struct upload_stream* upload_stream_new( struct upload_stream *s; assert(c && ss && name && length); - s = pa_xmalloc(sizeof(struct upload_stream)); + s = pa_xnew(struct upload_stream, 1); s->type = UPLOAD_STREAM; s->connection = c; s->sample_spec = *ss; @@ -291,7 +299,7 @@ static struct record_stream* record_stream_new( if (!(source_output = pa_source_output_new(source, __FILE__, name, ss, map, -1))) return NULL; - s = pa_xmalloc(sizeof(struct record_stream)); + s = pa_xnew(struct record_stream, 1); s->connection = c; s->source_output = source_output; s->source_output->push = source_output_push_cb; @@ -301,7 +309,15 @@ static struct record_stream* record_stream_new( s->source_output->owner = c->protocol->module; s->source_output->client = c->client; - s->memblockq = pa_memblockq_new(maxlength, 0, base = pa_frame_size(ss), 0, 0, c->protocol->core->memblock_stat); + s->memblockq = pa_memblockq_new( + 0, + maxlength, + 0, + base = pa_frame_size(ss), + 1, + 0, + NULL, + c->protocol->core->memblock_stat); assert(s->memblockq); s->fragment_size = (fragment_size/base)*base; @@ -332,19 +348,40 @@ static struct playback_stream* playback_stream_new( size_t tlength, size_t prebuf, size_t minreq, - pa_cvolume *volume) { + pa_cvolume *volume, + uint32_t syncid) { - struct playback_stream *s; + struct playback_stream *s, *sync; pa_sink_input *sink_input; + pa_memblock *silence; + uint32_t idx; + int64_t start_index; + assert(c && sink && ss && name && maxlength); + /* Find syncid group */ + for (sync = pa_idxset_first(c->output_streams, &idx); sync; sync = pa_idxset_next(c->output_streams, &idx)) { + + if (sync->type != PLAYBACK_STREAM) + continue; + + if (sync->syncid == syncid) + break; + } + + /* Synced streams must connect to the same sink */ + if (sync && sync->sink_input->sink != sink) + return NULL; + if (!(sink_input = pa_sink_input_new(sink, __FILE__, name, ss, map, 0, -1))) return NULL; - s = pa_xmalloc(sizeof(struct playback_stream)); + s = pa_xnew(struct playback_stream, 1); s->type = PLAYBACK_STREAM; s->connection = c; + s->syncid = syncid; s->sink_input = sink_input; + s->underrun = 1; s->sink_input->peek = sink_input_peek_cb; s->sink_input->drop = sink_input_drop_cb; @@ -353,24 +390,56 @@ static struct playback_stream* playback_stream_new( s->sink_input->userdata = s; s->sink_input->owner = c->protocol->module; s->sink_input->client = c->client; - - s->memblockq = pa_memblockq_new(maxlength, tlength, pa_frame_size(ss), prebuf, minreq, c->protocol->core->memblock_stat); - assert(s->memblockq); + if (sync) { + /* Sync id found, now find head of list */ + PA_LLIST_FIND_HEAD(struct playback_stream, sync, &sync); + + /* Prepend ourselves */ + PA_LLIST_PREPEND(struct playback_stream, sync, s); + + /* Set our start index to the current read index of the other grozp member(s) */ + assert(sync->next); + start_index = pa_memblockq_get_read_index(sync->next->memblockq); + } else { + /* This ia a new sync group */ + PA_LLIST_INIT(struct playback_stream, s); + start_index = 0; + } + + silence = pa_silence_memblock_new(ss, 0, c->protocol->core->memblock_stat); + + s->memblockq = pa_memblockq_new( + start_index, + maxlength, + tlength, + pa_frame_size(ss), + prebuf, + minreq, + silence, + c->protocol->core->memblock_stat); + + pa_memblock_unref(silence); + s->requested_bytes = 0; s->drain_request = 0; s->sink_input->volume = *volume; pa_idxset_put(c->output_streams, s, &s->index); + return s; } static void playback_stream_free(struct playback_stream* p) { + struct playback_stream *head; assert(p && p->connection); if (p->drain_request) - pa_pstream_send_error(p->connection->pstream, p->drain_tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(p->connection->pstream, p->drain_tag, PA_ERR_NOENTITY); + + PA_LLIST_FIND_HEAD(struct playback_stream, p, &head); + PA_LLIST_REMOVE(struct playback_stream, head, p); pa_idxset_remove_by_data(p->connection->output_streams, p, NULL); pa_sink_input_disconnect(p->sink_input); @@ -436,7 +505,7 @@ static void request_bytes(struct playback_stream *s) { pa_tagstruct_putu32(t, l); pa_pstream_send_tagstruct(s->connection->pstream, t); -/* pa_log(__FILE__": Requesting %u bytes\n", l); */ +/* pa_log(__FILE__": Requesting %u bytes\n", l); */ } static void send_memblock(struct connection *c) { @@ -461,7 +530,7 @@ static void send_memblock(struct connection *c) { if (schunk.length > r->fragment_size) schunk.length = r->fragment_size; - pa_pstream_send_memblock(c->pstream, r->index, 0, &schunk); + pa_pstream_send_memblock(c->pstream, r->index, 0, PA_SEEK_RELATIVE, &schunk); pa_memblockq_drop(r->memblockq, &chunk, schunk.length); pa_memblock_unref(schunk.memblock); @@ -501,9 +570,27 @@ static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk) { assert(i && i->userdata && chunk); s = i->userdata; - if (pa_memblockq_peek(s->memblockq, chunk) < 0) + if (pa_memblockq_get_length(s->memblockq) <= 0 && !s->underrun) { + pa_tagstruct *t; + + /* Report that we're empty */ + + t = pa_tagstruct_new(NULL, 0); + pa_tagstruct_putu32(t, PA_COMMAND_UNDERFLOW); + pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */ + pa_tagstruct_putu32(t, s->index); + pa_pstream_send_tagstruct(s->connection->pstream, t); + + s->underrun = 1; + } + + if (pa_memblockq_peek(s->memblockq, chunk) < 0) { + pa_log(__FILE__": peek: failure\n"); return -1; + } +/* pa_log(__FILE__": peek: %u\n", chunk->length); */ + return 0; } @@ -513,6 +600,7 @@ static void sink_input_drop_cb(pa_sink_input *i, const pa_memchunk *chunk, size_ s = i->userdata; pa_memblockq_drop(s->memblockq, chunk, length); + request_bytes(s); if (s->drain_request && !pa_memblockq_is_readable(s->memblockq)) { @@ -520,7 +608,7 @@ static void sink_input_drop_cb(pa_sink_input *i, const pa_memchunk *chunk, size_ s->drain_request = 0; } -/* pa_log(__FILE__": after_drop: %u\n", pa_memblockq_get_length(s->memblockq)); */ +/* pa_log(__FILE__": after_drop: %u %u\n", pa_memblockq_get_length(s->memblockq), pa_memblockq_is_readable(s->memblockq)); */ } static void sink_input_kill_cb(pa_sink_input *i) { @@ -546,7 +634,11 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) assert(o && o->userdata && chunk); s = o->userdata; - pa_memblockq_push_align(s->memblockq, chunk, 0); + if (pa_memblockq_push_align(s->memblockq, chunk) < 0) { + pa_log_warn(__FILE__": Failed to push data into output queue.\n"); + return; + } + if (!pa_pstream_is_pending(s->connection->pstream)) send_memblock(s->connection); } @@ -578,7 +670,7 @@ static void command_create_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GC struct connection *c = userdata; struct playback_stream *s; size_t maxlength, tlength, prebuf, minreq; - uint32_t sink_index; + uint32_t sink_index, syncid; const char *name, *sink_name; pa_sample_spec ss; pa_channel_map map; @@ -601,6 +693,7 @@ static void command_create_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GC PA_TAG_U32, &tlength, PA_TAG_U32, &prebuf, PA_TAG_U32, &minreq, + PA_TAG_U32, &syncid, PA_TAG_CVOLUME, &volume, PA_TAG_INVALID) < 0 || !pa_tagstruct_eof(t) || @@ -610,23 +703,23 @@ static void command_create_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GC } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } - if (sink_index != (uint32_t) -1) + if (sink_index != PA_INVALID_INDEX) sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index); else sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK, 1); if (!sink) { - pa_log("%s: Can't find a suitable sink.\n", __FILE__); - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_log_warn(__FILE__": Can't find a suitable sink.\n"); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } - if (!(s = playback_stream_new(c, sink, &ss, &map, name, maxlength, tlength, prebuf, minreq, &volume))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID); + if (!(s = playback_stream_new(c, sink, &ss, &map, name, maxlength, tlength, prebuf, minreq, &volume, syncid))) { + pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID); return; } @@ -656,14 +749,14 @@ static void command_delete_stream(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t comma } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (command == PA_COMMAND_DELETE_PLAYBACK_STREAM) { struct playback_stream *s; if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != PLAYBACK_STREAM)) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST); + pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST); return; } @@ -671,7 +764,7 @@ static void command_delete_stream(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t comma } else if (command == PA_COMMAND_DELETE_RECORD_STREAM) { struct record_stream *s; if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST); + pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST); return; } @@ -680,7 +773,7 @@ static void command_delete_stream(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t comma struct upload_stream *s; assert(command == PA_COMMAND_DELETE_UPLOAD_STREAM); if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != UPLOAD_STREAM)) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST); + pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST); return; } @@ -717,7 +810,7 @@ static void command_create_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_ } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -727,12 +820,12 @@ static void command_create_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_ source = pa_namereg_get(c->protocol->core, source_name, PA_NAMEREG_SOURCE, 1); if (!source) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } if (!(s = record_stream_new(c, source, &ss, &map, name, maxlength, fragment_size))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID); + pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID); return; } @@ -758,7 +851,7 @@ static void command_exit(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -782,7 +875,7 @@ static void command_auth(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t if (!c->authorized) { if (memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) != 0) { pa_log(__FILE__": Denied access to client with invalid authorization key.\n"); - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -826,7 +919,7 @@ static void command_lookup(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uin } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -842,7 +935,7 @@ static void command_lookup(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uin } if (idx == PA_IDXSET_INVALID) - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); else { pa_tagstruct *reply; reply = pa_tagstruct_new(NULL, 0); @@ -867,12 +960,12 @@ static void command_drain_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(s = pa_idxset_get_by_index(c->output_streams, idx)) || s->type != PLAYBACK_STREAM) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -881,10 +974,10 @@ static void command_drain_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC pa_memblockq_prebuf_disable(s->memblockq); if (!pa_memblockq_is_readable(s->memblockq)) { -/* pa_log("immediate drain: %u\n", pa_memblockq_get_length(s->memblockq)); */ +/* pa_log("immediate drain: %u\n", pa_memblockq_get_length(s->memblockq)); */ pa_pstream_send_simple_ack(c->pstream, tag); } else { -/* pa_log("slow drain triggered\n"); */ +/* pa_log("slow drain triggered\n"); */ s->drain_request = 1; s->drain_tag = tag; @@ -903,7 +996,7 @@ static void command_stat(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -937,12 +1030,12 @@ static void command_get_playback_latency(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_ } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(s = pa_idxset_get_by_index(c->output_streams, idx)) || s->type != PLAYBACK_STREAM) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -980,12 +1073,12 @@ static void command_get_record_latency(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UN } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(s = pa_idxset_get_by_index(c->record_streams, idx))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1026,17 +1119,17 @@ static void command_create_upload_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_ } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if ((length % pa_frame_size(&ss)) != 0 || length <= 0 || !*name) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID); + pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID); return; } if (!(s = upload_stream_new(c, &ss, &map, name, length))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID); + pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID); return; } @@ -1063,12 +1156,12 @@ static void command_finish_upload_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_ } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != UPLOAD_STREAM)) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST); + pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST); return; } @@ -1095,7 +1188,7 @@ static void command_play_sample(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED ui } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1105,12 +1198,12 @@ static void command_play_sample(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED ui sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK, 1); if (!sink) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } if (pa_scache_play_item(c->protocol->core, name, sink, &volume) < 0) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1129,12 +1222,12 @@ static void command_remove_sample(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (pa_scache_remove_item(c->protocol->core, name) < 0) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1261,7 +1354,7 @@ static void command_get_info(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, u } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1292,7 +1385,7 @@ static void command_get_info(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, u } if (!sink && !source && !client && !module && !si && !so && !sce) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1331,7 +1424,7 @@ static void command_get_info_list(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t comma } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1394,7 +1487,7 @@ static void command_get_server_info(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSE } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1444,7 +1537,7 @@ static void command_subscribe(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1478,7 +1571,7 @@ static void command_set_volume(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1493,7 +1586,7 @@ static void command_set_volume(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, } if (!si && !sink) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1509,7 +1602,7 @@ static void command_cork_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_ struct connection *c = userdata; uint32_t idx; int b; - struct playback_stream *s; + struct playback_stream *s, *sync; assert(c && t); if (pa_tagstruct_getu32(t, &idx) < 0 || @@ -1520,20 +1613,82 @@ static void command_cork_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_ } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(s = pa_idxset_get_by_index(c->output_streams, idx)) || s->type != PLAYBACK_STREAM) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } + fprintf(stderr, "Corking %i\n", b); + pa_sink_input_cork(s->sink_input, b); + pa_memblockq_prebuf_force(s->memblockq); + + /* Do the same for all other members in the sync group */ + for (sync = s->prev; sync; sync = sync->prev) { + pa_sink_input_cork(sync->sink_input, b); + pa_memblockq_prebuf_force(sync->memblockq); + } + + for (sync = s->next; sync; sync = sync->next) { + pa_sink_input_cork(sync->sink_input, b); + pa_memblockq_prebuf_force(sync->memblockq); + } + + pa_pstream_send_simple_ack(c->pstream, tag); +} + +static void command_flush_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) { + struct connection *c = userdata; + uint32_t idx; + struct playback_stream *s, *sync; + assert(c && t); + + if (pa_tagstruct_getu32(t, &idx) < 0 || + !pa_tagstruct_eof(t)) { + protocol_error(c); + return; + } + + if (!c->authorized) { + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); + return; + } + + if (!(s = pa_idxset_get_by_index(c->output_streams, idx)) || s->type != PLAYBACK_STREAM) { + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); + return; + } + + pa_memblockq_flush(s->memblockq); + s->underrun = 0; + + /* Do the same for all other members in the sync group */ + for (sync = s->prev; sync; sync = sync->prev) { + pa_memblockq_flush(sync->memblockq); + sync->underrun = 0; + } + + for (sync = s->next; sync; sync = sync->next) { + pa_memblockq_flush(sync->memblockq); + sync->underrun = 0; + } + pa_pstream_send_simple_ack(c->pstream, tag); + pa_sink_notify(s->sink_input->sink); + request_bytes(s); + + for (sync = s->prev; sync; sync = sync->prev) + request_bytes(sync); + + for (sync = s->next; sync; sync = sync->next) + request_bytes(sync); } -static void command_flush_or_trigger_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) { +static void command_trigger_or_prebuf_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) { struct connection *c = userdata; uint32_t idx; struct playback_stream *s; @@ -1546,23 +1701,26 @@ static void command_flush_or_trigger_playback_stream(PA_GCC_UNUSED pa_pdispatch } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(s = pa_idxset_get_by_index(c->output_streams, idx)) || s->type != PLAYBACK_STREAM) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } - if (command == PA_COMMAND_PREBUF_PLAYBACK_STREAM) - pa_memblockq_prebuf_reenable(s->memblockq); - else if (command == PA_COMMAND_TRIGGER_PLAYBACK_STREAM) - pa_memblockq_prebuf_disable(s->memblockq); - else { - assert(command == PA_COMMAND_FLUSH_PLAYBACK_STREAM); - pa_memblockq_flush(s->memblockq); - /*pa_log(__FILE__": flush: %u\n", pa_memblockq_get_length(s->memblockq));*/ + switch (command) { + case PA_COMMAND_PREBUF_PLAYBACK_STREAM: + pa_memblockq_prebuf_force(s->memblockq); + break; + + case PA_COMMAND_TRIGGER_PLAYBACK_STREAM: + pa_memblockq_prebuf_disable(s->memblockq); + break; + + default: + abort(); } pa_sink_notify(s->sink_input->sink); @@ -1585,16 +1743,17 @@ static void command_cork_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UN } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(s = pa_idxset_get_by_index(c->record_streams, idx))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } pa_source_output_cork(s->source_output, b); + pa_memblockq_prebuf_force(s->memblockq); pa_pstream_send_simple_ack(c->pstream, tag); } @@ -1611,12 +1770,12 @@ static void command_flush_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_U } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(s = pa_idxset_get_by_index(c->record_streams, idx))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1638,7 +1797,7 @@ static void command_set_default_sink_or_source(PA_GCC_UNUSED pa_pdispatch *pd, u } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1660,7 +1819,7 @@ static void command_set_stream_name(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t com } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1668,7 +1827,7 @@ static void command_set_stream_name(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t com struct playback_stream *s; if (!(s = pa_idxset_get_by_index(c->output_streams, idx)) || s->type != PLAYBACK_STREAM) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1678,7 +1837,7 @@ static void command_set_stream_name(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t com struct record_stream *s; if (!(s = pa_idxset_get_by_index(c->record_streams, idx))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1700,7 +1859,7 @@ static void command_kill(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint3 } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1708,7 +1867,7 @@ static void command_kill(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint3 pa_client *client; if (!(client = pa_idxset_get_by_index(c->protocol->core->clients, idx))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1717,7 +1876,7 @@ static void command_kill(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint3 pa_sink_input *s; if (!(s = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1728,7 +1887,7 @@ static void command_kill(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint3 assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT); if (!(s = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1753,12 +1912,12 @@ static void command_load_module(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED ui } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(m = pa_module_load(c->protocol->core, name, argument))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_INITFAILED); + pa_pstream_send_error(c->pstream, tag, PA_ERR_MODINITFAILED); return; } @@ -1782,12 +1941,12 @@ static void command_unload_module(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (!(m = pa_idxset_get_by_index(c->protocol->core->modules, idx))) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1813,12 +1972,12 @@ static void command_add_autoload(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED u } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } if (pa_autoload_add(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE, module, argument, &idx) < 0) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST); + pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST); return; } @@ -1847,7 +2006,7 @@ static void command_remove_autoload(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSE } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1857,7 +2016,7 @@ static void command_remove_autoload(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSE r = pa_autoload_remove_by_index(c->protocol->core, idx); if (r < 0) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1893,7 +2052,7 @@ static void command_get_autoload_info(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNU } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1904,7 +2063,7 @@ static void command_get_autoload_info(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNU a = pa_autoload_get_by_index(c->protocol->core, idx); if (!a) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY); + pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY); return; } @@ -1927,7 +2086,7 @@ static void command_get_autoload_info_list(PA_GCC_UNUSED pa_pdispatch *pd, PA_GC } if (!c->authorized) { - pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS); + pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS); return; } @@ -1958,7 +2117,7 @@ static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, void *user } } -static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk, void *userdata) { +static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata) { struct connection *c = userdata; struct output_stream *stream; assert(p && chunk && userdata); @@ -1975,13 +2134,30 @@ static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, uint32_t ps->requested_bytes = 0; else ps->requested_bytes -= chunk->length; - - pa_memblockq_push_align(ps->memblockq, chunk, delta); - assert(ps->sink_input); -/* pa_log(__FILE__": after_recv: %u\n", pa_memblockq_get_length(p->memblockq)); */ + pa_memblockq_seek(ps->memblockq, offset, seek); + + if (pa_memblockq_push_align(ps->memblockq, chunk) < 0) { + pa_tagstruct *t; + + pa_log_warn(__FILE__": failed to push data into queue\n"); + + /* Pushing this block into the queue failed, so we simulate + * it by skipping ahead */ + + pa_memblockq_seek(ps->memblockq, chunk->length, PA_SEEK_RELATIVE); + + /* Notify the user */ + t = pa_tagstruct_new(NULL, 0); + pa_tagstruct_putu32(t, PA_COMMAND_OVERFLOW); + pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */ + pa_tagstruct_putu32(t, ps->index); + pa_pstream_send_tagstruct(p, t); + } + + ps->underrun = 0; + pa_sink_notify(ps->sink_input->sink); -/* pa_log(__FILE__": Recieved %u bytes.\n", chunk->length); */ } else { struct upload_stream *u = (struct upload_stream*) stream; diff --git a/src/polypcore/protocol-simple.c b/src/polypcore/protocol-simple.c index 4d3f8e1d..fac54239 100644 --- a/src/polypcore/protocol-simple.c +++ b/src/polypcore/protocol-simple.c @@ -52,6 +52,8 @@ struct connection { pa_memblockq *input_memblockq, *output_memblockq; pa_defer_event *defer_event; + int dead; + struct { pa_memblock *current_memblock; size_t memblock_index, fragment_size; @@ -130,7 +132,7 @@ static int do_read(struct connection *c) { } if ((r = pa_iochannel_read(c->io, (uint8_t*) c->playback.current_memblock->data+c->playback.memblock_index, l)) <= 0) { - pa_log(__FILE__": read() failed: %s\n", r == 0 ? "EOF" : strerror(errno)); + pa_log_debug(__FILE__": read() failed: %s\n", r == 0 ? "EOF" : strerror(errno)); return -1; } @@ -142,7 +144,7 @@ static int do_read(struct connection *c) { c->playback.memblock_index += r; assert(c->input_memblockq); - pa_memblockq_push_align(c->input_memblockq, &chunk, 0); + pa_memblockq_push_align(c->input_memblockq, &chunk); assert(c->sink_input); pa_sink_notify(c->sink_input->sink); @@ -170,32 +172,46 @@ static int do_write(struct connection *c) { pa_memblockq_drop(c->output_memblockq, &chunk, r); pa_memblock_unref(chunk.memblock); + + pa_source_notify(c->source_output->source); return 0; } - static void do_work(struct connection *c) { assert(c); assert(c->protocol && c->protocol->core && c->protocol->core->mainloop && c->protocol->core->mainloop->defer_enable); c->protocol->core->mainloop->defer_enable(c->defer_event, 0); - if (pa_iochannel_is_writable(c->io)) - if (do_write(c) < 0) - goto fail; + if (c->dead) + return; - if (pa_iochannel_is_readable(c->io)) + if (pa_iochannel_is_readable(c->io)) { if (do_read(c) < 0) goto fail; + } else if (pa_iochannel_is_hungup(c->io)) + goto fail; - if (pa_iochannel_is_hungup(c->io)) - c->protocol->core->mainloop->defer_enable(c->defer_event, 1); + if (pa_iochannel_is_writable(c->io)) { + if (do_write(c) < 0) + goto fail; + } return; fail: - connection_free(c); + + if (c->sink_input) { + c->dead = 1; + + pa_iochannel_free(c->io); + c->io = NULL; + + pa_memblockq_prebuf_disable(c->input_memblockq); + pa_sink_notify(c->sink_input->sink); + } else + connection_free(c); } /*** sink_input callbacks ***/ @@ -205,8 +221,13 @@ static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk) { assert(i && i->userdata && chunk); c = i->userdata; - if (pa_memblockq_peek(c->input_memblockq, chunk) < 0) + if (pa_memblockq_peek(c->input_memblockq, chunk) < 0) { + + if (c->dead) + connection_free(c); + return -1; + } return 0; } @@ -240,7 +261,7 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) struct connection *c = o->userdata; assert(o && c && chunk); - pa_memblockq_push(c->output_memblockq, chunk, 0); + pa_memblockq_push(c->output_memblockq, chunk); /* do something */ assert(c->protocol && c->protocol->core && c->protocol->core->mainloop && c->protocol->core->mainloop->defer_enable); @@ -307,6 +328,7 @@ static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata) c->playback.current_memblock = NULL; c->playback.memblock_index = 0; c->playback.fragment_size = 0; + c->dead = 0; pa_iochannel_socket_peer_to_string(io, cname, sizeof(cname)); c->client = pa_client_new(p->core, __FILE__, cname); @@ -339,7 +361,15 @@ static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata) c->sink_input->userdata = c; l = (size_t) (pa_bytes_per_second(&p->sample_spec)*PLAYBACK_BUFFER_SECONDS); - c->input_memblockq = pa_memblockq_new(l, 0, pa_frame_size(&p->sample_spec), l/2, l/PLAYBACK_BUFFER_FRAGMENTS, p->core->memblock_stat); + c->input_memblockq = pa_memblockq_new( + 0, + l, + 0, + pa_frame_size(&p->sample_spec), + (size_t) -1, + l/PLAYBACK_BUFFER_FRAGMENTS, + NULL, + p->core->memblock_stat); assert(c->input_memblockq); pa_iochannel_socket_set_rcvbuf(io, l/PLAYBACK_BUFFER_FRAGMENTS*5); c->playback.fragment_size = l/10; @@ -368,7 +398,15 @@ static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata) c->source_output->userdata = c; l = (size_t) (pa_bytes_per_second(&p->sample_spec)*RECORD_BUFFER_SECONDS); - c->output_memblockq = pa_memblockq_new(l, 0, pa_frame_size(&p->sample_spec), 0, 0, p->core->memblock_stat); + c->output_memblockq = pa_memblockq_new( + 0, + l, + 0, + pa_frame_size(&p->sample_spec), + 1, + 0, + NULL, + p->core->memblock_stat); pa_iochannel_socket_set_sndbuf(io, l/RECORD_BUFFER_FRAGMENTS*2); } diff --git a/src/polypcore/pstream.c b/src/polypcore/pstream.c index eec93a0f..c697dc3d 100644 --- a/src/polypcore/pstream.c +++ b/src/polypcore/pstream.c @@ -40,12 +40,14 @@ #include "pstream.h" -typedef enum pa_pstream_descriptor_index { +enum { PA_PSTREAM_DESCRIPTOR_LENGTH, PA_PSTREAM_DESCRIPTOR_CHANNEL, - PA_PSTREAM_DESCRIPTOR_DELTA, + PA_PSTREAM_DESCRIPTOR_OFFSET_HI, + PA_PSTREAM_DESCRIPTOR_OFFSET_LO, + PA_PSTREAM_DESCRIPTOR_SEEK, PA_PSTREAM_DESCRIPTOR_MAX -} pa_pstream_descriptor_index; +}; typedef uint32_t pa_pstream_descriptor[PA_PSTREAM_DESCRIPTOR_MAX]; @@ -58,7 +60,8 @@ struct item_info { /* memblock info */ pa_memchunk chunk; uint32_t channel; - uint32_t delta; + int64_t offset; + pa_seek_mode_t seek_mode; /* packet info */ pa_packet *packet; @@ -94,7 +97,7 @@ struct pa_pstream { void (*recieve_packet_callback) (pa_pstream *p, pa_packet *packet, void *userdata); void *recieve_packet_callback_userdata; - void (*recieve_memblock_callback) (pa_pstream *p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk, void *userdata); + void (*recieve_memblock_callback) (pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata); void *recieve_memblock_callback_userdata; void (*drain_callback)(pa_pstream *p, void *userdata); @@ -103,8 +106,8 @@ struct pa_pstream { pa_memblock_stat *memblock_stat; }; -static void do_write(pa_pstream *p); -static void do_read(pa_pstream *p); +static int do_write(pa_pstream *p); +static int do_read(pa_pstream *p); static void do_something(pa_pstream *p) { assert(p); @@ -112,31 +115,47 @@ static void do_something(pa_pstream *p) { p->mainloop->defer_enable(p->defer_event, 0); pa_pstream_ref(p); - - if (!p->dead && pa_iochannel_is_readable(p->io)) - do_read(p); - if (!p->dead && pa_iochannel_is_writable(p->io)) - do_write(p); + if (!p->dead && pa_iochannel_is_readable(p->io)) { + if (do_read(p) < 0) + goto fail; + } else if (!p->dead && pa_iochannel_is_hungup(p->io)) + goto fail; + + if (!p->dead && pa_iochannel_is_writable(p->io)) { + if (do_write(p) < 0) + goto fail; + } + + pa_pstream_unref(p); + return; + +fail: - /* In case the line was hungup, make sure to rerun this function - as soon as possible, until all data has been read. */ + p->dead = 1; - if (!p->dead && pa_iochannel_is_hungup(p->io)) - p->mainloop->defer_enable(p->defer_event, 1); + if (p->die_callback) + p->die_callback(p, p->die_callback_userdata); pa_pstream_unref(p); } static void io_callback(pa_iochannel*io, void *userdata) { pa_pstream *p = userdata; - assert(p && p->io == io); + + assert(p); + assert(p->io == io); + do_something(p); } static void defer_callback(pa_mainloop_api *m, pa_defer_event *e, void*userdata) { pa_pstream *p = userdata; - assert(p && p->defer_event == e && p->mainloop == m); + + assert(p); + assert(p->defer_event == e); + assert(p->mainloop == m); + do_something(p); } @@ -144,7 +163,8 @@ pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_memblock_sta pa_pstream *p; assert(io); - p = pa_xmalloc(sizeof(pa_pstream)); + p = pa_xnew(pa_pstream, 1); + p->ref = 1; p->io = io; pa_iochannel_set_callback(io, io_callback, p); @@ -228,7 +248,7 @@ void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet) { /* pa_log(__FILE__": push-packet %p\n", packet); */ - i = pa_xmalloc(sizeof(struct item_info)); + i = pa_xnew(struct item_info, 1); i->type = PA_PSTREAM_ITEM_PACKET; i->packet = pa_packet_ref(packet); @@ -236,7 +256,7 @@ void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet) { p->mainloop->defer_enable(p->defer_event, 1); } -void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk) { +void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa_seek_mode_t seek_mode, const pa_memchunk *chunk) { struct item_info *i; assert(p && channel != (uint32_t) -1 && chunk && p->ref >= 1); @@ -245,11 +265,12 @@ void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, uint32_t delta, co /* pa_log(__FILE__": push-memblock %p\n", chunk); */ - i = pa_xmalloc(sizeof(struct item_info)); + i = pa_xnew(struct item_info, 1); i->type = PA_PSTREAM_ITEM_MEMBLOCK; i->chunk = *chunk; i->channel = channel; - i->delta = delta; + i->offset = offset; + i->seek_mode = seek_mode; pa_memblock_ref(i->chunk.memblock); @@ -264,7 +285,7 @@ void pa_pstream_set_recieve_packet_callback(pa_pstream *p, void (*callback) (pa_ p->recieve_packet_callback_userdata = userdata; } -void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, void (*callback) (pa_pstream *p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk, void *userdata), void *userdata) { +void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, void (*callback) (pa_pstream *p, uint32_t channel, int64_t delta, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata), void *userdata) { assert(p && callback); p->recieve_memblock_callback = callback; @@ -286,17 +307,21 @@ static void prepare_next_write_item(pa_pstream *p) { 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_CHANNEL] = htonl((uint32_t) -1); - p->write.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA] = 0; + p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = 0; + p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0; + p->write.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK] = 0; } else { assert(p->write.current->type == PA_PSTREAM_ITEM_MEMBLOCK && p->write.current->chunk.memblock); p->write.data = (uint8_t*) p->write.current->chunk.memblock->data + p->write.current->chunk.index; p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->chunk.length); p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl(p->write.current->channel); - p->write.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA] = htonl(p->write.current->delta); + 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)); + p->write.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK] = htonl(p->write.current->seek_mode); } } -static void do_write(pa_pstream *p) { +static int do_write(pa_pstream *p) { void *d; size_t l; ssize_t r; @@ -306,7 +331,7 @@ static void do_write(pa_pstream *p) { prepare_next_write_item(p); if (!p->write.current) - return; + return 0; assert(p->write.data); @@ -319,7 +344,7 @@ static void do_write(pa_pstream *p) { } if ((r = pa_iochannel_write(p->io, d, l)) < 0) - goto die; + return -1; p->write.index += r; @@ -332,15 +357,10 @@ static void do_write(pa_pstream *p) { p->drain_callback(p, p->drain_userdata); } - return; - -die: - p->dead = 1; - if (p->die_callback) - p->die_callback(p, p->die_callback_userdata); + return 0; } -static void do_read(pa_pstream *p) { +static int do_read(pa_pstream *p) { void *d; size_t l; ssize_t r; @@ -356,7 +376,7 @@ static void do_read(pa_pstream *p) { } if ((r = pa_iochannel_read(p->io, d, l)) <= 0) - goto die; + return -1; p->read.index += r; @@ -365,8 +385,8 @@ static void do_read(pa_pstream *p) { /* Frame size too large */ if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) > FRAME_SIZE_MAX) { - pa_log(__FILE__": Frame size too large\n"); - goto die; + pa_log_warn(__FILE__": Frame size too large\n"); + return -1; } assert(!p->read.packet && !p->read.memblock); @@ -374,13 +394,16 @@ static void do_read(pa_pstream *p) { if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]) == (uint32_t) -1) { /* Frame is a packet frame */ p->read.packet = pa_packet_new(ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH])); - assert(p->read.packet); p->read.data = p->read.packet->data; } else { /* Frame is a memblock frame */ p->read.memblock = pa_memblock_new(ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]), p->memblock_stat); - assert(p->read.memblock); p->read.data = p->read.memblock->data; + + if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK]) > PA_SEEK_RELATIVE_END) { + pa_log_warn(__FILE__": Invalid seek mode\n"); + return -1; + } } } else if (p->read.index > PA_PSTREAM_DESCRIPTOR_SIZE) { @@ -396,13 +419,26 @@ static void do_read(pa_pstream *p) { chunk.index = p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE - l; chunk.length = l; - if (p->recieve_memblock_callback) + if (p->recieve_memblock_callback) { + int64_t offset; + + 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]), - ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA]), + offset, + ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK]), &chunk, p->recieve_memblock_callback_userdata); + } + + /* Drop seek info for following callbacks */ + p->read.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK] = + p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = + p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0; } } @@ -427,13 +463,7 @@ static void do_read(pa_pstream *p) { } } - return; - -die: - p->dead = 1; - if (p->die_callback) - p->die_callback(p, p->die_callback_userdata); - + return 0; } void pa_pstream_set_die_callback(pa_pstream *p, void (*callback)(pa_pstream *p, void *userdata), void *userdata) { @@ -453,20 +483,24 @@ int pa_pstream_is_pending(pa_pstream *p) { void pa_pstream_set_drain_callback(pa_pstream *p, void (*cb)(pa_pstream *p, void *userdata), void *userdata) { assert(p); + assert(p->ref >= 1); p->drain_callback = cb; p->drain_userdata = userdata; } void pa_pstream_unref(pa_pstream*p) { - assert(p && p->ref >= 1); + assert(p); + assert(p->ref >= 1); - if (!(--(p->ref))) + if (--p->ref == 0) pstream_free(p); } pa_pstream* pa_pstream_ref(pa_pstream*p) { - assert(p && p->ref >= 1); + assert(p); + assert(p->ref >= 1); + p->ref++; return p; } diff --git a/src/polypcore/pstream.h b/src/polypcore/pstream.h index 10ce58f6..741ba9b5 100644 --- a/src/polypcore/pstream.h +++ b/src/polypcore/pstream.h @@ -25,6 +25,7 @@ #include <inttypes.h> #include <polyp/mainloop-api.h> +#include <polyp/def.h> #include <polypcore/packet.h> #include <polypcore/memblock.h> #include <polypcore/iochannel.h> @@ -37,10 +38,10 @@ void pa_pstream_unref(pa_pstream*p); pa_pstream* pa_pstream_ref(pa_pstream*p); void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet); -void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk); +void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk); void pa_pstream_set_recieve_packet_callback(pa_pstream *p, void (*callback) (pa_pstream *p, pa_packet *packet, void *userdata), void *userdata); -void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, void (*callback) (pa_pstream *p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk, void *userdata), void *userdata); +void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, void (*callback) (pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata), void *userdata); void pa_pstream_set_drain_callback(pa_pstream *p, void (*cb)(pa_pstream *p, void *userdata), void *userdata); void pa_pstream_set_die_callback(pa_pstream *p, void (*callback)(pa_pstream *p, void *userdata), void *userdata); diff --git a/src/polypcore/sample-util.c b/src/polypcore/sample-util.c index e3bb4aa6..e588446d 100644 --- a/src/polypcore/sample-util.c +++ b/src/polypcore/sample-util.c @@ -34,6 +34,15 @@ #include "sample-util.h" +pa_memblock *pa_silence_memblock_new(const pa_sample_spec *spec, size_t length, pa_memblock_stat*s) { + assert(spec); + + if (length == 0) + length = pa_bytes_per_second(spec)/10; /* 100 ms */ + + return pa_silence_memblock(pa_memblock_new(length, s), spec); +} + pa_memblock *pa_silence_memblock(pa_memblock* b, const pa_sample_spec *spec) { assert(b && b->data && spec); pa_silence_memory(b->data, b->length, spec); diff --git a/src/polypcore/sample-util.h b/src/polypcore/sample-util.h index 486d284b..7ea01a30 100644 --- a/src/polypcore/sample-util.h +++ b/src/polypcore/sample-util.h @@ -28,6 +28,7 @@ #include <polypcore/memchunk.h> pa_memblock *pa_silence_memblock(pa_memblock* b, const pa_sample_spec *spec); +pa_memblock *pa_silence_memblock_new(const pa_sample_spec *spec, size_t length, pa_memblock_stat*s); void pa_silence_memchunk(pa_memchunk *c, const pa_sample_spec *spec); void pa_silence_memory(void *p, size_t length, const pa_sample_spec *spec); diff --git a/src/polypcore/sink.c b/src/polypcore/sink.c index 9bc478c3..1f374f5e 100644 --- a/src/polypcore/sink.c +++ b/src/polypcore/sink.c @@ -270,6 +270,8 @@ int pa_sink_render(pa_sink*s, size_t length, pa_memchunk *result) { result->memblock = pa_memblock_new(length, s->core->memblock_stat); assert(result->memblock); +/* pa_log("mixing %i\n", n); */ + result->length = pa_mix(info, n, result->memblock->data, length, &s->sample_spec, &s->sw_volume); result->index = 0; } diff --git a/src/polypcore/sink.h b/src/polypcore/sink.h index 5fd9784f..fa120ebf 100644 --- a/src/polypcore/sink.h +++ b/src/polypcore/sink.h @@ -34,7 +34,7 @@ typedef struct pa_sink pa_sink; #include <polypcore/source.h> #include <polypcore/module.h> -#define PA_MAX_INPUTS_PER_SINK 6 +#define PA_MAX_INPUTS_PER_SINK 32 typedef enum pa_sink_state { PA_SINK_RUNNING, diff --git a/src/tests/memblockq-test.c b/src/tests/memblockq-test.c new file mode 100644 index 00000000..b01084da --- /dev/null +++ b/src/tests/memblockq-test.c @@ -0,0 +1,147 @@ +/* $Id$ */ + +/*** + This file is part of polypaudio. + + polypaudio 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 of the License, + or (at your option) any later version. + + polypaudio 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 + General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with polypaudio; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 + USA. +***/ + +#ifdef HAVE_CONFIG_H +#include <config.h> +#endif + +#include <stdlib.h> +#include <assert.h> +#include <stdio.h> + +#include <polypcore/memblockq.h> +#include <polypcore/log.h> + +int main(int argc, char *argv[]) { + int ret; + pa_memblockq *bq; + pa_memchunk chunk1, chunk2, chunk3, chunk4; + pa_memblock *silence; + + pa_log_set_maximal_level(PA_LOG_DEBUG); + + silence = pa_memblock_new_fixed("__", 2, 1, NULL); + assert(silence); + + bq = pa_memblockq_new(0, 40, 10, 2, 4, 4, silence, NULL); + assert(bq); + + chunk1.memblock = pa_memblock_new_fixed("AA", 2, 1, NULL); + chunk1.index = 0; + chunk1.length = 2; + assert(chunk1.memblock); + + chunk2.memblock = pa_memblock_new_fixed("TTBB", 4, 1, NULL); + chunk2.index = 2; + chunk2.length = 2; + assert(chunk2.memblock); + + chunk3.memblock = pa_memblock_new_fixed("ZZZZ", 4, 1, NULL); + chunk3.index = 0; + chunk3.length = 4; + assert(chunk3.memblock); + + chunk4.memblock = pa_memblock_new_fixed("KKKKKKKK", 8, 1, NULL); + chunk4.index = 0; + chunk4.length = 8; + assert(chunk4.memblock); + + ret = pa_memblockq_push(bq, &chunk1); + assert(ret == 0); + + ret = pa_memblockq_push(bq, &chunk1); + assert(ret == 0); + + ret = pa_memblockq_push(bq, &chunk2); + assert(ret == 0); + + ret = pa_memblockq_push(bq, &chunk2); + assert(ret == 0); + + pa_memblockq_seek(bq, -6, 0); + ret = pa_memblockq_push(bq, &chunk3); + assert(ret == 0); + + pa_memblockq_seek(bq, -2, 0); + ret = pa_memblockq_push(bq, &chunk3); + assert(ret == 0); + + pa_memblockq_seek(bq, -10, 0); + ret = pa_memblockq_push(bq, &chunk4); + assert(ret == 0); + + pa_memblockq_seek(bq, 10, 0); + + ret = pa_memblockq_push(bq, &chunk1); + assert(ret == 0); + + pa_memblockq_seek(bq, -6, 0); + ret = pa_memblockq_push(bq, &chunk2); + assert(ret == 0); + + /* Test splitting */ + pa_memblockq_seek(bq, -12, 0); + ret = pa_memblockq_push(bq, &chunk1); + assert(ret == 0); + + pa_memblockq_seek(bq, 20, 0); + + /* Test merging */ + ret = pa_memblockq_push(bq, &chunk3); + assert(ret == 0); + pa_memblockq_seek(bq, -2, 0); + + chunk3.index += 2; + chunk3.length -= 2; + + ret = pa_memblockq_push(bq, &chunk3); + assert(ret == 0); + + printf(">"); + + pa_memblockq_shorten(bq, 6); + + for (;;) { + pa_memchunk out; + char *e; + size_t n; + + if (pa_memblockq_peek(bq, &out) < 0) + break; + + for (e = (char*) out.memblock->data + out.index, n = 0; n < out.length; n++) + printf("%c", *e); + + pa_memblock_unref(out.memblock); + pa_memblockq_drop(bq, &out, out.length); + } + + printf("<\n"); + + pa_memblockq_free(bq); + pa_memblock_unref(silence); + pa_memblock_unref(chunk1.memblock); + pa_memblock_unref(chunk2.memblock); + pa_memblock_unref(chunk3.memblock); + pa_memblock_unref(chunk4.memblock); + + return 0; +} diff --git a/src/tests/sync-playback.c b/src/tests/sync-playback.c new file mode 100644 index 00000000..5df790c9 --- /dev/null +++ b/src/tests/sync-playback.c @@ -0,0 +1,192 @@ +/* $Id$ */ + +/*** + This file is part of polypaudio. + + polypaudio 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 of the License, + or (at your option) any later version. + + polypaudio 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 + General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with polypaudio; if not, write to the Free Software + Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 + USA. +***/ + +#ifdef HAVE_CONFIG_H +#include <config.h> +#endif + +#include <signal.h> +#include <string.h> +#include <errno.h> +#include <unistd.h> +#include <assert.h> +#include <stdio.h> +#include <stdlib.h> +#include <getopt.h> +#include <math.h> + +#include <polyp/polypaudio.h> +#include <polyp/mainloop.h> + +#define NSTREAMS 4 +#define SINE_HZ 440 +#define SAMPLE_HZ 8000 + +static pa_context *context = NULL; +static pa_stream *streams[NSTREAMS]; +static pa_mainloop_api *mainloop_api = NULL; + +static float data[SAMPLE_HZ]; /* one second space */ + +static int n_streams_ready = 0; + +static const pa_sample_spec sample_spec = { + .format = PA_SAMPLE_FLOAT32, + .rate = SAMPLE_HZ, + .channels = 1 +}; + +static const pa_buffer_attr buffer_attr = { + .maxlength = SAMPLE_HZ*sizeof(float)*NSTREAMS, /* exactly space for the entire play time */ + .tlength = 0, + .prebuf = 0, /* Setting prebuf to 0 guarantees us the the streams will run synchronously, no matter what */ + .minreq = 0 +}; + +static void nop_free_cb(void *p) {} + +static void underflow_cb(struct pa_stream *s, void *userdata) { + int i = (int) userdata; + + fprintf(stderr, "Stream %i finished\n", i); + + if (++n_streams_ready >= 2*NSTREAMS) { + fprintf(stderr, "We're done\n"); + mainloop_api->quit(mainloop_api, 0); + } +} + +/* This routine is called whenever the stream state changes */ +static void stream_state_callback(pa_stream *s, void *userdata) { + assert(s); + + switch (pa_stream_get_state(s)) { + case PA_STREAM_UNCONNECTED: + case PA_STREAM_CREATING: + case PA_STREAM_TERMINATED: + break; + + case PA_STREAM_READY: { + + int r, i = (int) userdata; + + fprintf(stderr, "Writing data to stream %i.\n", i); + + r = pa_stream_write(s, data, sizeof(data), nop_free_cb, sizeof(data) * i, PA_SEEK_ABSOLUTE); + assert(r == 0); + + /* Be notified when this stream is drained */ + pa_stream_set_underflow_callback(s, underflow_cb, userdata); + + /* All streams have been set up, let's go! */ + if (++n_streams_ready >= NSTREAMS) { + fprintf(stderr, "Uncorking\n"); + pa_operation_unref(pa_stream_cork(s, 0, NULL, NULL)); + } + + break; + } + + default: + case PA_STREAM_FAILED: + fprintf(stderr, "Stream error: %s\n", pa_strerror(pa_context_errno(pa_stream_get_context(s)))); + abort(); + } +} + +/* This is called whenever the context status changes */ +static void context_state_callback(pa_context *c, void *userdata) { + assert(c); + + switch (pa_context_get_state(c)) { + case PA_CONTEXT_CONNECTING: + case PA_CONTEXT_AUTHORIZING: + case PA_CONTEXT_SETTING_NAME: + break; + + case PA_CONTEXT_READY: { + + int i; + fprintf(stderr, "Connection established.\n"); + + for (i = 0; i < NSTREAMS; i++) { + char name[64]; + + fprintf(stderr, "Creating stream %i\n", i); + + snprintf(name, sizeof(name), "stream #%i", i); + + streams[i] = pa_stream_new(c, name, &sample_spec, NULL); + assert(streams[i]); + pa_stream_set_state_callback(streams[i], stream_state_callback, (void*) i); + pa_stream_connect_playback(streams[i], NULL, &buffer_attr, PA_STREAM_START_CORKED, NULL, i == 0 ? NULL : streams[0]); + } + + break; + } + + case PA_CONTEXT_TERMINATED: + mainloop_api->quit(mainloop_api, 0); + break; + + case PA_CONTEXT_FAILED: + default: + fprintf(stderr, "Context error: %s\n", pa_strerror(pa_context_errno(c))); + abort(); + } +} + +int main(int argc, char *argv[]) { + pa_mainloop* m = NULL; + int i, ret = 0; + + for (i = 0; i < SAMPLE_HZ; i++) + data[i] = (float) sin(((double) i/SAMPLE_HZ)*2*M_PI*SINE_HZ)/2; + + for (i = 0; i < NSTREAMS; i++) + streams[i] = NULL; + + /* Set up a new main loop */ + m = pa_mainloop_new(); + assert(m); + + mainloop_api = pa_mainloop_get_api(m); + + context = pa_context_new(mainloop_api, argv[0]); + assert(context); + + pa_context_set_state_callback(context, context_state_callback, NULL); + + pa_context_connect(context, NULL, 1, NULL); + + if (pa_mainloop_run(m, &ret) < 0) + fprintf(stderr, "pa_mainloop_run() failed.\n"); + + pa_context_unref(context); + + for (i = 0; i < NSTREAMS; i++) + if (streams[i]) + pa_stream_unref(streams[i]); + + pa_mainloop_free(m); + + return ret; +} diff --git a/src/utils/pacat.c b/src/utils/pacat.c index a3c3f2c8..4e126c8a 100644 --- a/src/utils/pacat.c +++ b/src/utils/pacat.c @@ -80,7 +80,7 @@ static void do_stream_write(size_t length) { if (l > buffer_length) l = buffer_length; - pa_stream_write(stream, (uint8_t*) buffer + buffer_index, l, NULL, 0); + pa_stream_write(stream, (uint8_t*) buffer + buffer_index, l, NULL, 0, PA_SEEK_RELATIVE); buffer_length -= l; buffer_index += l; @@ -106,8 +106,8 @@ static void stream_write_callback(pa_stream *s, size_t length, void *userdata) { /* This is called whenever new data may is available */ static void stream_read_callback(pa_stream *s, size_t length, void *userdata) { + const void *data; assert(s && length); - void *data; if (stdio_event) mainloop_api->io_enable(stdio_event, PA_IO_EVENT_OUTPUT); @@ -175,7 +175,7 @@ static void context_state_callback(pa_context *c, void *userdata) { if (mode == PLAYBACK) { pa_cvolume cv; - pa_stream_connect_playback(stream, device, NULL, 0, pa_cvolume_set(&cv, PA_CHANNELS_MAX, volume)); + pa_stream_connect_playback(stream, device, NULL, 0, pa_cvolume_set(&cv, PA_CHANNELS_MAX, volume), NULL); } else pa_stream_connect_record(stream, device, NULL, 0); diff --git a/src/utils/pactl.c b/src/utils/pactl.c index 4c22c925..e3305f0f 100644 --- a/src/utils/pactl.c +++ b/src/utils/pactl.c @@ -515,7 +515,7 @@ static void stream_write_callback(pa_stream *s, size_t length, void *userdata) { quit(1); } - pa_stream_write(s, d, length, free, 0); + pa_stream_write(s, d, length, free, 0, PA_SEEK_RELATIVE); sample_length -= length; diff --git a/src/utils/paplay.c b/src/utils/paplay.c index 9f73b834..5f985ee9 100644 --- a/src/utils/paplay.c +++ b/src/utils/paplay.c @@ -113,7 +113,7 @@ static void stream_write_callback(pa_stream *s, size_t length, void *userdata) { f = readf_function(sndfile, data, n); if (f > 0) - pa_stream_write(s, data, f*k, free, 0); + pa_stream_write(s, data, f*k, free, 0, PA_SEEK_RELATIVE); if (f < n) { sf_close(sndfile); @@ -166,7 +166,7 @@ static void context_state_callback(pa_context *c, void *userdata) { pa_stream_set_state_callback(stream, stream_state_callback, NULL); pa_stream_set_write_callback(stream, stream_write_callback, NULL); - pa_stream_connect_playback(stream, device, NULL, 0, pa_cvolume_set(&cv, PA_CHANNELS_MAX, volume)); + pa_stream_connect_playback(stream, device, NULL, 0, pa_cvolume_set(&cv, sample_spec.channels, volume), NULL); break; } |