Support thread-safe state restoration

8 files changed, 80 insertions, 26 deletions

diff --git a/src/server/BlockImpl.hpp b/src/server/BlockImpl.hpp
index 773cb115..1924d5fe 100644
--- a/src/server/BlockImpl.hpp
+++ b/src/server/BlockImpl.hpp
@@ -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
@@ -48,6 +48,7 @@ class GraphImpl;
 class PluginImpl;
 class PortImpl;
 class ProcessContext;
+class Worker;
 
 /** A Block in a Graph (which is also a Block).
  *
@@ -103,7 +104,7 @@ public:
 	virtual LilvState* load_preset(const Raul::URI& uri) { return NULL; }
 
 	/** Restore `state`. */
-	virtual void apply_state(LilvState* state) {}
+	virtual void apply_state(Worker* worker, LilvState* state) {}
 
 	/** Save current state as preset. */
 	virtual boost::optional<Resource>
diff --git a/src/server/Engine.cpp b/src/server/Engine.cpp
index beb38a24..ee56be3c 100644
--- a/src/server/Engine.cpp
+++ b/src/server/Engine.cpp
@@ -78,6 +78,7 @@ Engine::Engine(Ingen::World* world)
 	, _post_processor(new PostProcessor(*this))
 	, _root_graph(NULL)
 	, _worker(new Worker(world->log(), event_queue_size()))
+	, _sync_worker(new Worker(world->log(), event_queue_size(), true))
 	, _listener(NULL)
 	, _process_context(*this)
 	, _rand_engine(0)
diff --git a/src/server/Engine.hpp b/src/server/Engine.hpp
index d4ff4420..37c3c713 100644
--- a/src/server/Engine.hpp
+++ b/src/server/Engine.hpp
@@ -116,6 +116,7 @@ public:
 	UndoStack*       undo_stack()       const { return _undo_stack; }
 	UndoStack*       redo_stack()       const { return _redo_stack; }
 	Worker*          worker()           const { return _worker; }
+	Worker*          sync_worker()      const { return _sync_worker; }
 
 	ProcessContext& process_context() { return _process_context; }
 
@@ -141,6 +142,7 @@ private:
 	PostProcessor*   _post_processor;
 	GraphImpl*       _root_graph;
 	Worker*          _worker;
+	Worker*          _sync_worker;
 	SocketListener*  _listener;
 
 	ProcessContext _process_context;
diff --git a/src/server/LV2Block.cpp b/src/server/LV2Block.cpp
index e1f67897..de91699b 100644
--- a/src/server/LV2Block.cpp
+++ b/src/server/LV2Block.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
@@ -44,6 +44,7 @@
 #include "LV2Plugin.hpp"
 #include "OutputPort.hpp"
 #include "ProcessContext.hpp"
+#include "Worker.hpp"
 
 using namespace std;
 
@@ -75,7 +76,7 @@ LV2Block::~LV2Block()
 }
 
 void
-LV2Block::load_default_state()
+LV2Block::load_default_state(Worker* worker)
 {
 	const LilvPlugin* lplug    = _lv2_plugin->lilv_plugin();
 	const LilvNode*   uri_node = lilv_plugin_get_uri(lplug);
@@ -83,7 +84,7 @@ LV2Block::load_default_state()
 
 	LilvState* default_state = load_preset(_lv2_plugin->uri());
 	if (default_state) {
-		apply_state(default_state);
+		apply_state(worker, default_state);
 		lilv_state_free(default_state);
 	}
 }
@@ -445,7 +446,7 @@ LV2Block::instantiate(BufferFactory& bufs)
 		}
 	}
 
