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

extend rtpoll API to allow registration of arbitray functions to be executed in the event loop. Add priority system for specifying the order of these functions.

Browse source 
git-svn-id: file:///home/lennart/svn/public/pulseaudio/branches/lennart@1816 fefdeb5f-60dc-0310-8127-8f9354f1896f
This commit is contained in:
Lennart Poettering 2007-09-12 22:57:29 +00:00
parent cf3e9da9dd
commit ef83a19525
11 changed files with 139 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/modules/module-alsa-sink.c
View file

@ -322,7 +322,7 @@ static int build_pollfd(struct userdata *u) {
if (u->alsa_rtpoll_item)
pa_rtpoll_item_free(u->alsa_rtpoll_item);
u->alsa_rtpoll_item = pa_rtpoll_item_new(u->rtpoll, n);
u->alsa_rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, n);
pollfd = pa_rtpoll_item_get_pollfd(u->alsa_rtpoll_item, NULL);
if ((err = snd_pcm_poll_descriptors(u->pcm_handle, pollfd, n)) < 0) {
@ -745,7 +745,7 @@ int pa__init(pa_module*m) {
pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
u->rtpoll = pa_rtpoll_new();
u->alsa_rtpoll_item = NULL;
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, u->thread_mq.inq);
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
snd_config_update_free_global();
if ((err = snd_pcm_open(&u->pcm_handle, dev = pa_modargs_get_value(ma, "device", DEFAULT_DEVICE), SND_PCM_STREAM_PLAYBACK, SND_PCM_NONBLOCK)) < 0) {

4
src/modules/module-alsa-source.c
View file

@ -310,7 +310,7 @@ static int build_pollfd(struct userdata *u) {
if (u->alsa_rtpoll_item)
pa_rtpoll_item_free(u->alsa_rtpoll_item);
u->alsa_rtpoll_item = pa_rtpoll_item_new(u->rtpoll, n);
u->alsa_rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, n);
pollfd = pa_rtpoll_item_get_pollfd(u->alsa_rtpoll_item, NULL);
if ((err = snd_pcm_poll_descriptors(u->pcm_handle, pollfd, n)) < 0) {
@ -726,7 +726,7 @@ int pa__init(pa_module*m) {
pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
u->rtpoll = pa_rtpoll_new();
u->alsa_rtpoll_item = NULL;
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, u->thread_mq.inq);
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
snd_config_update_free_global();
if ((err = snd_pcm_open(&u->pcm_handle, dev = pa_modargs_get_value(ma, "device", DEFAULT_DEVICE), SND_PCM_STREAM_CAPTURE, SND_PCM_NONBLOCK)) < 0) {

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

@ -401,7 +401,7 @@ static void sink_input_attach_cb(pa_sink_input *i) {
pa_assert(o);
pa_assert(!o->rtpoll_item);
o->rtpoll_item = pa_rtpoll_item_new_asyncmsgq(i->sink->rtpoll, o->asyncmsgq);
o->rtpoll_item = pa_rtpoll_item_new_asyncmsgq(i->sink->rtpoll, PA_RTPOLL_NORMAL, o->asyncmsgq);
}
/* Called from I/O thread context */
@ -721,7 +721,7 @@ static int update_master(struct userdata *u, struct output *o) {
pa_assert(!u->rtpoll);
u->rtpoll = pa_rtpoll_new();
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, u->thread_mq.inq);
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
pa_sink_set_rtpoll(u->sink, u->rtpoll);

2
src/modules/module-null-sink.c
View file

@ -199,7 +199,7 @@ int pa__init(pa_module*m) {
m->userdata = u;
pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
u->rtpoll = pa_rtpoll_new();
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, u->thread_mq.inq);
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
if (!(u->sink = pa_sink_new(m->core, __FILE__, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME), 0, &ss, &map))) {
pa_log("Failed to create sink.");

6
src/modules/module-oss.c
View file

@ -577,7 +577,7 @@ static int unsuspend(struct userdata *u) {
pa_assert(!u->rtpoll_item);
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, 1);
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
pollfd->fd = u->fd;
pollfd->events = 0;
@ -1167,8 +1167,8 @@ int pa__init(pa_module*m) {
u->use_mmap = use_mmap;
pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
u->rtpoll = pa_rtpoll_new();
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, u->thread_mq.inq);
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, 1);
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
pollfd->fd = fd;
pollfd->events = 0;

4
src/modules/module-pipe-sink.c
View file

@ -236,7 +236,7 @@ int pa__init(pa_module*m) {
pa_memchunk_reset(&u->memchunk);
pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
u->rtpoll = pa_rtpoll_new();
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, u->thread_mq.inq);
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
u->filename = pa_xstrdup(pa_modargs_get_value(ma, "file", DEFAULT_FILE_NAME));
@ -274,7 +274,7 @@ int pa__init(pa_module*m) {
pa_sink_set_description(u->sink, t = pa_sprintf_malloc("Unix FIFO sink '%s'", u->filename));
pa_xfree(t);
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, 1);
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
pollfd->fd = u->fd;
pollfd->events = pollfd->revents = 0;

4
src/modules/module-pipe-source.c
View file

@ -215,7 +215,7 @@ int pa__init(pa_module*m) {
pa_memchunk_reset(&u->memchunk);
pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
u->rtpoll = pa_rtpoll_new();
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, u->thread_mq.inq);
pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
u->filename = pa_xstrdup(pa_modargs_get_value(ma, "file", DEFAULT_FILE_NAME));
@ -252,7 +252,7 @@ int pa__init(pa_module*m) {
pa_source_set_description(u->source, t = pa_sprintf_malloc("Unix FIFO source '%s'", u->filename));
pa_xfree(t);
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, 1);
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
pollfd->fd = u->fd;
pollfd->events = pollfd->revents = 0;

122
src/pulsecore/rtpoll.c
View file

@ -45,7 +45,6 @@
#include "rtpoll.h"
struct pa_rtpoll {
struct pollfd *pollfd, *pollfd2;
unsigned n_pollfd_alloc, n_pollfd_used;
@ -72,6 +71,8 @@ struct pa_rtpoll_item {
pa_rtpoll *rtpoll;
int dead;
pa_rtpoll_priority_t priority;
struct pollfd *pollfd;
unsigned n_pollfd;
@ -257,11 +258,37 @@ void pa_rtpoll_free(pa_rtpoll *p) {
pa_xfree(p);
}
static void reset_revents(pa_rtpoll_item *i) {
struct pollfd *f;
unsigned n;
pa_assert(i);
if (!(f = pa_rtpoll_item_get_pollfd(i, &n)))
return;
for (; n > 0; n--)
f[n-1].revents = 0;
}
static void reset_all_revents(pa_rtpoll *p) {
pa_rtpoll_item *i;
pa_assert(p);
for (i = p->items; i; i = i->next) {
if (i->dead)
continue;
reset_revents(i);
}
}
int pa_rtpoll_run(pa_rtpoll *p, int wait) {
pa_rtpoll_item *i;
int r = 0;
int no_events = 0;
int saved_errno;
int saved_errno = 0;
struct timespec timeout;
pa_assert(p);
@ -270,7 +297,8 @@ int pa_rtpoll_run(pa_rtpoll *p, int wait) {
p->running = 1;
for (i = p->items; i; i = i->next) {
for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
int k;
if (i->dead)
continue;
@ -278,10 +306,12 @@ int pa_rtpoll_run(pa_rtpoll *p, int wait) {
if (!i->before_cb)
continue;
if (i->before_cb(i) < 0) {
if ((k = i->before_cb(i)) != 0) {
/* Hmm, this one doesn't let us enter the poll, so rewind everything */
reset_all_revents(p);
for (i = i->prev; i; i = i->prev) {
if (i->dead)
@ -293,6 +323,9 @@ int pa_rtpoll_run(pa_rtpoll *p, int wait) {
i->after_cb(i);
}
if (k < 0)
r = k;
goto finish;
}
}
@ -329,7 +362,13 @@ int pa_rtpoll_run(pa_rtpoll *p, int wait) {
r = poll(p->pollfd, p->n_pollfd_used, p->timer_enabled > 0 ? (timeout.tv_sec*1000) + (timeout.tv_nsec / 1000000) : -1);
#endif
saved_errno = errno;
if (r < 0)
reset_all_revents(p);
if (r < 0 && (errno == EAGAIN || errno == EINTR))
r = 0;
saved_errno = r < 0 ? errno : 0;
if (p->timer_enabled) {
if (p->period > 0) {
@ -340,18 +379,13 @@ int pa_rtpoll_run(pa_rtpoll *p, int wait) {
/* Guarantee that the next timeout will happen in the future */
if (pa_timespec_cmp(&p->next_elapse, &now) < 0)
pa_timespec_add(&p->next_elapse, (pa_timespec_diff(&now, &p->next_elapse) / p->period + 1) * p->period);
pa_timespec_add(&p->next_elapse, (pa_timespec_diff(&now, &p->next_elapse) / p->period + 1) * p->period);
} else
p->timer_enabled = 0;
}
if (r == 0 || (r < 0 && (errno == EAGAIN || errno == EINTR))) {
r = 0;
no_events = 1;
}
for (i = p->items; i; i = i->next) {
for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
if (i->dead)
continue;
@ -359,13 +393,6 @@ int pa_rtpoll_run(pa_rtpoll *p, int wait) {
if (!i->after_cb)
continue;
if (no_events) {
unsigned j;
for (j = 0; j < i->n_pollfd; j++)
i->pollfd[j].revents = 0;
}
i->after_cb(i);
}
@ -386,7 +413,7 @@ finish:
}
}
if (r < 0)
if (saved_errno != 0)
errno = saved_errno;
return r;
@ -484,11 +511,10 @@ void pa_rtpoll_set_timer_disabled(pa_rtpoll *p) {
update_timer(p);
}
pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, unsigned n_fds) {
pa_rtpoll_item *i;
pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsigned n_fds) {
pa_rtpoll_item *i, *j, *l = NULL;
pa_assert(p);
pa_assert(n_fds > 0);
if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
i = pa_xnew(pa_rtpoll_item, 1);
@ -497,15 +523,25 @@ pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, unsigned n_fds) {
i->dead = 0;
i->n_pollfd = n_fds;
i->pollfd = NULL;
i->priority = prio;
i->userdata = NULL;
i->before_cb = NULL;
i->after_cb = NULL;
PA_LLIST_PREPEND(pa_rtpoll_item, p->items, i);
for (j = p->items; j; j = j->next) {
if (prio <= j->priority)
break;
p->rebuild_needed = 1;
p->n_pollfd_used += n_fds;
l = j;
}
PA_LLIST_INSERT_AFTER(pa_rtpoll_item, p->items, j ? j->prev : l, i);
if (n_fds > 0) {
p->rebuild_needed = 1;
p->n_pollfd_used += n_fds;
}
return i;
}
@ -525,8 +561,9 @@ void pa_rtpoll_item_free(pa_rtpoll_item *i) {
struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds) {
pa_assert(i);
if (i->rtpoll->rebuild_needed)
rtpoll_rebuild(i->rtpoll);
if (i->n_pollfd > 0)
if (i->rtpoll->rebuild_needed)
rtpoll_rebuild(i->rtpoll);
if (n_fds)
*n_fds = i->n_pollfd;
@ -536,12 +573,14 @@ struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds) {
void pa_rtpoll_item_set_before_callback(pa_rtpoll_item *i, int (*before_cb)(pa_rtpoll_item *i)) {
pa_assert(i);
pa_assert(i->priority < PA_RTPOLL_NEVER);
i->before_cb = before_cb;
}
void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i)) {
pa_assert(i);
pa_assert(i->priority < PA_RTPOLL_NEVER);
i->after_cb = after_cb;
}
@ -559,22 +598,28 @@ void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i) {
}
static int fdsem_before(pa_rtpoll_item *i) {
return pa_fdsem_before_poll(i->userdata);
if (pa_fdsem_before_poll(i->userdata) < 0)
return 1; /* 1 means immediate restart of the loop */
return 0;
}
static void fdsem_after(pa_rtpoll_item *i) {
pa_assert(i);
pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
pa_fdsem_after_poll(i->userdata);
}
pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_fdsem *f) {
pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_fdsem *f) {
pa_rtpoll_item *i;
struct pollfd *pollfd;
pa_assert(p);
pa_assert(f);
i = pa_rtpoll_item_new(p, 1);
i = pa_rtpoll_item_new(p, prio, 1);
pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
@ -589,22 +634,29 @@ pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_fdsem *f) {
}
static int asyncmsgq_before(pa_rtpoll_item *i) {
return pa_asyncmsgq_before_poll(i->userdata);
pa_assert(i);
if (pa_asyncmsgq_before_poll(i->userdata) < 0)
return 1; /* 1 means immediate restart of the loop */
return 0;
}
static void asyncmsgq_after(pa_rtpoll_item *i) {
pa_assert(i);
pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
pa_asyncmsgq_after_poll(i->userdata);
}
pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq(pa_rtpoll *p, pa_asyncmsgq *q) {
pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q) {
pa_rtpoll_item *i;
struct pollfd *pollfd;
pa_assert(p);
pa_assert(q);
i = pa_rtpoll_item_new(p, 1);
i = pa_rtpoll_item_new(p, prio, 1);
pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
pollfd->fd = pa_asyncmsgq_get_fd(q);

22
src/pulsecore/rtpoll.h
View file

@ -26,6 +26,7 @@
#include <poll.h>
#include <sys/types.h>
#include <limits.h>
#include <pulse/sample.h>
#include <pulsecore/asyncmsgq.h>
@ -53,9 +54,17 @@
typedef struct pa_rtpoll pa_rtpoll;
typedef struct pa_rtpoll_item pa_rtpoll_item;
typedef enum pa_rtpoll_priority {
PA_RTPOLL_EARLY = -100, /* For veeery important stuff, like handling control messages */
PA_RTPOLL_NORMAL = 0, /* For normal stuff */
PA_RTPOLL_LATE = +100, /* For housekeeping */
PA_RTPOLL_NEVER = INT_MAX, /* For stuff that doesn't register any callbacks, but only fds to listen on */
} pa_rtpoll_priority_t;
pa_rtpoll *pa_rtpoll_new(void);
void pa_rtpoll_free(pa_rtpoll *p);
/* Install the rtpoll in the current thread */
void pa_rtpoll_install(pa_rtpoll *p);
/* Sleep on the rtpoll until the time event, or any of the fd events
@ -68,7 +77,8 @@ void pa_rtpoll_set_timer_periodic(pa_rtpoll *p, pa_usec_t usec);
void pa_rtpoll_set_timer_relative(pa_rtpoll *p, pa_usec_t usec);
void pa_rtpoll_set_timer_disabled(pa_rtpoll *p);
pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, unsigned n_fds);
/* A new fd wakeup item for pa_rtpoll */
pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsigned n_fds);
void pa_rtpoll_item_free(pa_rtpoll_item *i);
/* Please note that this pointer might change on every call and when
@ -76,12 +86,18 @@ void pa_rtpoll_item_free(pa_rtpoll_item *i);
* using the pointer and don't save the result anywhere */
struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds);
/* Set the callback that shall be called immediately before entering
* the sleeping poll: If the callback returns a negative value, the
* poll is skipped. */
void pa_rtpoll_item_set_before_callback(pa_rtpoll_item *i, int (*before_cb)(pa_rtpoll_item *i));
/* Set the callback that shall be called immediately after having
* entered the sleeping poll */
void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i));
void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata);
void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i);
pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_fdsem *s);
pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq(pa_rtpoll *p, pa_asyncmsgq *q);
pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_fdsem *s);
pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q);
#endif

