Add parallelism-aware graph traversal

19 files changed, 433 insertions, 120 deletions

diff --git a/ingen/Status.hpp b/ingen/Status.hpp
index 88e9d707..c1002a17 100644
--- a/ingen/Status.hpp
+++ b/ingen/Status.hpp
@@ -46,7 +46,8 @@ enum class Status {
 	PROTOTYPE_NOT_FOUND,
 	PORT_NOT_FOUND,
 	TYPE_MISMATCH,
-	UNKNOWN_TYPE
+	UNKNOWN_TYPE,
+	COMPILATION_FAILED
 };
 
 static inline const char*
@@ -80,6 +81,7 @@ ingen_status_string(Status st)
 	case Status::PORT_NOT_FOUND:      return "Port not found";
 	case Status::TYPE_MISMATCH:       return "Type mismatch";
 	case Status::UNKNOWN_TYPE:        return "Unknown type";
+	case Status::COMPILATION_FAILED:  return "Graph compilation failed";
 	}
 
 	return "Unknown error";
diff --git a/src/Configuration.cpp b/src/Configuration.cpp
index 5fbf551f..13e2d722 100644
--- a/src/Configuration.cpp
+++ b/src/Configuration.cpp
@@ -50,7 +50,7 @@ Configuration::Configuration(Forge& forge)
 	add("atomicBundles",  "atomic-bundles", 'a', "Execute bundles atomically", GLOBAL, forge.Bool, forge.make(false));
 	add("clientPort",     "client-port",    'C', "Client port", GLOBAL, forge.Int, Atom());
 	add("connect",        "connect",        'c', "Connect to engine URI", SESSION, forge.String, forge.alloc("unix:///tmp/ingen.sock"));
-	add("engine",         "engine",         'e', "Run (JACK) engine", GLOBAL, forge.Bool, forge.make(false));
+	add("engine",         "engine",         'e', "Run (JACK) engine", SESSION, forge.Bool, forge.make(false));
 	add("enginePort",     "engine-port",    'E', "Engine listen port", GLOBAL, forge.Int, forge.make(16180));
 	add("socket",         "socket",         'S', "Engine socket path", GLOBAL, forge.String, forge.alloc("/tmp/ingen.sock"));
 	add("gui",            "gui",            'g', "Launch the GTK graphical interface", SESSION, forge.Bool, forge.make(false));
@@ -66,7 +66,7 @@ Configuration::Configuration(Forge& forge)
 	add("path",           "path",           'L', "Target path for loaded graph", SESSION, forge.String, Atom());
 	add("queueSize",      "queue-size",     'q', "Event queue size", GLOBAL, forge.Int, forge.make(4096));
 	add("flushLog",       "flush-log",      'f', "Flush logs after every entry", GLOBAL, forge.Bool, forge.make(false));
-	add("dump",           "dump",           'd', "Dump communication", GLOBAL, forge.Bool, forge.make(false));
+	add("dump",           "dump",           'd', "Print debug output", GLOBAL, forge.Bool, forge.make(false));
 	add("trace",          "trace",          't', "Show LV2 plugin trace messages", GLOBAL, forge.Bool, forge.make(false));
 	add("humanNames",     "human-names",     0,  "Show human names in GUI", GUI, forge.Bool, forge.make(true));
 	add("portLabels",     "port-labels",     0,  "Show port labels in GUI", GUI, forge.Bool, forge.make(true));
