merge 'lennart' branch back into trunk.

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1971 fefdeb5f-60dc-0310-8127-8f9354f1896f

1 files changed, 299 insertions, 726 deletions

diff --git a/src/modules/module-jack-sink.c b/src/modules/module-jack-sink.c
index a40ebe29..5019d656 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, 2007 Lennart Poettering and Tanu Kaskinen
+  Copyright 2006 Lennart Poettering
 
   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,858 +25,431 @@
 #include <config.h>
 #endif
 
-#include <pthread.h>
+#include <stdlib.h>
+#include <sys/stat.h>
+#include <stdio.h>
 #include <assert.h>
 #include <errno.h>
+#include <string.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <limits.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/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/strbuf.h>
+#include <pulsecore/log.h>
+#include <pulsecore/thread.h>
+#include <pulsecore/thread-mq.h>
+#include <pulsecore/rtpoll.h>
+#include <pulsecore/sample-util.h>
 
 #include "module-jack-sink-symdef.h"
 
-PA_MODULE_AUTHOR("Lennart Poettering & Tanu Kaskinen")
-PA_MODULE_DESCRIPTION("Jack Sink")
+/* General overview:
+ *
+ * Because JACK has a very unflexible event loop management, which
+ * doesn't allow us to add our own event sources to the event thread
+ * we cannot use the JACK real-time thread for dispatching our PA
+ * work. Instead, we run an additional RT thread which does most of
+ * the PA handling, and have the JACK RT thread request data from it
+ * via pa_asyncmsgq. The cost is an additional context switch which
+ * should hopefully not be that expensive if RT scheduling is
+ * enabled. A better fix would only be possible with additional event
+ * source support in JACK.
+ */
+
+PA_MODULE_AUTHOR("Lennart Poettering")
+PA_MODULE_DESCRIPTION("JACK Sink")
 PA_MODULE_VERSION(PACKAGE_VERSION)
 PA_MODULE_USAGE(
         "sink_name=<name of sink> "
         "server_name=<jack server name> "
         "client_name=<jack client name> "
         "channels=<number of channels> "
-        "connect=<connect ports automatically?> "
-        "buffersize=<intermediate buffering in frames> "
+        "connect=<connect ports?> "
         "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;
     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;
+    void *buffer[PA_CHANNELS_MAX];
 
-    /* For avoiding j_buffersize changes at a wrong moment. */
-    pthread_mutex_t buffersize_mutex;
+    pa_thread_mq thread_mq;
+    pa_asyncmsgq *jack_msgq;
+    pa_rtpoll *rtpoll;
+    pa_rtpoll_item *rtpoll_item;
 
-    /* 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;
+    pa_thread *thread;
 
-    pthread_t filler_thread; /* Keeps the ringbuffer filled. */
+    jack_nframes_t frames_in_buffer;
+    jack_nframes_t saved_frame_time;
+    pa_bool_t saved_frame_time_valid;
+};
 
-    int ringbuffer_is_full;
-    int filler_thread_is_running;
-    int quit_requested;
+static const char* const valid_modargs[] = {
+    "sink_name",
+    "server_name",
+    "client_name",
+    "channels",
+    "connect",
+    "channel_map",
+    NULL
+};
 
-    int pipe_fd_type;
-    int pipe_fds[2];
-    pa_io_event *io_event;
+enum {
+    SINK_MESSAGE_RENDER = PA_SINK_MESSAGE_MAX,
+    SINK_MESSAGE_ON_SHUTDOWN
 };
 
