Fix real-time issues with buffer allocation

16 files changed, 207 insertions, 125 deletions

diff --git a/src/server/Buffer.cpp b/src/server/Buffer.cpp
index c5f9a1c2..71dd320a 100644
--- a/src/server/Buffer.cpp
+++ b/src/server/Buffer.cpp
@@ -91,7 +91,7 @@ Buffer::Buffer(BufferFactory& bufs,
 			   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);
+			_value_buffer = bufs.get_buffer(value_type, 0, 0);
 		}
 	}
 }
@@ -110,12 +110,12 @@ Buffer::recycle()
 }
 
 void
-Buffer::set_type(LV2_URID type, LV2_URID value_type)
+Buffer::set_type(GetFn get, LV2_URID type, LV2_URID value_type)
 {
 	_type       = type;
 	_value_type = value_type;
 	if (type == _factory.uris().atom_Sequence && value_type) {
-		_value_buffer = _factory.get_buffer(value_type, 0, 0, false);
+		_value_buffer = (_factory.*get)(value_type, 0, 0);
 	}
 }
 
@@ -312,8 +312,7 @@ Buffer::append_event(int64_t        frames,
 
 	LV2_Atom* atom = get<LV2_Atom>();
 	if (atom->type == _factory.uris().atom_Chunk) {
-		// Chunk initialized with prepare_output_write(), clear
-		clear();
+		clear();  // Chunk initialized with prepare_output_write(), clear
 	}
 
 	if (sizeof(LV2_Atom) + atom->size + lv2_atom_pad_size(size) > _capacity) {
@@ -384,7 +383,7 @@ Buffer::set_value(const Atom& value)
 	}
 
 	if (value.size() > _value_buffer->size()) {
-		_value_buffer = _factory.get_buffer(value.type(), 0, value.size(), false, false);
+		_value_buffer = _factory.claim_buffer(value.type(), 0, value.size());
 	}
 
 	memcpy(_value_buffer->get<LV2_Atom*>(), value.atom(), sizeof(LV2_Atom) + value.size());
diff --git a/src/server/Buffer.hpp b/src/server/Buffer.hpp
index d60365f2..6404479b 100644
--- a/src/server/Buffer.hpp
+++ b/src/server/Buffer.hpp
@@ -64,7 +64,12 @@ public:
 		return is_audio() ? _capacity : sizeof(LV2_Atom) + get<LV2_Atom>()->size;
 	}
 
-	void set_type(LV2_URID type, LV2_URID value_type);
+	typedef BufferRef (BufferFactory::*GetFn)(LV2_URID, LV2_URID, uint32_t);
+
+	/** Set the buffer type and optional value type for this buffer.
+	 * @param get Called to get auxiliary buffers if necessary.
+	 */
+	void set_type(GetFn get, LV2_URID type, LV2_URID value_type);
 
 	inline bool is_audio() const {
 		return _type == _factory.uris().atom_Sound;
diff --git a/src/server/BufferFactory.cpp b/src/server/BufferFactory.cpp
index 749e83d0..2fa98b67 100644
--- a/src/server/BufferFactory.cpp
+++ b/src/server/BufferFactory.cpp
@@ -100,37 +100,51 @@ BufferFactory::default_size(LV2_URID type) const
 	}
 }
 