diff --git a/src/server/BlockImpl.cpp b/src/server/BlockImpl.cpp
index 02611137..d3a5b02d 100644
--- a/src/server/BlockImpl.cpp
+++ b/src/server/BlockImpl.cpp
@@ -42,10 +42,10 @@ BlockImpl::BlockImpl(PluginImpl*         plugin,
 	, _plugin(plugin)
 	, _ports(NULL)
 	, _polyphony((polyphonic && parent) ? parent->internal_poly() : 1)
+	, _mark(Mark::UNVISITED)
 	, _polyphonic(polyphonic)
 	, _activated(false)
 	, _enabled(true)
-	, _traversed(false)
 {
 	assert(_plugin);
 	assert(_polyphony > 0);
diff --git a/src/server/BlockImpl.hpp b/src/server/BlockImpl.hpp
index 47eaa6eb..2d7211ab 100644
--- a/src/server/BlockImpl.hpp
+++ b/src/server/BlockImpl.hpp
@@ -178,9 +178,10 @@ public:
 	uint32_t         num_ports() const { return _ports ? _ports->size() : 0; }
 	virtual uint32_t polyphony() const { return _polyphony; }
 
-	/** Used by the process order finding algorithm (ie during connections) */
-	bool traversed() const { return _traversed; }
-	void traversed(bool b) { _traversed = b; }
+	/** Mark used during graph compilation */
+	enum class Mark { UNVISITED, VISITING, VISITED };
+	Mark get_mark() const { return _mark; }
+	void set_mark(Mark m) { _mark = m; }
 
 protected:
 	PortImpl* nth_port_by_type(uint32_t n, bool input, PortType type);
@@ -190,10 +191,10 @@ protected:
 	uint32_t                _polyphony;
 	std::set<BlockImpl*>    _providers; ///< Blocks connected to this one's input ports
 	std::set<BlockImpl*>    _dependants; ///< Blocks this one's output ports are connected to
+	Mark                    _mark; ///< Mark for graph compilation algorithm
 	bool                    _polyphonic;
 	bool                    _activated;
 	bool                    _enabled;
-	bool                    _traversed; ///< Flag for process order algorithm
 };
 
 } // namespace Server