+static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *memchunk) {
+    struct userdata *u = PA_SINK(o)->userdata;
 
-struct options {
-    char* sink_name;
-    int sink_name_given;
+    switch (code) {
 
-    char* server_name; /* May be NULL */
-    int server_name_given;
+        case SINK_MESSAGE_RENDER:
 
-    char* client_name;
-    int client_name_given;
+            /* Handle the request from the JACK thread */
 
-    unsigned channels;
-    int channels_given;
+            if (u->sink->thread_info.state == PA_SINK_RUNNING) {
+                pa_memchunk chunk;
+                size_t nbytes;
+                void *p;
 
-    int connect;
-    int connect_given;
+                pa_assert(offset > 0);
+                nbytes = offset * pa_frame_size(&u->sink->sample_spec);
 
-    unsigned buffersize;
-    int buffersize_given;
+                pa_sink_render_full(u->sink, nbytes, &chunk);
 
-    pa_channel_map map;
-    int map_given;
-};
+                p = (uint8_t*) pa_memblock_acquire(chunk.memblock) + chunk.index;
+                pa_deinterleave(p, u->buffer, u->channels, sizeof(float), offset);
+                pa_memblock_release(chunk.memblock);
 
+                pa_memblock_unref(chunk.memblock);
+            } else {
+                unsigned c;
+                pa_sample_spec ss;
 
-static const char* const valid_modargs[] = {
-    "sink_name",
-    "server_name",
-    "client_name",
-    "channels",
-    "connect",
-    "buffersize",
-    "channel_map",
-    NULL
-};
+                /* Humm, we're not RUNNING, hence let's write some silence */
 
+                ss = u->sink->sample_spec;
+                ss.channels = 1;
 
-/* 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);
+                for (c = 0; c < u->channels; c++)
+                    pa_silence_memory(u->buffer[c], offset * pa_sample_size(&ss), &ss);
+            }
 
-/* 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);
+            u->frames_in_buffer = offset;
+            u->saved_frame_time = * (jack_nframes_t*) data;
+            u->saved_frame_time_valid = TRUE;
 
-/* The ringbuffer filler thread runs in this function. */
-static void* fill_ringbuffer(void* arg);
+            return 0;
 
-/* request_render asks asynchronously the mainloop to call io_event_cb. */
-static void request_render(struct userdata* u);
+        case SINK_MESSAGE_ON_SHUTDOWN:
+            pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
+            return 0;
 
+        case PA_SINK_MESSAGE_GET_LATENCY: {
+            jack_nframes_t l, ft, d;
+            size_t n;
 
-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;
+            /* This is the "worst-case" latency */
+            l = jack_port_get_total_latency(u->client, u->port[0]) + u->frames_in_buffer;
+
+            if (u->saved_frame_time_valid) {
+                /* Adjust the worst case latency by the time that
+                 * passed since we last handed data to JACK */
+
+                ft = jack_frame_time(u->client);
+                d = ft > u->saved_frame_time ? ft - u->saved_frame_time : 0;
+                l = l > d ? l - d : 0;
+            }
+
+            /* Convert it to usec */
+            n = l * pa_frame_size(&u->sink->sample_spec);
+            *((pa_usec_t*) data) = pa_bytes_to_usec(n, &u->sink->sample_spec);
+
+            return 0;
         }
-        
-        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;
 
-    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;
-    }
+    return pa_sink_process_msg(o, code, data, offset, memchunk);
+}
 
-    if (o.connect)
-        connect_ports(self);
+static int jack_process(jack_nframes_t nframes, void *arg) {
+    struct userdata *u = arg;
+    unsigned c;
+    jack_nframes_t frame_time;
+    pa_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;
+    /* We just forward the request to our other RT thread */
 
-    pa_xfree(o.sink_name);
-    pa_xfree(o.server_name);
-    pa_xfree(o.client_name);
-    
-    return 0;
+    for (c = 0; c < u->channels; c++)
+        pa_assert_se(u->buffer[c] = jack_port_get_buffer(u->port[c], nframes));
 
-fail:
-    pa_xfree(o.sink_name);
-    pa_xfree(o.server_name);
-    pa_xfree(o.client_name);
-    pa__done(c, self);
+    frame_time = jack_frame_time(u->client);
 
-    return -1;
+    pa_assert_se(pa_asyncmsgq_send(u->jack_msgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_RENDER, &frame_time, nframes, NULL) == 0);
+    return 0;
 }
 
+static void thread_func(void *userdata) {
+    struct userdata *u = userdata;
 
-static int parse_options(struct options* o, const char* argument) {
-    pa_modargs *ma = NULL;
-    const char* arg_val;
-    pa_strbuf* strbuf;
-    
-    assert(o);
+    pa_assert(u);
 
-    if (!(ma = pa_modargs_new(argument, valid_modargs))) {
-        pa_log_error("Failed to parse module arguments.");
-        goto fail;
-    }
+    pa_log_debug("Thread starting up");
 
-    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;
-    }
+    if (u->core->high_priority)
+        pa_make_realtime();
 
-    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;
-    }
+    pa_thread_mq_install(&u->thread_mq);
+    pa_rtpoll_install(u->rtpoll);
 
-    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;
-    }
+    for (;;) {
+        int ret;
 
-    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.");
+        if ((ret = pa_rtpoll_run(u->rtpoll, TRUE)) < 0)
             goto fail;
-        }
 
