Implement LV2 worker extension.

git-svn-id: http://svn.drobilla.net/lad/trunk/jalv@4101 a436a847-0d15-0410-975c-d299462d15a1

3 files changed, 255 insertions, 10 deletions

diff --git a/src/jalv.c b/src/jalv.c
index 2525501..dc4cbee 100644
--- a/src/jalv.c
+++ b/src/jalv.c
@@ -40,6 +40,7 @@
 #include "lv2/lv2plug.in/ns/ext/time/time.h"
 #include "lv2/lv2plug.in/ns/ext/uri-map/uri-map.h"
 #include "lv2/lv2plug.in/ns/ext/urid/urid.h"
+#include "lv2/lv2plug.in/ns/ext/worker/worker.h"
 
 #include "lilv/lilv.h"
 
@@ -91,11 +92,13 @@ static LV2_Feature map_feature       = { NS_EXT "urid#map", NULL };
 static LV2_Feature unmap_feature     = { NS_EXT "urid#unmap", NULL };
 static LV2_Feature instance_feature  = { NS_EXT "instance-access", NULL };
 static LV2_Feature make_path_feature = { LV2_STATE__makePath, NULL };
+static LV2_Feature schedule_feature  = { LV2_WORKER__schedule, NULL };
 
-const LV2_Feature* features[7] = {
+const LV2_Feature* features[8] = {
 	&uri_map_feature, &map_feature, &unmap_feature,
 	&instance_feature,
 	&make_path_feature,
+	&schedule_feature,
 	NULL
 };
 
@@ -425,6 +428,23 @@ jack_process_cb(jack_nframes_t nframes, void* data)
 	/* Run plugin for this cycle */
 	lilv_instance_run(host->instance, nframes);
 
+	/* Process any replies from the worker. */
+	if (host->worker.responses) {
+		uint32_t read_space = jack_ringbuffer_read_space(host->worker.responses);
+		while (read_space) {
+			uint32_t size = 0;
+			jack_ringbuffer_read(host->worker.responses, (char*)&size, sizeof(size));
+			
+			jack_ringbuffer_read(
+				host->worker.responses, host->worker.response, size);
+
+			host->worker.iface->work_response(
+				host->instance->lv2_handle, size, host->worker.response);
+
+			read_space -= sizeof(size) + size;
+		}
+	}
+
 	/* Check if it's time to send updates to the UI */
 	host->event_delta_t += nframes;
 	bool           send_ui_updates = false;
@@ -614,6 +634,54 @@ jalv_emit_ui_events(Jalv* host)
 	return true;
 }
 