diff --git a/src/server/CompiledGraph.cpp b/src/server/CompiledGraph.cpp
new file mode 100644
index 00000000..21bd5337
--- /dev/null
+++ b/src/server/CompiledGraph.cpp
@@ -0,0 +1,311 @@
+/*
+  This file is part of Ingen.
+  Copyright 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 <algorithm>
+
+#include "ingen/Configuration.hpp"
+#include "ingen/Log.hpp"
+
+#include "CompiledGraph.hpp"
+#include "Engine.hpp"
+#include "GraphImpl.hpp"
+#include "ThreadManager.hpp"
+
+namespace Ingen {
+namespace Server {
+
+/** Graph contains ambiguous feedback with no delay nodes. */
+class FeedbackException : public std::exception {
+public:
+	FeedbackException(const BlockImpl* node, const BlockImpl* root=NULL)
+		: node(node)
+		, root(root)
+	{}
+
+	const BlockImpl* node;
+	const BlockImpl* root;
+};
+
+CompiledGraph::CompiledGraph(GraphImpl* graph)
+	: _log(graph->engine().log())
+	, _path(graph->path())
+	, _master(Task::Mode::SEQUENTIAL)
+{
+	compile_graph(graph);
+}
+
+CompiledGraph*
+CompiledGraph::compile(GraphImpl* graph)
+{
+	try {
+		return new CompiledGraph(graph);
+	} catch (FeedbackException e) {
+		Log& log = graph->engine().log();
+		if (e.node && e.root) {
+			log.error(fmt("Feedback compiling %1% from %2%\n")
+			          % e.node->path() % e.root->path());
+		} else {
+			log.error(fmt("Feedback compiling %1%\n")
+			          % e.node->path());
+		}
+		return NULL;
+	}
+}
+
+void
+CompiledGraph::compile_set(const std::set<BlockImpl*>& blocks,
+                           Task&                       task,
+                           std::set<BlockImpl*>&       k)
+{
+	// Keep compiling working set until all nodes are visited
+	for (BlockImpl* block : 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);
+	}
+}
+
+void
+CompiledGraph::compile_graph(GraphImpl* graph)
+{
+	ThreadManager::assert_thread(THREAD_PRE_PROCESS);
+
+	// Start with sink nodes (no outputs, or connected only to graph outputs)
+	std::set<BlockImpl*> blocks;
+	for (auto& b : graph->blocks()) {
+		// 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
+			_log.info(fmt("Initial: %1%\n") % b.path());
+			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();
+		// 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
+			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);
+		}
+	}
+
+	_master.simplify();
+
+	if (graph->engine().world()->conf().option("dump").get<int32_t>()) {
+		dump(std::cout);
+	}
+}
+
+/** Throw a FeedbackException iff `dependant` has `root` as a dependency. */
+static void
+check_feedback(const BlockImpl* root, BlockImpl* dependant)
+{
+	if (dependant == root) {
+		throw FeedbackException(root);
+	}
+
+	for (auto& d : dependant->dependants()) {
+		const BlockImpl::Mark mark = d->get_mark();
+		switch (mark) {
+		case BlockImpl::Mark::UNVISITED:
+			d->set_mark(BlockImpl::Mark::VISITING);
+			check_feedback(root, d);
+			break;
+		case BlockImpl::Mark::VISITING:
+			throw FeedbackException(d, root);
+		case BlockImpl::Mark::VISITED:
+			break;
+		}
+		d->set_mark(mark);
+	}
+}
+
+void
+CompiledGraph::compile_dependant(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. */
+		check_feedback(root, block);
+		k.insert(block);
+	} else {
+		// Dependant has only this provider, 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);
+		} else {
+			// Append to given sequential task
+			compile_block(block, task, k);
+		}
+	}
+}
+
+void
+CompiledGraph::compile_block(BlockImpl* n, Task& task, std::set<BlockImpl*>& k)
+{
+	static unsigned indent = 0;
+
+	switch (n->get_mark()) {
+	case BlockImpl::Mark::UNVISITED:
+		indent += 4;
+		for (unsigned i = 0; i < indent; ++i) {
+			_log.info(" ");
+		}
+
+		_log.info(fmt("Compile block %1% (%2% dependants, %3% providers) {\n")
+		          % n->path() % n->dependants().size() % n->providers().size());
+
+		n->set_mark(BlockImpl::Mark::VISITING);
+
+		// Execute this task before the dependants to follow
+		task.push_back(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);
+			}
+		} else {
+			// Multiple dependants, create a new parallel task
+			Task par(Task::Mode::PARALLEL);
+			for (auto& d : n->dependants()) {
+				compile_dependant(n, d, par, k);
+			}
+			task.push_back(par);
+		}
+		n->set_mark(BlockImpl::Mark::VISITED);
+		for (unsigned i = 0; i < indent; ++i) {
+			_log.info(" ");
+		}
+		_log.info("} " + n->path() + "\n");
+		indent -= 4;
+		break;
+
+	case BlockImpl::Mark::VISITING:
+		throw FeedbackException(n);
+
+	case BlockImpl::Mark::VISITED:
+		break;
+	}
+}
+
+void
+CompiledGraph::run(RunContext& context)
+{
+	_master.run(context);
+}
+
+void
+CompiledGraph::dump(std::ostream& os) const
+{
+	os << "(compiled-graph " << _path;
+	_master.dump(os, 2, false);
+	os << ")" << std::endl;
+}
+
+void
+CompiledGraph::Task::run(RunContext& context)
+{
+	switch (_mode) {
+	case Mode::SINGLE:
+		_block->process(context);
+		break;
+	case Mode::SEQUENTIAL:
+	case Mode::PARALLEL:
+		for (Task& task : *this) {
+			task.run(context);
+		}
+		break;
+	}
+}
+
+void
+CompiledGraph::Task::simplify()
+{
+	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 (size() == 1) {
+			const Task t(front());
+			*this = t;
+		}
+	}
+}
+
+void
+CompiledGraph::Task::dump(std::ostream& os, unsigned indent, bool first) const
+{
+	if (!first) {
+		os << std::endl;
+		for (unsigned i = 0; i < indent; ++i) {
+			os << " ";
+		}
+	}
+
+	if (_mode == Mode::SINGLE) {
+		os << _block->path();
+	} else {
+		os << ((_mode == Mode::SEQUENTIAL) ? "(seq " : "(par ");
+		for (size_t i = 0; i < size(); ++i) {
+			(*this)[i].dump(os, indent + 5, i == 0);
+		}
+		os << ")";
+	}
+}
+
+} // namespace Server
+} // namespace Ingen
diff --git a/src/server/CompiledGraph.hpp b/src/server/CompiledGraph.hpp
index 9f4071a5..663752e3 100644
--- a/src/server/CompiledGraph.hpp
+++ b/src/server/CompiledGraph.hpp
@@ -17,41 +17,81 @@
 #ifndef INGEN_ENGINE_COMPILEDGRAPH_HPP
 #define INGEN_ENGINE_COMPILEDGRAPH_HPP
 
