From eaafb79858fff11add29c8a1126a18e90d7785f9 Mon Sep 17 00:00:00 2001 From: Tanu Kaskinen Date: Mon, 20 Aug 2007 06:22:21 +0000 Subject: Modified the JACK sink heavily: * Made the sink realtime-safe. * To achieve the previous item, internal buffering was be added. New module argument: buffersize. * Removed the user's need to set the JACK transport to playing state before he could hear anything from PulseAudio. * In process of achieving the previous item, latency calculation got more inaccurate: the reported latency is now always a multiple of the JACK processing block size, and constant. * The JACK ports now have a running numbering in their names. git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1680 fefdeb5f-60dc-0310-8127-8f9354f1896f --- src/modules/module-jack-sink.c | 957 ++++++++++++++++++++++++++++++----------- 1 file changed, 716 insertions(+), 241 deletions(-) diff --git a/src/modules/module-jack-sink.c b/src/modules/module-jack-sink.c index c6a7e33f..09a72e3a 100644 --- a/src/modules/module-jack-sink.c +++ b/src/modules/module-jack-sink.c @@ -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 @@ -25,33 +25,32 @@ #include #endif -#include -#include -#include +#include #include #include -#include -#include -#include -#include -#include #include +#include +#include +#include +#include +#include #include -#include -#include #include #include +#include +#include #include +#include +#include #include -#include -#include +#include #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( @@ -59,335 +58,806 @@ PA_MODULE_USAGE( "server_name= " "client_name= " "channels= " - "connect= " + "connect= " + "buffersize= " "channel_map=") #define DEFAULT_SINK_NAME "jack_out" +#define DEFAULT_CLIENT_NAME "PulseAudio(output)" +#define DEFAULT_RINGBUFFER_SIZE 4096 -struct userdata { - pa_core *core; - pa_module *module; +struct userdata { pa_sink *sink; unsigned channels; + unsigned frame_size; + + jack_port_t* j_ports[PA_CHANNELS_MAX]; + jack_client_t *j_client; - jack_port_t* port[PA_CHANNELS_MAX]; - jack_client_t *client; + jack_nframes_t j_buffersize; - pthread_mutex_t mutex; - pthread_cond_t cond; + /* For avoiding j_buffersize changes at a wrong moment. */ + pthread_mutex_t buffersize_mutex; - void * buffer[PA_CHANNELS_MAX]; - jack_nframes_t frames_requested; + /* The intermediate store where the pulse side writes to and the jack side + reads from. */ + jack_ringbuffer_t* ringbuffer; + + /* For signaling when there's room in the ringbuffer. */ + pthread_mutex_t cond_mutex; + pthread_cond_t ringbuffer_cond; + + pthread_t filler_thread; /* Keeps the ringbuffer filled. */ + + int ringbuffer_is_full; + int filler_thread_is_running; 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; + +struct options { + char* sink_name; + int sink_name_given; + + char* server_name; /* May be NULL */ + int server_name_given; + + char* client_name; + int client_name_given; + + unsigned channels; + int channels_given; + + int connect; + int connect_given; + + unsigned buffersize; + int buffersize_given; + + pa_channel_map map; + int map_given; }; + static const char* const valid_modargs[] = { "sink_name", "server_name", "client_name", "channels", "connect", + "buffersize", "channel_map", NULL }; -static void stop_sink(struct userdata *u) { - assert (u); - - jack_client_close(u->client); - u->client = NULL; - u->core->mainloop->io_free(u->io_event); - u->io_event = NULL; - pa_sink_disconnect(u->sink); - pa_sink_unref(u->sink); - u->sink = 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; - - assert(m); - assert(e); - assert(flags == PA_IO_EVENT_INPUT); - assert(u); - assert(u->pipe_fds[0] == fd); - - pa_read(fd, &x, 1, &u->pipe_fd_type); - if (u->quit_requested) { - stop_sink(u); - u->quit_requested = 0; - return; - } - - pthread_mutex_lock(&u->mutex); +/* Initialization functions. */ +static int parse_options(struct options* o, const char* argument); +static void set_default_channels(pa_module* self, struct options* o); +static int create_sink(pa_module* self, struct options *o); +static void connect_ports(pa_module* self); +static int start_filling_ringbuffer(pa_module* self); - if (u->frames_requested > 0) { - unsigned fs; - jack_nframes_t frame_idx; - pa_memchunk chunk; +/* Various callbacks. */ +static void jack_error_func(const char* t); +static pa_usec_t sink_get_latency_cb(pa_sink* s); +static int jack_process(jack_nframes_t nframes, void* arg); +static int jack_blocksize_cb(jack_nframes_t nframes, void* arg); +static void jack_shutdown(void* arg); +static void io_event_cb(pa_mainloop_api* m, pa_io_event* e, int fd, + pa_io_event_flags_t flags, void* userdata); - fs = pa_frame_size(&u->sink->sample_spec); +/* The ringbuffer filler thread runs in this function. */ +static void* fill_ringbuffer(void* arg); - pa_sink_render_full(u->sink, u->frames_requested * fs, &chunk); +/* request_render asks asynchronously the mainloop to call io_event_cb. */ +static void request_render(struct userdata* u); - for (frame_idx = 0; frame_idx < u->frames_requested; frame_idx ++) { - unsigned c; - 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; - - *d = *s; - } +int pa__init(pa_core* c, pa_module* self) { + struct userdata* u = NULL; + struct options o; + unsigned i; + + assert(c); + assert(self); + + o.sink_name = NULL; + o.server_name = NULL; + o.client_name = NULL; + + self->userdata = pa_xnew0(struct userdata, 1); + u = self->userdata; + + u->pipe_fds[0] = u->pipe_fds[1] = -1; + u->pipe_fd_type = 0; + u->ringbuffer_is_full = 0; + u->filler_thread_is_running = 0; + u->quit_requested = 0; + pthread_mutex_init(&u->buffersize_mutex, NULL); + pthread_mutex_init(&u->cond_mutex, NULL); + pthread_cond_init(&u->ringbuffer_cond, NULL); + + if (parse_options(&o, self->argument) != 0) + goto fail; + + jack_set_error_function(jack_error_func); + + if (!(u->j_client = jack_client_open( + o.client_name, + o.server_name ? JackServerName : JackNullOption, + NULL, o.server_name))) { + pa_log_error("jack_client_open() failed."); + goto fail; + } + pa_log_info("Successfully connected as '%s'", + jack_get_client_name(u->j_client)); + + if (!o.channels_given) + set_default_channels(self, &o); + + u->channels = o.channels; + + if (!o.map_given) + pa_channel_map_init_auto(&o.map, u->channels, PA_CHANNEL_MAP_ALSA); + + for (i = 0; i < u->channels; i++) { + char* port_name = pa_sprintf_malloc( + "out_%i:%s", i+1, + pa_channel_position_to_string(o.map.map[i])); + + if (!(u->j_ports[i] = jack_port_register( + u->j_client, port_name, + JACK_DEFAULT_AUDIO_TYPE, + JackPortIsOutput|JackPortIsTerminal, 0))) { + pa_log("jack_port_register() failed."); + goto fail; } + + pa_xfree(port_name); + } + + if (pipe(u->pipe_fds) < 0) { + pa_log("pipe() failed: %s", pa_cstrerror(errno)); + goto fail; + } + pa_make_nonblock_fd(u->pipe_fds[1]); + + if (create_sink(self, &o) != 0) + goto fail; - pa_memblock_unref(chunk.memblock); - - u->frames_requested = 0; - - pthread_cond_signal(&u->cond); + u->frame_size = pa_frame_size(&u->sink->sample_spec); + u->j_buffersize = jack_get_buffer_size(u->j_client); + + /* If the ringbuffer size were equal to the jack buffer size, a full block + would never fit in the ringbuffer, because the ringbuffer can never be + totally full: one slot is always wasted. */ + if (o.buffersize <= u->j_buffersize) { + o.buffersize = u->j_buffersize + 1; + } + /* The actual ringbuffer size will be rounded up to the nearest power of + two. */ + if (!(u->ringbuffer = jack_ringbuffer_create( + o.buffersize * u->frame_size))) { + pa_log("jack_ringbuffer_create() failed."); + goto fail; + } + assert((u->ringbuffer->size % sizeof(float)) == 0); + pa_log_info("buffersize is %u frames (%u samples, %u bytes).", + u->ringbuffer->size / u->frame_size, + u->ringbuffer->size / sizeof(float), + u->ringbuffer->size); + + jack_set_process_callback(u->j_client, jack_process, u); + jack_set_buffer_size_callback(u->j_client, jack_blocksize_cb, u); + jack_on_shutdown(u->j_client, jack_shutdown, u); + + if (jack_activate(u->j_client)) { + pa_log("jack_activate() failed."); + goto fail; } - pthread_mutex_unlock(&u->mutex); -} + if (o.connect) + connect_ports(self); -static void request_render(struct userdata *u) { - char c = 'x'; - assert(u); + u->io_event = c->mainloop->io_new(c->mainloop, u->pipe_fds[0], + PA_IO_EVENT_INPUT, io_event_cb, self); + + if (start_filling_ringbuffer(self) != 0) + goto fail; - assert(u->pipe_fds[1] >= 0); - pa_write(u->pipe_fds[1], &c, 1, &u->pipe_fd_type); -} + pa_xfree(o.sink_name); + pa_xfree(o.server_name); + pa_xfree(o.client_name); + + return 0; -static void jack_shutdown(void *arg) { - struct userdata *u = arg; - assert(u); +fail: + pa_xfree(o.sink_name); + pa_xfree(o.server_name); + pa_xfree(o.client_name); + pa__done(c, self); - u->quit_requested = 1; - request_render(u); + return -1; } -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); +static int parse_options(struct options* o, const char* argument) { + pa_modargs *ma = NULL; + const char* arg_val; + pa_strbuf* strbuf; + + assert(o); - u->frames_requested = nframes; + if (!(ma = pa_modargs_new(argument, valid_modargs))) { + pa_log_error("Failed to parse module arguments."); + goto fail; + } - for (c = 0; c < u->channels; c++) { - u->buffer[c] = jack_port_get_buffer(u->port[c], nframes); - assert(u->buffer[c]); + strbuf = pa_strbuf_new(); + if ((arg_val = pa_modargs_get_value(ma, "sink_name", NULL))) { + pa_strbuf_puts(strbuf, arg_val); + o->sink_name = pa_strbuf_tostring(strbuf); + o->sink_name_given = 1; + } else { + pa_strbuf_puts(strbuf, DEFAULT_SINK_NAME); + o->sink_name = pa_strbuf_tostring(strbuf); + o->sink_name_given = 0; + } + pa_strbuf_free(strbuf); + + strbuf = pa_strbuf_new(); + if ((arg_val = pa_modargs_get_value(ma, "server_name", NULL))) { + pa_strbuf_puts(strbuf, arg_val); + o->server_name = pa_strbuf_tostring(strbuf); + o->server_name_given = 1; + } else { + o->server_name = NULL; + o->server_name_given = 0; + } + pa_strbuf_free(strbuf); + + strbuf = pa_strbuf_new(); + if ((arg_val = pa_modargs_get_value(ma, "client_name", NULL))) { + pa_strbuf_puts(strbuf, arg_val); + o->client_name = pa_strbuf_tostring(strbuf); + o->client_name_given = 1; + } else { + pa_strbuf_puts(strbuf, DEFAULT_CLIENT_NAME); + o->client_name = pa_strbuf_tostring(strbuf); + o->client_name_given = 1; + } + pa_strbuf_free(strbuf); + + if (pa_modargs_get_value(ma, "channels", NULL)) { + o->channels_given = 1; + if (pa_modargs_get_value_u32(ma, "channels", &o->channels) < 0 || + o->channels == 0 || + o->channels >= PA_CHANNELS_MAX) { + pa_log_error("Failed to parse the \"channels\" argument."); + goto fail; } + } else { + o->channels = 0; /* The actual default value is the number of physical + input ports in jack (unknown at the moment), or if + that's zero, then the default_sample_spec.channels + of the core. */ + o->channels_given = 0; + } - 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); + if (pa_modargs_get_value(ma, "connect", NULL)) { + o->connect_given = 1; + if (pa_modargs_get_value_boolean(ma, "connect", &o->connect) < 0) { + pa_log_error("Failed to parse the \"connect\" argument."); + goto fail; + } + } else { + o->connect = 1; + o->connect_given = 0; } - return 0; -} + if (pa_modargs_get_value(ma, "buffersize", NULL)) { + o->buffersize_given = 1; + if (pa_modargs_get_value_u32(ma, "buffersize", &o->buffersize) < 0) { + pa_log_error("Failed to parse the \"buffersize\" argument."); + goto fail; + } + } else { + o->buffersize = DEFAULT_RINGBUFFER_SIZE; + o->buffersize_given = 0; + } -static pa_usec_t sink_get_latency_cb(pa_sink *s) { - struct userdata *u; - jack_nframes_t n, l, d; + if (pa_modargs_get_value(ma, "channel_map", NULL)) { + o->map_given = 1; + if (pa_modargs_get_channel_map(ma, &o->map) < 0) { + pa_log_error("Failed to parse the \"channel_map\" argument."); + goto fail; + } - assert(s); - u = s->userdata; + /* channel_map specifies the channel count too. */ + if (o->channels_given && (o->channels != o->map.channels)) { + pa_log_error( + "\"channels\" and \"channel_map\" arguments conficted. If you " + "use the \"channel_map\" argument, you can omit the " + "\"channels\" argument."); + goto fail; + } else { + o->channels = o->map.channels; + o->channels_given = 1; + } + } else { + /* The actual default value is the default alsa mappings, but that + can't be set until the channel count is known. Here we initialize + the map to some valid value, although the value won't be used. */ + pa_channel_map_init_stereo(&o->map); + o->map_given = 0; + } - if (jack_transport_query(u->client, NULL) != JackTransportRolling) - return 0; + pa_modargs_free(ma); - n = jack_get_current_transport_frame(u->client); + return 0; - if (n < u->timestamp) - return 0; +fail: + if (ma) + pa_modargs_free(ma); - d = n - u->timestamp; - l = jack_port_get_total_latency(u->client, u->port[0]) + u->frames_in_buffer; + return -1; +} - if (d >= l) - return 0; - return pa_bytes_to_usec((l - d) * pa_frame_size(&s->sample_spec), &s->sample_spec); +static void set_default_channels(pa_module* self, struct options* o) { + struct userdata* u; + const char **ports, **p; + + assert(self); + assert(o); + assert(self->userdata); + + u = self->userdata; + + assert(u->j_client); + assert(self->core); + + o->channels = 0; + + ports = jack_get_ports(u->j_client, NULL, JACK_DEFAULT_AUDIO_TYPE, + JackPortIsPhysical|JackPortIsInput); + + for (p = ports; *p; p++) + o->channels++; + + free(ports); + + if (o->channels >= PA_CHANNELS_MAX) + o->channels = PA_CHANNELS_MAX - 1; + + if (o->channels == 0) + o->channels = self->core->default_sample_spec.channels; } -static void jack_error_func(const char*t) { - pa_log_warn("JACK error >%s<", t); -} -int pa__init(pa_core *c, pa_module*m) { - struct userdata *u = NULL; +static int create_sink(pa_module* self, struct options* o) { + struct userdata* u; pa_sample_spec ss; - pa_channel_map map; - pa_modargs *ma = NULL; - jack_status_t status; - const char *server_name, *client_name; - uint32_t channels = 0; - int do_connect = 1; - unsigned i; - const char **ports = NULL, **p; char *t; + + assert(self); + assert(o); + assert(self->userdata); + + u = self->userdata; + + assert(u->j_client); + + ss.channels = u->channels; + ss.rate = jack_get_sample_rate(u->j_client); + ss.format = PA_SAMPLE_FLOAT32NE; + assert(pa_sample_spec_valid(&ss)); - assert(c); - assert(m); - - jack_set_error_function(jack_error_func); - - if (!(ma = pa_modargs_new(m->argument, valid_modargs))) { - pa_log("failed to parse module arguments."); - goto fail; + if (!(u->sink = pa_sink_new(self->core, __FILE__, o->sink_name, 0, &ss, + &o->map))) { + pa_log("failed to create sink."); + return -1; } + + u->sink->userdata = u; + pa_sink_set_owner(u->sink, self); + + pa_sink_set_description( + u->sink, + t = pa_sprintf_malloc("Jack sink (%s)", + jack_get_client_name(u->j_client))); + pa_xfree(t); + + u->sink->get_latency = sink_get_latency_cb; + + return 0; +} - if (pa_modargs_get_value_boolean(ma, "connect", &do_connect) < 0) { - pa_log("failed to parse connect= argument."); - goto fail; + +static void connect_ports(pa_module* self) { + struct userdata* u; + unsigned i; + const char **ports, **p; + + assert(self); + assert(self->userdata); + + u = self->userdata; + + assert(u->j_client); + + ports = jack_get_ports(u->j_client, NULL, JACK_DEFAULT_AUDIO_TYPE, + JackPortIsPhysical|JackPortIsInput); + + for (i = 0, p = ports; i < u->channels; i++, p++) { + assert(u->j_ports[i]); + + if (!*p) { + pa_log("Not enough physical output ports, leaving unconnected."); + break; + } + + pa_log_info("connecting %s to %s", + jack_port_name(u->j_ports[i]), *p); + + if (jack_connect(u->j_client, jack_port_name(u->j_ports[i]), *p)) { + pa_log("Failed to connect %s to %s, leaving unconnected.", + jack_port_name(u->j_ports[i]), *p); + break; + } } + + free(ports); +} - server_name = pa_modargs_get_value(ma, "server_name", NULL); - client_name = pa_modargs_get_value(ma, "client_name", "PulseAudio"); - u = pa_xnew0(struct userdata, 1); - m->userdata = u; - u->core = c; - u->module = m; - u->pipe_fds[0] = u->pipe_fds[1] = -1; - u->pipe_fd_type = 0; +static int start_filling_ringbuffer(pa_module* self) { + struct userdata* u; + pthread_attr_t thread_attributes; - pthread_mutex_init(&u->mutex, NULL); - pthread_cond_init(&u->cond, NULL); + assert(self); + assert(self->userdata); - if (pipe(u->pipe_fds) < 0) { - pa_log("pipe() failed: %s", pa_cstrerror(errno)); + u = self->userdata; + + pthread_attr_init(&thread_attributes); + + if (pthread_attr_setinheritsched(&thread_attributes, + PTHREAD_INHERIT_SCHED) != 0) { + pa_log("pthread_attr_setinheritsched() failed."); goto fail; } - - 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."); + else if (pthread_create(&u->filler_thread, &thread_attributes, + fill_ringbuffer, u) != 0) { + pa_log("pthread_create() failed."); goto fail; } + + u->filler_thread_is_running = 1; + + pthread_attr_destroy(&thread_attributes); - ports = jack_get_ports(u->client, NULL, NULL, JackPortIsPhysical|JackPortIsInput); - - channels = 0; - for (p = ports; *p; p++) - channels++; + return 0; + +fail: + pthread_attr_destroy(&thread_attributes); + return -1; +} - if (!channels) - channels = c->default_sample_spec.channels; - if (pa_modargs_get_value_u32(ma, "channels", &channels) < 0 || channels <= 0 || channels >= PA_CHANNELS_MAX) { - pa_log("failed to parse channels= argument."); - goto fail; - } +static void jack_error_func(const char* t) { + pa_log_warn("JACK error >%s<", t); +} - pa_channel_map_init_auto(&map, channels, PA_CHANNEL_MAP_ALSA); - if (pa_modargs_get_channel_map(ma, &map) < 0 || map.channels != channels) { - pa_log("failed to parse channel_map= argument."); - goto fail; - } - pa_log_info("Successfully connected as '%s'", jack_get_client_name(u->client)); +static pa_usec_t sink_get_latency_cb(pa_sink* s) { + /* The latency is approximately the sum of the first port's latency, + buffersize of jack and the ringbuffer size. Maybe instead of using just + the first port, the max of all ports' latencies should be used? */ + struct userdata* u; + jack_nframes_t l; + + assert(s); + assert(s->userdata); - ss.channels = u->channels = channels; - ss.rate = jack_get_sample_rate(u->client); - ss.format = PA_SAMPLE_FLOAT32NE; + u = s->userdata; + + l = jack_port_get_total_latency(u->j_client, u->j_ports[0]) + + u->j_buffersize + u->ringbuffer->size / u->frame_size; + + return pa_bytes_to_usec(l * u->frame_size, &s->sample_spec); +} - 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; - } +static int jack_process(jack_nframes_t nframes, void* arg) { + struct userdata* u = arg; + float* j_buffers[PA_CHANNELS_MAX]; + unsigned nsamples = u->channels * nframes; + unsigned sample_idx_part1, sample_idx_part2; + 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 chan; + unsigned samples_left_over; + + for (chan = 0; chan < u->channels; chan++) { + j_buffers[chan] = jack_port_get_buffer(u->j_ports[chan], nframes); + } + + jack_ringbuffer_get_read_vector(u->ringbuffer, data); + + /* We assume that the possible discontinuity doesn't happen in the middle + * of a sample. Should be a safe assumption. */ + assert(((data[0].len % sizeof(float)) == 0) || + (data[1].len == 0)); + + /* Copy from the first part of data until enough samples are copied or the + first part ends. */ + sample_idx_part1 = 0; + chan = 0; + frame_idx = 0; + while (sample_idx_part1 < nsamples && + ((sample_idx_part1 + 1) * sizeof(float)) <= data[0].len) { + float *s = ((float*) data[0].buf) + sample_idx_part1; + float *d = j_buffers[chan] + frame_idx; + *d = *s; + + sample_idx_part1++; + chan = (chan + 1) % u->channels; + frame_idx = sample_idx_part1 / u->channels; + } + + samples_left_over = nsamples - sample_idx_part1; + + /* Copy from the second part of data until enough samples are copied or the + second part ends. */ + sample_idx_part2 = 0; + while (sample_idx_part2 < samples_left_over && + ((sample_idx_part2 + 1) * sizeof(float)) <= data[1].len) { + float *s = ((float*) data[1].buf) + sample_idx_part2; + float *d = j_buffers[chan] + frame_idx; + *d = *s; + + sample_idx_part2++; + chan = (chan + 1) % u->channels; + frame_idx = (sample_idx_part1 + sample_idx_part2) / u->channels; } + + samples_left_over -= sample_idx_part2; + + /* If there's still samples left, fill the buffers with zeros. */ + while (samples_left_over > 0) { + float *d = j_buffers[chan] + frame_idx; + *d = 0.0; + + samples_left_over--; + chan = (chan + 1) % u->channels; + frame_idx = (nsamples - samples_left_over) / u->channels; + } + + jack_ringbuffer_read_advance( + u->ringbuffer, (sample_idx_part1 + sample_idx_part2) * sizeof(float)); + + /* Tell the rendering part that there is room in the ringbuffer. */ + u->ringbuffer_is_full = 0; + pthread_cond_signal(&u->ringbuffer_cond); + + return 0; +} - 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; + +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 j_buffersize field in userdata, we have + to create a new ringbuffer, if the new buffer size is bigger or equal to + the old ringbuffer size. */ + struct userdata* u = arg; + + assert(u); + + /* We don't want to change the blocksize and the ringbuffer while rendering + is going on. */ + pthread_mutex_lock(&u->buffersize_mutex); + + u->j_buffersize = nframes; + + if ((u->ringbuffer->size / u->frame_size) <= 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); + + /* The actual ringbuffer size will be rounded up to the nearest power + of two. */ + if (!(u->ringbuffer = + jack_ringbuffer_create((nframes + 1) * u->frame_size))) { + pa_log_error( + "jack_ringbuffer_create() failed while changing jack's buffer " + "size, module exiting."); + jack_client_close(u->j_client); + u->quit_requested = 1; + } + assert((u->ringbuffer->size % sizeof(float)) == 0); + pa_log_info("buffersize is %u frames (%u samples, %u bytes).", + u->ringbuffer->size / u->frame_size, + u->ringbuffer->size / sizeof(float), + u->ringbuffer->size); } + + pthread_mutex_unlock(&u->buffersize_mutex); + + return 0; +} - 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; - jack_set_process_callback(u->client, jack_process, u); - jack_on_shutdown(u->client, jack_shutdown, u); +static void jack_shutdown(void* arg) { + struct userdata* u = arg; + assert(u); - if (jack_activate(u->client)) { - pa_log("jack_activate() failed"); - goto fail; - } + u->quit_requested = 1; + request_render(u); +} - if (do_connect) { - for (i = 0, p = ports; i < ss.channels; i++, p++) { - if (!*p) { - pa_log("not enough physical output ports, leaving unconnected."); - break; - } +static void io_event_cb(pa_mainloop_api* m, pa_io_event* e, int fd, + pa_io_event_flags_t flags, void* userdata) { + pa_module* self = userdata; + struct userdata* u; + char x; + jack_ringbuffer_data_t buffer[2]; /* The write area in the ringbuffer may + be split in two parts. */ + pa_memchunk chunk; /* This is the data source. */ + unsigned part1_length, part2_length; + unsigned sample_idx_part1, sample_idx_part2; + unsigned chan; + unsigned frame_size; + int rem; + + assert(m); + assert(e); + assert(flags == PA_IO_EVENT_INPUT); + assert(self); + assert(self->userdata); - pa_log_info("connecting %s to %s", jack_port_name(u->port[i]), *p); + u = self->userdata; + + assert(u->pipe_fds[0] == fd); - if (jack_connect(u->client, jack_port_name(u->port[i]), *p)) { - pa_log("failed to connect %s to %s, leaving unconnected.", jack_port_name(u->port[i]), *p); - break; - } - } + pa_read(fd, &x, 1, &u->pipe_fd_type); + if (u->quit_requested) { + pa_module_unload_request(self); + return; } - u->io_event = c->mainloop->io_new(c->mainloop, u->pipe_fds[0], PA_IO_EVENT_INPUT, io_event_cb, u); - - free(ports); - pa_modargs_free(ma); + frame_size = u->frame_size; + + /* No blocksize changes during rendering, please. */ + pthread_mutex_lock(&u->buffersize_mutex); + + jack_ringbuffer_get_write_vector(u->ringbuffer, buffer); + assert(((buffer[0].len % sizeof(float)) == 0) || (buffer[1].len == 0)); + + part1_length = buffer[0].len / sizeof(float); + part2_length = buffer[1].len / sizeof(float); + + /* If the amount of free space is not a multiple of the frame size, we have + to adjust the lengths in order to not get confused with which sample is + which channel. */ + if ((rem = (part1_length + part2_length) % u->channels) != 0) { + if (part2_length >= rem) { + part2_length -= rem; + } else { + part1_length -= rem - part2_length; + part2_length = 0; + } + } + + /* pa_sink_render_full doesn't accept zero length, so we have do the + copying only if there's data to copy, which actually makes a kind of + sense. */ + if (part1_length > 0 || part2_length > 0) { + pa_sink_render_full(u->sink, + (part1_length + part2_length) * sizeof(float), + &chunk); + + /* Write to the first part of the buffer. */ + for (sample_idx_part1 = 0; + sample_idx_part1 < part1_length; + sample_idx_part1++) { + float *s = + ((float*) ((uint8_t*) chunk.memblock->data + chunk.index)) + + sample_idx_part1; + float *d = ((float*) buffer[0].buf) + sample_idx_part1; + *d = *s; + } + + /* Write to the second part of the buffer. */ + for (sample_idx_part2 = 0; + sample_idx_part2 < part2_length; + sample_idx_part2++) { + float *s = + ((float*) ((uint8_t*) chunk.memblock->data + chunk.index)) + + sample_idx_part1 + sample_idx_part2; + float *d = ((float*) buffer[1].buf) + sample_idx_part2; + *d = *s; + } + + pa_memblock_unref(chunk.memblock); + + jack_ringbuffer_write_advance( + u->ringbuffer, (part1_length + part2_length) * sizeof(float)); + } + + /* Blocksize can be changed again. */ + pthread_mutex_unlock(&u->buffersize_mutex); +} - return 0; -fail: - if (ma) - pa_modargs_free(ma); +static void* fill_ringbuffer(void* arg) { + struct userdata* u = arg; + + assert(u); + + while (!u->quit_requested) { + if (u->ringbuffer_is_full) { + pthread_mutex_lock(&u->cond_mutex); + pthread_cond_wait(&u->ringbuffer_cond, + &u->cond_mutex); + pthread_mutex_unlock(&u->cond_mutex); + } + /* No, it's not full yet, but this must be set to one as soon as + possible, because if the jack thread manages to process another + block before we set this to one, we may end up waiting without + a reason. */ + u->ringbuffer_is_full = 1; - free(ports); + request_render(u); + } + + return NULL; +} - pa__done(c, m); - return -1; +static void request_render(struct userdata* u) { + char c = 'x'; + + assert(u); + + assert(u->pipe_fds[1] >= 0); + pa_write(u->pipe_fds[1], &c, 1, &u->pipe_fd_type); } -void pa__done(pa_core *c, pa_module*m) { - struct userdata *u; - assert(c && m); +void pa__done(pa_core* c, pa_module* self) { + struct userdata* u; + + assert(c); + assert(self); - if (!(u = m->userdata)) + if (!self->userdata) return; - if (u->client) - jack_client_close(u->client); + u = self->userdata; + + if (u->filler_thread_is_running) { + u->quit_requested = 1; + pthread_cond_signal(&u->ringbuffer_cond); + pthread_join(u->filler_thread, NULL); + } + + if (u->j_client) + jack_client_close(u->j_client); if (u->io_event) c->mainloop->io_free(u->io_event); @@ -396,13 +866,18 @@ void pa__done(pa_core *c, pa_module*m) { 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]); + pa_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); + pa_close(u->pipe_fds[1]); + + pthread_mutex_destroy(&u->buffersize_mutex); + pthread_cond_destroy(&u->ringbuffer_cond); + pthread_mutex_destroy(&u->cond_mutex); + pa_xfree(self->userdata); + self->userdata = NULL; } -- cgit