2
src/pulsecore/sink.c
View file

@ -126,7 +126,7 @@ pa_sink* pa_sink_new(
n = pa_sprintf_malloc("%s.monitor", name);
if (!(s->monitor_source = pa_source_new(core, driver, n, 0, spec, map)))
pa_log_warn("failed to create monitor source.");
pa_log_warn("Failed to create monitor source.");
else {
char *d;
s->monitor_source->monitor_of = s;

16
src/tests/rtpoll-test.c
View file

@ -38,16 +38,21 @@ static void after(pa_rtpoll_item *i) {
pa_log("after");
}
static int worker(pa_rtpoll_item *w) {
pa_log("worker");
return 0;
}
int main(int argc, char *argv[]) {
pa_rtpoll *p;
pa_rtpoll_item *i;
pa_rtpoll_item *i, *w;
struct pollfd *pollfd;
pa_rtsig_configure(SIGRTMIN+10, SIGRTMAX);
p = pa_rtpoll_new();
i = pa_rtpoll_item_new(p, 1);
i = pa_rtpoll_item_new(p, PA_RTPOLL_EARLY, 1);
pa_rtpoll_item_set_before_callback(i, before);
pa_rtpoll_item_set_after_callback(i, after);
@ -55,6 +60,9 @@ int main(int argc, char *argv[]) {
pollfd->fd = 0;
pollfd->events = POLLIN;
w = pa_rtpoll_item_new(p, PA_RTPOLL_NORMAL, 0);
pa_rtpoll_item_set_before_callback(w, worker);
pa_rtpoll_install(p);
pa_rtpoll_set_timer_periodic(p, 10000000); /* 10 s */
@ -62,7 +70,7 @@ int main(int argc, char *argv[]) {
pa_rtpoll_item_free(i);
i = pa_rtpoll_item_new(p, 1);
i = pa_rtpoll_item_new(p, PA_RTPOLL_EARLY, 1);
pa_rtpoll_item_set_before_callback(i, before);
pa_rtpoll_item_set_after_callback(i, after);
@ -74,6 +82,8 @@ int main(int argc, char *argv[]) {
pa_rtpoll_item_free(i);
pa_rtpoll_item_free(w);
pa_rtpoll_free(p);
return 0;