+#include <ostream>
+#include <set>
 #include <vector>
-#include <list>
 
 #include "raul/Maid.hpp"
 #include "raul/Noncopyable.hpp"
+#include "raul/Path.hpp"
 
 namespace Ingen {
+
+class Log;
+
 namespace Server {
 
 class BlockImpl;
+class GraphImpl;
+class RunContext;
 
-/** All information required about a block to execute it in an audio thread.
+/** A graph ``compiled'' into a quickly executable form.
+ *
+ * This is a flat sequence of nodes ordered such that the process thread can
+ * execute the nodes in order and have nodes always executed before any of
+ * their dependencies.
  */
-class CompiledBlock {
+class CompiledGraph : public Raul::Maid::Disposable
+                    , public Raul::Noncopyable
+{
 public:
-	CompiledBlock(BlockImpl* b) : _block(b) {}
+	class Task : public std::vector<Task> {
+	public:
+		enum class Mode {
+			SINGLE,      ///< Single block to run
+			SEQUENTIAL,  ///< Elements must be run sequentially in order
+			PARALLEL     ///< Elements may be run in any order in parallel
+		};
+
+		Task(Mode mode, BlockImpl* block=NULL)
+			: _mode(mode)
+			, _block(block)
+		{}
+
+		void run(RunContext& context);
+		void dump(std::ostream& os, unsigned indent, bool first) const;
+		void simplify();
+
+		Mode       mode()  const { return _mode; }
+		BlockImpl* block() const { return _block; }
+
+	private:
+		Mode       _mode;   ///< Execution mode
+		BlockImpl* _block;  ///< Used for SINGLE only
+	};
 
-	BlockImpl* block() const { return _block; }
+	static CompiledGraph* compile(GraphImpl* graph);
+
+	void run(RunContext& context);
+
+	void dump(std::ostream& os) const;
 
 private:
-	BlockImpl* _block;
-};
+	CompiledGraph(GraphImpl* graph);
 
-/** A graph ``compiled'' into a flat structure with the correct order so
- * the audio thread(s) can execute it without threading problems (since
- * the preprocessor thread modifies the graph).
- *
- * The blocks contained here are sorted in the order they must be executed.
- * The parallel processing algorithm guarantees no block will be executed
- * before its providers, using this order as well as semaphores.
- */
-class CompiledGraph : public std::vector<CompiledBlock>
-                    , public Raul::Maid::Disposable
-                    , public Raul::Noncopyable
-{
+	typedef std::set<BlockImpl*> BlockSet;
+
+	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;
 };
 
 } // namespace Server
diff --git a/src/server/GraphImpl.cpp b/src/server/GraphImpl.cpp
index 86966b7c..22276a34 100644
--- a/src/server/GraphImpl.cpp
+++ b/src/server/GraphImpl.cpp
@@ -233,11 +233,8 @@ GraphImpl::process(RunContext& context)
 void
 GraphImpl::run(RunContext& context)
 {
-	if (_compiled_graph && _compiled_graph->size() > 0) {
-		// Run all blocks
-		for (size_t i = 0; i < _compiled_graph->size(); ++i) {
-			(*_compiled_graph)[i].block()->process(context);
-		}
+	if (_compiled_graph) {
+		_compiled_graph->run(context);
 	}
 }
 
@@ -250,10 +247,11 @@ GraphImpl::set_buffer_size(RunContext&    context,
 	BlockImpl::set_buffer_size(context, bufs, type, size);
 
 	if (_compiled_graph) {
-		for (size_t i = 0; i < _compiled_graph->size(); ++i) {
-			const CompiledBlock& block = (*_compiled_graph)[i];
-			block.block()->set_buffer_size(context, bufs, type, size);
-		}
+		// FIXME
+		// for (size_t i = 0; i < _compiled_graph->size(); ++i) {
+		// 	const CompiledBlock& block = (*_compiled_graph)[i];
+		// 	block.block()->set_buffer_size(context, bufs, type, size);
+		// }
 	}
 }
 
@@ -353,55 +351,5 @@ GraphImpl::build_ports_array()
 	return result;
 }
 
-static inline void
-compile_recursive(BlockImpl* n, CompiledGraph* output)
-{
-	if (n == NULL || n->traversed())
-		return;
-
-	n->traversed(true);
-	assert(output != NULL);
-
-	for (auto& p : n->providers())
-		if (!p->traversed())
-			compile_recursive(p, output);
-
-	output->push_back(CompiledBlock(n));
-}
-
-CompiledGraph*
-GraphImpl::compile()
-{
-	ThreadManager::assert_thread(THREAD_PRE_PROCESS);
-
-	CompiledGraph* const compiled_graph = new CompiledGraph();
-
-	for (auto& b : _blocks) {
-		b.traversed(false);
-	}
-
-	for (auto& b : _blocks) {
-		// Either a sink or connected to our output ports:
-		if (!b.traversed() && b.dependants().empty()) {
-			compile_recursive(&b, compiled_graph);
-		}
-	}
-
-	// Traverse any blocks we didn't hit yet
-	for (auto& b : _blocks) {
-		if (!b.traversed()) {
-			compile_recursive(&b, compiled_graph);
-		}
-	}
-
-	if (compiled_graph->size() != _blocks.size()) {
-		_engine.log().error(fmt("Failed to compile graph %1%\n") % _path);
-		delete compiled_graph;
-		return NULL;
-	}
-
-	return compiled_graph;
-}
-
 } // namespace Server
 } // namespace Ingen
