On 02/20/2018 07:02 PM, Georg Chini wrote:
> On 20.02.2018 16:38, Raman Shyshniou wrote:
>> Currently the pipe-source will remain running even if no
>> writer is connected and therefore no data is produced.
>> This patch adds the autosuspend=<bool> option to prevent this.
>> Source will stay suspended if no writer is connected.
>> This option is enabled by default.
>> ---
>> src/modules/module-pipe-source.c | 279 +++++++++++++++++++++++++++++----------
>> 1 file changed, 212 insertions(+), 67 deletions(-)
>>
>> diff --git a/src/modules/module-pipe-source.c b/src/modules/module-pipe-source.c
>> index f8284c1..359cdbf 100644
>> --- a/src/modules/module-pipe-source.c
>> +++ b/src/modules/module-pipe-source.c
>> @@ -33,6 +33,7 @@
>> #include <sys/filio.h>
>> #endif
>> +#include <pulse/rtclock.h>
>> #include <pulse/xmalloc.h>
>> #include <pulsecore/core-error.h>
>> @@ -57,11 +58,24 @@ PA_MODULE_USAGE(
>> "format=<sample format> "
>> "rate=<sample rate> "
>> "channels=<number of channels> "
>> - "channel_map=<channel map>");
>> + "channel_map=<channel map> "
>> + "autosuspend=<boolean>");
>> #define DEFAULT_FILE_NAME "/tmp/music.input"
>> #define DEFAULT_SOURCE_NAME "fifo_input"
>> +struct pipe_source_msg {
>> + pa_msgobject parent;
>> +};
>> +
>> +typedef struct pipe_source_msg pipe_source_msg;
>> +PA_DEFINE_PRIVATE_CLASS(pipe_source_msg, pa_msgobject);
>> +
>> +enum {
>> + PIPE_SOURCE_SUSPEND,
>> + PIPE_SOURCE_RESUME
>> +};
>> +
>> struct userdata {
>> pa_core *core;
>> pa_module *module;
>> @@ -71,12 +85,14 @@ struct userdata {
>> pa_thread_mq thread_mq;
>> pa_rtpoll *rtpoll;
>> + pipe_source_msg *msg;
>> + pa_usec_t timestamp;
>> + bool autosuspend;
>> + size_t pipe_size;
>> +
>> char *filename;
>> + int corkfd;
>> int fd;
>> -
>> - pa_memchunk memchunk;
>> -
>> - pa_rtpoll_item *rtpoll_item;
>> };
>> static const char* const valid_modargs[] = {
>> @@ -87,9 +103,41 @@ static const char* const valid_modargs[] = {
>> "rate",
>> "channels",
>> "channel_map",
>> + "autosuspend",
>> NULL
>> };
>> +/* Called from main context */
>> +static int pipe_source_process_msg(
>> + pa_msgobject *o,
>> + int code,
>> + void *data,
>> + int64_t offset,
>> + pa_memchunk *chunk) {
>> +
>> + struct userdata *u = (struct userdata *) data;
>> +
>> + pa_assert(u);
>> +
>> + switch (code) {
>> + case PIPE_SOURCE_SUSPEND:
>> + pa_log_debug("Suspending source %s because no writers left", u->source->name);
>> + pa_source_suspend(u->source, true, PA_SUSPEND_APPLICATION);
>> + break;
>> +
>> + case PIPE_SOURCE_RESUME:
>> + pa_log_debug("Resuming source %s", u->source->name);
>> + pa_source_suspend(u->source, false, PA_SUSPEND_APPLICATION);
>> + break;
>> +
>> + default:
>> + pa_assert_not_reached();
>> + }
>> +
>> + return 0;
>> +}
>> +
>> +/* Called from thread context */
>> static int source_process_msg(
>> pa_msgobject *o,
>> int code,
>> @@ -101,17 +149,19 @@ static int source_process_msg(
>> switch (code) {
>> - case PA_SOURCE_MESSAGE_GET_LATENCY: {
>> - size_t n = 0;
>> + case PA_SOURCE_MESSAGE_SET_STATE:
>> -#ifdef FIONREAD
>> - int l;
>> + if (u->source->thread_info.state == PA_SOURCE_SUSPENDED || u->source->thread_info.state == PA_SOURCE_INIT) {
>> + if (PA_SOURCE_IS_OPENED(PA_PTR_TO_UINT(data)))
>> + u->timestamp = pa_rtclock_now();
>> + }
>> - if (ioctl(u->fd, FIONREAD, &l) >= 0 && l > 0)
>> - n = (size_t) l;
>> -#endif
>> + break;
>> +
>> + case PA_SOURCE_MESSAGE_GET_LATENCY: {
>> +
>> + *((int64_t*) data) = PA_CLIP_SUB((int64_t)pa_rtclock_now(), (int64_t)u->timestamp);
>> - *((int64_t*) data) = pa_bytes_to_usec(n, &u->source->sample_spec);
>> return 0;
>> }
>> }
>> @@ -121,7 +171,11 @@ static int source_process_msg(
>> static void thread_func(void *userdata) {
>> struct userdata *u = userdata;
>> + pa_rtpoll_item *rtpoll_item;
>> + struct pollfd *pollfd;
>> + pa_memchunk chunk;
>> int read_type = 0;
>> + size_t fs;
>> pa_assert(u);
>> @@ -129,68 +183,140 @@ static void thread_func(void *userdata) {
>> pa_thread_mq_install(&u->thread_mq);
>> + fs = pa_frame_size(&u->source->sample_spec);
>> +
>> + pa_memchunk_reset(&chunk);
>> + chunk.memblock = pa_memblock_new(u->core->mempool, u->pipe_size);
>> +
>> + rtpoll_item = NULL;
>> + pollfd = NULL;
>> +
>> + u->timestamp = pa_rtclock_now();
>> +
>> + /* Close our writer here to suspend source if no writers left */
>> + pa_assert_se(pa_close(u->corkfd) == 0);
>> + u->corkfd = -1;
>> +
>> for (;;) {
>> int ret;
>> - struct pollfd *pollfd;
>> - pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
>> + if (chunk.length) {
>> + /* We have a pending data, let's stop polling pipe.
>> + * Setting up pollfd->events = 0 is not enough to stop
>> + * POLLHUP spam if all writers are closed pipe.
>> + * We need to stop polling pipe completely */
>> + if (rtpoll_item) {
>> + pa_rtpoll_item_free(rtpoll_item);
>> + rtpoll_item = NULL;
>> + }
>> + } else {
>> + /* We have no pending data, let's start polling pipe */
>> + if (rtpoll_item == NULL) {
>> + rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
>> + pollfd = pa_rtpoll_item_get_pollfd(rtpoll_item, NULL);
>> + pollfd->events = POLLIN;
>> + pollfd->fd = u->fd;
>> + }
>> + }
>
> Your code will allocate/deallocate the rtpoll_item on each
> iteration. This is unnecessary and CPU intensive (I was told).
> I would still prefer my approach and I only see disadvantages
> with your way.
>
No, it's not like that.
rtpoll_item will be allocated:
- On thread start
- We just processed all pending data, i.e. source was just resumed.
rtpoll_item will be freed:
- We were got any data from pipe, but source was just suspended.
So there will be only one free per suspend and one allocate per resume.
>> - /* Try to read some data and pass it on to the source driver */
>> - if (u->source->thread_info.state == PA_SOURCE_RUNNING && pollfd->revents) {
>> - ssize_t l;
>> - void *p;
>> + if ((ret = pa_rtpoll_run(u->rtpoll)) < 0)
>> + goto fail;
>> - if (!u->memchunk.memblock) {
>> - u->memchunk.memblock = pa_memblock_new(u->core->mempool, pa_pipe_buf(u->fd));
>> - u->memchunk.index = u->memchunk.length = 0;
>> - }
>> + if (ret == 0)
>> + goto finish;
>> - pa_assert(pa_memblock_get_length(u->memchunk.memblock) > u->memchunk.index);
>> + if (rtpoll_item) {
>> + pollfd = pa_rtpoll_item_get_pollfd(rtpoll_item, NULL);
>> - p = pa_memblock_acquire(u->memchunk.memblock);
>> - l = pa_read(u->fd, (uint8_t*) p + u->memchunk.index, pa_memblock_get_length(u->memchunk.memblock) - u->memchunk.index, &read_type);
>> - pa_memblock_release(u->memchunk.memblock);
>> + if (pollfd->revents & ~(POLLIN|POLLHUP)) {
>> + pa_log("FD error: %s%s%s%s",
>> + pollfd->revents & POLLOUT ? "POLLOUT " : "",
>> + pollfd->revents & POLLERR ? "POLLERR " : "",
>> + pollfd->revents & POLLPRI ? "POLLPRI " : "",
>> + pollfd->revents & POLLNVAL ? "POLLNVAL " : "");
>> - pa_assert(l != 0); /* EOF cannot happen, since we opened the fifo for both reading and writing */
>> + goto fail;
>> + }
>> + } else
>> + pollfd = NULL;
>> - if (l < 0) {
>> + /* Try to read some data if there are any events */
>> + if (pollfd && pollfd->revents) {
>> + ssize_t l;
>> + void *p;
>> - if (errno == EINTR)
>> - continue;
>> - else if (errno != EAGAIN) {
>> - pa_log("Failed to read data from FIFO: %s", pa_cstrerror(errno));
>> - goto fail;
>> - }
>> + pa_assert(chunk.index < fs);
>> - } else {
>> + p = pa_memblock_acquire(chunk.memblock);
>> + l = pa_read(u->fd, (uint8_t *) p + chunk.index, pa_memblock_get_length(chunk.memblock) - chunk.index, &read_type);
>> + pa_memblock_release(chunk.memblock);
>> - u->memchunk.length = (size_t) l;
>> - pa_source_post(u->source, &u->memchunk);
>> - u->memchunk.index += (size_t) l;
>> + if (l > 0) {
>> + chunk.length = (size_t) l;
>> + u->timestamp = pa_rtclock_now();
>> - if (u->memchunk.index >= pa_memblock_get_length(u->memchunk.memblock)) {
>> - pa_memblock_unref(u->memchunk.memblock);
>> - pa_memchunk_reset(&u->memchunk);
>> + if (u->corkfd >= 0) {
>> + if (u->autosuspend)
>> + pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(u->msg), PIPE_SOURCE_RESUME, u, 0, NULL, NULL);
>> +
>> + pa_assert_se(pa_close(u->corkfd) == 0);
>> + u->corkfd = -1;
>> + }
>> + } else if (l == 0) {
>> + /* Writer was disconnected: discard unalligned data tail */
>> + chunk.index = 0;
>> +
>> + if (u->corkfd < 0) {
>> + if (u->autosuspend)
>> + pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(u->msg), PIPE_SOURCE_SUSPEND, u, 0, NULL, NULL);
>> +
>> + if ((u->corkfd = pa_open_cloexec(u->filename, O_WRONLY, 0)) < 0) {
>> + pa_log("open('%s'): %s", u->filename, pa_cstrerror(errno));
>> + goto fail;
>> + }
>> }
>> + } else {
>> + if (errno == EINTR || errno == EAGAIN)
>> + continue;
>> - pollfd->revents = 0;
>> + pa_log("Failed to read data from FIFO: %s", pa_cstrerror(errno));
>> + goto fail;
>> }
>> }
>> - /* Hmm, nothing to do. Let's sleep */
>> - pollfd->events = (short) (u->source->thread_info.state == PA_SOURCE_RUNNING ? POLLIN : 0);
>> + /* Post pending data.
>> + * Let's keep frame boundaries */
>> + if (PA_SOURCE_IS_OPENED(u->source->thread_info.state)) {
>
> Just checking if the source is open before you post the data is also not
> the right thing to do. When you are resuming from autosuspend, you
> send a message to the main thread. You do not know if this message
> has already been processed when you reach this part of the code.
When I send the message to the main thread I expect the u->source->thread_info.state
will be changed only after source_process_msg() will be called, i.e. after at least
one pa_rtpoll_run() will be called.
That's why I wait for source to be resumed removing fd from polling.
>
>> + size_t total_len = chunk.index + chunk.length;
>> + size_t frames = total_len / fs;
>> + size_t tail = total_len % fs;
>> - if ((ret = pa_rtpoll_run(u->rtpoll)) < 0)
>> - goto fail;
>> + if (frames > 0) {
>> + pa_memblock *memblock;
>> - if (ret == 0)
>> - goto finish;
>> + memblock = pa_memblock_new(u->core->mempool, u->pipe_size);
>> + chunk.length = frames * fs;
>> + chunk.index = 0;
>> - pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
>> + if (tail) {
>> + void *src, *dst;
>> - if (pollfd->revents & ~POLLIN) {
>> - pa_log("FIFO shutdown.");
>> - goto fail;
>> + dst = pa_memblock_acquire(memblock);
>> + src = pa_memblock_acquire(chunk.memblock);
>> +
>> + memcpy(dst, (uint8_t *) src + chunk.length, tail);
>> +
>> + pa_memblock_release(chunk.memblock);
>> + pa_memblock_release(memblock);
>> + }
>> +
>> + pa_source_post(u->source, &chunk);
>> + pa_memblock_unref(chunk.memblock);
>> + chunk.memblock = memblock;
>> + }
>> +
>> + chunk.index = tail;
>> + chunk.length = 0;
>> }
>> }
>> @@ -201,6 +327,11 @@ fail:
>> pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
>> finish:
>> + if (rtpoll_item)
>> + pa_rtpoll_item_free(rtpoll_item);
>> +
>> + pa_memblock_unref(chunk.memblock);
>> +
>> pa_log_debug("Thread shutting down");
>> }
>> @@ -210,7 +341,6 @@ int pa__init(pa_module *m) {
>> pa_sample_spec ss;
>> pa_channel_map map;
>> pa_modargs *ma;
>> - struct pollfd *pollfd;
>> pa_source_new_data data;
>> pa_assert(m);
>> @@ -230,7 +360,6 @@ int pa__init(pa_module *m) {
>> m->userdata = u = pa_xnew0(struct userdata, 1);
>> u->core = m->core;
>> u->module = m;
>> - pa_memchunk_reset(&u->memchunk);
>> u->rtpoll = pa_rtpoll_new();
>> if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
>> @@ -238,13 +367,31 @@ int pa__init(pa_module *m) {
>> goto fail;
>> }
>> + if (!(u->msg = pa_msgobject_new(pipe_source_msg)))
>> + goto fail;
>> +
>> + u->msg->parent.process_msg = pipe_source_process_msg;
>> +
>> + u->autosuspend = true;
>> +
>> + if (pa_modargs_get_value_boolean(ma, "autosuspend", &u->autosuspend) < 0) {
>> + pa_log("Failed to parse autosuspend argument.");
>> + goto fail;
>> + }
>> +
>> u->filename = pa_runtime_path(pa_modargs_get_value(ma, "file", DEFAULT_FILE_NAME));
>> if (mkfifo(u->filename, 0666) < 0) {
>> pa_log("mkfifo('%s'): %s", u->filename, pa_cstrerror(errno));
>> goto fail;
>> }
>> - if ((u->fd = pa_open_cloexec(u->filename, O_RDWR, 0)) < 0) {
>> +
>> + if ((u->corkfd = pa_open_cloexec(u->filename, O_RDWR, 0)) < 0) {
>> + pa_log("open('%s'): %s", u->filename, pa_cstrerror(errno));
>> + goto fail;
>> + }
>> +
>> + if ((u->fd = pa_open_cloexec(u->filename, O_RDONLY, 0)) < 0) {
>> pa_log("open('%s'): %s", u->filename, pa_cstrerror(errno));
>> goto fail;
>> }
>> @@ -289,12 +436,10 @@ int pa__init(pa_module *m) {
>> pa_source_set_asyncmsgq(u->source, u->thread_mq.inq);
>> pa_source_set_rtpoll(u->source, u->rtpoll);
>> - pa_source_set_fixed_latency(u->source, pa_bytes_to_usec(pa_pipe_buf(u->fd), &u->source->sample_spec));
>> - u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
>> - pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
>> - pollfd->fd = u->fd;
>> - pollfd->events = pollfd->revents = 0;
>> + u->pipe_size = pa_pipe_buf(u->fd);
>> + u->pipe_size = PA_MIN(pa_mempool_block_size_max(u->core->mempool), u->pipe_size);
>> + pa_source_set_fixed_latency(u->source, pa_bytes_to_usec(u->pipe_size, &u->source->sample_spec));
>> if (!(u->thread = pa_thread_new("pipe-source", thread_func, u))) {
>> pa_log("Failed to create thread.");
>> @@ -346,12 +491,6 @@ void pa__done(pa_module *m) {
>> if (u->source)
>> pa_source_unref(u->source);
>> - 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);
>> @@ -360,6 +499,12 @@ void pa__done(pa_module *m) {
>> pa_xfree(u->filename);
>> }
>> + if (u->msg)
>> + pa_xfree(u->msg);
>> +
>> + if (u->corkfd >= 0)
>> + pa_assert_se(pa_close(u->corkfd) == 0);
>> +
>> if (u->fd >= 0)
>> pa_assert_se(pa_close(u->fd) == 0);
>>
>
>
--
Raman