+Buffer*
+BufferFactory::try_get_buffer(LV2_URID type)
+{
+	std::atomic<Buffer*>& head_ptr = free_list(type);
+	Buffer*               head     = NULL;
+	Buffer*               next;
+	do {
+		head = head_ptr.load();
+		if (!head) {
+			break;
+		}
+		next = head->_next;
+	} while (!head_ptr.compare_exchange_weak(head, next));
+
+	return head;
+}
+
 BufferRef
 BufferFactory::get_buffer(LV2_URID type,
                           LV2_URID value_type,
-                          uint32_t capacity,
-                          bool     real_time,
-                          bool     force_create)
+                          uint32_t capacity)
 {
-	std::atomic<Buffer*>& head_ptr = free_list(type);
-	Buffer*               try_head = NULL;
-
-	if (!force_create) {
-		Buffer* next;
-		do {
-			try_head = head_ptr.load();
-			if (!try_head)
-				break;
-			next = try_head->_next;
-		} while (!head_ptr.compare_exchange_weak(try_head, next));
+	Buffer* try_head = try_get_buffer(type);
+	if (!try_head) {
+		return create(type, value_type, capacity);
 	}
 
+	try_head->_next = NULL;
+	try_head->set_type(&BufferFactory::get_buffer, type, value_type);
+	return BufferRef(try_head);
+}
+
+BufferRef
+BufferFactory::claim_buffer(LV2_URID type,
+                            LV2_URID value_type,
+                            uint32_t capacity)
+{
+	Buffer* try_head = try_get_buffer(type);
 	if (!try_head) {
-		if (!real_time) {
-			return create(type, value_type, capacity);
-		} else {
-			_engine.world()->log().error("Failed to obtain buffer");
-			return BufferRef();
-		}
+		_engine.world()->log().rt_error("Failed to obtain buffer");
+		return BufferRef();
 	}
 
 	try_head->_next = NULL;
-	try_head->set_type(type, value_type);
+	try_head->set_type(&BufferFactory::claim_buffer, type, value_type);
 	return BufferRef(try_head);
 }
 
diff --git a/src/server/BufferFactory.hpp b/src/server/BufferFactory.hpp
index ccb01899..09722b4e 100644
--- a/src/server/BufferFactory.hpp
+++ b/src/server/BufferFactory.hpp
@@ -50,12 +50,22 @@ public:
 	uint32_t audio_buffer_size() const;
 	uint32_t default_size(LV2_URID type) const;
 
+	/** Dynamically allocate a new Buffer. */
+	BufferRef create(LV2_URID type,
+	                 LV2_URID value_type,
+	                 uint32_t capacity = 0);
+
+	/** Get a new buffer, reusing if possible, allocating if otherwise. */
 	BufferRef get_buffer(LV2_URID type,
 	                     LV2_URID value_type,
-	                     uint32_t capacity,
-	                     bool     real_time,
-	                     bool     force_create = false);
+	                     uint32_t capacity);
+
+	/** Claim an existing buffer, never allocates, real-time safe. */
+	BufferRef claim_buffer(LV2_URID type,
+	                       LV2_URID value_type,
+	                       uint32_t capacity);
 
+	/** Return a reference to a shared silent buffer. */
 	BufferRef silent_buffer();
 
 	void set_block_length(SampleCount block_length);
@@ -69,7 +79,7 @@ private:
 	friend class Buffer;
 	void recycle(Buffer* buf);
 
