[PATCH] "pipe-sink" added option "use_system_clock_for_timing"

[samo_pogacnik@xxxxxxx (Samo Pogačnik)]

Sorry for the missing annotation, so here it is:

This is a follow up of the thread:
[PATCH] "pipe-sink" module timig based upon "null-sink" module
mechanics.

As proposed, old "pipe-sink" behaviour has been preserved and a new
option "use_system_clock_for_timing" defines new behaviour. i.e:
...
load-module module-pipe-sink sink_name=pipe_sink format=s16le
rate=44100 channels=2 use_system_clock_for_timing=true 
...
Using this option, even the simplest tools like "cat" can properly
dump raw audio from the pipe.

Additionally, logging of dropped bytes of audio data was implemented
as suggested:
"Yeah, it's not good to log every dropped chunk. Maybe log only the
first dropped chunk and later the total dropped amount when the pipe
starts accepting writes again".
I also added "dropped_bytes" statistics cleanup upon every resume from
suspended state of the "pipe-sink" module.

Afterwards i would like to add "auto-draining" of the pipe, as already
propsed
upon each resume and maybe also upon full pipe.

And lastly additinal option migth be usefull too reuse-existing-pipe.

regards, Samo

Dne 09.12.2017 (sob) ob 20:09 +0100 je Samo PogaÄ?nik napisal(a):
> ---
> Â src/modules/module-pipe-sink.c | 236
> ++++++++++++++++++++++++++++++++++++++---
> Â 1 file changed, 221 insertions(+), 15 deletions(-)
> 
> diff --git a/src/modules/module-pipe-sink.c b/src/modules/module-
> pipe-sink.c
> index 8396a63..e2bf949 100644
> --- a/src/modules/module-pipe-sink.c
> +++ b/src/modules/module-pipe-sink.c
> @@ -33,7 +33,9 @@
> Â #include <sys/filio.h>
> Â #endif
> Â 
> -#include <pulse/xmalloc.h>
> +#include <pulse/timeval.h>
> +#include <pulse/util.h>
> +#include <pulse/rtclock.h>
> Â 
> Â #include <pulsecore/core-error.h>
> Â #include <pulsecore/sink.h>
> @@ -59,7 +61,9 @@ PA_MODULE_USAGE(
> Â Â Â Â Â Â Â Â Â "format=<sample format> "
> Â Â Â Â Â Â Â Â Â "rate=<sample rate> "
> Â Â Â Â Â Â Â Â Â "channels=<number of channels> "
> -Â Â Â Â Â Â Â Â "channel_map=<channel map>");
> +Â Â Â Â Â Â Â Â "channel_map=<channel map> "
> +Â Â Â Â Â Â Â Â "use_system_clock_for_timing=<yes or no> "
> +);
> Â 
> Â #define DEFAULT_FILE_NAME "fifo_output"
> Â #define DEFAULT_SINK_NAME "fifo_output"
> @@ -76,12 +80,19 @@ struct userdata {
> Â Â Â Â Â char *filename;
> Â Â Â Â Â int fd;
> Â Â Â Â Â size_t buffer_size;
> +Â Â Â Â size_t bytes_dropped;
> +Â Â Â Â bool log_dropped;
> +Â Â Â Â bool pipe_sink_open;
> Â 
> Â Â Â Â Â pa_memchunk memchunk;
> Â 
> Â Â Â Â Â pa_rtpoll_item *rtpoll_item;
> Â 
> Â Â Â Â Â int write_type;
> +Â Â Â Â pa_usec_t block_usec;
> +Â Â Â Â pa_usec_t timestamp;
> +
> +Â Â Â Â bool use_system_clock_for_timing;
> Â };
> Â 
> Â static const char* const valid_modargs[] = {
> @@ -92,6 +103,7 @@ static const char* const valid_modargs[] = {
> Â Â Â Â Â "rate",
> Â Â Â Â Â "channels",
> Â Â Â Â Â "channel_map",
> +Â Â Â Â "use_system_clock_for_timing",
> Â Â Â Â Â NULL
> Â };
> Â 
> @@ -99,27 +111,148 @@ static int sink_process_msg(pa_msgobject *o,
> int code, void *data, int64_t offse
> Â Â Â Â Â struct userdata *u = PA_SINK(o)->userdata;
> Â 
> Â Â Â Â Â switch (code) {
> +Â Â Â Â Â Â Â Â case PA_SINK_MESSAGE_SET_STATE:
> +Â Â Â Â Â Â Â Â Â Â Â Â if (pa_sink_get_state(u->sink) == PA_SINK_SUSPENDED ||
> pa_sink_get_state(u->sink) == PA_SINK_INIT) {
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â if (PA_PTR_TO_UINT(data) == PA_SINK_RUNNING ||
> PA_PTR_TO_UINT(data) == PA_SINK_IDLE)
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â u->timestamp = pa_rtclock_now();
> +Â Â Â Â Â Â Â Â Â Â Â Â }
> +Â Â Â Â Â Â Â Â Â Â Â Â break;
> Â 
> -Â Â Â Â Â Â Â Â case PA_SINK_MESSAGE_GET_LATENCY: {
> -Â Â Â Â Â Â Â Â Â Â Â Â size_t n = 0;
> +Â Â Â Â Â Â Â Â case PA_SINK_MESSAGE_GET_LATENCY:
> +Â Â Â Â Â Â Â Â Â Â Â Â if (u->use_system_clock_for_timing) {
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â pa_usec_t now;
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â now = pa_rtclock_now();
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â *((int64_t*) data) = (int64_t)u->timestamp -
> (int64_t)now;
> +Â Â Â Â Â Â Â Â Â Â Â Â } else {
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â size_t n = 0;
> Â 
> Â #ifdef FIONREAD
> -Â Â Â Â Â Â Â Â Â Â Â Â int l;
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â int l;
> Â 
> -Â Â Â Â Â Â Â Â Â Â Â Â if (ioctl(u->fd, FIONREAD, &l) >= 0 && l > 0)
> -Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â n = (size_t) l;
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â if (ioctl(u->fd, FIONREAD, &l) >= 0 && l > 0)
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â n = (size_t) l;
> Â #endif
> Â 
> -Â Â Â Â Â Â Â Â Â Â Â Â n += u->memchunk.length;
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â n += u->memchunk.length;
> Â 
> -Â Â Â Â Â Â Â Â Â Â Â Â *((int64_t*) data) = pa_bytes_to_usec(n, &u->sink-
> >sample_spec);
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â *((int64_t*) data) = pa_bytes_to_usec(n, &u->sink-
> >sample_spec);
> +Â Â Â Â Â Â Â Â Â Â Â Â }
> Â Â Â Â Â Â Â Â Â Â Â Â Â return 0;
> -Â Â Â Â Â Â Â Â }
> Â Â Â Â Â }
> Â 
> Â Â Â Â Â return pa_sink_process_msg(o, code, data, offset, chunk);
> Â }
> Â 
> +static void sink_update_requested_latency_cb(pa_sink *s) {
> +Â Â Â Â struct userdata *u;
> +Â Â Â Â size_t nbytes;
> +
> +Â Â Â Â pa_sink_assert_ref(s);
> +Â Â Â Â pa_assert_se(u = s->userdata);
> +
> +Â Â Â Â u->block_usec = pa_sink_get_requested_latency_within_thread(s);
> +
> +Â Â Â Â if (u->block_usec == (pa_usec_t) -1)
> +Â Â Â Â Â Â Â Â u->block_usec = s->thread_info.max_latency;
> +
> +Â Â Â Â nbytes = pa_usec_to_bytes(u->block_usec, &s->sample_spec);
> +Â Â Â Â pa_sink_set_max_rewind_within_thread(s, nbytes);
> +Â Â Â Â pa_sink_set_max_request_within_thread(s, nbytes);
> +}
> +
> +static size_t pipe_sink_write(struct userdata *u, void *p) {
> +Â Â Â Â ssize_t l;
> +Â Â Â Â size_t index, length;
> +
> +Â Â Â Â pa_assert(u);
> +
> +Â Â Â Â index = u->memchunk.index;
> +Â Â Â Â length = u->memchunk.length;
> +
> +Â Â Â Â for (;;) {
> +Â Â Â Â Â Â Â Â l = pa_write(u->fd, (uint8_t*) p + index, length, &u-
> >write_type);
> +
> +Â Â Â Â Â Â Â Â pa_assert(l != 0);
> +
> +Â Â Â Â Â Â Â Â if (l < 0) {
> +Â Â Â Â Â Â Â Â Â Â Â Â if (errno == EAGAIN)
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â break;
> +Â Â Â Â Â Â Â Â Â Â Â Â else if (errno != EINTR) {
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â pa_log("Failed to write data to FIFO (%d): %s",
> errno, pa_cstrerror(errno));
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â break;
> +Â Â Â Â Â Â Â Â Â Â Â Â }
> +Â Â Â Â Â Â Â Â } else {
> +Â Â Â Â Â Â Â Â Â Â Â Â index += (size_t) l;
> +Â Â Â Â Â Â Â Â Â Â Â Â length -= (size_t) l;
> +
> +Â Â Â Â Â Â Â Â Â Â Â Â if (length <= 0) {
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â break;
> +Â Â Â Â Â Â Â Â Â Â Â Â }
> +Â Â Â Â Â Â Â Â }
> +Â Â Â Â }
> +
> +Â Â Â Â return index;
> +}
> +
> +static void process_render_use_timing(struct userdata *u, pa_usec_t
> now) {
> +Â Â Â Â size_t dropped = 0;
> +Â Â Â Â size_t consumed = 0;
> +
> +Â Â Â Â pa_assert(u);
> +
> +Â Â Â Â if (!u->pipe_sink_open) {
> +Â Â Â Â Â Â Â Â /* Clear bytes_dropped statistics. */
> +Â Â Â Â Â Â Â Â pa_log_debug("pipe-sink cleared dropped bytes statistics
> (sum: %zu -> 0 dropped bytes)", u->bytes_dropped);
> +Â Â Â Â Â Â Â Â u->bytes_dropped = 0;
> +Â Â Â Â Â Â Â Â u->log_dropped = true;
> +Â Â Â Â Â Â Â Â u->pipe_sink_open = true;
> +Â Â Â Â }
> +
> +Â Â Â Â /* This is the configured latency. Sink inputs connected to us
> +Â Â Â Â might not have a single frame more than the maxrequest value
> +Â Â Â Â queued. Hence: at maximum read this many bytes from the sink
> +Â Â Â Â inputs. */
> +
> +Â Â Â Â /* Fill the buffer up the latency size */
> +Â Â Â Â while (u->timestamp < now + u->block_usec) {
> +Â Â Â Â Â Â Â Â void *p;
> +Â Â Â Â Â Â Â Â size_t written = 0;
> +
> +Â Â Â Â Â Â Â Â pa_sink_render(u->sink, u->sink->thread_info.max_request,
> &u->memchunk);
> +
> +Â Â Â Â Â Â Â Â pa_assert(u->memchunk.length > 0);
> +
> +Â Â Â Â Â Â Â Â p = pa_memblock_acquire(u->memchunk.memblock);
> +
> +Â Â Â Â Â Â Â Â written = pipe_sink_write(u, p);
> +
> +Â Â Â Â Â Â Â Â pa_memblock_release(u->memchunk.memblock);
> +
> +Â Â Â Â Â Â Â Â pa_memblock_unref(u->memchunk.memblock);
> +
> +Â Â Â Â Â Â Â Â u->timestamp += pa_bytes_to_usec(u->memchunk.length, &u-
> >sink->sample_spec);
> +
> +Â Â Â Â Â Â Â Â dropped += (u->memchunk.length - written);
> +Â Â Â Â Â Â Â Â consumed += u->memchunk.length;
> +
> +Â Â Â Â Â Â Â Â if (consumed >= u->sink->thread_info.max_request)
> +Â Â Â Â Â Â Â Â Â Â Â Â break;
> +Â Â Â Â }
> +
> +Â Â Â Â u->bytes_dropped += dropped;
> +
> +Â Â Â Â if (dropped == 0) {
> +Â Â Â Â Â Â Â Â if (!u->log_dropped)
> +Â Â Â Â Â Â Â Â Â Â Â Â pa_log_debug("pipe-sink stopped dropping (sum: %zu
> dropped bytes)", u->bytes_dropped);
> +Â Â Â Â Â Â Â Â u->log_dropped = true;
> +Â Â Â Â }
> +
> +Â Â Â Â if ((dropped != 0) && u->log_dropped) {
> +Â Â Â Â Â Â Â Â pa_log_debug("pipe-sink just dropped %zu bytes (sum: %zu
> dropped bytes)", dropped, u->bytes_dropped);
> +Â Â Â Â Â Â Â Â u->log_dropped = false;
> +Â Â Â Â }
> +}
> +
> Â static int process_render(struct userdata *u) {
> Â Â Â Â Â pa_assert(u);
> Â 
> @@ -164,6 +297,56 @@ static int process_render(struct userdata *u) {
> Â Â Â Â Â }
> Â }
> Â 
> +static void thread_func_use_timing(void *userdata) {
> +Â Â Â Â struct userdata *u = userdata;
> +
> +Â Â Â Â pa_assert(u);
> +
> +Â Â Â Â pa_log_debug("Thread (use timing) starting up");
> +
> +Â Â Â Â pa_thread_mq_install(&u->thread_mq);
> +
> +Â Â Â Â u->timestamp = pa_rtclock_now();
> +
> +Â Â Â Â for (;;) {
> +Â Â Â Â Â Â Â Â pa_usec_t now = 0;
> +Â Â Â Â Â Â Â Â int ret;
> +
> +Â Â Â Â Â Â Â Â if (PA_SINK_IS_OPENED(u->sink->thread_info.state))
> +Â Â Â Â Â Â Â Â Â Â Â Â now = pa_rtclock_now();
> +
> +Â Â Â Â Â Â Â Â if (PA_UNLIKELY(u->sink->thread_info.rewind_requested))
> +Â Â Â Â Â Â Â Â Â Â Â Â pa_sink_process_rewind(u->sink, 0);
> +
> +Â Â Â Â Â Â Â Â /* Render some data and write it to the fifo */
> +Â Â Â Â Â Â Â Â if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
> +Â Â Â Â Â Â Â Â Â Â Â Â if (u->timestamp <= now)
> +Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â Â process_render_use_timing(u, now);
> +
> +Â Â Â Â Â Â Â Â Â Â Â Â pa_rtpoll_set_timer_absolute(u->rtpoll, u->timestamp);
> +Â Â Â Â Â Â Â Â } else {
> +Â Â Â Â Â Â Â Â Â Â Â Â pa_rtpoll_set_timer_disabled(u->rtpoll);
> +Â Â Â Â Â Â Â Â Â Â Â Â u->pipe_sink_open = false;
> +Â Â Â Â Â Â Â Â }
> +
> +Â Â Â Â Â Â Â Â /* Hmm, nothing to do. Let's sleep */
> +Â Â Â Â Â Â Â Â if ((ret = pa_rtpoll_run(u->rtpoll)) < 0)
> +Â Â Â Â Â Â Â Â Â Â Â Â goto fail;
> +
> +Â Â Â Â Â Â Â Â if (ret == 0)
> +Â Â Â Â Â Â Â Â Â Â Â Â goto finish;
> +Â Â Â Â }
> +
> +fail:
> +Â Â Â Â /* If this was no regular exit from the loop we have to continue
> +Â Â Â Â Â * processing messages until we received PA_MESSAGE_SHUTDOWN */
> +Â Â Â Â pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core),
> PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
> +Â Â Â Â pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
> +
> +finish:
> +Â Â Â Â pa_log_debug("Thread (use timing) shutting down");
> +}
> +
> Â static void thread_func(void *userdata) {
> Â Â Â Â Â struct userdata *u = userdata;
> Â 
> @@ -227,6 +410,7 @@ int pa__init(pa_module *m) {
> Â Â Â Â Â pa_modargs *ma;
> Â Â Â Â Â struct pollfd *pollfd;
> Â Â Â Â Â pa_sink_new_data data;
> +Â Â Â Â pa_thread_func_t thread_routine;
> Â 
> Â Â Â Â Â pa_assert(m);
> Â 
> @@ -249,6 +433,11 @@ int pa__init(pa_module *m) {
> Â Â Â Â Â pa_memchunk_reset(&u->memchunk);
> Â Â Â Â Â u->rtpoll = pa_rtpoll_new();
> Â 
> +Â Â Â Â if (pa_modargs_get_value_boolean(ma,
> "use_system_clock_for_timing", &u->use_system_clock_for_timing) < 0)
> {
> +Â Â Â Â Â Â Â Â pa_log("Failed to parse use_system_clock_for_timing
> argument.");
> +Â Â Â Â Â Â Â Â goto fail;
> +Â Â Â Â }
> +
> Â Â Â Â Â if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u-
> >rtpoll) < 0) {
> Â Â Â Â Â Â Â Â Â pa_log("pa_thread_mq_init() failed.");
> Â Â Â Â Â Â Â Â Â goto fail;
> @@ -294,7 +483,10 @@ int pa__init(pa_module *m) {
> Â Â Â Â Â Â Â Â Â goto fail;
> Â Â Â Â Â }
> Â 
> -Â Â Â Â u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY);
> +Â Â Â Â if (u->use_system_clock_for_timing)
> +Â Â Â Â Â Â Â Â u->sink = pa_sink_new(m->core, &data,
> PA_SINK_LATENCY|PA_SINK_DYNAMIC_LATENCY);
> +Â Â Â Â else
> +Â Â Â Â Â Â Â Â u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY);
> Â Â Â Â Â pa_sink_new_data_done(&data);
> Â 
> Â Â Â Â Â if (!u->sink) {
> @@ -303,21 +495,34 @@ int pa__init(pa_module *m) {
> Â Â Â Â Â }
> Â 
> Â Â Â Â Â u->sink->parent.process_msg = sink_process_msg;
> +Â Â Â Â if (u->use_system_clock_for_timing)
> +Â Â Â Â Â Â Â Â u->sink->update_requested_latency =
> sink_update_requested_latency_cb;
> Â Â Â Â Â u->sink->userdata = u;
> Â 
> Â Â Â Â Â pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
> Â Â Â Â Â pa_sink_set_rtpoll(u->sink, u->rtpoll);
> Â 
> +Â Â Â Â u->bytes_dropped = 0;
> +Â Â Â Â u->log_dropped = true;
> +Â Â Â Â u->pipe_sink_open = false;
> Â Â Â Â Â u->buffer_size = pa_frame_align(pa_pipe_buf(u->fd), &u->sink-
> >sample_spec);
> +Â Â Â Â if (u->use_system_clock_for_timing) {
> +Â Â Â Â Â Â Â Â u->block_usec = pa_bytes_to_usec(u->buffer_size, &u->sink-
> >sample_spec);
> +Â Â Â Â Â Â Â Â pa_sink_set_latency_range(u->sink, 0, u->block_usec);
> +Â Â Â Â Â Â Â Â pa_sink_set_max_rewind(u->sink, u->buffer_size);
> +Â Â Â Â Â Â Â Â thread_routine = thread_func_use_timing;
> +Â Â Â Â } else {
> +Â Â Â Â Â Â Â Â pa_sink_set_fixed_latency(u->sink, pa_bytes_to_usec(u-
> >buffer_size, &u->sink->sample_spec));
> +Â Â Â Â Â Â Â Â thread_routine = thread_func;
> +Â Â Â Â }
> Â Â Â Â Â pa_sink_set_max_request(u->sink, u->buffer_size);
> -Â Â Â Â pa_sink_set_fixed_latency(u->sink, pa_bytes_to_usec(u-
> >buffer_size, &u->sink->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;
> Â 
> -Â Â Â Â if (!(u->thread = pa_thread_new("pipe-sink", thread_func, u))) {
> +Â Â Â Â if (!(u->thread = pa_thread_new("pipe-sink", thread_routine,
> u))) {
> Â Â Â Â Â Â Â Â Â pa_log("Failed to create thread.");
> Â Â Â Â Â Â Â Â Â goto fail;
> Â Â Â Â Â }
> @@ -367,8 +572,9 @@ void pa__done(pa_module *m) {
> Â Â Â Â Â if (u->sink)
> Â Â Â Â Â Â Â Â Â pa_sink_unref(u->sink);
> Â 
> -Â Â Â Â if (u->memchunk.memblock)
> -Â Â Â Â Â Â Â Â pa_memblock_unref(u->memchunk.memblock);
> +Â Â Â Â if (!u->use_system_clock_for_timing)
> +Â Â Â Â Â Â Â Â if (u->memchunk.memblock)
> +Â Â Â Â Â Â Â Â Â Â Â Â pa_memblock_unref(u->memchunk.memblock);
> Â 
> Â Â Â Â Â if (u->rtpoll_item)
> Â Â Â Â Â Â Â Â Â pa_rtpoll_item_free(u->rtpoll_item);