+LV2_Worker_Status
+schedule_work(LV2_Worker_Schedule_Handle handle,
+              uint32_t                   size,
+              const void*                data)
+{
+	Jalv* jalv = (Jalv*)handle;
+	jack_ringbuffer_write(jalv->worker.requests, (const char*)&size, sizeof(size));
+	jack_ringbuffer_write(jalv->worker.requests, data, size);
+	zix_sem_post(&jalv->worker.sem);
+	return LV2_WORKER_SUCCESS;
+}
+
+LV2_Worker_Status
+worker_respond(LV2_Worker_Respond_Handle handle,
+               uint32_t                  size,
+               const void*               data)
+{
+	Jalv* jalv = (Jalv*)handle;
+	jack_ringbuffer_write(jalv->worker.responses, (const char*)&size, sizeof(size));
+	jack_ringbuffer_write(jalv->worker.responses, data, size);
+	return LV2_WORKER_SUCCESS;
+}
+
+void*
+worker(void* data)
+{
+	Jalv* jalv = (Jalv*)data;
+	void* buf  = NULL;
+	while (true) {
+		zix_sem_wait(&jalv->worker.sem);
+		if (jalv->exit) {
+			break;
+		}
+
+		uint32_t size = 0;
+		jack_ringbuffer_read(jalv->worker.requests, (char*)&size, sizeof(size));
+
+		buf = realloc(buf, size);
+		jack_ringbuffer_read(jalv->worker.requests, buf, size);
+
+		jalv->worker.iface->work(
+			jalv->instance->lv2_handle, worker_respond, jalv, size, buf);
+	}
+
+	free(buf);
+	return NULL;
+}
+
 static void
 signal_handler(int ignored)
 {
@@ -672,10 +740,14 @@ main(int argc, char** argv)
 	LV2_State_Make_Path make_path = { &host, jalv_make_path };
 	make_path_feature.data = &make_path;
 
+	LV2_Worker_Schedule schedule = { &host, schedule_work };
+	schedule_feature.data = &schedule;
+
 	zix_sem_init(&exit_sem, 0);
 	host.done = &exit_sem;
 
 	zix_sem_init(&host.paused, 0);
+	zix_sem_init(&host.worker.sem, 0);
 
 	signal(SIGINT, signal_handler);
 	signal(SIGTERM, signal_handler);
@@ -698,6 +770,7 @@ main(int argc, char** argv)
 	host.midi_class     = lilv_new_uri(world, LILV_URI_MIDI_EVENT);
 	host.preset_class   = lilv_new_uri(world, NS_PSET "Preset");
 	host.label_pred     = lilv_new_uri(world, LILV_NS_RDFS "label");
+	host.work_schedule  = lilv_new_uri(world, LV2_WORKER__schedule);
 	host.optional       = lilv_new_uri(world, LILV_NS_LV2
 	                                   "connectionOptional");
 
@@ -756,6 +829,7 @@ main(int argc, char** argv)
 		}
 	}
 
+	/* Create ringbuffers for UI if necessary */
 	if (host.ui) {
 		fprintf(stderr, "UI:        %s\n",
 		        lilv_node_as_uri(lilv_ui_get_uri(host.ui)));
@@ -824,6 +898,18 @@ main(int argc, char** argv)
 		jalv_allocate_port_buffers(&host);
 	}
 
