Zero-copy to/from driver ports where possible

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

31 files changed, 466 insertions, 314 deletions

diff --git a/src/server/BlockImpl.hpp b/src/server/BlockImpl.hpp
index 2dcc4762..06111a88 100644
--- a/src/server/BlockImpl.hpp
+++ b/src/server/BlockImpl.hpp
@@ -14,8 +14,8 @@
   along with Ingen.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-#ifndef INGEN_ENGINE_NODEIMPL_HPP
-#define INGEN_ENGINE_NODEIMPL_HPP
+#ifndef INGEN_ENGINE_BLOCKIMPL_HPP
+#define INGEN_ENGINE_BLOCKIMPL_HPP
 
 #include <list>
 
diff --git a/src/server/Buffer.cpp b/src/server/Buffer.cpp
index 1ee7156a..d4aa37c4 100644
--- a/src/server/Buffer.cpp
+++ b/src/server/Buffer.cpp
@@ -42,7 +42,9 @@ namespace Server {
 Buffer::Buffer(BufferFactory& bufs,
                LV2_URID       type,
                LV2_URID       value_type,
-               uint32_t       capacity)
+               uint32_t       capacity,
+               bool           external,
+               void*          buf)
 	: _factory(bufs)
 	, _type(type)
 	, _value_type(value_type)
