mirrors/pulseaudio
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pulseaudio/pulseaudio.git synced 2025-11-02 09:01:46 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

rework latency querying API (this needs more testing)

Browse source 
git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@648 fefdeb5f-60dc-0310-8127-8f9354f1896f
This commit is contained in:
Lennart Poettering 2006-04-07 00:24:48 +00:00
parent cc302f2d17
commit 920f045380
4 changed files with 326 additions and 223 deletions
Download patch file Download diff file Expand all files Collapse all files

44
src/polyp/def.h
View file

@ -80,7 +80,7 @@ typedef enum pa_stream_direction {
/** Some special flags for stream connections. \since 0.6 */ /** Some special flags for stream connections. \since 0.6 */
typedef enum pa_stream_flags { typedef enum pa_stream_flags {
PA_STREAM_START_CORKED = 1, /**< Create the stream corked, requiring an explicit pa_stream_cork() call to uncork it. */ PA_STREAM_START_CORKED = 1, /**< Create the stream corked, requiring an explicit pa_stream_cork() call to uncork it. */
PA_STREAM_INTERPOLATE_LATENCY = 2 /**< Interpolate the latency for PA_STREAM_INTERPOLATE_LATENCY = 2, /**< Interpolate the latency for
* this stream. When enabled, * this stream. When enabled,
* you can use * you can use
* pa_stream_interpolated_xxx() * pa_stream_interpolated_xxx()
@ -95,6 +95,7 @@ typedef enum pa_stream_flags {
* information. This is * information. This is
* especially useful on long latency * especially useful on long latency
* network connections. */ * network connections. */
PA_STREAM_NOT_MONOTONOUS = 4, /**< Don't force the time to run monotonically */
} pa_stream_flags_t; } pa_stream_flags_t;
/** Playback and record buffer metrics */ /** Playback and record buffer metrics */
@ -124,7 +125,7 @@ enum {
PA_ERR_INVALIDSERVER, /**< Invalid server */ PA_ERR_INVALIDSERVER, /**< Invalid server */
PA_ERR_MODINITFAILED, /**< Module initialization failed */ PA_ERR_MODINITFAILED, /**< Module initialization failed */
PA_ERR_BADSTATE, /**< Bad state */ PA_ERR_BADSTATE, /**< Bad state */
PA_ERR_NODATA, /**< No data */ PA_ERR_NODATA, /**< No data */
PA_ERR_VERSION, /**< Incompatible protocol version \since 0.8 */ PA_ERR_VERSION, /**< Incompatible protocol version \since 0.8 */
PA_ERR_MAX /**< Not really an error but the first invalid error code */ PA_ERR_MAX /**< Not really an error but the first invalid error code */
}; };
@ -171,7 +172,7 @@ typedef enum pa_subscription_event_type {
* pa_stream_write() takes to be played may be estimated by * pa_stream_write() takes to be played may be estimated by
* sink_usec+buffer_usec+transport_usec. The output buffer to which * sink_usec+buffer_usec+transport_usec. The output buffer to which
* buffer_usec relates may be manipulated freely (with * buffer_usec relates may be manipulated freely (with
* pa_stream_write()'s delta argument, pa_stream_flush() and friends), * pa_stream_write()'s seek argument, pa_stream_flush() and friends),
* the buffers sink_usec/source_usec relates to is a first-in * the buffers sink_usec/source_usec relates to is a first-in
* first-out buffer which cannot be flushed or manipulated in any * first-out buffer which cannot be flushed or manipulated in any
* way. The total input latency a sample that is recorded takes to be * way. The total input latency a sample that is recorded takes to be
@ -180,12 +181,7 @@ typedef enum pa_subscription_event_type {
* sign issues!) When connected to a monitor source sink_usec contains * sign issues!) When connected to a monitor source sink_usec contains
* the latency of the owning sink.*/ * the latency of the owning sink.*/
typedef struct pa_latency_info { typedef struct pa_latency_info {
pa_usec_t buffer_usec; /**< Time in usecs the current buffer takes to play. For both playback and record streams. */ struct timeval timestamp; /**< The time when this latency info was current */
pa_usec_t sink_usec; /**< Time in usecs a sample takes to be played on the sink. For playback streams and record streams connected to a monitor source. */
pa_usec_t source_usec; /**< Time in usecs a sample takes from being recorded to being delivered to the application. Only for record streams. \since 0.5*/
pa_usec_t transport_usec; /**< Estimated time in usecs a sample takes to be transferred to/from the daemon. For both playback and record streams. \since 0.5 */
int playing; /**< Non-zero when the stream is currently playing. Only for playback streams. */
uint32_t queue_length; /**< Queue size in bytes. For both playback and record streams. */
int synchronized_clocks; /**< Non-zero if the local and the int synchronized_clocks; /**< Non-zero if the local and the
* remote machine have synchronized * remote machine have synchronized
* clocks. If synchronized clocks are * clocks. If synchronized clocks are
@ -194,10 +190,31 @@ typedef struct pa_latency_info {
* detects synchronized clocks is very * detects synchronized clocks is very
* limited und unreliable itself. \since * limited und unreliable itself. \since
* 0.5 */ * 0.5 */
struct timeval timestamp; /**< The time when this latency info was current */
uint64_t counter; /**< The byte counter current when the latency info was requested. \since 0.6 */ pa_usec_t buffer_usec; /**< Time in usecs the current buffer takes to play. For both playback and record streams. */
int64_t write_index; /**< Current absolute write index in the buffer. \since 0.8 */ pa_usec_t sink_usec; /**< Time in usecs a sample takes to be played on the sink. For playback streams and record streams connected to a monitor source. */
int64_t read_index; /**< Current absolute read index in the buffer. \since 0.8 */ pa_usec_t source_usec; /**< Time in usecs a sample takes from being recorded to being delivered to the application. Only for record streams. \since 0.5*/
pa_usec_t transport_usec; /**< Estimated time in usecs a sample takes to be transferred to/from the daemon. For both playback and record streams. \since 0.5 */
int playing; /**< Non-zero when the stream is currently playing. Only for playback streams. */
int write_index_corrupt; /**< Non-Zero if the write_index is not up to date because a local write command corrupted it */
int64_t write_index; /**< Current write index into the
* playback buffer in bytes. Think twice before
* using this for seeking purposes: it
* might be out of date a the time you
* want to use it. Consider using
* PA_SEEK_RELATIVE instead. \since
* 0.8 */
int64_t read_index; /**< Current read index into the
* playback buffer in bytes. Think twice before
* using this for seeking purposes: it
* might be out of date a the time you
* want to use it. Consider using
* PA_SEEK_RELATIVE_ON_READ
* instead. \since 0.8 */
uint32_t buffer_length; /* Current buffer length. This is usually identical to write_index-read_index. */
} pa_latency_info; } pa_latency_info;
/** A structure for the spawn api. This may be used to integrate auto /** A structure for the spawn api. This may be used to integrate auto
@ -227,7 +244,6 @@ typedef enum pa_seek_mode {
PA_SEEK_RELATIVE_END = 3, /**< Seek relatively to the current end of the buffer queue */ PA_SEEK_RELATIVE_END = 3, /**< Seek relatively to the current end of the buffer queue */
} pa_seek_mode_t; } pa_seek_mode_t;
PA_C_DECL_END PA_C_DECL_END
#endif #endif

