Fix some real-time safety issues

13 files changed, 62 insertions, 40 deletions

diff --git a/src/Log.cpp b/src/Log.cpp
index 9b5a09ce..259d9a3e 100644
--- a/src/Log.cpp
+++ b/src/Log.cpp
@@ -32,6 +32,15 @@ Log::Log(LV2_Log_Log* log, URIs& uris)
 {}
 
 void
+Log::rt_error(const char* msg)
+{
+#ifndef NDEBUG
+	va_list args;
+	vtprintf(_uris.log_Error, msg, args);
+#endif
+}
+
+void
 Log::error(const std::string& msg)
 {
 	va_list args;
diff --git a/src/server/Buffer.cpp b/src/server/Buffer.cpp
index 1a1f804f..c5f9a1c2 100644
--- a/src/server/Buffer.cpp
+++ b/src/server/Buffer.cpp
@@ -67,7 +67,7 @@ Buffer::Buffer(BufferFactory& bufs,
 #endif
 
 		if (ret) {
-			bufs.engine().log().error("Failed to allocate event buffer\n");
+			bufs.engine().log().rt_error("Failed to allocate event buffer\n");
 			throw std::bad_alloc();
 		}
 	}
diff --git a/src/server/ControlBindings.cpp b/src/server/ControlBindings.cpp
index 2e18442c..0063c682 100644
--- a/src/server/ControlBindings.cpp
+++ b/src/server/ControlBindings.cpp
@@ -80,9 +80,9 @@ ControlBindings::binding_key(const Atom& binding) const
 			lv2_atom_object_body_get(
 				binding.size(), obj, (LV2_URID)uris.midi_controllerNumber, &num, NULL);
 			if (!num) {
-				_engine.log().error("Controller binding missing number\n");
+				_engine.log().rt_error("Controller binding missing number\n");
 			} else if (num->type != uris.atom_Int) {
-				_engine.log().error("Controller number not an integer\n");
+				_engine.log().rt_error("Controller number not an integer\n");
 			} else {
 				key = Key(Type::MIDI_CC, ((LV2_Atom_Int*)num)->body);
 			}
@@ -90,15 +90,15 @@ ControlBindings::binding_key(const Atom& binding) const
 			lv2_atom_object_body_get(
 				binding.size(), obj, (LV2_URID)uris.midi_noteNumber, &num, NULL);
 			if (!num) {
-				_engine.log().error("Note binding missing number\n");
+				_engine.log().rt_error("Note binding missing number\n");
 			} else if (num->type != uris.atom_Int) {
-				_engine.log().error("Note number not an integer\n");
+				_engine.log().rt_error("Note number not an integer\n");
 			} else {
 				key = Key(Type::MIDI_NOTE, ((LV2_Atom_Int*)num)->body);
 			}
 		}
 	} else if (binding.type()) {
-		_engine.log().error(fmt("Unknown binding type %1%\n") % binding.type());
+		_engine.log().rt_error("Unknown binding type\n");
 	}
 	return key;
 }
diff --git a/src/server/Engine.cpp b/src/server/Engine.cpp
index 6e61f463..1a9d82b0 100644
--- a/src/server/Engine.cpp
+++ b/src/server/Engine.cpp
@@ -99,7 +99,9 @@ Engine::Engine(Ingen::World* world)
 	_control_bindings = new ControlBindings(*this);
 
 	for (int i = 0; i < world->conf().option("threads").get<int32_t>(); ++i) {
-		_run_contexts.push_back(new RunContext(*this, i, i > 0));
+		Raul::RingBuffer* ring = new Raul::RingBuffer(24 * event_queue_size());
+		_notifications.push_back(ring);
+		_run_contexts.push_back(new RunContext(*this, ring, i, i > 0));
 	}
 
 	_world->lv2_features().add_feature(_worker->schedule_feature());
@@ -149,6 +151,13 @@ Engine::~Engine()
 		_post_processor->process();
 	}
 
+	for (RunContext* ctx : _run_contexts) {
+		delete ctx;
+	}
+	for (Raul::RingBuffer* ring : _notifications) {
+		delete ring;
+	}
+
 	const SPtr<Store> store = this->store();
 	if (store) {
 		for (auto& s : *store.get()) {
diff --git a/src/server/Engine.hpp b/src/server/Engine.hpp
index 5be18faf..a4b85562 100644
--- a/src/server/Engine.hpp
+++ b/src/server/Engine.hpp
@@ -205,11 +205,12 @@ private:
 	Worker*           _sync_worker;
 	SocketListener*   _listener;
 
-	std::vector<RunContext*> _run_contexts;
-	uint64_t                 _cycle_start_time;
-	Load                     _event_load;
-	Load                     _run_load;
-	Clock                    _clock;
+	std::vector<Raul::RingBuffer*> _notifications;
+	std::vector<RunContext*>       _run_contexts;
+	uint64_t                       _cycle_start_time;
+	Load                           _event_load;
+	Load                           _run_load;
+	Clock                          _clock;
 
 	std::mt19937                          _rand_engine;
 	std::uniform_real_distribution<float> _uniform_dist;
diff --git a/src/server/InputPort.cpp b/src/server/InputPort.cpp
index 5133758c..4e1ced3e 100644
--- a/src/server/InputPort.cpp
+++ b/src/server/InputPort.cpp
@@ -130,7 +130,7 @@ InputPort::remove_arc(RunContext& context, const OutputPort* tail)
 	}
 
 	if (!arc) {
-		context.engine().log().error("Attempt to remove non-existent arc\n");
+		context.engine().log().rt_error("Attempt to remove non-existent arc\n");
 		return NULL;
 	}
 
diff --git a/src/server/JackDriver.cpp b/src/server/JackDriver.cpp
index 7f0e7ae0..fb672ffa 100644
--- a/src/server/JackDriver.cpp
+++ b/src/server/JackDriver.cpp
@@ -362,7 +362,7 @@ JackDriver::pre_process_port(RunContext& context, EnginePort* port)
 				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");
+					_engine.log().rt_error("Failed to write to MIDI buffer, events lost!\n");
 				}
 			}
 		}
