Hi all!
I intended to get rid of the "jack_transport; play; exit" phase of
starting jackd and pulseaudio. But then I installed a realtime patched
kernel and started to run jackd with very low latency settings, and
found out that the jack sink module is unusable in such setup. So I had
more to fix - the processing callback had to be made realtime-safe. The
result is attached. This is the first patch I ever send anywhere, so it
may be in a wrong format or anything. It's the output of "svn diff".
What I did was adding internal buffering to the sink. That's implemented
with Jack's utility ringbuffer. Now the processing function just reads
from the buffer and tells the pulse core (correct terminology?) to
render some more audio, but the processing function doesn't wait for
that to happen. I guess that's about it.
A new module parameter is added too. buffersize takes an unsigned
integer as an argument, which is the amount of frames the user wishes to
be in the ringbuffer, defaulting to 4096. The actual buffersize is
rounded up to nearest power of two, and is always greater than the jack
blocksize (frames per processing call).
The blocksize change callback isn't tested, since it seems that my jack
doesn't support dynamic changing of the blocksize. I believe I enabled
it when configuring the jack build, but maybe it's just not yet
implemented. Or maybe I didn't enable it.
--
Tanu Kaskinen
Index: src/modules/module-jack-sink.c
===================================================================
--- src/modules/module-jack-sink.c (revision 1437)
+++ src/modules/module-jack-sink.c (working copy)
@@ -3,7 +3,7 @@
/***
This file is part of PulseAudio.
- Copyright 2006 Lennart Poettering
+ Copyright 2006, 2007 Lennart Poettering and Tanu Kaskinen
PulseAudio is free software; you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published
@@ -37,6 +37,7 @@
#include <pthread.h>
#include <jack/jack.h>
+#include <jack/ringbuffer.h>
#include <pulse/xmalloc.h>
@@ -51,7 +52,7 @@
#include "module-jack-sink-symdef.h"
-PA_MODULE_AUTHOR("Lennart Poettering")
+PA_MODULE_AUTHOR("Lennart Poettering & Tanu Kaskinen")
PA_MODULE_DESCRIPTION("Jack Sink")
PA_MODULE_VERSION(PACKAGE_VERSION)
PA_MODULE_USAGE(
@@ -60,6 +61,7 @@
"client_name=<jack client name> "
"channels=<number of channels> "
"connect=<connect ports?> "
+ "buffersize=<number of frames> "
"channel_map=<channel map>")
#define DEFAULT_SINK_NAME "jack_out"
@@ -74,20 +76,21 @@
jack_port_t* port[PA_CHANNELS_MAX];
jack_client_t *client;
-
- pthread_mutex_t mutex;
+
+ void * buffer[PA_CHANNELS_MAX];
+ jack_nframes_t blocksize; /* The amount of frames we process in one call; the jack buffer size. */
+ int bufferptrs_are_dirty; /* Set to one when blocksize changes, and zeroed again in the process callback. */
+
+ pthread_mutex_t mutex; /* For avoiding blocksize changes at wrong moment. */
pthread_cond_t cond;
-
- void * buffer[PA_CHANNELS_MAX];
- jack_nframes_t frames_requested;
+
+ jack_ringbuffer_t *ringbuffer;
+
int quit_requested;
int pipe_fd_type;
int pipe_fds[2];
pa_io_event *io_event;
-
- jack_nframes_t frames_in_buffer;
- jack_nframes_t timestamp;
};
static const char* const valid_modargs[] = {
@@ -96,6 +99,7 @@
"client_name",
"channels",
"connect",
+ "buffersize", /* amount of internal buffering in frames */
"channel_map",
NULL
};
@@ -110,12 +114,16 @@
pa_sink_disconnect(u->sink);
pa_sink_unref(u->sink);
u->sink = NULL;
+ jack_ringbuffer_free(u->ringbuffer);
+ u->ringbuffer = NULL;
pa_module_unload_request(u->module);
}
static void io_event_cb(pa_mainloop_api *m, pa_io_event *e, int fd, pa_io_event_flags_t flags, void *userdata) {
struct userdata *u = userdata;
char x;
+ unsigned fs;
+ unsigned c;
assert(m);
assert(e);
@@ -131,35 +139,47 @@
return;
}
+ fs = pa_frame_size(&u->sink->sample_spec);
+
+ /* No blocksize changes or ringbuffer modifications during rendering, please. */
pthread_mutex_lock(&u->mutex);
-
- if (u->frames_requested > 0) {
- unsigned fs;
+
+ /* Do something only if there's room in the ringbuffer. */
+ if (jack_ringbuffer_write_space(u->ringbuffer) >= (u->blocksize * fs)) {
jack_nframes_t frame_idx;
+ jack_ringbuffer_data_t buffer[2]; /* The write area in the ringbuffer may be
+ split in two parts. */
pa_memchunk chunk;
- fs = pa_frame_size(&u->sink->sample_spec);
+ pa_sink_render_full(u->sink, u->blocksize * fs, &chunk);
+
+ jack_ringbuffer_get_write_vector(u->ringbuffer, buffer);
- pa_sink_render_full(u->sink, u->frames_requested * fs, &chunk);
-
- for (frame_idx = 0; frame_idx < u->frames_requested; frame_idx ++) {
- unsigned c;
-
+ /* Write to the first part of the buffer until the entire chunk is written or the first part ends. */
+ for (frame_idx = 0; (frame_idx < u->blocksize) && ((frame_idx * fs) < buffer[0].len); frame_idx++) {
for (c = 0; c < u->channels; c++) {
float *s = ((float*) ((uint8_t*) chunk.memblock->data + chunk.index)) + (frame_idx * u->channels) + c;
- float *d = ((float*) u->buffer[c]) + frame_idx;
+ float *d = ((float*) buffer[0].buf) + (frame_idx * u->channels) + c;
*d = *s;
}
}
+
+ /* Write to the second part of the buffer until the entire chunk is written. */
+ for (; frame_idx < u->blocksize; frame_idx++) {
+ for (c = 0; c < u->channels; c++) {
+ float *s = ((float*) ((uint8_t*) chunk.memblock->data + chunk.index)) + (frame_idx * u->channels) + c;
+ float *d = ((float*) (buffer[1].buf - buffer[0].len)) + (frame_idx * u->channels) + c;
+
+ *d = *s;
+ }
+ }
+
+ jack_ringbuffer_write_advance(u->ringbuffer, u->blocksize * fs);
pa_memblock_unref(chunk.memblock);
-
- u->frames_requested = 0;
-
- pthread_cond_signal(&u->cond);
}
-
+
pthread_mutex_unlock(&u->mutex);
}
@@ -181,55 +201,128 @@
static int jack_process(jack_nframes_t nframes, void *arg) {
struct userdata *u = arg;
- assert(u);
-
- if (jack_transport_query(u->client, NULL) == JackTransportRolling) {
- unsigned c;
-
- pthread_mutex_lock(&u->mutex);
-
- u->frames_requested = nframes;
-
+ jack_nframes_t frame_idx;
+ jack_ringbuffer_data_t data[2]; /* In case the readable area in the ringbuffer is non-continuous,
+ * the data will be split in two parts. */
+ unsigned fs;
+ unsigned c;
+
+ fs = pa_frame_size(&u->sink->sample_spec);
+
+ if (u->bufferptrs_are_dirty) {
for (c = 0; c < u->channels; c++) {
+ assert(u->port[c]);
u->buffer[c] = jack_port_get_buffer(u->port[c], nframes);
- assert(u->buffer[c]);
}
+ u->bufferptrs_are_dirty = 0;
+ }
+
+ jack_ringbuffer_get_read_vector(u->ringbuffer, data);
+
+ /* We only copy either whole bufferfuls of valid data or zeros only.
+ * Partial blocks shouldn't even be possible, since the ringbuffer size is a power of two
+ * as is the blocksize and the data gets written to the ringbuffer one block at a time. */
+
+ if ((data[0].len + data[1].len) < (nframes * fs)) {
+ /* Write zeros and jump out of the function.
+ * User hears a moment of silence and the audio source notices nothing. */
+
+ for (frame_idx = 0; frame_idx < nframes; frame_idx++) {
+ for (c = 0; c < u->channels; c++) {
+ float *d = ((float*) u->buffer[c]) + frame_idx;
+
+ *d = 0.0;
+ }
+ }
+
+ return 0;
+ }
+
+ /* Copy from the first part of data until enough frames are copied or the first part ends. */
+ for (frame_idx = 0; frame_idx < nframes && (frame_idx * fs) < data[0].len; frame_idx++) {
+ for (c = 0; c < u->channels; c++) {
+ float *s = ((float*) data[0].buf) + (frame_idx * u->channels) + c;
+ float *d = ((float*) u->buffer[c]) + frame_idx;
+
+ *d = *s;
+ }
+ }
+
+ /* Copy from the second part of data until enough frames are copied. */
+ for (; frame_idx < nframes; frame_idx++) {
+ for (c = 0; c < u->channels; c++) {
+ float *s = ((float*) (data[1].buf - data[0].len)) + (frame_idx * u->channels) + c;
+ float *d = ((float*) u->buffer[c]) + frame_idx;
+
+ *d = *s;
+ }
+ }
+
+ jack_ringbuffer_read_advance(u->ringbuffer, nframes * fs);
+ request_render(u);
+
+ return 0;
+}
+
+static int jack_blocksize_cb(jack_nframes_t nframes, void *arg) {
+ /* This gets called in the processing thread, so do we have to be realtime safe?
+ * No, we can do whatever we want. User gets silence while we do it.
+ *
+ * In addition to just updating the blocksize field in userdata, we have to create
+ * a new ringbuffer, if the new blocksize is bigger than the old ringbuffer size. */
+ struct userdata *u = arg;
+ unsigned fs;
+
+ assert(u);
+ fs = pa_frame_size(&u->sink->sample_spec);
+
+ /* We don't want to change the blocksize and the ringbuffer while rendering
+ * is going on. */
+ pthread_mutex_lock(&u->mutex);
+
+ u->bufferptrs_are_dirty = 1;
+
+ u->blocksize = nframes;
+
+ if ((u->ringbuffer->size / fs) <= nframes) {
+ /* We have to create a new ringbuffer. What are we going to do with the old
+ * data in the old buffer? We throw it away, because we're lazy coders.
+ * The listening experience is likely to get ruined anyway during the
+ * blocksize change. */
+ jack_ringbuffer_free(u->ringbuffer);
+
+ if (!(u->ringbuffer = jack_ringbuffer_create((nframes + 1) * fs))) {
+ pa_log("jack_ringbuffer_create() failed while changing the blocksize, module exiting.");
+ jack_client_close(u->client);
+ u->quit_requested = 1;
+ }
+
+ pa_log_info("buffersize is %u frames.", u->ringbuffer->size / fs);
+
+ /* Fill the new buffer. */
request_render(u);
-
- pthread_cond_wait(&u->cond, &u->mutex);
-
- u->frames_in_buffer = nframes;
- u->timestamp = jack_get_current_transport_frame(u->client);
-
- pthread_mutex_unlock(&u->mutex);
}
-
+
+ pthread_mutex_unlock(&u->mutex);
+
return 0;
}
static pa_usec_t sink_get_latency_cb(pa_sink *s) {
+ /* The latency is the sum of the first channel's latency, blocksize of jack and the ringbuffer size.
+ * Maybe instead of using just the first channel, the max of all channels' latencies should be used?*/
struct userdata *u;
- jack_nframes_t n, l, d;
+ jack_nframes_t l;
+ unsigned fs;
assert(s);
u = s->userdata;
+ fs = pa_frame_size(&s->sample_spec);
- if (jack_transport_query(u->client, NULL) != JackTransportRolling)
- return 0;
+ l = jack_port_get_total_latency(u->client, u->port[0]) + u->blocksize + u->ringbuffer->size / fs;
- n = jack_get_current_transport_frame(u->client);
-
- if (n < u->timestamp)
- return 0;
-
- d = n - u->timestamp;
- l = jack_port_get_total_latency(u->client, u->port[0]) + u->frames_in_buffer;
-
- if (d >= l)
- return 0;
-
- return pa_bytes_to_usec((l - d) * pa_frame_size(&s->sample_spec), &s->sample_spec);
+ return pa_bytes_to_usec(l * fs, &s->sample_spec);
}
static void jack_error_func(const char*t) {
@@ -244,6 +337,7 @@
jack_status_t status;
const char *server_name, *client_name;
uint32_t channels = 0;
+ uint32_t buffersize = 4096;
int do_connect = 1;
unsigned i;
const char **ports = NULL, **p;
@@ -273,9 +367,8 @@
u->module = m;
u->pipe_fds[0] = u->pipe_fds[1] = -1;
u->pipe_fd_type = 0;
-
+
pthread_mutex_init(&u->mutex, NULL);
- pthread_cond_init(&u->cond, NULL);
if (pipe(u->pipe_fds) < 0) {
pa_log("pipe() failed: %s", pa_cstrerror(errno));
@@ -283,7 +376,7 @@
}
pa_make_nonblock_fd(u->pipe_fds[1]);
-
+
if (!(u->client = jack_client_open(client_name, server_name ? JackServerName : JackNullOption, &status, server_name))) {
pa_log("jack_client_open() failed.");
goto fail;
@@ -316,28 +409,49 @@
ss.format = PA_SAMPLE_FLOAT32NE;
assert(pa_sample_spec_valid(&ss));
-
+
for (i = 0; i < ss.channels; i++) {
if (!(u->port[i] = jack_port_register(u->client, pa_channel_position_to_string(map.map[i]), JACK_DEFAULT_AUDIO_TYPE, JackPortIsOutput|JackPortIsTerminal, 0))) {
pa_log("jack_port_register() failed.");
goto fail;
}
}
+
+ u->bufferptrs_are_dirty = 1;
+
+ if (pa_modargs_get_value_u32(ma, "buffersize", &buffersize) < 0 || buffersize <= 0) {
+ pa_log("failed to parse buffersize= argument.");
+ goto fail;
+ }
if (!(u->sink = pa_sink_new(c, __FILE__, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME), 0, &ss, &map))) {
pa_log("failed to create sink.");
goto fail;
}
-
+
u->sink->userdata = u;
pa_sink_set_owner(u->sink, m);
pa_sink_set_description(u->sink, t = pa_sprintf_malloc("Jack sink (%s)", jack_get_client_name(u->client)));
pa_xfree(t);
u->sink->get_latency = sink_get_latency_cb;
-
+
+ u->blocksize = jack_get_buffer_size(u->client);
+
+ if (buffersize <= u->blocksize) {
+ buffersize = u->blocksize + 1;
+ }
+
+ if (!(u->ringbuffer = jack_ringbuffer_create(buffersize * pa_frame_size(&u->sink->sample_spec)))) {
+ pa_log("jack_ringbuffer_create() failed");
+ goto fail;
+ }
+
+ pa_log_info("buffersize is %u frames.", u->ringbuffer->size / pa_frame_size(&u->sink->sample_spec));
+
jack_set_process_callback(u->client, jack_process, u);
+ jack_set_buffer_size_callback(u->client, jack_blocksize_cb, u);
jack_on_shutdown(u->client, jack_shutdown, u);
-
+
if (jack_activate(u->client)) {
pa_log("jack_activate() failed");
goto fail;
@@ -362,7 +476,12 @@
}
u->io_event = c->mainloop->io_new(c->mainloop, u->pipe_fds[0], PA_IO_EVENT_INPUT, io_event_cb, u);
-
+
+ /* Fill the ringbuffer. */
+ for (i = 0; i < (u->ringbuffer->size / pa_frame_size(&u->sink->sample_spec) / u->blocksize); i++) {
+ request_render(u);
+ }
+
free(ports);
pa_modargs_free(ma);
@@ -396,13 +515,15 @@
pa_sink_disconnect(u->sink);
pa_sink_unref(u->sink);
}
+
+ if (u->ringbuffer)
+ jack_ringbuffer_free(u->ringbuffer);
if (u->pipe_fds[0] >= 0)
close(u->pipe_fds[0]);
if (u->pipe_fds[1] >= 0)
close(u->pipe_fds[1]);
-
+
pthread_mutex_destroy(&u->mutex);
- pthread_cond_destroy(&u->cond);
pa_xfree(u);
}
_______________________________________________
pulseaudio-discuss mailing list
[email protected]
https://tango.0pointer.de/mailman/listinfo/pulseaudio-discuss
