Preliminary alternative parallel traversal/benchmarking work

6 files changed, 124 insertions, 117 deletions

diff --git a/src/server/CompiledGraph.cpp b/src/server/CompiledGraph.cpp
index b808cb90..5553140c 100644
--- a/src/server/CompiledGraph.cpp
+++ b/src/server/CompiledGraph.cpp
@@ -1,6 +1,6 @@
 /*
   This file is part of Ingen.
-  Copyright 2015-2016 David Robillard <http://drobilla.net/>
+  Copyright 2015-2017 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
@@ -43,7 +43,7 @@ public:
 CompiledGraph::CompiledGraph(GraphImpl* graph)
 	: _log(graph->engine().log())
 	, _path(graph->path())
-	, _master(Task::Mode::SEQUENTIAL)
+	, _master(new Task(Task::Mode::PARALLEL))
 {
 	compile_graph(graph);
 }
@@ -76,7 +76,7 @@ CompiledGraph::compile_set(const std::set<BlockImpl*>& blocks,
 		// Each block is the start of a new sequential task
 		Task seq(Task::Mode::SEQUENTIAL);
 		compile_block(block, seq, k);
-		task.push_back(seq);
+		task.push_front(std::move(seq));
 	}
 }
 
@@ -91,41 +91,29 @@ CompiledGraph::compile_graph(GraphImpl* graph)
 		// Mark all blocks as unvisited initially
 		b.set_mark(BlockImpl::Mark::UNVISITED);
 
-		if (b.providers().empty()) {
-			// Block has no dependencies, add to initial working set
+		if (b.dependants().empty()) {
+			// Block has no dependants, add to initial working set
 			blocks.insert(&b);
 		}
 	}
 
-	// Compile initial working set into master task
-	Task                 start(Task::Mode::PARALLEL);
-	std::set<BlockImpl*> next;
-	compile_set(blocks, start, next);
-	_master.push_back(start);
-
 	// Keep compiling working set until all connected nodes are visited
-	while (!next.empty()) {
-		blocks.clear();
+	while (!blocks.empty()) {
+        std::set<BlockImpl*> next;
+
 		// The working set is a parallel task...
 		Task par(Task::Mode::PARALLEL);
-		for (BlockImpl* block : next)  {
-			// ... where each block is the start of a new sequential task
+		for (BlockImpl* block : blocks)  {
+			// ... where each block is the tail of a new sequential task
 			Task seq(Task::Mode::SEQUENTIAL);
-			compile_block(block, seq, blocks);
-			par.push_back(seq);
-		}
-		_master.push_back(par);
-		next = blocks;
-	}
-
-	// Compile any nodes that weren't reached (disconnected cycles)
-	for (auto& b : graph->blocks()) {
-		if (b.get_mark() == BlockImpl::Mark::UNVISITED) {
-			compile_block(&b, _master, next);
+			compile_block(block, seq, next);
+			par.push_front(std::move(seq));
 		}
+		_master->push_front(std::move(par));
+		blocks = next;
 	}
 
-	_master.simplify();
+	_master = Task::simplify(std::move(_master));
 
 	if (graph->engine().world()->conf().option("trace").get<int32_t>()) {
 		dump([this](const std::string& msg) {
@@ -160,23 +148,23 @@ check_feedback(const BlockImpl* root, BlockImpl* dependant)
 }
 
 void
-CompiledGraph::compile_dependant(const BlockImpl*      root,
-                                 BlockImpl*            block,
-                                 Task&                 task,
-                                 std::set<BlockImpl*>& k)
+CompiledGraph::compile_provider(const BlockImpl*      root,
+                                BlockImpl*            block,
+                                Task&                 task,
+                                std::set<BlockImpl*>& k)
 {
-	if (block->providers().size() > 1) {
-		/* Dependant has other providers, so this is the start of a sequential task.
-		   Add dependant to future working set and stop traversal. */
+	if (block->dependants().size() > 1) {
+		/* Provider has other dependants, so this is the start of a sequential task.
+		   Add provider to future working set and stop traversal. */
 		check_feedback(root, block);
 		k.insert(block);
 	} else {
-		// Dependant has only this provider, add here
+		// Provider has only this dependant, add here
 		if (task.mode() == Task::Mode::PARALLEL) {
 			// Inside a parallel task, compile into a new sequential child
 			Task seq(Task::Mode::SEQUENTIAL);
 			compile_block(block, seq, k);
-			task.push_back(seq);
+			task.push_front(std::move(seq));
 		} else {
 			// Append to given sequential task
 			compile_block(block, task, k);
@@ -191,21 +179,21 @@ CompiledGraph::compile_block(BlockImpl* n, Task& task, std::set<BlockImpl*>& k)
 	case BlockImpl::Mark::UNVISITED:
 		n->set_mark(BlockImpl::Mark::VISITING);
 
-		// Execute this task before the dependants to follow
-		task.push_back(Task(Task::Mode::SINGLE, n));
+		// Execute this task after the providers to follow
+		task.push_front(Task(Task::Mode::SINGLE, n));
 
-		if (n->dependants().size() < 2) {
-			// Single dependant, append to this sequential task
-			for (auto& d : n->dependants()) {
-				compile_dependant(n, d, task, k);
+		if (n->providers().size() < 2) {
+			// Single provider, append to this sequential task
+			for (auto& d : n->providers()) {
+				compile_provider(n, d, task, k);
 			}
 		} else {
-			// Multiple dependants, create a new parallel task
+			// Multiple providers, create a new parallel task
 			Task par(Task::Mode::PARALLEL);
-			for (auto& d : n->dependants()) {
-				compile_dependant(n, d, par, k);
+			for (auto& d : n->providers()) {
+				compile_provider(n, d, par, k);
 			}
-			task.push_back(par);
+			task.push_front(std::move(par));
 		}
 		n->set_mark(BlockImpl::Mark::VISITED);
 		break;
@@ -221,7 +209,7 @@ CompiledGraph::compile_block(BlockImpl* n, Task& task, std::set<BlockImpl*>& k)
 void
 CompiledGraph::run(RunContext& context)
 {
-	_master.run(context);
+	_master->run(context);
 }
 
 void
@@ -229,7 +217,7 @@ CompiledGraph::dump(std::function<void (const std::string&)> sink) const
 {
 	sink("(compiled-graph ");
 	sink(_path);
-	_master.dump(sink, 2, false);
+	_master->dump(sink, 2, false);
 	sink(")\n");
 }
 
diff --git a/src/server/CompiledGraph.hpp b/src/server/CompiledGraph.hpp
index eeeb6111..db9aa288 100644
--- a/src/server/CompiledGraph.hpp
+++ b/src/server/CompiledGraph.hpp
@@ -1,6 +1,6 @@
 /*
   This file is part of Ingen.
-  Copyright 2007-2016 David Robillard <http://drobilla.net/>
+  Copyright 2007-2017 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
@@ -63,14 +63,14 @@ private:
 	void compile_graph(GraphImpl* graph);
 	void compile_set(const BlockSet& blocks, Task& task, BlockSet& k);
 	void compile_block(BlockImpl* block, Task& task, BlockSet& k);
-	void compile_dependant(const BlockImpl* root,
-	                       BlockImpl*       block,
-	                       Task&            task,
-	                       BlockSet&        k);
-
-	Log&             _log;
-	const Raul::Path _path;
-	Task             _master;
+	void compile_provider(const BlockImpl* root,
+	                      BlockImpl*       block,
+	                      Task&            task,
+	                      BlockSet&        k);
+
+	Log&                  _log;
+	const Raul::Path      _path;
+	std::unique_ptr<Task> _master;
 };
 
 } // namespace Server
diff --git a/src/server/Engine.cpp b/src/server/Engine.cpp
index 794088bc..5561c36e 100644
--- a/src/server/Engine.cpp
+++ b/src/server/Engine.cpp
@@ -102,6 +102,7 @@ Engine::Engine(Ingen::World* world)
 		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));
+		_run_contexts.back()->set_priority(90); // FIXME
 	}
 
 	_world->lv2_features().add_feature(_worker->schedule_feature());
diff --git a/src/server/Task.cpp b/src/server/Task.cpp
index e19855d3..bf1f5028 100644
--- a/src/server/Task.cpp
+++ b/src/server/Task.cpp
@@ -1,6 +1,6 @@
 /*
   This file is part of Ingen.
-  Copyright 2015-2016 David Robillard <http://drobilla.net/>
+  Copyright 2015-2017 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
@@ -30,14 +30,14 @@ Task::run(RunContext& context)
 		_block->process(context);
 		break;
 	case Mode::SEQUENTIAL:
-		for (Task& task : *this) {
-			task.run(context);
+		for (const auto& task : _children) {
+			task->run(context);
 		}
 		break;
 	case Mode::PARALLEL:
 		// Initialize (not) done state of sub-tasks
-		for (Task& task : *this) {
-			task.set_done(false);
+		for (const auto& task : _children) {
+			task->set_done(false);
 		}
 
 		// Grab the first sub-task
@@ -65,12 +65,12 @@ Task::steal(RunContext& context)
 {
 	if (_mode == Mode::PARALLEL) {
 		const unsigned i = _next++;
-		if (i < size()) {
-			return &(*this)[i];
+		if (i < _children.size()) {
+			return _children[i].get();
 		}
 	}
 
-	return NULL;
+	return nullptr;
 }
 
 Task*
@@ -82,52 +82,65 @@ Task::get_task(RunContext& context)
 		return t;
 	}
 
+   int num_spins = 0;
 	while (true) {
 		// Push done end index as forward as possible
-		for (; _done_end < size() && (*this)[_done_end].done(); ++_done_end) {}
+		while (_done_end < _children.size() && _children[_done_end]->done()) {
+			++_done_end;
+		}
 
-		if (_done_end >= size()) {
-			return NULL;  // All child tasks are finished
+		if (_done_end >= _children.size()) {
+			return nullptr;  // All child tasks are finished
 		}
 
 		// All child tasks claimed, but some are unfinished, steal a task
-		t = context.engine().steal_task(context.id() + 1);
-		if (t) {
+		if ((t = context.engine().steal_task(context.id() + 1))) {
 			return t;
 		}
 
+      ++num_spins;
 		/* All child tasks are claimed, and we failed to steal any tasks.  Spin
 		   to prevent blocking, though it would probably be wiser to wait for a
 		   signal in non-main threads, and maybe even in the main thread
 		   depending on your real-time safe philosophy... more experimentation
 		   here is needed. */
 	}
+
+   if (num_spins > 0) {
+     fprintf(stderr, "Num spins: %u\n", num_spins);
+   }
 }
 
-void
-Task::simplify()
+std::unique_ptr<Task>
+Task::simplify(std::unique_ptr<Task>&& task)
 {
-	if (_mode != Mode::SINGLE) {
-		for (std::vector<Task>::iterator t = begin(); t != end();) {
-			t->simplify();
-			if (t->mode() != Mode::SINGLE && t->empty()) {
-				// Empty task, erase
-				t = erase(t);
-			} else if (t->mode() == _mode) {
-				// Subtask with the same type, fold child into parent
-				const Task child(*t);
-				t = erase(t);
-				t = insert(t, child.begin(), child.end());
-			} else {
-				++t;
-			}
-		}
+	if (task->mode() == Mode::SINGLE) {
+		return std::move(task);
+	}
 
-		if (size() == 1) {
-			const Task t(front());
-			*this = t;
+	for (auto t = task->_children.begin(); t != task->_children.end();) {
+		*t = simplify(std::move(*t));
+		if ((*t)->empty()) {
+			// Empty task, erase
+			t = task->_children.erase(t);
+		} else if ((*t)->mode() == task->_mode) {
+			// Subtask with the same type, fold child into parent
+			std::unique_ptr<Task> child(std::move(*t));
+			t = task->_children.erase(t);
+			t = task->_children.insert(
+				t,
+				std::make_move_iterator(child->_children.begin()),
+				std::make_move_iterator(child->_children.end()));
+		} else {
+			++t;
 		}
 	}
+
+	if (task->_children.size() == 1) {
+		return std::move(task->_children.front());
+	}
+
+	return std::move(task);
 }
 
 void
@@ -144,8 +157,8 @@ Task::dump(std::function<void (const std::string&)> sink, unsigned indent, bool
 		sink(_block->path());
 	} else {
 		sink(((_mode == Mode::SEQUENTIAL) ? "(seq " : "(par "));
-		for (size_t i = 0; i < size(); ++i) {
-			(*this)[i].dump(sink, indent + 5, i == 0);
+		for (size_t i = 0; i < _children.size(); ++i) {
+			_children[i]->dump(sink, indent + 5, i == 0);
 		}
 		sink(")");
 	}
diff --git a/src/server/Task.hpp b/src/server/Task.hpp
index 2c1a0cc2..982a6206 100644
--- a/src/server/Task.hpp
+++ b/src/server/Task.hpp
@@ -1,6 +1,6 @@
 /*
   This file is part of Ingen.
-  Copyright 2007-2016 David Robillard <http://drobilla.net/>
+  Copyright 2007-2017 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
@@ -18,8 +18,9 @@
 #define INGEN_ENGINE_TASK_HPP
 
 #include <cassert>
+#include <deque>
+#include <memory>
 #include <ostream>
-#include <vector>
 
 namespace Ingen {
 namespace Server {
@@ -27,7 +28,7 @@ namespace Server {
 class BlockImpl;
 class RunContext;
 
-class Task : public std::vector<Task> {
+class Task {
 public:
 	enum class Mode {
 		SINGLE,      ///< Single block to run
@@ -35,7 +36,7 @@ public:
 		PARALLEL     ///< Elements may be run in any order in parallel
 	};
 
-	Task(Mode mode, BlockImpl* block=NULL)
+	Task(Mode mode, BlockImpl* block = nullptr)
 		: _block(block)
 		, _mode(mode)
 		, _done_end(0)
@@ -45,23 +46,13 @@ public:
 		assert(!(mode == Mode::SINGLE && !block));
 	}
 
-	Task& operator=(const Task& copy) {
-		*static_cast<std::vector<Task>*>(this) = copy;
-		_block    = copy._block;
-		_mode     = copy._mode;
-		_done_end = copy._done_end;
-		_next     = copy._next.load();
-		_done     = copy._done.load();
-		return *this;
-	}
-
-	Task(const Task& copy)
-		: std::vector<Task>(copy)
-		, _block(copy._block)
-		, _mode(copy._mode)
-		, _done_end(copy._done_end)
-		, _next(copy._next.load())
-		, _done(copy._done.load())
+	Task(Task&& task)
+		: _children(std::move(task._children))
+		, _block(task._block)
+		, _mode(task._mode)
+		, _done_end(task._done_end)
+		, _next(task._next.load())
+		, _done(task._done.load())
 	{}
 
 	/** Run task in the given context. */
@@ -70,12 +61,20 @@ public:
 	/** Pretty print task to the given stream (recursively). */
 	void dump(std::function<void (const std::string&)> sink, unsigned indent, bool first) const;
 
+	/** Return true iff this is an empty task. */
+	bool empty() const { return _mode != Mode::SINGLE && _children.empty(); }
+
 	/** Simplify task expression. */
-	void simplify();
+	static std::unique_ptr<Task> simplify(std::unique_ptr<Task>&& task);
 
 	/** Steal a child task from this task (succeeds for PARALLEL only). */
 	Task* steal(RunContext& context);
 
+	/** Prepend a child to this task. */
+	void push_front(Task&& task) {
+		_children.push_front(std::unique_ptr<Task>(new Task(std::move(task))));
+	}
+
 	Mode       mode()  const { return _mode; }
 	BlockImpl* block() const { return _block; }
 	bool       done()  const { return _done; }
@@ -83,8 +82,14 @@ public:
 	void set_done(bool done) { _done = done; }
 
 private:
+	typedef std::deque<std::unique_ptr<Task>> Children;
+
+	Task(const Task&) = delete;
+	Task& operator=(const Task&) = delete;
+
 	Task* get_task(RunContext& context);
 
+	Children              _children;  ///< Vector of child tasks
 	BlockImpl*            _block;     ///< Used for SINGLE only
 	Mode                  _mode;      ///< Execution mode
 	unsigned              _done_end;  ///< Index of rightmost done sub-task
diff --git a/src/server/ingen_lv2.cpp b/src/server/ingen_lv2.cpp
index 9119ae14..00b59ca7 100644
--- a/src/server/ingen_lv2.cpp
+++ b/src/server/ingen_lv2.cpp
@@ -507,9 +507,9 @@ ingen_instantiate(const LV2_Descriptor*    descriptor,
 	}
 
 	IngenPlugin* plugin = new IngenPlugin();
-	plugin->map = map;
-	plugin->world = new Ingen::World(
-		plugin->argc, plugin->argv, map, unmap, log);
+	plugin->map   = map;
+	plugin->world = new Ingen::World(map, unmap, log);
+	plugin->world->load_configuration(plugin->argc, plugin->argv);
 
 	LV2_URID bufsz_max    = map->map(map->handle, LV2_BUF_SIZE__maxBlockLength);
 	LV2_URID bufsz_seq    = map->map(map->handle, LV2_BUF_SIZE__sequenceSize);