diff --git a/src/server/PortImpl.cpp b/src/server/PortImpl.cpp
index c99a26fb..1f1b3695 100644
--- a/src/server/PortImpl.cpp
+++ b/src/server/PortImpl.cpp
@@ -251,9 +251,11 @@ PortImpl::set_voice_value(const RunContext& context,
 			}
 			_voices->at(voice).set_state.set(context, time, value);
 		} else {
+#ifndef NDEBUG
 			fprintf(stderr,
 			        "error: %s set non-sequence atom port value (buffer type %u)\n",
 			        path().c_str(), buffer(voice)->type());
+#endif
 		}
 	default:
 		break;
diff --git a/src/server/PortImpl.hpp b/src/server/PortImpl.hpp
index e497176c..94fdd50c 100644
--- a/src/server/PortImpl.hpp
+++ b/src/server/PortImpl.hpp
@@ -119,8 +119,8 @@ public:
 	   accessed in the process thread, which is required for applying control
 	   bindings from incoming MIDI data.
 	*/
-	void set_minimum(const Atom& min) { _min = min; }
-	void set_maximum(const Atom& max) { _max = max; }
+	void set_minimum(const Atom& min) { _min.set_rt(min); }
+	void set_maximum(const Atom& max) { _max.set_rt(max); }
 
 	inline BufferRef buffer(uint32_t voice) const {
 		return _voices->at((_poly == 1) ? 0 : voice).buffer;
diff --git a/src/server/PreProcessor.cpp b/src/server/PreProcessor.cpp
index c23c99d6..07c51fee 100644
--- a/src/server/PreProcessor.cpp
+++ b/src/server/PreProcessor.cpp
@@ -129,12 +129,14 @@ PreProcessor::process(RunContext& context, PostProcessor& dest, size_t limit)
 
 	if (n_processed > 0) {
 		Engine& engine = context.engine();
+#ifndef NDEBUG
 		if (engine.world()->conf().option("trace").get<int32_t>()) {
 			const uint64_t start = engine.cycle_start_time(context);
 			const uint64_t end   = engine.current_time(context);
 			fprintf(stderr, "Processed %zu events in %u us\n",
 			        n_processed, (unsigned)(end - start));
 		}
+#endif
 
 		Event* next = (Event*)last->next();
 		last->next(NULL);
diff --git a/src/server/RunContext.cpp b/src/server/RunContext.cpp
index e107d247..1ab73605 100644
--- a/src/server/RunContext.cpp
+++ b/src/server/RunContext.cpp
@@ -46,10 +46,12 @@ struct Notification
 	LV2_URID  type;
 };
 
-RunContext::RunContext(Engine& engine, unsigned id, bool threaded)
+RunContext::RunContext(Engine&           engine,
+                       Raul::RingBuffer* event_sink,
+                       unsigned          id,
+                       bool              threaded)
 	: _engine(engine)
-	, _event_sink(
-		new Raul::RingBuffer(engine.event_queue_size() * sizeof(Notification)))
+	, _event_sink(event_sink)
 	, _task(nullptr)
 	, _thread(threaded ? new std::thread(&RunContext::run, this) : nullptr)
 	, _id(id)
@@ -58,7 +60,6 @@ RunContext::RunContext(Engine& engine, unsigned id, bool threaded)
 	, _offset(0)
 	, _nframes(0)
 	, _realtime(true)
-	, _copy(false)
 {}
 
 RunContext::RunContext(const RunContext& copy)
@@ -72,15 +73,10 @@ RunContext::RunContext(const RunContext& copy)
 	, _offset(copy._offset)
 	, _nframes(copy._nframes)
 	, _realtime(copy._realtime)
-	, _copy(true)
 {}
 
 RunContext::~RunContext()
-{
-	if (!_copy) {
-		delete _event_sink;
-	}
-}
+{}
 
 bool
 RunContext::must_notify(const PortImpl* port) const
@@ -101,9 +97,9 @@ RunContext::notify(LV2_URID    key,
 		return false;
 	}
 	if (_event_sink->write(sizeof(n), &n) != sizeof(n)) {
-		_engine.log().error("Error writing header to notification ring\n");
+		_engine.log().rt_error("Error writing header to notification ring\n");
 	} else if (_event_sink->write(size, body) != size) {
-		_engine.log().error("Error writing body to notification ring\n");
+		_engine.log().rt_error("Error writing body to notification ring\n");
 	} else {
 		return true;
 	}
@@ -135,13 +131,13 @@ RunContext::emit_notifications(FrameTime end)
 						note.port->set_property(uris.ingen_value, value);
 					}
 				} else {
-					_engine.log().error("Error unmapping notification key URI\n");
+					_engine.log().rt_error("Error unmapping notification key URI\n");
 				}
 			} else {
-				_engine.log().error("Error reading body from notification ring\n");
+				_engine.log().rt_error("Error reading body from notification ring\n");
 			}
 		} else {
-			_engine.log().error("Error reading header from notification ring\n");
+			_engine.log().rt_error("Error reading header from notification ring\n");
 		}
 	}
 }
