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

1 files 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 <config.h>
 #endif
 
-#include <stdlib.h>
-#include <sys/stat.h>
-#include <stdio.h>
+#include <pthread.h>
 #include <assert.h>
 #include <errno.h>
-#include <string.h>
-#include <fcntl.h>
-#include <unistd.h>
-#include <limits.h>
-#include <pthread.h>
 
 #include <jack/jack.h>
+#include <jack/ringbuffer.h>
+#include <jack/types.h>
 
+#include <pulse/mainloop-api.h>
+#include <pulse/sample.h>
+#include <pulse/channelmap.h>
 #include <pulse/xmalloc.h>
 
-#include <pulsecore/core-error.h>
-#include <pulsecore/iochannel.h>
 #include <pulsecore/sink.h>
 #include <pulsecore/module.h>
+#include <pulsecore/core.h>
+#include <pulsecore/log.h>
 #include <pulsecore/core-util.h>
+#include <pulsecore/core-error.h>
+#include <pulsecore/pipe.h>
 #include <pulsecore/modargs.h>
-#include <pulsecore/log.h>
-#include <pulse/mainloop-api.h>
+#include <pulsecore/strbuf.h>
 
 #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=<jack server name> "
         "client_name=<jack client name> "
         "channels=<number of channels> "
-        "connect=<connect ports?> "
+        "connect=<connect ports automatically?> "
+        "buffersize=<intermediate buffering in frames> "
         "channel_map=<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;
 }