@@ -50,40 +52,54 @@ Buffer::Buffer(BufferFactory& bufs,
 	, _latest_event(0)
 	, _next(NULL)
 	, _refs(0)
+	, _external(external)
 {
+	if (!external) {
 #ifdef HAVE_POSIX_MEMALIGN
-	int ret = posix_memalign((void**)&_atom, 16, capacity);
+		int ret = posix_memalign((void**)&_buf, 16, capacity);
+		if (!ret) {
+			memset(_buf, 0, capacity);
+		}
 #else
-	_atom = (LV2_Atom*)malloc(capacity);
-	int ret = (_atom != NULL) ? 0 : -1;
+		_buf = (LV2_buf*)calloc(1, capacity);
+		int ret = (_buf != NULL) ? 0 : -1;
 #endif
 
-	if (ret) {
-		bufs.engine().log().error("Failed to allocate event buffer\n");
-		throw std::bad_alloc();
+		if (ret) {
+			bufs.engine().log().error("Failed to allocate event buffer\n");
+			throw std::bad_alloc();
+		}
 	}
 
-	memset(_atom, 0, capacity);
-	_atom->size = capacity - sizeof(LV2_Atom);
-	_atom->type = type;
+	if (type != bufs.uris().atom_Sound) {
+		/* Audio buffers are not atoms, the buffer is the start of a float
+		   array which is already silent since the buffer is zeroed.  All other
+		   buffers are atoms. */
+		if (_buf) {
+			LV2_Atom* atom = get<LV2_Atom>();
+			atom->size = capacity - sizeof(LV2_Atom);
+			atom->type = type;
 
-	if (type == bufs.uris().atom_Sound) {
-		// Audio port (Vector of float)
-		LV2_Atom_Vector* vec = (LV2_Atom_Vector*)_atom;
-		vec->body.child_size = sizeof(float);
-		vec->body.child_type = bufs.uris().atom_Float;
-	}
+			clear();
+		}
 
-	if (type == bufs.uris().atom_Sequence && value_type) {
-		_value_buffer = bufs.get_buffer(value_type, 0, 0, false);
+		if (value_type && value_type != type) {
+			/* Buffer with a different value type.  These buffers (probably
+			   sequences) have a "value" that persists independently of the buffer
+			   contents.  This is used to represent things like a Sequence of
+			   Float, which acts like an individual float (has a value), but the
+			   buffer itself only transmits changes and does not necessarily
+			   contain the current value. */
+			_value_buffer = bufs.get_buffer(value_type, 0, 0, false);
+		}
 	}
-
-	clear();
 }
 
 Buffer::~Buffer()
 {
-	free(_atom);
+	if (!_external) {
+		free(_buf);
+	}
 }
 
 void
@@ -104,13 +120,14 @@ Buffer::set_type(LV2_URID type, LV2_URID value_type)
 void
 Buffer::clear()
 {
-	if (is_audio() || is_control()) {
-		_atom->size = _capacity - sizeof(LV2_Atom);
-		set_block(0, 0, nframes());
+	if (is_audio() && _buf) {
+		memset(_buf, 0, _capacity);
+	} else if (is_control()) {
+		get<LV2_Atom_Float>()->body = 0;
 	} else if (is_sequence()) {
-		LV2_Atom_Sequence* seq = (LV2_Atom_Sequence*)_atom;
-		_atom->type    = _factory.uris().atom_Sequence;
-		_atom->size    = sizeof(LV2_Atom_Sequence_Body);
+		LV2_Atom_Sequence* seq = get<LV2_Atom_Sequence>();
+		seq->atom.type = _factory.uris().atom_Sequence;
+		seq->atom.size = sizeof(LV2_Atom_Sequence_Body);
 		seq->body.unit = 0;
 		seq->body.pad  = 0;
 		_latest_event  = 0;
@@ -121,7 +138,7 @@ void
 Buffer::render_sequence(const Context& context, const Buffer* src, bool add)
 {
 	const LV2_URID           atom_Float = _factory.uris().atom_Float;
-	const LV2_Atom_Sequence* seq        = (const LV2_Atom_Sequence*)src->atom();
+	const LV2_Atom_Sequence* seq        = src->get<const LV2_Atom_Sequence>();
 	const LV2_Atom_Float*    init       = (const LV2_Atom_Float*)src->value();
 	float                    value      = init ? init->body : 0.0f;
 	SampleCount              offset     = context.offset();
@@ -138,8 +155,18 @@ Buffer::render_sequence(const Context& context, const Buffer* src, bool add)
 void
 Buffer::copy(const Context& context, const Buffer* src)
 {
-	if (_type == src->type() && src->_atom->size + sizeof(LV2_Atom) <= _capacity) {
-		memcpy(_atom, src->_atom, sizeof(LV2_Atom) + src->_atom->size);
+	if (!_buf) {
+		return;
+	} else if (is_audio() && src->is_audio()) {
+		memcpy(_buf, src->_buf, src->_capacity);
+	} else if (_type == src->type()) {
+		const LV2_Atom* src_atom = src->get<const LV2_Atom>();
+		if (lv2_atom_total_size(src_atom) <= _capacity) {
+			memcpy(_buf, src_atom, lv2_atom_total_size(src_atom));
+		} else {
+			clear();
+		}
+
 		if (value() && src->value()) {
 			memcpy(value(), src->value(), lv2_atom_total_size(src->value()));
 		}
@@ -158,9 +185,13 @@ Buffer::copy(const Context& context, const Buffer* src)
 void
 Buffer::resize(uint32_t capacity)
 {
-	_atom     = (LV2_Atom*)realloc(_atom, capacity);
-	_capacity = capacity;
-	clear();
+	if (!_external) {
+		_buf      = realloc(_buf, capacity);
+		_capacity = capacity;
+		clear();
+	} else {
+		_factory.engine().log().error("Attempt to resize external buffer\n");
+	}
 }
 
 void*
@@ -168,16 +199,15 @@ Buffer::port_data(PortType port_type, SampleCount offset)
 {
 	switch (port_type.id()) {
 	case PortType::ID::CONTROL:
-	case PortType::ID::CV:
-	case PortType::ID::AUDIO:
-		if (_atom->type == _factory.uris().atom_Float) {
-			return (float*)LV2_ATOM_BODY(_atom);
-		} else if (_atom->type == _factory.uris().atom_Sound) {
-			return (float*)LV2_ATOM_CONTENTS(LV2_Atom_Vector, _atom) + offset;
+		if (_type == _factory.uris().atom_Float) {
+			return &get<LV2_Atom_Float>()->body;
 		}
 		break;
+	case PortType::ID::CV:
+	case PortType::ID::AUDIO:
+		return (Sample*)_buf;
 	default:
-		return _atom;
+		return _buf;
 	}
 	return NULL;
 }
@@ -246,8 +276,10 @@ void
 Buffer::prepare_write(Context& context)
 {
 	if (_type == _factory.uris().atom_Sequence) {
-		_atom->type   = (LV2_URID)_factory.uris().atom_Sequence;
-		_atom->size   = sizeof(LV2_Atom_Sequence_Body);
+		LV2_Atom* atom = get<LV2_Atom>();
+
+		atom->type    = (LV2_URID)_factory.uris().atom_Sequence;
+		atom->size    = sizeof(LV2_Atom_Sequence_Body);
 		_latest_event = 0;
 	}
 }
@@ -256,8 +288,10 @@ void
 Buffer::prepare_output_write(Context& context)
 {
 	if (_type == _factory.uris().atom_Sequence) {
-		_atom->type   = (LV2_URID)_factory.uris().atom_Chunk;
-		_atom->size   = _capacity - sizeof(LV2_Atom);
+		LV2_Atom* atom = get<LV2_Atom>();
+
+		atom->type    = (LV2_URID)_factory.uris().atom_Chunk;
+		atom->size    = _capacity - sizeof(LV2_Atom);
 		_latest_event = 0;
 	}
 }
@@ -269,16 +303,18 @@ Buffer::append_event(int64_t        frames,
                      const uint8_t* data)
 {
 	assert(frames >= _latest_event);
-	if (_atom->type == _factory.uris().atom_Chunk) {
+
+	LV2_Atom* atom = get<LV2_Atom>();
+	if (atom->type == _factory.uris().atom_Chunk) {
 		// Chunk initialized with prepare_output_write(), clear
 		clear();
 	}
 
-	if (sizeof(LV2_Atom) + _atom->size + lv2_atom_pad_size(size) > _capacity) {
+	if (sizeof(LV2_Atom) + atom->size + lv2_atom_pad_size(size) > _capacity) {
 		return false;
 	}
 
-	LV2_Atom_Sequence* seq = (LV2_Atom_Sequence*)_atom;
+	LV2_Atom_Sequence* seq = (LV2_Atom_Sequence*)atom;
 	LV2_Atom_Event*    ev  = (LV2_Atom_Event*)(
 		(uint8_t*)seq + lv2_atom_total_size(&seq->atom));
 
@@ -287,7 +323,7 @@ Buffer::append_event(int64_t        frames,
 	ev->body.type   = type;
 	memcpy(ev + 1, data, size);
 
-	_atom->size += sizeof(LV2_Atom_Event) + lv2_atom_pad_size(size);
+	atom->size += sizeof(LV2_Atom_Event) + lv2_atom_pad_size(size);
 
 	_latest_event = frames;
 
@@ -297,8 +333,9 @@ Buffer::append_event(int64_t        frames,
 SampleCount
 Buffer::next_value_offset(SampleCount offset, SampleCount end) const
 {
-	SampleCount earliest = end;
-	LV2_ATOM_SEQUENCE_FOREACH((LV2_Atom_Sequence*)_atom, ev) {
+	SampleCount              earliest = end;
+	const LV2_Atom_Sequence* seq      = get<const LV2_Atom_Sequence>();
+	LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
 		if (ev->time.frames >  offset   &&
 		    ev->time.frames <  earliest &&
 		    ev->body.type   == _value_type) {
@@ -312,13 +349,13 @@ Buffer::next_value_offset(SampleCount offset, SampleCount end) const
 const LV2_Atom*
 Buffer::value() const
 {
-	return _value_buffer ? _value_buffer->atom() : NULL;
+	return _value_buffer ? _value_buffer->get<const LV2_Atom>() : NULL;
 }
 
 LV2_Atom*
 Buffer::value()
 {
-	return _value_buffer ? _value_buffer->atom() : NULL;
+	return _value_buffer ? _value_buffer->get<LV2_Atom>() : NULL;
 }
 
 void
@@ -328,9 +365,10 @@ Buffer::update_value_buffer(SampleCount offset)
 		return;
 	}
 
-	LV2_ATOM_SEQUENCE_FOREACH((LV2_Atom_Sequence*)_atom, ev) {
+	LV2_Atom_Sequence* seq = get<LV2_Atom_Sequence>();
+	LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
 		if (ev->time.frames <= offset && ev->body.type == _value_type) {
-			memcpy(_value_buffer->atom(),
+			memcpy(_value_buffer->get<LV2_Atom>(),
 			       &ev->body,
 			       lv2_atom_total_size(&ev->body));
 			break;
diff --git a/src/server/Buffer.hpp b/src/server/Buffer.hpp
index ada8a43d..11037dd1 100644
--- a/src/server/Buffer.hpp
+++ b/src/server/Buffer.hpp
@@ -45,7 +45,9 @@ public:
 	Buffer(BufferFactory& bufs,
 	       LV2_URID       type,
 	       LV2_URID       value_type,
-	       uint32_t       capacity);
+	       uint32_t       capacity,
+	       bool           external = false,
+	       void*          buf = NULL);
 
 	void clear();
 	void resize(uint32_t size);
@@ -73,18 +75,12 @@ public:
 		return _type == _factory.uris().atom_Sequence;
 	}
 
-	inline bool empty() const {
-		return (_atom->type != _type ||
-		        (_type == _factory.uris().atom_Sequence &&
-		         _atom->size <= sizeof(LV2_Atom_Sequence_Body)));
-	}
-
-	/// Audio buffers only
+	/// Audio or float buffers only
 	inline const Sample* samples() const {
 		if (is_control()) {
-			return (const Sample*)LV2_ATOM_BODY_CONST(atom());
+			return (const Sample*)LV2_ATOM_BODY_CONST(get<LV2_Atom_Float>());
 		} else if (is_audio()) {
-			return (const Sample*)LV2_ATOM_CONTENTS_CONST(LV2_Atom_Vector, atom());
+			return (const Sample*)_buf;
 		}
 		return NULL;
 	}
@@ -92,9 +88,9 @@ public:
 	/// Audio buffers only
 	inline Sample* samples() {
 		if (is_control()) {
-			return (Sample*)LV2_ATOM_BODY(atom());
+			return (Sample*)LV2_ATOM_BODY(get<LV2_Atom_Float>());
 		} else if (is_audio()) {
-			return (Sample*)LV2_ATOM_CONTENTS(LV2_Atom_Vector, atom());
+			return (Sample*)_buf;
 		}
 		return NULL;
 	}
@@ -104,7 +100,7 @@ public:
 		if (is_control()) {
 			return 1;
 		} else if (is_audio()) {
-			return (_capacity - sizeof(LV2_Atom_Vector)) / sizeof(Sample);
+			return (_capacity / sizeof(Sample));
 		}
 		return 0;
 	}
@@ -152,8 +148,9 @@ public:
 	{
 		if (add) {
 			add_block(val, start, end);
+		} else {
+			set_block(val, start, end);
 		}
-		set_block(val, start, end);
 	}
 
 	/// Audio buffers only
@@ -184,11 +181,13 @@ public:
 	/// Set/add to audio buffer from the Sequence of Float in `src`
 	void render_sequence(const Context& context, const Buffer* src, bool add);
 
-	LV2_Atom*       atom()       { return _atom; }
-	const LV2_Atom* atom() const { return _atom; }
-
 	void set_capacity(uint32_t capacity) { _capacity = capacity; }
 
+	void set_buffer(void* buf) { assert(_external); _buf = buf; }
+
+	template<typename T> const T* get() const { return reinterpret_cast<const T*>(_buf); }
+	template<typename T> T*       get()       { return reinterpret_cast<T*>(_buf); }
+
 	inline void ref() { ++_refs; }
 
 	inline void deref() {
@@ -199,7 +198,7 @@ public:
 
 protected:
 	BufferFactory& _factory;
-	LV2_Atom*      _atom;
+	void*          _buf;
 	LV2_URID       _type;
 	LV2_URID       _value_type;
 	uint32_t       _capacity;
@@ -213,8 +212,9 @@ protected:
 private:
 	void recycle();
 
-	Buffer*               _next;  ///< Intrusive linked list for BufferFactory
-	std::atomic<unsigned> _refs;  ///< Intrusive reference count
+	Buffer*               _next;      ///< Intrusive linked list for BufferFactory
+	std::atomic<unsigned> _refs;      ///< Intrusive reference count
+	bool                  _external;  ///< Buffer is externally allocated
 };
 
 } // namespace Server
diff --git a/src/server/BufferFactory.cpp b/src/server/BufferFactory.cpp
index 599f8fc5..aeaa0d44 100644
--- a/src/server/BufferFactory.cpp
+++ b/src/server/BufferFactory.cpp
@@ -71,7 +71,7 @@ BufferFactory::set_block_length(SampleCount block_length)
 uint32_t
 BufferFactory::audio_buffer_size(SampleCount nframes)
 {
-	return sizeof(LV2_Atom_Vector) + (nframes * sizeof(float));
+	return nframes * sizeof(Sample);
 }
 
 uint32_t
diff --git a/src/server/ControlBindings.cpp b/src/server/ControlBindings.cpp
index c4a95476..cdce838c 100644
--- a/src/server/ControlBindings.cpp
+++ b/src/server/ControlBindings.cpp
@@ -411,7 +411,7 @@ ControlBindings::pre_process(ProcessContext& context, Buffer* buffer)
 		return;
 	}
 
-	LV2_Atom_Sequence* seq = (LV2_Atom_Sequence*)buffer->atom();
+	LV2_Atom_Sequence* seq = buffer->get<LV2_Atom_Sequence>();
 	LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
 		if (ev->body.type == uris.midi_MidiEvent) {
 			const uint8_t* buf = (const uint8_t*)LV2_ATOM_BODY(&ev->body);
diff --git a/src/server/DuplexPort.cpp b/src/server/DuplexPort.cpp
index 05bf1b5a..f116ae09 100644
--- a/src/server/DuplexPort.cpp
+++ b/src/server/DuplexPort.cpp
@@ -29,19 +29,18 @@ namespace Ingen {
 namespace Server {
 
 DuplexPort::DuplexPort(BufferFactory&      bufs,
-                       BlockImpl*          parent,
+                       GraphImpl*          parent,
                        const Raul::Symbol& symbol,
                        uint32_t            index,
                        bool                polyphonic,
-                       uint32_t            poly,
                        PortType            type,
-                       LV2_URID            buffer_type,
+                       LV2_URID            buf_type,
+                       size_t              buf_size,
                        const Atom&         value,
-                       size_t              buffer_size,
                        bool                is_output)
-	: PortImpl(bufs, parent, symbol, index, poly, type, buffer_type, value, buffer_size)
-	, InputPort(bufs, parent, symbol, index, poly, type, buffer_type, value, buffer_size)
-	, OutputPort(bufs, parent, symbol, index, poly, type, buffer_type, value, buffer_size)
+	: PortImpl(bufs, parent, symbol, index, parent->polyphony(), type, buf_type, value, buf_size)
+	, InputPort(bufs, parent, symbol, index, parent->polyphony(), type, buf_type, value, buf_size)
+	, OutputPort(bufs, parent, symbol, index, parent->polyphony(), type, buf_type, value, buf_size)
 	, _is_output(is_output)
 {
 	if (polyphonic) {
@@ -78,8 +77,8 @@ DuplexPort::duplicate(Engine&             engine,
 	DuplexPort* dup = new DuplexPort(
 		bufs, parent, symbol, _index,
 		polyphonic.type() == bufs.uris().atom_Bool && polyphonic.get<int32_t>(),
-		_poly, _type, _buffer_type,
-		_value, _buffer_size, _is_output);
+		_type, _buffer_type, _buffer_size,
+		_value, _is_output);
 
 	dup->set_properties(properties());
 
@@ -129,11 +128,28 @@ DuplexPort::get_buffers(BufferFactory&      bufs,
                         uint32_t            poly,
                         bool                real_time) const
 {
-	if (_is_output) {
-		return InputPort::get_buffers(bufs, voices, poly, real_time);
-	} else {
-		return OutputPort::get_buffers(bufs, voices, poly, real_time);
+	if (!_is_driver_port) {
+		if (_is_output) {
+			return InputPort::get_buffers(bufs, voices, poly, real_time);
+		} else {
+			return OutputPort::get_buffers(bufs, voices, poly, real_time);
+		}
 	}
+	return false;
+}
+
+void
+DuplexPort::set_is_driver_port(BufferFactory& bufs)
+{
+	_voices->at(0).buffer = new Buffer(bufs, buffer_type(), _value.type(), 0, true, NULL);
+	PortImpl::set_is_driver_port(bufs);
+}
+
+void
+DuplexPort::set_driver_buffer(void* buf, uint32_t capacity)
+{
+	_voices->at(0).buffer->set_buffer(buf);
+	_voices->at(0).buffer->set_capacity(capacity);
 }
 
 uint32_t
@@ -192,9 +208,8 @@ DuplexPort::post_process(Context& context)
 		   (external perspective) is ready. */
 		InputPort::pre_process(context);
 		InputPort::pre_run(context);
-	} else {
-		monitor(context);
 	}
+	monitor(context);
 }
 
 SampleCount
diff --git a/src/server/DuplexPort.hpp b/src/server/DuplexPort.hpp
index 1dd1292c..36e37b0b 100644
--- a/src/server/DuplexPort.hpp
+++ b/src/server/DuplexPort.hpp
@@ -43,15 +43,14 @@ class DuplexPort : public InputPort
 {
 public:
 	DuplexPort(BufferFactory&      bufs,
-	           BlockImpl*          parent,
+	           GraphImpl*          parent,
 	           const Raul::Symbol& symbol,
 	           uint32_t            index,
 	           bool                polyphonic,
-	           uint32_t            poly,
 	           PortType            type,
-	           LV2_URID            buffer_type,
+	           LV2_URID            buf_type,
+	           size_t              buf_size,
 	           const Atom&         value,
-	           size_t              buffer_size,
 	           bool                is_output);
 
 	virtual ~DuplexPort();
@@ -77,6 +76,16 @@ public:
 	                 uint32_t            poly,
 	                 bool                real_time) const;
 
+
+	virtual void set_is_driver_port(BufferFactory& bufs);
+
+	/** Set the external driver-provided buffer.
+	 *
+	 * This may only be called in the process thread, after an earlier call to
+	 * prepare_driver_buffer().
+	 */
+	void set_driver_buffer(void* buf, uint32_t capacity);
+
 	void pre_process(Context& context);
 	void post_process(Context& context);
 
diff --git a/src/server/GraphImpl.cpp b/src/server/GraphImpl.cpp
index 256d607a..535269a6 100644
--- a/src/server/GraphImpl.cpp
+++ b/src/server/GraphImpl.cpp
@@ -311,29 +311,6 @@ GraphImpl::num_ports_non_rt() const
 	return _inputs.size() + _outputs.size();
 }
 
-DuplexPort*
-GraphImpl::create_port(BufferFactory&      bufs,
-                       const Raul::Symbol& symbol,
-                       PortType            type,
-                       LV2_URID            buffer_type,
-                       uint32_t            buffer_size,
-                       bool                is_output,
-                       bool                polyphonic)
-{
-	if (type == PortType::UNKNOWN) {
-		bufs.engine().log().error(fmt("Unknown port type %1%\n")
-		                          % type.uri());
-		return NULL;
-	}
-
-	Atom value;
-	if (type == PortType::CONTROL || type == PortType::CV)
-		value = bufs.forge().make(0.0f);
-
-	return new DuplexPort(bufs, this, symbol, num_ports_non_rt(), polyphonic, _polyphony,
-	                      type, buffer_type, value, buffer_size, is_output);
-}
-
 void
 GraphImpl::remove_port(DuplexPort& port)
 {
diff --git a/src/server/GraphImpl.hpp b/src/server/GraphImpl.hpp
index c48754a8..bedf902c 100644
--- a/src/server/GraphImpl.hpp
+++ b/src/server/GraphImpl.hpp
@@ -117,19 +117,6 @@ public:
 
 	uint32_t num_ports_non_rt() const;
 
-	/** Create a port to be later added to this graph.
-	 * Not realtime safe.  This function is to be called by events in the
-	 * pre-process thread to create ports which will later be installed in the
-	 * process thread.
-	 */
-	DuplexPort* create_port(BufferFactory&      bufs,
-	                        const Raul::Symbol& symbol,
-	                        PortType            type,
-	                        LV2_URID            buffer_type,
-	                        uint32_t            buffer_size,
-	                        bool                is_output,
-	                        bool                polyphonic);
-
 	typedef boost::intrusive::slist<
 		DuplexPort, boost::intrusive::constant_time_size<true> > Ports;
 
diff --git a/src/server/InternalBlock.cpp b/src/server/InternalBlock.cpp
new file mode 100644
index 00000000..5fa0bdc0
--- /dev/null
+++ b/src/server/InternalBlock.cpp
@@ -0,0 +1,51 @@
+/*
+  This file is part of Ingen.
+  Copyright 2007-2015 David Robillard <http://drobilla.net/>
+
+  Ingen is free software: you can redistribute it and/or modify it under the
+  terms of the GNU Affero General Public License as published by the Free
+  Software Foundation, either version 3 of the License, or 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 Affero General Public License for details.
+
+  You should have received a copy of the GNU Affero General Public License
+  along with Ingen.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "Buffer.hpp"
+#include "InternalBlock.hpp"
+#include "PortImpl.hpp"
+
+namespace Ingen {
+namespace Server {
+
+InternalBlock::InternalBlock(PluginImpl*         plugin,
+                             const Raul::Symbol& symbol,
+                             bool                poly,
+                             GraphImpl*          parent,
+                             SampleRate          rate)
+	: BlockImpl(plugin, symbol, poly, parent, rate)
+{}
+
+
+void
+InternalBlock::pre_process(ProcessContext& context) {
+	/* Output sequences are initialized in LV2 format, an atom:Chunk with size
+	   set to the capacity of the buffer.  Internal nodes don't care, so clear
+	   to an empty sequences so appending events results in a valid output. */
+	for (uint32_t i = 0; i < num_ports(); ++i) {
+		PortImpl* const port = _ports->at(i);
+		if (port->is_input()) {
+			port->pre_process(context);
+		} else {
+			for (uint32_t v = 0; v < port->poly(); ++v) {
+				port->buffer(v)->clear();
+			}
+		}
+	}
+}
+
+} // namespace Server
+} // namespace Ingen
diff --git a/src/server/InternalBlock.hpp b/src/server/InternalBlock.hpp
new file mode 100644
index 00000000..c88e87d1
--- /dev/null
+++ b/src/server/InternalBlock.hpp
@@ -0,0 +1,44 @@
+/*
+  This file is part of Ingen.
+  Copyright 2007-2015 David Robillard <http://drobilla.net/>
+
+  Ingen is free software: you can redistribute it and/or modify it under the
+  terms of the GNU Affero General Public License as published by the Free
+  Software Foundation, either version 3 of the License, or 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 Affero General Public License for details.
+
+  You should have received a copy of the GNU Affero General Public License
+  along with Ingen.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef INGEN_ENGINE_INTERNALBLOCK_HPP
+#define INGEN_ENGINE_INTERNALBLOCK_HPP
+
+#include "BlockImpl.hpp"
+
+namespace Ingen {
+namespace Server {
+
+/** An internal Block implemented inside Ingen.
+ *
+ * \ingroup engine
+ */
+class InternalBlock : public BlockImpl
+{
+public:
+	InternalBlock(PluginImpl*         plugin,
+	              const Raul::Symbol& symbol,
+	              bool                poly,
+	              GraphImpl*          parent,
+	              SampleRate          rate);
+
+	virtual void pre_process(ProcessContext& context);
+};
+
+} // namespace Server
+} // namespace Ingen
+
+#endif // INGEN_ENGINE_BLOCKIMPL_HPP
diff --git a/src/server/JackDriver.cpp b/src/server/JackDriver.cpp
index 94658cc3..6b0d2a05 100644
--- a/src/server/JackDriver.cpp
+++ b/src/server/JackDriver.cpp
@@ -212,6 +212,14 @@ void
 JackDriver::add_port(ProcessContext& context, EnginePort* port)
 {
 	_ports.push_back(*port);
+
+	DuplexPort* graph_port = port->graph_port();
+	if (graph_port->is_a(PortType::AUDIO) || graph_port->is_a(PortType::CV)) {
+		const SampleCount nframes = context.nframes();
+		jack_port_t*      jport   = (jack_port_t*)port->handle();
+		void*             jbuf    = jack_port_get_buffer(jport, nframes);
+		graph_port->set_driver_buffer(jbuf, nframes * sizeof(float));
+	}
 }
 
 void
@@ -250,6 +258,8 @@ JackDriver::unregister_port(EnginePort& port)
 	if (jack_port_unregister(_client, (jack_port_t*)port.handle())) {
 		_engine.log().error("Failed to unregister Jack port\n");
 	}
+
+	port.set_handle(NULL);
 }
 
 void
@@ -302,94 +312,95 @@ JackDriver::port_property_internal(const jack_port_t* jport,
 EnginePort*
 JackDriver::create_port(DuplexPort* graph_port)
 {
-	if (graph_port &&
-	    (graph_port->is_a(PortType::AUDIO) ||
-	     graph_port->is_a(PortType::CV) ||
-	     (graph_port->is_a(PortType::ATOM) &&
-	      graph_port->buffer_type() == _engine.world()->uris().atom_Sequence))) {
-		EnginePort* eport = new EnginePort(graph_port);
+	EnginePort* eport = NULL;
+	if (graph_port->is_a(PortType::AUDIO) || graph_port->is_a(PortType::CV)) {
+		// Audio buffer port, use Jack buffer directly
+		eport = new EnginePort(graph_port);
+		graph_port->set_is_driver_port(*_engine.buffer_factory());
+	} else if (graph_port->is_a(PortType::ATOM) &&
+	           graph_port->buffer_type() == _engine.world()->uris().atom_Sequence) {
+		// Sequence port, make Jack port but use internal LV2 format buffer
+		eport = new EnginePort(graph_port);
+	}
+
+	if (eport) {
 		register_port(*eport);
-		graph_port->setup_buffers(*_engine.buffer_factory(),
-		                          graph_port->poly(),
-		                          false);
-		graph_port->set_is_driver_port(true);
-		return eport;
-	} else {
-		return NULL;
 	}
+
+	return eport;
 }
 
 void
 JackDriver::pre_process_port(ProcessContext& context, EnginePort* port)
 {
+	const URIs&       uris       = context.engine().world()->uris();
 	const SampleCount nframes    = context.nframes();
 	jack_port_t*      jack_port  = (jack_port_t*)port->handle();
-	PortImpl*         graph_port = port->graph_port();
-	void*             buffer     = jack_port_get_buffer(jack_port, nframes);
-
-	port->set_buffer(buffer);
-
-	if (!graph_port->is_input()) {
-		graph_port->buffer(0)->clear();
-		return;
-	}
-
-	if (graph_port->is_a(PortType::AUDIO)) {
-		Buffer* graph_buf = graph_port->buffer(0).get();
-		memcpy(graph_buf->samples(), buffer, nframes * sizeof(float));
-
-	} else if (graph_port->buffer_type() == graph_port->bufs().uris().atom_Sequence) {
-		Buffer* graph_buf = (Buffer*)graph_port->buffer(0).get();
-
-		const jack_nframes_t event_count = jack_midi_get_event_count(buffer);
-
+	DuplexPort*       graph_port = port->graph_port();
+	Buffer*           graph_buf  = graph_port->buffer(0).get();
+	void*             jack_buf   = jack_port_get_buffer(jack_port, nframes);
+
+	if (graph_port->is_a(PortType::AUDIO) || graph_port->is_a(PortType::CV)) {
+		graph_port->set_driver_buffer(jack_buf, nframes * sizeof(float));
+		if (graph_port->is_input()) {
+			graph_port->monitor(context);
+		} else {
+			graph_port->buffer(0)->clear(); // TODO: Avoid when possible
+		}
+	} else if (graph_port->buffer_type() == uris.atom_Sequence) {
 		graph_buf->prepare_write(context);
-
-		// Copy events from Jack port buffer into graph port buffer
-		for (jack_nframes_t i = 0; i < event_count; ++i) {
-			jack_midi_event_t ev;
-			jack_midi_event_get(&ev, buffer, i);
-
-			if (!graph_buf->append_event(
-				    ev.time, ev.size, _midi_event_type, ev.buffer)) {
-				_engine.log().warn("Failed to write to MIDI buffer, events lost!\n");
+		if (graph_port->is_input()) {
+			// Copy events from Jack port buffer into graph port buffer
+			const jack_nframes_t event_count = jack_midi_get_event_count(jack_buf);
+			for (jack_nframes_t i = 0; i < event_count; ++i) {
+				jack_midi_event_t ev;
+				jack_midi_event_get(&ev, jack_buf, i);
+				if (!graph_buf->append_event(
+					    ev.time, ev.size, _midi_event_type, ev.buffer)) {
+					_engine.log().warn("Failed to write to MIDI buffer, events lost!\n");
+				}
 			}
 		}
+		graph_port->monitor(context);
 	}
 }
 
 void
 JackDriver::post_process_port(ProcessContext& context, EnginePort* port)
 {
+	const URIs&       uris       = context.engine().world()->uris();
 	const SampleCount nframes    = context.nframes();
 	jack_port_t*      jack_port  = (jack_port_t*)port->handle();
-	PortImpl*         graph_port = port->graph_port();
-	void*             buffer     = port->buffer();
-
-	if (graph_port->is_input()) {
-		return;
-	}
-
-	if (!buffer) {
-		// First cycle for a new output, so pre_process wasn't called
-		buffer = jack_port_get_buffer(jack_port, nframes);
-		port->set_buffer(buffer);
-	}
+	DuplexPort*       graph_port = port->graph_port();
+	void*             jack_buf   = port->buffer();
+
+	if (port->graph_port()->is_output()) {
+		if (!jack_buf) {
+			// First cycle for a new output, so pre_process wasn't called
+			jack_buf = jack_port_get_buffer(jack_port, nframes);
+			port->set_buffer(jack_buf);
+		}
 
-	graph_port->post_process(context);
-	Buffer* const graph_buf = graph_port->buffer(0).get();
-	if (graph_port->is_a(PortType::AUDIO)) {
-		memcpy(buffer, graph_buf->samples(), nframes * sizeof(Sample));
-	} else if (graph_port->buffer_type() == graph_port->bufs().uris().atom_Sequence) {
-		jack_midi_clear_buffer(buffer);
-		LV2_Atom_Sequence* seq = (LV2_Atom_Sequence*)graph_buf->atom();
-		LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
-			const uint8_t* buf = (const uint8_t*)LV2_ATOM_BODY(&ev->body);
-			if (ev->body.type == graph_port->bufs().uris().midi_MidiEvent) {
-				jack_midi_event_write(buffer, ev->time.frames, buf, ev->body.size);
+		if (graph_port->buffer_type() == uris.atom_Sequence) {
+			// Copy LV2 MIDI events to Jack MIDI buffer
+			Buffer* const      graph_buf = graph_port->buffer(0).get();
+			LV2_Atom_Sequence* seq       = graph_buf->get<LV2_Atom_Sequence>();
+
+			jack_midi_clear_buffer(jack_buf);
+			LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
+				const uint8_t* buf = (const uint8_t*)LV2_ATOM_BODY(&ev->body);
+				if (ev->body.type == _midi_event_type) {
+					jack_midi_event_write(
+						jack_buf, ev->time.frames, buf, ev->body.size);
+				}
 			}
 		}
 	}
+
+	// Reset graph port buffer pointer to no longer point to the Jack buffer
+	if (graph_port->is_driver_port()) {
+		graph_port->set_driver_buffer(NULL, 0);
+	}
 }
 
 void
@@ -457,7 +468,7 @@ JackDriver::_process_cb(jack_nframes_t nframes)
 		return 0;
 	}
 
-	/* Note that Jack can not call this function for a cycle, if overloaded,
+	/* Note that Jack may not call this function for a cycle, if overloaded,
 	   so a rolling counter here would not always be correct. */
 	const jack_nframes_t start_of_current_cycle = jack_last_frame_time(_client);
 
diff --git a/src/server/PortImpl.cpp b/src/server/PortImpl.cpp
index 7119f94c..89b99b4c 100644
--- a/src/server/PortImpl.cpp
+++ b/src/server/PortImpl.cpp
@@ -168,7 +168,7 @@ PortImpl::activate(BufferFactory& bufs)
 void
 PortImpl::deactivate()
 {
-	if (is_output()) {
+	if (is_output() && !_is_driver_port) {
 		for (uint32_t v = 0; v < _poly; ++v) {
 			if (_voices->at(v).buffer) {
 				_voices->at(v).buffer->clear();
@@ -302,7 +302,7 @@ bool
 PortImpl::prepare_poly(BufferFactory& bufs, uint32_t poly)
 {
 	ThreadManager::assert_thread(THREAD_PRE_PROCESS);
-	if (_parent->path().is_root() ||
+	if (_is_driver_port || _parent->path().is_root() ||
 	    (_type == PortType::ATOM && !_value.is_valid())) {
 		return false;
 	}
@@ -385,6 +385,12 @@ PortImpl::recycle_buffers()
 }
 
 void
+PortImpl::set_is_driver_port(BufferFactory& bufs)
+{
+	_is_driver_port = true;
+}
+
+void
 PortImpl::clear_buffers()
 {
 	switch (_type.id()) {
@@ -442,13 +448,13 @@ PortImpl::monitor(Context& context, bool send_now)
 		break;
 	case PortType::ATOM:
 		if (_buffer_type == _bufs.uris().atom_Sequence) {
-			const LV2_Atom* atom = buffer(0)->atom();
-			if (buffer(0)->value() && !_monitored) {
-				// Value sequence not fully monitored, monitor as control
-				key = uris.ingen_value;
-				val = ((LV2_Atom_Float*)buffer(0)->value())->body;
-			} else if (_monitored && !buffer(0)->empty()) {
-				// Monitoring explictly enabled, send everything
+			const LV2_Atom* atom  = buffer(0)->get<const LV2_Atom>();
+			const LV2_Atom* value = buffer(0)->value();
+			if (atom->type != _bufs.uris().atom_Sequence) {
+				/* Buffer contents are not actually a Sequence.  Probably an
+				   uninitialized Chunk, so do nothing. */
+			} else if (_monitored) {
+				/* Sequence explicitly monitored, send everything. */
 				const LV2_Atom_Sequence* seq = (const LV2_Atom_Sequence*)atom;
 				LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
 					context.notify(uris.ingen_activity,
@@ -458,8 +464,12 @@ PortImpl::monitor(Context& context, bool send_now)
 					               ev->body.type,
 					               LV2_ATOM_BODY(&ev->body));
 				}
-			} else if (!buffer(0)->empty()) {
-				// Just send activity for blinkenlights
+			} else if (value && value->type == _bufs.uris().atom_Float) {
+				/* Float sequence, monitor as a control. */
+				key = uris.ingen_value;
+				val = ((LV2_Atom_Float*)buffer(0)->value())->body;
+			} else if (atom->size > sizeof(LV2_Atom_Sequence_Body)) {
+				/* General sequence, send activity for blinkenlights. */
 				const int32_t one = 1;
 				context.notify(uris.ingen_activity,
 				               context.start(),
diff --git a/src/server/PortImpl.hpp b/src/server/PortImpl.hpp
index 3eb33e6d..d16ec10f 100644
--- a/src/server/PortImpl.hpp
+++ b/src/server/PortImpl.hpp
@@ -140,8 +140,17 @@ public:
 	                       FrameTime      time,
 	                       Sample         value);
 
-	void set_is_driver_port(bool b) { _is_driver_port = b; }
-	bool is_driver_port() const     { return _is_driver_port; }
+	/** Prepare this port to use an external driver-provided buffer.
+	 *
+	 * This will avoid allocating a buffer for the port, instead the driver
+	 * buffer is used directly.  This only makes sense for ports on the
+	 * top-level graph, which are monophonic.  Non-real-time, must be called
+	 * before using the port, followed by a call to set_driver_buffer() in the
+	 * processing thread.
+	 */
+	virtual void set_is_driver_port(BufferFactory& bufs);
+
+	bool is_driver_port() const { return _is_driver_port; }
 
 	/** Called once per process cycle */
 	virtual void pre_process(Context& context) = 0;
diff --git a/src/server/events/Connect.cpp b/src/server/events/Connect.cpp
index 1f22e6f2..5eff8854 100644
--- a/src/server/events/Connect.cpp
+++ b/src/server/events/Connect.cpp
@@ -123,11 +123,13 @@ Connect::pre_process()
 
 	lock.unlock();
 
-	_voices = new Raul::Array<PortImpl::Voice>(_head->poly());
-	_head->get_buffers(*_engine.buffer_factory(),
-	                   _voices,
-	                   _head->poly(),
-	                   false);
+	if (!_head->is_driver_port()) {
+		_voices = new Raul::Array<PortImpl::Voice>(_head->poly());
+		_head->get_buffers(*_engine.buffer_factory(),
+		                   _voices,
+		                   _head->poly(),
+		                   false);
+	}
 
 	if (_graph->enabled()) {
 		_compiled_graph = _graph->compile();
@@ -141,7 +143,9 @@ Connect::execute(ProcessContext& context)
 {
 	if (_status == Status::SUCCESS) {
 		_head->add_arc(context, _arc.get());
-		_engine.maid()->dispose(_head->set_voices(context, _voices));
+		if (!_head->is_driver_port()) {
+			_engine.maid()->dispose(_head->set_voices(context, _voices));
+		}
 		_head->connect_buffers();
 		_graph->set_compiled_graph(_compiled_graph);
 	}
diff --git a/src/server/events/CreatePort.cpp b/src/server/events/CreatePort.cpp
index 173b8a73..0b58ee89 100644
--- a/src/server/events/CreatePort.cpp
+++ b/src/server/events/CreatePort.cpp
@@ -125,12 +125,19 @@ CreatePort::pre_process()
 	                         poly_i->second.type() == uris.forge.Bool &&
 	                         poly_i->second.get<int32_t>());
 
-	if (!(_graph_port = _graph->create_port(
-		      bufs, Raul::Symbol(_path.symbol()),
-		      _port_type, _buf_type, buf_size, _is_output, polyphonic))) {
-		return Event::pre_process_done(Status::CREATION_FAILED, _path);
+	// Create 0 value if the port requires one
+	Atom value;
+	if (_port_type == PortType::CONTROL || _port_type == PortType::CV) {
+		value = bufs.forge().make(0.0f);
 	}
 
+	// Create port
+	_graph_port = new DuplexPort(bufs, _graph, Raul::Symbol(_path.symbol()),
+	                             _graph->num_ports_non_rt(),
+	                             polyphonic,
+	                             _port_type, _buf_type, buf_size,
+	                             value, _is_output);
+
 	_graph_port->properties().insert(_properties.begin(), _properties.end());
 
 	_engine.store()->add(_graph_port);
@@ -141,8 +148,7 @@ CreatePort::pre_process()
 	}
 
 	if (!_graph->parent()) {
-		_engine_port = _engine.driver()->create_port(
-			dynamic_cast<DuplexPort*>(_graph_port));
+		_engine_port = _engine.driver()->create_port(_graph_port);
 	}
 
 	_ports_array = new Raul::Array<PortImpl*>(old_n_ports + 1, NULL);
diff --git a/src/server/events/Disconnect.cpp b/src/server/events/Disconnect.cpp
index 87c192b4..66aee38e 100644
--- a/src/server/events/Disconnect.cpp
+++ b/src/server/events/Disconnect.cpp
@@ -91,21 +91,25 @@ Disconnect::Impl::Impl(Engine&     e,
 	_dst_input_port->decrement_num_arcs();
 
 	if (_dst_input_port->num_arcs() == 0) {
-		_voices = new Raul::Array<PortImpl::Voice>(_dst_input_port->poly());
-		_dst_input_port->get_buffers(*_engine.buffer_factory(),
-		                             _voices,
-		                             _dst_input_port->poly(),
-		                             false);
-
-		const bool is_control = _dst_input_port->is_a(PortType::CONTROL) ||
-			_dst_input_port->is_a(PortType::CV);
-		const float value = is_control ? _dst_input_port->value().get<float>() : 0;
-		for (uint32_t i = 0; i < _voices->size(); ++i) {
-			if (is_control) {
-				Buffer* buf = _voices->at(i).buffer.get();
-				buf->set_block(value, 0, buf->nframes());
+		if (!_dst_input_port->is_driver_port()) {
+			_voices = new Raul::Array<PortImpl::Voice>(_dst_input_port->poly());
+			_dst_input_port->get_buffers(*_engine.buffer_factory(),
+			                             _voices,
+			                             _dst_input_port->poly(),
+			                             false);
+
+			if (_dst_input_port->is_a(PortType::CONTROL) ||
+			    _dst_input_port->is_a(PortType::CV)) {
+				// Reset buffer to control value
+				const float value = _dst_input_port->value().get<float>();
+				for (uint32_t i = 0; i < _voices->size(); ++i) {
+					Buffer* buf = _voices->at(i).buffer.get();
+					buf->set_block(value, 0, buf->nframes());
+				}
 			} else {
-				_voices->at(i).buffer->clear();
+				for (uint32_t i = 0; i < _voices->size(); ++i) {
+					_voices->at(i).buffer->clear();
+				}
 			}
 		}
 	}
@@ -177,8 +181,11 @@ Disconnect::Impl::execute(ProcessContext& context, bool set_dst_buffers)
 {
 	ArcImpl* const port_arc =
 		_dst_input_port->remove_arc(context, _src_output_port);
+
 	if (!port_arc) {
 		return false;
+	} else if (_dst_input_port->is_driver_port()) {
+		return true;
 	}
 
 	if (set_dst_buffers) {
diff --git a/src/server/events/SetPortValue.cpp b/src/server/events/SetPortValue.cpp
index e069e474..fd10c94f 100644
--- a/src/server/events/SetPortValue.cpp
+++ b/src/server/events/SetPortValue.cpp
@@ -107,7 +107,7 @@ SetPortValue::apply(Context& context)
 			_status = Status::NO_SPACE;
 		}
 	} else if (buf->type() == uris.atom_URID) {
-		((LV2_Atom_URID*)buf->atom())->body = _value.get<int32_t>();
+		buf->get<LV2_Atom_URID>()->body = _value.get<int32_t>();
 	} else {
 		_status = Status::BAD_VALUE_TYPE;
 	}
diff --git a/src/server/ingen_lv2.cpp b/src/server/ingen_lv2.cpp
index 8265cce8..d74a829b 100644
--- a/src/server/ingen_lv2.cpp
+++ b/src/server/ingen_lv2.cpp
@@ -118,67 +118,46 @@ public:
 	{}
 
 	void pre_process_port(ProcessContext& context, EnginePort* port) {
-		PortImpl* graph_port = port->graph_port();
-		void*     buffer     = port->buffer();
-
-		if (!graph_port->is_input() || !buffer) {
-			return;
-		}
-
-		Buffer* const graph_buf = graph_port->buffer(0).get();
-		if (graph_port->is_a(PortType::AUDIO) ||
-		    graph_port->is_a(PortType::CV)) {
-			memcpy(graph_buf->samples(),
-			       buffer,
-			       context.nframes() * sizeof(float));
-		} else if (graph_port->is_a(PortType::CONTROL)) {
-			graph_buf->samples()[0] = ((float*)buffer)[0];
-		} else {
-			LV2_Atom_Sequence* seq      = (LV2_Atom_Sequence*)buffer;
-			bool               enqueued = false;
-			URIs&              uris     = graph_port->bufs().uris();
-			graph_buf->prepare_write(context);
-			LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
-				if (!graph_buf->append_event(
-					    ev->time.frames, ev->body.size, ev->body.type,
-					    (const uint8_t*)LV2_ATOM_BODY(&ev->body))) {
-					_engine.log().warn("Failed to write to buffer, event lost!\n");
+		const URIs&       uris       = _engine.world()->uris();
+		const SampleCount nframes    = context.nframes();
+		DuplexPort*       graph_port = port->graph_port();
+		Buffer*           graph_buf  = graph_port->buffer(0).get();
+		void*             lv2_buf    = port->buffer();
+
+		if (graph_port->is_a(PortType::AUDIO) || graph_port->is_a(PortType::CV)) {
+			graph_port->set_driver_buffer(lv2_buf, nframes * sizeof(float));
+		} else if (graph_port->buffer_type() == uris.atom_Sequence) {
+			graph_port->set_driver_buffer(lv2_buf, lv2_atom_total_size((LV2_Atom*)lv2_buf));
+			if (graph_port->symbol() == "control_in") {  // TODO: Safe to use index?
+				LV2_Atom_Sequence* seq      = (LV2_Atom_Sequence*)lv2_buf;
+				bool               enqueued = false;
+				LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
+					if (AtomReader::is_message(uris, &ev->body)) {
+						enqueued = enqueue_message(&ev->body) || enqueued;
+					}
 				}
 
-				if (AtomReader::is_message(uris, &ev->body)) {
-					enqueue_message(&ev->body);
-					enqueued = true;
+				if (enqueued) {
+					// Enqueued a message for processing, raise semaphore
+					_main_sem.post();
 				}
 			}
+		}
 
-			if (enqueued) {
-				_main_sem.post();
-			}
+		if (graph_port->is_input()) {
+			graph_port->monitor(context);
+		} else {
+			graph_buf->prepare_write(context);
 		}
 	}
 
 	void post_process_port(ProcessContext& context, EnginePort* port) {
-		PortImpl* graph_port = port->graph_port();
-		void*     buffer     = port->buffer();
+		DuplexPort* graph_port = port->graph_port();
 
-		if (graph_port->is_input() || !buffer) {
-			return;
-		}
-
-		Buffer* graph_buf = graph_port->buffer(0).get();
-		if (graph_port->is_a(PortType::AUDIO) ||
-		    graph_port->is_a(PortType::CV)) {
-			memcpy(buffer,
-			       graph_buf->samples(),
-			       context.nframes() * sizeof(float));
-		} else if (graph_port->is_a(PortType::CONTROL)) {
-			((float*)buffer)[0] = graph_buf->samples()[0];
-		} else if (graph_port->index() != 1) {
-			/* Copy Sequence output to LV2 buffer, except notify output which
-			   is written by flush_to_ui() (TODO: merge) */
-			memcpy(buffer,
-			       graph_buf->atom(),
-			       sizeof(LV2_Atom) + graph_buf->atom()->size);
+		// No copying necessary, host buffers are used directly
+		// Reset graph port buffer pointer to no longer point to the Jack buffer
+		if (graph_port->is_driver_port()) {
+			graph_port->set_driver_buffer(NULL, 0);
 		}
 	}
 
@@ -258,6 +237,7 @@ public:
 	                           const Atom&       value) {}
 
 	virtual EnginePort* create_port(DuplexPort* graph_port) {
+		graph_port->set_is_driver_port(*_engine.buffer_factory());
 		return new EnginePort(graph_port);
 	}
 
@@ -278,12 +258,14 @@ public:
 	}
 
 	/** Called in run thread for events received at control input port. */
-	void enqueue_message(const LV2_Atom* atom) {
+	bool enqueue_message(const LV2_Atom* atom) {
 		if (_from_ui.write(lv2_atom_total_size(atom), atom) == 0) {
 #ifndef NDEBUG
 			_engine.log().error("Control input buffer overflow\n");
 #endif
+			return false;
 		}
+		return true;
 	}
 
 	Raul::Semaphore& main_sem() { return _main_sem; }
@@ -585,6 +567,7 @@ ingen_instantiate(const LV2_Descriptor*    descriptor,
 
 	std::lock_guard<std::mutex> lock(plugin->world->rdf_mutex());
 
+	fprintf(stderr, "LV2 parse resource %s from %s\n", graph->uri.c_str(), graph->filename.c_str());
 	plugin->world->parser()->parse_file(plugin->world,
 	                                    plugin->world->interface().get(),
 	                                    graph->filename);
diff --git a/src/server/internals/Controller.cpp b/src/server/internals/Controller.cpp
index d52dc59f..8913d293 100644
--- a/src/server/internals/Controller.cpp
+++ b/src/server/internals/Controller.cpp
@@ -47,7 +47,7 @@ ControllerNode::ControllerNode(InternalPlugin*     plugin,
                                bool                polyphonic,
                                GraphImpl*          parent,
                                SampleRate          srate)
-	: BlockImpl(plugin, symbol, false, parent, srate)
+	: InternalBlock(plugin, symbol, false, parent, srate)
 	, _learning(false)
 {
 	const Ingen::URIs& uris = bufs.uris();
@@ -103,7 +103,7 @@ void
 ControllerNode::run(ProcessContext& context)
 {
 	Buffer* const      midi_in = _midi_in_port->buffer(0).get();
-	LV2_Atom_Sequence* seq     = (LV2_Atom_Sequence*)midi_in->atom();
+	LV2_Atom_Sequence* seq     = midi_in->get<LV2_Atom_Sequence>();
 	LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
 		const uint8_t* buf = (const uint8_t*)LV2_ATOM_BODY(&ev->body);
 		if (ev->body.type == _midi_in_port->bufs().uris().midi_MidiEvent &&
diff --git a/src/server/internals/Controller.hpp b/src/server/internals/Controller.hpp
index 8f5bd766..4eb1f508 100644
--- a/src/server/internals/Controller.hpp
+++ b/src/server/internals/Controller.hpp
@@ -17,7 +17,7 @@
 #ifndef INGEN_INTERNALS_CONTROLLER_HPP
 #define INGEN_INTERNALS_CONTROLLER_HPP
 
-#include "BlockImpl.hpp"
+#include "InternalBlock.hpp"
 
 namespace Ingen {
 namespace Server {
@@ -35,7 +35,7 @@ namespace Internals {
  *
  * \ingroup engine
  */
-class ControllerNode : public BlockImpl
+class ControllerNode : public InternalBlock
 {
 public:
 	ControllerNode(InternalPlugin*     plugin,
diff --git a/src/server/internals/Delay.cpp b/src/server/internals/Delay.cpp
index 16b20a3f..d0413689 100644
--- a/src/server/internals/Delay.cpp
+++ b/src/server/internals/Delay.cpp
@@ -55,7 +55,7 @@ DelayNode::DelayNode(InternalPlugin*     plugin,
                      bool                polyphonic,
                      GraphImpl*          parent,
                      SampleRate          srate)
-	: BlockImpl(plugin, symbol, polyphonic, parent, srate)
+	: InternalBlock(plugin, symbol, polyphonic, parent, srate)
 	, _buffer(0)
 	, _buffer_length(0)
 	, _buffer_mask(0)
diff --git a/src/server/internals/Delay.hpp b/src/server/internals/Delay.hpp
index 51e8e276..24f2854e 100644
--- a/src/server/internals/Delay.hpp
+++ b/src/server/internals/Delay.hpp
@@ -19,7 +19,7 @@
 
 #include <math.h>
 
-#include "BlockImpl.hpp"
+#include "InternalBlock.hpp"
 #include "types.hpp"
 
 namespace Ingen {
@@ -32,7 +32,7 @@ class BufferFactory;
 
 namespace Internals {
 
-class DelayNode : public BlockImpl
+class DelayNode : public InternalBlock
 {
 public:
 	DelayNode(InternalPlugin*     plugin,
diff --git a/src/server/internals/Note.cpp b/src/server/internals/Note.cpp
index feb3890f..c3ebaf71 100644
--- a/src/server/internals/Note.cpp
+++ b/src/server/internals/Note.cpp
@@ -52,7 +52,7 @@ NoteNode::NoteNode(InternalPlugin*     plugin,
                    bool                polyphonic,
                    GraphImpl*          parent,
                    SampleRate          srate)
-	: BlockImpl(plugin, symbol, polyphonic, parent, srate)
+	: InternalBlock(plugin, symbol, polyphonic, parent, srate)
 	, _voices(new Raul::Array<Voice>(_polyphony))
 	, _prepared_voices(NULL)
 	, _sustain(false)
@@ -171,7 +171,7 @@ void
 NoteNode::run(ProcessContext& context)
 {
 	Buffer* const      midi_in = _midi_in_port->buffer(0).get();
-	LV2_Atom_Sequence* seq     = (LV2_Atom_Sequence*)midi_in->atom();
+	LV2_Atom_Sequence* seq     = midi_in->get<LV2_Atom_Sequence>();
 	LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
 		const uint8_t*  buf  = (const uint8_t*)LV2_ATOM_BODY_CONST(&ev->body);
 		const FrameTime time = context.start() + (FrameTime)ev->time.frames;
diff --git a/src/server/internals/Note.hpp b/src/server/internals/Note.hpp
index 111d3f1a..6e993f06 100644
--- a/src/server/internals/Note.hpp
+++ b/src/server/internals/Note.hpp
@@ -17,7 +17,7 @@
 #ifndef INGEN_INTERNALS_NOTE_HPP
 #define INGEN_INTERNALS_NOTE_HPP
 
-#include "BlockImpl.hpp"
+#include "InternalBlock.hpp"
 #include "types.hpp"
 
 namespace Ingen {
@@ -35,7 +35,7 @@ namespace Internals {
  *
  * \ingroup engine
  */
-class NoteNode : public BlockImpl
+class NoteNode : public InternalBlock
 {
 public:
 	NoteNode(InternalPlugin*     plugin,
diff --git a/src/server/internals/Time.cpp b/src/server/internals/Time.cpp
index b03f427c..13f63538 100644
--- a/src/server/internals/Time.cpp
+++ b/src/server/internals/Time.cpp
@@ -42,7 +42,7 @@ TimeNode::TimeNode(InternalPlugin*     plugin,
                    bool                polyphonic,
                    GraphImpl*          parent,
                    SampleRate          srate)
-	: BlockImpl(plugin, symbol, false, parent, srate)
+	: InternalBlock(plugin, symbol, false, parent, srate)
 {
 	const Ingen::URIs& uris = bufs.uris();
 	_ports = new Raul::Array<PortImpl*>(1);
@@ -60,7 +60,7 @@ void
 TimeNode::run(ProcessContext& context)
 {
 	BufferRef          buf = _notify_port->buffer(0);
-	LV2_Atom_Sequence* seq = (LV2_Atom_Sequence*)buf->atom();
+	LV2_Atom_Sequence* seq = buf->get<LV2_Atom_Sequence>();
 
 	// Initialise output to the empty sequence
 	seq->atom.type = _notify_port->bufs().uris().atom_Sequence;
diff --git a/src/server/internals/Time.hpp b/src/server/internals/Time.hpp
index bd98827c..f8dfbd64 100644
--- a/src/server/internals/Time.hpp
+++ b/src/server/internals/Time.hpp
@@ -17,7 +17,7 @@
 #ifndef INGEN_INTERNALS_TIME_HPP
 #define INGEN_INTERNALS_TIME_HPP
 
-#include "BlockImpl.hpp"
+#include "InternalBlock.hpp"
 
 namespace Ingen {
 namespace Server {
@@ -34,7 +34,7 @@ namespace Internals {
  *
  * \ingroup engine
  */
-class TimeNode : public BlockImpl
+class TimeNode : public InternalBlock
 {
 public:
 	TimeNode(InternalPlugin*     plugin,
diff --git a/src/server/internals/Trigger.cpp b/src/server/internals/Trigger.cpp
index 1fa57ba9..d34b38c1 100644
--- a/src/server/internals/Trigger.cpp
+++ b/src/server/internals/Trigger.cpp
@@ -47,7 +47,7 @@ TriggerNode::TriggerNode(InternalPlugin*     plugin,
                          bool                polyphonic,
                          GraphImpl*          parent,
                          SampleRate          srate)
-	: BlockImpl(plugin, symbol, false, parent, srate)
+	: InternalBlock(plugin, symbol, false, parent, srate)
 	, _learning(false)
 {
 	const Ingen::URIs& uris = bufs.uris();
@@ -99,7 +99,7 @@ void
 TriggerNode::run(ProcessContext& context)
 {
 	Buffer* const      midi_in = _midi_in_port->buffer(0).get();
-	LV2_Atom_Sequence* seq     = (LV2_Atom_Sequence*)midi_in->atom();
+	LV2_Atom_Sequence* seq     = midi_in->get<LV2_Atom_Sequence>();
 	LV2_ATOM_SEQUENCE_FOREACH(seq, ev) {
 		const uint8_t* buf = (const uint8_t*)LV2_ATOM_BODY(&ev->body);
 		if (ev->body.type == _midi_in_port->bufs().uris().midi_MidiEvent &&
diff --git a/src/server/internals/Trigger.hpp b/src/server/internals/Trigger.hpp
index f3fae245..f486ef38 100644
--- a/src/server/internals/Trigger.hpp
+++ b/src/server/internals/Trigger.hpp
@@ -17,7 +17,7 @@
 #ifndef INGEN_INTERNALS_TRIGGER_HPP
 #define INGEN_INTERNALS_TRIGGER_HPP
 
-#include "BlockImpl.hpp"
+#include "InternalBlock.hpp"
 
 namespace Ingen {
 namespace Server {
@@ -38,7 +38,7 @@ namespace Internals {
  *
  * \ingroup engine
  */
-class TriggerNode : public BlockImpl
+class TriggerNode : public InternalBlock
 {
 public:
 	TriggerNode(InternalPlugin*     plugin,
diff --git a/src/server/mix.cpp b/src/server/mix.cpp
index d7bf2229..6fa8626c 100644
--- a/src/server/mix.cpp
+++ b/src/server/mix.cpp
@@ -26,9 +26,10 @@ namespace Server {
 static inline bool
 is_end(const Buffer* buf, const LV2_Atom_Event* ev)
 {
+	const LV2_Atom* atom = buf->get<const LV2_Atom>();
 	return lv2_atom_sequence_is_end(
-		(const LV2_Atom_Sequence_Body*)LV2_ATOM_BODY_CONST(buf->atom()),
-		buf->atom()->size,
+		(const LV2_Atom_Sequence_Body*)LV2_ATOM_BODY_CONST(atom),
+		atom->size,
 		ev);
 }
 
@@ -72,9 +73,8 @@ mix(const Context&      context,
 		for (uint32_t i = 0; i < num_srcs; ++i) {
 			iters[i] = NULL;
 			if (srcs[i]->is_sequence()) {
-				iters[i] = lv2_atom_sequence_begin(
-					(const LV2_Atom_Sequence_Body*)LV2_ATOM_BODY_CONST(
-						srcs[i]->atom()));
+				const LV2_Atom_Sequence* seq = srcs[i]->get<const LV2_Atom_Sequence>();
+				iters[i] = lv2_atom_sequence_begin(&seq->body);
 				if (is_end(srcs[i], iters[i])) {
 					iters[i] = NULL;
 				}
diff --git a/src/server/wscript b/src/server/wscript
index c3e963fa..7dde5361 100644
--- a/src/server/wscript
+++ b/src/server/wscript
@@ -16,6 +16,7 @@ def build(bld):
             EventWriter.cpp
             GraphImpl.cpp
             InputPort.cpp
+            InternalBlock.cpp
             InternalPlugin.cpp
             LV2Block.cpp
             LV2Plugin.cpp