diff --git a/src/server/RunContext.hpp b/src/server/RunContext.hpp
index feeb4874..d2f24f8d 100644
--- a/src/server/RunContext.hpp
+++ b/src/server/RunContext.hpp
@@ -50,11 +50,15 @@ public:
 	/** Create a new run context.
 	 *
 	 * @param engine The engine this context is running within.
+	 * @param event_sink Sink for notification events (peaks etc)
 	 * @param id The ID of this context.
 	 * @param threaded If true, then this context is a worker which will launch
 	 * a thread and execute tasks as they become available.
 	 */
-	RunContext(Engine& engine, unsigned id, bool threaded);
+	RunContext(Engine&           engine,
+	           Raul::RingBuffer* event_sink,
+	           unsigned          id,
+	           bool              threaded);
 
 	/** Create a sub-context of `parent`.
 	 *
@@ -147,7 +151,6 @@ protected:
 	SampleCount _nframes;    ///< Number of frames past offset to process
 	SampleCount _rate;       ///< Sample rate in Hz
 	bool        _realtime;   ///< True iff context is hard realtime
-	bool        _copy;       ///< True iff this is a copy (shared event_sink)
 };
 
 } // namespace Server
diff --git a/src/server/ingen_lv2.cpp b/src/server/ingen_lv2.cpp
index 40d5b664..ff2d7189 100644
--- a/src/server/ingen_lv2.cpp
+++ b/src/server/ingen_lv2.cpp
@@ -291,7 +291,7 @@ public:
 		for (uint32_t read = 0; read < read_space;) {
 			LV2_Atom atom;
 			if (!_from_ui.read(sizeof(LV2_Atom), &atom)) {
-				_engine.log().error("Error reading head from from-UI ring\n");
+				_engine.log().rt_error("Error reading head from from-UI ring\n");
 				break;
 			}
 
@@ -299,7 +299,7 @@ public:
 			memcpy(buf, &atom, sizeof(LV2_Atom));
 
 			if (!_from_ui.read(atom.size, (char*)buf + sizeof(LV2_Atom))) {
-				_engine.log().error("Error reading body from from-UI ring\n");
+				_engine.log().rt_error("Error reading body from from-UI ring\n");
 				break;
 			}
 
@@ -311,13 +311,13 @@ public:
 
 	void flush_to_ui(RunContext& context) {
 		if (_ports.size() < 2) {
-			_engine.log().error("Standard control ports are not present\n");
+			_engine.log().rt_error("Standard control ports are not present\n");
 			return;
 		}
 
 		LV2_Atom_Sequence* seq = (LV2_Atom_Sequence*)_ports[1]->buffer();
 		if (!seq) {
-			_engine.log().error("Notify output not connected\n");
+			_engine.log().rt_error("Notify output not connected\n");
 			return;
 		}
 
@@ -329,7 +329,7 @@ public:
 		for (uint32_t read = 0; read < read_space;) {
 			LV2_Atom atom;
 			if (!_to_ui.peek(sizeof(LV2_Atom), &atom)) {
-				_engine.log().error("Error reading head from to-UI ring\n");
+				_engine.log().rt_error("Error reading head from to-UI ring\n");
 				break;
 			}
 
@@ -347,7 +347,7 @@ public:
 
 			_to_ui.skip(sizeof(LV2_Atom));
 			if (!_to_ui.read(ev->body.size, LV2_ATOM_BODY(&ev->body))) {
-				_engine.log().error("Error reading body from to-UI ring\n");
+				_engine.log().rt_error("Error reading body from to-UI ring\n");
 				break;
 			}
 
@@ -600,7 +600,7 @@ ingen_connect_port(LV2_Handle instance, uint32_t port, void* data)
 	if (port < driver->ports().size()) {
 		driver->ports().at(port)->set_buffer(data);
 	} else {
-		engine->log().error(fmt("Connect to non-existent port %1%\n") % port);
+		engine->log().rt_error("Connect to non-existent port\n");
 	}
 }