+	/* Create thread and ringbuffers for worker if necessary */
+	if (lilv_plugin_has_feature(host.plugin, host.work_schedule)) {
+		host.worker.iface = lilv_instance_get_extension_data(
+			host.instance, LV2_WORKER__Interface);
+		zix_thread_create(&host.worker.thread, 4096, worker, &host);
+		host.worker.requests  = jack_ringbuffer_create(4096);
+		host.worker.responses = jack_ringbuffer_create(4096);
+		host.worker.response  = malloc(4096);
+		jack_ringbuffer_mlock(host.worker.requests);
+		jack_ringbuffer_mlock(host.worker.responses);
+	}
+
 	/* Apply loaded state to plugin instance if necessary */
 	if (state) {
 		jalv_apply_state(&host, state);
@@ -890,9 +976,15 @@ main(int argc, char** argv)
 
 	/* Wait for finish signal from UI or signal handler */
 	zix_sem_wait(&exit_sem);
+	host.exit = true;
 
 	fprintf(stderr, "Exiting...\n");
 
+	if (host.worker.requests) {
+		zix_sem_post(&host.worker.sem);
+		zix_thread_join(host.worker.thread, NULL);
+	}
+
 	/* Deactivate JACK */
 	jack_deactivate(host.jack_client);
 	for (uint32_t i = 0; i < host.num_ports; ++i) {
@@ -913,6 +1005,11 @@ main(int argc, char** argv)
 		jack_ringbuffer_free(host.ui_events);
 		jack_ringbuffer_free(host.plugin_events);
 	}
+	if (host.worker.requests) {
+		jack_ringbuffer_free(host.worker.requests);
+		jack_ringbuffer_free(host.worker.responses);
+		free(host.worker.response);
+	}
 	lilv_node_free(native_ui_type);
 	lilv_node_free(host.input_class);
 	lilv_node_free(host.output_class);
@@ -922,6 +1019,7 @@ main(int argc, char** argv)
 	lilv_node_free(host.midi_class);
 	lilv_node_free(host.preset_class);
 	lilv_node_free(host.label_pred);
+	lilv_node_free(host.work_schedule);
 	lilv_node_free(host.optional);
 	symap_free(host.symap);
 	suil_host_free(ui_host);
diff --git a/src/jalv_internal.h b/src/jalv_internal.h
index 82f414c..6bbc291 100644
--- a/src/jalv_internal.h
+++ b/src/jalv_internal.h
@@ -29,10 +29,12 @@
 
 #include "lv2/lv2plug.in/ns/ext/atom/atom.h"
 #include "lv2/lv2plug.in/ns/ext/atom/forge.h"
-#include "lv2/lv2plug.in/ns/ext/urid/urid.h"
 #include "lv2/lv2plug.in/ns/ext/state/state.h"
+#include "lv2/lv2plug.in/ns/ext/urid/urid.h"
+#include "lv2/lv2plug.in/ns/ext/worker/worker.h"
 
 #include "zix/sem.h"
+#include "zix/thread.h"
 
 #include "sratom/sratom.h"
 
@@ -104,6 +106,15 @@ typedef enum {
 } JalvPlayState;
 
 typedef struct {
+	jack_ringbuffer_t*          requests;   ///< Requests to the worker
+	jack_ringbuffer_t*          responses;  ///< Responses from the worker
+	void*                       response;   ///< Worker response buffer
+	ZixSem                      sem;        ///< Worker semaphore
+	ZixThread                   thread;     ///< Worker thread
+	const LV2_Worker_Interface* iface;      ///< Plugin worker interface
+} JalvWorker;
+	
+typedef struct {
 	JalvOptions        opts;           ///< Command-line options
 	JalvURIDs          urids;          ///< URIDs
 	LV2_Atom_Forge     forge;          ///< Atom forge
@@ -116,6 +127,7 @@ typedef struct {
 	jack_client_t*     jack_client;    ///< Jack client
 	jack_ringbuffer_t* ui_events;      ///< Port events from UI
 	jack_ringbuffer_t* plugin_events;  ///< Port events from plugin
+	JalvWorker         worker;         ///< Worker thread implementation
 	ZixSem*            done;           ///< Exit semaphore
 	ZixSem             paused;         ///< Paused signal from process thread
 	JalvPlayState      play_state;     ///< Current play state
@@ -132,20 +144,22 @@ typedef struct {
 	uint32_t           longest_sym;    ///< Longest port symbol
 	jack_nframes_t     sample_rate;    ///< Sample rate
 	jack_nframes_t     event_delta_t;  ///< Frames since last update sent to UI
-	LilvNode*          input_class;    ///< Input port class (URI)
-	LilvNode*          output_class;   ///< Output port class (URI)
-	LilvNode*          control_class;  ///< Control port class (URI)
 	LilvNode*          audio_class;    ///< Audio port class (URI)
-	LilvNode*          event_class;    ///< Event port class (URI)
 	LilvNode*          chunk_class;    ///< Atom sequence class (URI)
-	LilvNode*          seq_class;      ///< Atom sequence class (URI)
-	LilvNode*          msg_port_class; ///< Atom event port class (URI)
-	LilvNode*          midi_class;     ///< MIDI event class (URI)
-	LilvNode*          preset_class;   ///< Preset class (URI)
+	LilvNode*          control_class;  ///< Control port class (URI)
+	LilvNode*          event_class;    ///< Event port class (URI)
+	LilvNode*          input_class;    ///< Input port class (URI)
 	LilvNode*          label_pred;     ///< rdfs:label
+	LilvNode*          midi_class;     ///< MIDI event class (URI)
+	LilvNode*          msg_port_class; ///< Atom event port class (URI)
 	LilvNode*          optional;       ///< lv2:connectionOptional port property
+	LilvNode*          output_class;   ///< Output port class (URI)
+	LilvNode*          preset_class;   ///< Preset class (URI)
+	LilvNode*          seq_class;      ///< Atom sequence class (URI)
+	LilvNode*          work_schedule;  ///< lv2:connectionOptional port property
 	uint32_t           midi_event_id;  ///< MIDI event class ID in event context
 	bool               buf_size_set;   ///< True iff buffer size callback fired
+	bool               exit;           ///< True if execution is finished 
 } Jalv;
 
 int
diff --git a/src/zix/thread.h b/src/zix/thread.h
new file mode 100644
index 0000000..ff5a727
--- /dev/null
+++ b/src/zix/thread.h
@@ -0,0 +1,133 @@
+/*
+  Copyright 2012 David Robillard <http://drobilla.net>
+
+  Permission to use, copy, modify, and/or distribute this software for any
+  purpose with or without fee is hereby granted, provided that the above
+  copyright notice and this permission notice appear in all copies.
+
+  THIS SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+  WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+  MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+  ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+  WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+  ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+  OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+
+#ifndef ZIX_THREAD_H
+#define ZIX_THREAD_H
+
+#ifdef _WIN32
+#    include <windows.h>
+#else
+#    include <errno.h>
+#    include <pthread.h>
+#endif
+
+#include "zix/common.h"
+
+#ifdef __cplusplus
+extern "C" {
+#else
+#    include <stdbool.h>
+#endif
+
+/**
+   @addtogroup zix
+   @{
+   @name Thread
+   @{
+*/
+
+#ifdef _WIN32
+typedef HANDLE ZixThread;
+#else
+typedef pthread_t ZixThread;
+#endif
+
+/**
+   Initialize @c thread to a new thread.
+
+   The thread will immediately be launched, calling @c function with @c arg
+   as the only parameter.
+*/
+static inline ZixStatus
+zix_thread_create(ZixThread* thread,
+                  size_t     stack_size,
+                  void*      (*function)(void*),
+                  void*      arg);
+
+/**
+   Join @c thread (block until @c thread exits).
+*/
+static inline ZixStatus
+zix_thread_join(ZixThread thread, void** retval);
+
+#ifdef _WIN32
+
+static inline ZixStatus
+zix_thread_create(ZixThread* thread,
+                  size_t     stack_size,
+                  void*      (*function)(void*),
+                  void*      arg)
+{
+	*thread = CreateThread(NULL, stack_size,
+	                       (LPTHREAD_START_ROUTINE)function, arg,
+	                       0, NULL);
+	return *thread ? ZIX_STATUS_SUCCESS : ZIX_STATUS_ERROR;
+}
+
+static inline ZixStatus
+zix_thread_join(ZixThread thread, void** retval)
+{
+	return WaitForSingleObject(thread, INFINITE)
+		? ZIX_STATUS_SUCCESS : ZIX_STATUS_ERROR;
+}
+
+#else  /* !defined(_WIN32) */
+
+static inline ZixStatus
+zix_thread_create(ZixThread* thread,
+                  size_t     stack_size,
+                  void*      (*function)(void*),
+                  void*      arg)
+{
+	pthread_attr_t attr;
+	pthread_attr_init(&attr);
+	pthread_attr_setstacksize(&attr, stack_size);
+
+	const int ret = pthread_create(thread, NULL, function, arg);
+	pthread_attr_destroy(&attr);
+
+	if (ret == EAGAIN) {
+		return ZIX_STATUS_NO_MEM;
+	} else if (ret == EINVAL) {
+		return ZIX_STATUS_BAD_ARG;
+	} else if (ret == EPERM) {
+		return ZIX_STATUS_BAD_PERMS;
+	} else if (ret) {
+		return ZIX_STATUS_ERROR;
+	}
+
+	return ZIX_STATUS_SUCCESS;
+}
+
+static inline ZixStatus
+zix_thread_join(ZixThread thread, void** retval)
+{
+	return pthread_join(thread, retval)
+		? ZIX_STATUS_ERROR : ZIX_STATUS_SUCCESS;
+}
+
+#endif
+
+/**
+   @}
+   @}
+*/
+
+#ifdef __cplusplus
+}  /* extern "C" */
+#endif
+
+#endif  /* ZIX_THREAD_H */