-	load_default_state();
+	load_default_state(NULL);
 
 	// FIXME: Polyphony + worker?
 	if (lilv_plugin_has_feature(plug, uris.work_schedule)) {
@@ -526,16 +527,21 @@ LV2Block::work_respond(LV2_Worker_Respond_Handle handle,
 	return LV2_WORKER_SUCCESS;
 }
 
-void
+LV2_Worker_Status
 LV2Block::work(uint32_t size, const void* data)
 {
 	if (_worker_iface) {
-		LV2_Handle inst = lilv_instance_get_handle(instance(0));
-		if (_worker_iface->work(inst, work_respond, this, size, data)) {
+		std::lock_guard<std::mutex> lock(_work_mutex);
+
+		LV2_Handle        inst = lilv_instance_get_handle(instance(0));
+		LV2_Worker_Status st   = _worker_iface->work(inst, work_respond, this, size, data);
+		if (st) {
 			parent_graph()->engine().log().error(
 				fmt("Error calling %1% work method\n") % _path);
 		}
+		return st;
 	}
+	return LV2_WORKER_ERR_UNKNOWN;
 }
 
 void
@@ -584,10 +590,21 @@ LV2Block::load_preset(const Raul::URI& uri)
 }
 
 void
-LV2Block::apply_state(LilvState* state)
+LV2Block::apply_state(Worker* worker, LilvState* state)
 {
+	World*            world = parent_graph()->engine().world();
+	SPtr<LV2_Feature> sched;
+	if (worker) {
+		sched = worker->schedule_feature()->feature(world, this);
+	}
+
+	const LV2_Feature* state_features[2] = { NULL, NULL };
+	if (sched) {
+		state_features[0] = sched.get();
+	}
+
 	for (uint32_t v = 0; v < _polyphony; ++v) {
-		lilv_state_restore(state, instance(v), NULL, NULL, 0, NULL);
+		lilv_state_restore(state, instance(v), NULL, NULL, 0, state_features);
 	}
 }
 
diff --git a/src/server/LV2Block.hpp b/src/server/LV2Block.hpp
index 6483f8b0..6e78ad83 100644
--- a/src/server/LV2Block.hpp
+++ b/src/server/LV2Block.hpp
@@ -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
@@ -17,6 +17,8 @@
 #ifndef INGEN_ENGINE_LV2BLOCK_HPP
 #define INGEN_ENGINE_LV2BLOCK_HPP
 
+#include <mutex>
+
 #include "lilv/lilv.h"
 #include "lv2/lv2plug.in/ns/ext/worker/worker.h"
 #include "raul/Maid.hpp"
@@ -60,14 +62,14 @@ public:
 	void activate(BufferFactory& bufs);
 	void deactivate();
 
-	void work(uint32_t size, const void* data);
+	LV2_Worker_Status work(uint32_t size, const void* data);
 
 	void run(ProcessContext& context);
 	void post_process(ProcessContext& context);
 
 	LilvState* load_preset(const Raul::URI& uri);
 
-	void apply_state(LilvState* state);
+	void apply_state(Worker* worker, LilvState* state);
 
 	boost::optional<Resource> save_preset(const Raul::URI&  bundle,
 	                                      const Properties& props);
@@ -83,7 +85,7 @@ protected:
 	                                 uint32_t   voice,
 	                                 bool       preparing);
 
-	void load_default_state();
+	void load_default_state(Worker* worker);
 
 	inline LilvInstance* instance(uint32_t voice) {
 		return (LilvInstance*)(*_instances)[voice].get();
@@ -122,6 +124,7 @@ protected:
 	Instances*                      _instances;
 	Instances*                      _prepared_instances;
 	const LV2_Worker_Interface*     _worker_iface;
+	std::mutex                      _work_mutex;
 	Responses                       _responses;
 	SPtr<LV2Features::FeatureArray> _features;
 };
diff --git a/src/server/Worker.cpp b/src/server/Worker.cpp
index a89bd4b3..1ff7a89c 100644
--- a/src/server/Worker.cpp
+++ b/src/server/Worker.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
@@ -46,11 +46,27 @@ schedule(LV2_Worker_Schedule_Handle handle,
 	return worker->request(block, size, data);
 }
 