diff --git a/src/server/GraphImpl.hpp b/src/server/GraphImpl.hpp
index 7352da39..124614de 100644
--- a/src/server/GraphImpl.hpp
+++ b/src/server/GraphImpl.hpp
@@ -165,16 +165,6 @@ public:
 	Raul::Array<PortImpl*>* external_ports()                           { return _ports; }
 	void                    external_ports(Raul::Array<PortImpl*>* pa) { _ports = pa; }
 
-	/** Compile the graph into a version suitable for real-time execution.
-	 *
-	 * The CompiledGraph is a flat list that the graph will execute in order
-	 * when its run() method is called.  The returned object is newly allocated
-	 * and owned by the caller.  This function is non-realtime and does not
-	 * affect processing, to take effect the returned object must be installed
-	 * in the audio thread with set_compiled_graph().
-	 */
-	CompiledGraph* compile();
-
 	Raul::Array<PortImpl*>* build_ports_array();
 
 	/** Whether to run this graph's DSP bits in the audio thread */
diff --git a/src/server/events/Connect.cpp b/src/server/events/Connect.cpp
index 66f726a6..57bfb975 100644
--- a/src/server/events/Connect.cpp
+++ b/src/server/events/Connect.cpp
@@ -27,6 +27,7 @@
 #include "OutputPort.hpp"
 #include "PortImpl.hpp"
 #include "types.hpp"
+#include "internals/BlockDelay.hpp"
 
 namespace Ingen {
 namespace Server {
@@ -111,19 +112,31 @@ Connect::pre_process()
 	   provider...
 	*/
 	if (tail_block != head_block && tail_block->parent() == head_block->parent()) {
+		// Connection is between blocks inside a graph, compile graph
+
 		// The tail block is now a dependency (provider) of the head block
 		head_block->providers().insert(tail_block);
-		tail_block->dependants().insert(head_block);
+
+		if (!dynamic_cast<Internals::BlockDelayNode*>(tail_block)) {
+			/* Arcs leaving a delay node are ignored for the purposes of
+			   compilation, since the output is from the previous cycle and
+			   does not affect execution order.  Otherwise, the head block is
+			   now a dependant of the head block. */
+			tail_block->dependants().insert(head_block);
+		}
+
+		if (_graph->enabled()) {
+			if (!(_compiled_graph = CompiledGraph::compile(_graph))) {
+				head_block->providers().erase(tail_block);
+				tail_block->dependants().erase(head_block);
+				return Event::pre_process_done(Status::COMPILATION_FAILED);
+			}
+		}
 	}
 
 	_graph->add_arc(_arc);
 	_head->increment_num_arcs();
 
-	tail_output->inherit_neighbour(_head, _tail_remove, _tail_add);
-	_head->inherit_neighbour(tail_output, _head_remove, _head_add);
-
-	lock.unlock();
-
 	if (!_head->is_driver_port()) {
 		_voices = new Raul::Array<PortImpl::Voice>(_head->poly());
 		_head->get_buffers(*_engine.buffer_factory(),
@@ -132,9 +145,8 @@ Connect::pre_process()
 		                   false);
 	}
 
-	if (_graph->enabled()) {
-		_compiled_graph = _graph->compile();
-	}
+	tail_output->inherit_neighbour(_head, _tail_remove, _tail_add);
+	_head->inherit_neighbour(tail_output, _head_remove, _head_add);
 
 	return Event::pre_process_done(Status::SUCCESS);
 }
@@ -148,7 +160,9 @@ Connect::execute(RunContext& context)
 			_engine.maid()->dispose(_head->set_voices(context, _voices));
 		}
 		_head->connect_buffers();
