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

thread-mq: Support a mainloop as thread/consumer

Browse source 
This is needed by the tunnel module rewrite, which runs pa_mainloop in
the IO thread instead of pa_rtpoll.

Signed-off-by: Alexander Couzens <lynxis@fe80.eu>
This commit is contained in:
Alexander Couzens 2013-08-06 05:29:45 +02:00 • committed by Tanu Kaskinen
parent 7682cab60e
commit 9fa3466366
2 changed files with 68 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

70
src/pulsecore/thread-mq.c
View file

@ -30,6 +30,8 @@
#include <pulsecore/semaphore.h>
#include <pulsecore/macro.h>
#include <pulse/mainloop-api.h>
#include "thread-mq.h"
PA_STATIC_TLS_DECLARE_NO_FREE(thread_mq);
@ -38,10 +40,15 @@ static void asyncmsgq_read_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io
pa_thread_mq *q = userdata;
pa_asyncmsgq *aq;
pa_assert(pa_asyncmsgq_read_fd(q->outq) == fd);
pa_assert(events == PA_IO_EVENT_INPUT);
if (pa_asyncmsgq_read_fd(q->outq) == fd)
pa_asyncmsgq_ref(aq = q->outq);
else if (pa_asyncmsgq_read_fd(q->inq) == fd)
pa_asyncmsgq_ref(aq = q->inq);
else
pa_assert_not_reached();
pa_asyncmsgq_read_after_poll(aq);
for (;;) {
@ -55,6 +62,12 @@ static void asyncmsgq_read_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io
while (pa_asyncmsgq_get(aq, &object, &code, &data, &offset, &chunk, 0) >= 0) {
int ret;
if (!object && code == PA_MESSAGE_SHUTDOWN) {
pa_asyncmsgq_done(aq, 0);
api->quit(api, 0);
break;
}
ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
pa_asyncmsgq_done(aq, ret);
}
@ -66,7 +79,7 @@ static void asyncmsgq_read_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io
pa_asyncmsgq_unref(aq);
}
static void asyncmsgq_write_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io_event_flags_t events, void *userdata) {
static void asyncmsgq_write_inq_cb(pa_mainloop_api *api, pa_io_event *e, int fd, pa_io_event_flags_t events, void *userdata) {
pa_thread_mq *q = userdata;
pa_assert(pa_asyncmsgq_write_fd(q->inq) == fd);
@ -76,19 +89,51 @@ static void asyncmsgq_write_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_i
pa_asyncmsgq_write_before_poll(q->inq);
}
static void asyncmsgq_write_outq_cb(pa_mainloop_api *api, pa_io_event *e, int fd, pa_io_event_flags_t events, void *userdata) {
pa_thread_mq *q = userdata;
pa_assert(pa_asyncmsgq_write_fd(q->outq) == fd);
pa_assert(events == PA_IO_EVENT_INPUT);
pa_asyncmsgq_write_after_poll(q->outq);
pa_asyncmsgq_write_before_poll(q->outq);
}
void pa_thread_mq_init_thread_mainloop(pa_thread_mq *q, pa_mainloop_api *main_mainloop, pa_mainloop_api *thread_mainloop) {
pa_assert(q);
pa_assert(main_mainloop);
pa_assert(thread_mainloop);
pa_assert_se(q->inq = pa_asyncmsgq_new(0));
pa_assert_se(q->outq = pa_asyncmsgq_new(0));
q->main_mainloop = main_mainloop;
q->thread_mainloop = thread_mainloop;
pa_assert_se(pa_asyncmsgq_read_before_poll(q->outq) == 0);
pa_asyncmsgq_write_before_poll(q->outq);
pa_assert_se(q->read_main_event = main_mainloop->io_new(main_mainloop, pa_asyncmsgq_read_fd(q->outq), PA_IO_EVENT_INPUT, asyncmsgq_read_cb, q));
pa_assert_se(q->write_thread_event = main_mainloop->io_new(thread_mainloop, pa_asyncmsgq_write_fd(q->outq), PA_IO_EVENT_INPUT, asyncmsgq_write_outq_cb, q));
pa_asyncmsgq_read_before_poll(q->inq);
pa_asyncmsgq_write_before_poll(q->inq);
pa_assert_se(q->read_thread_event = thread_mainloop->io_new(thread_mainloop, pa_asyncmsgq_read_fd(q->inq), PA_IO_EVENT_INPUT, asyncmsgq_read_cb, q));
pa_assert_se(q->write_main_event = main_mainloop->io_new(main_mainloop, pa_asyncmsgq_write_fd(q->inq), PA_IO_EVENT_INPUT, asyncmsgq_write_inq_cb, q));
}
void pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rtpoll) {
pa_assert(q);
pa_assert(mainloop);
q->mainloop = mainloop;
q->main_mainloop = mainloop;
pa_assert_se(q->inq = pa_asyncmsgq_new(0));
pa_assert_se(q->outq = pa_asyncmsgq_new(0));
pa_assert_se(pa_asyncmsgq_read_before_poll(q->outq) == 0);
pa_assert_se(q->read_event = mainloop->io_new(mainloop, pa_asyncmsgq_read_fd(q->outq), PA_IO_EVENT_INPUT, asyncmsgq_read_cb, q));
pa_assert_se(q->read_main_event = mainloop->io_new(mainloop, pa_asyncmsgq_read_fd(q->outq), PA_IO_EVENT_INPUT, asyncmsgq_read_cb, q));
pa_asyncmsgq_write_before_poll(q->inq);
pa_assert_se(q->write_event = mainloop->io_new(mainloop, pa_asyncmsgq_write_fd(q->inq), PA_IO_EVENT_INPUT, asyncmsgq_write_cb, q));
pa_assert_se(q->write_main_event = mainloop->io_new(mainloop, pa_asyncmsgq_write_fd(q->inq), PA_IO_EVENT_INPUT, asyncmsgq_write_inq_cb, q));
pa_rtpoll_item_new_asyncmsgq_read(rtpoll, PA_RTPOLL_EARLY, q->inq);
pa_rtpoll_item_new_asyncmsgq_write(rtpoll, PA_RTPOLL_LATE, q->outq);
@ -106,15 +151,22 @@ void pa_thread_mq_done(pa_thread_mq *q) {
if (!pa_asyncmsgq_dispatching(q->outq))
pa_asyncmsgq_flush(q->outq, true);
q->mainloop->io_free(q->read_event);
q->mainloop->io_free(q->write_event);
q->read_event = q->write_event = NULL;
q->main_mainloop->io_free(q->read_main_event);
q->main_mainloop->io_free(q->write_main_event);
q->read_main_event = q->write_main_event = NULL;
if (q->thread_mainloop) {
q->thread_mainloop->io_free(q->read_thread_event);
q->thread_mainloop->io_free(q->write_thread_event);
q->read_thread_event = q->write_thread_event = NULL;
}
pa_asyncmsgq_unref(q->inq);
pa_asyncmsgq_unref(q->outq);
q->inq = q->outq = NULL;
q->mainloop = NULL;
q->main_mainloop = NULL;
q->thread_mainloop = NULL;
}
void pa_thread_mq_install(pa_thread_mq *q) {

7
src/pulsecore/thread-mq.h
View file

@ -31,12 +31,15 @@
* attached to the thread using pa_thread_mq_install(). */
typedef struct pa_thread_mq {
pa_mainloop_api *mainloop;
pa_mainloop_api *main_mainloop;
pa_mainloop_api *thread_mainloop;
pa_asyncmsgq *inq, *outq;
pa_io_event *read_event, *write_event;
pa_io_event *read_main_event, *write_main_event;
pa_io_event *read_thread_event, *write_thread_event;
} pa_thread_mq;
void pa_thread_mq_init(pa_thread_mq *q, pa_mainloop_api *mainloop, pa_rtpoll *rtpoll);
void pa_thread_mq_init_thread_mainloop(pa_thread_mq *q, pa_mainloop_api *main_mainloop, pa_mainloop_api *thread_mainloop);
void pa_thread_mq_done(pa_thread_mq *q);
/* Install the specified pa_thread_mq object for the current thread */