diff options
26 files changed, 702 insertions, 372 deletions
diff --git a/src/modules/module-alsa-sink.c b/src/modules/module-alsa-sink.c index a805845c..24765a04 100644 --- a/src/modules/module-alsa-sink.c +++ b/src/modules/module-alsa-sink.c @@ -347,7 +347,7 @@ static int suspend(struct userdata *u) { u->alsa_rtpoll_item = NULL; } - pa_log_debug("Device suspended..."); + pa_log_info("Device suspended..."); return 0; } @@ -361,7 +361,7 @@ static int unsuspend(struct userdata *u) { pa_assert(u); pa_assert(!u->pcm_handle); - pa_log_debug("Trying resume..."); + pa_log_info("Trying resume..."); snd_config_update_free_global(); if ((err = snd_pcm_open(&u->pcm_handle, u->device_name, SND_PCM_STREAM_PLAYBACK, SND_PCM_NONBLOCK)) < 0) { @@ -406,7 +406,7 @@ static int unsuspend(struct userdata *u) { u->first = 1; - pa_log_debug("Resumed successfully..."); + pa_log_info("Resumed successfully..."); return 0; @@ -457,7 +457,8 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse break; - case PA_SINK_DISCONNECTED: + case PA_SINK_UNLINKED: + case PA_SINK_INIT: ; } @@ -607,13 +608,7 @@ static void thread_func(void *userdata) { goto fail; for (;;) { - pa_msgobject *object; - int code; - void *data; - pa_memchunk chunk; - int64_t offset; - -/* pa_log("loop"); */ + int ret; /* Render some data and write it to the dsp */ if (PA_SINK_OPENED(u->sink->thread_info.state)) { @@ -635,28 +630,25 @@ static void thread_func(void *userdata) { } } -/* pa_log("loop2"); */ + /* Now give the sink inputs some to time to process their data */ + if ((ret = pa_sink_process_inputs(u->sink)) < 0) + goto fail; + if (ret > 0) + continue; /* Check whether there is a message for us to process */ - if (pa_asyncmsgq_get(u->thread_mq.inq, &object, &code, &data, &offset, &chunk, 0) == 0) { - int ret; - - if (!object && code == PA_MESSAGE_SHUTDOWN) { - pa_asyncmsgq_done(u->thread_mq.inq, 0); - goto finish; - } - - ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk); - pa_asyncmsgq_done(u->thread_mq.inq, ret); + if ((ret = pa_thread_mq_process(&u->thread_mq) < 0)) + goto finish; + if (ret > 0) continue; - } - + /* Hmm, nothing to do. Let's sleep */ if (pa_rtpoll_run(u->rtpoll) < 0) { pa_log("poll() failed: %s", pa_cstrerror(errno)); goto fail; } + /* Tell ALSA about this and process its response */ if (PA_SINK_OPENED(u->sink->thread_info.state)) { struct pollfd *pollfd; unsigned short revents = 0; @@ -680,9 +672,7 @@ static void thread_func(void *userdata) { goto fail; } -/* pa_log("got alsa event"); */ } - } fail: @@ -828,6 +818,7 @@ int pa__init(pa_module*m) { 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, t = pa_sprintf_malloc( "ALSA PCM on %s (%s)%s", dev, @@ -835,7 +826,7 @@ int pa__init(pa_module*m) { use_mmap ? " via DMA" : "")); pa_xfree(t); - u->sink->is_hardware = 1; + u->sink->flags = PA_SINK_HARDWARE|PA_SINK_CAN_SUSPEND|PA_SINK_HW_VOLUME_CTRL|PA_SINK_LATENCY; u->frame_size = frame_size; u->fragment_size = frag_size = period_size * frame_size; @@ -894,6 +885,8 @@ int pa__init(pa_module*m) { if (u->sink->get_mute) u->sink->get_mute(u->sink); + pa_sink_put(u->sink); + pa_modargs_free(ma); return 0; @@ -917,7 +910,7 @@ void pa__done(pa_module*m) { return; if (u->sink) - pa_sink_disconnect(u->sink); + pa_sink_unlink(u->sink); if (u->thread) { pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL); diff --git a/src/modules/module-alsa-source.c b/src/modules/module-alsa-source.c index 6ea99ec7..0d76cabc 100644 --- a/src/modules/module-alsa-source.c +++ b/src/modules/module-alsa-source.c @@ -328,7 +328,7 @@ static int suspend(struct userdata *u) { u->alsa_rtpoll_item = NULL; } - pa_log_debug("Device suspended..."); + pa_log_info("Device suspended..."); return 0; } @@ -342,7 +342,7 @@ static int unsuspend(struct userdata *u) { pa_assert(u); pa_assert(!u->pcm_handle); - pa_log_debug("Trying resume..."); + pa_log_info("Trying resume..."); snd_config_update_free_global(); if ((err = snd_pcm_open(&u->pcm_handle, u->device_name, SND_PCM_STREAM_CAPTURE, SND_PCM_NONBLOCK)) < 0) { @@ -387,7 +387,7 @@ static int unsuspend(struct userdata *u) { /* FIXME: We need to reload the volume somehow */ - pa_log_debug("Resumed successfully..."); + pa_log_info("Resumed successfully..."); return 0; @@ -438,7 +438,8 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off break; - case PA_SOURCE_DISCONNECTED: + case PA_SOURCE_UNLINKED: + case PA_SOURCE_INIT: ; } @@ -590,15 +591,9 @@ static void thread_func(void *userdata) { snd_pcm_start(u->pcm_handle); for (;;) { - pa_msgobject *object; - int code; - void *data; - int64_t offset; - pa_memchunk chunk; + int ret; -/* pa_log("loop"); */ - - /* Render some data and write it to the dsp */ + /* Read some data and pass it to the sources */ if (PA_SOURCE_OPENED(u->source->thread_info.state)) { if (u->use_mmap) { @@ -611,29 +606,25 @@ static void thread_func(void *userdata) { } } -/* pa_log("loop2"); */ + /* Now give the source outputs some to time to process their data */ + if ((ret = pa_source_process_outputs(u->source)) < 0) + goto fail; + if (ret > 0) + continue; /* Check whether there is a message for us to process */ - if (pa_asyncmsgq_get(u->thread_mq.inq, &object, &code, &data, &offset, &chunk, 0) == 0) { - int ret; - -/* pa_log("processing msg"); */ - - if (!object && code == PA_MESSAGE_SHUTDOWN) { - pa_asyncmsgq_done(u->thread_mq.inq, 0); - goto finish; - } - - ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk); - pa_asyncmsgq_done(u->thread_mq.inq, ret); + if ((ret = pa_thread_mq_process(&u->thread_mq) < 0)) + goto finish; + if (ret > 0) continue; - } + /* Hmm, nothing to do. Let's sleep */ if (pa_rtpoll_run(u->rtpoll) < 0) { pa_log("poll() failed: %s", pa_cstrerror(errno)); goto fail; } + /* Tell ALSA about this and process its response */ if (PA_SOURCE_OPENED(u->source->thread_info.state)) { struct pollfd *pollfd; unsigned short revents = 0; @@ -657,7 +648,6 @@ static void thread_func(void *userdata) { goto fail; } -/* pa_log("got alsa event"); */ } } @@ -802,6 +792,7 @@ int pa__init(pa_module*m) { pa_source_set_module(u->source, m); pa_source_set_asyncmsgq(u->source, u->thread_mq.inq); + pa_source_set_rtpoll(u->source, u->rtpoll); pa_source_set_description(u->source, t = pa_sprintf_malloc( "ALSA PCM on %s (%s)%s", dev, @@ -809,7 +800,7 @@ int pa__init(pa_module*m) { use_mmap ? " via DMA" : "")); pa_xfree(t); - u->source->is_hardware = 1; + u->source->flags = PA_SOURCE_HARDWARE|PA_SOURCE_CAN_SUSPEND|PA_SOURCE_LATENCY|PA_SOURCE_HW_VOLUME_CTRL; u->frame_size = frame_size; u->fragment_size = frag_size = period_size * frame_size; @@ -863,6 +854,8 @@ int pa__init(pa_module*m) { if (u->source->get_mute) u->source->get_mute(u->source); + pa_source_put(u->source); + pa_modargs_free(ma); return 0; @@ -886,7 +879,7 @@ void pa__done(pa_module*m) { return; if (u->source) - pa_source_disconnect(u->source); + pa_source_unlink(u->source); if (u->thread) { pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL); diff --git a/src/modules/module-null-sink.c b/src/modules/module-null-sink.c index 3b512371..78d99ceb 100644 --- a/src/modules/module-null-sink.c +++ b/src/modules/module-null-sink.c @@ -128,11 +128,7 @@ static void thread_func(void *userdata) { pa_rtclock_get(&u->timestamp); for (;;) { - pa_msgobject *object; - int code; - void *data; - pa_memchunk chunk; - int64_t offset; + int ret; /* Render some data and drop it immediately */ if (u->sink->thread_info.state == PA_SINK_RUNNING) { @@ -141,30 +137,25 @@ static void thread_func(void *userdata) { pa_rtclock_get(&now); if (pa_timespec_cmp(&u->timestamp, &now) <= 0) { - - pa_sink_render(u->sink, u->block_size, &chunk); - pa_memblock_unref(chunk.memblock); - - pa_timespec_add(&u->timestamp, pa_bytes_to_usec(chunk.length, &u->sink->sample_spec)); + pa_sink_skip(u->sink, u->block_size); + pa_timespec_add(&u->timestamp, pa_bytes_to_usec(u->block_size, &u->sink->sample_spec)); } pa_rtpoll_set_timer_absolute(u->rtpoll, &u->timestamp); } else pa_rtpoll_set_timer_disabled(u->rtpoll); - /* Check whether there is a message for us to process */ - if (pa_asyncmsgq_get(u->thread_mq.inq, &object, &code, &data, &offset, &chunk, 0) == 0) { - int ret; - - if (!object && code == PA_MESSAGE_SHUTDOWN) { - pa_asyncmsgq_done(u->thread_mq.inq, 0); - goto finish; - } + /* Now give the sink inputs some to time to process their data */ + if ((ret = pa_sink_process_inputs(u->sink)) < 0) + goto fail; + if (ret > 0) + continue; - ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk); - pa_asyncmsgq_done(u->thread_mq.inq, ret); + /* Check whether there is a message for us to process */ + if ((ret = pa_thread_mq_process(&u->thread_mq) < 0)) + goto finish; + if (ret > 0) continue; - } /* Hmm, nothing to do. Let's sleep */ if (pa_rtpoll_run(u->rtpoll) < 0) { @@ -217,9 +208,11 @@ int pa__init(pa_module*m) { u->sink->parent.process_msg = sink_process_msg; u->sink->userdata = u; + 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->block_size = pa_bytes_per_second(&ss) / 20; /* 50 ms */ @@ -231,6 +224,8 @@ int pa__init(pa_module*m) { goto fail; } + pa_sink_put(u->sink); + pa_modargs_free(ma); return 0; @@ -253,7 +248,7 @@ void pa__done(pa_module*m) { return; if (u->sink) - pa_sink_disconnect(u->sink); + pa_sink_unlink(u->sink); if (u->thread) { pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL); diff --git a/src/modules/module-oss.c b/src/modules/module-oss.c index 85fdd9c8..bbb1e0d2 100644 --- a/src/modules/module-oss.c +++ b/src/modules/module-oss.c @@ -443,7 +443,7 @@ static int suspend(struct userdata *u) { pa_assert(u); pa_assert(u->fd >= 0); - pa_log_debug("Suspending..."); + pa_log_info("Suspending..."); if (u->out_mmap_memblocks) { unsigned i; @@ -483,7 +483,7 @@ static int suspend(struct userdata *u) { u->rtpoll_item = NULL; } - pa_log_debug("Device suspended..."); + pa_log_info("Device suspended..."); return 0; } @@ -501,7 +501,7 @@ static int unsuspend(struct userdata *u) { m = u->mode; - pa_log_debug("Trying resume..."); + pa_log_info("Trying resume..."); if ((u->fd = pa_oss_open(u->device_name, &m, NULL)) < 0) { pa_log_warn("Resume failed, device busy (%s)", pa_cstrerror(errno)); @@ -582,7 +582,7 @@ static int unsuspend(struct userdata *u) { pollfd->events = 0; pollfd->revents = 0; - pa_log_debug("Resumed successfully..."); + pa_log_info("Resumed successfully..."); return 0; @@ -651,7 +651,8 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse break; - case PA_SINK_DISCONNECTED: + case PA_SINK_UNLINKED: + case PA_SINK_INIT: ; } @@ -748,7 +749,8 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off } break; - case PA_SOURCE_DISCONNECTED: + case PA_SOURCE_UNLINKED: + case PA_SOURCE_INIT: ; } @@ -807,20 +809,15 @@ static void thread_func(void *userdata) { trigger(u, 0); for (;;) { - pa_msgobject *object; - int code; - void *data; - pa_memchunk chunk; - int64_t offset; + int ret; /* pa_log("loop"); */ /* Render some data and write it to the dsp */ - if (u->sink && u->sink->thread_info.state != PA_SINK_DISCONNECTED && u->fd >= 0 && (revents & POLLOUT)) { + if (u->sink && u->sink->thread_info.state != PA_SINK_UNLINKED && u->fd >= 0 && (revents & POLLOUT)) { if (u->use_mmap) { - int ret; if ((ret = mmap_write(u)) < 0) goto fail; @@ -908,10 +905,9 @@ static void thread_func(void *userdata) { /* Try to read some data and pass it on to the source driver */ - if (u->source && u->source->thread_info.state != PA_SOURCE_DISCONNECTED && u->fd >= 0 && ((revents & POLLIN))) { + if (u->source && u->source->thread_info.state != PA_SOURCE_UNLINKED && u->fd >= 0 && ((revents & POLLIN))) { if (u->use_mmap) { - int ret; if ((ret = mmap_read(u)) < 0) goto fail; @@ -995,21 +991,23 @@ static void thread_func(void *userdata) { /* pa_log("loop2"); */ - /* Check whether there is a message for us to process */ - if (pa_asyncmsgq_get(u->thread_mq.inq, &object, &code, &data, &offset, &chunk, 0) == 0) { - int ret; - -/* pa_log("processing msg"); */ - - if (!object && code == PA_MESSAGE_SHUTDOWN) { - pa_asyncmsgq_done(u->thread_mq.inq, 0); - goto finish; - } + /* Now give the sink inputs some to time to process their data */ + if ((ret = pa_sink_process_inputs(u->sink)) < 0) + goto fail; + if (ret > 0) + continue; - ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk); - pa_asyncmsgq_done(u->thread_mq.inq, ret); + /* Now give the source outputs some to time to process their data */ + if ((ret = pa_source_process_outputs(u->source)) < 0) + goto fail; + if (ret > 0) + continue; + + /* Check whether there is a message for us to process */ + if ((ret = pa_thread_mq_process(&u->thread_mq) < 0)) + goto finish; + if (ret > 0) continue; - } if (u->fd >= 0) { struct pollfd *pollfd; @@ -1019,7 +1017,6 @@ static void thread_func(void *userdata) { ((u->source && PA_SOURCE_OPENED(u->source->thread_info.state)) ? POLLIN : 0) | ((u->sink && PA_SINK_OPENED(u->sink->thread_info.state)) ? POLLOUT : 0); } - /* Hmm, nothing to do. Let's sleep */ if (pa_rtpoll_run(u->rtpoll) < 0) { @@ -1212,6 +1209,7 @@ int pa__init(pa_module*m) { pa_source_set_module(u->source, m); pa_source_set_asyncmsgq(u->source, u->thread_mq.inq); + pa_source_set_rtpoll(u->source, u->rtpoll); pa_source_set_description(u->source, t = pa_sprintf_malloc( "OSS PCM on %s%s%s%s%s", dev, @@ -1220,7 +1218,7 @@ int pa__init(pa_module*m) { hwdesc[0] ? ")" : "", use_mmap ? " via DMA" : "")); pa_xfree(t); - u->source->is_hardware = 1; + u->source->flags = PA_SOURCE_HARDWARE|PA_SOURCE_CAN_SUSPEND|PA_SOURCE_LATENCY|PA_SOURCE_HW_VOLUME_CTRL; u->source->refresh_volume = 1; if (use_mmap) @@ -1266,6 +1264,7 @@ int pa__init(pa_module*m) { 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, t = pa_sprintf_malloc( "OSS PCM on %s%s%s%s%s", dev, @@ -1274,7 +1273,7 @@ int pa__init(pa_module*m) { hwdesc[0] ? ")" : "", use_mmap ? " via DMA" : "")); pa_xfree(t); - u->sink->is_hardware = 1; + u->sink->flags = PA_SINK_HARDWARE|PA_SINK_CAN_SUSPEND|PA_SINK_LATENCY|PA_SINK_HW_VOLUME_CTRL; u->sink->refresh_volume = 1; if (use_mmap) @@ -1298,6 +1297,11 @@ go_on: if (u->sink) pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), PA_SINK_MESSAGE_GET_VOLUME, &u->sink->volume, 0, NULL, NULL); + if (u->sink) + pa_sink_put(u->sink); + if (u->source) + pa_source_put(u->source); + pa_modargs_free(ma); return 0; @@ -1324,10 +1328,10 @@ void pa__done(pa_module*m) { return; if (u->sink) - pa_sink_disconnect(u->sink); + pa_sink_unlink(u->sink); if (u->source) - pa_source_disconnect(u->source); + pa_source_unlink(u->source); if (u->thread) { pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL); diff --git a/src/modules/module-pipe-sink.c b/src/modules/module-pipe-sink.c index a1101ab8..d5ef194f 100644 --- a/src/modules/module-pipe-sink.c +++ b/src/modules/module-pipe-sink.c @@ -124,17 +124,12 @@ static void thread_func(void *userdata) { pa_rtpoll_install(u->rtpoll); for (;;) { - pa_msgobject *object; - int code; - void *data; - pa_memchunk chunk; - int64_t offset; + int ret; struct pollfd *pollfd; - /* Render some data and write it to the fifo */ - pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL); - + + /* Render some data and write it to the fifo */ if (u->sink->thread_info.state == PA_SINK_RUNNING && pollfd->revents) { ssize_t l; void *p; @@ -173,19 +168,17 @@ static void thread_func(void *userdata) { } } - /* Check whether there is a message for us to process */ - if (pa_asyncmsgq_get(u->thread_mq.inq, &object, &code, &data, &offset, &chunk, 0) == 0) { - int ret; - - if (!object && code == PA_MESSAGE_SHUTDOWN) { - pa_asyncmsgq_done(u->thread_mq.inq, 0); - goto finish; - } + /* Now give the sink inputs some to time to process their data */ + if ((ret = pa_sink_process_inputs(u->sink)) < 0) + goto fail; + if (ret > 0) + continue; - ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk); - pa_asyncmsgq_done(u->thread_mq.inq, ret); + /* Check whether there is a message for us to process */ + if ((ret = pa_thread_mq_process(&u->thread_mq) < 0)) + goto finish; + if (ret > 0) continue; - } /* Hmm, nothing to do. Let's sleep */ pollfd->events = u->sink->thread_info.state == PA_SINK_RUNNING ? POLLOUT : 0; @@ -271,9 +264,11 @@ int pa__init(pa_module*m) { u->sink->parent.process_msg = sink_process_msg; u->sink->userdata = u; + 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, t = pa_sprintf_malloc("Unix FIFO sink '%s'", u->filename)); pa_xfree(t); @@ -281,12 +276,14 @@ int pa__init(pa_module*m) { pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL); pollfd->fd = u->fd; pollfd->events = pollfd->revents = 0; - + if (!(u->thread = pa_thread_new(thread_func, u))) { pa_log("Failed to create thread."); goto fail; } + pa_sink_put(u->sink); + pa_modargs_free(ma); return 0; @@ -309,7 +306,7 @@ void pa__done(pa_module*m) { return; if (u->sink) - pa_sink_disconnect(u->sink); + pa_sink_unlink(u->sink); if (u->thread) { pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL); diff --git a/src/modules/module-pipe-source.c b/src/modules/module-pipe-source.c index 2209d1ee..6dc94648 100644 --- a/src/modules/module-pipe-source.c +++ b/src/modules/module-pipe-source.c @@ -45,6 +45,7 @@ #include <pulsecore/log.h> #include <pulsecore/thread.h> #include <pulsecore/thread-mq.h> +#include <pulsecore/rtpoll.h> #include "module-pipe-source-symdef.h" @@ -66,13 +67,17 @@ struct userdata { pa_core *core; pa_module *module; pa_source *source; + pa_thread *thread; pa_thread_mq thread_mq; - + pa_rtpoll *rtpoll; + char *filename; int fd; pa_memchunk memchunk; + + pa_rtpoll_item *rtpoll_item; }; static const char* const valid_modargs[] = { @@ -86,14 +91,7 @@ static const char* const valid_modargs[] = { }; static void thread_func(void *userdata) { - enum { - POLLFD_ASYNCQ, - POLLFD_FIFO, - POLLFD_MAX, - }; - struct userdata *u = userdata; - struct pollfd pollfd[POLLFD_MAX]; int read_type = 0; pa_assert(u); @@ -101,40 +99,18 @@ static void thread_func(void *userdata) { pa_log_debug("Thread starting up"); pa_thread_mq_install(&u->thread_mq); - - memset(&pollfd, 0, sizeof(pollfd)); + pa_rtpoll_install(u->rtpoll); - pollfd[POLLFD_ASYNCQ].fd = pa_asyncmsgq_get_fd(u->thread_mq.inq); - pollfd[POLLFD_ASYNCQ].events = POLLIN; - pollfd[POLLFD_FIFO].fd = u->fd; - for (;;) { - pa_msgobject *object; - int code; - void *data; - pa_memchunk chunk; - int r; - int64_t offset; - - /* Check whether there is a message for us to process */ - if (pa_asyncmsgq_get(u->thread_mq.inq, &object, &code, &data, &offset, &chunk, 0) == 0) { - int ret; - - if (!object && code == PA_MESSAGE_SHUTDOWN) { - pa_asyncmsgq_done(u->thread_mq.inq, 0); - goto finish; - } - - ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk); - pa_asyncmsgq_done(u->thread_mq.inq, ret); - continue; - } + int ret; + struct pollfd *pollfd; + pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL); + /* Try to read some data and pass it on to the source driver */ - - if (u->source->thread_info.state == PA_SOURCE_RUNNING && pollfd[POLLFD_FIFO].revents) { - void *p; + if (u->source->thread_info.state == PA_SOURCE_RUNNING && pollfd->revents) { ssize_t l; + void *p; if (!u->memchunk.memblock) { u->memchunk.memblock = pa_memblock_new(u->core->mempool, PIPE_BUF); @@ -169,38 +145,35 @@ static void thread_func(void *userdata) { pa_memchunk_reset(&u->memchunk); } - pollfd[POLLFD_FIFO].revents = 0; - continue; + pollfd->revents = 0; } } - pollfd[POLLFD_FIFO].events = u->source->thread_info.state == PA_SOURCE_RUNNING ? POLLIN : 0; - - /* Hmm, nothing to do. Let's sleep */ - - if (pa_asyncmsgq_before_poll(u->thread_mq.inq) < 0) + /* Now give the source outputs some to time to process their data */ + if ((ret = pa_source_process_outputs(u->source)) < 0) + goto fail; + if (ret > 0) continue; -/* pa_log("polling for %i", pollfd[POLLFD_FIFO].events); */ - r = poll(pollfd, POLLFD_MAX, -1); -/* pa_log("polling got %i (r=%i) %i", r > 0 ? pollfd[POLLFD_FIFO].revents : 0, r, r > 0 ? pollfd[POLLFD_ASYNCQ].revents: 0); */ - - pa_asyncmsgq_after_poll(u->thread_mq.inq); + /* Check whether there is a message for us to process */ + if ((ret = pa_thread_mq_process(&u->thread_mq) < 0)) + goto finish; + if (ret > 0) + continue; - if (r < 0) { - if (errno == EINTR) - continue; + /* Hmm, nothing to do. Let's sleep */ + pollfd->events = u->source->thread_info.state == PA_SOURCE_RUNNING ? POLLIN : 0; + if (pa_rtpoll_run(u->rtpoll) < 0) { pa_log("poll() failed: %s", pa_cstrerror(errno)); goto fail; } - if (pollfd[POLLFD_FIFO].revents & ~POLLIN) { + pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL); + if (pollfd->revents & ~POLLIN) { pa_log("FIFO shutdown."); goto fail; } - - pa_assert((pollfd[POLLFD_ASYNCQ].revents & ~POLLIN) == 0); } fail: @@ -220,6 +193,7 @@ int pa__init(pa_module*m) { pa_channel_map map; pa_modargs *ma; char *t; + struct pollfd *pollfd; pa_assert(m); @@ -240,6 +214,8 @@ int pa__init(pa_module*m) { m->userdata = u; 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); u->filename = pa_xstrdup(pa_modargs_get_value(ma, "file", DEFAULT_FILE_NAME)); @@ -268,17 +244,26 @@ int pa__init(pa_module*m) { } u->source->userdata = u; + u->source->flags = 0; pa_source_set_module(u->source, m); pa_source_set_asyncmsgq(u->source, u->thread_mq.inq); + pa_source_set_rtpoll(u->source, u->rtpoll); 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); + pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL); + pollfd->fd = u->fd; + pollfd->events = pollfd->revents = 0; + if (!(u->thread = pa_thread_new(thread_func, u))) { pa_log("Failed to create thread."); goto fail; } + pa_source_put(u->source); + pa_modargs_free(ma); return 0; @@ -301,7 +286,7 @@ void pa__done(pa_module*m) { return; if (u->source) - pa_source_disconnect(u->source); + pa_source_unlink(u->source); if (u->thread) { pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL); @@ -316,6 +301,12 @@ void pa__done(pa_module*m) { if (u->memchunk.memblock) pa_memblock_unref(u->memchunk.memblock); + if (u->rtpoll_item) + pa_rtpoll_item_free(u->rtpoll_item); + + if (u->rtpoll) + pa_rtpoll_free(u->rtpoll); + if (u->filename) { unlink(u->filename); pa_xfree(u->filename); diff --git a/src/modules/module-rescue-streams.c b/src/modules/module-rescue-streams.c index e0eed036..24077dc2 100644 --- a/src/modules/module-rescue-streams.c +++ b/src/modules/module-rescue-streams.c @@ -138,8 +138,8 @@ int pa__init(pa_module*m) { } m->userdata = u = pa_xnew(struct userdata, 1); - u->sink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_DISCONNECT], (pa_hook_cb_t) sink_hook_callback, NULL); - u->source_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_DISCONNECT], (pa_hook_cb_t) source_hook_callback, NULL); + u->sink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_UNLINK], (pa_hook_cb_t) sink_hook_callback, NULL); + u->source_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_UNLINK], (pa_hook_cb_t) source_hook_callback, NULL); pa_modargs_free(ma); return 0; diff --git a/src/modules/module-sine.c b/src/modules/module-sine.c index f2830ff0..94036f6c 100644 --- a/src/modules/module-sine.c +++ b/src/modules/module-sine.c @@ -98,7 +98,7 @@ static void sink_input_kill_cb(pa_sink_input *i) { u = i->userdata; pa_assert(u); - pa_sink_input_disconnect(u->sink_input); + pa_sink_input_unlink(u->sink_input); pa_sink_input_unref(u->sink_input); u->sink_input = NULL; @@ -195,7 +195,7 @@ void pa__done(pa_module*m) { return; if (u->sink_input) { - pa_sink_input_disconnect(u->sink_input); + pa_sink_input_unlink(u->sink_input); pa_sink_input_unref(u->sink_input); } diff --git a/src/modules/module-suspend-on-idle.c b/src/modules/module-suspend-on-idle.c index e9226ca2..79850e71 100644 --- a/src/modules/module-suspend-on-idle.c +++ b/src/modules/module-suspend-on-idle.c @@ -53,16 +53,16 @@ struct userdata { pa_hook_slot *sink_new_slot, *source_new_slot, - *sink_disconnect_slot, - *source_disconnect_slot, + *sink_unlink_slot, + *source_unlink_slot, *sink_state_changed_slot, *source_state_changed_slot; pa_hook_slot *sink_input_new_slot, *source_output_new_slot, - *sink_input_disconnect_slot, - *source_output_disconnect_slot, + *sink_input_unlink_slot, + *source_output_unlink_slot, *sink_input_move_slot, *source_output_move_slot, *sink_input_move_post_slot, @@ -139,8 +139,8 @@ static pa_hook_result_t sink_input_new_hook_cb(pa_core *c, pa_sink_input *s, str pa_sink_input_assert_ref(s); pa_assert(u); - pa_assert_se((d = pa_hashmap_get(u->device_infos, s->sink))); - resume(d); + if ((d = pa_hashmap_get(u->device_infos, s->sink))) + resume(d); return PA_HOOK_OK; } @@ -152,35 +152,35 @@ static pa_hook_result_t source_output_new_hook_cb(pa_core *c, pa_source_output * pa_source_output_assert_ref(s); pa_assert(u); - pa_assert_se((d = pa_hashmap_get(u->device_infos, s->source))); - resume(d); + if ((d = pa_hashmap_get(u->device_infos, s->source))) + resume(d); return PA_HOOK_OK; } -static pa_hook_result_t sink_input_disconnect_hook_cb(pa_core *c, pa_sink_input *s, struct userdata *u) { +static pa_hook_result_t sink_input_unlink_hook_cb(pa_core *c, pa_sink_input *s, struct userdata *u) { pa_assert(c); pa_sink_input_assert_ref(s); pa_assert(u); if (pa_sink_used_by(s->sink) <= 0) { struct device_info *d; - pa_assert_se((d = pa_hashmap_get(u->device_infos, s->sink))); - restart(d); + if ((d = pa_hashmap_get(u->device_infos, s->sink))) + restart(d); } return PA_HOOK_OK; } -static pa_hook_result_t source_output_disconnect_hook_cb(pa_core *c, pa_source_output *s, struct userdata *u) { +static pa_hook_result_t source_output_unlink_hook_cb(pa_core *c, pa_source_output *s, struct userdata *u) { pa_assert(c); pa_source_output_assert_ref(s); pa_assert(u); if (pa_source_used_by(s->source) <= 0) { struct device_info *d; - pa_assert_se((d = pa_hashmap_get(u->device_infos, s->source))); - restart(d); + if ((d = pa_hashmap_get(u->device_infos, s->source))) + restart(d); } return PA_HOOK_OK; @@ -193,8 +193,8 @@ static pa_hook_result_t sink_input_move_hook_cb(pa_core *c, pa_sink_input *s, st if (pa_sink_used_by(s->sink) <= 1) { struct device_info *d; - pa_assert_se((d = pa_hashmap_get(u->device_infos, s->sink))); - restart(d); + if ((d = pa_hashmap_get(u->device_infos, s->sink))) + restart(d); } return PA_HOOK_OK; @@ -206,8 +206,8 @@ static pa_hook_result_t sink_input_move_post_hook_cb(pa_core *c, pa_sink_input * pa_sink_input_assert_ref(s); pa_assert(u); - pa_assert_se((d = pa_hashmap_get(u->device_infos, s->sink))); - resume(d); + if ((d = pa_hashmap_get(u->device_infos, s->sink))) + resume(d); return PA_HOOK_OK; } @@ -219,8 +219,9 @@ static pa_hook_result_t source_output_move_hook_cb(pa_core *c, pa_source_output if (pa_source_used_by(s->source) <= 1) { struct device_info *d; - pa_assert_se((d = pa_hashmap_get(u->device_infos, s->source))); - restart(d); + + if ((d = pa_hashmap_get(u->device_infos, s->source))) + restart(d); } return PA_HOOK_OK; @@ -232,24 +233,36 @@ static pa_hook_result_t source_output_move_post_hook_cb(pa_core *c, pa_source_ou pa_source_output_assert_ref(s); pa_assert(u); - pa_assert_se((d = pa_hashmap_get(u->device_infos, s->source))); - resume(d); + if ((d = pa_hashmap_get(u->device_infos, s->source))) + resume(d); return PA_HOOK_OK; } static pa_hook_result_t device_new_hook_cb(pa_core *c, pa_object *o, struct userdata *u) { struct device_info *d; + pa_source *source; + pa_sink *sink; pa_assert(c); pa_object_assert_ref(o); pa_assert(u); + source = pa_source_isinstance(o) ? PA_SOURCE(o) : NULL; + sink = pa_sink_isinstance(o) ? PA_SINK(o) : NULL; + + pa_assert(source || sink); + + if (source && !(source->flags & PA_SOURCE_CAN_SUSPEND)) + return PA_HOOK_OK; + + if (sink && !(sink->flags & PA_SINK_CAN_SUSPEND)) + return PA_HOOK_OK; + d = pa_xnew(struct device_info, 1); d->userdata = u; - d->source = pa_source_isinstance(o) ? pa_source_ref(PA_SOURCE(o)) : NULL; - d->sink = pa_sink_isinstance(o) ? pa_sink_ref(PA_SINK(o)) : NULL; - pa_assert(d->source || d->sink); + d->source = source ? pa_source_ref(source) : NULL; + d->sink = sink ? pa_sink_ref(sink) : NULL; d->time_event = c->mainloop->time_new(c->mainloop, NULL, timeout_cb, d); pa_hashmap_put(u->device_infos, o, d); @@ -273,15 +286,15 @@ static void device_info_free(struct device_info *d) { pa_xfree(d); } -static pa_hook_result_t device_disconnect_hook_cb(pa_core *c, pa_object *o, struct userdata *u) { +static pa_hook_result_t device_unlink_hook_cb(pa_core *c, pa_object *o, struct userdata *u) { struct device_info *d; pa_assert(c); pa_object_assert_ref(o); pa_assert(u); - pa_assert_se((d = pa_hashmap_remove(u->device_infos, o))); - device_info_free(d); + if ((d = pa_hashmap_remove(u->device_infos, o))) + device_info_free(d); return PA_HOOK_OK; } @@ -353,15 +366,15 @@ int pa__init(pa_module*m) { u->sink_new_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_NEW_POST], (pa_hook_cb_t) device_new_hook_cb, u); u->source_new_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_NEW_POST], (pa_hook_cb_t) device_new_hook_cb, u); - u->sink_disconnect_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_DISCONNECT_POST], (pa_hook_cb_t) device_disconnect_hook_cb, u); - u->source_disconnect_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_DISCONNECT_POST], (pa_hook_cb_t) device_disconnect_hook_cb, u); + u->sink_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_UNLINK_POST], (pa_hook_cb_t) device_unlink_hook_cb, u); + u->source_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_UNLINK_POST], (pa_hook_cb_t) device_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) device_state_changed_hook_cb, u); u->source_state_changed_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_STATE_CHANGED], (pa_hook_cb_t) device_state_changed_hook_cb, u); u->sink_input_new_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT], (pa_hook_cb_t) sink_input_new_hook_cb, u); u->source_output_new_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_PUT], (pa_hook_cb_t) source_output_new_hook_cb, u); - u->sink_input_disconnect_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_INPUT_DISCONNECT_POST], (pa_hook_cb_t) sink_input_disconnect_hook_cb, u); - u->source_output_disconnect_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_DISCONNECT_POST], (pa_hook_cb_t) source_output_disconnect_hook_cb, u); + u->sink_input_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_INPUT_UNLINK_POST], (pa_hook_cb_t) sink_input_unlink_hook_cb, u); + u->source_output_unlink_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_UNLINK_POST], (pa_hook_cb_t) source_output_unlink_hook_cb, u); u->sink_input_move_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_INPUT_MOVE], (pa_hook_cb_t) sink_input_move_hook_cb, u); u->source_output_move_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_MOVE], (pa_hook_cb_t) source_output_move_hook_cb, u); u->sink_input_move_post_slot = pa_hook_connect(&m->core->hooks[PA_CORE_HOOK_SINK_INPUT_MOVE_POST], (pa_hook_cb_t) sink_input_move_post_hook_cb, u); @@ -392,22 +405,22 @@ void pa__done(pa_module*m) { if (u->sink_new_slot) pa_hook_slot_free(u->sink_new_slot); - if (u->sink_disconnect_slot) - pa_hook_slot_free(u->sink_disconnect_slot); + if (u->sink_unlink_slot) + pa_hook_slot_free(u->sink_unlink_slot); if (u->sink_state_changed_slot) pa_hook_slot_free(u->sink_state_changed_slot); if (u->source_new_slot) pa_hook_slot_free(u->source_new_slot); - if (u->source_disconnect_slot) - pa_hook_slot_free(u->source_disconnect_slot); + if (u->source_unlink_slot) + pa_hook_slot_free(u->source_unlink_slot); if (u->source_state_changed_slot) pa_hook_slot_free(u->source_state_changed_slot); if (u->sink_input_new_slot) pa_hook_slot_free(u->sink_input_new_slot); - if (u->sink_input_disconnect_slot) - pa_hook_slot_free(u->sink_input_disconnect_slot); + if (u->sink_input_unlink_slot) + pa_hook_slot_free(u->sink_input_unlink_slot); if (u->sink_input_move_slot) pa_hook_slot_free(u->sink_input_move_slot); if (u->sink_input_move_post_slot) @@ -415,8 +428,8 @@ void pa__done(pa_module*m) { if (u->source_output_new_slot) pa_hook_slot_free(u->source_output_new_slot); - if (u->source_output_disconnect_slot) - pa_hook_slot_free(u->source_output_disconnect_slot); + if (u->source_output_unlink_slot) + pa_hook_slot_free(u->source_output_unlink_slot); if (u->source_output_move_slot) pa_hook_slot_free(u->source_output_move_slot); if (u->source_output_move_post_slot) diff --git a/src/pulsecore/cli-text.c b/src/pulsecore/cli-text.c index c919e46d..617ae812 100644 --- a/src/pulsecore/cli-text.c +++ b/src/pulsecore/cli-text.c @@ -97,7 +97,7 @@ char *pa_sink_list_to_string(pa_core *c) { [PA_SINK_RUNNING] = "RUNNING", [PA_SINK_SUSPENDED] = "SUSPENDED", [PA_SINK_IDLE] = "IDLE", - [PA_SINK_DISCONNECTED] = "DISCONNECTED" + [PA_SINK_UNLINKED] = "UNLINKED" }; assert(c); @@ -114,7 +114,7 @@ char *pa_sink_list_to_string(pa_core *c) { " %c index: %u\n" "\tname: <%s>\n" "\tdriver: <%s>\n" - "\tis hardware: <%i>\n" + "\tflags: %s%s%s%s\n" "\tstate: %s\n" "\tvolume: <%s>\n" "\tmute: <%i>\n" @@ -127,7 +127,10 @@ char *pa_sink_list_to_string(pa_core *c) { sink->index, sink->name, sink->driver, - !!sink->is_hardware, + sink->flags & PA_SINK_HW_VOLUME_CTRL ? "HW_VOLUME_CTRL " : "", + sink->flags & PA_SINK_LATENCY ? "LATENCY " : "", + sink->flags & PA_SINK_HARDWARE ? "HARDWARE " : "", + sink->flags & PA_SINK_CAN_SUSPEND ? "CAN_SUSPEND " : "", state_table[pa_sink_get_state(sink)], pa_cvolume_snprint(cv, sizeof(cv), pa_sink_get_volume(sink)), !!pa_sink_get_mute(sink), @@ -154,7 +157,7 @@ char *pa_source_list_to_string(pa_core *c) { [PA_SOURCE_RUNNING] = "RUNNING", [PA_SOURCE_SUSPENDED] = "SUSPENDED", [PA_SOURCE_IDLE] = "IDLE", - [PA_SOURCE_DISCONNECTED] = "DISCONNECTED" + [PA_SOURCE_UNLINKED] = "UNLINKED" }; assert(c); @@ -172,7 +175,7 @@ char *pa_source_list_to_string(pa_core *c) { " %c index: %u\n" "\tname: <%s>\n" "\tdriver: <%s>\n" - "\tis hardware: <%i>\n" + "\tflags: %s%s%s%s\n" "\tstate: %s\n" "\tvolume: <%s>\n" "\tmute: <%u>\n" @@ -184,7 +187,10 @@ char *pa_source_list_to_string(pa_core *c) { source->index, source->name, source->driver, - !!source->is_hardware, + source->flags & PA_SOURCE_HW_VOLUME_CTRL ? "HW_VOLUME_CTRL " : "", + source->flags & PA_SOURCE_LATENCY ? "LATENCY " : "", + source->flags & PA_SOURCE_HARDWARE ? "HARDWARE " : "", + source->flags & PA_SOURCE_CAN_SUSPEND ? "CAN_SUSPEND " : "", state_table[pa_source_get_state(source)], pa_cvolume_snprint(cv, sizeof(cv), pa_source_get_volume(source)), !!pa_source_get_mute(source), @@ -212,7 +218,7 @@ char *pa_source_output_list_to_string(pa_core *c) { static const char* const state_table[] = { [PA_SOURCE_OUTPUT_RUNNING] = "RUNNING", [PA_SOURCE_OUTPUT_CORKED] = "CORKED", - [PA_SOURCE_OUTPUT_DISCONNECTED] = "DISCONNECTED" + [PA_SOURCE_OUTPUT_UNLINKED] = "UNLINKED" }; assert(c); @@ -231,6 +237,7 @@ char *pa_source_output_list_to_string(pa_core *c) { " index: %u\n" "\tname: '%s'\n" "\tdriver: <%s>\n" + "\tflags: %s%s\n" "\tstate: %s\n" "\tsource: <%u> '%s'\n" "\tlatency: <%0.0f usec>\n" @@ -240,6 +247,8 @@ char *pa_source_output_list_to_string(pa_core *c) { o->index, o->name, o->driver, + o->flags & PA_SOURCE_OUTPUT_VARIABLE_RATE ? "VARIABLE_RATE " : "", + o->flags & PA_SOURCE_OUTPUT_DONT_MOVE ? "DONT_MOVE " : "", state_table[pa_source_output_get_state(o)], o->source->index, o->source->name, (double) pa_source_output_get_latency(o), @@ -263,7 +272,7 @@ char *pa_sink_input_list_to_string(pa_core *c) { [PA_SINK_INPUT_RUNNING] = "RUNNING", [PA_SINK_INPUT_DRAINED] = "DRAINED", [PA_SINK_INPUT_CORKED] = "CORKED", - [PA_SINK_INPUT_DISCONNECTED] = "DISCONNECTED" + [PA_SINK_INPUT_UNLINKED] = "UNLINKED" }; assert(c); @@ -282,6 +291,7 @@ char *pa_sink_input_list_to_string(pa_core *c) { " index: %u\n" "\tname: <%s>\n" "\tdriver: <%s>\n" + "\tflags: %s%s\n" "\tstate: %s\n" "\tsink: <%u> '%s'\n" "\tvolume: <%s>\n" @@ -293,6 +303,8 @@ char *pa_sink_input_list_to_string(pa_core *c) { i->index, i->name, i->driver, + i->flags & PA_SINK_INPUT_VARIABLE_RATE ? "VARIABLE_RATE " : "", + i->flags & PA_SINK_INPUT_DONT_MOVE ? "DONT_MOVE " : "", state_table[pa_sink_input_get_state(i)], i->sink->index, i->sink->name, pa_cvolume_snprint(cv, sizeof(cv), pa_sink_input_get_volume(i)), diff --git a/src/pulsecore/core.c b/src/pulsecore/core.c index 7ca2e753..808bc9a8 100644 --- a/src/pulsecore/core.c +++ b/src/pulsecore/core.c @@ -133,7 +133,7 @@ pa_core* pa_core_new(pa_mainloop_api *m, int shared) { c->module_idle_time = 20; c->scache_idle_time = 20; - c->resample_method = PA_RESAMPLER_SRC_SINC_FASTEST; + c->resample_method = PA_RESAMPLER_SPEEX_FLOAT_BASE; c->is_system_instance = 0; c->disallow_module_loading = 0; diff --git a/src/pulsecore/core.h b/src/pulsecore/core.h index 61cb47d5..c8d75800 100644 --- a/src/pulsecore/core.h +++ b/src/pulsecore/core.h @@ -44,23 +44,23 @@ typedef struct pa_core pa_core; typedef enum pa_core_hook { PA_CORE_HOOK_SINK_NEW_POST, - PA_CORE_HOOK_SINK_DISCONNECT, - PA_CORE_HOOK_SINK_DISCONNECT_POST, + PA_CORE_HOOK_SINK_UNLINK, + PA_CORE_HOOK_SINK_UNLINK_POST, PA_CORE_HOOK_SINK_STATE_CHANGED, PA_CORE_HOOK_SOURCE_NEW_POST, - PA_CORE_HOOK_SOURCE_DISCONNECT, - PA_CORE_HOOK_SOURCE_DISCONNECT_POST, + PA_CORE_HOOK_SOURCE_UNLINK, + PA_CORE_HOOK_SOURCE_UNLINK_POST, PA_CORE_HOOK_SOURCE_STATE_CHANGED, PA_CORE_HOOK_SINK_INPUT_NEW, PA_CORE_HOOK_SINK_INPUT_PUT, - PA_CORE_HOOK_SINK_INPUT_DISCONNECT, - PA_CORE_HOOK_SINK_INPUT_DISCONNECT_POST, + PA_CORE_HOOK_SINK_INPUT_UNLINK, + PA_CORE_HOOK_SINK_INPUT_UNLINK_POST, PA_CORE_HOOK_SINK_INPUT_MOVE, PA_CORE_HOOK_SINK_INPUT_MOVE_POST, PA_CORE_HOOK_SOURCE_OUTPUT_NEW, PA_CORE_HOOK_SOURCE_OUTPUT_PUT, - PA_CORE_HOOK_SOURCE_OUTPUT_DISCONNECT, - PA_CORE_HOOK_SOURCE_OUTPUT_DISCONNECT_POST, + PA_CORE_HOOK_SOURCE_OUTPUT_UNLINK, + PA_CORE_HOOK_SOURCE_OUTPUT_UNLINK_POST, PA_CORE_HOOK_SOURCE_OUTPUT_MOVE, PA_CORE_HOOK_SOURCE_OUTPUT_MOVE_POST, PA_CORE_HOOK_MAX diff --git a/src/pulsecore/play-memblockq.c b/src/pulsecore/play-memblockq.c index 0d7efc5c..68ce21b1 100644 --- a/src/pulsecore/play-memblockq.c +++ b/src/pulsecore/play-memblockq.c @@ -58,7 +58,7 @@ static void memblockq_stream_unlink(memblockq_stream *u) { if (!u->sink_input) return; - pa_sink_input_disconnect(u->sink_input); + pa_sink_input_unlink(u->sink_input); pa_sink_input_unref(u->sink_input); u->sink_input = NULL; diff --git a/src/pulsecore/play-memchunk.c b/src/pulsecore/play-memchunk.c index f2925d1b..3ebbc14c 100644 --- a/src/pulsecore/play-memchunk.c +++ b/src/pulsecore/play-memchunk.c @@ -58,7 +58,7 @@ static void memchunk_stream_unlink(memchunk_stream *u) { if (!u->sink_input) return; - pa_sink_input_disconnect(u->sink_input); + pa_sink_input_unlink(u->sink_input); pa_sink_input_unref(u->sink_input); u->sink_input = NULL; diff --git a/src/pulsecore/protocol-esound.c b/src/pulsecore/protocol-esound.c index c5505e57..fc881c81 100644 --- a/src/pulsecore/protocol-esound.c +++ b/src/pulsecore/protocol-esound.c @@ -212,13 +212,13 @@ static void connection_unlink(connection *c) { return; if (c->sink_input) { - pa_sink_input_disconnect(c->sink_input); + pa_sink_input_unlink(c->sink_input); pa_sink_input_unref(c->sink_input); c->sink_input = NULL; } if (c->source_output) { - pa_source_output_disconnect(c->source_output); + pa_source_output_unlink(c->source_output); pa_source_output_unref(c->source_output); c->source_output = NULL; } diff --git a/src/pulsecore/protocol-native.c b/src/pulsecore/protocol-native.c index a0b70669..ea7f43ca 100644 --- a/src/pulsecore/protocol-native.c +++ b/src/pulsecore/protocol-native.c @@ -387,7 +387,7 @@ static void record_stream_unlink(record_stream *s) { return; if (s->source_output) { - pa_source_output_disconnect(s->source_output); + pa_source_output_unlink(s->source_output); pa_source_output_unref(s->source_output); s->source_output = NULL; } @@ -460,11 +460,10 @@ static record_stream* record_stream_new( data.source = source; data.driver = __FILE__; data.name = name; - data.start_corked = corked; pa_source_output_new_data_set_sample_spec(&data, ss); pa_source_output_new_data_set_channel_map(&data, map); - if (!(source_output = pa_source_output_new(c->protocol->core, &data, 0))) + if (!(source_output = pa_source_output_new(c->protocol->core, &data, corked ? PA_SOURCE_OUTPUT_START_CORKED : 0))) return NULL; s = pa_msgobject_new(record_stream); @@ -504,7 +503,7 @@ static void playback_stream_unlink(playback_stream *s) { return; if (s->sink_input) { - pa_sink_input_disconnect(s->sink_input); + pa_sink_input_unlink(s->sink_input); pa_sink_input_unref(s->sink_input); s->sink_input = NULL; } @@ -653,10 +652,9 @@ static playback_stream* playback_stream_new( pa_sink_input_new_data_set_volume(&data, volume); data.module = c->protocol->module; data.client = c->client; - data.start_corked = corked; data.sync_base = ssync ? ssync->sink_input : NULL; - if (!(sink_input = pa_sink_input_new(c->protocol->core, &data, 0))) + if (!(sink_input = pa_sink_input_new(c->protocol->core, &data, corked ? PA_SINK_INPUT_START_CORKED : 0))) return NULL; s = pa_msgobject_new(playback_stream); @@ -1729,10 +1727,7 @@ static void sink_fill_tagstruct(pa_tagstruct *t, pa_sink *sink) { PA_TAG_STRING, sink->monitor_source ? sink->monitor_source->name : NULL, PA_TAG_USEC, pa_sink_get_latency(sink), PA_TAG_STRING, sink->driver, - PA_TAG_U32, - (sink->get_volume ? PA_SINK_HW_VOLUME_CTRL : 0) | /* FIXME */ - (sink->get_latency ? PA_SINK_LATENCY : 0) | /* FIXME */ - (sink->is_hardware ? PA_SINK_HARDWARE : 0), + PA_TAG_U32, sink->flags, PA_TAG_INVALID); } @@ -1754,10 +1749,7 @@ static void source_fill_tagstruct(pa_tagstruct *t, pa_source *source) { PA_TAG_STRING, source->monitor_of ? source->monitor_of->name : NULL, PA_TAG_USEC, pa_source_get_latency(source), PA_TAG_STRING, source->driver, - PA_TAG_U32, - (source->get_volume ? PA_SOURCE_HW_VOLUME_CTRL : 0) | /* FIXME */ - (source->get_latency ? PA_SOURCE_LATENCY : 0) | /* FIXME */ - (source->is_hardware ? PA_SOURCE_HARDWARE : 0), + PA_TAG_U32, source->flags, PA_TAG_INVALID); } diff --git a/src/pulsecore/protocol-simple.c b/src/pulsecore/protocol-simple.c index bc48ac37..db7c9a69 100644 --- a/src/pulsecore/protocol-simple.c +++ b/src/pulsecore/protocol-simple.c @@ -110,13 +110,13 @@ static void connection_unlink(connection *c) { return; if (c->sink_input) { - pa_sink_input_disconnect(c->sink_input); + pa_sink_input_unlink(c->sink_input); pa_sink_input_unref(c->sink_input); c->sink_input = NULL; } if (c->source_output) { - pa_source_output_disconnect(c->source_output); + pa_source_output_unlink(c->source_output); pa_source_output_unref(c->source_output); c->source_output = NULL; } diff --git a/src/pulsecore/sink-input.c b/src/pulsecore/sink-input.c index 0cdcbc0b..c33d8e70 100644 --- a/src/pulsecore/sink-input.c +++ b/src/pulsecore/sink-input.c @@ -108,7 +108,7 @@ pa_sink_input* pa_sink_input_new( data->sink = pa_namereg_get(core, NULL, PA_NAMEREG_SINK, 1); pa_return_null_if_fail(data->sink); - pa_return_null_if_fail(pa_sink_get_state(data->sink) != PA_SINK_DISCONNECTED); + pa_return_null_if_fail(pa_sink_get_state(data->sink) != PA_SINK_UNLINKED); pa_return_null_if_fail(!data->sync_base || (data->sync_base->sink == data->sink && pa_sink_input_get_state(data->sync_base) == PA_SINK_INPUT_CORKED)); if (!data->sample_spec_is_set) @@ -167,7 +167,7 @@ pa_sink_input* pa_sink_input_new( i->parent.process_msg = pa_sink_input_process_msg; i->core = core; - i->state = data->start_corked ? PA_SINK_INPUT_CORKED : PA_SINK_INPUT_RUNNING; + i->state = PA_SINK_INPUT_INIT; i->flags = flags; i->name = pa_xstrdup(data->name); i->driver = pa_xstrdup(data->driver); @@ -194,9 +194,12 @@ pa_sink_input* pa_sink_input_new( i->peek = NULL; i->drop = NULL; + i->process = NULL; i->kill = NULL; i->get_latency = NULL; - i->underrun = NULL; + i->attach = NULL; + i->detach = NULL; + i->suspend = NULL; i->userdata = NULL; i->thread_info.state = i->state; @@ -245,11 +248,11 @@ static int sink_input_set_state(pa_sink_input *i, pa_sink_input_state_t state) { return 0; } -void pa_sink_input_disconnect(pa_sink_input *i) { +void pa_sink_input_unlink(pa_sink_input *i) { pa_assert(i); - pa_return_if_fail(i->state != PA_SINK_INPUT_DISCONNECTED); + pa_assert(PA_SINK_INPUT_LINKED(i->state)); - pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_DISCONNECT], i); + pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_UNLINK], i); if (i->sync_prev) i->sync_prev->sync_next = i->sync_next; @@ -264,16 +267,19 @@ void pa_sink_input_disconnect(pa_sink_input *i) { pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_REMOVE, i->index); - sink_input_set_state(i, PA_SINK_INPUT_DISCONNECTED); + sink_input_set_state(i, PA_SINK_INPUT_UNLINKED); pa_sink_update_status(i->sink); i->peek = NULL; i->drop = NULL; + i->process = NULL; i->kill = NULL; i->get_latency = NULL; - i->underrun = NULL; + i->attach = NULL; + i->detach = NULL; + i->suspend = NULL; - pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_DISCONNECT_POST], i); + pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_UNLINK_POST], i); i->sink = NULL; pa_sink_input_unref(i); } @@ -284,8 +290,8 @@ static void sink_input_free(pa_object *o) { pa_assert(i); pa_assert(pa_sink_input_refcnt(i) == 0); - if (i->state != PA_SINK_INPUT_DISCONNECTED) - pa_sink_input_disconnect(i); + if (PA_SINK_INPUT_LINKED(i->state)) + pa_sink_input_unlink(i); pa_log_info("Freeing output %u \"%s\"", i->index, i->name); @@ -306,6 +312,11 @@ static void sink_input_free(pa_object *o) { void pa_sink_input_put(pa_sink_input *i) { pa_sink_input_assert_ref(i); + pa_assert(i->state == PA_SINK_INPUT_INIT); + pa_assert(i->peek); + pa_assert(i->drop); + + i->thread_info.state = i->state = i->flags & PA_SINK_INPUT_START_CORKED ? PA_SINK_INPUT_CORKED : PA_SINK_INPUT_RUNNING; i->thread_info.volume = i->volume; i->thread_info.muted = i->muted; @@ -314,10 +325,15 @@ void pa_sink_input_put(pa_sink_input *i) { 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); + + /* Please note that if you change something here, you have to + change something in pa_sink_input_move() with the ghost stream + registration too. */ } void pa_sink_input_kill(pa_sink_input*i) { pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->state)); if (i->kill) i->kill(i); @@ -327,6 +343,7 @@ pa_usec_t pa_sink_input_get_latency(pa_sink_input *i) { pa_usec_t r = 0; pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_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; @@ -344,10 +361,11 @@ int pa_sink_input_peek(pa_sink_input *i, pa_memchunk *chunk, pa_cvolume *volume) int volume_is_norm; pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->thread_info.state)); pa_assert(chunk); pa_assert(volume); - if (!i->peek || !i->drop || i->thread_info.state == PA_SINK_INPUT_DISCONNECTED || i->thread_info.state == PA_SINK_INPUT_CORKED) + if (!i->peek || !i->drop || i->thread_info.state == PA_SINK_INPUT_UNLINKED || i->thread_info.state == PA_SINK_INPUT_CORKED) goto finish; pa_assert(i->thread_info.state == PA_SINK_INPUT_RUNNING || i->thread_info.state == PA_SINK_INPUT_DRAINED); @@ -420,9 +438,6 @@ int pa_sink_input_peek(pa_sink_input *i, pa_memchunk *chunk, pa_cvolume *volume) finish: - if (ret < 0 && !pa_atomic_load(&i->thread_info.drained) && i->underrun) - i->underrun(i); - if (ret >= 0) pa_atomic_store(&i->thread_info.drained, 0); else if (ret < 0) @@ -448,13 +463,18 @@ finish: /* Called from thread context */ void pa_sink_input_drop(pa_sink_input *i, size_t length) { pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->thread_info.state)); pa_assert(length > 0); if (i->thread_info.move_silence > 0) { - pa_assert(i->thread_info.move_silence >= length); - - i->thread_info.move_silence -= length; + if (i->thread_info.move_silence >= length) { + i->thread_info.move_silence -= length; + length = 0; + } else { + length -= i->thread_info.move_silence; + i->thread_info.move_silence = 0; + } if (i->thread_info.move_silence <= 0) { pa_assert(i->thread_info.silence_memblock); @@ -462,7 +482,8 @@ void pa_sink_input_drop(pa_sink_input *i, size_t length) { i->thread_info.silence_memblock = NULL; } - return; + if (length <= 0) + return; } if (i->thread_info.resampled_chunk.memblock) { @@ -531,6 +552,7 @@ void pa_sink_input_drop(pa_sink_input *i, size_t length) { 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_LINKED(i->state)); if (pa_cvolume_equal(&i->volume, volume)) return; @@ -543,6 +565,7 @@ 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) { pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->state)); return &i->volume; } @@ -550,6 +573,7 @@ const pa_cvolume *pa_sink_input_get_volume(pa_sink_input *i) { void pa_sink_input_set_mute(pa_sink_input *i, int mute) { pa_assert(i); pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->state)); if (!i->muted == !mute) return; @@ -562,18 +586,21 @@ void pa_sink_input_set_mute(pa_sink_input *i, int mute) { int pa_sink_input_get_mute(pa_sink_input *i) { pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->state)); return !!i->muted; } void pa_sink_input_cork(pa_sink_input *i, int b) { pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->state)); sink_input_set_state(i, b ? PA_SINK_INPUT_CORKED : PA_SINK_INPUT_RUNNING); } int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate) { pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->state)); pa_return_val_if_fail(i->thread_info.resampler, -1); if (i->sample_spec.rate == rate) @@ -615,6 +642,7 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, int immediately) { pa_sink_input_move_info info; pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->state)); pa_sink_assert_ref(dest); origin = i->sink; @@ -622,6 +650,9 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, int immediately) { if (dest == origin) return 0; + if (i->flags & PA_SINK_INPUT_DONT_MOVE) + return -1; + if (i->sync_next || i->sync_prev) { pa_log_warn("Moving synchronised streams not supported."); return -1; @@ -714,6 +745,10 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, int immediately) { if (info.ghost_sink_input) { /* Basically, do what pa_sink_input_put() does ...*/ + info.ghost_sink_input->thread_info.state = info.ghost_sink_input->state = PA_SINK_INPUT_RUNNING; + info.ghost_sink_input->thread_info.volume = info.ghost_sink_input->volume; + info.ghost_sink_input->thread_info.muted = info.ghost_sink_input->muted; + pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_NEW, info.ghost_sink_input->index); pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT], info.ghost_sink_input); pa_sink_input_unref(info.ghost_sink_input); @@ -774,6 +809,7 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t pa_sink_input *i = PA_SINK_INPUT(o); pa_sink_input_assert_ref(i); + pa_assert(PA_SINK_INPUT_LINKED(i->thread_info.state)); switch (code) { case PA_SINK_INPUT_MESSAGE_SET_VOLUME: @@ -841,3 +877,4 @@ pa_sink_input_state_t pa_sink_input_get_state(pa_sink_input *i) { return i->state; } + diff --git a/src/pulsecore/sink-input.h b/src/pulsecore/sink-input.h index d728d462..e1d89ffb 100644 --- a/src/pulsecore/sink-input.h +++ b/src/pulsecore/sink-input.h @@ -39,14 +39,21 @@ typedef struct pa_sink_input pa_sink_input; #include <pulsecore/core.h> typedef enum pa_sink_input_state { + PA_SINK_INPUT_INIT, /*< The stream is not active yet, because pa_sink_put() has not been called yet */ PA_SINK_INPUT_DRAINED, /*< The stream stopped playing because there was no data to play */ PA_SINK_INPUT_RUNNING, /*< The stream is alive and kicking */ PA_SINK_INPUT_CORKED, /*< The stream was corked on user request */ - PA_SINK_INPUT_DISCONNECTED /*< The stream is dead */ + PA_SINK_INPUT_UNLINKED /*< The stream is dead */ } pa_sink_input_state_t; +static inline int PA_SINK_INPUT_LINKED(pa_sink_input_state_t x) { + return x == PA_SINK_INPUT_DRAINED || x == PA_SINK_INPUT_RUNNING || x == PA_SINK_INPUT_CORKED; +} + typedef enum pa_sink_input_flags { PA_SINK_INPUT_VARIABLE_RATE = 1, + PA_SINK_INPUT_DONT_MOVE = 2, + PA_SINK_INPUT_START_CORKED = 4 } pa_sink_input_flags_t; struct pa_sink_input { @@ -75,11 +82,42 @@ struct pa_sink_input { pa_cvolume volume; int muted; + /* Returns the chunk of audio data (but doesn't drop it + * yet!). Returns -1 on failure. Called from IO thread context. */ int (*peek) (pa_sink_input *i, pa_memchunk *chunk); + + /* Drops the specified number of bytes, usually called right after + * peek(), but not necessarily. Called from IO thread context. */ void (*drop) (pa_sink_input *i, size_t length); + + /* If non-NULL this function is called in each IO event loop and + * can be used to do additional processing even when the device is + * suspended and peek() is never called. Should return 1 when + * "some work" has been done and the IO event loop should be + * reiterated immediately. Called from IO thread context. */ + int (*process) (pa_sink_input *i); /* may be NULL */ + + /* If non-NULL this function is called when the input is first + * connected to a sink. Called from IO thread context */ + void (*attach) (pa_sink_input *i); /* may be NULL */ + + /* If non-NULL this function is called when the output is + * disconnected from its sink. Called from IO thread context */ + void (*detach) (pa_sink_input *i); /* may be NULL */ + + /* If non-NULL called whenever the the sink this input is attached + * to suspends or resumes. Called from main context */ + void (*suspend) (pa_sink_input *i, int b); /* may be NULL */ + + /* Supposed to unlink and destroy this stream. Called from main + * context. */ void (*kill) (pa_sink_input *i); /* may 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. */ pa_usec_t (*get_latency) (pa_sink_input *i); /* may be NULL */ - void (*underrun) (pa_sink_input *i); /* may be NULL */ pa_resample_method_t resample_method; @@ -138,7 +176,6 @@ typedef struct pa_sink_input_new_data { pa_resample_method_t resample_method; - int start_corked; pa_sink_input *sync_base; } pa_sink_input_new_data; @@ -156,7 +193,7 @@ pa_sink_input* pa_sink_input_new( pa_sink_input_flags_t flags); void pa_sink_input_put(pa_sink_input *i); -void pa_sink_input_disconnect(pa_sink_input* i); +void pa_sink_input_unlink(pa_sink_input* i); void pa_sink_input_set_name(pa_sink_input *i, const char *name); diff --git a/src/pulsecore/sink.c b/src/pulsecore/sink.c index 2c7c661c..b61a8507 100644 --- a/src/pulsecore/sink.c +++ b/src/pulsecore/sink.c @@ -92,7 +92,8 @@ pa_sink* pa_sink_new( s->parent.process_msg = pa_sink_process_msg; s->core = core; - s->state = PA_SINK_IDLE; + s->state = PA_SINK_INIT; + s->flags = 0; s->name = pa_xstrdup(name); s->description = NULL; s->driver = pa_xstrdup(driver); @@ -107,8 +108,6 @@ pa_sink* pa_sink_new( s->muted = 0; s->refresh_volume = s->refresh_mute = 0; - s->is_hardware = 0; - s->get_latency = NULL; s->set_volume = NULL; s->get_volume = NULL; @@ -118,6 +117,7 @@ pa_sink* pa_sink_new( s->userdata = NULL; s->asyncmsgq = NULL; + s->rtpoll = NULL; s->silence = NULL; r = pa_idxset_put(core->sinks, s, &s->index); @@ -145,11 +145,22 @@ pa_sink* pa_sink_new( s->thread_info.soft_muted = s->muted; s->thread_info.state = s->state; - pa_subscription_post(core, PA_SUBSCRIPTION_EVENT_SINK | PA_SUBSCRIPTION_EVENT_NEW, s->index); + return s; +} - pa_hook_fire(&core->hooks[PA_CORE_HOOK_SINK_NEW_POST], s); +void pa_sink_put(pa_sink* s) { + pa_sink_assert_ref(s); - return s; + pa_assert(s->state == PA_SINK_INIT); + pa_assert(s->asyncmsgq); + pa_assert(s->rtpoll); + + s->thread_info.state = s->state = PA_SINK_IDLE; + + pa_source_put(s->monitor_source); + + pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK | PA_SUBSCRIPTION_EVENT_NEW, s->index); + pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SINK_NEW_POST], s); } static int sink_set_state(pa_sink *s, pa_sink_state_t state) { @@ -160,6 +171,21 @@ static int sink_set_state(pa_sink *s, pa_sink_state_t state) { if (s->state == state) return 0; + if (state == PA_SINK_SUSPENDED && !(s->flags & PA_SINK_CAN_SUSPEND)) + return -1; + + if ((s->state == PA_SINK_SUSPENDED && PA_SINK_OPENED(state)) || + (PA_SINK_OPENED(s->state) && state == PA_SINK_SUSPENDED)) { + pa_sink_input *i; + uint32_t idx; + + /* We're suspending or resuming, tell everyone about it */ + + for (i = PA_SINK_INPUT(pa_idxset_first(s->inputs, &idx)); i; i = PA_SINK_INPUT(pa_idxset_next(s->inputs, &idx))) + if (i->suspend) + i->suspend(i, state == PA_SINK_SUSPENDED); + } + if (s->set_state) if ((ret = s->set_state(s, state)) < 0) return -1; @@ -168,18 +194,19 @@ static int sink_set_state(pa_sink *s, pa_sink_state_t state) { return -1; s->state = state; - - pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SINK_STATE_CHANGED], s); + + if (state != PA_SINK_UNLINKED) /* if we enter UNLINKED state pa_sink_unlink() will fire the apropriate events */ + pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SINK_STATE_CHANGED], s); return 0; } -void pa_sink_disconnect(pa_sink* s) { +void pa_sink_unlink(pa_sink* s) { pa_sink_input *i, *j = NULL; pa_assert(s); - pa_return_if_fail(s->state != PA_SINK_DISCONNECTED); + pa_assert(PA_SINK_LINKED(s->state)); - pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SINK_DISCONNECT], s); + pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SINK_UNLINK], s); pa_namereg_unregister(s->core, s->name); pa_idxset_remove_by_data(s->core->sinks, s, NULL); @@ -191,9 +218,9 @@ void pa_sink_disconnect(pa_sink* s) { } if (s->monitor_source) - pa_source_disconnect(s->monitor_source); + pa_source_unlink(s->monitor_source); - sink_set_state(s, PA_SINK_DISCONNECTED); + sink_set_state(s, PA_SINK_UNLINKED); s->get_latency = NULL; s->get_volume = NULL; @@ -204,7 +231,7 @@ void pa_sink_disconnect(pa_sink* s) { pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK | PA_SUBSCRIPTION_EVENT_REMOVE, s->index); - pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SINK_DISCONNECT_POST], s); + pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SINK_UNLINK_POST], s); } static void sink_free(pa_object *o) { @@ -214,8 +241,8 @@ static void sink_free(pa_object *o) { pa_assert(s); pa_assert(pa_sink_refcnt(s) == 0); - if (s->state != PA_SINK_DISCONNECTED) - pa_sink_disconnect(s); + if (PA_SINK_LINKED(s->state)) + pa_sink_unlink(s); pa_log_info("Freeing sink %u \"%s\"", s->index, s->name); @@ -250,8 +277,18 @@ void pa_sink_set_asyncmsgq(pa_sink *s, pa_asyncmsgq *q) { pa_source_set_asyncmsgq(s->monitor_source, q); } +void pa_sink_set_rtpoll(pa_sink *s, pa_rtpoll *p) { + pa_sink_assert_ref(s); + pa_assert(p); + + s->rtpoll = p; + if (s->monitor_source) + pa_source_set_rtpoll(s->monitor_source, p); +} + int pa_sink_update_status(pa_sink*s) { pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); if (s->state == PA_SINK_SUSPENDED) return 0; @@ -261,6 +298,7 @@ int pa_sink_update_status(pa_sink*s) { int pa_sink_suspend(pa_sink *s, int suspend) { pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); if (suspend) return sink_set_state(s, PA_SINK_SUSPENDED); @@ -270,6 +308,7 @@ int pa_sink_suspend(pa_sink *s, int suspend) { void pa_sink_ping(pa_sink *s) { pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_PING, NULL, 0, NULL, NULL); } @@ -364,6 +403,7 @@ void pa_sink_render(pa_sink*s, size_t length, pa_memchunk *result) { unsigned n; pa_sink_assert_ref(s); + pa_assert(PA_SINK_OPENED(s->thread_info.state)); pa_assert(length); pa_assert(result); @@ -429,6 +469,7 @@ void pa_sink_render_into(pa_sink*s, pa_memchunk *target) { unsigned n; pa_sink_assert_ref(s); + pa_assert(PA_SINK_OPENED(s->thread_info.state)); pa_assert(target); pa_assert(target->memblock); pa_assert(target->length); @@ -494,6 +535,7 @@ void pa_sink_render_into_full(pa_sink *s, pa_memchunk *target) { size_t l, d; pa_sink_assert_ref(s); + pa_assert(PA_SINK_OPENED(s->thread_info.state)); pa_assert(target); pa_assert(target->memblock); pa_assert(target->length); @@ -518,6 +560,7 @@ void pa_sink_render_into_full(pa_sink *s, pa_memchunk *target) { void pa_sink_render_full(pa_sink *s, size_t length, pa_memchunk *result) { pa_sink_assert_ref(s); + pa_assert(PA_SINK_OPENED(s->thread_info.state)); pa_assert(length); pa_assert(result); @@ -529,10 +572,46 @@ void pa_sink_render_full(pa_sink *s, size_t length, pa_memchunk *result) { pa_sink_render_into_full(s, result); } +void pa_sink_skip(pa_sink *s, size_t length) { + pa_sink_input *i; + void *state = NULL; + + pa_sink_assert_ref(s); + pa_assert(PA_SINK_OPENED(s->thread_info.state)); + pa_assert(length > 0); + + if (pa_source_used_by(s->monitor_source)) { + pa_memchunk chunk; + + /* If something is connected to our monitor source, we have to + * pass valid data to it */ + + while (length > 0) { + pa_sink_render(s, length, &chunk); + pa_memblock_unref(chunk.memblock); + + pa_assert(chunk.length <= length); + length -= chunk.length; + } + + } else { + /* Ok, noone cares about the rendered data, so let's not even render it */ + + while ((i = pa_hashmap_iterate(s->thread_info.inputs, &state, NULL))) { + pa_sink_input_assert_ref(i); + pa_sink_input_drop(i, length); + } + } +} + pa_usec_t pa_sink_get_latency(pa_sink *s) { pa_usec_t usec = 0; pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); + + if (!PA_SINK_OPENED(s->state)) + return 0; if (s->get_latency) return s->get_latency(s); @@ -547,6 +626,7 @@ void pa_sink_set_volume(pa_sink *s, const pa_cvolume *volume) { int changed; pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); pa_assert(volume); changed = !pa_cvolume_equal(volume, &s->volume); @@ -566,6 +646,7 @@ const pa_cvolume *pa_sink_get_volume(pa_sink *s) { struct pa_cvolume old_volume; pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); old_volume = s->volume; @@ -585,6 +666,7 @@ void pa_sink_set_mute(pa_sink *s, int mute) { int changed; pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); changed = s->muted != mute; s->muted = mute; @@ -603,6 +685,7 @@ int pa_sink_get_mute(pa_sink *s) { int old_muted; pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); old_muted = s->muted; @@ -659,6 +742,7 @@ unsigned pa_sink_used_by(pa_sink *s) { unsigned ret; pa_sink_assert_ref(s); + pa_assert(PA_SINK_LINKED(s->state)); ret = pa_idxset_size(s->inputs); @@ -671,6 +755,7 @@ unsigned pa_sink_used_by(pa_sink *s) { 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(PA_SINK_LINKED(s->thread_info.state)); switch ((pa_sink_message_t) code) { @@ -694,13 +779,19 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse i->thread_info.sync_next->thread_info.sync_prev = i; } + if (i->attach) + i->attach(i); + return 0; } case PA_SINK_MESSAGE_REMOVE_INPUT: { pa_sink_input *i = PA_SINK_INPUT(userdata); - /* Since the caller sleeps in pa_sink_input_disconnect(), + if (i->detach) + i->detach(i); + + /* Since the caller sleeps in pa_sink_input_unlink(), * we can safely access data outside of thread_info even * though it is mutable */ @@ -833,3 +924,21 @@ int pa_sink_suspend_all(pa_core *c, int suspend) { return ret; } + +int pa_sink_process_inputs(pa_sink *s) { + pa_sink_input *i; + void *state = NULL; + int r; + + pa_sink_assert_ref(s); + + if (!PA_SINK_LINKED(s->thread_info.state)) + return 0; + + while ((i = PA_SINK_INPUT(pa_hashmap_iterate(s->thread_info.inputs, &state, NULL)))) + if (i->process) + if ((r = i->process(i))) + return r; + + return 0; +} diff --git a/src/pulsecore/sink.h b/src/pulsecore/sink.h index fadc76eb..b3fcff49 100644 --- a/src/pulsecore/sink.h +++ b/src/pulsecore/sink.h @@ -40,26 +40,36 @@ typedef struct pa_sink pa_sink; #include <pulsecore/module.h> #include <pulsecore/refcnt.h> #include <pulsecore/msgobject.h> +#include <pulsecore/rtpoll.h> #define PA_MAX_INPUTS_PER_SINK 32 typedef enum pa_sink_state { + PA_SINK_INIT, PA_SINK_RUNNING, PA_SINK_SUSPENDED, PA_SINK_IDLE, - PA_SINK_DISCONNECTED + PA_SINK_UNLINKED } pa_sink_state_t; +static inline int PA_SINK_OPENED(pa_sink_state_t x) { + return x == PA_SINK_RUNNING || x == PA_SINK_IDLE; +} + +static inline int PA_SINK_LINKED(pa_sink_state_t x) { + return x == PA_SINK_RUNNING || x == PA_SINK_IDLE || x == PA_SINK_SUSPENDED; +} + struct pa_sink { pa_msgobject parent; uint32_t index; pa_core *core; pa_sink_state_t state; + pa_sink_flags_t flags; char *name; char *description, *driver; /* may be NULL */ - int is_hardware; pa_module *module; /* may be NULL */ @@ -67,21 +77,22 @@ struct pa_sink { pa_channel_map channel_map; pa_idxset *inputs; - pa_source *monitor_source; /* may be NULL */ + pa_source *monitor_source; pa_cvolume volume; int muted; int refresh_volume; int refresh_mute; - int (*set_state)(pa_sink *s, pa_sink_state_t state); - int (*set_volume)(pa_sink *s); /* dito */ - int (*get_volume)(pa_sink *s); /* dito */ - int (*get_mute)(pa_sink *s); /* dito */ - int (*set_mute)(pa_sink *s); /* dito */ + int (*set_state)(pa_sink *s, pa_sink_state_t state); /* may be NULL */ + int (*set_volume)(pa_sink *s); /* dito */ + int (*get_volume)(pa_sink *s); /* dito */ + int (*get_mute)(pa_sink *s); /* dito */ + int (*set_mute)(pa_sink *s); /* dito */ pa_usec_t (*get_latency)(pa_sink *s); /* dito */ pa_asyncmsgq *asyncmsgq; + pa_rtpoll *rtpoll; /* Contains copies of the above data so that the real-time worker * thread can work without access locking */ @@ -114,7 +125,7 @@ typedef enum pa_sink_message { PA_SINK_MESSAGE_MAX } pa_sink_message_t; -/* To be used exclusively by the sink driver */ +/* To be called exclusively by the sink driver, from main context */ pa_sink* pa_sink_new( pa_core *core, @@ -124,18 +135,21 @@ pa_sink* pa_sink_new( const pa_sample_spec *spec, const pa_channel_map *map); -void pa_sink_disconnect(pa_sink* s); +void pa_sink_put(pa_sink *s); +void pa_sink_unlink(pa_sink* s); void pa_sink_set_module(pa_sink *sink, pa_module *m); void pa_sink_set_description(pa_sink *s, const char *description); void pa_sink_set_asyncmsgq(pa_sink *s, pa_asyncmsgq *q); +void pa_sink_set_rtpoll(pa_sink *s, pa_rtpoll *p); -/* Usable by everyone */ +/* May be called by everyone, from main context */ pa_usec_t pa_sink_get_latency(pa_sink *s); int pa_sink_update_status(pa_sink*s); int pa_sink_suspend(pa_sink *s, int suspend); +int pa_sink_suspend_all(pa_core *c, int suspend); /* Sends a ping message to the sink thread, to make it wake up and * check for data to process even if there is no real message is @@ -150,19 +164,17 @@ int pa_sink_get_mute(pa_sink *sink); unsigned pa_sink_used_by(pa_sink *s); #define pa_sink_get_state(s) ((s)->state) -/* To be used exclusively by the sink driver thread */ +/* To be called exclusively by the sink driver, from IO context */ void pa_sink_render(pa_sink*s, size_t length, pa_memchunk *result); void pa_sink_render_full(pa_sink *s, size_t length, pa_memchunk *result); void pa_sink_render_into(pa_sink*s, pa_memchunk *target); void pa_sink_render_into_full(pa_sink *s, pa_memchunk *target); -int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk); +void pa_sink_skip(pa_sink *s, size_t length); -static inline int PA_SINK_OPENED(pa_sink_state_t x) { - return x == PA_SINK_RUNNING || x == PA_SINK_IDLE; -} +int pa_sink_process_inputs(pa_sink *s); -int pa_sink_suspend_all(pa_core *c, int suspend); +int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk); #endif diff --git a/src/pulsecore/sound-file-stream.c b/src/pulsecore/sound-file-stream.c index e0d51a7d..36c7f03c 100644 --- a/src/pulsecore/sound-file-stream.c +++ b/src/pulsecore/sound-file-stream.c @@ -69,7 +69,7 @@ static void file_stream_unlink(file_stream *u) { if (!u->sink_input) return; - pa_sink_input_disconnect(u->sink_input); + pa_sink_input_unlink(u->sink_input); pa_sink_input_unref(u->sink_input); u->sink_input = NULL; diff --git a/src/pulsecore/source-output.c b/src/pulsecore/source-output.c index 25322469..ccd968ae 100644 --- a/src/pulsecore/source-output.c +++ b/src/pulsecore/source-output.c @@ -86,7 +86,7 @@ pa_source_output* pa_source_output_new( data->source = pa_namereg_get(core, NULL, PA_NAMEREG_SOURCE, 1); pa_return_null_if_fail(data->source); - pa_return_null_if_fail(pa_source_get_state(data->source) != PA_SOURCE_DISCONNECTED); + pa_return_null_if_fail(pa_source_get_state(data->source) != PA_SOURCE_UNLINKED); if (!data->sample_spec_is_set) data->sample_spec = data->source->sample_spec; @@ -135,7 +135,7 @@ pa_source_output* pa_source_output_new( o->parent.process_msg = pa_source_output_process_msg; o->core = core; - o->state = data->start_corked ? PA_SOURCE_OUTPUT_CORKED : PA_SOURCE_OUTPUT_RUNNING; + o->state = PA_SOURCE_OUTPUT_INIT; o->flags = flags; o->name = pa_xstrdup(data->name); o->driver = pa_xstrdup(data->driver); @@ -148,8 +148,12 @@ pa_source_output* pa_source_output_new( o->channel_map = data->channel_map; o->push = NULL; + o->process = NULL; o->kill = NULL; o->get_latency = NULL; + o->detach = NULL; + o->attach = NULL; + o->suspend = NULL; o->userdata = NULL; o->thread_info.state = o->state; @@ -183,11 +187,11 @@ static int source_output_set_state(pa_source_output *o, pa_source_output_state_t return 0; } -void pa_source_output_disconnect(pa_source_output*o) { +void pa_source_output_unlink(pa_source_output*o) { pa_assert(o); - pa_return_if_fail(o->state != PA_SOURCE_OUTPUT_DISCONNECTED); + pa_assert(PA_SOURCE_OUTPUT_LINKED(o->state)); - pa_hook_fire(&o->source->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_DISCONNECT], o); + pa_hook_fire(&o->source->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_UNLINK], o); pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_REMOVE_OUTPUT, o, 0, NULL); @@ -196,14 +200,18 @@ void pa_source_output_disconnect(pa_source_output*o) { pa_subscription_post(o->source->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_REMOVE, o->index); - source_output_set_state(o, PA_SOURCE_OUTPUT_DISCONNECTED); + source_output_set_state(o, PA_SOURCE_OUTPUT_UNLINKED); pa_source_update_status(o->source); o->push = NULL; + o->process = NULL; o->kill = NULL; o->get_latency = NULL; + o->attach = NULL; + o->detach = NULL; + o->suspend = NULL; - pa_hook_fire(&o->source->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_DISCONNECT_POST], o); + pa_hook_fire(&o->source->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_UNLINK_POST], o); o->source = NULL; pa_source_output_unref(o); @@ -214,8 +222,8 @@ static void source_output_free(pa_object* mo) { pa_assert(pa_source_output_refcnt(o) == 0); - if (o->state != PA_SOURCE_OUTPUT_DISCONNECTED) - pa_source_output_disconnect(o); + if (PA_SOURCE_LINKED(o->state)) + pa_source_output_unlink(o); pa_log_info("Freeing output %u \"%s\"", o->index, o->name); @@ -230,6 +238,11 @@ static void source_output_free(pa_object* mo) { void pa_source_output_put(pa_source_output *o) { pa_source_output_assert_ref(o); + pa_assert(o->state == PA_SOURCE_OUTPUT_INIT); + pa_assert(o->push); + + o->thread_info.state = o->state = o->flags & PA_SOURCE_OUTPUT_START_CORKED ? PA_SOURCE_OUTPUT_CORKED : PA_SOURCE_OUTPUT_RUNNING; + pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, o, 0, NULL); pa_source_update_status(o->source); @@ -240,7 +253,8 @@ void pa_source_output_put(pa_source_output *o) { void pa_source_output_kill(pa_source_output*o) { pa_source_output_assert_ref(o); - + pa_assert(PA_SOURCE_OUTPUT_LINKED(o->state)); + if (o->kill) o->kill(o); } @@ -249,6 +263,7 @@ pa_usec_t pa_source_output_get_latency(pa_source_output *o) { pa_usec_t r = 0; pa_source_output_assert_ref(o); + pa_assert(PA_SOURCE_OUTPUT_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; @@ -264,10 +279,11 @@ void pa_source_output_push(pa_source_output *o, const pa_memchunk *chunk) { pa_memchunk rchunk; pa_source_output_assert_ref(o); + pa_assert(PA_SOURCE_OUTPUT_LINKED(o->thread_info.state)); pa_assert(chunk); pa_assert(chunk->length); - if (!o->push || o->state == PA_SOURCE_OUTPUT_DISCONNECTED || o->state == PA_SOURCE_OUTPUT_CORKED) + if (!o->push || o->state == PA_SOURCE_OUTPUT_UNLINKED || o->state == PA_SOURCE_OUTPUT_CORKED) return; pa_assert(o->state == PA_SOURCE_OUTPUT_RUNNING); @@ -288,12 +304,14 @@ void pa_source_output_push(pa_source_output *o, const pa_memchunk *chunk) { void pa_source_output_cork(pa_source_output *o, int b) { pa_source_output_assert_ref(o); + pa_assert(PA_SOURCE_OUTPUT_LINKED(o->state)); source_output_set_state(o, b ? PA_SOURCE_OUTPUT_CORKED : PA_SOURCE_OUTPUT_RUNNING); } int pa_source_output_set_rate(pa_source_output *o, uint32_t rate) { pa_source_output_assert_ref(o); + pa_assert(PA_SOURCE_OUTPUT_LINKED(o->state)); pa_return_val_if_fail(o->thread_info.resampler, -1); if (o->sample_spec.rate == rate) @@ -333,6 +351,7 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) { pa_resampler *new_resampler = NULL; pa_source_output_assert_ref(o); + pa_assert(PA_SOURCE_OUTPUT_LINKED(o->state)); pa_source_assert_ref(dest); origin = o->source; @@ -340,6 +359,9 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) { if (dest == origin) return 0; + if (o->flags & PA_SOURCE_OUTPUT_DONT_MOVE) + return -1; + if (pa_idxset_size(dest->outputs) >= PA_MAX_OUTPUTS_PER_SOURCE) { pa_log_warn("Failed to move source output: too many outputs per source."); return -1; @@ -405,6 +427,7 @@ int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int pa_source_output *o = PA_SOURCE_OUTPUT(mo); pa_source_output_assert_ref(o); + pa_assert(PA_SOURCE_OUTPUT_LINKED(o->thread_info.state)); switch (code) { diff --git a/src/pulsecore/source-output.h b/src/pulsecore/source-output.h index b17adcb5..2027e37a 100644 --- a/src/pulsecore/source-output.h +++ b/src/pulsecore/source-output.h @@ -36,13 +36,20 @@ typedef struct pa_source_output pa_source_output; #include <pulsecore/client.h> typedef enum pa_source_output_state { + PA_SOURCE_OUTPUT_INIT, PA_SOURCE_OUTPUT_RUNNING, PA_SOURCE_OUTPUT_CORKED, - PA_SOURCE_OUTPUT_DISCONNECTED + PA_SOURCE_OUTPUT_UNLINKED } pa_source_output_state_t; +static inline int PA_SOURCE_OUTPUT_LINKED(pa_source_output_state_t x) { + return x == PA_SOURCE_OUTPUT_RUNNING || x == PA_SOURCE_OUTPUT_CORKED; +} + typedef enum pa_source_output_flags { - PA_SOURCE_OUTPUT_VARIABLE_RATE = 1 + PA_SOURCE_OUTPUT_VARIABLE_RATE = 1, + PA_SOURCE_OUTPUT_DONT_MOVE = 2, + PA_SOURCE_OUTPUT_START_CORKED = 4 } pa_source_output_flags_t; struct pa_source_output { @@ -62,8 +69,37 @@ struct pa_source_output { pa_sample_spec sample_spec; pa_channel_map channel_map; + /* Pushes a new memchunk into the output. Called from IO thread + * context. */ void (*push)(pa_source_output *o, const pa_memchunk *chunk); + + /* If non-NULL this function is called in each IO event loop and + * can be used to do additional processing even when the device is + * suspended and peek() is never called. Should return 1 when + * "some work" has been done and the IO event loop should be + * reiterated immediately. Called from IO thread context. */ + int (*process) (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 */ + + /* If non-NULL this function is called when the output is + * disconnected from its source. Called from IO thread context */ + void (*detach) (pa_source_output *o); /* may be NULL */ + + /* If non-NULL called whenever the the source this output is attached + * to suspends or resumes. Called from main context */ + void (*suspend) (pa_source_output *o, int b); /* may be NULL */ + + /* Supposed to unlink and destroy this stream. Called from main + * context. */ void (*kill)(pa_source_output* o); /* may 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. */ pa_usec_t (*get_latency) (pa_source_output *o); /* may be NULL */ pa_resample_method_t resample_method; @@ -102,8 +138,6 @@ typedef struct pa_source_output_new_data { int channel_map_is_set; pa_resample_method_t resample_method; - - int start_corked; } pa_source_output_new_data; pa_source_output_new_data* pa_source_output_new_data_init(pa_source_output_new_data *data); @@ -119,7 +153,7 @@ pa_source_output* pa_source_output_new( pa_source_output_flags_t flags); void pa_source_output_put(pa_source_output *o); -void pa_source_output_disconnect(pa_source_output*o); +void pa_source_output_unlink(pa_source_output*o); void pa_source_output_set_name(pa_source_output *i, const char *name); diff --git a/src/pulsecore/source.c b/src/pulsecore/source.c index 30a45cb6..ee2203e7 100644 --- a/src/pulsecore/source.c +++ b/src/pulsecore/source.c @@ -84,7 +84,8 @@ pa_source* pa_source_new( s->parent.process_msg = pa_source_process_msg; s->core = core; - s->state = PA_SOURCE_IDLE; + s->state = PA_SOURCE_INIT; + s->flags = 0; s->name = pa_xstrdup(name); s->description = NULL; s->driver = pa_xstrdup(driver); @@ -100,8 +101,6 @@ pa_source* pa_source_new( s->muted = 0; s->refresh_volume = s->refresh_muted = 0; - s->is_hardware = 0; - s->get_latency = NULL; s->set_volume = NULL; s->get_volume = NULL; @@ -111,6 +110,7 @@ pa_source* pa_source_new( s->userdata = NULL; s->asyncmsgq = NULL; + s->rtpoll = NULL; r = pa_idxset_put(core->sources, s, &s->index); assert(s->index != PA_IDXSET_INVALID && r >= 0); @@ -123,13 +123,22 @@ pa_source* pa_source_new( s->thread_info.soft_muted = s->muted; s->thread_info.state = s->state; - pa_subscription_post(core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_NEW, s->index); - - pa_hook_fire(&core->hooks[PA_CORE_HOOK_SOURCE_NEW_POST], s); - return s; } +void pa_source_put(pa_source *s) { + pa_source_assert_ref(s); + + pa_assert(s->state == PA_SINK_INIT); + pa_assert(s->rtpoll); + pa_assert(s->asyncmsgq); + + s->thread_info.state = s->state = PA_SOURCE_IDLE; + + 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_NEW_POST], s); +} + static int source_set_state(pa_source *s, pa_source_state_t state) { int ret; @@ -138,6 +147,21 @@ static int source_set_state(pa_source *s, pa_source_state_t state) { if (s->state == state) return 0; + if (state == PA_SOURCE_SUSPENDED && !(s->flags & PA_SOURCE_CAN_SUSPEND)) + return -1; + + if ((s->state == PA_SOURCE_SUSPENDED && PA_SOURCE_OPENED(state)) || + (PA_SOURCE_OPENED(s->state) && state == PA_SOURCE_SUSPENDED)) { + pa_source_output *o; + uint32_t idx; + + /* We're suspending or resuming, tell everyone about it */ + + for (o = PA_SOURCE_OUTPUT(pa_idxset_first(s->outputs, &idx)); o; o = PA_SOURCE_OUTPUT(pa_idxset_next(s->outputs, &idx))) + if (o->suspend) + o->suspend(o, state == PA_SINK_SUSPENDED); + } + if (s->set_state) if ((ret = s->set_state(s, state)) < 0) return -1; @@ -146,17 +170,19 @@ static int source_set_state(pa_source *s, pa_source_state_t state) { return -1; s->state = state; - pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_STATE_CHANGED], s); + + if (state != PA_SOURCE_UNLINKED) /* if we enter UNLINKED state pa_source_unlink() will fire the apropriate events */ + pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_STATE_CHANGED], s); return 0; } -void pa_source_disconnect(pa_source *s) { +void pa_source_unlink(pa_source *s) { pa_source_output *o, *j = NULL; pa_assert(s); - pa_return_if_fail(s->state != PA_SOURCE_DISCONNECTED); + pa_assert(PA_SOURCE_LINKED(s->state)); - pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_DISCONNECT], s); + pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_UNLINK], s); pa_namereg_unregister(s->core, s->name); pa_idxset_remove_by_data(s->core->sources, s, NULL); @@ -167,7 +193,7 @@ void pa_source_disconnect(pa_source *s) { j = o; } - source_set_state(s, PA_SOURCE_DISCONNECTED); + source_set_state(s, PA_SOURCE_UNLINKED); s->get_latency = NULL; s->get_volume = NULL; @@ -178,7 +204,7 @@ void pa_source_disconnect(pa_source *s) { pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE | PA_SUBSCRIPTION_EVENT_REMOVE, s->index); - pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_DISCONNECT_POST], s); + pa_hook_fire(&s->core->hooks[PA_CORE_HOOK_SOURCE_UNLINK_POST], s); } static void source_free(pa_object *o) { @@ -188,8 +214,8 @@ static void source_free(pa_object *o) { pa_assert(s); pa_assert(pa_source_refcnt(s) == 0); - if (s->state != PA_SOURCE_DISCONNECTED) - pa_source_disconnect(s); + if (PA_SOURCE_LINKED(s->state)) + pa_source_unlink(s); pa_log_info("Freeing source %u \"%s\"", s->index, s->name); @@ -208,6 +234,7 @@ static void source_free(pa_object *o) { int pa_source_update_status(pa_source*s) { pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); if (s->state == PA_SOURCE_SUSPENDED) return 0; @@ -217,6 +244,7 @@ int pa_source_update_status(pa_source*s) { int pa_source_suspend(pa_source *s, int suspend) { pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); if (suspend) return source_set_state(s, PA_SOURCE_SUSPENDED); @@ -226,6 +254,7 @@ int pa_source_suspend(pa_source *s, int suspend) { void pa_source_ping(pa_source *s) { pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_PING, NULL, 0, NULL, NULL); } @@ -235,6 +264,7 @@ void pa_source_post(pa_source*s, const pa_memchunk *chunk) { void *state = NULL; pa_source_assert_ref(s); + pa_assert(PA_SOURCE_OPENED(s->thread_info.state)); pa_assert(chunk); if (s->thread_info.state != PA_SOURCE_RUNNING) @@ -267,6 +297,10 @@ pa_usec_t pa_source_get_latency(pa_source *s) { pa_usec_t usec; pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); + + if (!PA_SOURCE_OPENED(s->state)) + return 0; if (s->get_latency) return s->get_latency(s); @@ -281,6 +315,7 @@ void pa_source_set_volume(pa_source *s, const pa_cvolume *volume) { int changed; pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); pa_assert(volume); changed = !pa_cvolume_equal(volume, &s->volume); @@ -298,7 +333,9 @@ void pa_source_set_volume(pa_source *s, const pa_cvolume *volume) { const pa_cvolume *pa_source_get_volume(pa_source *s) { pa_cvolume old_volume; + pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); old_volume = s->volume; @@ -318,6 +355,7 @@ void pa_source_set_mute(pa_source *s, int mute) { int changed; pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); changed = s->muted != mute; s->muted = mute; @@ -336,6 +374,7 @@ int pa_source_get_mute(pa_source *s) { int old_muted; pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); old_muted = s->muted; @@ -384,8 +423,16 @@ void pa_source_set_asyncmsgq(pa_source *s, pa_asyncmsgq *q) { s->asyncmsgq = q; } +void pa_source_set_rtpoll(pa_source *s, pa_rtpoll *p) { + pa_source_assert_ref(s); + pa_assert(p); + + s->rtpoll = p; +} + unsigned pa_source_used_by(pa_source *s) { pa_source_assert_ref(s); + pa_assert(PA_SOURCE_LINKED(s->state)); return pa_idxset_size(s->outputs); } @@ -393,19 +440,28 @@ unsigned pa_source_used_by(pa_source *s) { 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(PA_SOURCE_LINKED(s->thread_info.state)); switch ((pa_source_message_t) code) { case PA_SOURCE_MESSAGE_ADD_OUTPUT: { pa_source_output *o = PA_SOURCE_OUTPUT(userdata); pa_hashmap_put(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index), pa_source_output_ref(o)); + + if (o->attach) + o->attach(o); + return 0; } case PA_SOURCE_MESSAGE_REMOVE_OUTPUT: { pa_source_output *o = PA_SOURCE_OUTPUT(userdata); + + if (o->detach) + o->detach(o); + if (pa_hashmap_remove(s->thread_info.outputs, PA_UINT32_TO_PTR(o->index))) pa_source_output_unref(o); - + return 0; } @@ -452,3 +508,21 @@ int pa_source_suspend_all(pa_core *c, int suspend) { return ret; } + +int pa_source_process_outputs(pa_source *s) { + pa_source_output *o; + void *state = NULL; + int r; + + pa_source_assert_ref(s); + + if (!PA_SOURCE_LINKED(s->state)) + return 0; + + while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL))) + if (o->process) + if ((r = o->process(o))) + return r; + + return 0; +} diff --git a/src/pulsecore/source.h b/src/pulsecore/source.h index ffe51079..ddc66156 100644 --- a/src/pulsecore/source.h +++ b/src/pulsecore/source.h @@ -42,26 +42,36 @@ typedef struct pa_source pa_source; #include <pulsecore/module.h> #include <pulsecore/asyncmsgq.h> #include <pulsecore/msgobject.h> +#include <pulsecore/rtpoll.h> #define PA_MAX_OUTPUTS_PER_SOURCE 32 typedef enum pa_source_state { + PA_SOURCE_INIT, PA_SOURCE_RUNNING, PA_SOURCE_SUSPENDED, PA_SOURCE_IDLE, - PA_SOURCE_DISCONNECTED + PA_SOURCE_UNLINKED } pa_source_state_t; +static inline int PA_SOURCE_OPENED(pa_source_state_t x) { + return x == PA_SOURCE_RUNNING || x == PA_SOURCE_IDLE; +} + +static inline int PA_SOURCE_LINKED(pa_source_state_t x) { + return x == PA_SOURCE_RUNNING || x == PA_SOURCE_IDLE || x == PA_SOURCE_SUSPENDED; +} + struct pa_source { pa_msgobject parent; uint32_t index; pa_core *core; pa_source_state_t state; + pa_source_flags_t flags; char *name; char *description, *driver; /* may be NULL */ - int is_hardware; pa_module *module; /* may be NULL */ @@ -76,7 +86,7 @@ struct pa_source { int refresh_volume; int refresh_muted; - int (*set_state)(pa_source*source, pa_source_state_t state); /* may be NULL */ + int (*set_state)(pa_source*source, pa_source_state_t state); /* may be NULL */ int (*set_volume)(pa_source *s); /* dito */ int (*get_volume)(pa_source *s); /* dito */ int (*set_mute)(pa_source *s); /* dito */ @@ -84,7 +94,10 @@ struct pa_source { pa_usec_t (*get_latency)(pa_source *s); /* dito */ pa_asyncmsgq *asyncmsgq; + pa_rtpoll *rtpoll; + /* Contains copies of the above data so that the real-time worker + * thread can work without access locking */ struct { pa_source_state_t state; pa_hashmap *outputs; @@ -111,7 +124,7 @@ typedef enum pa_source_message { PA_SOURCE_MESSAGE_MAX } pa_source_message_t; -/* To be used exclusively by the source driver */ +/* To be called exclusively by the source driver, from main context */ pa_source* pa_source_new( pa_core *core, @@ -121,18 +134,22 @@ pa_source* pa_source_new( const pa_sample_spec *spec, const pa_channel_map *map); -void pa_source_disconnect(pa_source *s); +void pa_source_put(pa_source *s); +void pa_source_unlink(pa_source *s); void pa_source_set_module(pa_source *s, pa_module *m); void pa_source_set_description(pa_source *s, const char *description); void pa_source_set_asyncmsgq(pa_source *s, pa_asyncmsgq *q); +void pa_source_set_rtpoll(pa_source *s, pa_rtpoll *p); -/* Callable by everyone */ +/* May be called by everyone, from main context */ pa_usec_t pa_source_get_latency(pa_source *s); int pa_source_update_status(pa_source*s); int pa_source_suspend(pa_source *s, int suspend); +int pa_source_suspend_all(pa_core *c, int suspend); + void pa_source_ping(pa_source *s); void pa_source_set_volume(pa_source *source, const pa_cvolume *volume); @@ -143,15 +160,12 @@ int pa_source_get_mute(pa_source *source); unsigned pa_source_used_by(pa_source *s); #define pa_source_get_state(s) ((pa_source_state_t) (s)->state) -/* To be used exclusively by the source driver thread */ +/* To be called exclusively by the source driver, from IO context */ void pa_source_post(pa_source*s, const pa_memchunk *b); -int pa_source_process_msg(pa_msgobject *o, int code, void *userdata, int64_t, pa_memchunk *chunk); -static inline int PA_SOURCE_OPENED(pa_source_state_t x) { - return x == PA_SOURCE_RUNNING || x == PA_SOURCE_IDLE; -} +int pa_source_process_outputs(pa_source *o); -int pa_source_suspend_all(pa_core *c, int suspend); +int pa_source_process_msg(pa_msgobject *o, int code, void *userdata, int64_t, pa_memchunk *chunk); #endif |