Partial support for message/value ports and the message context.

This use case now works:
 - Add an event input and the "print" plugin from imum.lv2 to ingen
 - Connect the event input to the input of "print"
 - Hook Ingen up to JACK and play some MIDI events
   (or get events to the print plugin from anywhere else)
 - The "print" plugin will print the received events to the console
   in the message context (i.e. the audio thread is realtime safe)


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

35 files changed, 382 insertions, 84 deletions

diff --git a/src/common/interface/PortType.hpp b/src/common/interface/PortType.hpp
index 72dbb327..f659cc81 100644
--- a/src/common/interface/PortType.hpp
+++ b/src/common/interface/PortType.hpp
@@ -39,6 +39,7 @@ public:
 		CONTROL = 2,
 		EVENTS  = 3,
 		VALUE   = 7,
+		MESSAGE = 8,
 	};
 
 	PortType(const Raul::URI& uri)
@@ -52,6 +53,8 @@ public:
 			_symbol = EVENTS;
 		} else if (uri.str() == type_uri(VALUE)) {
 			_symbol = VALUE;
+		} else if (uri.str() == type_uri(MESSAGE)) {
+			_symbol = MESSAGE;
 		}
 	}
 
@@ -71,6 +74,7 @@ public:
 	inline bool is_control() { return _symbol == CONTROL; }
 	inline bool is_events()  { return _symbol == EVENTS; }
 	inline bool is_value()   { return _symbol == VALUE; }
+	inline bool is_message() { return _symbol == MESSAGE; }
 
 private:
 
@@ -80,6 +84,7 @@ private:
 		case 2:  return "lv2:ControlPort";
 		case 3:  return "lv2ev:EventPort";
 		case 7:  return "obj:ValuePort";
+		case 8:  return "obj:MessagePort";
 		default: return "";
 		}
 	}
diff --git a/src/engine/Buffer.cpp b/src/engine/Buffer.cpp
index 4b66b63f..15752846 100644
--- a/src/engine/Buffer.cpp
+++ b/src/engine/Buffer.cpp
@@ -35,7 +35,7 @@ Buffer::create(Engine& engine, Shared::PortType type, size_t size)
 		return new AudioBuffer(type, size);
 	else if (type.is_events())
 		return new EventBuffer(size);
-	else if (type.is_value())
+	else if (type.is_value() || type.is_message())
 		return new ObjectBuffer(std::max(size,
 					sizeof(LV2_Object) + sizeof(void*)));
 	else
diff --git a/src/engine/BufferFactory.cpp b/src/engine/BufferFactory.cpp
index cfbe72e6..d84cf658 100644
--- a/src/engine/BufferFactory.cpp
+++ b/src/engine/BufferFactory.cpp
@@ -97,7 +97,7 @@ BufferFactory::create(Shared::PortType type, size_t size)
 		if (size == 0)
 			size = _engine.audio_driver()->buffer_size() * 4; // FIXME
 		buffer = new EventBuffer(size);
-	} else if (type.is_value()) {
+	} else if (type.is_value() || type.is_message()) {
 		if (size == 0)
 			size = 32; // FIXME
 		buffer = new ObjectBuffer(std::max(size, sizeof(LV2_Object) + sizeof(void*)));
diff --git a/src/engine/BufferFactory.hpp b/src/engine/BufferFactory.hpp
index 8ec29697..25164db9 100644
--- a/src/engine/BufferFactory.hpp
+++ b/src/engine/BufferFactory.hpp
@@ -49,7 +49,8 @@ private:
 		case PortType::AUDIO:   return _free_audio;
 		case PortType::CONTROL: return _free_control;
 		case PortType::EVENTS:  return _free_event;
-		case PortType::VALUE:   return _free_object;
+		case PortType::VALUE:
+		case PortType::MESSAGE: return _free_object;
 		default: throw;
 		}
 	}
diff --git a/src/engine/ConnectionImpl.cpp b/src/engine/ConnectionImpl.cpp
index 24f42545..27786e00 100644
--- a/src/engine/ConnectionImpl.cpp
+++ b/src/engine/ConnectionImpl.cpp
@@ -18,12 +18,15 @@
 #include <algorithm>
 #include "raul/Maid.hpp"
 #include "util.hpp"
+#include "AudioBuffer.hpp"
+#include "BufferFactory.hpp"
 #include "ConnectionImpl.hpp"
+#include "Engine.hpp"
+#include "EventBuffer.hpp"
+#include "InputPort.hpp"
+#include "MessageContext.hpp"
 #include "PortImpl.hpp"
-#include "AudioBuffer.hpp"
 #include "ProcessContext.hpp"