-		_graph->set_compiled_graph(_compiled_graph);
+		if (_compiled_graph) {
+			_graph->set_compiled_graph(_compiled_graph);
+		}
 	}
 }
 
diff --git a/src/server/events/Copy.cpp b/src/server/events/Copy.cpp
index 0ff53843..04c77316 100644
--- a/src/server/events/Copy.cpp
+++ b/src/server/events/Copy.cpp
@@ -126,7 +126,7 @@ Copy::engine_to_engine()
 
 	// Compile graph with new block added for insertion in audio thread
 	if (_parent->enabled()) {
-		_compiled_graph = _parent->compile();
+		_compiled_graph = CompiledGraph::compile(_parent);
 	}
 
 	return Event::pre_process_done(Status::SUCCESS);
diff --git a/src/server/events/CreateBlock.cpp b/src/server/events/CreateBlock.cpp
index 28afe4b2..d1060fa4 100644
--- a/src/server/events/CreateBlock.cpp
+++ b/src/server/events/CreateBlock.cpp
@@ -138,7 +138,7 @@ CreateBlock::pre_process()
 	   TODO: Since the block is not connected at this point, a full compilation
 	   could be avoided and the block simply appended. */
 	if (_graph->enabled()) {
-		_compiled_graph = _graph->compile();
+		_compiled_graph = CompiledGraph::compile(_graph);
 	}
 
 	_update.put_block(_block);