-        /* 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 (ret == 0)
+            goto finish;
     }
 
-    pa_modargs_free(ma);
-
-    return 0;
-
 fail:
-    if (ma)
-      pa_modargs_free(ma);
+    /* 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);
 
-    return -1;
+finish:
+    pa_log_debug("Thread shutting down");
 }
 
+static void jack_error_func(const char*t) {
+    char *s;
 
-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;
+    s = pa_xstrndup(t, strcspn(t, "\n\r"));
+    pa_log_warn("JACK error >%s<", s);
+    pa_xfree(s);
 }
 
+static void jack_init(void *arg) {
+    struct userdata *u = arg;
 
-static int create_sink(pa_module* self, struct options* o) {
-    struct userdata* u;
-    pa_sample_spec ss;
-    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));
+    pa_log_info("JACK thread starting up.");
 
-    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 (u->core->high_priority)
+        pa_make_realtime();
 }
 
+static void jack_shutdown(void* arg) {
+    struct userdata *u = arg;
 
-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);
+    pa_log_info("JACK thread shutting down..");
+    pa_asyncmsgq_post(u->jack_msgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_ON_SHUTDOWN, NULL, 0, NULL, NULL);
 }
 
+int pa__init(pa_module*m) {
+    struct userdata *u = NULL;
+    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;
 
-static int start_filling_ringbuffer(pa_module* self) {
-    struct userdata* u;
-    pthread_attr_t thread_attributes;
+    pa_assert(m);
 
-    assert(self);
-    assert(self->userdata);
+    jack_set_error_function(jack_error_func);
 
-    u = self->userdata;
-    
-    pthread_attr_init(&thread_attributes);
-    
-    if (pthread_attr_setinheritsched(&thread_attributes,
-                                     PTHREAD_INHERIT_SCHED) != 0) {
-        pa_log("pthread_attr_setinheritsched() failed.");
+    if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
+        pa_log("Failed to parse module arguments.");
         goto fail;
     }
-    else if (pthread_create(&u->filler_thread, &thread_attributes,
-                            fill_ringbuffer, u) != 0) {
-        pa_log("pthread_create() failed.");
+
+    if (pa_modargs_get_value_boolean(ma, "connect", &do_connect) < 0) {
+        pa_log("Failed to parse connect= argument.");
         goto fail;
     }
-    
-    u->filler_thread_is_running = 1;
-    
-    pthread_attr_destroy(&thread_attributes);
-
-    return 0;
-    
-fail:
-    pthread_attr_destroy(&thread_attributes);
-    return -1;
-}
 
+    server_name = pa_modargs_get_value(ma, "server_name", NULL);
+    client_name = pa_modargs_get_value(ma, "client_name", "PulseAudio JACK Sink");
+
+    u = pa_xnew0(struct userdata, 1);
+    u->core = m->core;
+    u->module = m;
+    m->userdata = u;
+    u->saved_frame_time_valid = FALSE;
+    pa_thread_mq_init(&u->thread_mq, m->core->mainloop);
+    u->rtpoll = pa_rtpoll_new();
+    pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY, u->thread_mq.inq);
+
+    /* The queue linking the JACK thread and our RT thread */
+    u->jack_msgq = pa_asyncmsgq_new(0);
+
+    /* The msgq from the JACK RT thread should have an even higher
+     * priority than the normal message queues, to match the guarantee
+     * all other drivers make: supplying the audio device with data is
+     * the top priority -- and as long as that is possible we don't do
+     * anything else */
+    u->rtpoll_item = pa_rtpoll_item_new_asyncmsgq(u->rtpoll, PA_RTPOLL_EARLY-1, u->jack_msgq);
+
+    if (!(u->client = jack_client_open(client_name, server_name ? JackServerName : JackNullOption, &status, server_name))) {
+        pa_log("jack_client_open() failed.");
+        goto fail;
+    }
 
-static void jack_error_func(const char* t) {
-    pa_log_warn("JACK error >%s<", t);
-}
-
+    ports = jack_get_ports(u->client, NULL, NULL, JackPortIsPhysical|JackPortIsInput);
 
-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);
-
-    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);
-}
+    channels = 0;
+    for (p = ports; *p; p++)
+        channels++;
 
+    if (!channels)
+        channels = m->core->default_sample_spec.channels;
 
-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;
+    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;
     }