-	BufferRef create(LV2_URID type, LV2_URID value_type, uint32_t capacity=0);
+	Buffer* try_get_buffer(LV2_URID type);
 
 	inline std::atomic<Buffer*>& free_list(LV2_URID type) {
 		if (type == _uris.atom_Float) {
diff --git a/src/server/ControlBindings.cpp b/src/server/ControlBindings.cpp
index 0063c682..fc3ac7de 100644
--- a/src/server/ControlBindings.cpp
+++ b/src/server/ControlBindings.cpp
@@ -1,6 +1,6 @@
 /*
   This file is part of Ingen.
-  Copyright 2007-2015 David Robillard <http://drobilla.net/>
+  Copyright 2007-2016 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
@@ -399,8 +399,8 @@ ControlBindings::remove(PortImpl* port)
 void
 ControlBindings::pre_process(RunContext& context, Buffer* buffer)
 {
-	uint16_t       value    = 0;
-	SPtr<Bindings> bindings = _bindings;
+	uint16_t  value    = 0;
+	Bindings* bindings = _bindings.get();
 	_feedback->clear();
 
 	Ingen::World*      world = context.engine().world();
diff --git a/src/server/DuplexPort.cpp b/src/server/DuplexPort.cpp
index 5eff4add..41312e65 100644
--- a/src/server/DuplexPort.cpp
+++ b/src/server/DuplexPort.cpp
@@ -62,7 +62,8 @@ DuplexPort::DuplexPort(BufferFactory&      bufs,
 		}
 	}
 
-	get_buffers(bufs, _voices, parent->polyphony(), false);
+	get_buffers(bufs, &BufferFactory::get_buffer,
+	            _voices, parent->polyphony(), 0);
 }
 
 DuplexPort::~DuplexPort()
@@ -130,16 +131,26 @@ DuplexPort::on_property(const Raul::URI& uri, const Atom& value)
 
 bool
 DuplexPort::get_buffers(BufferFactory&      bufs,
+                        PortImpl::GetFn     get,
                         Raul::Array<Voice>* voices,
                         uint32_t            poly,
-                        bool                real_time) const
+                        size_t              num_in_arcs) const
 {
-	if (!_is_driver_port) {
-		if (_is_output) {
-			return InputPort::get_buffers(bufs, voices, poly, real_time);
-		} else {
-			return PortImpl::get_buffers(bufs, voices, poly, real_time);
-		}
+	if (!_is_driver_port && is_output()) {
+		return InputPort::get_buffers(bufs, get, voices, poly, num_in_arcs);
+	} else if (!_is_driver_port && is_input()) {
+		return PortImpl::get_buffers(bufs, get, voices, poly, num_in_arcs);
+	}
+	return false;
+}
+
+bool
+DuplexPort::setup_buffers(RunContext& ctx, BufferFactory& bufs, uint32_t poly)
+{
+	if (!_is_driver_port && is_output()) {
+		return InputPort::setup_buffers(ctx, bufs, poly);
+	} else if (!_is_driver_port && is_input()) {
+		return PortImpl::setup_buffers(ctx, bufs, poly);
 	}
 	return false;
 }
diff --git a/src/server/DuplexPort.hpp b/src/server/DuplexPort.hpp
index adf32a88..a51c5f55 100644
--- a/src/server/DuplexPort.hpp
+++ b/src/server/DuplexPort.hpp
@@ -70,10 +70,10 @@ public:
 	bool apply_poly(RunContext& context, Raul::Maid& maid, uint32_t poly);
 
 	bool get_buffers(BufferFactory&      bufs,
+	                 PortImpl::GetFn     get,
 	                 Raul::Array<Voice>* voices,
 	                 uint32_t            poly,
-	                 bool                real_time) const;
-
+	                 size_t              num_in_arcs) const;
 
 	virtual void set_is_driver_port(BufferFactory& bufs);
 
@@ -84,6 +84,8 @@ public:
 	 */
 	void set_driver_buffer(void* buf, uint32_t capacity);
 
+	bool setup_buffers(RunContext& ctx, BufferFactory& bufs, uint32_t poly) override;
+
 	void pre_process(RunContext& context);
 	void post_process(RunContext& context);
 
diff --git a/src/server/GraphImpl.cpp b/src/server/GraphImpl.cpp
index c1ee6a2b..dd7be036 100644
--- a/src/server/GraphImpl.cpp
+++ b/src/server/GraphImpl.cpp
@@ -191,15 +191,17 @@ GraphImpl::apply_internal_poly(RunContext&    context,
 	for (auto& b : _blocks) {
 		for (uint32_t j = 0; j < b.num_ports(); ++j) {
 			PortImpl* const port = b.port_impl(j);
-			if (port->is_input() && dynamic_cast<InputPort*>(port)->direct_connect())
-				port->setup_buffers(bufs, port->poly(), true);
+			if (port->is_input() && dynamic_cast<InputPort*>(port)->direct_connect()) {
+				port->setup_buffers(context, bufs, port->poly());
+			}
 			port->connect_buffers();
 		}
 	}
 
 	const bool polyphonic = parent_graph() && (poly == parent_graph()->internal_poly_process());
-	for (auto& o : _outputs)
-		o.setup_buffers(bufs, polyphonic ? poly : 1, true);
+	for (auto& o : _outputs) {
+		o.setup_buffers(context, bufs, polyphonic ? poly : 1);
+	}
 
 	_poly_process = poly;
 	return true;
diff --git a/src/server/InputPort.cpp b/src/server/InputPort.cpp
index fc2abd56..1b4c4501 100644
--- a/src/server/InputPort.cpp
+++ b/src/server/InputPort.cpp
@@ -68,40 +68,28 @@ InputPort::apply_poly(RunContext& context, Raul::Maid& maid, uint32_t poly)
 
 bool
 InputPort::get_buffers(BufferFactory&      bufs,
+                       PortImpl::GetFn     get,
                        Raul::Array<Voice>* voices,
                        uint32_t            poly,
-                       bool                real_time) const
+                       size_t              num_in_arcs) const
 {
-	const size_t num_arcs = real_time ? _arcs.size() : _num_arcs;
-
 	if (is_a(PortType::ATOM) && !_value.is_valid()) {
 		poly = 1;
 	}
 
-	if (is_a(PortType::AUDIO) && num_arcs == 0) {
+	if (is_a(PortType::AUDIO) && num_in_arcs == 0) {
 		// Audio input with no arcs, use shared zero buffer
 		for (uint32_t v = 0; v < poly; ++v) {
 			voices->at(v).buffer = bufs.silent_buffer();
 		}
 		return false;
-
-	} else if (num_arcs == 1 && !is_a(PortType::ATOM)) {
-		if (real_time) {
-			if (!_arcs.front().must_mix()) {
-				// Single non-mixing connection, use buffers directly
-				for (uint32_t v = 0; v < poly; ++v) {
-					voices->at(v).buffer = _arcs.front().buffer(v);
-				}
-				return false;
-			}
-		}
 	}
 
 	// Otherwise, allocate local buffers
 	for (uint32_t v = 0; v < poly; ++v) {
 		voices->at(v).buffer.reset();
-		voices->at(v).buffer = bufs.get_buffer(
-			buffer_type(), _value.type(), _buffer_size, real_time);
+		voices->at(v).buffer = (bufs.*get)(
+			buffer_type(), _value.type(), _buffer_size);
 		voices->at(v).buffer->clear();
 		if (_value.is_valid()) {
 			voices->at(v).buffer->set_value(_value);
@@ -110,6 +98,32 @@ InputPort::get_buffers(BufferFactory&      bufs,
 	return true;
 }
 
+bool
+InputPort::pre_get_buffers(BufferFactory&      bufs,
+                           Raul::Array<Voice>* voices,
+                           uint32_t            poly) const
+{
+	return get_buffers(bufs, &BufferFactory::get_buffer, voices, poly, _num_arcs);
+}
+
+bool
+InputPort::setup_buffers(RunContext& ctx, BufferFactory& bufs, uint32_t poly)
+{
+	if (is_a(PortType::ATOM) && !_value.is_valid()) {
+		poly = 1;
+	}
+
+	if (_arcs.size() == 1 && !is_a(PortType::ATOM) && !_arcs.front().must_mix()) {
+		// Single non-mixing connection, use buffers directly
+		for (uint32_t v = 0; v < poly; ++v) {
+			_voices->at(v).buffer = _arcs.front().buffer(v);
+		}
+		return false;
+	}
+
+	return get_buffers(bufs, &BufferFactory::claim_buffer, _voices, poly, _arcs.size());
+}
+
 void
 InputPort::add_arc(RunContext& context, ArcImpl* c)
 {
diff --git a/src/server/InputPort.hpp b/src/server/InputPort.hpp
index 1a27ac93..59e79463 100644
--- a/src/server/InputPort.hpp
+++ b/src/server/InputPort.hpp
@@ -85,14 +85,17 @@ public:
 	ArcImpl* remove_arc(RunContext&     context,
 	                    const PortImpl* tail);
 
-	/** Set `voices` as the buffers to be used for this port.
+	/** Like `get_buffers`, but for the pre-process thread.
 	 *
-	 * @return true iff buffers are locally owned by the port
+	 * This uses the "current" number of arcs fromthe perspective of the
+	 * pre-process thread to allocate buffers for application of a
+	 * connection/disconnection/etc in the next process cycle.
 	 */
-	bool get_buffers(BufferFactory&      bufs,
-	                 Raul::Array<Voice>* voices,
-	                 uint32_t            poly,
-	                 bool                real_time) const;
+	bool pre_get_buffers(BufferFactory&      bufs,
+	                     Raul::Array<Voice>* voices,
+	                     uint32_t            poly) const;
+
+	bool setup_buffers(RunContext& ctx, BufferFactory& bufs, uint32_t poly) override;
 
 	/** Set up buffer pointers. */
 	void pre_process(RunContext& context);
@@ -106,13 +109,19 @@ public:
 	SampleCount next_value_offset(SampleCount offset, SampleCount end) const;
 	void        update_values(SampleCount offset, uint32_t voice);
 
-	size_t num_arcs() const { return _num_arcs; } ///< Pre-process thread
-	void increment_num_arcs() { ++_num_arcs; }
-	void decrement_num_arcs() { --_num_arcs; }
+	size_t num_arcs() const override { return _num_arcs; }
+	void   increment_num_arcs()      { ++_num_arcs; }
+	void   decrement_num_arcs()      { --_num_arcs; }
 
 	bool direct_connect() const;
 
 protected:
+	bool get_buffers(BufferFactory&      bufs,
+	                 PortImpl::GetFn     get,
+	                 Raul::Array<Voice>* voices,
+	                 uint32_t            poly,
+	                 size_t              num_in_arcs) const override;
+
 	size_t _num_arcs;  ///< Pre-process thread
 	Arcs   _arcs;      ///< Audio thread
 };
diff --git a/src/server/PortImpl.cpp b/src/server/PortImpl.cpp
index c27cdb9e..245173a7 100644
--- a/src/server/PortImpl.cpp
+++ b/src/server/PortImpl.cpp
@@ -68,6 +68,7 @@ PortImpl::PortImpl(BufferFactory&      bufs,
 	, _max(bufs.forge().make(1.0f))
 	, _voices(new Raul::Array<Voice>(static_cast<size_t>(poly)))
 	, _prepared_voices(NULL)
+	, _connected_flag(false)
 	, _monitored(false)
 	, _force_monitor_update(false)
 	, _is_morph(false)
@@ -98,9 +99,9 @@ PortImpl::PortImpl(BufferFactory&      bufs,
 		if (_parent->graph_type() != Node::GraphType::GRAPH) {
 			add_property(bufs.uris().rdf_type, bufs.uris().lv2_OutputPort.urid);
 		}
-
-		setup_buffers(bufs, poly, false);
 	}
+
+	get_buffers(bufs, &BufferFactory::get_buffer, _voices, poly, 0);
 }
 
 PortImpl::~PortImpl()
@@ -110,17 +111,26 @@ PortImpl::~PortImpl()
 
 bool
 PortImpl::get_buffers(BufferFactory&      bufs,
+                      GetFn               get,
                       Raul::Array<Voice>* voices,
                       uint32_t            poly,
-                      bool                real_time) const
+                      size_t              num_in_arcs) const
 {
-	for (uint32_t v = 0; v < poly; ++v)
-		voices->at(v).buffer = bufs.get_buffer(
-			buffer_type(), _value.type(), _buffer_size, real_time);
+	for (uint32_t v = 0; v < poly; ++v) {
+		voices->at(v).buffer.reset();
+		voices->at(v).buffer = (bufs.*get)(
+			buffer_type(), _value.type(), _buffer_size);
+	}
 
 	return true;
 }
 
+bool
+PortImpl::setup_buffers(RunContext& ctx, BufferFactory& bufs, uint32_t poly)
+{
+	return get_buffers(bufs, &BufferFactory::claim_buffer, _voices, poly, 0);
+}
+
 void
 PortImpl::set_type(PortType port_type, LV2_URID buffer_type)
 {
@@ -171,10 +181,6 @@ PortImpl::supports(const URIs::Quark& value_type) const
 void
 PortImpl::activate(BufferFactory& bufs)
 {
-	setup_buffers(bufs, _poly, false);
-	connect_buffers();
-	clear_buffers();
-
 	/* Set the time since the last monitor update to a random value within the
 	   monitor period, to spread the load out over time.  Otherwise, every
 	   port would try to send an update at exactly the same time, every time.
@@ -184,6 +190,8 @@ PortImpl::activate(BufferFactory& bufs)
 	_frames_since_monitor = bufs.engine().frand() * period;
 	_monitor_value        = 0.0f;
 	_peak                 = 0.0f;
+
+	_connected_flag.clear();  // Trigger buffer re-connect next cycle
 }
 
 void
@@ -328,24 +336,19 @@ PortImpl::prepare_poly(BufferFactory& bufs, uint32_t poly)
 	if (_is_driver_port || _parent->is_main() ||
 	    (_type == PortType::ATOM && !_value.is_valid())) {
 		return false;
-	}
-
-	if (_poly == poly) {
+	} else if (_poly == poly) {
 		return true;
-	}
-
-	if (_prepared_voices && _prepared_voices->size() != poly) {
+	} else if (_prepared_voices && _prepared_voices->size() != poly) {
 		delete _prepared_voices;
 		_prepared_voices = NULL;
 	}
 
-	if (!_prepared_voices)
+	if (!_prepared_voices) {
 		_prepared_voices = new Raul::Array<Voice>(poly, *_voices, Voice());
+	}
 
-	get_buffers(bufs,
-	            _prepared_voices,
-	            _prepared_voices->size(),
-	            false);
+	get_buffers(bufs, &BufferFactory::get_buffer,
+	            _prepared_voices, _prepared_voices->size(), num_arcs());
 
 	return true;
 }
@@ -356,9 +359,7 @@ PortImpl::apply_poly(RunContext& context, Raul::Maid& maid, uint32_t poly)
 	if (_parent->is_main() ||
 	    (_type == PortType::ATOM && !_value.is_valid())) {
 		return false;
-	}
-
-	if (!_prepared_voices) {
+	} else if (!_prepared_voices) {
 		return true;
 	}
 
@@ -379,6 +380,8 @@ PortImpl::apply_poly(RunContext& context, Raul::Maid& maid, uint32_t poly)
 	assert(this->poly() == poly);
 	assert(!_prepared_voices);
 
+	connect_buffers();
+
 	return true;
 }
 
@@ -396,8 +399,9 @@ PortImpl::set_buffer_size(RunContext& context, BufferFactory& bufs, size_t size)
 void
 PortImpl::connect_buffers(SampleCount offset)
 {
-	for (uint32_t v = 0; v < _poly; ++v)
+	for (uint32_t v = 0; v < _poly; ++v) {
 		PortImpl::parent_block()->set_port_buffer(v, _index, buffer(v), offset);
+	}
 }
 
 void
@@ -547,6 +551,11 @@ PortImpl::update_values(SampleCount offset, uint32_t voice)
 void
 PortImpl::pre_process(RunContext& context)
 {
+	if (!_connected_flag.test_and_set()) {
+		connect_buffers();
+		clear_buffers();
+	}
+
 	for (uint32_t v = 0; v < _poly; ++v)
 		_voices->at(v).buffer->prepare_output_write(context);
 }
diff --git a/src/server/PortImpl.hpp b/src/server/PortImpl.hpp
index 05ca3ff4..fd35b52d 100644
--- a/src/server/PortImpl.hpp
+++ b/src/server/PortImpl.hpp
@@ -124,6 +124,9 @@ public:
 	virtual bool apply_poly(
 		RunContext& context, Raul::Maid& maid, uint32_t poly);
 
+	/** Return the number of arcs (pre-process thraed). */
+	virtual size_t num_arcs() const { return 0; }
+
 	const Atom& value() const { return _value; }
 	void        set_value(const Atom& v) { _value = v; }
 
@@ -175,15 +178,8 @@ public:
 	/** Empty buffer contents completely (ie silence) */
 	virtual void clear_buffers();
 
-public:
-	virtual bool get_buffers(BufferFactory&      bufs,
-	                         Raul::Array<Voice>* voices,
-	                         uint32_t            poly,
-	                         bool                real_time) const;
-
-	void setup_buffers(BufferFactory& bufs, uint32_t poly, bool real_time) {
-		get_buffers(bufs, _voices, poly, real_time);
-	}
+	/** Claim and apply buffers in the real-time thread. */
+	virtual bool setup_buffers(RunContext& ctx, BufferFactory& bufs, uint32_t poly);
 
 	void activate(BufferFactory& bufs);
 	void deactivate();
@@ -272,6 +268,22 @@ public:
 	bool is_toggled()     const { return _is_toggled; }
 
 protected:
+	typedef BufferRef (BufferFactory::*GetFn)(LV2_URID, LV2_URID, uint32_t);
+
+	/** Set `voices` as the buffers to be used for this port.
+	 *
+	 * This is real-time safe only if `get` is as well, use in the real-time
+	 * thread should pass &BufferFactory::claim_buffer.
+	 *
+	 * @param get Method used to get new buffers as necessary.
+	 * @return true iff buffers are locally owned by the port
+	 */
+	virtual bool get_buffers(BufferFactory&      bufs,
+	                         GetFn               get,
+	                         Raul::Array<Voice>* voices,
+	                         uint32_t            poly,
+	                         size_t              num_in_arcs) const;
+
 	BufferFactory&      _bufs;
 	uint32_t            _index;
 	uint32_t            _poly;
@@ -287,6 +299,7 @@ protected:
 	Raul::Array<Voice>* _voices;
 	Raul::Array<Voice>* _prepared_voices;
 	BufferRef           _user_buffer;
+	std::atomic_flag    _connected_flag;
 	bool                _monitored;
 	bool                _force_monitor_update;
 	bool                _is_morph;
diff --git a/src/server/events/Connect.cpp b/src/server/events/Connect.cpp
index c24fc2bb..eb758940 100644
--- a/src/server/events/Connect.cpp
+++ b/src/server/events/Connect.cpp
@@ -139,10 +139,9 @@ Connect::pre_process(PreProcessContext& ctx)
 
 	if (!_head->is_driver_port()) {
 		_voices = new Raul::Array<PortImpl::Voice>(_head->poly());
-		_head->get_buffers(*_engine.buffer_factory(),
-		                   _voices,
-		                   _head->poly(),
-		                   false);
+		_head->pre_get_buffers(*_engine.buffer_factory(),
+		                       _voices,
+		                       _head->poly());
 	}
 
 	tail_output->inherit_neighbour(_head, _tail_remove, _tail_add);
diff --git a/src/server/events/Disconnect.cpp b/src/server/events/Disconnect.cpp
index 8875eb0c..3233cac4 100644
--- a/src/server/events/Disconnect.cpp
+++ b/src/server/events/Disconnect.cpp
@@ -90,10 +90,9 @@ Disconnect::Impl::Impl(Engine&     e,
 	if (_head->num_arcs() == 0) {
 		if (!_head->is_driver_port()) {
 			_voices = new Raul::Array<PortImpl::Voice>(_head->poly());
-			_head->get_buffers(*_engine.buffer_factory(),
-			                   _voices,
-			                   _head->poly(),
-			                   false);
+			_head->pre_get_buffers(*_engine.buffer_factory(),
+			                       _voices,
+			                       _head->poly());
 
 			if (_head->is_a(PortType::CONTROL) ||
 			    _head->is_a(PortType::CV)) {
@@ -189,9 +188,7 @@ Disconnect::Impl::execute(RunContext& context, bool set_head_buffers)
 		if (_voices) {
 			_engine.maid()->dispose(_head->set_voices(context, _voices));
 		} else {
-			_head->setup_buffers(*_engine.buffer_factory(),
-			                     _head->poly(),
-			                     true);
+			_head->setup_buffers(context, *_engine.buffer_factory(), _head->poly());
 		}
 		_head->connect_buffers();
 	} else {
diff --git a/src/server/events/SetPortValue.cpp b/src/server/events/SetPortValue.cpp
index 4ab0271c..ff0827e4 100644
--- a/src/server/events/SetPortValue.cpp
+++ b/src/server/events/SetPortValue.cpp
@@ -74,9 +74,7 @@ SetPortValue::pre_process(PreProcessContext& ctx)
 		_buffer = _engine.buffer_factory()->get_buffer(
 			_port->buffer_type(),
 			0,
-			_engine.buffer_factory()->default_size(_port->buffer_type()),
-			false,
-			false);
+			_engine.buffer_factory()->default_size(_port->buffer_type()));
 	}
 
 	return Event::pre_process_done(Status::SUCCESS);
diff --git a/src/server/internals/BlockDelay.cpp b/src/server/internals/BlockDelay.cpp
index 3dee2feb..d52ae721 100644
--- a/src/server/internals/BlockDelay.cpp
+++ b/src/server/internals/BlockDelay.cpp
@@ -68,8 +68,8 @@ BlockDelayNode::~BlockDelayNode()
 void
 BlockDelayNode::activate(BufferFactory& bufs)
 {
-	_buffer = bufs.get_buffer(
-		bufs.uris().atom_Sound, 0, bufs.audio_buffer_size(), false, true);
+	_buffer = bufs.create(
+		bufs.uris().atom_Sound, 0, bufs.audio_buffer_size());
 
 	BlockImpl::activate(bufs);
 }