[pulseaudio-commits] [SCM] PulseAudio Sound Server branch, master, updated. add6c0361ada8c2aa632392efcec538a07af5bce
Lennart Poettering
gitmailer-noreply at 0pointer.de
Fri Jun 20 13:32:47 PDT 2008
This is an automated email from the git hooks/post-receive script. It was
generated because of a push to the "PulseAudio Sound Server" repository.
The master branch has been updated
from 11e55fe01fcf85f9954577795b0517eac8373390 (commit)
- Log -----------------------------------------------------------------
add6c03... Rework module-combine to work with glitch-free core; add new max_request field to pa_sink
1420e1d... fix interpretation of remix parameter
6c980c2... add new abstract device class
2b112fe... add new function pa_smoother_reset()
-----------------------------------------------------------------------
Summary of changes:
src/modules/module-alsa-sink.c | 3 +
src/modules/module-combine.c | 543 ++++++++++++++++++++------------------
src/modules/module-ladspa-sink.c | 36 +++-
src/modules/module-null-sink.c | 28 +-
src/modules/module-oss.c | 2 +
src/modules/module-remap-sink.c | 40 +++-
src/modules/module-tunnel.c | 6 +-
src/pulse/proplist.h | 2 +-
src/pulsecore/cli-text.c | 20 ++-
src/pulsecore/protocol-native.c | 10 +-
src/pulsecore/sink-input.c | 128 +++++++---
src/pulsecore/sink-input.h | 28 ++-
src/pulsecore/sink.c | 286 ++++++++++++++++-----
src/pulsecore/sink.h | 42 ++-
src/pulsecore/source-output.c | 89 ++++--
src/pulsecore/source-output.h | 27 ++-
src/pulsecore/source.c | 240 +++++++++++++----
src/pulsecore/source.h | 53 +++-
src/pulsecore/time-smoother.c | 8 +
src/pulsecore/time-smoother.h | 2 +
20 files changed, 1088 insertions(+), 505 deletions(-)
-----------------------------------------------------------------------
commit 2b112fe9db9005b3e445e7dc462f5ec7202b8fe3
Author: Lennart Poettering <lennart at poettering.net>
Date: Fri Jun 20 22:17:40 2008 +0200
add new function pa_smoother_reset()
diff --git a/src/pulsecore/time-smoother.c b/src/pulsecore/time-smoother.c
index 2723edb..fe5a4f1 100644
--- a/src/pulsecore/time-smoother.c
+++ b/src/pulsecore/time-smoother.c
@@ -449,3 +449,11 @@ pa_usec_t pa_smoother_translate(pa_smoother *s, pa_usec_t x, pa_usec_t y_delay)
return (pa_usec_t) ((double) y_delay / nde);
}
+
+void pa_smoother_reset(pa_smoother *s) {
+ pa_assert(s);
+
+ s->n_history = 0;
+ s->abc_valid = FALSE;
+
+}
diff --git a/src/pulsecore/time-smoother.h b/src/pulsecore/time-smoother.h
index 8f450d0..2051e64 100644
--- a/src/pulsecore/time-smoother.h
+++ b/src/pulsecore/time-smoother.h
@@ -44,4 +44,6 @@ void pa_smoother_set_time_offset(pa_smoother *s, pa_usec_t x_offset);
void pa_smoother_pause(pa_smoother *s, pa_usec_t x);
void pa_smoother_resume(pa_smoother *s, pa_usec_t x);
+void pa_smoother_reset(pa_smoother *s);
+
#endif
commit 6c980c27377487741e893d7d9c94828716acd3fb
Author: Lennart Poettering <lennart at poettering.net>
Date: Fri Jun 20 22:18:03 2008 +0200
add new abstract device class
diff --git a/src/pulse/proplist.h b/src/pulse/proplist.h
index 9d0647e..f75cca5 100644
--- a/src/pulse/proplist.h
+++ b/src/pulse/proplist.h
@@ -69,7 +69,7 @@ PA_C_DECL_BEGIN
* device.bus_path
* device.serial
* device.vendor_product_id
- * device.class sound, modem, monitor, filter
+ * device.class sound, modem, monitor, filter, abstract
* device.form_factor laptop-speakers, external-speakers, telephone, tv-capture, webcam-capture, microphone-capture, headset
* device.connector isa, pci, usb, firewire, bluetooth
* device.access_mode mmap, mmap_rewrite, serial
commit 1420e1d1b5430d1963f652972f2bf72fadfc561e
Author: Lennart Poettering <lennart at poettering.net>
Date: Fri Jun 20 22:29:30 2008 +0200
fix interpretation of remix parameter
diff --git a/src/modules/module-remap-sink.c b/src/modules/module-remap-sink.c
index b9e7301..b8c76b6 100644
--- a/src/modules/module-remap-sink.c
+++ b/src/modules/module-remap-sink.c
@@ -348,7 +348,7 @@ int pa__init(pa_module*m) {
pa_sink_input_new_data_set_sample_spec(&sink_input_data, &ss);
pa_sink_input_new_data_set_channel_map(&sink_input_data, &stream_map);
- u->sink_input = pa_sink_input_new(m->core, &sink_input_data, PA_SINK_INPUT_DONT_MOVE | (remix ? PA_SINK_INPUT_NO_REMIX : 0));
+ u->sink_input = pa_sink_input_new(m->core, &sink_input_data, PA_SINK_INPUT_DONT_MOVE | (remix ? 0 : PA_SINK_INPUT_NO_REMIX));
pa_sink_input_new_data_done(&sink_input_data);
if (!u->sink_input)
commit add6c0361ada8c2aa632392efcec538a07af5bce
Author: Lennart Poettering <lennart at poettering.net>
Date: Fri Jun 20 22:32:41 2008 +0200
Rework module-combine to work with glitch-free core; add new max_request field to pa_sink
diff --git a/src/modules/module-alsa-sink.c b/src/modules/module-alsa-sink.c
index 7bad073..6765775 100644
--- a/src/modules/module-alsa-sink.c
+++ b/src/modules/module-alsa-sink.c
@@ -600,6 +600,8 @@ static int update_sw_params(struct userdata *u) {
return err;
}
+ pa_sink_set_max_request(u->sink, u->hwbuf_size - u->hwbuf_unused_frames * u->frame_size);
+
return 0;
}
@@ -1316,6 +1318,7 @@ int pa__init(pa_module*m) {
fix_tsched_watermark(u);
u->sink->thread_info.max_rewind = use_tsched ? u->hwbuf_size : 0;
+ u->sink->thread_info.max_request = u->hwbuf_size;
pa_sink_set_latency_range(u->sink,
!use_tsched ? pa_bytes_to_usec(u->hwbuf_size, &ss) : (pa_usec_t) -1,
diff --git a/src/modules/module-combine.c b/src/modules/module-combine.c
index 0632257..a90d409 100644
--- a/src/modules/module-combine.c
+++ b/src/modules/module-combine.c
@@ -1,7 +1,7 @@
/***
This file is part of PulseAudio.
- Copyright 2004-2006 Lennart Poettering
+ Copyright 2004-2008 Lennart Poettering
PulseAudio is free software; you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published
@@ -45,6 +45,7 @@
#include <pulsecore/rtpoll.h>
#include <pulsecore/rtclock.h>
#include <pulsecore/core-error.h>
+#include <pulsecore/time-smoother.h>
#include "module-combine-symdef.h"
@@ -54,7 +55,6 @@ PA_MODULE_VERSION(PACKAGE_VERSION);
PA_MODULE_LOAD_ONCE(FALSE);
PA_MODULE_USAGE(
"sink_name=<name for the sink> "
- "master=<master sink> "
"slaves=<slave sinks> "
"adjust_time=<seconds> "
"resample_method=<method> "
@@ -64,13 +64,15 @@ PA_MODULE_USAGE(
"channel_map=<channel map>");
#define DEFAULT_SINK_NAME "combined"
+
#define MEMBLOCKQ_MAXLENGTH (1024*1024*16)
#define DEFAULT_ADJUST_TIME 10
+#define REQUEST_LATENCY_USEC (PA_USEC_PER_MSEC * 200)
+
static const char* const valid_modargs[] = {
"sink_name",
- "master",
"slaves",
"adjust_time",
"resample_method",
@@ -89,12 +91,15 @@ struct output {
pa_asyncmsgq *inq, /* Message queue from the sink thread to this sink input */
*outq; /* Message queue from this sink input to the sink thread */
- pa_rtpoll_item *inq_rtpoll_item, *outq_rtpoll_item;
+ pa_rtpoll_item *inq_rtpoll_item_read, *inq_rtpoll_item_write;
+ pa_rtpoll_item *outq_rtpoll_item_read, *outq_rtpoll_item_write;
pa_memblockq *memblockq;
pa_usec_t total_latency;
+ pa_atomic_t max_request;
+
PA_LLIST_FIELDS(struct output);
};
@@ -111,29 +116,33 @@ struct userdata {
uint32_t adjust_time;
pa_bool_t automatic;
- size_t block_size;
- pa_hook_slot *sink_new_slot, *sink_unlink_slot, *sink_state_changed_slot;
+ pa_hook_slot *sink_put_slot, *sink_unlink_slot, *sink_state_changed_slot;
pa_resample_method_t resample_method;
struct timeval adjust_timestamp;
- struct output *master;
+ pa_usec_t block_usec;
+
pa_idxset* outputs; /* managed in main context */
struct {
PA_LLIST_HEAD(struct output, active_outputs); /* managed in IO thread context */
pa_atomic_t running; /* we cache that value here, so that every thread can query it cheaply */
- struct timeval timestamp;
+ pa_usec_t timestamp;
pa_bool_t in_null_mode;
+ pa_smoother *smoother;
+ uint64_t counter;
} thread_info;
};
enum {
SINK_MESSAGE_ADD_OUTPUT = PA_SINK_MESSAGE_MAX,
SINK_MESSAGE_REMOVE_OUTPUT,
- SINK_MESSAGE_NEED
+ SINK_MESSAGE_NEED,
+ SINK_MESSAGE_UPDATE_LATENCY,
+ SINK_MESSAGE_UPDATE_MAX_REQUEST
};
enum {
@@ -142,14 +151,13 @@ enum {
static void output_free(struct output *o);
static int output_create_sink_input(struct output *o);
-static void update_master(struct userdata *u, struct output *o);
-static void pick_master(struct userdata *u, struct output *except);
static void adjust_rates(struct userdata *u) {
struct output *o;
- pa_usec_t max_sink_latency = 0, min_total_latency = (pa_usec_t) -1, target_latency;
+ pa_usec_t max_sink_latency = 0, min_total_latency = (pa_usec_t) -1, target_latency, avg_total_latency = 0;
uint32_t base_rate;
uint32_t idx;
+ unsigned n = 0;
pa_assert(u);
pa_sink_assert_ref(u->sink);
@@ -157,9 +165,6 @@ static void adjust_rates(struct userdata *u) {
if (pa_idxset_size(u->outputs) <= 0)
return;
- if (!u->master)
- return;
-
if (!PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)))
return;
@@ -169,23 +174,28 @@ static void adjust_rates(struct userdata *u) {
if (!o->sink_input || !PA_SINK_IS_OPENED(pa_sink_get_state(o->sink)))
continue;
- sink_latency = pa_sink_get_latency(o->sink);
- o->total_latency = sink_latency + pa_sink_input_get_latency(o->sink_input);
+ o->total_latency = pa_sink_input_get_latency(o->sink_input, &sink_latency);
+ o->total_latency += sink_latency;
if (sink_latency > max_sink_latency)
max_sink_latency = sink_latency;
if (min_total_latency == (pa_usec_t) -1 || o->total_latency < min_total_latency)
min_total_latency = o->total_latency;
+
+ avg_total_latency += o->total_latency;
+ n++;
}
if (min_total_latency == (pa_usec_t) -1)
return;
+ avg_total_latency /= n;
+
target_latency = max_sink_latency > min_total_latency ? max_sink_latency : min_total_latency;
- pa_log_info("[%s] target latency is %0.0f usec.", u->sink->name, (float) target_latency);
- pa_log_info("[%s] master %s latency %0.0f usec.", u->sink->name, u->master->sink->name, (float) u->master->total_latency);
+ pa_log_info("[%s] avg total latency is %0.2f msec.", u->sink->name, (double) avg_total_latency / PA_USEC_PER_MSEC);
+ pa_log_info("[%s] target latency is %0.2f msec.", u->sink->name, (double) target_latency / PA_USEC_PER_MSEC);
base_rate = u->sink->sample_spec.rate;
@@ -208,6 +218,8 @@ static void adjust_rates(struct userdata *u) {
pa_sink_input_set_rate(o->sink_input, r);
}
}
+
+ pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UPDATE_LATENCY, NULL, (int64_t) avg_total_latency, NULL);
}
static void time_callback(pa_mainloop_api*a, pa_time_event* e, const struct timeval *tv, void *userdata) {
@@ -225,6 +237,36 @@ static void time_callback(pa_mainloop_api*a, pa_time_event* e, const struct time
u->sink->core->mainloop->time_restart(e, &n);
}
+static void process_render_null(struct userdata *u, pa_usec_t now) {
+ size_t ate = 0;
+ pa_assert(u);
+
+ if (u->thread_info.in_null_mode)
+ u->thread_info.timestamp = now;
+
+ while (u->thread_info.timestamp < now + u->block_usec) {
+ pa_memchunk chunk;
+
+ pa_sink_render(u->sink, u->sink->thread_info.max_request, &chunk);
+ pa_memblock_unref(chunk.memblock);
+
+ u->thread_info.counter += chunk.length;
+
+/* pa_log_debug("Ate %lu bytes.", (unsigned long) chunk.length); */
+ u->thread_info.timestamp += pa_bytes_to_usec(chunk.length, &u->sink->sample_spec);
+
+ ate += chunk.length;
+
+ if (ate >= u->sink->thread_info.max_request)
+ break;
+ }
+
+/* pa_log_debug("Ate in sum %lu bytes (of %lu)", (unsigned long) ate, (unsigned long) nbytes); */
+
+ pa_smoother_put(u->thread_info.smoother, now,
+ pa_bytes_to_usec(u->thread_info.counter, &u->sink->sample_spec) - (u->thread_info.timestamp - now));
+}
+
static void thread_func(void *userdata) {
struct userdata *u = userdata;
@@ -238,38 +280,23 @@ static void thread_func(void *userdata) {
pa_thread_mq_install(&u->thread_mq);
pa_rtpoll_install(u->rtpoll);
- pa_rtclock_get(&u->thread_info.timestamp);
+ u->thread_info.timestamp = pa_rtclock_usec();
u->thread_info.in_null_mode = FALSE;
for (;;) {
int ret;
/* If no outputs are connected, render some data and drop it immediately. */
- if (u->sink->thread_info.state == PA_SINK_RUNNING && !u->thread_info.active_outputs) {
- struct timeval now;
+ if (PA_SINK_IS_OPENED(u->sink->thread_info.state) && !u->thread_info.active_outputs) {
+ pa_usec_t now;
- /* Just rewind if necessary, since we are in NULL mode, we
- * don't have to pass this on */
- pa_sink_process_rewind(u->sink, u->sink->thread_info.rewind_nbytes);
- u->sink->thread_info.rewind_nbytes = 0;
+ now = pa_rtclock_usec();
- pa_rtclock_get(&now);
+ if (!u->thread_info.in_null_mode || u->thread_info.timestamp <= now)
+ process_render_null(u, now);
- if (!u->thread_info.in_null_mode || pa_timeval_cmp(&u->thread_info.timestamp, &now) <= 0) {
- pa_memchunk chunk;
-
- pa_sink_render_full(u->sink, u->block_size, &chunk);
- pa_memblock_unref(chunk.memblock);
-
- if (!u->thread_info.in_null_mode)
- u->thread_info.timestamp = now;
-
- pa_timeval_add(&u->thread_info.timestamp, pa_bytes_to_usec(u->block_size, &u->sink->sample_spec));
- }
-
- pa_rtpoll_set_timer_absolute(u->rtpoll, &u->thread_info.timestamp);
+ pa_rtpoll_set_timer_absolute(u->rtpoll, u->thread_info.timestamp);
u->thread_info.in_null_mode = TRUE;
-
} else {
pa_rtpoll_set_timer_disabled(u->rtpoll);
u->thread_info.in_null_mode = FALSE;
@@ -301,7 +328,7 @@ static void render_memblock(struct userdata *u, struct output *o, size_t length)
pa_assert(o);
/* We are run by the sink thread, on behalf of an output (o). The
- * other output is waiting for us, hence it is safe to access its
+ * output is waiting for us, hence it is safe to access its
* mainblockq and asyncmsgq directly. */
/* If we are not running, we cannot produce any data */
@@ -321,6 +348,8 @@ static void render_memblock(struct userdata *u, struct output *o, size_t length)
/* Render data! */
pa_sink_render(u->sink, length, &chunk);
+ u->thread_info.counter += chunk.length;
+
/* OK, let's send this data to the other threads */
for (j = u->thread_info.active_outputs; j; j = j->next)
@@ -377,6 +406,42 @@ static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk
}
/* Called from I/O thread context */
+static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes) {
+ struct output *o;
+
+ pa_sink_input_assert_ref(i);
+ pa_assert(nbytes > 0);
+ pa_assert_se(o = i->userdata);
+
+ pa_memblockq_rewind(o->memblockq, nbytes);
+}
+
+/* Called from I/O thread context */
+static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
+ struct output *o;
+
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(o = i->userdata);
+
+ pa_memblockq_set_maxrewind(o->memblockq, nbytes);
+}
+
+/* Called from I/O thread context */
+static void sink_input_update_max_request_cb(pa_sink_input *i, size_t nbytes) {
+ struct output *o;
+
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(o = i->userdata);
+
+ if (pa_atomic_load(&o->max_request) == (int) nbytes)
+ return;
+
+ pa_atomic_store(&o->max_request, (int) nbytes);
+
+ pa_asyncmsgq_post(o->outq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_UPDATE_MAX_REQUEST, NULL, 0, NULL, NULL);
+}
+
+/* Called from I/O thread context */
static void sink_input_attach_cb(pa_sink_input *i) {
struct output *o;
@@ -384,11 +449,17 @@ static void sink_input_attach_cb(pa_sink_input *i) {
pa_assert_se(o = i->userdata);
/* Set up the queue from the sink thread to us */
- pa_assert(!o->inq_rtpoll_item);
- o->inq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq_read(
+ pa_assert(!o->inq_rtpoll_item_read && !o->outq_rtpoll_item_write);
+
+ o->inq_rtpoll_item_read = pa_rtpoll_item_new_asyncmsgq_read(
i->sink->rtpoll,
PA_RTPOLL_LATE, /* This one is not that important, since we check for data in _peek() anyway. */
o->inq);
+
+ o->outq_rtpoll_item_write = pa_rtpoll_item_new_asyncmsgq_write(
+ i->sink->rtpoll,
+ PA_RTPOLL_EARLY,
+ o->outq);
}
/* Called from I/O thread context */
@@ -399,9 +470,13 @@ static void sink_input_detach_cb(pa_sink_input *i) {
pa_assert_se(o = i->userdata);
/* Shut down the queue from the sink thread to us */
- pa_assert(o->inq_rtpoll_item);
- pa_rtpoll_item_free(o->inq_rtpoll_item);
- o->inq_rtpoll_item = NULL;
+ pa_assert(o->inq_rtpoll_item_read && o->outq_rtpoll_item_write);
+
+ pa_rtpoll_item_free(o->inq_rtpoll_item_read);
+ o->inq_rtpoll_item_read = NULL;
+
+ pa_rtpoll_item_free(o->outq_rtpoll_item_write);
+ o->outq_rtpoll_item_write = NULL;
}
/* Called from main context */
@@ -438,8 +513,7 @@ static int sink_input_process_msg(pa_msgobject *obj, int code, void *data, int64
else
pa_memblockq_flush(o->memblockq);
- break;
-
+ return 0;
}
return pa_sink_input_process_msg(obj, code, data, offset, chunk);
@@ -452,11 +526,10 @@ static void disable_output(struct output *o) {
if (!o->sink_input)
return;
- pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
pa_sink_input_unlink(o->sink_input);
+ pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
pa_sink_input_unref(o->sink_input);
o->sink_input = NULL;
-
}
/* Called from main context */
@@ -488,8 +561,6 @@ static void suspend(struct userdata *u) {
for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
disable_output(o);
- pick_master(u, NULL);
-
pa_log_info("Device suspended...");
}
@@ -509,8 +580,6 @@ static void unsuspend(struct userdata *u) {
enable_output(o);
}
- pick_master(u, NULL);
-
pa_log_info("Resumed successfully...");
}
@@ -547,7 +616,25 @@ static int sink_set_state(pa_sink *sink, pa_sink_state_t state) {
return 0;
}
-/* Called from thread context of the master */
+/* Called from IO context */
+static void update_max_request(struct userdata *u) {
+ size_t max_request = 0;
+ struct output *o;
+
+ for (o = u->thread_info.active_outputs; o; o = o->next) {
+ size_t mr = (size_t) pa_atomic_load(&o->max_request);
+
+ if (mr > max_request)
+ max_request = mr;
+ }
+
+ if (max_request <= 0)
+ max_request = pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec);
+
+ pa_sink_set_max_request(u->sink, max_request);
+}
+
+/* Called from thread context of the io thread */
static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SINK(o)->userdata;
@@ -555,41 +642,47 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
case PA_SINK_MESSAGE_SET_STATE:
pa_atomic_store(&u->thread_info.running, PA_PTR_TO_UINT(data) == PA_SINK_RUNNING);
- break;
- case PA_SINK_MESSAGE_GET_LATENCY:
+ if (PA_PTR_TO_UINT(data) == PA_SINK_SUSPENDED)
+ pa_smoother_pause(u->thread_info.smoother, pa_rtclock_usec());
+ else
+ pa_smoother_resume(u->thread_info.smoother, pa_rtclock_usec());
- /* This code will only be called when running in NULL
- * mode, i.e. when no output is attached. See
- * sink_get_latency_cb() below */
+ break;
- if (u->thread_info.in_null_mode) {
- struct timeval now;
+ case PA_SINK_MESSAGE_GET_LATENCY: {
+ pa_usec_t x, y, c, *delay = data;
- if (pa_timeval_cmp(&u->thread_info.timestamp, pa_rtclock_get(&now)) > 0) {
- *((pa_usec_t*) data) = pa_timeval_diff(&u->thread_info.timestamp, &now);
- break;
- }
- }
+ x = pa_rtclock_usec();
+ y = pa_smoother_get(u->thread_info.smoother, x);
- *((pa_usec_t*) data) = 0;
+ c = pa_bytes_to_usec(u->thread_info.counter, &u->sink->sample_spec);
- break;
+ if (y < c)
+ *delay = c - y;
+ else
+ *delay = 0;
+
+ return 0;
+ }
case SINK_MESSAGE_ADD_OUTPUT: {
struct output *op = data;
PA_LLIST_PREPEND(struct output, u->thread_info.active_outputs, op);
- pa_assert(!op->outq_rtpoll_item);
-
- /* Create pa_asyncmsgq to the sink thread */
+ pa_assert(!op->outq_rtpoll_item_read && !op->inq_rtpoll_item_write);
- op->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq_read(
+ op->outq_rtpoll_item_read = pa_rtpoll_item_new_asyncmsgq_read(
u->rtpoll,
PA_RTPOLL_EARLY-1, /* This item is very important */
op->outq);
+ op->inq_rtpoll_item_write = pa_rtpoll_item_new_asyncmsgq_write(
+ u->rtpoll,
+ PA_RTPOLL_EARLY,
+ op->inq);
+ update_max_request(u);
return 0;
}
@@ -598,56 +691,48 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
PA_LLIST_REMOVE(struct output, u->thread_info.active_outputs, op);
- /* Remove the q that leads from this output to the sink thread */
+ pa_assert(op->outq_rtpoll_item_read && op->inq_rtpoll_item_write);
- pa_assert(op->outq_rtpoll_item);
- pa_rtpoll_item_free(op->outq_rtpoll_item);
- op->outq_rtpoll_item = NULL;
+ pa_rtpoll_item_free(op->outq_rtpoll_item_read);
+ op->outq_rtpoll_item_read = NULL;
+ pa_rtpoll_item_free(op->inq_rtpoll_item_write);
+ op->inq_rtpoll_item_write = NULL;
+
+ update_max_request(u);
return 0;
}
case SINK_MESSAGE_NEED:
- render_memblock(u, data, (size_t) offset);
+ render_memblock(u, (struct output*) data, (size_t) offset);
return 0;
- }
-
- return pa_sink_process_msg(o, code, data, offset, chunk);
-}
-/* Called from main context */
-/* static pa_usec_t sink_get_latency_cb(pa_sink *s) { */
-/* struct userdata *u; */
+ case SINK_MESSAGE_UPDATE_LATENCY: {
+ pa_usec_t x, y, latency = (pa_usec_t) offset;
-/* pa_sink_assert_ref(s); */
-/* pa_assert_se(u = s->userdata); */
+ x = pa_rtclock_usec();
+ y = pa_bytes_to_usec(u->thread_info.counter, &u->sink->sample_spec);
-/* if (u->master) { */
-/* /\* If we have a master sink, we just return the latency of it */
-/* * and add our own buffering on top *\/ */
-
-/* if (!u->master->sink_input) */
-/* return 0; */
-
-/* return */
-/* pa_sink_input_get_latency(u->master->sink_input) + */
-/* pa_sink_get_latency(u->master->sink); */
+ if (y > latency)
+ y -= latency;
+ else
+ y = 0;
-/* } else { */
-/* pa_usec_t usec = 0; */
+ pa_smoother_put(u->thread_info.smoother, x, y);
+ return 0;
+ }
-/* /\* We have no master, hence let's ask our own thread which */
-/* * implements the NULL sink *\/ */
+ case SINK_MESSAGE_UPDATE_MAX_REQUEST:
-/* if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0) */
-/* return 0; */
+ update_max_request(u);
+ break;
+ }
-/* return usec; */
-/* } */
-/* } */
+ return pa_sink_process_msg(o, code, data, offset, chunk);
+}
static void update_description(struct userdata *u) {
- int first = 1;
+ pa_bool_t first = TRUE;
char *t;
struct output *o;
uint32_t idx;
@@ -666,7 +751,7 @@ static void update_description(struct userdata *u) {
if (first) {
e = pa_sprintf_malloc("%s %s", t, pa_strnull(pa_proplist_gets(o->sink->proplist, PA_PROP_DEVICE_DESCRIPTION)));
- first = 0;
+ first = FALSE;
} else
e = pa_sprintf_malloc("%s, %s", t, pa_strnull(pa_proplist_gets(o->sink->proplist, PA_PROP_DEVICE_DESCRIPTION)));
@@ -678,57 +763,19 @@ static void update_description(struct userdata *u) {
pa_xfree(t);
}
-static void update_master(struct userdata *u, struct output *o) {
- pa_assert(u);
-
- if (u->master == o)
- return;
-
- if ((u->master = o))
- pa_log_info("Master sink is now '%s'", o->sink_input->sink->name);
- else
- pa_log_info("No master selected, lacking suitable outputs.");
-}
-
-static void pick_master(struct userdata *u, struct output *except) {
- struct output *o;
- uint32_t idx;
- pa_assert(u);
-
- if (u->master &&
- u->master != except &&
- u->master->sink_input &&
- PA_SINK_IS_OPENED(pa_sink_get_state(u->master->sink))) {
- update_master(u, u->master);
- return;
- }
-
- for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
- if (o != except &&
- o->sink_input &&
- PA_SINK_IS_OPENED(pa_sink_get_state(o->sink))) {
- update_master(u, o);
- return;
- }
-
- update_master(u, NULL);
-}
-
static int output_create_sink_input(struct output *o) {
pa_sink_input_new_data data;
- char *t;
pa_assert(o);
if (o->sink_input)
return 0;
- t = pa_sprintf_malloc("Simultaneous output on %s", pa_strnull(pa_proplist_gets(o->sink->proplist, PA_PROP_DEVICE_DESCRIPTION)));
-
pa_sink_input_new_data_init(&data);
data.sink = o->sink;
data.driver = __FILE__;
- pa_proplist_sets(data.proplist, PA_PROP_DEVICE_DESCRIPTION, t);
+ pa_proplist_setf(data.proplist, PA_PROP_MEDIA_NAME, "Simultaneous output on %s", pa_strnull(pa_proplist_gets(o->sink->proplist, PA_PROP_DEVICE_DESCRIPTION)));
+ pa_proplist_sets(data.proplist, PA_PROP_MEDIA_ROLE, "filter");
pa_sink_input_new_data_set_sample_spec(&data, &o->userdata->sink->sample_spec);
pa_sink_input_new_data_set_channel_map(&data, &o->userdata->sink->channel_map);
data.module = o->userdata->module;
@@ -738,24 +785,27 @@ static int output_create_sink_input(struct output *o) {
pa_sink_input_new_data_done(&data);
- pa_xfree(t);
-
if (!o->sink_input)
return -1;
o->sink_input->parent.process_msg = sink_input_process_msg;
o->sink_input->pop = sink_input_pop_cb;
+ o->sink_input->process_rewind = sink_input_process_rewind_cb;
+ o->sink_input->update_max_rewind = sink_input_update_max_rewind_cb;
+ o->sink_input->update_max_request = sink_input_update_max_request_cb;
o->sink_input->attach = sink_input_attach_cb;
o->sink_input->detach = sink_input_detach_cb;
o->sink_input->kill = sink_input_kill_cb;
o->sink_input->userdata = o;
+ pa_sink_input_set_requested_latency(o->sink_input, REQUEST_LATENCY_USEC);
return 0;
}
static struct output *output_new(struct userdata *u, pa_sink *sink) {
struct output *o;
+ pa_sink_state_t state;
pa_assert(u);
pa_assert(sink);
@@ -765,8 +815,8 @@ static struct output *output_new(struct userdata *u, pa_sink *sink) {
o->userdata = u;
o->inq = pa_asyncmsgq_new(0);
o->outq = pa_asyncmsgq_new(0);
- o->inq_rtpoll_item = NULL;
- o->outq_rtpoll_item = NULL;
+ o->inq_rtpoll_item_write = o->inq_rtpoll_item_read = NULL;
+ o->outq_rtpoll_item_write = o->outq_rtpoll_item_read = NULL;
o->sink = sink;
o->sink_input = NULL;
o->memblockq = pa_memblockq_new(
@@ -778,22 +828,30 @@ static struct output *output_new(struct userdata *u, pa_sink *sink) {
0,
0,
NULL);
+ pa_atomic_store(&o->max_request, 0);
+ PA_LLIST_INIT(struct output, o);
pa_assert_se(pa_idxset_put(u->outputs, o, NULL) == 0);
- if (u->sink && PA_SINK_IS_LINKED(pa_sink_get_state(u->sink)))
+ state = pa_sink_get_state(u->sink);
+
+ if (state != PA_SINK_INIT)
pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_ADD_OUTPUT, o, 0, NULL);
else {
/* If the sink is not yet started, we need to do the activation ourselves */
PA_LLIST_PREPEND(struct output, u->thread_info.active_outputs, o);
- o->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq_read(
+ o->outq_rtpoll_item_read = pa_rtpoll_item_new_asyncmsgq_read(
u->rtpoll,
PA_RTPOLL_EARLY-1, /* This item is very important */
o->outq);
+ o->inq_rtpoll_item_write = pa_rtpoll_item_new_asyncmsgq_write(
+ u->rtpoll,
+ PA_RTPOLL_EARLY,
+ o->inq);
}
- if (PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)) || pa_sink_get_state(u->sink) == PA_SINK_INIT) {
+ if (PA_SINK_IS_OPENED(state) || state == PA_SINK_INIT) {
pa_sink_suspend(sink, FALSE);
if (PA_SINK_IS_OPENED(pa_sink_get_state(sink)))
@@ -801,7 +859,6 @@ static struct output *output_new(struct userdata *u, pa_sink *sink) {
goto fail;
}
-
update_description(u);
return o;
@@ -831,7 +888,25 @@ fail:
return NULL;
}
-static pa_hook_result_t sink_new_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
+static pa_bool_t is_suitable_sink(struct userdata *u, pa_sink *s) {
+ const char *t;
+
+ pa_sink_assert_ref(s);
+
+ if (!(s->flags & PA_SINK_HARDWARE))
+ return FALSE;
+
+ if (s == u->sink)
+ return FALSE;
+
+ if ((t = pa_proplist_gets(s->proplist, PA_PROP_DEVICE_CLASS)))
+ if (strcmp(t, "sound"))
+ return FALSE;
+
+ return TRUE;
+}
+
+static pa_hook_result_t sink_put_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
struct output *o;
pa_core_assert_ref(c);
@@ -839,7 +914,7 @@ static pa_hook_result_t sink_new_hook_cb(pa_core *c, pa_sink *s, struct userdata
pa_assert(u);
pa_assert(u->automatic);
- if (!(s->flags & PA_SINK_HARDWARE) || s == u->sink)
+ if (!is_suitable_sink(u, s))
return PA_HOOK_OK;
pa_log_info("Configuring new sink: %s", s->name);
@@ -852,27 +927,34 @@ static pa_hook_result_t sink_new_hook_cb(pa_core *c, pa_sink *s, struct userdata
if (o->sink_input)
pa_sink_input_put(o->sink_input);
- pick_master(u, NULL);
-
return PA_HOOK_OK;
}
-static pa_hook_result_t sink_unlink_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
+static struct output* find_output(struct userdata *u, pa_sink *s) {
struct output *o;
uint32_t idx;
- pa_assert(c);
- pa_sink_assert_ref(s);
pa_assert(u);
+ pa_assert(s);
- if (s == u->sink)
- return PA_HOOK_OK;
+ if (u->sink == s)
+ return NULL;
for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
if (o->sink == s)
- break;
+ return o;
+
+ return NULL;
+}
+
+static pa_hook_result_t sink_unlink_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
+ struct output *o;
+
+ pa_assert(c);
+ pa_sink_assert_ref(s);
+ pa_assert(u);
- if (!o)
+ if (!(o = find_output(u, s)))
return PA_HOOK_OK;
pa_log_info("Unconfiguring sink: %s", s->name);
@@ -884,30 +966,18 @@ static pa_hook_result_t sink_unlink_hook_cb(pa_core *c, pa_sink *s, struct userd
static pa_hook_result_t sink_state_changed_hook_cb(pa_core *c, pa_sink *s, struct userdata* u) {
struct output *o;
- uint32_t idx;
pa_sink_state_t state;
- if (s == u->sink)
- return PA_HOOK_OK;
-
- for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
- if (o->sink == s)
- break;
-
- if (!o)
+ if (!(o = find_output(u, s)))
return PA_HOOK_OK;
state = pa_sink_get_state(s);
- if (PA_SINK_IS_OPENED(state) && PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)) && !o->sink_input) {
+ if (PA_SINK_IS_OPENED(state) && PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)) && !o->sink_input)
enable_output(o);
- pick_master(u, NULL);
- }
- if (state == PA_SINK_SUSPENDED && o->sink_input) {
+ if (state == PA_SINK_SUSPENDED && o->sink_input)
disable_output(o);
- pick_master(u, o);
- }
return PA_HOOK_OK;
}
@@ -915,8 +985,7 @@ static pa_hook_result_t sink_state_changed_hook_cb(pa_core *c, pa_sink *s, struc
int pa__init(pa_module*m) {
struct userdata *u;
pa_modargs *ma = NULL;
- const char *master_name, *slaves, *rm;
- pa_sink *master_sink = NULL;
+ const char *slaves, *rm;
int resample_method = PA_RESAMPLER_TRIVIAL;
pa_sample_spec ss;
pa_channel_map map;
@@ -938,12 +1007,10 @@ int pa__init(pa_module*m) {
}
}
- u = pa_xnew(struct userdata, 1);
+ m->userdata = u = pa_xnew(struct userdata, 1);
u->core = m->core;
u->module = m;
- m->userdata = u;
u->sink = NULL;
- u->master = NULL;
u->time_event = NULL;
u->adjust_time = DEFAULT_ADJUST_TIME;
u->rtpoll = pa_rtpoll_new();
@@ -952,67 +1019,39 @@ int pa__init(pa_module*m) {
u->resample_method = resample_method;
u->outputs = pa_idxset_new(NULL, NULL);
memset(&u->adjust_timestamp, 0, sizeof(u->adjust_timestamp));
- u->sink_new_slot = u->sink_unlink_slot = u->sink_state_changed_slot = NULL;
+ u->sink_put_slot = u->sink_unlink_slot = u->sink_state_changed_slot = NULL;
PA_LLIST_HEAD_INIT(struct output, u->thread_info.active_outputs);
pa_atomic_store(&u->thread_info.running, FALSE);
u->thread_info.in_null_mode = FALSE;
+ u->thread_info.counter = 0;
+ u->thread_info.smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE, 10);
if (pa_modargs_get_value_u32(ma, "adjust_time", &u->adjust_time) < 0) {
pa_log("Failed to parse adjust_time value");
goto fail;
}
- master_name = pa_modargs_get_value(ma, "master", NULL);
slaves = pa_modargs_get_value(ma, "slaves", NULL);
- if (!master_name != !slaves) {
- pa_log("No master or slave sinks specified");
- goto fail;
- }
-
- if (master_name) {
- if (!(master_sink = pa_namereg_get(m->core, master_name, PA_NAMEREG_SINK, 1))) {
- pa_log("Invalid master sink '%s'", master_name);
- goto fail;
- }
-
- ss = master_sink->sample_spec;
- u->automatic = FALSE;
- } else {
- master_sink = NULL;
- ss = m->core->default_sample_spec;
- u->automatic = TRUE;
- }
+ u->automatic = !slaves;
+ ss = m->core->default_sample_spec;
- if ((pa_modargs_get_sample_spec(ma, &ss) < 0)) {
+ if ((pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0)) {
pa_log("Invalid sample specification.");
goto fail;
}
- if (master_sink && ss.channels == master_sink->sample_spec.channels)
- map = master_sink->channel_map;
- else {
- pa_assert_se(pa_channel_map_init_auto(&map, ss.channels, PA_CHANNEL_MAP_AUX));
- pa_channel_map_init_auto(&map, ss.channels, PA_CHANNEL_MAP_DEFAULT);
- }
-
- if ((pa_modargs_get_channel_map(ma, NULL, &map) < 0)) {
- pa_log("Invalid channel map.");
- goto fail;
- }
-
- if (ss.channels != map.channels) {
- pa_log("Channel map and sample specification don't match.");
- goto fail;
- }
-
pa_sink_new_data_init(&data);
- data.name = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
data.namereg_fail = FALSE;
data.driver = __FILE__;
data.module = m;
+ pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
pa_sink_new_data_set_sample_spec(&data, &ss);
pa_sink_new_data_set_channel_map(&data, &map);
pa_proplist_sets(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Simultaneous Output");
+ pa_proplist_sets(data.proplist, PA_PROP_DEVICE_CLASS, "filter");
+
+ if (slaves)
+ pa_proplist_sets(data.proplist, "combine.slaves", slaves);
u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY);
pa_sink_new_data_done(&data);
@@ -1023,34 +1062,30 @@ int pa__init(pa_module*m) {
}
u->sink->parent.process_msg = sink_process_msg;
-/* u->sink->get_latency = sink_get_latency_cb; */
u->sink->set_state = sink_set_state;
u->sink->userdata = u;
pa_sink_set_rtpoll(u->sink, u->rtpoll);
pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
- u->block_size = pa_bytes_per_second(&ss) / 20; /* 50 ms */
- if (u->block_size <= 0)
- u->block_size = pa_frame_size(&ss);
+ pa_sink_set_latency_range(u->sink, REQUEST_LATENCY_USEC, REQUEST_LATENCY_USEC);
+ u->block_usec = u->sink->thread_info.max_latency;
+
+ u->sink->thread_info.max_request =
+ pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec);
if (!u->automatic) {
const char*split_state;
char *n = NULL;
pa_assert(slaves);
- /* The master and slaves have been specified manually */
-
- if (!(u->master = output_new(u, master_sink))) {
- pa_log("Failed to create master sink input on sink '%s'.", master_sink->name);
- goto fail;
- }
+ /* The slaves have been specified manually */
split_state = NULL;
while ((n = pa_split(slaves, ",", &split_state))) {
pa_sink *slave_sink;
- if (!(slave_sink = pa_namereg_get(m->core, n, PA_NAMEREG_SINK, 1)) || slave_sink == u->sink) {
+ if (!(slave_sink = pa_namereg_get(m->core, n, PA_NAMEREG_SINK, TRUE)) || slave_sink == u->sink) {
pa_log("Invalid slave sink '%s'", n);
pa_xfree(n);
goto fail;
@@ -1067,17 +1102,16 @@ int pa__init(pa_module*m) {
if (pa_idxset_size(u->outputs) <= 1)
pa_log_warn("No slave sinks specified.");
- u->sink_new_slot = NULL;
+ u->sink_put_slot = NULL;
} else {
pa_sink *s;
- /* We're in automatic mode, we elect one hw sink to the master
- * and attach all other hw sinks as slaves to it */
+ /* We're in automatic mode, we add every sink that matches our needs */
for (s = pa_idxset_first(m->core->sinks, &idx); s; s = pa_idxset_next(m->core->sinks, &idx)) {
- if (!(s->flags & PA_SINK_HARDWARE) || s == u->sink)
+ if (!is_suitable_sink(u, s))
continue;
if (!output_new(u, s)) {
@@ -1086,14 +1120,12 @@ int pa__init(pa_module*m) {
}
}
- u->sink_new_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_PUT], PA_HOOK_LATE, (pa_hook_cb_t) sink_new_hook_cb, u);
+ u->sink_put_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_PUT], PA_HOOK_LATE, (pa_hook_cb_t) sink_put_hook_cb, u);
}
u->sink_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_UNLINK], PA_HOOK_EARLY, (pa_hook_cb_t) sink_unlink_hook_cb, u);
u->sink_state_changed_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_STATE_CHANGED], PA_HOOK_NORMAL, (pa_hook_cb_t) sink_state_changed_hook_cb, u);
- pick_master(u, NULL);
-
if (!(u->thread = pa_thread_new(thread_func, u))) {
pa_log("Failed to create thread.");
goto fail;
@@ -1130,19 +1162,21 @@ fail:
static void output_free(struct output *o) {
pa_assert(o);
- pick_master(o->userdata, o);
-
disable_output(o);
pa_assert_se(pa_idxset_remove_by_data(o->userdata->outputs, o, NULL));
update_description(o->userdata);
- if (o->inq_rtpoll_item)
- pa_rtpoll_item_free(o->inq_rtpoll_item);
+ if (o->inq_rtpoll_item_read)
+ pa_rtpoll_item_free(o->inq_rtpoll_item_read);
+ if (o->inq_rtpoll_item_write)
+ pa_rtpoll_item_free(o->inq_rtpoll_item_write);
- if (o->outq_rtpoll_item)
- pa_rtpoll_item_free(o->outq_rtpoll_item);
+ if (o->outq_rtpoll_item_read)
+ pa_rtpoll_item_free(o->outq_rtpoll_item_read);
+ if (o->outq_rtpoll_item_write)
+ pa_rtpoll_item_free(o->outq_rtpoll_item_write);
if (o->inq)
pa_asyncmsgq_unref(o->inq);
@@ -1165,8 +1199,8 @@ void pa__done(pa_module*m) {
if (!(u = m->userdata))
return;
- if (u->sink_new_slot)
- pa_hook_slot_free(u->sink_new_slot);
+ if (u->sink_put_slot)
+ pa_hook_slot_free(u->sink_put_slot);
if (u->sink_unlink_slot)
pa_hook_slot_free(u->sink_unlink_slot);
@@ -1200,5 +1234,8 @@ void pa__done(pa_module*m) {
if (u->time_event)
u->core->mainloop->time_free(u->time_event);
+ if (u->thread_info.smoother)
+ pa_smoother_free(u->thread_info.smoother);
+
pa_xfree(u);
}
diff --git a/src/modules/module-ladspa-sink.c b/src/modules/module-ladspa-sink.c
index d11ac26..ca1c05c 100644
--- a/src/modules/module-ladspa-sink.c
+++ b/src/modules/module-ladspa-sink.c
@@ -260,7 +260,7 @@ static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
pa_sink_input_assert_ref(i);
pa_assert_se(u = i->userdata);
- if (!u->sink || !PA_SINK_IS_OPENED(u->sink->thread_info.state))
+ if (!u->sink)
return;
pa_memblockq_set_maxrewind(u->memblockq, nbytes);
@@ -268,13 +268,39 @@ static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
}
/* Called from I/O thread context */
+static void sink_input_update_max_request_cb(pa_sink_input *i, size_t nbytes) {
+ struct userdata *u;
+
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(u = i->userdata);
+
+ if (!u->sink)
+ return;
+
+ pa_sink_set_max_request(u->sink, nbytes);
+}
+
+/* Called from I/O thread context */
+static void sink_input_update_sink_latency_range_cb(pa_sink_input *i) {
+ struct userdata *u;
+
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(u = i->userdata);
+
+ if (!u->sink)
+ return;
+
+ pa_sink_update_latency_range(u->sink, i->sink->thread_info.min_latency, i->sink->thread_info.max_latency);
+}
+
+/* Called from I/O thread context */
static void sink_input_detach_cb(pa_sink_input *i) {
struct userdata *u;
pa_sink_input_assert_ref(i);
pa_assert_se(u = i->userdata);
- if (!u->sink || !PA_SINK_IS_OPENED(u->sink->thread_info.state))
+ if (!u->sink)
return;
pa_sink_detach_within_thread(u->sink);
@@ -289,14 +315,14 @@ static void sink_input_attach_cb(pa_sink_input *i) {
pa_sink_input_assert_ref(i);
pa_assert_se(u = i->userdata);
- if (!u->sink || !PA_SINK_IS_OPENED(u->sink->thread_info.state))
+ if (!u->sink)
return;
pa_sink_set_asyncmsgq(u->sink, i->sink->asyncmsgq);
pa_sink_set_rtpoll(u->sink, i->sink->rtpoll);
pa_sink_attach_within_thread(u->sink);
- pa_sink_set_latency_range(u->sink, u->master->min_latency, u->master->max_latency);
+ pa_sink_set_latency_range(u->sink, u->master->thread_info.min_latency, u->master->thread_info.max_latency);
}
/* Called from main context */
@@ -705,6 +731,8 @@ int pa__init(pa_module*m) {
u->sink_input->pop = sink_input_pop_cb;
u->sink_input->process_rewind = sink_input_process_rewind_cb;
u->sink_input->update_max_rewind = sink_input_update_max_rewind_cb;
+ u->sink_input->update_max_request = sink_input_update_max_request_cb;
+ u->sink_input->update_sink_latency_range = sink_input_update_sink_latency_range_cb;
u->sink_input->kill = sink_input_kill_cb;
u->sink_input->attach = sink_input_attach_cb;
u->sink_input->detach = sink_input_detach_cb;
diff --git a/src/modules/module-null-sink.c b/src/modules/module-null-sink.c
index 38da343..604ab15 100644
--- a/src/modules/module-null-sink.c
+++ b/src/modules/module-null-sink.c
@@ -155,34 +155,32 @@ static void process_rewind(struct userdata *u, pa_usec_t now) {
}
static void process_render(struct userdata *u, pa_usec_t now) {
- size_t nbytes;
size_t ate = 0;
pa_assert(u);
/* This is the configured latency. Sink inputs connected to us
- might not have a single frame more than this value queued. Hence:
- at maximum read this many bytes from the sink inputs. */
-
- nbytes = pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec);
+ might not have a single frame more than the maxrequest value
+ queed. Hence: at maximum read this many bytes from the sink
+ inputs. */
/* Fill the buffer up the the latency size */
while (u->timestamp < now + u->block_usec) {
pa_memchunk chunk;
- pa_sink_render(u->sink, nbytes, &chunk);
+ pa_sink_render(u->sink, u->sink->thread_info.max_request, &chunk);
pa_memblock_unref(chunk.memblock);
- pa_log_debug("Ate %lu bytes.", (unsigned long) chunk.length);
+/* pa_log_debug("Ate %lu bytes.", (unsigned long) chunk.length); */
u->timestamp += pa_bytes_to_usec(chunk.length, &u->sink->sample_spec);
ate += chunk.length;
- if (ate >= nbytes)
+ if (ate >= u->sink->thread_info.max_request)
break;
}
- pa_log_debug("Ate in sum %lu bytes (of %lu)", (unsigned long) ate, (unsigned long) nbytes);
+/* pa_log_debug("Ate in sum %lu bytes (of %lu)", (unsigned long) ate, (unsigned long) nbytes); */
}
static void thread_func(void *userdata) {
@@ -201,7 +199,7 @@ static void thread_func(void *userdata) {
int ret;
/* Render some data and drop it immediately */
- if (u->sink->thread_info.state == PA_SINK_RUNNING) {
+ if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
pa_usec_t now;
now = pa_rtclock_usec();
@@ -254,10 +252,9 @@ int pa__init(pa_module*m) {
goto fail;
}
- u = pa_xnew0(struct userdata, 1);
+ m->userdata = u = pa_xnew0(struct userdata, 1);
u->core = m->core;
u->module = m;
- m->userdata = u;
u->rtpoll = pa_rtpoll_new();
pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
@@ -268,6 +265,7 @@ int pa__init(pa_module*m) {
pa_sink_new_data_set_sample_spec(&data, &ss);
pa_sink_new_data_set_channel_map(&data, &map);
pa_proplist_sets(data.proplist, PA_PROP_DEVICE_DESCRIPTION, pa_modargs_get_value(ma, "description", "Null Output"));
+ pa_proplist_sets(data.proplist, PA_PROP_DEVICE_CLASS, "abstract");
u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY);
pa_sink_new_data_done(&data);
@@ -285,9 +283,11 @@ int pa__init(pa_module*m) {
pa_sink_set_rtpoll(u->sink, u->rtpoll);
pa_sink_set_latency_range(u->sink, (pa_usec_t) -1, MAX_LATENCY_USEC);
- u->block_usec = u->sink->max_latency;
+ u->block_usec = u->sink->thread_info.max_latency;
- u->sink->thread_info.max_rewind = pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec);
+ u->sink->thread_info.max_rewind =
+ u->sink->thread_info.max_request =
+ pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec);
if (!(u->thread = pa_thread_new(thread_func, u))) {
pa_log("Failed to create thread.");
diff --git a/src/modules/module-oss.c b/src/modules/module-oss.c
index 5eb71cb..21b6737 100644
--- a/src/modules/module-oss.c
+++ b/src/modules/module-oss.c
@@ -1366,6 +1366,8 @@ int pa__init(pa_module*m) {
pa_sink_set_rtpoll(u->sink, u->rtpoll);
u->sink->refresh_volume = TRUE;
+ u->sink->thread_info.max_request = u->out_hwbuf_size;
+
if (use_mmap)
u->out_mmap_memblocks = pa_xnew0(pa_memblock*, u->out_nfrags);
}
diff --git a/src/modules/module-remap-sink.c b/src/modules/module-remap-sink.c
index b8c76b6..ac39fa1 100644
--- a/src/modules/module-remap-sink.c
+++ b/src/modules/module-remap-sink.c
@@ -179,20 +179,46 @@ static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
pa_sink_input_assert_ref(i);
pa_assert_se(u = i->userdata);
- if (!u->sink || !PA_SINK_IS_OPENED(u->sink->thread_info.state))
+ if (!u->sink)
return;
pa_sink_set_max_rewind(u->sink, nbytes);
}
/* Called from I/O thread context */
+static void sink_input_update_max_request_cb(pa_sink_input *i, size_t nbytes) {
+ struct userdata *u;
+
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(u = i->userdata);
+
+ if (!u->sink)
+ return;
+
+ pa_sink_set_max_request(u->sink, nbytes);
+}
+
+/* Called from I/O thread context */
+static void sink_input_update_sink_latency_range_cb(pa_sink_input *i) {
+ struct userdata *u;
+
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(u = i->userdata);
+
+ if (!u->sink)
+ return;
+
+ pa_sink_update_latency_range(u->sink, i->sink->thread_info.min_latency, i->sink->thread_info.max_latency);
+}
+
+/* Called from I/O thread context */
static void sink_input_detach_cb(pa_sink_input *i) {
struct userdata *u;
pa_sink_input_assert_ref(i);
pa_assert_se(u = i->userdata);
- if (!u->sink || !PA_SINK_IS_OPENED(u->sink->thread_info.state))
+ if (!u->sink)
return;
pa_sink_detach_within_thread(u->sink);
@@ -207,14 +233,14 @@ static void sink_input_attach_cb(pa_sink_input *i) {
pa_sink_input_assert_ref(i);
pa_assert_se(u = i->userdata);
- if (!u->sink || !PA_SINK_IS_OPENED(u->sink->thread_info.state))
+ if (!u->sink)
return;
pa_sink_set_asyncmsgq(u->sink, i->sink->asyncmsgq);
pa_sink_set_rtpoll(u->sink, i->sink->rtpoll);
pa_sink_attach_within_thread(u->sink);
- pa_sink_set_latency_range(u->sink, u->master->min_latency, u->master->max_latency);
+ pa_sink_set_latency_range(u->sink, u->master->thread_info.min_latency, u->master->thread_info.max_latency);
}
/* Called from main context */
@@ -357,9 +383,11 @@ int pa__init(pa_module*m) {
u->sink_input->pop = sink_input_pop_cb;
u->sink_input->process_rewind = sink_input_process_rewind_cb;
u->sink_input->update_max_rewind = sink_input_update_max_rewind_cb;
- u->sink_input->kill = sink_input_kill_cb;
+ u->sink_input->update_max_request = sink_input_update_max_request_cb;
+ u->sink_input->update_sink_latency_range = sink_input_update_sink_latency_range_cb;
u->sink_input->attach = sink_input_attach_cb;
u->sink_input->detach = sink_input_detach_cb;
+ u->sink_input->kill = sink_input_kill_cb;
u->sink_input->state_change = sink_input_state_change_cb;
u->sink_input->userdata = u;
diff --git a/src/modules/module-tunnel.c b/src/modules/module-tunnel.c
index ed7f402..86f3081 100644
--- a/src/modules/module-tunnel.c
+++ b/src/modules/module-tunnel.c
@@ -1699,8 +1699,8 @@ int pa__init(pa_module*m) {
}
m->userdata = u = pa_xnew0(struct userdata, 1);
- u->module = m;
u->core = m->core;
+ u->module = m;
u->client = NULL;
u->pdispatch = NULL;
u->pstream = NULL;
@@ -1779,7 +1779,7 @@ int pa__init(pa_module*m) {
u->sink->set_volume = sink_set_volume;
u->sink->set_mute = sink_set_mute;
- u->sink->refresh_volume = u->sink->refresh_mute = FALSE;
+ u->sink->refresh_volume = u->sink->refresh_muted = FALSE;
pa_sink_set_latency_range(u->sink, MIN_NETWORK_LATENCY_USEC, 0);
@@ -1812,8 +1812,8 @@ int pa__init(pa_module*m) {
}
u->source->parent.process_msg = source_process_msg;
- u->source->userdata = u;
u->source->set_state = source_set_state;
+ u->source->userdata = u;
pa_source_set_latency_range(u->source, MIN_NETWORK_LATENCY_USEC, 0);
diff --git a/src/pulsecore/cli-text.c b/src/pulsecore/cli-text.c
index 02e670d..c92fca2 100644
--- a/src/pulsecore/cli-text.c
+++ b/src/pulsecore/cli-text.c
@@ -116,6 +116,9 @@ char *pa_sink_list_to_string(pa_core *c) {
for (sink = pa_idxset_first(c->sinks, &idx); sink; sink = pa_idxset_next(c->sinks, &idx)) {
char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cv[PA_CVOLUME_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t;
+ pa_usec_t min_latency, max_latency;
+
+ pa_sink_get_latency_range(sink, &min_latency, &max_latency);
pa_strbuf_printf(
s,
@@ -128,6 +131,8 @@ char *pa_sink_list_to_string(pa_core *c) {
"\tmuted: %s\n"
"\tcurrent latency: %0.2f ms\n"
"\tconfigured latency: %0.2f ms; range is %0.2f .. %0.2f ms\n"
+ "\tmax request: %lu KiB\n"
+ "\tmax rewind: %lu KiB\n"
"\tmonitor source: %u\n"
"\tsample spec: %s\n"
"\tchannel map: %s\n"
@@ -147,7 +152,9 @@ char *pa_sink_list_to_string(pa_core *c) {
pa_cvolume_snprint(cv, sizeof(cv), pa_sink_get_volume(sink)),
pa_yes_no(pa_sink_get_mute(sink)),
(double) pa_sink_get_latency(sink) / PA_USEC_PER_MSEC,
- (double) pa_sink_get_requested_latency(sink) / PA_USEC_PER_MSEC, (double) sink->min_latency / PA_USEC_PER_MSEC, (double) sink->max_latency / PA_USEC_PER_MSEC,
+ (double) pa_sink_get_requested_latency(sink) / PA_USEC_PER_MSEC, (double) min_latency / PA_USEC_PER_MSEC, (double) max_latency / PA_USEC_PER_MSEC,
+ (unsigned long) pa_sink_get_max_request(sink) / 1024,
+ (unsigned long) pa_sink_get_max_rewind(sink) / 1024,
sink->monitor_source ? sink->monitor_source->index : PA_INVALID_INDEX,
pa_sample_spec_snprint(ss, sizeof(ss), &sink->sample_spec),
pa_channel_map_snprint(cm, sizeof(cm), &sink->channel_map),
@@ -184,6 +191,9 @@ char *pa_source_list_to_string(pa_core *c) {
for (source = pa_idxset_first(c->sources, &idx); source; source = pa_idxset_next(c->sources, &idx)) {
char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], cv[PA_CVOLUME_SNPRINT_MAX], *t;
+ pa_usec_t min_latency, max_latency;
+
+ pa_source_get_latency_range(source, &min_latency, &max_latency);
pa_strbuf_printf(
s,
@@ -196,6 +206,7 @@ char *pa_source_list_to_string(pa_core *c) {
"\tmuted: %s\n"
"\tcurrent latency: %0.2f ms\n"
"\tconfigured latency: %0.2f ms; range is %0.2f .. %0.2f ms\n"
+ "\tmax rewind: %lu KiB\n"
"\tsample spec: %s\n"
"\tchannel map: %s\n"
"\tused by: %u\n"
@@ -214,7 +225,8 @@ char *pa_source_list_to_string(pa_core *c) {
pa_cvolume_snprint(cv, sizeof(cv), pa_source_get_volume(source)),
pa_yes_no(pa_source_get_mute(source)),
(double) pa_source_get_latency(source) / PA_USEC_PER_MSEC,
- (double) pa_source_get_requested_latency(source) / PA_USEC_PER_MSEC, (double) source->min_latency / PA_USEC_PER_MSEC, (double) source->max_latency / PA_USEC_PER_MSEC,
+ (double) pa_source_get_requested_latency(source) / PA_USEC_PER_MSEC, (double) min_latency / PA_USEC_PER_MSEC, (double) max_latency / PA_USEC_PER_MSEC,
+ (unsigned long) pa_source_get_max_rewind(source) / 1024,
pa_sample_spec_snprint(ss, sizeof(ss), &source->sample_spec),
pa_channel_map_snprint(cm, sizeof(cm), &source->channel_map),
pa_source_used_by(source),
@@ -285,7 +297,7 @@ char *pa_source_output_list_to_string(pa_core *c) {
o->flags & PA_SOURCE_OUTPUT_FIX_CHANNELS ? "FIX_CHANNELS " : "",
state_table[pa_source_output_get_state(o)],
o->source->index, o->source->name,
- (double) pa_source_output_get_latency(o) / PA_USEC_PER_MSEC,
+ (double) pa_source_output_get_latency(o, NULL) / PA_USEC_PER_MSEC,
clt,
pa_sample_spec_snprint(ss, sizeof(ss), &o->sample_spec),
pa_channel_map_snprint(cm, sizeof(cm), &o->channel_map),
@@ -361,7 +373,7 @@ char *pa_sink_input_list_to_string(pa_core *c) {
i->sink->index, i->sink->name,
pa_cvolume_snprint(cv, sizeof(cv), pa_sink_input_get_volume(i)),
pa_yes_no(pa_sink_input_get_mute(i)),
- (double) pa_sink_input_get_latency(i) / PA_USEC_PER_MSEC,
+ (double) pa_sink_input_get_latency(i, NULL) / PA_USEC_PER_MSEC,
clt,
pa_sample_spec_snprint(ss, sizeof(ss), &i->sample_spec),
pa_channel_map_snprint(cm, sizeof(cm), &i->channel_map),
diff --git a/src/pulsecore/protocol-native.c b/src/pulsecore/protocol-native.c
index e6e7c03..923a566 100644
--- a/src/pulsecore/protocol-native.c
+++ b/src/pulsecore/protocol-native.c
@@ -2535,6 +2535,7 @@ static void module_fill_tagstruct(pa_tagstruct *t, pa_module *module) {
static void sink_input_fill_tagstruct(connection *c, pa_tagstruct *t, pa_sink_input *s) {
pa_sample_spec fixed_ss;
+ pa_usec_t sink_latency;
pa_assert(t);
pa_sink_input_assert_ref(s);
@@ -2549,8 +2550,8 @@ static void sink_input_fill_tagstruct(connection *c, pa_tagstruct *t, pa_sink_in
pa_tagstruct_put_sample_spec(t, &fixed_ss);
pa_tagstruct_put_channel_map(t, &s->channel_map);
pa_tagstruct_put_cvolume(t, &s->volume);
- pa_tagstruct_put_usec(t, pa_sink_input_get_latency(s));
- pa_tagstruct_put_usec(t, pa_sink_get_latency(s->sink));
+ pa_tagstruct_put_usec(t, pa_sink_input_get_latency(s, &sink_latency));
+ pa_tagstruct_put_usec(t, sink_latency);
pa_tagstruct_puts(t, pa_resample_method_to_string(pa_sink_input_get_resample_method(s)));
pa_tagstruct_puts(t, s->driver);
if (c->version >= 11)
@@ -2561,6 +2562,7 @@ static void sink_input_fill_tagstruct(connection *c, pa_tagstruct *t, pa_sink_in
static void source_output_fill_tagstruct(connection *c, pa_tagstruct *t, pa_source_output *s) {
pa_sample_spec fixed_ss;
+ pa_usec_t source_latency;
pa_assert(t);
pa_source_output_assert_ref(s);
@@ -2574,8 +2576,8 @@ static void source_output_fill_tagstruct(connection *c, pa_tagstruct *t, pa_sour
pa_tagstruct_putu32(t, s->source->index);
pa_tagstruct_put_sample_spec(t, &fixed_ss);
pa_tagstruct_put_channel_map(t, &s->channel_map);
- pa_tagstruct_put_usec(t, pa_source_output_get_latency(s));
- pa_tagstruct_put_usec(t, pa_source_get_latency(s->source));
+ pa_tagstruct_put_usec(t, pa_source_output_get_latency(s, &source_latency));
+ pa_tagstruct_put_usec(t, source_latency);
pa_tagstruct_puts(t, pa_resample_method_to_string(pa_source_output_get_resample_method(s)));
pa_tagstruct_puts(t, s->driver);
diff --git a/src/pulsecore/sink-input.c b/src/pulsecore/sink-input.c
index 8257edf..43e7bb2 100644
--- a/src/pulsecore/sink-input.c
+++ b/src/pulsecore/sink-input.c
@@ -42,7 +42,6 @@
#define MEMBLOCKQ_MAXLENGTH (32*1024*1024)
#define CONVERT_BUFFER_LENGTH (PA_PAGE_SIZE)
-#define MOVE_BUFFER_LENGTH (PA_PAGE_SIZE*256)
static PA_DEFINE_CHECK_TYPE(pa_sink_input, pa_msgobject);
@@ -92,12 +91,16 @@ void pa_sink_input_new_data_done(pa_sink_input_new_data *data) {
pa_proplist_free(data->proplist);
}
+/* Called from main context */
static void reset_callbacks(pa_sink_input *i) {
pa_assert(i);
i->pop = NULL;
i->process_rewind = NULL;
i->update_max_rewind = NULL;
+ i->update_max_request = NULL;
+ i->update_sink_requested_latency = NULL;
+ i->update_sink_latency_range = NULL;
i->attach = NULL;
i->detach = NULL;
i->suspend = NULL;
@@ -107,6 +110,7 @@ static void reset_callbacks(pa_sink_input *i) {
i->state_change = NULL;
}
+/* Called from main context */
pa_sink_input* pa_sink_input_new(
pa_core *core,
pa_sink_input_new_data *data,
@@ -125,7 +129,7 @@ pa_sink_input* pa_sink_input_new(
pa_return_null_if_fail(!data->driver || pa_utf8_valid(data->driver));
if (!data->sink)
- data->sink = pa_namereg_get(core, NULL, PA_NAMEREG_SINK, 1);
+ data->sink = pa_namereg_get(core, NULL, PA_NAMEREG_SINK, TRUE);
pa_return_null_if_fail(data->sink);
pa_return_null_if_fail(pa_sink_get_state(data->sink) != PA_SINK_UNLINKED);
@@ -218,8 +222,6 @@ pa_sink_input* pa_sink_input_new(
i->sink = data->sink;
i->client = data->client;
- i->direct_outputs = pa_idxset_new(NULL, NULL);
-
i->resample_method = data->resample_method;
i->sample_spec = data->sample_spec;
i->channel_map = data->channel_map;
@@ -237,6 +239,8 @@ pa_sink_input* pa_sink_input_new(
} else
i->sync_next = i->sync_prev = NULL;
+ i->direct_outputs = pa_idxset_new(NULL, NULL);
+
reset_callbacks(i);
i->userdata = NULL;
@@ -279,6 +283,7 @@ pa_sink_input* pa_sink_input_new(
return i;
}
+/* Called from main context */
static void update_n_corked(pa_sink_input *i, pa_sink_input_state_t state) {
pa_assert(i);
@@ -290,6 +295,7 @@ static void update_n_corked(pa_sink_input *i, pa_sink_input_state_t state) {
pa_sink_update_status(i->sink);
}
+/* Called from main context */
static int sink_input_set_state(pa_sink_input *i, pa_sink_input_state_t state) {
pa_sink_input *ssync;
pa_assert(i);
@@ -300,8 +306,7 @@ static int sink_input_set_state(pa_sink_input *i, pa_sink_input_state_t state) {
if (i->state == state)
return 0;
- if (pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) < 0)
- return -1;
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) == 0);
update_n_corked(i, state);
i->state = state;
@@ -315,12 +320,20 @@ static int sink_input_set_state(pa_sink_input *i, pa_sink_input_state_t state) {
ssync->state = state;
}
- if (state != PA_SINK_INPUT_UNLINKED)
+ if (state != PA_SINK_INPUT_UNLINKED) {
pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_STATE_CHANGED], i);
+ for (ssync = i->sync_prev; ssync; ssync = ssync->sync_prev)
+ pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_STATE_CHANGED], ssync);
+
+ for (ssync = i->sync_next; ssync; ssync = ssync->sync_next)
+ pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_STATE_CHANGED], ssync);
+ }
+
return 0;
}
+/* Called from main context */
void pa_sink_input_unlink(pa_sink_input *i) {
pa_bool_t linked;
pa_source_output *o, *p = NULL;
@@ -358,7 +371,7 @@ void pa_sink_input_unlink(pa_sink_input *i) {
if (linked)
if (i->sink->asyncmsgq)
- pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_REMOVE_INPUT, i, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_REMOVE_INPUT, i, 0, NULL) == 0);
reset_callbacks(i);
@@ -371,6 +384,7 @@ void pa_sink_input_unlink(pa_sink_input *i) {
pa_sink_input_unref(i);
}
+/* Called from main context */
static void sink_input_free(pa_object *o) {
pa_sink_input* i = PA_SINK_INPUT(o);
@@ -403,13 +417,17 @@ static void sink_input_free(pa_object *o) {
pa_xfree(i);
}
+/* Called from main context */
void pa_sink_input_put(pa_sink_input *i) {
pa_sink_input_state_t state;
pa_sink_input_assert_ref(i);
pa_assert(i->state == PA_SINK_INPUT_INIT);
+
+ /* The following fields must be initialized properly */
pa_assert(i->pop);
pa_assert(i->process_rewind);
+ pa_assert(i->kill);
i->thread_info.volume = i->volume;
i->thread_info.muted = i->muted;
@@ -419,33 +437,36 @@ void pa_sink_input_put(pa_sink_input *i) {
update_n_corked(i, state);
i->state = state;
- pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_ADD_INPUT, i, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_ADD_INPUT, i, 0, NULL) == 0);
pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_NEW, i->index);
pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT], i);
}
+/* Called from main context */
void pa_sink_input_kill(pa_sink_input*i) {
pa_sink_input_assert_ref(i);
pa_assert(PA_SINK_INPUT_IS_LINKED(i->state));
- if (i->kill)
- i->kill(i);
+ i->kill(i);
}
-pa_usec_t pa_sink_input_get_latency(pa_sink_input *i) {
- pa_usec_t r = 0;
+/* Called from main context */
+pa_usec_t pa_sink_input_get_latency(pa_sink_input *i, pa_usec_t *sink_latency) {
+ pa_usec_t r[2] = { 0, 0 };
pa_sink_input_assert_ref(i);
pa_assert(PA_SINK_INPUT_IS_LINKED(i->state));
- if (pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_GET_LATENCY, &r, 0, NULL) < 0)
- r = 0;
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_GET_LATENCY, r, 0, NULL) == 0);
if (i->get_latency)
- r += i->get_latency(i);
+ r[0] += i->get_latency(i);
+
+ if (sink_latency)
+ *sink_latency = r[1];
- return r;
+ return r[0];
}
/* Called from thread context */
@@ -698,42 +719,54 @@ void pa_sink_input_update_max_rewind(pa_sink_input *i, size_t nbytes /* in the
i->update_max_rewind(i, i->thread_info.resampler ? pa_resampler_request(i->thread_info.resampler, nbytes) : nbytes);
}
+/* Called from thread context */
+void pa_sink_input_update_max_request(pa_sink_input *i, size_t nbytes /* in the sink's sample spec */) {
+ pa_sink_input_assert_ref(i);
+ pa_assert(PA_SINK_INPUT_IS_LINKED(i->thread_info.state));
+ pa_assert(pa_frame_aligned(nbytes, &i->sink->sample_spec));
+
+ if (i->update_max_request)
+ i->update_max_request(i, i->thread_info.resampler ? pa_resampler_request(i->thread_info.resampler, nbytes) : nbytes);
+}
+
+/* Called from thread context */
static pa_usec_t fixup_latency(pa_sink *s, pa_usec_t usec) {
pa_sink_assert_ref(s);
if (usec == (pa_usec_t) -1)
return usec;
- if (s->max_latency > 0 && usec > s->max_latency)
- usec = s->max_latency;
+ if (s->thread_info.max_latency > 0 && usec > s->thread_info.max_latency)
+ usec = s->thread_info.max_latency;
- if (s->min_latency > 0 && usec < s->min_latency)
- usec = s->min_latency;
+ if (s->thread_info.min_latency > 0 && usec < s->thread_info.min_latency)
+ usec = s->thread_info.min_latency;
return usec;
}
+/* Called from thread context */
pa_usec_t pa_sink_input_set_requested_latency_within_thread(pa_sink_input *i, pa_usec_t usec) {
pa_sink_input_assert_ref(i);
usec = fixup_latency(i->sink, usec);
-
i->thread_info.requested_sink_latency = usec;
pa_sink_invalidate_requested_latency(i->sink);
return usec;
}
+/* Called from main context */
pa_usec_t pa_sink_input_set_requested_latency(pa_sink_input *i, pa_usec_t usec) {
pa_sink_input_assert_ref(i);
- usec = fixup_latency(i->sink, usec);
-
if (PA_SINK_INPUT_IS_LINKED(i->state))
- pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_REQUESTED_LATENCY, NULL, (int64_t) usec, NULL, NULL);
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
else {
/* If this sink input is not realized yet, we have to touch
* the thread info data directly */
+
+ usec = fixup_latency(i->sink, usec);
i->thread_info.requested_sink_latency = usec;
i->sink->thread_info.requested_latency_valid = FALSE;
}
@@ -741,13 +774,14 @@ pa_usec_t pa_sink_input_set_requested_latency(pa_sink_input *i, pa_usec_t usec)
return usec;
}
+/* Called from main context */
pa_usec_t pa_sink_input_get_requested_latency(pa_sink_input *i) {
pa_usec_t usec = 0;
pa_sink_input_assert_ref(i);
if (PA_SINK_INPUT_IS_LINKED(i->state))
- pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
else
/* If this sink input is not realized yet, we have to touch
* the thread info data directly */
@@ -756,19 +790,22 @@ pa_usec_t pa_sink_input_get_requested_latency(pa_sink_input *i) {
return usec;
}
+/* Called from main context */
void pa_sink_input_set_volume(pa_sink_input *i, const pa_cvolume *volume) {
pa_sink_input_assert_ref(i);
pa_assert(PA_SINK_INPUT_IS_LINKED(i->state));
+ pa_assert(volume);
if (pa_cvolume_equal(&i->volume, volume))
return;
i->volume = *volume;
- pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), 0, NULL, pa_xfree);
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_VOLUME, &i->volume, 0, NULL) == 0);
pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, i->index);
}
+/* Called from main context */
const pa_cvolume *pa_sink_input_get_volume(pa_sink_input *i) {
pa_sink_input_assert_ref(i);
pa_assert(PA_SINK_INPUT_IS_LINKED(i->state));
@@ -776,6 +813,7 @@ const pa_cvolume *pa_sink_input_get_volume(pa_sink_input *i) {
return &i->volume;
}
+/* Called from main context */
void pa_sink_input_set_mute(pa_sink_input *i, pa_bool_t mute) {
pa_assert(i);
pa_sink_input_assert_ref(i);
@@ -790,13 +828,15 @@ void pa_sink_input_set_mute(pa_sink_input *i, pa_bool_t mute) {
pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, i->index);
}
-int pa_sink_input_get_mute(pa_sink_input *i) {
+/* Called from main context */
+pa_bool_t pa_sink_input_get_mute(pa_sink_input *i) {
pa_sink_input_assert_ref(i);
pa_assert(PA_SINK_INPUT_IS_LINKED(i->state));
- return !!i->muted;
+ return i->muted;
}
+/* Called from main context */
void pa_sink_input_cork(pa_sink_input *i, pa_bool_t b) {
pa_sink_input_assert_ref(i);
pa_assert(PA_SINK_INPUT_IS_LINKED(i->state));
@@ -804,6 +844,7 @@ void pa_sink_input_cork(pa_sink_input *i, pa_bool_t b) {
sink_input_set_state(i, b ? PA_SINK_INPUT_CORKED : PA_SINK_INPUT_RUNNING);
}
+/* Called from main context */
int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate) {
pa_sink_input_assert_ref(i);
pa_assert(PA_SINK_INPUT_IS_LINKED(i->state));
@@ -820,6 +861,7 @@ int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate) {
return 0;
}
+/* Called from main context */
void pa_sink_input_set_name(pa_sink_input *i, const char *name) {
const char *old;
pa_sink_input_assert_ref(i);
@@ -843,12 +885,14 @@ void pa_sink_input_set_name(pa_sink_input *i, const char *name) {
}
}
+/* Called from main context */
pa_resample_method_t pa_sink_input_get_resample_method(pa_sink_input *i) {
pa_sink_input_assert_ref(i);
return i->resample_method;
}
+/* Called from main context */
int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest) {
pa_resampler *new_resampler;
pa_sink *origin;
@@ -915,7 +959,7 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest) {
hook_data.destination = dest;
pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_MOVE], &hook_data);
- pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_START_MOVE, i, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_START_MOVE, i, 0, NULL) == 0);
pa_idxset_remove_by_data(origin->inputs, i, NULL);
pa_idxset_put(dest->inputs, i, NULL);
@@ -949,7 +993,7 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest) {
pa_sink_update_status(origin);
pa_sink_update_status(dest);
- pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_FINISH_MOVE, i, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_FINISH_MOVE, i, 0, NULL) == 0);
if (i->moved)
i->moved(i);
@@ -964,6 +1008,7 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest) {
return 0;
}
+/* Called from IO thread context */
void pa_sink_input_set_state_within_thread(pa_sink_input *i, pa_sink_input_state_t state) {
pa_sink_input_assert_ref(i);
@@ -993,14 +1038,13 @@ void pa_sink_input_set_state_within_thread(pa_sink_input *i, pa_sink_input_state
i->thread_info.state = state;
}
-/* Called from thread context */
+/* Called from thread context, except when it is not. */
int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
pa_sink_input *i = PA_SINK_INPUT(o);
-
pa_sink_input_assert_ref(i);
- pa_assert(PA_SINK_INPUT_IS_LINKED(i->thread_info.state));
switch (code) {
+
case PA_SINK_INPUT_MESSAGE_SET_VOLUME:
i->thread_info.volume = *((pa_cvolume*) userdata);
pa_sink_input_request_rewind(i, 0, TRUE, FALSE);
@@ -1013,8 +1057,13 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t
case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
pa_usec_t *r = userdata;
+ pa_usec_t sink_usec = 0;
+
+ r[0] += pa_bytes_to_usec(pa_memblockq_get_length(i->thread_info.render_memblockq), &i->sink->sample_spec);
+
+ if (i->sink->parent.process_msg(PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_GET_LATENCY, &sink_usec, 0, NULL) >= 0)
+ r[1] += sink_usec;
- *r += pa_bytes_to_usec(pa_memblockq_get_length(i->thread_info.render_memblockq), &i->sink->sample_spec);
return 0;
}
@@ -1039,10 +1088,12 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t
return 0;
}
- case PA_SINK_INPUT_MESSAGE_SET_REQUESTED_LATENCY:
+ case PA_SINK_INPUT_MESSAGE_SET_REQUESTED_LATENCY: {
+ pa_usec_t *usec = userdata;
- pa_sink_input_set_requested_latency_within_thread(i, (pa_usec_t) offset);
+ *usec = pa_sink_input_set_requested_latency_within_thread(i, *usec);
return 0;
+ }
case PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY: {
pa_usec_t *r = userdata;
@@ -1055,6 +1106,7 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t
return -1;
}
+/* Called from main thread */
pa_sink_input_state_t pa_sink_input_get_state(pa_sink_input *i) {
pa_sink_input_assert_ref(i);
@@ -1074,6 +1126,7 @@ pa_bool_t pa_sink_input_safe_to_remove(pa_sink_input *i) {
return TRUE;
}
+/* Called from IO context */
void pa_sink_input_request_rewind(pa_sink_input *i, size_t nbytes /* in our sample spec */, pa_bool_t rewrite, pa_bool_t flush) {
size_t lbq;
@@ -1133,6 +1186,7 @@ void pa_sink_input_request_rewind(pa_sink_input *i, size_t nbytes /* in our sam
pa_sink_request_rewind(i->sink, nbytes - lbq);
}
+/* Called from main context */
pa_memchunk* pa_sink_input_get_silence(pa_sink_input *i, pa_memchunk *ret) {
pa_sink_input_assert_ref(i);
pa_assert(ret);
diff --git a/src/pulsecore/sink-input.h b/src/pulsecore/sink-input.h
index 2bb7a75..c07a740 100644
--- a/src/pulsecore/sink-input.h
+++ b/src/pulsecore/sink-input.h
@@ -71,8 +71,9 @@ struct pa_sink_input {
pa_sink_input_state_t state;
pa_sink_input_flags_t flags;
- pa_proplist *proplist;
char *driver; /* may be NULL */
+ pa_proplist *proplist;
+
pa_module *module; /* may be NULL */
pa_client *client; /* may be NULL */
@@ -112,6 +113,18 @@ struct pa_sink_input {
* changes. Called from IO context. */
void (*update_max_rewind) (pa_sink_input *i, size_t nbytes); /* may be NULL */
+ /* Called whenever the maxiumum request size of the sink
+ * changes. Called from IO context. */
+ void (*update_max_request) (pa_sink_input *i, size_t nbytes); /* may be NULL */
+
+ /* Called whenever the configured latency of the sink
+ * changes. Called from IO context. */
+ void (*update_sink_requested_latency) (pa_sink_input *i); /* may be NULL */
+
+ /* Called whenver the latency range of the sink changes. Called
+ * from IO context. */
+ void (*update_sink_latency_range) (pa_sink_input *i); /* may be NULL */
+
/* If non-NULL this function is called when the input is first
* connected to a sink or when the rtpoll/asyncmsgq fields
* change. You usually don't need to implement this function
@@ -133,12 +146,12 @@ struct pa_sink_input {
/* Supposed to unlink and destroy this stream. Called from main
* context. */
- void (*kill) (pa_sink_input *i); /* may be NULL */
+ void (*kill) (pa_sink_input *i); /* may NOT be NULL */
/* Return the current latency (i.e. length of bufferd audio) of
- this stream. Called from main context. If NULL a
- PA_SINK_INPUT_MESSAGE_GET_LATENCY message is sent to the IO thread
- instead. */
+ this stream. Called from main context. This is added to what the
+ PA_SINK_INPUT_MESSAGE_GET_LATENCY message sent to the IO thread
+ returns */
pa_usec_t (*get_latency) (pa_sink_input *i); /* may be NULL */
/* If non_NULL this function is called from thread context if the
@@ -258,12 +271,12 @@ int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate);
/* External code may request disconnection with this function */
void pa_sink_input_kill(pa_sink_input*i);
-pa_usec_t pa_sink_input_get_latency(pa_sink_input *i);
+pa_usec_t pa_sink_input_get_latency(pa_sink_input *i, pa_usec_t *sink_latency);
void pa_sink_input_set_volume(pa_sink_input *i, const pa_cvolume *volume);
const pa_cvolume *pa_sink_input_get_volume(pa_sink_input *i);
void pa_sink_input_set_mute(pa_sink_input *i, pa_bool_t mute);
-int pa_sink_input_get_mute(pa_sink_input *i);
+pa_bool_t pa_sink_input_get_mute(pa_sink_input *i);
pa_resample_method_t pa_sink_input_get_resample_method(pa_sink_input *i);
@@ -279,6 +292,7 @@ int pa_sink_input_peek(pa_sink_input *i, size_t length, pa_memchunk *chunk, pa_c
void pa_sink_input_drop(pa_sink_input *i, size_t length);
void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in the sink's sample spec */);
void pa_sink_input_update_max_rewind(pa_sink_input *i, size_t nbytes /* in the sink's sample spec */);
+void pa_sink_input_update_max_request(pa_sink_input *i, size_t nbytes /* in the sink's sample spec */);
void pa_sink_input_set_state_within_thread(pa_sink_input *i, pa_sink_input_state_t state);
diff --git a/src/pulsecore/sink.c b/src/pulsecore/sink.c
index 7b6bb1b..4d3b42c 100644
--- a/src/pulsecore/sink.c
+++ b/src/pulsecore/sink.c
@@ -103,6 +103,7 @@ void pa_sink_new_data_done(pa_sink_new_data *data) {
pa_proplist_free(data->proplist);
}
+/* Called from main context */
static void reset_callbacks(pa_sink *s) {
pa_assert(s);
@@ -115,6 +116,7 @@ static void reset_callbacks(pa_sink *s) {
s->update_requested_latency = NULL;
}
+/* Called from main context */
pa_sink* pa_sink_new(
pa_core *core,
pa_sink_new_data *data,
@@ -190,7 +192,7 @@ pa_sink* pa_sink_new(
s->volume = data->volume;
s->muted = data->muted;
- s->refresh_volume = s->refresh_mute = FALSE;
+ s->refresh_volume = s->refresh_muted = FALSE;
reset_callbacks(s);
s->userdata = NULL;
@@ -205,17 +207,17 @@ pa_sink* pa_sink_new(
&s->sample_spec,
0);
- s->min_latency = DEFAULT_MIN_LATENCY;
- s->max_latency = s->min_latency;
-
s->thread_info.inputs = pa_hashmap_new(pa_idxset_trivial_hash_func, pa_idxset_trivial_compare_func);
- s->thread_info.soft_volume = s->volume;
- s->thread_info.soft_muted = s->muted;
+ pa_cvolume_reset(&s->thread_info.soft_volume, s->sample_spec.channels);
+ s->thread_info.soft_muted = FALSE;
s->thread_info.state = s->state;
s->thread_info.rewind_nbytes = 0;
s->thread_info.max_rewind = 0;
+ s->thread_info.max_request = 0;
s->thread_info.requested_latency_valid = FALSE;
s->thread_info.requested_latency = 0;
+ s->thread_info.min_latency = DEFAULT_MIN_LATENCY;
+ s->thread_info.max_latency = 0;
pa_assert_se(pa_idxset_put(core->sinks, s, &s->index) >= 0);
@@ -246,15 +248,15 @@ pa_sink* pa_sink_new(
return NULL;
}
- s->monitor_source->min_latency = s->min_latency;
- s->monitor_source->max_latency = s->max_latency;
-
s->monitor_source->monitor_of = s;
+
+ pa_source_set_latency_range(s->monitor_source, s->thread_info.min_latency, s->thread_info.max_latency);
pa_source_set_max_rewind(s->monitor_source, s->thread_info.max_rewind);
return s;
}
+/* Called from main context */
static int sink_set_state(pa_sink *s, pa_sink_state_t state) {
int ret;
pa_bool_t suspend_change;
@@ -273,8 +275,7 @@ static int sink_set_state(pa_sink *s, pa_sink_state_t state) {
return -1;
if (s->asyncmsgq)
- if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) < 0)
- return -1;
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) == 0);
s->state = state;
@@ -295,18 +296,26 @@ static int sink_set_state(pa_sink *s, pa_sink_state_t state) {
return 0;
}
+/* Called from main context */
void pa_sink_put(pa_sink* s) {
pa_sink_assert_ref(s);
pa_assert(s->state == PA_SINK_INIT);
+
+ /* The following fields must be initialized properly when calling _put() */
pa_assert(s->asyncmsgq);
pa_assert(s->rtpoll);
+ pa_assert(s->thread_info.max_request > 0);
+ pa_assert(!s->thread_info.min_latency || !s->thread_info.max_latency ||
+ s->thread_info.min_latency <= s->thread_info.max_latency);
- pa_assert(!s->min_latency || !s->max_latency || s->min_latency <= s->max_latency);
-
- if (!(s->flags & PA_SINK_HW_VOLUME_CTRL))
+ if (!(s->flags & PA_SINK_HW_VOLUME_CTRL)) {
s->flags |= PA_SINK_DECIBEL_VOLUME;
+ s->thread_info.soft_volume = s->volume;
+ s->thread_info.soft_muted = s->muted;
+ }
+
pa_assert_se(sink_set_state(s, PA_SINK_IDLE) == 0);
pa_source_put(s->monitor_source);
@@ -315,6 +324,7 @@ void pa_sink_put(pa_sink* s) {
pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SINK_PUT], s);
}
+/* Called from main context */
void pa_sink_unlink(pa_sink* s) {
pa_bool_t linked;
pa_sink_input *i, *j = NULL;
@@ -360,6 +370,7 @@ void pa_sink_unlink(pa_sink* s) {
}
}
+/* Called from main context */
static void sink_free(pa_object *o) {
pa_sink *s = PA_SINK(o);
pa_sink_input *i;
@@ -396,6 +407,7 @@ static void sink_free(pa_object *o) {
pa_xfree(s);
}
+/* Called from main context */
void pa_sink_set_asyncmsgq(pa_sink *s, pa_asyncmsgq *q) {
pa_sink_assert_ref(s);
@@ -405,6 +417,7 @@ void pa_sink_set_asyncmsgq(pa_sink *s, pa_asyncmsgq *q) {
pa_source_set_asyncmsgq(s->monitor_source, q);
}
+/* Called from main context */
void pa_sink_set_rtpoll(pa_sink *s, pa_rtpoll *p) {
pa_sink_assert_ref(s);
@@ -413,6 +426,7 @@ void pa_sink_set_rtpoll(pa_sink *s, pa_rtpoll *p) {
pa_source_set_rtpoll(s->monitor_source, p);
}
+/* Called from main context */
int pa_sink_update_status(pa_sink*s) {
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->state));
@@ -423,6 +437,7 @@ int pa_sink_update_status(pa_sink*s) {
return sink_set_state(s, pa_sink_used_by(s) ? PA_SINK_RUNNING : PA_SINK_IDLE);
}
+/* Called from main context */
int pa_sink_suspend(pa_sink *s, pa_bool_t suspend) {
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->state));
@@ -433,11 +448,12 @@ int pa_sink_suspend(pa_sink *s, pa_bool_t suspend) {
return sink_set_state(s, pa_sink_used_by(s) ? PA_SINK_RUNNING : PA_SINK_IDLE);
}
+/* Called from IO thread context */
void pa_sink_process_rewind(pa_sink *s, size_t nbytes) {
pa_sink_input *i;
void *state = NULL;
pa_sink_assert_ref(s);
- pa_assert(PA_SINK_IS_LINKED(s->state));
+ pa_assert(PA_SINK_IS_LINKED(s->thread_info.state));
/* Make sure the sink code already reset the counter! */
pa_assert(s->thread_info.rewind_nbytes <= 0);
@@ -452,11 +468,11 @@ void pa_sink_process_rewind(pa_sink *s, size_t nbytes) {
pa_sink_input_process_rewind(i, nbytes);
}
- if (s->monitor_source && PA_SOURCE_IS_OPENED(pa_source_get_state(s->monitor_source)))
+ if (s->monitor_source && PA_SOURCE_IS_OPENED(s->monitor_source->thread_info.state))
pa_source_process_rewind(s->monitor_source, nbytes);
-
}
+/* Called from IO thread context */
static unsigned fill_mix_info(pa_sink *s, size_t *length, pa_mix_info *info, unsigned maxinfo) {
pa_sink_input *i;
unsigned n = 0;
@@ -496,6 +512,7 @@ static unsigned fill_mix_info(pa_sink *s, size_t *length, pa_mix_info *info, uns
return n;
}
+/* Called from IO thread context */
static void inputs_drop(pa_sink *s, pa_mix_info *info, unsigned n, pa_memchunk *result) {
pa_sink_input *i;
void *state = NULL;
@@ -591,6 +608,7 @@ static void inputs_drop(pa_sink *s, pa_mix_info *info, unsigned n, pa_memchunk *
pa_source_post(s->monitor_source, result);
}
+/* Called from IO thread context */
void pa_sink_render(pa_sink*s, size_t length, pa_memchunk *result) {
pa_mix_info info[MAX_MIX_CHANNELS];
unsigned n;
@@ -664,6 +682,7 @@ void pa_sink_render(pa_sink*s, size_t length, pa_memchunk *result) {
pa_sink_unref(s);
}
+/* Called from IO thread context */
void pa_sink_render_into(pa_sink*s, pa_memchunk *target) {
pa_mix_info info[MAX_MIX_CHANNELS];
unsigned n;
@@ -742,6 +761,7 @@ void pa_sink_render_into(pa_sink*s, pa_memchunk *target) {
pa_sink_unref(s);
}
+/* Called from IO thread context */
void pa_sink_render_into_full(pa_sink *s, pa_memchunk *target) {
pa_memchunk chunk;
size_t l, d;
@@ -773,6 +793,7 @@ void pa_sink_render_into_full(pa_sink *s, pa_memchunk *target) {
pa_sink_unref(s);
}
+/* Called from IO thread context */
void pa_sink_render_full(pa_sink *s, size_t length, pa_memchunk *result) {
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_OPENED(s->thread_info.state));
@@ -791,6 +812,7 @@ void pa_sink_render_full(pa_sink *s, size_t length, pa_memchunk *result) {
pa_sink_render_into_full(s, result);
}
+/* Called from main thread */
pa_usec_t pa_sink_get_latency(pa_sink *s) {
pa_usec_t usec = 0;
@@ -802,14 +824,14 @@ pa_usec_t pa_sink_get_latency(pa_sink *s) {
if (!PA_SINK_IS_OPENED(s->state))
return 0;
- if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
- return 0;
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) == 0);
return usec;
}
+/* Called from main thread */
void pa_sink_set_volume(pa_sink *s, const pa_cvolume *volume) {
- int changed;
+ pa_bool_t changed;
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->state));
@@ -822,34 +844,36 @@ void pa_sink_set_volume(pa_sink *s, const pa_cvolume *volume) {
s->set_volume = NULL;
if (!s->set_volume)
- pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), 0, NULL, pa_xfree);
+ pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_VOLUME, volume, 0, NULL);
if (changed)
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
}
+/* Called from main thread */
const pa_cvolume *pa_sink_get_volume(pa_sink *s) {
- struct pa_cvolume old_volume;
-
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->state));
- old_volume = s->volume;
+ if (s->refresh_volume) {
+ struct pa_cvolume old_volume = s->volume;
- if (s->get_volume && s->get_volume(s) < 0)
- s->get_volume = NULL;
+ if (s->get_volume && s->get_volume(s) < 0)
+ s->get_volume = NULL;
- if (!s->get_volume && s->refresh_volume)
- pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_VOLUME, &s->volume, 0, NULL);
+ if (!s->get_volume)
+ pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_VOLUME, &s->volume, 0, NULL);
- if (!pa_cvolume_equal(&old_volume, &s->volume))
- pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
+ if (!pa_cvolume_equal(&old_volume, &s->volume))
+ pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
+ }
return &s->volume;
}
+/* Called from main thread */
void pa_sink_set_mute(pa_sink *s, pa_bool_t mute) {
- int changed;
+ pa_bool_t changed;
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->state));
@@ -867,26 +891,29 @@ void pa_sink_set_mute(pa_sink *s, pa_bool_t mute) {
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
}
+/* Called from main thread */
pa_bool_t pa_sink_get_mute(pa_sink *s) {
- pa_bool_t old_muted;
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->state));
- old_muted = s->muted;
+ if (s->refresh_muted) {
+ pa_bool_t old_muted = s->muted;
- if (s->get_mute && s->get_mute(s) < 0)
- s->get_mute = NULL;
+ if (s->get_mute && s->get_mute(s) < 0)
+ s->get_mute = NULL;
- if (!s->get_mute && s->refresh_mute)
- pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_MUTE, &s->muted, 0, NULL);
+ if (!s->get_mute)
+ pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_MUTE, &s->muted, 0, NULL);
- if (old_muted != s->muted)
- pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
+ if (old_muted != s->muted)
+ pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
+ }
return s->muted;
}
+/* Called from main thread */
void pa_sink_set_description(pa_sink *s, const char *description) {
const char *old;
pa_sink_assert_ref(s);
@@ -918,6 +945,7 @@ void pa_sink_set_description(pa_sink *s, const char *description) {
}
}
+/* Called from main thread */
unsigned pa_sink_linked_by(pa_sink *s) {
unsigned ret;
@@ -935,6 +963,7 @@ unsigned pa_sink_linked_by(pa_sink *s) {
return ret;
}
+/* Called from main thread */
unsigned pa_sink_used_by(pa_sink *s) {
unsigned ret;
@@ -950,10 +979,10 @@ unsigned pa_sink_used_by(pa_sink *s) {
return ret - s->n_corked;
}
+/* Called from IO thread, except when it is not */
int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
pa_sink *s = PA_SINK(o);
pa_sink_assert_ref(s);
- pa_assert(s->thread_info.state != PA_SINK_UNLINKED);
switch ((pa_sink_message_t) code) {
@@ -991,6 +1020,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
pa_sink_input_set_state_within_thread(i, i->state);
pa_sink_input_update_max_rewind(i, s->thread_info.max_rewind);
+ pa_sink_input_update_max_request(i, s->thread_info.max_request);
pa_sink_invalidate_requested_latency(s);
@@ -1106,6 +1136,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
i->attach(i);
pa_sink_input_update_max_rewind(i, s->thread_info.max_rewind);
+ pa_sink_input_update_max_request(i, s->thread_info.max_request);
pa_sink_input_set_requested_latency_within_thread(i, i->thread_info.requested_sink_latency);
@@ -1156,9 +1187,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
case PA_SINK_MESSAGE_DETACH:
- /* We're detaching all our input streams so that the
- * asyncmsgq and rtpoll fields can be changed without
- * problems */
+ /* Detach all streams */
pa_sink_detach_within_thread(s);
return 0;
@@ -1172,9 +1201,40 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
pa_usec_t *usec = userdata;
*usec = pa_sink_get_requested_latency_within_thread(s);
+
+ if (*usec == (pa_usec_t) -1)
+ *usec = s->thread_info.max_latency;
+
+ return 0;
+ }
+
+ case PA_SINK_MESSAGE_SET_LATENCY_RANGE: {
+ pa_usec_t *r = userdata;
+
+ pa_sink_update_latency_range(s, r[0], r[1]);
+
+ return 0;
+ }
+
+ case PA_SINK_MESSAGE_GET_LATENCY_RANGE: {
+ pa_usec_t *r = userdata;
+
+ r[0] = s->thread_info.min_latency;
+ r[1] = s->thread_info.max_latency;
+
return 0;
}
+ case PA_SINK_MESSAGE_GET_MAX_REWIND:
+
+ *((size_t*) userdata) = s->thread_info.max_rewind;
+ return 0;
+
+ case PA_SINK_MESSAGE_GET_MAX_REQUEST:
+
+ *((size_t*) userdata) = s->thread_info.max_request;
+ return 0;
+
case PA_SINK_MESSAGE_GET_LATENCY:
case PA_SINK_MESSAGE_MAX:
;
@@ -1183,6 +1243,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
return -1;
}
+/* Called from main thread */
int pa_sink_suspend_all(pa_core *c, pa_bool_t suspend) {
pa_sink *sink;
uint32_t idx;
@@ -1196,20 +1257,23 @@ int pa_sink_suspend_all(pa_core *c, pa_bool_t suspend) {
return ret;
}
+/* Called from main thread */
void pa_sink_detach(pa_sink *s) {
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->state));
- pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_DETACH, NULL, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_DETACH, NULL, 0, NULL) == 0);
}
+/* Called from main thread */
void pa_sink_attach(pa_sink *s) {
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->state));
- pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_ATTACH, NULL, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_ATTACH, NULL, 0, NULL) == 0);
}
+/* Called from IO thread */
void pa_sink_detach_within_thread(pa_sink *s) {
pa_sink_input *i;
void *state = NULL;
@@ -1225,6 +1289,7 @@ void pa_sink_detach_within_thread(pa_sink *s) {
pa_source_detach_within_thread(s->monitor_source);
}
+/* Called from IO thread */
void pa_sink_attach_within_thread(pa_sink *s) {
pa_sink_input *i;
void *state = NULL;
@@ -1240,6 +1305,7 @@ void pa_sink_attach_within_thread(pa_sink *s) {
pa_source_attach_within_thread(s->monitor_source);
}
+/* Called from IO thread */
void pa_sink_request_rewind(pa_sink*s, size_t nbytes) {
pa_sink_assert_ref(s);
pa_assert(PA_SINK_IS_LINKED(s->thread_info.state));
@@ -1258,6 +1324,7 @@ void pa_sink_request_rewind(pa_sink*s, size_t nbytes) {
s->request_rewind(s);
}
+/* Called from IO thread */
pa_usec_t pa_sink_get_requested_latency_within_thread(pa_sink *s) {
pa_usec_t result = (pa_usec_t) -1;
pa_sink_input *i;
@@ -1282,11 +1349,11 @@ pa_usec_t pa_sink_get_requested_latency_within_thread(pa_sink *s) {
result = monitor_latency;
if (result != (pa_usec_t) -1) {
- if (s->max_latency > 0 && result > s->max_latency)
- result = s->max_latency;
+ if (s->thread_info.max_latency > 0 && result > s->thread_info.max_latency)
+ result = s->thread_info.max_latency;
- if (s->min_latency > 0 && result < s->min_latency)
- result = s->min_latency;
+ if (s->thread_info.min_latency > 0 && result < s->thread_info.min_latency)
+ result = s->thread_info.min_latency;
}
s->thread_info.requested_latency = result;
@@ -1295,6 +1362,7 @@ pa_usec_t pa_sink_get_requested_latency_within_thread(pa_sink *s) {
return result;
}
+/* Called from main thread */
pa_usec_t pa_sink_get_requested_latency(pa_sink *s) {
pa_usec_t usec = 0;
@@ -1304,15 +1372,11 @@ pa_usec_t pa_sink_get_requested_latency(pa_sink *s) {
if (!PA_SINK_IS_OPENED(s->state))
return 0;
- if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) < 0)
- return 0;
-
- if (usec == (pa_usec_t) -1)
- usec = s->max_latency;
-
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
return usec;
}
+/* Called from IO thread */
void pa_sink_set_max_rewind(pa_sink *s, size_t max_rewind) {
pa_sink_input *i;
void *state = NULL;
@@ -1324,14 +1388,37 @@ void pa_sink_set_max_rewind(pa_sink *s, size_t max_rewind) {
s->thread_info.max_rewind = max_rewind;
- while ((i = pa_hashmap_iterate(s->thread_info.inputs, &state, NULL)))
- pa_sink_input_update_max_rewind(i, s->thread_info.max_rewind);
+ if (PA_SINK_IS_LINKED(s->thread_info.state)) {
+ while ((i = pa_hashmap_iterate(s->thread_info.inputs, &state, NULL)))
+ pa_sink_input_update_max_rewind(i, s->thread_info.max_rewind);
+ }
if (s->monitor_source)
pa_source_set_max_rewind(s->monitor_source, s->thread_info.max_rewind);
}
+/* Called from IO thread */
+void pa_sink_set_max_request(pa_sink *s, size_t max_request) {
+ pa_sink_input *i;
+ void *state = NULL;
+
+ pa_sink_assert_ref(s);
+
+ if (max_request == s->thread_info.max_request)
+ return;
+
+ s->thread_info.max_request = max_request;
+
+ if (PA_SINK_IS_LINKED(s->thread_info.state)) {
+ while ((i = pa_hashmap_iterate(s->thread_info.inputs, &state, NULL)))
+ pa_sink_input_update_max_request(i, s->thread_info.max_request);
+ }
+}
+
+/* Called from IO thread */
void pa_sink_invalidate_requested_latency(pa_sink *s) {
+ pa_sink_input *i;
+ void *state = NULL;
pa_sink_assert_ref(s);
@@ -1339,8 +1426,13 @@ void pa_sink_invalidate_requested_latency(pa_sink *s) {
if (s->update_requested_latency)
s->update_requested_latency(s);
+
+ while ((i = pa_hashmap_iterate(s->thread_info.inputs, &state, NULL)))
+ if (i->update_sink_requested_latency)
+ i->update_sink_requested_latency(i);
}
+/* Called from main thread */
void pa_sink_set_latency_range(pa_sink *s, pa_usec_t min_latency, pa_usec_t max_latency) {
pa_sink_assert_ref(s);
@@ -1359,8 +1451,82 @@ void pa_sink_set_latency_range(pa_sink *s, pa_usec_t min_latency, pa_usec_t max_
pa_assert(!min_latency || !max_latency ||
min_latency <= max_latency);
- s->min_latency = min_latency;
- s->max_latency = max_latency;
+ if (PA_SINK_IS_LINKED(s->state)) {
+ pa_usec_t r[2];
+
+ r[0] = min_latency;
+ r[1] = max_latency;
+
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_LATENCY_RANGE, r, 0, NULL) == 0);
+ } else {
+ s->thread_info.min_latency = min_latency;
+ s->thread_info.max_latency = max_latency;
+
+ s->monitor_source->thread_info.min_latency = min_latency;
+ s->monitor_source->thread_info.max_latency = max_latency;
+
+ s->thread_info.requested_latency_valid = s->monitor_source->thread_info.requested_latency_valid = FALSE;
+ }
+}
+
+/* Called from main thread */
+void pa_sink_get_latency_range(pa_sink *s, pa_usec_t *min_latency, pa_usec_t *max_latency) {
+ pa_sink_assert_ref(s);
+ pa_assert(min_latency);
+ pa_assert(max_latency);
+
+ if (PA_SINK_IS_LINKED(s->state)) {
+ pa_usec_t r[2] = { 0, 0 };
+
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY_RANGE, r, 0, NULL) == 0);
+
+ *min_latency = r[0];
+ *max_latency = r[1];
+ } else {
+ *min_latency = s->thread_info.min_latency;
+ *max_latency = s->thread_info.max_latency;
+ }
+}
+
+/* Called from IO thread */
+void pa_sink_update_latency_range(pa_sink *s, pa_usec_t min_latency, pa_usec_t max_latency) {
+ pa_sink_input *i;
+ void *state = NULL;
+
+ pa_sink_assert_ref(s);
+
+ s->thread_info.min_latency = min_latency;
+ s->thread_info.max_latency = max_latency;
+
+ while ((i = pa_hashmap_iterate(s->thread_info.inputs, &state, NULL)))
+ if (i->update_sink_latency_range)
+ i->update_sink_latency_range(i);
+
+ pa_sink_invalidate_requested_latency(s);
+
+ pa_source_update_latency_range(s->monitor_source, min_latency, max_latency);
+}
+
+size_t pa_sink_get_max_rewind(pa_sink *s) {
+ size_t r;
+ pa_sink_assert_ref(s);
+
+ if (!PA_SINK_IS_LINKED(s->state))
+ return s->thread_info.max_rewind;
+
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_MAX_REWIND, &r, 0, NULL) == 0);
+
+ return r;
+}
+
+size_t pa_sink_get_max_request(pa_sink *s) {
+ size_t r;
+ pa_sink_assert_ref(s);
+
+ if (!PA_SINK_IS_LINKED(s->state))
+ return s->thread_info.max_request;
+
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_MAX_REQUEST, &r, 0, NULL) == 0);
- pa_source_set_latency_range(s->monitor_source, min_latency, max_latency);
+ return r;
}
diff --git a/src/pulsecore/sink.h b/src/pulsecore/sink.h
index 4405896..b73944e 100644
--- a/src/pulsecore/sink.h
+++ b/src/pulsecore/sink.h
@@ -80,17 +80,15 @@ struct pa_sink {
pa_cvolume volume;
pa_bool_t muted;
- pa_bool_t refresh_volume;
- pa_bool_t refresh_mute;
+
+ pa_bool_t refresh_volume:1;
+ pa_bool_t refresh_muted:1;
pa_asyncmsgq *asyncmsgq;
pa_rtpoll *rtpoll;
pa_memchunk silence;
- pa_usec_t min_latency; /* we won't go below this latency */
- pa_usec_t max_latency; /* An upper limit for the latencies */
-
/* Called when the main loop requests a state change. Called from
* main loop context. If returns -1 the state change will be
* inhibited */
@@ -98,8 +96,9 @@ struct pa_sink {
/* Callled when the volume is queried. Called from main loop
* context. If this is NULL a PA_SINK_MESSAGE_GET_VOLUME message
- * will be sent to the IO thread instead. */
- int (*get_volume)(pa_sink *s); /* may be null */
+ * will be sent to the IO thread instead. If refresh_volume is
+ * FALSE neither this function is called nor a message is sent. */
+ int (*get_volume)(pa_sink *s); /* may be NULL */
/* Called when the volume shall be changed. Called from main loop
* context. If this is NULL a PA_SINK_MESSAGE_SET_VOLUME message
@@ -108,7 +107,8 @@ struct pa_sink {
/* Called when the mute setting is queried. Called from main loop
* context. If this is NULL a PA_SINK_MESSAGE_GET_MUTE message
- * will be sent to the IO thread instead. */
+ * will be sent to the IO thread instead. If refresh_mute is
+ * FALSE neither this function is called nor a message is sent.*/
int (*get_mute)(pa_sink *s); /* dito */
/* Called when the mute setting shall be changed. Called from main
@@ -130,17 +130,24 @@ struct pa_sink {
pa_sink_state_t state;
pa_hashmap *inputs;
pa_cvolume soft_volume;
- pa_bool_t soft_muted;
+ pa_bool_t soft_muted:1;
- pa_bool_t requested_latency_valid;
+ pa_bool_t requested_latency_valid:1;
pa_usec_t requested_latency;
- /* The number of bytes we need keep around to be able to satisfy
- * every DMA buffer rewrite */
+ /* The number of bytes streams need to keep around as history to
+ * be able to satisfy every DMA buffer rewrite */
size_t max_rewind;
+ /* The number of bytes streams need to keep around to satisfy
+ * every DMA write request */
+ size_t max_request;
+
/* Maximum of what clients requested to rewind in this cycle */
size_t rewind_nbytes;
+
+ pa_usec_t min_latency; /* we won't go below this latency */
+ pa_usec_t max_latency; /* An upper limit for the latencies */
} thread_info;
void *userdata;
@@ -163,6 +170,10 @@ typedef enum pa_sink_message {
PA_SINK_MESSAGE_FINISH_MOVE,
PA_SINK_MESSAGE_ATTACH,
PA_SINK_MESSAGE_DETACH,
+ PA_SINK_MESSAGE_SET_LATENCY_RANGE,
+ PA_SINK_MESSAGE_GET_LATENCY_RANGE,
+ PA_SINK_MESSAGE_GET_MAX_REWIND,
+ PA_SINK_MESSAGE_GET_MAX_REQUEST,
PA_SINK_MESSAGE_MAX
} pa_sink_message_t;
@@ -217,6 +228,10 @@ void pa_sink_attach(pa_sink *s);
/* The returned value is supposed to be in the time domain of the sound card! */
pa_usec_t pa_sink_get_latency(pa_sink *s);
pa_usec_t pa_sink_get_requested_latency(pa_sink *s);
+void pa_sink_get_latency_range(pa_sink *s, pa_usec_t *min_latency, pa_usec_t *max_latency);
+
+size_t pa_sink_get_max_rewind(pa_sink *s);
+size_t pa_sink_get_max_request(pa_sink *s);
int pa_sink_update_status(pa_sink*s);
int pa_sink_suspend(pa_sink *s, pa_bool_t suspend);
@@ -248,6 +263,9 @@ void pa_sink_detach_within_thread(pa_sink *s);
pa_usec_t pa_sink_get_requested_latency_within_thread(pa_sink *s);
void pa_sink_set_max_rewind(pa_sink *s, size_t max_rewind);
+void pa_sink_set_max_request(pa_sink *s, size_t max_request);
+
+void pa_sink_update_latency_range(pa_sink *s, pa_usec_t min_latency, pa_usec_t max_latency);
/* To be called exclusively by sink input drivers, from IO context */
diff --git a/src/pulsecore/source-output.c b/src/pulsecore/source-output.c
index dbb362f..3d1abe3 100644
--- a/src/pulsecore/source-output.c
+++ b/src/pulsecore/source-output.c
@@ -74,12 +74,15 @@ void pa_source_output_new_data_done(pa_source_output_new_data *data) {
pa_proplist_free(data->proplist);
}
+/* Called from main context */
static void reset_callbacks(pa_source_output *o) {
pa_assert(o);
o->push = NULL;
o->process_rewind = NULL;
o->update_max_rewind = NULL;
+ o->update_source_requested_latency = NULL;
+ o->update_source_latency_range = NULL;
o->attach = NULL;
o->detach = NULL;
o->suspend = NULL;
@@ -89,6 +92,7 @@ static void reset_callbacks(pa_source_output *o) {
o->state_change = NULL;
}
+/* Called from main context */
pa_source_output* pa_source_output_new(
pa_core *core,
pa_source_output_new_data *data,
@@ -107,7 +111,7 @@ pa_source_output* pa_source_output_new(
pa_return_null_if_fail(!data->driver || pa_utf8_valid(data->driver));
if (!data->source)
- data->source = pa_namereg_get(core, NULL, PA_NAMEREG_SOURCE, 1);
+ data->source = pa_namereg_get(core, NULL, PA_NAMEREG_SOURCE, TRUE);
pa_return_null_if_fail(data->source);
pa_return_null_if_fail(pa_source_get_state(data->source) != PA_SOURCE_UNLINKED);
@@ -232,6 +236,7 @@ pa_source_output* pa_source_output_new(
return o;
}
+/* Called from main context */
static void update_n_corked(pa_source_output *o, pa_source_output_state_t state) {
pa_assert(o);
@@ -243,14 +248,14 @@ static void update_n_corked(pa_source_output *o, pa_source_output_state_t state)
pa_source_update_status(o->source);
}
+/* Called from main context */
static int source_output_set_state(pa_source_output *o, pa_source_output_state_t state) {
pa_assert(o);
if (o->state == state)
return 0;
- if (pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) < 0)
- return -1;
+ pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) == 0);
update_n_corked(o, state);
o->state = state;
@@ -261,6 +266,7 @@ static int source_output_set_state(pa_source_output *o, pa_source_output_state_t
return 0;
}
+/* Called from main context */
void pa_source_output_unlink(pa_source_output*o) {
pa_bool_t linked;
pa_assert(o);
@@ -286,7 +292,7 @@ void pa_source_output_unlink(pa_source_output*o) {
if (linked)
if (o->source->asyncmsgq)
- pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_REMOVE_OUTPUT, o, 0, NULL) == 0);
reset_callbacks(o);
@@ -299,6 +305,7 @@ void pa_source_output_unlink(pa_source_output*o) {
pa_source_output_unref(o);
}
+/* Called from main context */
static void source_output_free(pa_object* mo) {
pa_source_output *o = PA_SOURCE_OUTPUT(mo);
@@ -325,45 +332,52 @@ static void source_output_free(pa_object* mo) {
pa_xfree(o);
}
+/* Called from main context */
void pa_source_output_put(pa_source_output *o) {
pa_source_output_state_t state;
pa_source_output_assert_ref(o);
pa_assert(o->state == PA_SOURCE_OUTPUT_INIT);
+
+ /* The following fields must be initialized properly */
pa_assert(o->push);
+ pa_assert(o->kill);
state = o->flags & PA_SOURCE_OUTPUT_START_CORKED ? PA_SOURCE_OUTPUT_CORKED : PA_SOURCE_OUTPUT_RUNNING;
update_n_corked(o, state);
o->state = state;
- pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, o, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, o, 0, NULL) == 0);
pa_subscription_post(o->source->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_NEW, o->index);
pa_hook_fire(&o->source->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_PUT], o);
}
+/* Called from main context */
void pa_source_output_kill(pa_source_output*o) {
pa_source_output_assert_ref(o);
pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(o->state));
- if (o->kill)
- o->kill(o);
+ o->kill(o);
}
-pa_usec_t pa_source_output_get_latency(pa_source_output *o) {
- pa_usec_t r = 0;
+/* Called from main context */
+pa_usec_t pa_source_output_get_latency(pa_source_output *o, pa_usec_t *source_latency) {
+ pa_usec_t r[2] = { 0, 0 };
pa_source_output_assert_ref(o);
pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(o->state));
- if (pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY, &r, 0, NULL) < 0)
- r = 0;
+ pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY, r, 0, NULL) == 0);
if (o->get_latency)
- r += o->get_latency(o);
+ r[0] += o->get_latency(o);
- return r;
+ if (source_latency)
+ *source_latency = r[1];
+
+ return r[0];
}
/* Called from thread context */
@@ -464,42 +478,44 @@ void pa_source_output_update_max_rewind(pa_source_output *o, size_t nbytes /* i
o->update_max_rewind(o, o->thread_info.resampler ? pa_resampler_result(o->thread_info.resampler, nbytes) : nbytes);
}
+/* Called from thread context */
static pa_usec_t fixup_latency(pa_source *s, pa_usec_t usec) {
pa_source_assert_ref(s);
if (usec == (pa_usec_t) -1)
return usec;
- if (s->max_latency > 0 && usec > s->max_latency)
- usec = s->max_latency;
+ if (s->thread_info.max_latency > 0 && usec > s->thread_info.max_latency)
+ usec = s->thread_info.max_latency;
- if (s->min_latency > 0 && usec < s->min_latency)
- usec = s->min_latency;
+ if (s->thread_info.min_latency > 0 && usec < s->thread_info.min_latency)
+ usec = s->thread_info.min_latency;
return usec;
}
+/* Called from thread context */
pa_usec_t pa_source_output_set_requested_latency_within_thread(pa_source_output *o, pa_usec_t usec) {
pa_source_output_assert_ref(o);
usec = fixup_latency(o->source, usec);
-
o->thread_info.requested_source_latency = usec;
pa_source_invalidate_requested_latency(o->source);
return usec;
}
+/* Called from main context */
pa_usec_t pa_source_output_set_requested_latency(pa_source_output *o, pa_usec_t usec) {
pa_source_output_assert_ref(o);
- usec = fixup_latency(o->source, usec);
-
if (PA_SOURCE_OUTPUT_IS_LINKED(o->state))
- pa_asyncmsgq_post(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_SET_REQUESTED_LATENCY, NULL, (int64_t) usec, NULL, NULL);
+ pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_SET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
else {
/* If this sink input is not realized yet, we have to touch
* the thread info data directly */
+
+ usec = fixup_latency(o->source, usec);
o->thread_info.requested_source_latency = usec;
o->source->thread_info.requested_latency_valid = FALSE;
}
@@ -507,13 +523,14 @@ pa_usec_t pa_source_output_set_requested_latency(pa_source_output *o, pa_usec_t
return usec;
}
+/* Called from main context */
pa_usec_t pa_source_output_get_requested_latency(pa_source_output *o) {
pa_usec_t usec = 0;
pa_source_output_assert_ref(o);
if (PA_SOURCE_OUTPUT_IS_LINKED(o->state))
- pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
else
/* If this sink input is not realized yet, we have to touch
* the thread info data directly */
@@ -522,6 +539,7 @@ pa_usec_t pa_source_output_get_requested_latency(pa_source_output *o) {
return usec;
}
+/* Called from main context */
void pa_source_output_cork(pa_source_output *o, pa_bool_t b) {
pa_source_output_assert_ref(o);
pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(o->state));
@@ -529,6 +547,7 @@ void pa_source_output_cork(pa_source_output *o, pa_bool_t b) {
source_output_set_state(o, b ? PA_SOURCE_OUTPUT_CORKED : PA_SOURCE_OUTPUT_RUNNING);
}
+/* Called from main context */
int pa_source_output_set_rate(pa_source_output *o, uint32_t rate) {
pa_source_output_assert_ref(o);
pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(o->state));
@@ -545,6 +564,7 @@ int pa_source_output_set_rate(pa_source_output *o, uint32_t rate) {
return 0;
}
+/* Called from main context */
void pa_source_output_set_name(pa_source_output *o, const char *name) {
const char *old;
pa_source_output_assert_ref(o);
@@ -568,12 +588,14 @@ void pa_source_output_set_name(pa_source_output *o, const char *name) {
}
}
+/* Called from main context */
pa_resample_method_t pa_source_output_get_resample_method(pa_source_output *o) {
pa_source_output_assert_ref(o);
return o->resample_method;
}
+/* Called from main context */
int pa_source_output_move_to(pa_source_output *o, pa_source *dest) {
pa_source *origin;
pa_resampler *new_resampler;
@@ -631,7 +653,7 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) {
pa_hook_fire(&o->source->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_MOVE], &hook_data);
/* Okey, let's move it */
- pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_REMOVE_OUTPUT, o, 0, NULL) == 0);
pa_idxset_remove_by_data(origin->outputs, o, NULL);
pa_idxset_put(dest->outputs, o, NULL);
@@ -664,7 +686,7 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) {
pa_source_update_status(origin);
pa_source_update_status(dest);
- pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, o, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, o, 0, NULL) == 0);
if (o->moved)
o->moved(o);
@@ -679,6 +701,7 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) {
return 0;
}
+/* Called from IO thread context */
void pa_source_output_set_state_within_thread(pa_source_output *o, pa_source_output_state_t state) {
pa_source_output_assert_ref(o);
@@ -691,19 +714,22 @@ void pa_source_output_set_state_within_thread(pa_source_output *o, pa_source_out
o->thread_info.state = state;
}
-/* Called from thread context */
+/* Called from IO thread context, except when it is not */
int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int64_t offset, pa_memchunk* chunk) {
pa_source_output *o = PA_SOURCE_OUTPUT(mo);
-
pa_source_output_assert_ref(o);
- pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(o->thread_info.state));
switch (code) {
case PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY: {
pa_usec_t *r = userdata;
+ pa_usec_t source_usec = 0;
+
+ r[0] += pa_bytes_to_usec(pa_memblockq_get_length(o->thread_info.delay_memblockq), &o->source->sample_spec);
+
+ if (o->source->parent.process_msg(PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_GET_LATENCY, &source_usec, 0, NULL) >= 0)
+ r[1] += source_usec;
- *r += pa_bytes_to_usec(pa_memblockq_get_length(o->thread_info.delay_memblockq), &o->source->sample_spec);
return 0;
}
@@ -718,10 +744,13 @@ int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int
pa_source_output_set_state_within_thread(o, PA_PTR_TO_UINT(userdata));
return 0;
- case PA_SOURCE_OUTPUT_MESSAGE_SET_REQUESTED_LATENCY:
+ case PA_SOURCE_OUTPUT_MESSAGE_SET_REQUESTED_LATENCY: {
+ pa_usec_t *usec = userdata;
+
+ *usec = pa_source_output_set_requested_latency_within_thread(o, *usec);
- pa_source_output_set_requested_latency_within_thread(o, (pa_usec_t) offset);
return 0;
+ }
case PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY: {
pa_usec_t *r = userdata;
diff --git a/src/pulsecore/source-output.h b/src/pulsecore/source-output.h
index fe842eb..61825b2 100644
--- a/src/pulsecore/source-output.h
+++ b/src/pulsecore/source-output.h
@@ -64,8 +64,9 @@ struct pa_source_output {
pa_source_output_state_t state;
pa_source_output_flags_t flags;
- pa_proplist *proplist;
char *driver; /* may be NULL */
+ pa_proplist *proplist;
+
pa_module *module; /* may be NULL */
pa_client *client; /* may be NULL */
@@ -81,16 +82,24 @@ struct pa_source_output {
/* Pushes a new memchunk into the output. Called from IO thread
* context. */
- void (*push)(pa_source_output *o, const pa_memchunk *chunk);
+ void (*push)(pa_source_output *o, const pa_memchunk *chunk); /* may NOT be NULL */
/* Only relevant for monitor sources right now: called when the
- * recorded stream is rewound. Called from IO context*/
- void (*process_rewind)(pa_source_output *o, size_t nbytes);
+ * recorded stream is rewound. Called from IO context */
+ void (*process_rewind)(pa_source_output *o, size_t nbytes); /* may be NULL */
/* Called whenever the maximum rewindable size of the source
* changes. Called from IO thread context. */
void (*update_max_rewind) (pa_source_output *o, size_t nbytes); /* may be NULL */
+ /* Called whenever the configured latency of the source
+ * changes. Called from IO context. */
+ void (*update_source_requested_latency) (pa_source_output *o); /* may be NULL */
+
+ /* Called whenver the latency range of the source changes. Called
+ * from IO context. */
+ void (*update_source_latency_range) (pa_source_output *o); /* may be NULL */
+
/* If non-NULL this function is called when the output is first
* connected to a source. Called from IO thread context */
void (*attach) (pa_source_output *o); /* may be NULL */
@@ -109,12 +118,12 @@ struct pa_source_output {
/* Supposed to unlink and destroy this stream. Called from main
* context. */
- void (*kill)(pa_source_output* o); /* may be NULL */
+ void (*kill)(pa_source_output* o); /* may NOT be NULL */
/* Return the current latency (i.e. length of bufferd audio) of
- this stream. Called from main context. If NULL a
- PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY message is sent to the IO
- thread instead. */
+ this stream. Called from main context. This is added to what the
+ PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY message sent to the IO thread
+ returns */
pa_usec_t (*get_latency) (pa_source_output *o); /* may be NULL */
/* If non_NULL this function is called from thread context if the
@@ -206,7 +215,7 @@ int pa_source_output_set_rate(pa_source_output *o, uint32_t rate);
/* External code may request disconnection with this funcion */
void pa_source_output_kill(pa_source_output*o);
-pa_usec_t pa_source_output_get_latency(pa_source_output *i);
+pa_usec_t pa_source_output_get_latency(pa_source_output *i, pa_usec_t *source_latency);
pa_resample_method_t pa_source_output_get_resample_method(pa_source_output *o);
diff --git a/src/pulsecore/source.c b/src/pulsecore/source.c
index ebd3817..8256a98 100644
--- a/src/pulsecore/source.c
+++ b/src/pulsecore/source.c
@@ -97,6 +97,7 @@ void pa_source_new_data_done(pa_source_new_data *data) {
pa_proplist_free(data->proplist);
}
+/* Called from main context */
static void reset_callbacks(pa_source *s) {
pa_assert(s);
@@ -108,6 +109,7 @@ static void reset_callbacks(pa_source *s) {
s->update_requested_latency = NULL;
}
+/* Called from main context */
pa_source* pa_source_new(
pa_core *core,
pa_source_new_data *data,
@@ -118,6 +120,8 @@ pa_source* pa_source_new(
char st[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX];
pa_assert(core);
+ pa_assert(data);
+ pa_assert(data->name);
s = pa_msgobject_new(pa_source);
@@ -126,6 +130,8 @@ pa_source* pa_source_new(
return NULL;
}
+ pa_source_new_data_set_name(data, name);
+
if (pa_hook_fire(&core->hooks[PA_CORE_HOOK_SOURCE_NEW], data) < 0) {
pa_xfree(s);
pa_namereg_unregister(core, name);
@@ -193,16 +199,15 @@ pa_source* pa_source_new(
&s->sample_spec,
0);
- s->min_latency = DEFAULT_MIN_LATENCY;
- s->max_latency = s->min_latency;
-
s->thread_info.outputs = pa_hashmap_new(pa_idxset_trivial_hash_func, pa_idxset_trivial_compare_func);
- s->thread_info.soft_volume = s->volume;
- s->thread_info.soft_muted = s->muted;
+ pa_cvolume_reset(&s->thread_info.soft_volume, s->sample_spec.channels);
+ s->thread_info.soft_muted = FALSE;
s->thread_info.state = s->state;
s->thread_info.max_rewind = 0;
s->thread_info.requested_latency_valid = FALSE;
s->thread_info.requested_latency = 0;
+ s->thread_info.min_latency = DEFAULT_MIN_LATENCY;
+ s->thread_info.max_latency = 0;
pa_assert_se(pa_idxset_put(core->sources, s, &s->index) >= 0);
@@ -215,6 +220,7 @@ pa_source* pa_source_new(
return s;
}
+/* Called from main context */
static int source_set_state(pa_source *s, pa_source_state_t state) {
int ret;
pa_bool_t suspend_change;
@@ -233,8 +239,7 @@ static int source_set_state(pa_source *s, pa_source_state_t state) {
return -1;
if (s->asyncmsgq)
- if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) < 0)
- return -1;
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) == 0);
s->state = state;
@@ -255,24 +260,32 @@ static int source_set_state(pa_source *s, pa_source_state_t state) {
return 0;
}
+/* Called from main context */
void pa_source_put(pa_source *s) {
pa_source_assert_ref(s);
pa_assert(s->state == PA_SINK_INIT);
+
+ /* The following fields must be initialized properly when calling _put() */
pa_assert(s->asyncmsgq);
pa_assert(s->rtpoll);
+ pa_assert(!s->thread_info.min_latency || !s->thread_info.max_latency ||
+ s->thread_info.min_latency <= s->thread_info.max_latency);
- pa_assert(!s->min_latency || !s->max_latency || s->min_latency <= s->max_latency);
-
- if (!(s->flags & PA_SOURCE_HW_VOLUME_CTRL))
+ if (!(s->flags & PA_SOURCE_HW_VOLUME_CTRL)) {
s->flags |= PA_SOURCE_DECIBEL_VOLUME;
+ s->thread_info.soft_volume = s->volume;
+ s->thread_info.soft_muted = s->muted;
+ }
+
pa_assert_se(source_set_state(s, PA_SOURCE_IDLE) == 0);
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_NEW, s->index);
pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_PUT], s);
}
+/* Called from main context */
void pa_source_unlink(pa_source *s) {
pa_bool_t linked;
pa_source_output *o, *j = NULL;
@@ -310,6 +323,7 @@ void pa_source_unlink(pa_source *s) {
}
}
+/* Called from main context */
static void source_free(pa_object *o) {
pa_source_output *so;
pa_source *s = PA_SOURCE(o);
@@ -341,18 +355,21 @@ static void source_free(pa_object *o) {
pa_xfree(s);
}
+/* Called from main context */
void pa_source_set_asyncmsgq(pa_source *s, pa_asyncmsgq *q) {
pa_source_assert_ref(s);
s->asyncmsgq = q;
}
+/* Called from main context */
void pa_source_set_rtpoll(pa_source *s, pa_rtpoll *p) {
pa_source_assert_ref(s);
s->rtpoll = p;
}
+/* Called from main context */
int pa_source_update_status(pa_source*s) {
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
@@ -363,6 +380,7 @@ int pa_source_update_status(pa_source*s) {
return source_set_state(s, pa_source_used_by(s) ? PA_SOURCE_RUNNING : PA_SOURCE_IDLE);
}
+/* Called from main context */
int pa_source_suspend(pa_source *s, pa_bool_t suspend) {
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
@@ -373,6 +391,7 @@ int pa_source_suspend(pa_source *s, pa_bool_t suspend) {
return source_set_state(s, pa_source_used_by(s) ? PA_SOURCE_RUNNING : PA_SOURCE_IDLE);
}
+/* Called from IO thread context */
void pa_source_process_rewind(pa_source *s, size_t nbytes) {
pa_source_output *o;
void *state = NULL;
@@ -391,6 +410,7 @@ void pa_source_process_rewind(pa_source *s, size_t nbytes) {
}
}
+/* Called from IO thread context */
void pa_source_post(pa_source*s, const pa_memchunk *chunk) {
pa_source_output *o;
void *state = NULL;
@@ -432,6 +452,7 @@ void pa_source_post(pa_source*s, const pa_memchunk *chunk) {
}
}
+/* Called from IO thread context */
void pa_source_post_direct(pa_source*s, pa_source_output *o, const pa_memchunk *chunk) {
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_OPENED(s->thread_info.state));
@@ -460,6 +481,7 @@ void pa_source_post_direct(pa_source*s, pa_source_output *o, const pa_memchunk *
pa_source_output_push(o, chunk);
}
+/* Called from main thread */
pa_usec_t pa_source_get_latency(pa_source *s) {
pa_usec_t usec;
@@ -469,14 +491,14 @@ pa_usec_t pa_source_get_latency(pa_source *s) {
if (!PA_SOURCE_IS_OPENED(s->state))
return 0;
- if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
- return 0;
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) == 0);
return usec;
}
+/* Called from main thread */
void pa_source_set_volume(pa_source *s, const pa_cvolume *volume) {
- int changed;
+ pa_bool_t changed;
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
@@ -489,34 +511,36 @@ void pa_source_set_volume(pa_source *s, const pa_cvolume *volume) {
s->set_volume = NULL;
if (!s->set_volume)
- pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), 0, NULL, pa_xfree);
+ pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, volume, 0, NULL);
if (changed)
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
}
+/* Called from main thread */
const pa_cvolume *pa_source_get_volume(pa_source *s) {
- pa_cvolume old_volume;
-
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
- old_volume = s->volume;
+ if (s->refresh_volume) {
+ pa_cvolume old_volume = s->volume;
- if (s->get_volume && s->get_volume(s) < 0)
- s->get_volume = NULL;
+ if (s->get_volume && s->get_volume(s) < 0)
+ s->get_volume = NULL;
- if (!s->get_volume && s->refresh_volume)
- pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_VOLUME, &s->volume, 0, NULL);
+ if (!s->get_volume)
+ pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_VOLUME, &s->volume, 0, NULL);
- if (!pa_cvolume_equal(&old_volume, &s->volume))
- pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
+ if (!pa_cvolume_equal(&old_volume, &s->volume))
+ pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
+ }
return &s->volume;
}
+/* Called from main thread */
void pa_source_set_mute(pa_source *s, pa_bool_t mute) {
- int changed;
+ pa_bool_t changed;
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
@@ -534,26 +558,29 @@ void pa_source_set_mute(pa_source *s, pa_bool_t mute) {
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
}
+/* Called from main thread */
pa_bool_t pa_source_get_mute(pa_source *s) {
- pa_bool_t old_muted;
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
- old_muted = s->muted;
+ if (s->refresh_muted) {
+ pa_bool_t old_muted = s->muted;
- if (s->get_mute && s->get_mute(s) < 0)
- s->get_mute = NULL;
+ if (s->get_mute && s->get_mute(s) < 0)
+ s->get_mute = NULL;
- if (!s->get_mute && s->refresh_muted)
- pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MUTE, &s->muted, 0, NULL);
+ if (!s->get_mute)
+ pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MUTE, &s->muted, 0, NULL);
- if (old_muted != s->muted)
- pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
+ if (old_muted != s->muted)
+ pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
+ }
return s->muted;
}
+/* Called from main thread */
void pa_source_set_description(pa_source *s, const char *description) {
const char *old;
pa_source_assert_ref(s);
@@ -577,6 +604,7 @@ void pa_source_set_description(pa_source *s, const char *description) {
}
}
+/* Called from main thread */
unsigned pa_source_linked_by(pa_source *s) {
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
@@ -584,6 +612,7 @@ unsigned pa_source_linked_by(pa_source *s) {
return pa_idxset_size(s->outputs);
}
+/* Called from main thread */
unsigned pa_source_used_by(pa_source *s) {
unsigned ret;
@@ -596,12 +625,13 @@ unsigned pa_source_used_by(pa_source *s) {
return ret - s->n_corked;
}
+/* Called from IO thread, except when it is not */
int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
pa_source *s = PA_SOURCE(object);
pa_source_assert_ref(s);
- pa_assert(s->thread_info.state != PA_SOURCE_UNLINKED);
switch ((pa_source_message_t) code) {
+
case PA_SOURCE_MESSAGE_ADD_OUTPUT: {
pa_source_output *o = PA_SOURCE_OUTPUT(userdata);
@@ -676,9 +706,7 @@ int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_
case PA_SOURCE_MESSAGE_DETACH:
- /* We're detaching all our output streams so that the
- * asyncmsgq and rtpoll fields can be changed without
- * problems */
+ /* Detach all streams */
pa_source_detach_within_thread(s);
return 0;
@@ -692,10 +720,45 @@ int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_
pa_usec_t *usec = userdata;
*usec = pa_source_get_requested_latency_within_thread(s);
+
+ if (*usec == (pa_usec_t) -1)
+ *usec = s->thread_info.max_latency;
+
+ return 0;
+ }
+
+ case PA_SOURCE_MESSAGE_SET_LATENCY_RANGE: {
+ pa_usec_t *r = userdata;
+
+ pa_source_update_latency_range(s, r[0], r[1]);
+
+ return 0;
+ }
+
+ case PA_SOURCE_MESSAGE_GET_LATENCY_RANGE: {
+ pa_usec_t *r = userdata;
+
+ r[0] = s->thread_info.min_latency;
+ r[1] = s->thread_info.max_latency;
+
return 0;
}
+ case PA_SOURCE_MESSAGE_GET_MAX_REWIND:
+
+ *((size_t*) userdata) = s->thread_info.max_rewind;
+ return 0;
+
case PA_SOURCE_MESSAGE_GET_LATENCY:
+
+ if (s->monitor_of) {
+ *((pa_usec_t*) userdata) = 0;
+ return 0;
+ }
+
+ /* Implementors need to overwrite this implementation! */
+ return -1;
+
case PA_SOURCE_MESSAGE_MAX:
;
}
@@ -703,6 +766,7 @@ int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_
return -1;
}
+/* Called from main thread */
int pa_source_suspend_all(pa_core *c, pa_bool_t suspend) {
uint32_t idx;
pa_source *source;
@@ -716,20 +780,23 @@ int pa_source_suspend_all(pa_core *c, pa_bool_t suspend) {
return ret;
}
+/* Called from main thread */
void pa_source_detach(pa_source *s) {
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
- pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_DETACH, NULL, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_DETACH, NULL, 0, NULL) == 0);
}
+/* Called from main thread */
void pa_source_attach(pa_source *s) {
pa_source_assert_ref(s);
pa_assert(PA_SOURCE_IS_LINKED(s->state));
- pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_ATTACH, NULL, 0, NULL);
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_ATTACH, NULL, 0, NULL) == 0);
}
+/* Called from IO thread */
void pa_source_detach_within_thread(pa_source *s) {
pa_source_output *o;
void *state = NULL;
@@ -742,6 +809,7 @@ void pa_source_detach_within_thread(pa_source *s) {
o->detach(o);
}
+/* Called from IO thread */
void pa_source_attach_within_thread(pa_source *s) {
pa_source_output *o;
void *state = NULL;
@@ -754,6 +822,7 @@ void pa_source_attach_within_thread(pa_source *s) {
o->attach(o);
}
+/* Called from IO thread */
pa_usec_t pa_source_get_requested_latency_within_thread(pa_source *s) {
pa_usec_t result = (pa_usec_t) -1;
pa_source_output *o;
@@ -771,11 +840,11 @@ pa_usec_t pa_source_get_requested_latency_within_thread(pa_source *s) {
result = o->thread_info.requested_source_latency;
if (result != (pa_usec_t) -1) {
- if (s->max_latency > 0 && result > s->max_latency)
- result = s->max_latency;
+ if (s->thread_info.max_latency > 0 && result > s->thread_info.max_latency)
+ result = s->thread_info.max_latency;
- if (s->min_latency > 0 && result < s->min_latency)
- result = s->min_latency;
+ if (s->thread_info.min_latency > 0 && result < s->thread_info.min_latency)
+ result = s->thread_info.min_latency;
}
s->thread_info.requested_latency = result;
@@ -784,6 +853,7 @@ pa_usec_t pa_source_get_requested_latency_within_thread(pa_source *s) {
return result;
}
+/* Called from main thread */
pa_usec_t pa_source_get_requested_latency(pa_source *s) {
pa_usec_t usec;
@@ -793,15 +863,12 @@ pa_usec_t pa_source_get_requested_latency(pa_source *s) {
if (!PA_SOURCE_IS_OPENED(s->state))
return 0;
- if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) < 0)
- return 0;
-
- if (usec == (pa_usec_t) -1)
- usec = s->max_latency;
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
return usec;
}
+/* Called from IO thread */
void pa_source_set_max_rewind(pa_source *s, size_t max_rewind) {
pa_source_output *o;
void *state = NULL;
@@ -813,11 +880,15 @@ void pa_source_set_max_rewind(pa_source *s, size_t max_rewind) {
s->thread_info.max_rewind = max_rewind;
- while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
- pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
+ if (PA_SOURCE_IS_LINKED(s->thread_info.state)) {
+ while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
+ pa_source_output_update_max_rewind(o, s->thread_info.max_rewind);
+ }
}
void pa_source_invalidate_requested_latency(pa_source *s) {
+ pa_source_output *o;
+ void *state = NULL;
pa_source_assert_ref(s);
@@ -826,6 +897,9 @@ void pa_source_invalidate_requested_latency(pa_source *s) {
if (s->update_requested_latency)
s->update_requested_latency(s);
+ while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
+ o->update_source_requested_latency(o);
+
if (s->monitor_of)
pa_sink_invalidate_requested_latency(s->monitor_of);
}
@@ -833,6 +907,12 @@ void pa_source_invalidate_requested_latency(pa_source *s) {
void pa_source_set_latency_range(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
pa_source_assert_ref(s);
+ /* min_latency == 0: no limit
+ * min_latency == (size_t) -1: default limit
+ * min_latency anything else: specified limit
+ *
+ * Similar for max_latency */
+
if (min_latency == (pa_usec_t) -1)
min_latency = DEFAULT_MIN_LATENCY;
@@ -842,6 +922,64 @@ void pa_source_set_latency_range(pa_source *s, pa_usec_t min_latency, pa_usec_t
pa_assert(!min_latency || !max_latency ||
min_latency <= max_latency);
- s->min_latency = min_latency;
- s->max_latency = max_latency;
+ if (PA_SINK_IS_LINKED(s->state)) {
+ pa_usec_t r[2];
+
+ r[0] = min_latency;
+ r[1] = max_latency;
+
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_LATENCY_RANGE, r, 0, NULL) == 0);
+ } else {
+ s->thread_info.min_latency = min_latency;
+ s->thread_info.max_latency = max_latency;
+
+ s->thread_info.requested_latency_valid = FALSE;
+ }
+}
+
+void pa_source_get_latency_range(pa_source *s, pa_usec_t *min_latency, pa_usec_t *max_latency) {
+ pa_source_assert_ref(s);
+ pa_assert(min_latency);
+ pa_assert(max_latency);
+
+ if (PA_SOURCE_IS_LINKED(s->state)) {
+ pa_usec_t r[2] = { 0, 0 };
+
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY_RANGE, r, 0, NULL) == 0);
+
+ *min_latency = r[0];
+ *max_latency = r[1];
+ } else {
+ *min_latency = s->thread_info.min_latency;
+ *max_latency = s->thread_info.max_latency;
+ }
+}
+
+/* Called from IO thread */
+void pa_source_update_latency_range(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency) {
+ pa_source_output *o;
+ void *state = NULL;
+
+ pa_source_assert_ref(s);
+
+ s->thread_info.min_latency = min_latency;
+ s->thread_info.max_latency = max_latency;
+
+ while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
+ if (o->update_source_latency_range)
+ o->update_source_latency_range(o);
+
+ pa_source_invalidate_requested_latency(s);
+}
+
+size_t pa_source_get_max_rewind(pa_source *s) {
+ size_t r;
+ pa_source_assert_ref(s);
+
+ if (!PA_SOURCE_IS_LINKED(s->state))
+ return s->thread_info.max_rewind;
+
+ pa_assert_se(pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MAX_REWIND, &r, 0, NULL) == 0);
+
+ return r;
}
diff --git a/src/pulsecore/source.h b/src/pulsecore/source.h
index 479222f..f4a17e8 100644
--- a/src/pulsecore/source.h
+++ b/src/pulsecore/source.h
@@ -83,22 +83,44 @@ struct pa_source {
pa_cvolume volume;
pa_bool_t muted;
- pa_bool_t refresh_volume;
- pa_bool_t refresh_muted;
+
+ pa_bool_t refresh_volume:1;
+ pa_bool_t refresh_muted:1;
pa_asyncmsgq *asyncmsgq;
pa_rtpoll *rtpoll;
pa_memchunk silence;
- pa_usec_t min_latency; /* we won't go below this latency setting */
- pa_usec_t max_latency; /* An upper limit for the latencies */
-
+ /* Called when the main loop requests a state change. Called from
+ * main loop context. If returns -1 the state change will be
+ * inhibited */
int (*set_state)(pa_source*source, pa_source_state_t state); /* may be NULL */
- int (*set_volume)(pa_source *s); /* dito */
+
+ /* Callled when the volume is queried. Called from main loop
+ * context. If this is NULL a PA_SOURCE_MESSAGE_GET_VOLUME message
+ * will be sent to the IO thread instead. If refresh_volume is
+ * FALSE neither this function is called nor a message is sent. */
int (*get_volume)(pa_source *s); /* dito */
- int (*set_mute)(pa_source *s); /* dito */
+
+ /* Called when the volume shall be changed. Called from main loop
+ * context. If this is NULL a PA_SOURCE_MESSAGE_SET_VOLUME message
+ * will be sent to the IO thread instead. */
+ int (*set_volume)(pa_source *s); /* dito */
+
+ /* Called when the mute setting is queried. Called from main loop
+ * context. If this is NULL a PA_SOURCE_MESSAGE_GET_MUTE message
+ * will be sent to the IO thread instead. If refresh_mute is
+ * FALSE neither this function is called nor a message is sent.*/
int (*get_mute)(pa_source *s); /* dito */
+
+ /* Called when the mute setting shall be changed. Called from main
+ * loop context. If this is NULL a PA_SOURCE_MESSAGE_SET_MUTE
+ * message will be sent to the IO thread instead. */
+ int (*set_mute)(pa_source *s); /* dito */
+
+ /* Called when a the requested latency is changed. Called from IO
+ * thread context. */
void (*update_requested_latency)(pa_source *s); /* dito */
/* Contains copies of the above data so that the real-time worker
@@ -107,14 +129,17 @@ struct pa_source {
pa_source_state_t state;
pa_hashmap *outputs;
pa_cvolume soft_volume;
- pa_bool_t soft_muted;
+ pa_bool_t soft_muted:1;
- pa_bool_t requested_latency_valid;
+ pa_bool_t requested_latency_valid:1;
pa_usec_t requested_latency;
/* Then number of bytes this source will be rewound for at
- * max */
+ * max. (Only used on monitor sources) */
size_t max_rewind;
+
+ pa_usec_t min_latency; /* we won't go below this latency */
+ pa_usec_t max_latency; /* An upper limit for the latencies */
} thread_info;
void *userdata;
@@ -135,6 +160,9 @@ typedef enum pa_source_message {
PA_SOURCE_MESSAGE_SET_STATE,
PA_SOURCE_MESSAGE_ATTACH,
PA_SOURCE_MESSAGE_DETACH,
+ PA_SOURCE_MESSAGE_SET_LATENCY_RANGE,
+ PA_SOURCE_MESSAGE_GET_LATENCY_RANGE,
+ PA_SOURCE_MESSAGE_GET_MAX_REWIND,
PA_SOURCE_MESSAGE_MAX
} pa_source_message_t;
@@ -186,8 +214,12 @@ void pa_source_attach(pa_source *s);
/* May be called by everyone, from main context */
+/* The returned value is supposed to be in the time domain of the sound card! */
pa_usec_t pa_source_get_latency(pa_source *s);
pa_usec_t pa_source_get_requested_latency(pa_source *s);
+void pa_source_get_latency_range(pa_source *s, pa_usec_t *min_latency, pa_usec_t *max_latency);
+
+size_t pa_source_get_max_rewind(pa_source *s);
int pa_source_update_status(pa_source*s);
int pa_source_suspend(pa_source *s, pa_bool_t suspend);
@@ -216,6 +248,7 @@ void pa_source_detach_within_thread(pa_source *s);
pa_usec_t pa_source_get_requested_latency_within_thread(pa_source *s);
void pa_source_set_max_rewind(pa_source *s, size_t max_rewind);
+void pa_source_update_latency_range(pa_source *s, pa_usec_t min_latency, pa_usec_t max_latency);
/* To be called exclusively by source output drivers, from IO context */
--
hooks/post-receive
PulseAudio Sound Server
More information about the pulseaudio-commits
mailing list