-#include "InputPort.hpp"
-#include "BufferFactory.hpp"
 
 namespace Ingen {
 
@@ -35,7 +38,8 @@ using namespace Shared;
  * user (InputPort).
  */
 ConnectionImpl::ConnectionImpl(BufferFactory& bufs, PortImpl* src_port, PortImpl* dst_port)
-	: _bufs(bufs)
+	: _queue(NULL)
+	, _bufs(bufs)
 	, _src_port(src_port)
 	, _dst_port(dst_port)
 	, _pending_disconnection(false)
@@ -44,15 +48,14 @@ ConnectionImpl::ConnectionImpl(BufferFactory& bufs, PortImpl* src_port, PortImpl
 	assert(dst_port);
 	assert(src_port != dst_port);
 	assert(src_port->path() != dst_port->path());
-	assert(src_port->type() == dst_port->type()
-			|| ( (src_port->type() == PortType::CONTROL || src_port->type() == PortType::AUDIO)
-				&& (dst_port->type() == PortType::CONTROL || dst_port->type() == PortType::AUDIO) ));
 
-	/*assert((src_port->parent_node()->poly() == dst_port->parent_node()->poly())
-		|| (src_port->parent_node()->poly() == 1 || dst_port->parent_node()->poly() == 1));*/
-
-	if (must_mix())
+	if (must_mix() || must_queue())
 		_local_buffer = bufs.get(dst_port->type(), dst_port->buffer_size());
+
+	if (must_queue())
+		_queue = new Raul::RingBuffer<LV2_Object>(src_port->buffer_size() * 2);
+
+	dump();
 }
 
 
@@ -60,7 +63,8 @@ void
 ConnectionImpl::dump() const
 {
 	cerr << _src_port->path() << " -> " << _dst_port->path()
-		<< (must_mix() ? " MIX" : " DIRECT") << endl;
+		<< (must_mix()   ? " (MIX) " : " (DIRECT) ")
+		<< (must_queue() ? " (QUEUE)" : " (NOQUEUE)") << endl;
 }
 
 
@@ -96,6 +100,27 @@ ConnectionImpl::apply_poly(Raul::Maid& maid, uint32_t poly)
 void
 ConnectionImpl::process(Context& context)
 {
+	if (must_queue()) {
+		SharedPtr<EventBuffer> src_buf = PtrCast<EventBuffer>(_src_port->buffer(0));
+		if (!src_buf) {
+			cerr << "ERROR: Queued connection but source is not an EventBuffer" << endl;
+			return;
+		}
+
+		SharedPtr<ObjectBuffer> local_buf = PtrCast<ObjectBuffer>(_local_buffer);
+		if (!local_buf) {
+			cerr << "ERROR: Queued connection but source is not an EventBuffer" << endl;
+			return;
+		}
+
+		if (_queue->read_space()) {
+			LV2_Object obj;
+			_queue->full_peek(sizeof(LV2_Object), &obj);
+			_queue->full_read(sizeof(LV2_Object) + obj.size, local_buf->object());
+		}
+		return;
+	}
+
 	if (!must_mix())
 		return;
 
@@ -108,5 +133,33 @@ ConnectionImpl::process(Context& context)
 }
 
 
+void
+ConnectionImpl::queue(Context& context)
+{
+	if (!must_queue())
+		return;
+
+	SharedPtr<EventBuffer> src_buf = PtrCast<EventBuffer>(_src_port->buffer(0));
+	if (!src_buf) {
+		cerr << "ERROR: Queued connection but source is not an EventBuffer" << endl;
+		return;
+	}
+
+	while (src_buf->is_valid()) {
+		LV2_Object* obj = src_buf->get_object();
+		/*cout << _src_port->path() << " -> " << _dst_port->path()
+			<< " QUEUE OBJECT TYPE " << obj->type << ":";
+		for (size_t i = 0; i < obj->size; ++i)
+			cout << " " << std::hex << (int)obj->body[i];
+		cout << endl;*/
+
+		_queue->write(sizeof(LV2_Object) + obj->size, obj);
+		src_buf->increment();
+
+		context.engine().message_context()->run(_dst_port->parent_node());
+	}
+}
+
+
 } // namespace Ingen
 
diff --git a/src/engine/ConnectionImpl.hpp b/src/engine/ConnectionImpl.hpp
index 76224269..9daea1f1 100644
--- a/src/engine/ConnectionImpl.hpp
+++ b/src/engine/ConnectionImpl.hpp
@@ -24,6 +24,7 @@
 #include "raul/Deletable.hpp"
 #include "interface/PortType.hpp"
 #include "interface/Connection.hpp"
+#include "object.lv2/object.h"
 #include "PortImpl.hpp"
 #include "PortImpl.hpp"
 
@@ -62,6 +63,7 @@ public:
 	void pending_disconnection(bool b) { _pending_disconnection = b; }
 
 	void process(Context& context);