+static LV2_Worker_Status
+schedule_sync(LV2_Worker_Schedule_Handle handle,
+              uint32_t                   size,
+              const void*                data)
+{
+	LV2Block* block  = (LV2Block*)handle;
+	Engine&   engine = block->parent_graph()->engine();
+	Worker*   worker = engine.sync_worker();
+
+	return worker->request(block, size, data);
+}
+
 LV2_Worker_Status
 Worker::request(LV2Block*   block,
                 uint32_t    size,
                 const void* data)
 {
+	if (_synchronous) {
+		return block->work(size, data);
+	}
+
 	Engine& engine = block->parent_graph()->engine();
 	if (_requests.write_space() < sizeof(MessageHeader) + size) {
 		engine.log().error("Work request ring overflow\n");
@@ -83,7 +99,7 @@ Worker::Schedule::feature(World* world, Node* n)
 	LV2_Worker_Schedule* data = (LV2_Worker_Schedule*)malloc(
 		sizeof(LV2_Worker_Schedule));
 	data->handle        = block;
-	data->schedule_work = schedule;
+	data->schedule_work = synchronous ? schedule_sync : schedule;
 
 	LV2_Feature* f = (LV2_Feature*)malloc(sizeof(LV2_Feature));
 	f->URI  = LV2_WORKER__schedule;
@@ -92,22 +108,31 @@ Worker::Schedule::feature(World* world, Node* n)
 	return SPtr<LV2_Feature>(f, &free_feature);
 }
 
-Worker::Worker(Log& log, uint32_t buffer_size)
-	: _schedule(new Schedule())
+Worker::Worker(Log& log, uint32_t buffer_size, bool synchronous)
+	: _schedule(new Schedule(synchronous))
 	, _log(log)
 	, _sem(0)
 	, _requests(buffer_size)
 	, _responses(buffer_size)
 	, _buffer((uint8_t*)malloc(buffer_size))
 	, _buffer_size(buffer_size)
-	, _thread(&Worker::run, this)
-{}
+	, _thread(nullptr)
+	, _exit_flag(false)
+	, _synchronous(synchronous)
+{
+	if (!synchronous) {
+		_thread = new std::thread(&Worker::run, this);
+	}
+}
 
 Worker::~Worker()
 {
 	_exit_flag = true;
 	_sem.post();
-	_thread.join();
+	if (_thread) {
+		_thread->join();
+		delete _thread;
+	}
 	free(_buffer);
 }
 
diff --git a/src/server/Worker.hpp b/src/server/Worker.hpp
index 1cd6aff8..0a3fdeaf 100644
--- a/src/server/Worker.hpp
+++ b/src/server/Worker.hpp
@@ -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
@@ -35,13 +35,17 @@ class LV2Block;
 class Worker
 {
 public:
-	Worker(Log& log, uint32_t buffer_size);
+	Worker(Log& log, uint32_t buffer_size, bool synchronous=false);
 	~Worker();
 
 	struct Schedule : public LV2Features::Feature {
+		Schedule(bool sync) : synchronous(sync) {}
+
 		const char* uri() const { return LV2_WORKER__schedule; }
 
 		SPtr<LV2_Feature> feature(World* world, Node* n);
+
+		const bool synchronous;
 	};
 
 	LV2_Worker_Status request(LV2Block*   block,
@@ -59,8 +63,9 @@ private:
 	Raul::RingBuffer _responses;
 	uint8_t* const   _buffer;
 	const uint32_t   _buffer_size;
+	std::thread*     _thread;
 	bool             _exit_flag;
-	std::thread      _thread;
+	bool             _synchronous;
 
 	void run();
 };
diff --git a/src/server/events/Delta.cpp b/src/server/events/Delta.cpp
index a765932b..898fd971 100644
--- a/src/server/events/Delta.cpp
+++ b/src/server/events/Delta.cpp
@@ -578,7 +578,7 @@ Delta::post_process()
 	if (_state) {
 		BlockImpl* block = dynamic_cast<BlockImpl*>(_object);
 		if (block) {
-			block->apply_state(_state);
+			block->apply_state(_engine.sync_worker(), _state);
 			block->set_enabled(true);
 		}
 		lilv_state_free(_state);