mirrors/pulseaudio
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pulseaudio/pulseaudio.git synced 2025-11-10 13:29:58 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

rework module-combine again

Browse source 
git-svn-id: file:///home/lennart/svn/public/pulseaudio/branches/lennart@1873 fefdeb5f-60dc-0310-8127-8f9354f1896f
This commit is contained in:
Lennart Poettering 2007-09-20 20:33:34 +00:00
parent c40c1682be
commit f3f44dab37
1 changed files with 245 additions and 187 deletions
Download patch file Download diff file Expand all files Collapse all files

222
src/modules/module-combine.c
View file

@ -134,7 +134,8 @@ struct userdata {
enum { enum {
SINK_MESSAGE_ADD_OUTPUT = PA_SINK_MESSAGE_MAX, SINK_MESSAGE_ADD_OUTPUT = PA_SINK_MESSAGE_MAX,
SINK_MESSAGE_REMOVE_OUTPUT, SINK_MESSAGE_REMOVE_OUTPUT,
SINK_MESSAGE_NEED SINK_MESSAGE_NEED,
SINK_MESSAGE_SET_MASTER
}; };
enum { enum {
@ -144,7 +145,7 @@ enum {
static void output_free(struct output *o); static void output_free(struct output *o);
static int output_create_sink_input(struct userdata *u, struct output *o); static int output_create_sink_input(struct userdata *u, struct output *o);
static int update_master(struct userdata *u, struct output *o); static int update_master(struct userdata *u, struct output *o);
static int pick_master(struct userdata *u); static int pick_master(struct userdata *u, struct output *except);
static void adjust_rates(struct userdata *u) { static void adjust_rates(struct userdata *u) {
struct output *o; struct output *o;
@ -275,6 +276,7 @@ finish:
pa_log_debug("Thread shutting down"); pa_log_debug("Thread shutting down");
} }
/* Called from I/O thread context */
static void render_memblock(struct userdata *u, struct output *o, size_t length) { static void render_memblock(struct userdata *u, struct output *o, size_t length) {
pa_assert(u); pa_assert(u);
pa_assert(o); pa_assert(o);
@ -305,20 +307,21 @@ static void render_memblock(struct userdata *u, struct output *o, size_t length)
/* Send to other outputs, which are not the requesting /* Send to other outputs, which are not the requesting
* one, and not the master */ * one, and not the master */
if (j != o && j != u->master && j->sink_input) if (j != o && j != u->thread_info.master && j->sink_input)
pa_asyncmsgq_post(j->inq, PA_MSGOBJECT(j->sink_input), SINK_INPUT_MESSAGE_POST, NULL, 0, &chunk, NULL); pa_asyncmsgq_post(j->inq, PA_MSGOBJECT(j->sink_input), SINK_INPUT_MESSAGE_POST, NULL, 0, &chunk, NULL);
/* Now push it into the master queue */ /* Now push it into the master queue */
pa_memblockq_push_align(u->master->memblockq, &chunk); pa_memblockq_push_align(u->thread_info.master->memblockq, &chunk);
/* And into the requesting output's queue */ /* And into the requesting output's queue */
if (o != u->master) if (o != u->thread_info.master)
pa_memblockq_push_align(o->memblockq, &chunk); pa_memblockq_push_align(o->memblockq, &chunk);
pa_memblock_unref(chunk.memblock); pa_memblock_unref(chunk.memblock);
} }
} }
/* Called from I/O thread context */
static void request_memblock(struct output *o, size_t length) { static void request_memblock(struct output *o, size_t length) {
pa_assert(o); pa_assert(o);
pa_sink_input_assert_ref(o->sink_input); pa_sink_input_assert_ref(o->sink_input);
@ -336,7 +339,7 @@ static void request_memblock(struct output *o, size_t length) {
/* OK, we need to prepare new data */ /* OK, we need to prepare new data */
if (o == o->userdata->master) if (o == o->userdata->thread_info.master)
/* OK, we're the master, so let's render some data */ /* OK, we're the master, so let's render some data */
render_memblock(o->userdata, o, length); render_memblock(o->userdata, o, length);
@ -371,6 +374,51 @@ static void sink_input_drop_cb(pa_sink_input *i, size_t length) {
pa_memblockq_drop(o->memblockq, length); pa_memblockq_drop(o->memblockq, length);
} }
/* Called from I/O thread context for the master */
static void create_master_rtpolls(struct userdata *u) {
struct output *k;
pa_assert(u);
pa_assert(!u->master->outq_rtpoll_item);
/* Set up the queues from the outputs to the master */
for (k = u->thread_info.outputs; k; k = k->next) {
pa_assert(!k->outq_rtpoll_item);
if (k == u->master)
continue;
k->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
u->master->sink->rtpoll,
PA_RTPOLL_EARLY+1, /* This one has a slightly lower priority than the normal message handling */
k->outq);
pa_log("1: %p now has rptoll item %p", k, k->outq_rtpoll_item);
}
}
/* Called from I/O thread context for the master */
static void free_master_rtpolls(struct userdata *u) {
struct output *k;
pa_assert(!u->master->outq_rtpoll_item);
for (k = u->thread_info.outputs; k; k = k->next) {
if (k == u->master)
continue;
if (k->outq_rtpoll_item) {
pa_rtpoll_item_free(k->outq_rtpoll_item);
k->outq_rtpoll_item = NULL;
}
pa_assert(!k->outq_rtpoll_item);
}
}
/* Called from I/O thread context */ /* Called from I/O thread context */
static void sink_input_attach_cb(pa_sink_input *i) { static void sink_input_attach_cb(pa_sink_input *i) {
struct output *o; struct output *o;
@ -378,26 +426,10 @@ static void sink_input_attach_cb(pa_sink_input *i) {
pa_sink_input_assert_ref(i); pa_sink_input_assert_ref(i);
pa_assert_se(o = i->userdata); pa_assert_se(o = i->userdata);
pa_assert(!o->inq_rtpoll_item); pa_log("attaching %s", i->sink->name);
if (o->userdata->master == o) { if (o->userdata->thread_info.master == o) {
struct output *k; create_master_rtpolls(o->userdata);
pa_assert(!o->outq_rtpoll_item);
/* Set up the queues from the outputs to the master */
for (k = o->userdata->thread_info.outputs; k; k = k->next) {
pa_assert(!k->outq_rtpoll_item);
if (o == k)
continue;
k->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
i->sink->rtpoll,
PA_RTPOLL_EARLY+1, /* This one has a slightly lower priority than the normal message handling */
k->outq);
}
/* Calling these two functions here is safe, because both /* Calling these two functions here is safe, because both
* threads that might access this sink are known to be * threads that might access this sink are known to be
@ -408,6 +440,7 @@ static void sink_input_attach_cb(pa_sink_input *i) {
} }
/* Set up the queues from the inputs to the master */ /* Set up the queues from the inputs to the master */
pa_assert(!o->inq_rtpoll_item);
o->inq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq( o->inq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
i->sink->rtpoll, i->sink->rtpoll,
PA_RTPOLL_NORMAL, /* This one has a lower priority than the normal message handling */ PA_RTPOLL_NORMAL, /* This one has a lower priority than the normal message handling */
@ -421,24 +454,15 @@ static void sink_input_detach_cb(pa_sink_input *i) {
pa_sink_input_assert_ref(i); pa_sink_input_assert_ref(i);
pa_assert_se(o = i->userdata); pa_assert_se(o = i->userdata);
pa_log("detaching %s", i->sink->name);
pa_assert(o->inq_rtpoll_item); pa_assert(o->inq_rtpoll_item);
pa_rtpoll_item_free(o->inq_rtpoll_item); pa_rtpoll_item_free(o->inq_rtpoll_item);
o->inq_rtpoll_item = NULL; o->inq_rtpoll_item = NULL;
if (o->userdata->master == o) { if (o->userdata->thread_info.master == o) {
struct output *k;
pa_sink_detach_within_thread(o->userdata->sink); pa_sink_detach_within_thread(o->userdata->sink);
free_master_rtpolls(o->userdata);
for (k = o->userdata->thread_info.outputs; k; k = k->next) {
if (o == k)
continue;
pa_assert(k->outq_rtpoll_item);
pa_rtpoll_item_free(k->outq_rtpoll_item);
k->outq_rtpoll_item = NULL;
}
} }
} }
@ -505,7 +529,7 @@ static int suspend(struct userdata *u) {
} }
} }
if (pick_master(u) < 0) if (pick_master(u, NULL) < 0)
pa_module_unload_request(u->module); pa_module_unload_request(u->module);
pa_log_info("Device suspended..."); pa_log_info("Device suspended...");
@ -529,14 +553,16 @@ static int unsuspend(struct userdata *u) {
if (PA_SINK_OPENED(pa_sink_get_state(o->sink))) { if (PA_SINK_OPENED(pa_sink_get_state(o->sink))) {
if (output_create_sink_input(u, o) < 0) if (output_create_sink_input(u, o) < 0)
output_free(o); output_free(o);
else
pa_sink_input_put(o->sink_input);
} }
} }
if (pick_master(u) < 0) if (pick_master(u, NULL) < 0)
pa_module_unload_request(u->module); pa_module_unload_request(u->module);
for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
if (o->sink_input && pa_sink_get_state(o->sink_input) == PA_SINK_INPUT_INIT)
pa_sink_input_put(o->sink_input);
pa_log_info("Resumed successfully..."); pa_log_info("Resumed successfully...");
return 0; return 0;
} }
@ -609,32 +635,67 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
break; break;
} }
case PA_SINK_MESSAGE_DETACH: case SINK_MESSAGE_SET_MASTER:
if ((u->thread_info.master = data)) {
/* We're detaching all our input streams artificially, so /* There's now a master, and we're being executed in
* that we can drive our sink from a different sink */ * its thread, let's register the asyncmsgqs from other
* outputs to us */
u->thread_info.master = NULL; if (u->thread_info.master->sink_input->thread_info.attached)
break; create_master_rtpolls(u);
case PA_SINK_MESSAGE_ATTACH: } else {
/* We're attached all our input streams artificially again */ if (u->thread_info.master->sink_input->thread_info.attached)
free_master_rtpolls(u);
u->thread_info.master = data; }
break; return 0;
case SINK_MESSAGE_ADD_OUTPUT: case SINK_MESSAGE_ADD_OUTPUT: {
PA_LLIST_PREPEND(struct output, u->thread_info.outputs, (struct output*) data); struct output *op = data;
break;
case SINK_MESSAGE_REMOVE_OUTPUT: PA_LLIST_PREPEND(struct output, u->thread_info.outputs, op);
PA_LLIST_REMOVE(struct output, u->thread_info.outputs, (struct output*) data);
break; pa_assert(!op->outq_rtpoll_item);
if (op != u->thread_info.master) {
/* Create pa_asyncmsgq to master */
op->outq_rtpoll_item = pa_rtpoll_item_new_asyncmsgq(
u->thread_info.master->sink->rtpoll,
PA_RTPOLL_EARLY+1, /* This one has a slightly lower priority than the normal message handling */
op->outq);
pa_log("2: %p now has rptoll item %p", op, op->outq_rtpoll_item);
}
pa_log("Added output %s", op->sink_input->sink->name);
return 0;
}
case SINK_MESSAGE_REMOVE_OUTPUT: {
struct output *op = data;
pa_log("Remove output %s", op->sink_input->sink->name);
PA_LLIST_REMOVE(struct output, u->thread_info.outputs, op);
/* Remove the q that leads from this output to the master output */
if (op->outq_rtpoll_item) {
pa_rtpoll_item_free(op->outq_rtpoll_item);
op->outq_rtpoll_item = NULL;
}
return 0;
}
case SINK_MESSAGE_NEED: case SINK_MESSAGE_NEED:
render_memblock(u, data, (size_t) offset); render_memblock(u, data, (size_t) offset);
break; return 0;
} }
return pa_sink_process_msg(o, code, data, offset, chunk); return pa_sink_process_msg(o, code, data, offset, chunk);
@ -708,8 +769,11 @@ static int update_master(struct userdata *u, struct output *o) {
pa_assert(u); pa_assert(u);
/* Make sure everything is detached from the old thread before we move our stuff to a new thread */ /* Make sure everything is detached from the old thread before we move our stuff to a new thread */
if (u->sink && PA_SINK_LINKED(pa_sink_get_state(u->sink))) if (u->sink && PA_SINK_LINKED(pa_sink_get_state(u->sink))) {
pa_sink_detach(u->sink); pa_sink_detach(u->sink);
pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_SET_MASTER, NULL, 0, NULL);
} else
u->thread_info.master = NULL;
if (o) { if (o) {
/* If we have a master sink we run our own sink in its thread */ /* If we have a master sink we run our own sink in its thread */
@ -759,22 +823,25 @@ static int update_master(struct userdata *u, struct output *o) {
} }
/* Now attach everything again */ /* Now attach everything again */
if (u->sink && PA_SINK_LINKED(pa_sink_get_state(u->sink))) if (u->sink && PA_SINK_LINKED(pa_sink_get_state(u->sink))) {
pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_SET_MASTER, u->master, 0, NULL);
pa_sink_attach(u->sink); pa_sink_attach(u->sink);
} else
u->thread_info.master = u->master;
return 0; return 0;
} }
static int pick_master(struct userdata *u) { static int pick_master(struct userdata *u, struct output *except) {
struct output *o; struct output *o;
uint32_t idx; uint32_t idx;
pa_assert(u); pa_assert(u);
if (u->master && u->master->sink_input && PA_SINK_OPENED(pa_sink_get_state(u->master->sink))) if (u->master && u->master != except && u->master->sink_input && PA_SINK_OPENED(pa_sink_get_state(u->master->sink)))
return update_master(u, u->master); return update_master(u, u->master);
for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx)) for (o = pa_idxset_first(u->outputs, &idx); o; o = pa_idxset_next(u->outputs, &idx))
if (o->sink_input && PA_SINK_OPENED(pa_sink_get_state(o->sink))) if (o != except && o->sink_input && PA_SINK_OPENED(pa_sink_get_state(o->sink)))
return update_master(u, o); return update_master(u, o);
return update_master(u, NULL); return update_master(u, NULL);
@ -901,7 +968,7 @@ static pa_hook_result_t sink_new_hook_cb(pa_core *c, pa_sink *s, struct userdata
return PA_HOOK_OK; return PA_HOOK_OK;
} }
if (pick_master(u) < 0) if (pick_master(u, NULL) < 0)
pa_module_unload_request(u->module); pa_module_unload_request(u->module);
if (o->sink_input) if (o->sink_input)
@ -932,9 +999,6 @@ static pa_hook_result_t sink_unlink_hook_cb(pa_core *c, pa_sink *s, struct userd
output_free(o); output_free(o);
if (pick_master(u) < 0)
pa_module_unload_request(u->module);
return PA_HOOK_OK; return PA_HOOK_OK;
} }
@ -958,7 +1022,7 @@ static pa_hook_result_t sink_state_changed_hook_cb(pa_core *c, pa_sink *s, struc
if (PA_SINK_OPENED(state) && PA_SINK_OPENED(pa_sink_get_state(u->sink)) && !o->sink_input) { if (PA_SINK_OPENED(state) && PA_SINK_OPENED(pa_sink_get_state(u->sink)) && !o->sink_input) {
output_create_sink_input(u, o); output_create_sink_input(u, o);
if (pick_master(u) < 0) if (pick_master(u, NULL) < 0)
pa_module_unload_request(u->module); pa_module_unload_request(u->module);
if (o->sink_input) if (o->sink_input)
@ -972,7 +1036,7 @@ static pa_hook_result_t sink_state_changed_hook_cb(pa_core *c, pa_sink *s, struc
pa_memblockq_flush(o->memblockq); pa_memblockq_flush(o->memblockq);
if (pick_master(u) < 0) if (pick_master(u, o) < 0)
pa_module_unload_request(u->module); pa_module_unload_request(u->module);
} }
@ -1142,7 +1206,7 @@ int pa__init(pa_module*m) {
u->sink_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_UNLINK], (pa_hook_cb_t) sink_unlink_hook_cb, u); u->sink_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_UNLINK], (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_cb_t) sink_state_changed_hook_cb, u); u->sink_state_changed_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_STATE_CHANGED], (pa_hook_cb_t) sink_state_changed_hook_cb, u);
if (pick_master(u) < 0) if (pick_master(u, NULL) < 0)
goto fail; goto fail;
/* Activate the sink and the sink inputs */ /* Activate the sink and the sink inputs */
@ -1176,28 +1240,24 @@ fail:
static void output_free(struct output *o) { static void output_free(struct output *o) {
pa_assert(o); pa_assert(o);
if (o->userdata) { /* Make sure the master points to a different output */
if (pick_master(o->userdata, o) < 0)
pa_module_unload_request(o->userdata->module);
if (o->userdata->sink && PA_SINK_LINKED(pa_sink_get_state(o->userdata->sink))) if (o->userdata->sink && PA_SINK_LINKED(pa_sink_get_state(o->userdata->sink)))
pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_REMOVE_OUTPUT, o, 0, NULL); pa_asyncmsgq_send(o->userdata->sink->asyncmsgq, PA_MSGOBJECT(o->userdata->sink), SINK_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
else else
PA_LLIST_REMOVE(struct output, o->userdata->thread_info.outputs, o); PA_LLIST_REMOVE(struct output, o->userdata->thread_info.outputs, o);
}
pa_assert_se(pa_idxset_remove_by_data(o->userdata->outputs, o, NULL)); pa_assert_se(pa_idxset_remove_by_data(o->userdata->outputs, o, NULL));
if (o->userdata->master == o) {
/* Make sure the master points to a different output */
o->userdata->master = NULL;
pick_master(o->userdata);
}
update_description(o->userdata);
if (o->sink_input) { if (o->sink_input) {
pa_sink_input_unlink(o->sink_input); pa_sink_input_unlink(o->sink_input);
pa_sink_input_unref(o->sink_input); pa_sink_input_unref(o->sink_input);
} }
update_description(o->userdata);
if (o->inq_rtpoll_item) if (o->inq_rtpoll_item)
pa_rtpoll_item_free(o->inq_rtpoll_item); pa_rtpoll_item_free(o->inq_rtpoll_item);
@ -1262,5 +1322,3 @@ void pa__done(pa_module*m) {
pa_xfree(u); pa_xfree(u);
} }