-    
-    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;
+
+    pa_channel_map_init_auto(&map, channels, PA_CHANNEL_MAP_ALSA);
+    if (pa_modargs_get_channel_map(ma, NULL, &map) < 0 || map.channels != channels) {
+        pa_log("Failed to parse channel_map= argument.");
+        goto fail;
     }
-    
-    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;
-}
 
+    pa_log_info("Successfully connected as '%s'", jack_get_client_name(u->client));
+
+    ss.channels = u->channels = channels;
+    ss.rate = jack_get_sample_rate(u->client);
+    ss.format = PA_SAMPLE_FLOAT32NE;
 
-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;
+    pa_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;
         }
-        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;
-}
 
+    if (!(u->sink = pa_sink_new(m->core, __FILE__, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME), 0, &ss, &map))) {
+        pa_log("failed to create sink.");
+        goto fail;
+    }
 
-static void jack_shutdown(void* arg) {
-    struct userdata* u = arg;
-    assert(u);
+    u->sink->parent.process_msg = sink_process_msg;
+    u->sink->userdata = u;
+    u->sink->flags = PA_SINK_LATENCY;
 
-    u->quit_requested = 1;
-    request_render(u);
-}
+    pa_sink_set_module(u->sink, m);
+    pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
+    pa_sink_set_rtpoll(u->sink, u->rtpoll);
+    pa_sink_set_description(u->sink, t = pa_sprintf_malloc("Jack sink (%s)", jack_get_client_name(u->client)));
+    pa_xfree(t);
 
+    jack_set_process_callback(u->client, jack_process, u);
+    jack_on_shutdown(u->client, jack_shutdown, u);
+    jack_set_thread_init_callback(u->client, jack_init, u);
 
-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);
-
-    u = self->userdata;
-    
-    assert(u->pipe_fds[0] == fd);
-
-    pa_read(fd, &x, 1, &u->pipe_fd_type);
-
-    if (u->quit_requested) {
-        pa_module_unload_request(self);
-        return;
+    if (!(u->thread = pa_thread_new(thread_func, u))) {
+        pa_log("Failed to create thread.");
+        goto fail;
     }
 
-    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 truncate the lengths so that we process only complete frames. */
-    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;
-        }
+    if (jack_activate(u->client)) {
+        pa_log("jack_activate() failed");
+        goto fail;
     }
-    
-    /* 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;
+
+    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;
+            }
+
+            pa_log_info("Connecting %s to %s", jack_port_name(u->port[i]), *p);
+
+            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_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);
-}
 
+    pa_sink_put(u->sink);
 
-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);
+    pa_modargs_free(ma);
 
-        request_render(u);
-    }
-    
-    return NULL;
-}
+    return 0;
+
+fail:
+    if (ma)
+        pa_modargs_free(ma);
+
+    free(ports);
 
+    pa__done(m);
 
-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);
+    return -1;
 }
 
-void pa__done(pa_core* c, pa_module* self) {
-    struct userdata* u;
-    
-    assert(c);
-    assert(self);
+void pa__done(pa_module*m) {
+    struct userdata *u;
+
+    pa_assert(m);
 
-    if (!self->userdata)
+    if (!(u = m->userdata))
         return;
 
-    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->client)
+        jack_client_close(u->client);
+
+    if (u->sink)
+        pa_sink_unlink(u->sink);
+
+    if (u->thread) {
+        pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
+        pa_thread_free(u->thread);
     }
-    
-    if (u->j_client)
-        jack_client_close(u->j_client);
 
-    if (u->io_event)
-        c->mainloop->io_free(u->io_event);
+    pa_thread_mq_done(&u->thread_mq);
 
-    if (u->sink) {
-        pa_sink_disconnect(u->sink);
+    if (u->sink)
         pa_sink_unref(u->sink);
-    }
-    
-    if (u->ringbuffer)
-        jack_ringbuffer_free(u->ringbuffer);
-
-    if (u->pipe_fds[0] >= 0)
-        pa_close(u->pipe_fds[0]);
-    if (u->pipe_fds[1] >= 0)
-        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;
+
+    if (u->rtpoll_item)
+        pa_rtpoll_item_free(u->rtpoll_item);
+
+    if (u->jack_msgq)
+        pa_asyncmsgq_unref(u->jack_msgq);
+
+    if (u->rtpoll)
+        pa_rtpoll_free(u->rtpoll);
+
+    pa_xfree(u);
 }