44
src/polyp/internal.h
View file

@ -83,6 +83,15 @@ struct pa_context {
pa_client_conf *conf; pa_client_conf *conf;
}; };
#define PA_MAX_LATENCY_CORRECTIONS 10
typedef struct pa_latency_correction {
uint32_t tag;
int valid;
int64_t value;
int absolute, corrupt;
} pa_latency_correction;
struct pa_stream { struct pa_stream {
int ref; int ref;
pa_context *context; pa_context *context;
@ -93,41 +102,48 @@ struct pa_stream {
pa_buffer_attr buffer_attr; pa_buffer_attr buffer_attr;
pa_sample_spec sample_spec; pa_sample_spec sample_spec;
pa_channel_map channel_map; pa_channel_map channel_map;
pa_stream_flags_t flags;
uint32_t channel; uint32_t channel;
uint32_t syncid; uint32_t syncid;
int channel_valid; int channel_valid;
uint32_t device_index; uint32_t device_index;
pa_stream_direction_t direction; pa_stream_direction_t direction;
uint32_t requested_bytes;
uint64_t counter;
pa_usec_t previous_time;
pa_usec_t previous_ipol_time;
pa_stream_state_t state; pa_stream_state_t state;
uint32_t requested_bytes;
pa_memchunk peek_memchunk; pa_memchunk peek_memchunk;
pa_memblockq *record_memblockq; pa_memblockq *record_memblockq;
pa_hashmap *counter_hashmap;
int interpolate;
int corked; int corked;
uint32_t ipol_usec; /* Store latest latency info */
struct timeval ipol_timestamp; pa_latency_info latency_info;
int latency_info_valid;
/* Use to make sure that time advances monotonically */
pa_usec_t previous_time;
/* Latency correction stuff */
pa_latency_correction latency_corrections[PA_MAX_LATENCY_CORRECTIONS];
int idx_latency_correction;
/* Latency interpolation stuff */
pa_time_event *ipol_event; pa_time_event *ipol_event;
int ipol_requested; int ipol_requested;
pa_usec_t ipol_usec;
int ipol_usec_valid;
struct timeval ipol_timestamp;
/* Callbacks */
pa_stream_notify_cb_t state_callback; pa_stream_notify_cb_t state_callback;
void *state_userdata; void *state_userdata;
pa_stream_request_cb_t read_callback; pa_stream_request_cb_t read_callback;
void *read_userdata; void *read_userdata;
pa_stream_request_cb_t write_callback; pa_stream_request_cb_t write_callback;
void *write_userdata; void *write_userdata;
pa_stream_notify_cb_t overflow_callback; pa_stream_notify_cb_t overflow_callback;
void *overflow_userdata; void *overflow_userdata;
pa_stream_notify_cb_t underflow_callback; pa_stream_notify_cb_t underflow_callback;
void *underflow_userdata; void *underflow_userdata;
}; };

433
src/polyp/stream.c
View file

@ -38,10 +38,10 @@
#include "internal.h" #include "internal.h"
#define LATENCY_IPOL_INTERVAL_USEC (10000L) #define LATENCY_IPOL_INTERVAL_USEC (10000L)
#define COUNTER_HASHMAP_MAXSIZE (5)
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) {
pa_stream *s; pa_stream *s;
int i;
assert(c); assert(c);
@ -67,6 +67,7 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
s->direction = PA_STREAM_NODIRECTION; s->direction = PA_STREAM_NODIRECTION;
s->name = pa_xstrdup(name); s->name = pa_xstrdup(name);
s->sample_spec = *ss; s->sample_spec = *ss;
s->flags = 0;
if (map) if (map)
s->channel_map = *map; s->channel_map = *map;
@ -87,20 +88,21 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
s->record_memblockq = NULL; s->record_memblockq = NULL;
s->counter_hashmap = pa_hashmap_new(NULL, NULL);
s->counter = 0;
s->previous_time = 0; s->previous_time = 0;
s->previous_ipol_time = 0; s->latency_info_valid = 0;
s->corked = 0; s->corked = 0;
s->interpolate = 0;
s->ipol_usec = 0; s->ipol_usec_valid = 0;
memset(&s->ipol_timestamp, 0, sizeof(s->ipol_timestamp)); s->ipol_timestamp.tv_sec = 0;
s->ipol_timestamp.tv_usec = 0;
s->ipol_event = NULL; s->ipol_event = NULL;
s->ipol_requested = 0; s->ipol_requested = 0;
for (i = 0; i < PA_MAX_LATENCY_CORRECTIONS; i++)
s->latency_corrections[i].valid = 0;
s->idx_latency_correction = 0;
PA_LLIST_PREPEND(pa_stream, c->streams, s); PA_LLIST_PREPEND(pa_stream, c->streams, s);
/* The context and stream will point at each other. We cannot ref count /* The context and stream will point at each other. We cannot ref count
@ -110,10 +112,6 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
return s; return s;
} }
static void hashmap_free_func(void *p, void *userdata) {
pa_xfree(p);
}
static void stream_free(pa_stream *s) { static void stream_free(pa_stream *s) {
assert(s && s->context && !s->channel_valid); assert(s && s->context && !s->channel_valid);
@ -132,9 +130,6 @@ static void stream_free(pa_stream *s) {
if (s->record_memblockq) if (s->record_memblockq)
pa_memblockq_free(s->record_memblockq); pa_memblockq_free(s->record_memblockq);
if (s->counter_hashmap)
pa_hashmap_free(s->counter_hashmap, hashmap_free_func, NULL);
pa_xfree(s->name); pa_xfree(s->name);
pa_xfree(s); pa_xfree(s);
} }
@ -315,10 +310,11 @@ static void ipol_callback(pa_mainloop_api *m, pa_time_event *e, PA_GCC_UNUSED co
if (s->state == PA_STREAM_READY && !s->ipol_requested) { if (s->state == PA_STREAM_READY && !s->ipol_requested) {
pa_operation *o; pa_operation *o;
o = pa_stream_get_latency_info(s, NULL, NULL);
if (o) if ((o = pa_stream_update_latency_info(s, NULL, NULL))) {
pa_operation_unref(o); pa_operation_unref(o);
s->ipol_requested = 1; s->ipol_requested = 1;
}
} }
pa_gettimeofday(&next); pa_gettimeofday(&next);
@ -375,7 +371,7 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
/* We add an extra ref as long as we're connected (i.e. in the dynarray) */ /* We add an extra ref as long as we're connected (i.e. in the dynarray) */
pa_stream_ref(s); pa_stream_ref(s);
if (s->interpolate) { if (s->flags & PA_STREAM_INTERPOLATE_LATENCY) {
struct timeval tv; struct timeval tv;
pa_gettimeofday(&tv); pa_gettimeofday(&tv);
@ -418,13 +414,11 @@ static int create_stream(
pa_stream_ref(s); pa_stream_ref(s);
s->direction = direction; s->direction = direction;
s->flags = flags;
if (sync_stream) if (sync_stream)
s->syncid = sync_stream->syncid; s->syncid = sync_stream->syncid;
s->interpolate = !!(flags & PA_STREAM_INTERPOLATE_LATENCY);
pa_stream_trash_ipol(s);
if (attr) if (attr)
s->buffer_attr = *attr; s->buffer_attr = *attr;
else { else {
@ -548,7 +542,33 @@ int pa_stream_write(
else else
s->requested_bytes = 0; s->requested_bytes = 0;
s->counter += length; /* Update latency request correction */
if (s->latency_corrections[s->idx_latency_correction].valid) {
if (seek == PA_SEEK_ABSOLUTE) {
s->latency_corrections[s->idx_latency_correction].corrupt = 0;
s->latency_corrections[s->idx_latency_correction].absolute = 1;
s->latency_corrections[s->idx_latency_correction].value = offset + length;
} else if (seek == PA_SEEK_RELATIVE) {
if (!s->latency_corrections[s->idx_latency_correction].corrupt)
s->latency_corrections[s->idx_latency_correction].value += offset + length;
} else
s->latency_corrections[s->idx_latency_correction].corrupt = 1;
}
/* Update the write index in the already available latency data */
if (s->latency_info_valid) {
if (seek == PA_SEEK_ABSOLUTE) {
s->latency_info.write_index_corrupt = 0;
s->latency_info.write_index = offset + length;
} else if (seek == PA_SEEK_RELATIVE) {
if (!s->latency_info.write_index_corrupt)
s->latency_info.write_index += offset + length;
} else
s->latency_info.write_index_corrupt = 1;
}
return 0; return 0;
} }
@ -590,7 +610,6 @@ int pa_stream_drop(pa_stream *s) {
s->peek_memchunk.index = 0; s->peek_memchunk.index = 0;
s->peek_memchunk.memblock = NULL; s->peek_memchunk.memblock = NULL;
s->counter += s->peek_memchunk.length;
return 0; return 0;
} }
@ -637,33 +656,34 @@ pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *us
static void stream_get_latency_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) { static void stream_get_latency_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
pa_operation *o = userdata; pa_operation *o = userdata;
pa_latency_info i, *p = NULL;
struct timeval local, remote, now; struct timeval local, remote, now;
pa_latency_info *i;
assert(pd); assert(pd);
assert(o); assert(o);
assert(o->stream); assert(o->stream);
assert(o->context); assert(o->context);
i.counter = *(uint64_t*)pa_hashmap_get(o->stream->counter_hashmap, (void*)(unsigned long)tag); i = &o->stream->latency_info;
pa_xfree(pa_hashmap_remove(o->stream->counter_hashmap, (void*)(unsigned long)tag)); o->stream->latency_info_valid = 0;
i->write_index_corrupt = 0;
if (command != PA_COMMAND_REPLY) { if (command != PA_COMMAND_REPLY) {
if (pa_context_handle_error(o->context, command, t) < 0) if (pa_context_handle_error(o->context, command, t) < 0)
goto finish; goto finish;
} else if (pa_tagstruct_get_usec(t, &i.buffer_usec) < 0 || } else if (pa_tagstruct_get_usec(t, &i->buffer_usec) < 0 ||
pa_tagstruct_get_usec(t, &i.sink_usec) < 0 || pa_tagstruct_get_usec(t, &i->sink_usec) < 0 ||
pa_tagstruct_get_usec(t, &i.source_usec) < 0 || pa_tagstruct_get_usec(t, &i->source_usec) < 0 ||
pa_tagstruct_get_boolean(t, &i.playing) < 0 || pa_tagstruct_get_boolean(t, &i->playing) < 0 ||
pa_tagstruct_getu32(t, &i.queue_length) < 0 ||
pa_tagstruct_get_timeval(t, &local) < 0 || pa_tagstruct_get_timeval(t, &local) < 0 ||
pa_tagstruct_get_timeval(t, &remote) < 0 || pa_tagstruct_get_timeval(t, &remote) < 0 ||
pa_tagstruct_gets64(t, &i.write_index) < 0 || pa_tagstruct_gets64(t, &i->write_index) < 0 ||
pa_tagstruct_gets64(t, &i.read_index) < 0 || pa_tagstruct_gets64(t, &i->read_index) < 0 ||
!pa_tagstruct_eof(t)) { !pa_tagstruct_eof(t)) {
pa_context_fail(o->context, PA_ERR_PROTOCOL); pa_context_fail(o->context, PA_ERR_PROTOCOL);
goto finish; goto finish;
} else { } else {
pa_gettimeofday(&now); pa_gettimeofday(&now);
@ -671,72 +691,125 @@ static void stream_get_latency_info_callback(pa_pdispatch *pd, uint32_t command,
/* local and remote seem to have synchronized clocks */ /* local and remote seem to have synchronized clocks */
if (o->stream->direction == PA_STREAM_PLAYBACK) if (o->stream->direction == PA_STREAM_PLAYBACK)
i.transport_usec = pa_timeval_diff(&remote, &local); i->transport_usec = pa_timeval_diff(&remote, &local);
else else
i.transport_usec = pa_timeval_diff(&now, &remote); i->transport_usec = pa_timeval_diff(&now, &remote);
i.synchronized_clocks = 1; i->synchronized_clocks = 1;
i.timestamp = remote; i->timestamp = remote;
} else { } else {
/* clocks are not synchronized, let's estimate latency then */ /* clocks are not synchronized, let's estimate latency then */
i.transport_usec = pa_timeval_diff(&now, &local)/2; i->transport_usec = pa_timeval_diff(&now, &local)/2;
i.synchronized_clocks = 0; i->synchronized_clocks = 0;
i.timestamp = local; i->timestamp = local;
pa_timeval_add(&i.timestamp, i.transport_usec); pa_timeval_add(&i->timestamp, i->transport_usec);
}
if (o->stream->interpolate) {
/* pa_log("new interpol data"); */
o->stream->ipol_timestamp = i.timestamp;
o->stream->ipol_usec = pa_stream_get_time(o->stream, &i);
o->stream->ipol_requested = 0;
} }
p = &i; if (o->stream->direction == PA_STREAM_PLAYBACK) {
/* Write index correction */
int n, j;
uint32_t ctag = tag;
/* Go through the saved correction values and add up the total correction.*/
for (n = 0, j = o->stream->idx_latency_correction;
n < PA_MAX_LATENCY_CORRECTIONS;
n++, j = (j + 1) % PA_MAX_LATENCY_CORRECTIONS) {
/* Step over invalid data or out-of-date data */
if (!o->stream->latency_corrections[j].valid ||
o->stream->latency_corrections[j].tag < ctag)
continue;
/* Make sure that everything is in order */
ctag = o->stream->latency_corrections[j].tag+1;
/* Now fix the write index */
if (o->stream->latency_corrections[j].corrupt) {
/* A corrupting seek was made */
i->write_index = 0;
i->write_index_corrupt = 1;
} else if (o->stream->latency_corrections[j].absolute) {
/* An absolute seek was made */
i->write_index = o->stream->latency_corrections[j].value;
i->write_index_corrupt = 0;
} else if (!i->write_index_corrupt) {
/* A relative seek was made */
i->write_index += o->stream->latency_corrections[j].value;
}
}
}
o->stream->latency_info_valid = 1;
o->stream->ipol_timestamp = now;
o->stream->ipol_usec_valid = 0;
}
/* Clear old correction entries */
if (o->stream->direction == PA_STREAM_PLAYBACK) {
int n;
for (n = 0; n < PA_MAX_LATENCY_CORRECTIONS; n++) {
if (!o->stream->latency_corrections[n].valid)
continue;
if (o->stream->latency_corrections[n].tag <= tag)
o->stream->latency_corrections[n].valid = 0;
}
} }
if (o->callback) { if (o->callback) {
pa_stream_get_latency_info_cb_t cb = (pa_stream_get_latency_info_cb_t) o->callback; pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
cb(o->stream, p, o->userdata); cb(o->stream, o->stream->latency_info_valid, o->userdata);
} }
finish: finish:
pa_operation_done(o); pa_operation_done(o);
pa_operation_unref(o); pa_operation_unref(o);
} }
pa_operation* pa_stream_get_latency_info(pa_stream *s, pa_stream_get_latency_info_cb_t cb, void *userdata) { pa_operation* pa_stream_update_latency_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
uint32_t tag; uint32_t tag;
pa_operation *o; pa_operation *o;
pa_tagstruct *t; pa_tagstruct *t;
struct timeval now; struct timeval now;
uint64_t *counter; int cidx;
assert(s); assert(s);
assert(s->ref >= 1); 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->state == PA_STREAM_READY, PA_ERR_BADSTATE);
PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE); PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
PA_CHECK_VALIDITY_RETURN_NULL(s->context, pa_hashmap_size(s->counter_hashmap) < COUNTER_HASHMAP_MAXSIZE, PA_ERR_INTERNAL);
/* Find a place to store the write_index correction data for this entry */
cidx = (s->idx_latency_correction + 1) % PA_MAX_LATENCY_CORRECTIONS;
/* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->latency_corrections[cidx].valid, PA_ERR_INTERNAL);
o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata); o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
t = pa_tagstruct_command( t = pa_tagstruct_command(
s->context, s->context,
s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY,
&tag); &tag);
pa_tagstruct_putu32(t, s->channel); pa_tagstruct_putu32(t, s->channel);
pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
pa_gettimeofday(&now);
pa_tagstruct_put_timeval(t, &now);
pa_pstream_send_tagstruct(s->context->pstream, t); pa_pstream_send_tagstruct(s->context->pstream, t);
pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_info_callback, o); pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_info_callback, o);
counter = pa_xmalloc(sizeof(uint64_t)); /* Fill in initial correction data */
*counter = s->counter; o->stream->idx_latency_correction = cidx;
pa_hashmap_put(s->counter_hashmap, (void*)(unsigned long)tag, counter); o->stream->latency_corrections[cidx].valid = 1;
o->stream->latency_corrections[cidx].tag = tag;
o->stream->latency_corrections[cidx].absolute = 0;
o->stream->latency_corrections[cidx].value = 0;
o->stream->latency_corrections[cidx].corrupt = 0;
return pa_operation_ref(o); return pa_operation_ref(o);
} }
@ -872,11 +945,11 @@ pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, voi
PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE); 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); PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
if (s->interpolate) { if (s->flags & PA_STREAM_INTERPOLATE_LATENCY) {
if (!s->corked && b) if (!s->corked && b) {
/* Pausing */ /* Refresh the interpolated data just befor pausing */
s->ipol_usec = pa_stream_get_interpolated_time(s); pa_stream_get_time(s, NULL);
else if (s->corked && !b) } else if (s->corked && !b)
/* Unpausing */ /* Unpausing */
pa_gettimeofday(&s->ipol_timestamp); pa_gettimeofday(&s->ipol_timestamp);
} }
@ -894,8 +967,7 @@ pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, voi
pa_pstream_send_tagstruct(s->context->pstream, t); pa_pstream_send_tagstruct(s->context->pstream, t);
pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o); pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
lo = pa_stream_get_latency_info(s, NULL, NULL); if ((lo = pa_stream_update_latency_info(s, NULL, NULL)))
if (lo)
pa_operation_unref(lo); pa_operation_unref(lo);
return pa_operation_ref(o); return pa_operation_ref(o);
@ -928,8 +1000,8 @@ pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *use
if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) { if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) {
pa_operation *lo; pa_operation *lo;
lo = pa_stream_get_latency_info(s, NULL, NULL);
if (lo) if ((lo = pa_stream_update_latency_info(s, NULL, NULL)))
pa_operation_unref(lo); pa_operation_unref(lo);
} }
@ -943,8 +1015,8 @@ pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *us
if ((o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata))) { if ((o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata))) {
pa_operation *lo; pa_operation *lo;
lo = pa_stream_get_latency_info(s, NULL, NULL);
if (lo) if ((lo = pa_stream_update_latency_info(s, NULL, NULL)))
pa_operation_unref(lo); pa_operation_unref(lo);
} }
@ -958,8 +1030,8 @@ pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *u
if ((o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata))) { if ((o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata))) {
pa_operation *lo; pa_operation *lo;
lo = pa_stream_get_latency_info(s, NULL, NULL);
if (lo) if ((lo = pa_stream_update_latency_info(s, NULL, NULL)))
pa_operation_unref(lo); pa_operation_unref(lo);
} }
@ -992,85 +1064,151 @@ pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_succe
return pa_operation_ref(o); return pa_operation_ref(o);
} }
uint64_t pa_stream_get_counter(pa_stream *s) { int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
assert(s); pa_usec_t usec = 0;
assert(s->ref >= 1);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (uint64_t) -1);
return s->counter;
}
pa_usec_t pa_stream_get_time(pa_stream *s, const pa_latency_info *i) {
pa_usec_t usec;
assert(s); assert(s);
assert(s->ref >= 1); assert(s->ref >= 1);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (pa_usec_t) -1); PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE, (pa_usec_t) -1); PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
PA_CHECK_VALIDITY(s->context, s->latency_info_valid, PA_ERR_NODATA);
usec = pa_bytes_to_usec(i->counter, &s->sample_spec);
if (i) { if (s->flags & PA_STREAM_INTERPOLATE_LATENCY && s->ipol_usec_valid )
usec = s->ipol_usec;
else {
if (s->direction == PA_STREAM_PLAYBACK) { if (s->direction == PA_STREAM_PLAYBACK) {
pa_usec_t latency = i->transport_usec + i->buffer_usec + i->sink_usec; /* The last byte that was written into the output device
if (usec < latency) * had this time value associated */
usec = pa_bytes_to_usec(s->latency_info.read_index < 0 ? 0 : (uint64_t) s->latency_info.read_index, &s->sample_spec);
/* Because the latency info took a little time to come
* to us, we assume that the real output time is actually
* a little ahead */
usec += s->latency_info.transport_usec;
/* However, the output device usually maintains a buffer
too, hence the real sample currently played is a little
back */
if (s->latency_info.sink_usec >= usec)
usec = 0; usec = 0;
else else
usec -= latency; usec -= s->latency_info.sink_usec;
} else if (s->direction == PA_STREAM_RECORD) { } else if (s->direction == PA_STREAM_RECORD) {
usec += i->source_usec + i->buffer_usec + i->transport_usec; /* The last byte written into the server side queue had
* this time value associated */
if (usec > i->sink_usec) usec = pa_bytes_to_usec(s->latency_info.write_index < 0 ? 0 : (uint64_t) s->latency_info.write_index, &s->sample_spec);
usec -= i->sink_usec;
else /* Add transport latency */
usec += s->latency_info.transport_usec;
/* Add latency of data in device buffer */
usec += s->latency_info.source_usec;
/* If this is a monitor source, we need to correct the
* time by the playback device buffer */
if (s->latency_info.sink_usec >= usec)
usec = 0; usec = 0;
else
usec -= s->latency_info.sink_usec;
}
if (s->flags & PA_STREAM_INTERPOLATE_LATENCY) {
s->ipol_usec_valid = 1;
s->ipol_usec = usec;
} }
} }
if (usec < s->previous_time) /* Interpolate if requested */
usec = s->previous_time; if (s->flags & PA_STREAM_INTERPOLATE_LATENCY) {
s->previous_time = usec; /* We just add the time that passed since the latency info was
* current */
if (!s->corked) {
struct timeval now;
usec += pa_timeval_diff(pa_gettimeofday(&now), &s->ipol_timestamp);
s->ipol_timestamp = now;
}
}
return usec; /* Make sure the time runs monotonically */
if (!(s->flags & PA_STREAM_NOT_MONOTONOUS)) {
if (usec < s->previous_time)
usec = s->previous_time;
else
s->previous_time = usec;
}
if (r_usec)
*r_usec = usec;
return 0;
} }
static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t t, pa_usec_t c, int *negative) { static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
assert(s); assert(s);
assert(s->ref >= 1); assert(s->ref >= 1);
if (negative) if (negative)
*negative = 0; *negative = 0;
if (c < t) { if (a >= b)
if (s->direction == PA_STREAM_RECORD) { return a-b;
if (negative) else {
*negative = 1; if (negative && s->direction == PA_STREAM_RECORD) {
*negative = 1;
return t-c; return b-a;
} else } else
return 0; return 0;
} else }
return c-t;
} }
pa_usec_t pa_stream_get_latency(pa_stream *s, const pa_latency_info *i, int *negative) { int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
pa_usec_t t, c; pa_usec_t t, c;
int r;
int64_t cindex;
assert(s); assert(s);
assert(s->ref >= 1); assert(s->ref >= 1);
assert(i); assert(r_usec);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (pa_usec_t) -1); PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE, (pa_usec_t) -1); PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
PA_CHECK_VALIDITY(s->context, s->latency_info_valid, PA_ERR_NODATA);
PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->latency_info.write_index_corrupt, PA_ERR_NODATA);
t = pa_stream_get_time(s, i); if ((r = pa_stream_get_time(s, &t)) < 0)
c = pa_bytes_to_usec(s->counter, &s->sample_spec); return r;
return time_counter_diff(s, t, c, negative); if (s->direction == PA_STREAM_PLAYBACK)
cindex = s->latency_info.write_index;
else
cindex = s->latency_info.read_index;
if (cindex < 0)
cindex = 0;
c = pa_bytes_to_usec(cindex, &s->sample_spec);
if (s->direction == PA_STREAM_PLAYBACK)
*r_usec = time_counter_diff(s, c, t, negative);
else
*r_usec = time_counter_diff(s, t, c, negative);
return 0;
}
const pa_latency_info* pa_stream_get_latency_info(pa_stream *s) {
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);
PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->latency_info_valid, PA_ERR_BADSTATE);
return &s->latency_info;
} }
const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) { const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
@ -1087,55 +1225,4 @@ const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
return &s->channel_map; return &s->channel_map;
} }
void pa_stream_trash_ipol(pa_stream *s) {
assert(s);
assert(s->ref >= 1);
if (!s->interpolate)
return;
memset(&s->ipol_timestamp, 0, sizeof(s->ipol_timestamp));
s->ipol_usec = 0;
}
pa_usec_t pa_stream_get_interpolated_time(pa_stream *s) {
pa_usec_t usec;
assert(s);
assert(s->ref >= 1);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (pa_usec_t) -1);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE, (pa_usec_t) -1);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->interpolate, PA_ERR_BADSTATE, (pa_usec_t) -1);
if (s->corked)
usec = s->ipol_usec;
else {
if (s->ipol_timestamp.tv_sec == 0)
usec = 0;
else
usec = s->ipol_usec + pa_timeval_age(&s->ipol_timestamp);
}
if (usec < s->previous_ipol_time)
usec = s->previous_ipol_time;
s->previous_ipol_time = usec;
return usec;
}
pa_usec_t pa_stream_get_interpolated_latency(pa_stream *s, int *negative) {
pa_usec_t t, c;
assert(s);
assert(s->ref >= 1);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (pa_usec_t) -1);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE, (pa_usec_t) -1);
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->interpolate, PA_ERR_BADSTATE, (pa_usec_t) -1);
t = pa_stream_get_interpolated_time(s);
c = pa_bytes_to_usec(s->counter, &s->sample_spec);
return time_counter_diff(s, t, c, negative);
}