+	void queue(Context& context);
 
 	/** Get the buffer for a particular voice.
 	 * A Connection is smart - it knows the destination port requesting the
@@ -69,7 +71,7 @@ public:
 	 * voice in a mono->poly connection).
 	 */
 	inline SharedPtr<Buffer> buffer(uint32_t voice) const {
-		if (must_mix()) {
+		if (must_mix() || must_queue()) {
 			return _local_buffer;
 		} else if ( ! _src_port->polyphonic()) {
 			return _src_port->buffer(0);
@@ -85,9 +87,14 @@ public:
 	/** Returns true if this connection must mix down voices into a local buffer */
 	inline bool must_mix() const { return _src_port->poly() > _dst_port->poly(); }
 
+	/** Returns true if this connection crosses contexts and must buffer */
+	inline bool must_queue() const { return _src_port->context() != _dst_port->context(); }
+
 protected:
 	void dump() const;
 
+	Raul::RingBuffer<LV2_Object>* _queue;
+
 	BufferFactory&    _bufs;
 	PortImpl* const   _src_port;
 	PortImpl* const   _dst_port;
diff --git a/src/engine/Engine.cpp b/src/engine/Engine.cpp
index a12f6e2e..07529caf 100644
--- a/src/engine/Engine.cpp
+++ b/src/engine/Engine.cpp
@@ -185,6 +185,8 @@ Engine::activate(size_t parallelism)
 {
 	assert(_audio_driver);
 
+	_message_context->Thread::start();
+
 	if (!_midi_driver)
 		_midi_driver = new DummyMidiDriver();
 
diff --git a/src/engine/EventBuffer.hpp b/src/engine/EventBuffer.hpp
index b779a837..d75c91ba 100644
--- a/src/engine/EventBuffer.hpp
+++ b/src/engine/EventBuffer.hpp
@@ -60,6 +60,10 @@ public:
 		return _buf->get_event(frames, subframes, type, size, data);
 	}
 
+	LV2_Object* get_object() const {
+		return _buf->get_object();
+	}
+
 	inline bool append(uint32_t       frames,
 	                   uint32_t       subframes,
 	                   uint16_t       type,
diff --git a/src/engine/JackAudioDriver.cpp b/src/engine/JackAudioDriver.cpp
index 86f946da..edb1ba9a 100644
--- a/src/engine/JackAudioDriver.cpp
+++ b/src/engine/JackAudioDriver.cpp
@@ -21,21 +21,22 @@
 #include <iostream>
 #include <cstdlib>
 #include "raul/List.hpp"
+#include "AudioBuffer.hpp"
+#include "DuplexPort.hpp"
 #include "Engine.hpp"
-#include "util.hpp"
 #include "Event.hpp"
-#include "ThreadManager.hpp"
-#include "QueuedEvent.hpp"
 #include "EventSource.hpp"
-#include "PostProcessor.hpp"
+#include "EventSource.hpp"
+#include "JackMidiDriver.hpp"
+#include "MessageContext.hpp"
+#include "MidiDriver.hpp"
 #include "PatchImpl.hpp"
 #include "PortImpl.hpp"
-#include "MidiDriver.hpp"
-#include "DuplexPort.hpp"
-#include "EventSource.hpp"
-#include "AudioBuffer.hpp"
+#include "PostProcessor.hpp"
 #include "ProcessSlave.hpp"
-#include "JackMidiDriver.hpp"
+#include "QueuedEvent.hpp"
+#include "ThreadManager.hpp"
+#include "util.hpp"
 
 using namespace std;
 using namespace Raul;
@@ -343,16 +344,9 @@ JackAudioDriver::_process_cb(jack_nframes_t nframes)
 	// FIXME: support nframes != buffer_size, even though that never damn well happens
 	assert(nframes == _buffer_size / sizeof(Sample));
 
-	// Jack can elect to not call this function for a cycle, if overloaded
-	// FIXME: this doesn't make sense, and the start time isn't used anyway
+	// Note that Jack can not call this function for a cycle, if overloaded
 	const jack_nframes_t start_of_current_cycle = jack_last_frame_time(_client);
-
-	const jack_nframes_t end_of_current_cycle = start_of_current_cycle + nframes;
-#ifndef NDEBUG
-	// FIXME: support changing cycle length
-	const jack_nframes_t start_of_last_cycle = start_of_current_cycle - nframes;
-	assert(start_of_current_cycle - start_of_last_cycle == nframes);
-#endif
+	const jack_nframes_t end_of_current_cycle   = start_of_current_cycle + nframes;
 
 	_transport_state = jack_transport_query(_client, &_position);
 
@@ -378,6 +372,10 @@ JackAudioDriver::_process_cb(jack_nframes_t nframes)
 	if (_root_patch)
 		_root_patch->process(_process_context);
 
+	// Signal message context to run if necessary
+	if (_engine.message_context()->has_requests())
+		_engine.message_context()->signal();
+
 	if (_engine.midi_driver())
 		_engine.midi_driver()->post_process(_process_context);
 
diff --git a/src/engine/LV2BlobFeature.hpp b/src/engine/LV2BlobFeature.hpp
index c672b7ac..13b6753e 100644
--- a/src/engine/LV2BlobFeature.hpp
+++ b/src/engine/LV2BlobFeature.hpp
@@ -39,10 +39,10 @@ struct BlobFeature : public Shared::LV2Features::Feature {
 	                     LV2_Reference*        reference,
 	                     LV2_Blob_Destroy      destroy_func,
 	                     uint32_t              type,
-	                     uint32_t              size) {}
+	                     size_t                size) {}
 
 	static LV2_Blob* reference_get(LV2_Blob_Support_Data data,
-	                               LV2_Reference*    ref) { return 0; }
+	                               LV2_Reference*        ref) { return 0; }
 
 	static void reference_copy(LV2_Blob_Support_Data data,
 	                           LV2_Reference*        dst,
diff --git a/src/engine/LV2EventBuffer.cpp b/src/engine/LV2EventBuffer.cpp
index 3f11790e..20df73c4 100644
--- a/src/engine/LV2EventBuffer.cpp
+++ b/src/engine/LV2EventBuffer.cpp
@@ -120,6 +120,20 @@ LV2EventBuffer::get_event(uint32_t* frames,
 }
 
 
+/** Get the object currently pointed to, or NULL if invalid.
+ */
+LV2_Object*
+LV2EventBuffer::get_object() const
+{
+	if (lv2_event_is_valid(&_iter)) {
+		uint8_t* data;
+		LV2_Event* ev = lv2_event_get(&_iter, &data);
+		return LV2_OBJECT_FROM_EVENT(ev);
+	}
+	return NULL;
+}
+
+
 /** Append an event to the buffer.
  *
  * \a timestamp must be >= the latest event in the buffer.
diff --git a/src/engine/LV2EventBuffer.hpp b/src/engine/LV2EventBuffer.hpp
index ce75d7cb..61cdba4c 100644
--- a/src/engine/LV2EventBuffer.hpp
+++ b/src/engine/LV2EventBuffer.hpp
@@ -20,6 +20,7 @@
 
 #include "event.lv2/event.h"
 #include "event.lv2/event-helpers.h"
+#include "object.lv2/object.h"
 
 namespace Ingen {
 
@@ -58,6 +59,8 @@ public:
 	               uint16_t* size,
 	               uint8_t** data) const;
 
+	LV2_Object* get_object() const;
+
 	bool append(uint32_t       frames,
 	            uint32_t       subframes,
 	            uint16_t       type,
diff --git a/src/engine/LV2Info.cpp b/src/engine/LV2Info.cpp
index 99c2ff85..93f4b386 100644
--- a/src/engine/LV2Info.cpp
+++ b/src/engine/LV2Info.cpp
@@ -38,6 +38,7 @@ LV2Info::LV2Info(Ingen::Shared::World* world)
 	, audio_class(slv2_value_new_uri(world->slv2_world, SLV2_PORT_CLASS_AUDIO))
 	, event_class(slv2_value_new_uri(world->slv2_world, SLV2_PORT_CLASS_EVENT))
 	, value_port_class(slv2_value_new_uri(world->slv2_world, LV2_OBJECT_URI "#ValuePort"))
+	, message_port_class(slv2_value_new_uri(world->slv2_world, LV2_OBJECT_URI "#MessagePort"))
 	, _world(world)
 {
 	assert(world);
@@ -58,6 +59,7 @@ LV2Info::~LV2Info()
 	slv2_value_free(audio_class);
 	slv2_value_free(event_class);
 	slv2_value_free(value_port_class);
+	slv2_value_free(message_port_class);
 }
 
 } // namespace Ingen
diff --git a/src/engine/LV2Info.hpp b/src/engine/LV2Info.hpp
index 76dfefa7..2a63f983 100644
--- a/src/engine/LV2Info.hpp
+++ b/src/engine/LV2Info.hpp
@@ -51,6 +51,7 @@ public:
 	SLV2Value audio_class;
 	SLV2Value event_class;
 	SLV2Value value_port_class;
+	SLV2Value message_port_class;
 
 	Ingen::Shared::World& world()     { return *_world; }
 	SLV2World             lv2_world() { return _world->slv2_world; }
diff --git a/src/engine/LV2Node.cpp b/src/engine/LV2Node.cpp
index dff994b9..e2f91927 100644
--- a/src/engine/LV2Node.cpp
+++ b/src/engine/LV2Node.cpp
@@ -189,13 +189,14 @@ LV2Node::instantiate(BufferFactory& bufs)
 	SLV2Value context_pred = slv2_value_new_uri(info->lv2_world(),
 			"http://lv2plug.in/ns/dev/contexts#context");
 
-	// FIXME: Why doesn't this just use lv2:default?
 	SLV2Value default_pred = slv2_value_new_uri(info->lv2_world(),
-			"http://lv2plug.in/ns/dev/string-port#default");
+			"http://lv2plug.in/ns/lv2core#default");
 
-	// FIXME: Make this a separate extension
-	SLV2Value size_pred = slv2_value_new_uri(info->lv2_world(),
-			"http://lv2plug.in/ns/dev/string-port#requiredSpace");
+	SLV2Value min_size_pred = slv2_value_new_uri(info->lv2_world(),
+			"http://lv2plug.in/ns/dev/resize-port#minimumSize");
+
+	//SLV2Value as_large_as_pred = slv2_value_new_uri(info->lv2_world(),
+	//		"http://lv2plug.in/ns/dev/resize-port#asLargeAs");
 
 	for (uint32_t j=0; j < num_ports; ++j) {
 		SLV2Port id = slv2_plugin_get_port_by_index(plug, j);
@@ -219,6 +220,11 @@ LV2Node::instantiate(BufferFactory& bufs)
 			port_buffer_size = _buffer_size;
 		} else if (slv2_port_is_a(plug, id, info->value_port_class)) {
 			data_type = PortType::VALUE;
+		} else if (slv2_port_is_a(plug, id, info->message_port_class)) {
+			data_type = PortType::MESSAGE;
+		}
+
+		if (data_type == PortType::VALUE || data_type == PortType::MESSAGE) {
 			port_buffer_size = 0;
 
 			// Get default value, and its length
@@ -236,7 +242,7 @@ LV2Node::instantiate(BufferFactory& bufs)
 			}
 
 			// Get minimum size, if set in data
-			SLV2Values sizes = slv2_port_get_value(plug, id, size_pred);
+			SLV2Values sizes = slv2_port_get_value(plug, id, min_size_pred);
 			for (uint32_t i = 0; i < slv2_values_size(sizes); ++i) {
 				SLV2Value d = slv2_values_get_at(sizes, i);
 				if (slv2_value_is_int(d)) {
diff --git a/src/engine/MessageContext.cpp b/src/engine/MessageContext.cpp
index ba03088d..5a180a10 100644
--- a/src/engine/MessageContext.cpp
+++ b/src/engine/MessageContext.cpp
@@ -24,20 +24,68 @@
 #include "PatchImpl.hpp"
 #include "PortImpl.hpp"
 #include "ProcessContext.hpp"
+#include "ThreadManager.hpp"
 
 using namespace std;
 
 namespace Ingen {
 
+
 void
 MessageContext::run(NodeImpl* node)
 {
+	if (ThreadManager::current_thread_id() == THREAD_PRE_PROCESS) {
+		assert(node);
+		Glib::Mutex::Lock lock(_mutex);
+		_request = node;
+		_sem.post();
+		_cond.wait(_mutex);
+	} else if (ThreadManager::current_thread_id() == THREAD_PROCESS) {
+		_requests.write(sizeof(NodeImpl*), &node);
+		// signal() will be called at the end of this process cycle
+	} else if (ThreadManager::current_thread_id() == THREAD_MESSAGE) {
+		cout << "Message context recursion at " << node->path() << endl;
+	} else {
+		cout << "[MessageContext] ERROR: Run requested from unknown thread" << endl;
+	}
+}
+
+
+void
+MessageContext::_run()
+{
+	NodeImpl* node = NULL;
+
+	while (true) {
+		_sem.wait();
+
+		// Run a node requested by the pre-process thread
+		{
+			Glib::Mutex::Lock lock(_mutex);
+			node = _request.get();
+			if (node) {
+				_cond.broadcast(); // Notify caller we got the message
+				run_node(node);
+			}
+		}
+
+		// Run nodes requested by the audio thread
+		while (has_requests()) {
+			_requests.full_read(sizeof(NodeImpl*), &node);
+			run_node(node);
+		}
+	}
+}
+
+
+void
+MessageContext::run_node(NodeImpl* node)
+{
 	node->message_run(*this);
 
 	void*      valid_ports = node->valid_ports();
 	PatchImpl* patch       = node->parent_patch();
 
-	//cout << "MESSAGE RUN " << node->path() << " {" << endl;
 	for (uint32_t i = 0; i < node->num_ports(); ++i) {
 		PortImpl* p = node->port_impl(i);
 		if (p->is_output() && p->context() == Context::MESSAGE &&
@@ -47,12 +95,11 @@ MessageContext::run(NodeImpl* node)
 				ConnectionImpl* ci = dynamic_cast<ConnectionImpl*>(c->get());
 				if (ci->src_port() == p) {
 					ci->dst_port()->pre_process(*_engine.message_context());
-					run(ci->dst_port()->parent_node());
+					run_node(ci->dst_port()->parent_node());
 				}
 			}
 		}
 	}
-	//cout << "}" << endl;
 
 	node->reset_valid_ports();
 }
diff --git a/src/engine/MessageContext.hpp b/src/engine/MessageContext.hpp
index 9d17d920..c871ad1c 100644
--- a/src/engine/MessageContext.hpp
+++ b/src/engine/MessageContext.hpp
@@ -18,7 +18,14 @@
 #ifndef MESSAGECONTEXT_H
 #define MESSAGECONTEXT_H
 
+#include <glibmm/thread.h>
+#include "raul/Thread.hpp"
+#include "raul/Semaphore.hpp"
+#include "raul/AtomicPtr.hpp"
+#include "object.lv2/object.h"
 #include "Context.hpp"
+#include "ThreadManager.hpp"
+#include "tuning.hpp"
 
 namespace Ingen {
 
@@ -33,14 +40,41 @@ class NodeImpl;
  *
  * \ingroup engine
  */
-class MessageContext : public Context
+class MessageContext : public Context, public Raul::Thread
 {
 public:
 	MessageContext(Engine& engine)
 		: Context(engine, MESSAGE)
-	{}
+		, Raul::Thread("message-context")
+		, _sem(0)
+		, _requests(message_context_queue_size)
+	{
+		Thread::set_context(THREAD_MESSAGE);
+	}
 
+	/** Request a run starting at node.
+	 *
+	 * Safe to call from either process thread or pre-process thread.
+	 */
 	void run(NodeImpl* node);
+
+	inline void signal() { _sem.post(); }
+	inline bool has_requests() const {
+		return _requests.read_space() >= sizeof(NodeImpl*) || _request.get();
+	}
+
+protected:
+	/** Thread run method (wait for and execute requests from process thread */
+	void _run();
+
+	/** Actually execute and propagate from node */
+	void run_node(NodeImpl* node);
+
+	Raul::Semaphore             _sem;
+	Raul::RingBuffer<NodeImpl*> _requests;
+	Glib::Mutex                 _mutex;
+	Glib::Cond                  _cond;
+	Raul::AtomicPtr<NodeImpl>   _request;
 };
 
 
diff --git a/src/engine/PatchImpl.cpp b/src/engine/PatchImpl.cpp
index 6388f648..6958c8b5 100644
--- a/src/engine/PatchImpl.cpp
+++ b/src/engine/PatchImpl.cpp
@@ -144,19 +144,22 @@ PatchImpl::process(ProcessContext& context)
 
 	NodeBase::pre_process(context);
 
-	/*if (_ports)
-		for (size_t i=0; i < _ports->size(); ++i)
-			if (_ports->at(i)->is_input() && _ports->at(i)->type() == PortType::MIDI)
-				cerr << _ports->at(i)->path() << " "
-					<< _ports->at(i)->buffer(0) << " # events: "
-					<< ((MidiBuffer*)_ports->at(i)->buffer(0))->event_count() << endl;*/
-
-	/* Run */
+	// Run all nodes
 	if (_compiled_patch && _compiled_patch->size() > 0) {
-		if (_engine.process_slaves().size() > 0)
+		if (_engine.process_slaves().size() > 0) {
 			process_parallel(context);
-		else
+		} else {
 			process_single(context);
+		}
+	}
+
+	// Queue any cross-context connections
+	for (Connections::iterator i = _connections.begin(); i != _connections.end(); ++i) {
+		ConnectionImpl* const c = (ConnectionImpl*)i->get();
+		if (c->src_port()->context() == Context::AUDIO &&
+				c->dst_port()->context() == Context::MESSAGE) {
+			c->queue(context);
+		}
 	}
 
 	NodeBase::post_process(context);
diff --git a/src/engine/PortImpl.cpp b/src/engine/PortImpl.cpp
index 23f7b5f8..45ee76da 100644
--- a/src/engine/PortImpl.cpp
+++ b/src/engine/PortImpl.cpp
@@ -18,6 +18,7 @@
 #include <iostream>
 #include "raul/Array.hpp"
 #include "raul/Maid.hpp"
+#include "contexts.lv2/contexts.h"
 #include "interface/PortType.hpp"
 #include "events/SendPortValue.hpp"
 #include "events/SendPortActivity.hpp"
@@ -75,7 +76,8 @@ PortImpl::PortImpl(BufferFactory&  bufs,
 	else
 		_polyphonic = true;
 
-	add_property("rdf:type", Atom(Atom::URI, type.uri()));
+	add_property("rdf:type",    Atom(Atom::URI, type.uri()));
+	set_context(_context);
 
 	if (type == PortType::EVENTS)
 		_broadcast = true; // send activity blips
@@ -188,6 +190,7 @@ PortImpl::broadcast_value(Context& context, bool force)
 		}
 		break;
 	case PortType::VALUE:
+	case PortType::MESSAGE:
 		LV2Object::to_atom(context.engine().world(), ((ObjectBuffer*)buffer(0).get())->object(), val);
 		break;
 	}
@@ -201,4 +204,19 @@ PortImpl::broadcast_value(Context& context, bool force)
 }
 
 
+void
+PortImpl::set_context(Context::ID c)
+{
+	_context = c;
+	switch (c) {
+	case Context::AUDIO:
+		set_property("ctx:context", Atom(Atom::URI, "ctx:AudioContext"));
+		break;
+	case Context::MESSAGE:
+		set_property("ctx:context", Atom(Atom::URI, "ctx:MessageContext"));
+		break;
+	}
+}
+
+
 } // namespace Ingen
