From a67c21f093202f142438689d3f7cfbdf4ea82eea Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Sun, 28 Oct 2007 19:13:50 +0000 Subject: merge 'lennart' branch back into trunk. git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1971 fefdeb5f-60dc-0310-8127-8f9354f1896f --- src/modules/module-null-sink.c | 167 +++++++++++++++++++++++++++++------------ 1 file changed, 120 insertions(+), 47 deletions(-) (limited to 'src/modules/module-null-sink.c') diff --git a/src/modules/module-null-sink.c b/src/modules/module-null-sink.c index 54a8e890..3a4edec7 100644 --- a/src/modules/module-null-sink.c +++ b/src/modules/module-null-sink.c @@ -28,7 +28,6 @@ #include #include #include -#include #include #include #include @@ -38,12 +37,17 @@ #include #include -#include +#include #include #include #include +#include #include #include +#include +#include +#include +#include #include "module-null-sink-symdef.h" @@ -64,11 +68,14 @@ struct userdata { pa_core *core; pa_module *module; pa_sink *sink; - pa_time_event *time_event; + + pa_thread *thread; + pa_thread_mq thread_mq; + pa_rtpoll *rtpoll; + size_t block_size; - uint64_t n_bytes; - struct timeval start_time; + struct timeval timestamp; }; static const char* const valid_modargs[] = { @@ -81,78 +88,132 @@ static const char* const valid_modargs[] = { NULL }; -static void time_callback(pa_mainloop_api *m, pa_time_event*e, const struct timeval *tv, void *userdata) { - struct userdata *u = userdata; - pa_memchunk chunk; - struct timeval ntv = *tv; - size_t l; +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; + + switch (code) { + case PA_SINK_MESSAGE_SET_STATE: - assert(u); + if (PA_PTR_TO_UINT(data) == PA_SINK_RUNNING) + pa_rtclock_get(&u->timestamp); - if (pa_sink_render(u->sink, u->block_size, &chunk) >= 0) { - l = chunk.length; - pa_memblock_unref(chunk.memblock); - } else - l = u->block_size; + break; - pa_timeval_add(&ntv, pa_bytes_to_usec(l, &u->sink->sample_spec)); - m->time_restart(e, &ntv); + case PA_SINK_MESSAGE_GET_LATENCY: { + struct timeval now; - u->n_bytes += l; + pa_rtclock_get(&now); + + if (pa_timeval_cmp(&u->timestamp, &now) > 0) + *((pa_usec_t*) data) = 0; + else + *((pa_usec_t*) data) = pa_timeval_diff(&u->timestamp, &now); + break; + } + } + + return pa_sink_process_msg(o, code, data, offset, chunk); } -static pa_usec_t get_latency(pa_sink *s) { - struct userdata *u = s->userdata; - pa_usec_t a, b; - struct timeval now; +static void thread_func(void *userdata) { + struct userdata *u = userdata; + + pa_assert(u); + + pa_log_debug("Thread starting up"); + + pa_thread_mq_install(&u->thread_mq); + pa_rtpoll_install(u->rtpoll); + + pa_rtclock_get(&u->timestamp); + + for (;;) { + int ret; + + /* Render some data and drop it immediately */ + if (u->sink->thread_info.state == PA_SINK_RUNNING) { + struct timeval now; + + pa_rtclock_get(&now); - a = pa_timeval_diff(pa_gettimeofday(&now), &u->start_time); - b = pa_bytes_to_usec(u->n_bytes, &s->sample_spec); + if (pa_timeval_cmp(&u->timestamp, &now) <= 0) { + pa_sink_skip(u->sink, u->block_size); + pa_timeval_add(&u->timestamp, pa_bytes_to_usec(u->block_size, &u->sink->sample_spec)); + } - return b > a ? b - a : 0; + pa_rtpoll_set_timer_absolute(u->rtpoll, &u->timestamp); + } else + pa_rtpoll_set_timer_disabled(u->rtpoll); + + /* Hmm, nothing to do. Let's sleep */ + if ((ret = pa_rtpoll_run(u->rtpoll, 1)) < 0) + goto fail; + + if (ret == 0) + goto finish; + } + +fail: + /* If this was no regular exit from the loop we have to continue + * processing messages until we received PA_MESSAGE_SHUTDOWN */ + pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL); + pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN); + +finish: + pa_log_debug("Thread shutting down"); } -int pa__init(pa_core *c, pa_module*m) { +int pa__init(pa_module*m) { struct userdata *u = NULL; pa_sample_spec ss; pa_channel_map map; pa_modargs *ma = NULL; - assert(c); - assert(m); + pa_assert(m); if (!(ma = pa_modargs_new(m->argument, valid_modargs))) { - pa_log("failed to parse module arguments."); + pa_log("Failed to parse module arguments."); goto fail; } - ss = c->default_sample_spec; + ss = m->core->default_sample_spec; if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) { - pa_log("invalid sample format specification or channel map."); + pa_log("Invalid sample format specification or channel map"); goto fail; } u = pa_xnew0(struct userdata, 1); - u->core = c; + u->core = m->core; u->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, PA_RTPOLL_EARLY, u->thread_mq.inq); - if (!(u->sink = pa_sink_new(c, __FILE__, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME), 0, &ss, &map))) { - pa_log("failed to create sink."); + 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."); goto fail; } - u->sink->get_latency = get_latency; + u->sink->parent.process_msg = sink_process_msg; u->sink->userdata = u; - pa_sink_set_owner(u->sink, m); + u->sink->flags = PA_SINK_LATENCY; + + pa_sink_set_module(u->sink, m); + pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq); + pa_sink_set_rtpoll(u->sink, u->rtpoll); pa_sink_set_description(u->sink, pa_modargs_get_value(ma, "description", "NULL sink")); - u->n_bytes = 0; - pa_gettimeofday(&u->start_time); + u->block_size = pa_bytes_per_second(&ss) / 20; /* 50 ms */ + if (u->block_size <= 0) + u->block_size = pa_frame_size(&ss); - u->time_event = c->mainloop->time_new(c->mainloop, &u->start_time, time_callback, u); + if (!(u->thread = pa_thread_new(thread_func, u))) { + pa_log("Failed to create thread."); + goto fail; + } - u->block_size = pa_bytes_per_second(&ss) / 10; + pa_sink_put(u->sink); pa_modargs_free(ma); @@ -162,22 +223,34 @@ fail: if (ma) pa_modargs_free(ma); - pa__done(c, m); + pa__done(m); return -1; } -void pa__done(pa_core *c, pa_module*m) { +void pa__done(pa_module*m) { struct userdata *u; - assert(c && m); + + pa_assert(m); if (!(u = m->userdata)) return; - pa_sink_disconnect(u->sink); - pa_sink_unref(u->sink); + if (u->sink) + pa_sink_unlink(u->sink); + + if (u->thread) { + pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL); + pa_thread_free(u->thread); + } + + pa_thread_mq_done(&u->thread_mq); + + if (u->sink) + pa_sink_unref(u->sink); - u->core->mainloop->time_free(u->time_event); + if (u->rtpoll) + pa_rtpoll_free(u->rtpoll); pa_xfree(u); } -- cgit