28
src/polyp/stream.h
View file

@ -51,9 +51,6 @@ typedef void (*pa_stream_request_cb_t)(pa_stream *p, size_t length, void *userda
/** A generic notification callback */ /** A generic notification callback */
typedef void (*pa_stream_notify_cb_t)(pa_stream *p, void *userdata); 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 */ /** Create a new, unconnected stream with the specified name and sample type */
pa_stream* pa_stream_new( pa_stream* pa_stream_new(
pa_context *c /**< The context to create this stream in */, pa_context *c /**< The context to create this stream in */,
@ -133,8 +130,8 @@ size_t pa_stream_readable_size(pa_stream *p);
/** Drain a playback stream. Use this for notification when the buffer is empty */ /** 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); pa_operation* pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata);
/** Get the playback latency of a stream */ /** Update the latency info of a stream */
pa_operation* pa_stream_get_latency_info(pa_stream *p, pa_stream_get_latency_info_cb_t cby, void *userdata); pa_operation* pa_stream_update_latency_info(pa_stream *p, pa_stream_success_cb_t cb, void *userdata);
/** Set the callback function that is called whenever the state of the stream changes */ /** Set the callback function that is called whenever the state of the stream changes */
void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata); void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata);
@ -173,34 +170,21 @@ pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *u
/** Rename the stream. \since 0.5 */ /** Rename the stream. \since 0.5 */
pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, 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
* this yourself using pa_stream_reset_counter(). \since 0.6 */
uint64_t pa_stream_get_counter(pa_stream *s);
/** Return the current playback/recording time. This is based on the /** Return the current playback/recording time. This is based on the
* counter accessible with pa_stream_get_counter(). This function * counter accessible with pa_stream_get_counter(). This function
* requires a pa_latency_info structure as argument, which should be * requires a pa_latency_info structure as argument, which should be
* acquired using pa_stream_get_latency(). \since 0.6 */ * acquired using pa_stream_get_latency(). \since 0.6 */
pa_usec_t pa_stream_get_time(pa_stream *s, const pa_latency_info *i); int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec);
/** Return the total stream latency. Thus function requires a /** Return the total stream latency. Thus function requires a
* pa_latency_info structure as argument, which should be aquired * pa_latency_info structure as argument, which should be aquired
* using pa_stream_get_latency(). In case the stream is a monitoring * using pa_stream_get_latency(). In case the stream is a monitoring
* stream the result can be negative, i.e. the captured samples are * stream the result can be negative, i.e. the captured samples are
* not yet played. In this case *negative is set to 1. \since 0.6 */ * not yet played. In this case *negative is set to 1. \since 0.6 */
pa_usec_t pa_stream_get_latency(pa_stream *s, const pa_latency_info *i, int *negative); int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative);
/** Return the interpolated playback/recording time. Requires the /** Return the latest latency data. \since 0.8 */
* PA_STREAM_INTERPOLATE_LATENCY bit set when creating the stream. In const pa_latency_info* pa_stream_get_latency_info(pa_stream *s);
* contrast to pa_stream_get_latency() this function doesn't require
* a whole roundtrip for response. \since 0.6 */
pa_usec_t pa_stream_get_interpolated_time(pa_stream *s);
/** Return the interpolated playback/recording latency. Requires the
* PA_STREAM_INTERPOLATE_LATENCY bit set when creating the
* stream. \since 0.6 */
pa_usec_t pa_stream_get_interpolated_latency(pa_stream *s, int *negative);
/** Return a pointer to the stream's sample specification. \since 0.6 */ /** Return a pointer to the stream's sample specification. \since 0.6 */
const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s); const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s);