diff --git a/src/engine/PortImpl.hpp b/src/engine/PortImpl.hpp
index e3de9f7f..e004e03f 100644
--- a/src/engine/PortImpl.hpp
+++ b/src/engine/PortImpl.hpp
@@ -108,7 +108,7 @@ public:
 	void raise_set_by_user_flag() { _set_by_user = true; }
 
 	Context::ID context() const            { return _context; }
-	void        set_context(Context::ID c) { _context = c; }
+	void        set_context(Context::ID c);
 
 protected:
 	PortImpl(BufferFactory&     bufs,
diff --git a/src/engine/ThreadManager.hpp b/src/engine/ThreadManager.hpp
index c088816f..6821949c 100644
--- a/src/engine/ThreadManager.hpp
+++ b/src/engine/ThreadManager.hpp
@@ -26,7 +26,8 @@ namespace Ingen {
 enum ThreadID {
 	THREAD_PRE_PROCESS,
 	THREAD_PROCESS,
-	THREAD_POST_PROCESS
+	THREAD_POST_PROCESS,
+	THREAD_MESSAGE,
 };
 
 
diff --git a/src/engine/events/Connect.cpp b/src/engine/events/Connect.cpp
index 18c31e03..a120aee4 100644
--- a/src/engine/events/Connect.cpp
+++ b/src/engine/events/Connect.cpp
@@ -75,9 +75,21 @@ Connect::pre_process()
 		return;
 	}
 
-	if ( ! (_src_port->type() == _dst_port->type()
-			|| ( (_src_port->type() == PortType::CONTROL || _src_port->type() == PortType::AUDIO)
-				&& (_dst_port->type() == PortType::CONTROL || _dst_port->type() == PortType::AUDIO) ))) {
+	const PortType src_type = _src_port->type();
+	const PortType dst_type = _dst_port->type();
+
+	if (	!(
+			// Equal types
+			(src_type == dst_type)
+
+			|| // or Control=>Audio or Audio=>Control
+			((src_type == PortType::CONTROL || src_type == PortType::AUDIO)
+				&& (dst_type == PortType::CONTROL || dst_type == PortType::AUDIO))
+
+			|| // or Events=>Message or Message=>Events
+			((src_type == PortType::EVENTS || src_type == PortType::MESSAGE)
+				&& (dst_type == PortType::EVENTS || dst_type == PortType::MESSAGE))
+			)) {
 		_error = TYPE_MISMATCH;
 		QueuedEvent::pre_process();
 		return;
diff --git a/src/engine/events/RequestMetadata.cpp b/src/engine/events/RequestMetadata.cpp
index 82165460..f51651fc 100644
--- a/src/engine/events/RequestMetadata.cpp
+++ b/src/engine/events/RequestMetadata.cpp
@@ -96,7 +96,7 @@ RequestMetadata::execute(ProcessContext& context)
 		if (port) {
 			if (port->type() == PortType::CONTROL || port->type() == PortType::AUDIO)
 				_value = ((AudioBuffer*)port->buffer(0).get())->value_at(0); // TODO: offset
-			else if (port->type() == PortType::VALUE)
+			else if (port->type() == PortType::VALUE || port->type() == PortType::MESSAGE)
 				LV2Object::to_atom(context.engine().world(),
 						((ObjectBuffer*)port->buffer(0).get())->object(), _value);
 		} else {
diff --git a/src/engine/tuning.hpp b/src/engine/tuning.hpp
index ea7c08c3..1ddaf71a 100644
--- a/src/engine/tuning.hpp
+++ b/src/engine/tuning.hpp
@@ -29,6 +29,7 @@ static const size_t event_queue_size           = 1024;
 static const size_t pre_processor_queue_size   = 1024;
 static const size_t post_processor_queue_size  = 1024;
 static const size_t maid_queue_size            = 1024;
+static const size_t message_context_queue_size = 1024;
 
 //static const timespec main_rate     = { 0, 500000000 }; // 1/2 second
 static const timespec main_rate       = { 0, 125000000 }; // 1/8 second
diff --git a/src/gui/Configuration.cpp b/src/gui/Configuration.cpp
index 2169093e..83de3321 100644
--- a/src/gui/Configuration.cpp
+++ b/src/gui/Configuration.cpp
@@ -24,7 +24,9 @@
 #include "client/PortModel.hpp"
 #include "client/PluginModel.hpp"
 #include "serialisation/Parser.hpp"
+#include "flowcanvas/Port.hpp"
 #include "App.hpp"
+#include "Port.hpp"
 
 using namespace std;
 
@@ -41,7 +43,7 @@ Configuration::Configuration()
 	, _control_port_color(0x4A8A0EC0)
 	, _event_port_color(  0x960909C0)
 //	, _osc_port_color(    0x5C3566C0)
-	, _object_port_color( 0x5C3566C0)
+	, _value_port_color(  0x4A4A4AC0)
 {
 }
 
@@ -95,7 +97,9 @@ Configuration::get_port_color(const PortModel* p)
 	/*} else if (p->type().is_osc()) {
 		return _osc_port_color;
 	*/} else if (p->type().is_value()) {
-		return _object_port_color;
+		return _value_port_color;
+	} else if (p->type().is_message()) {
+		return _event_port_color;
 	}
 
 	cerr << "[Configuration] Unknown port type " << p->type().uri()
diff --git a/src/gui/Configuration.hpp b/src/gui/Configuration.hpp
index f66d9f67..fef86999 100644
--- a/src/gui/Configuration.hpp
+++ b/src/gui/Configuration.hpp
@@ -20,16 +20,16 @@
 
 #include <stdint.h>
 #include <string>
+#include "raul/SharedPtr.hpp"
 
 namespace Ingen { namespace Client { class PortModel; } }
 using Ingen::Client::PortModel;
 using std::string;
 
-struct Coord { double x; double y; };
-
 namespace Ingen {
 namespace GUI {
 
+class Port;
 
 /** Singleton state manager for the entire app.
  *
@@ -52,7 +52,7 @@ public:
 	const string& patch_folder()                    { return _patch_folder; }
 	void          set_patch_folder(const string& f) { _patch_folder = f; }
 
-	uint32_t get_port_color(const PortModel* pi);
+	uint32_t get_port_color(const PortModel* p);
 
 	enum NameStyle { PATH, HUMAN, NONE };
 
@@ -68,7 +68,7 @@ private:
 	uint32_t _audio_port_color;
 	uint32_t _control_port_color;
 	uint32_t _event_port_color;
-	uint32_t _object_port_color;
+	uint32_t _value_port_color;
 };
 
 
diff --git a/src/gui/Connection.cpp b/src/gui/Connection.cpp
new file mode 100644
index 00000000..48cf7372
--- /dev/null
+++ b/src/gui/Connection.cpp
@@ -0,0 +1,43 @@
+/* This file is part of Ingen.
+ * Copyright (C) 2007-2009 Dave Robillard <http://drobilla.net>
+ *
+ * Ingen is free software; you can redistribute it and/or modify it under the
+ * terms of the GNU General Public License as published by the Free Software
+ * Foundation; either version 2 of the License, or (at your option) any later
+ * version.
+ *
+ * Ingen is distributed in the hope that it will be useful, but WITHOUT ANY
+ * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include <iostream>
+#include "client/ConnectionModel.hpp"
+#include "Connection.hpp"
+#include "Port.hpp"
+
+using namespace std;
+
+namespace Ingen {
+namespace GUI {
+
+Connection::Connection(boost::shared_ptr<FlowCanvas::Canvas>      canvas,
+                       boost::shared_ptr<ConnectionModel>         model,
+                       boost::shared_ptr<FlowCanvas::Connectable> src,
+                       boost::shared_ptr<FlowCanvas::Connectable> dst,
+                       uint32_t                                   color)
+	: FlowCanvas::Connection(canvas, src, dst, color)
+	, _connection_model(model)
+{
+	boost::shared_ptr<Port> src_port = boost::dynamic_pointer_cast<Port>(src);
+	if (src_port)
+		_bpath.property_dash() = src_port->dash();
+}
+
+
+}   // namespace GUI
+}   // namespace Ingen
diff --git a/src/gui/Connection.hpp b/src/gui/Connection.hpp
index 8abfa008..e9d5aa3c 100644
--- a/src/gui/Connection.hpp
+++ b/src/gui/Connection.hpp
@@ -22,10 +22,12 @@
 #include <string>
 #include "flowcanvas/Connection.hpp"
 #include "raul/SharedPtr.hpp"
-#include "client/ConnectionModel.hpp"
-using Ingen::Client::ConnectionModel;
 
 namespace Ingen {
+
+namespace Client { class ConnectionModel; }
+using Client::ConnectionModel;
+
 namespace GUI {
 
 
@@ -40,12 +42,7 @@ public:
 	           boost::shared_ptr<ConnectionModel>         model,
 	           boost::shared_ptr<FlowCanvas::Connectable> src,
 	           boost::shared_ptr<FlowCanvas::Connectable> dst,
-	           uint32_t                                   color)
-	: FlowCanvas::Connection(canvas, src, dst, color)
-	, _connection_model(model)
-	{}
-
-	virtual ~Connection() {}
+	           uint32_t                                   color);
 
 	SharedPtr<ConnectionModel> model() const { return _connection_model; }
 
diff --git a/src/gui/ControlPanel.cpp b/src/gui/ControlPanel.cpp
index fdfba8b7..4aea2f08 100644
--- a/src/gui/ControlPanel.cpp
+++ b/src/gui/ControlPanel.cpp
@@ -120,7 +120,7 @@ ControlPanel::add_port(SharedPtr<PortModel> pm)
 			Glib::RefPtr<Gnome::Glade::Xml> xml = GladeFactory::new_glade_reference("toggle_control");
 			xml->get_widget_derived("toggle_control", tc);
 			control = tc;
-		} else if (pm->type().is_value()) {
+		} else if (pm->type().is_value() || pm->type().is_message()) {
 			StringControl* sc;
 			Glib::RefPtr<Gnome::Glade::Xml> xml = GladeFactory::new_glade_reference("string_control");
 			xml->get_widget_derived("string_control", sc);
diff --git a/src/gui/Port.cpp b/src/gui/Port.cpp
index 012e8fcd..ba4e0482 100644
--- a/src/gui/Port.cpp
+++ b/src/gui/Port.cpp
@@ -35,6 +35,7 @@ namespace Ingen {
 using namespace Shared;
 namespace GUI {
 
+ArtVpathDash* Port::_dash;
 
 /** @a flip Make an input port appear as an output port, and vice versa.
  */
@@ -55,6 +56,10 @@ Port::Port(
 	delete _menu;
 	_menu = NULL;
 
+	ArtVpathDash* dash = this->dash();
+	_rect->property_dash() = dash;
+	set_border_width(dash ? 2.0 : 0.0);
+
 	pm->signal_moved.connect(sigc::mem_fun(this, &Port::moved));
 
 	if (pm->type().is_control()) {
@@ -148,9 +153,37 @@ Port::property_changed(const URI& key, const Atom& value)
 	} else if (value.type() == Atom::BOOL) {
 		if ((key.str() == "lv2:toggled"))
 			set_toggled(value.get_bool());
+	} else if (value.type() == Atom::URI) {
+		ArtVpathDash* dash = this->dash();
+		_rect->property_dash() = dash;
+		set_border_width(dash ? 2.0 : 0.0);
 	}
 }
 
 
+ArtVpathDash*
+Port::dash()
+{
+	SharedPtr<PortModel> pm = _port_model.lock();
+	if (!pm)
+		return NULL;
+
+	const Raul::Atom& context = pm->get_property("ctx:context");
+	if (!context.is_valid() || context.type() != Atom::URI
+			|| !strcmp(context.get_uri(), "ctx:AudioContext"))
+		return NULL;
+
+	if (!_dash) {
+		_dash = new ArtVpathDash();
+		_dash->n_dash = 2;
+		_dash->dash = art_new(double, 2);
+		_dash->dash[0] = 4;
+		_dash->dash[1] = 4;
+	}
+
+	return _dash;
+}
+
+
 } // namespace GUI
 } // namespace Ingen
diff --git a/src/gui/Port.hpp b/src/gui/Port.hpp
index db511041..ec361a9f 100644
--- a/src/gui/Port.hpp
+++ b/src/gui/Port.hpp
@@ -56,11 +56,15 @@ public:
 	void value_changed(const Raul::Atom& value);
 	void activity();
 
+	ArtVpathDash* dash();
+
 private:
 	void property_changed(const Raul::URI& key, const Raul::Atom& value);
 
 	void moved();
 
+	static ArtVpathDash* _dash;
+
 	WeakPtr<PortModel> _port_model;
 	bool               _flipped;
 };
diff --git a/src/gui/wscript b/src/gui/wscript
index ddcba166..c3e618b6 100644
--- a/src/gui/wscript
+++ b/src/gui/wscript
@@ -8,6 +8,7 @@ def build(bld):
 		BreadCrumbs.cpp
 		Configuration.cpp
 		ConnectWindow.cpp
+		Connection.cpp
 		ControlPanel.cpp
 		Controls.cpp
 		GladeFactory.cpp
diff --git a/src/ingen/main.cpp b/src/ingen/main.cpp
index 4cdeb552..a7762a72 100644
--- a/src/ingen/main.cpp
+++ b/src/ingen/main.cpp
@@ -115,7 +115,7 @@ main(int argc, char** argv)
 	world->rdf_world->add_prefix("ingenuity", "http://drobilla.net/ns/ingenuity#");
 	world->rdf_world->add_prefix("lv2",       "http://lv2plug.in/ns/lv2core#");
 	world->rdf_world->add_prefix("lv2ev",     "http://lv2plug.in/ns/ext/event#");
-	world->rdf_world->add_prefix("lv2midi",   "http://lv2plug.in/ns/ext/midi");
+	world->rdf_world->add_prefix("lv2midi",   "http://lv2plug.in/ns/ext/midi#");
 	world->rdf_world->add_prefix("owl",       "http://www.w3.org/2002/07/owl#");
 	world->rdf_world->add_prefix("rdfs",      "http://www.w3.org/2000/01/rdf-schema#");
 	world->rdf_world->add_prefix("sp",        "http://lv2plug.in/ns/dev/string-port#");
diff --git a/src/shared/LV2Object.cpp b/src/shared/LV2Object.cpp
index 54c81429..657bee29 100644
--- a/src/shared/LV2Object.cpp
+++ b/src/shared/LV2Object.cpp
@@ -75,16 +75,17 @@ from_atom(World* world, const Raul::Atom& atom, LV2_Object* object)
 		break;
 	case Raul::Atom::STRING:
 		object->type = map->object_class_string;
-		object->size = std::min(object->size, (uint32_t)strlen(atom.get_string()) + 1);
+		object->size = std::min((uint16_t)object->size, (uint16_t)(strlen(atom.get_string()) + 1));
 		str = ((char*)(object + 1));
 		str[object->size - 1] = '\0';
 		strncpy(str, atom.get_string(), object->size);
 		break;
 	case Raul::Atom::BLOB:
-		object->type = map->object_class_string;
-		*(uint32_t*)(object + 1) = map->uri_to_id(NULL, atom.get_blob_type());
+		cerr << "TODO: Blob support" << endl;
+		/*object->type = map->object_class_string;
+		*(uint16_t*)(object + 1) = map->uri_to_id(NULL, atom.get_blob_type());
 		memcpy(((char*)(object + 1) + sizeof(uint32_t)), atom.get_blob(),
-				std::min(atom.data_size(), (size_t)object->size));
+				std::min(atom.data_size(), (size_t)object->size));*/
 	default:
 		cerr << "Unsupported value type for toggle control" << endl;
 		return false;
diff --git a/src/shared/ResourceImpl.cpp b/src/shared/ResourceImpl.cpp
index 5f23caf3..6af25ef0 100644
--- a/src/shared/ResourceImpl.cpp
+++ b/src/shared/ResourceImpl.cpp
@@ -115,6 +115,9 @@ ResourceImpl::type(
 			} else if (!strcmp(atom.get_uri(), "obj:ValuePort")) {
 				data_type = PortType::VALUE;
 				port = true;
+			} else if (!strcmp(atom.get_uri(), "obj:MessagePort")) {
+				data_type = PortType::MESSAGE;
+				port = true;
 			}
 		}
 	}