@@ -149,7 +149,7 @@ CreateBlock::pre_process()
 void
 CreateBlock::execute(RunContext& context)
 {
-	if (_block) {
+	if (_status == Status::SUCCESS) {
 		_graph->set_compiled_graph(_compiled_graph);
 		_compiled_graph = NULL;  // Graph takes ownership
 	}
diff --git a/src/server/events/CreateGraph.cpp b/src/server/events/CreateGraph.cpp
index ca36f258..a2e4e6c4 100644
--- a/src/server/events/CreateGraph.cpp
+++ b/src/server/events/CreateGraph.cpp
@@ -163,7 +163,7 @@ CreateGraph::pre_process()
 		_parent->add_block(*_graph);
 		if (_parent->enabled()) {
 			_graph->enable();
-			_compiled_graph = _parent->compile();
+			_compiled_graph = CompiledGraph::compile(_parent);
 		}
 	}
 
diff --git a/src/server/events/Delete.cpp b/src/server/events/Delete.cpp
index 5ca70a3a..12c9a3f1 100644
--- a/src/server/events/Delete.cpp
+++ b/src/server/events/Delete.cpp
@@ -94,7 +94,7 @@ Delete::pre_process()
 		_disconnect_event->pre_process();
 
 		if (parent->enabled()) {
-			_compiled_graph = parent->compile();
+			_compiled_graph = CompiledGraph::compile(parent);
 		}
 	} else if (_port) {
 		parent->remove_port(*_port);
@@ -102,7 +102,7 @@ Delete::pre_process()
 		_disconnect_event->pre_process();
 
 		if (parent->enabled()) {
-			_compiled_graph = parent->compile();
+			_compiled_graph = CompiledGraph::compile(parent);
 			_ports_array    = parent->build_ports_array();
 			assert(_ports_array->size() == parent->num_ports_non_rt());
 		}
diff --git a/src/server/events/Delta.cpp b/src/server/events/Delta.cpp
index 49ea27ff..8c94fac7 100644
--- a/src/server/events/Delta.cpp
+++ b/src/server/events/Delta.cpp
@@ -374,8 +374,11 @@ Delta::pre_process()
 					if (value.type() == uris.forge.Bool) {
 						op = SpecialType::ENABLE;
 						// FIXME: defer this until all other metadata has been processed
-						if (value.get<int32_t>() && !_graph->enabled())
-							_compiled_graph = _graph->compile();
+						if (value.get<int32_t>() && !_graph->enabled()) {
+							if (!(_compiled_graph = CompiledGraph::compile(_graph))) {
+								_status = Status::COMPILATION_FAILED;
+							}
+						}
 					} else {
 						_status = Status::BAD_VALUE_TYPE;
 					}
diff --git a/src/server/events/Delta.hpp b/src/server/events/Delta.hpp
index 8b00fd3a..24330199 100644
--- a/src/server/events/Delta.hpp
+++ b/src/server/events/Delta.hpp
@@ -38,7 +38,7 @@ namespace Server {
 class CompiledGraph;
 class Engine;
 class GraphImpl;
-class ProcessContext;
+class RunContext;
 
 namespace Events {
 
diff --git a/src/server/events/Disconnect.cpp b/src/server/events/Disconnect.cpp
index 8d47ac5a..9d1b18a3 100644
--- a/src/server/events/Disconnect.cpp
+++ b/src/server/events/Disconnect.cpp
@@ -167,7 +167,7 @@ Disconnect::pre_process()
 	                 dynamic_cast<InputPort*>(head));
 
 	if (_graph->enabled())
-		_compiled_graph = _graph->compile();
+		_compiled_graph = CompiledGraph::compile(_graph);
 
 	return Event::pre_process_done(Status::SUCCESS);
 }
diff --git a/src/server/events/DisconnectAll.cpp b/src/server/events/DisconnectAll.cpp
index 380aced5..e4866d30 100644
--- a/src/server/events/DisconnectAll.cpp
+++ b/src/server/events/DisconnectAll.cpp
@@ -136,8 +136,11 @@ DisconnectAll::pre_process()
 			                 dynamic_cast<InputPort*>(a->head())));
 	}
 
-	if (!_deleting && _parent->enabled())
-		_compiled_graph = _parent->compile();
+	if (!_deleting && _parent->enabled()) {
+		if (!(_compiled_graph = CompiledGraph::compile(_parent))) {
+			return Event::pre_process_done(Status::COMPILATION_FAILED);
+		}
+	}
 
 	return Event::pre_process_done(Status::SUCCESS);
 }
diff --git a/src/server/internals/BlockDelay.cpp b/src/server/internals/BlockDelay.cpp
index 39bd09b9..3dee2feb 100644
--- a/src/server/internals/BlockDelay.cpp
+++ b/src/server/internals/BlockDelay.cpp
@@ -26,7 +26,7 @@
 #include "InputPort.hpp"
 #include "InternalPlugin.hpp"
 #include "OutputPort.hpp"
-#include "ProcessContext.hpp"
+#include "RunContext.hpp"
 #include "internals/BlockDelay.hpp"
 
 namespace Ingen {
@@ -75,7 +75,7 @@ BlockDelayNode::activate(BufferFactory& bufs)
 }
 
 void
-BlockDelayNode::run(ProcessContext& context)
+BlockDelayNode::run(RunContext& context)
 {
 	// Copy buffer from last cycle to output
 	_out_port->buffer(0)->copy(context, _buffer.get());
diff --git a/src/server/wscript b/src/server/wscript
index 8fda0b2a..fad48330 100644
--- a/src/server/wscript
+++ b/src/server/wscript
@@ -9,6 +9,7 @@ def build(bld):
             Broadcaster.cpp
             Buffer.cpp
             BufferFactory.cpp
+            CompiledGraph.cpp
             ClientUpdate.cpp
             ControlBindings.cpp
             DuplexPort.cpp