Как сделать из Ninja систему распределённой сборки?

Этап 0. Формулировка задачи

• Читаем граф сборки, вычленяем команды компиляции;
• Разбиваем компиляцию на два этапа, препроцессинг и собственно генерацию кода. Последнюю помечаем как возможную к удаленному выполнению;
• Выполняем препроцессинг, считываем результат в память;
• Отправляем препроцессированный файл и команду на генерацию кода на другой хост по сети;
• Выполняем команду кодогенерации, считываем объектный файл и отдаем в качестве ответа по сети;
• Полученный объектный файл сохраняем на диск и выводим в консоль сообщения компилятора.

1. Прототип 1. Программа имитирует компилятор, разделяя команду на 2, и самостоятельно вызывая компилятор.
2. Прототип 2. К этому добавим пересылку команды на компиляцию по сети, без самого файла.
3. Прототип 3. Пройдемся по графу сборки Ninja, выводя потенциально разбиваемые команды.

Этап 1. Разбиваем командную строку

#include <iostream>

#include "InvocationRewriter.hpp"
#include "LocalExecutor.hpp"

int main(int argc, char ** argv)
{
	StringVector args;
	for (int i = 1; i < argc; ++i)
		args.emplace_back(argv[i]);

	InvocationRewriter rewriter;
	StringVector ppArgs, ccArgs; // аргументы для препроцессинга и компиляции соотвественно.
	if (!rewriter.SplitInvocation(args, ppArgs, ccArgs))
	{
		std::cerr << "Usage: -c <filename> -o <filename> \n";
		return 1;
	}

	LocalExecutor localExecutor;
	const std::string cxxExecutable = "/usr/bin/g++"; // предполагаем, что мы работаем под GNU/Linux.
	const auto ppResult = localExecutor.Execute(cxxExecutable, ppArgs);
	if (!ppResult.m_result)
	{
		std::cerr << ppResult.m_output;
		return 1;
	}

	const auto ccResult = localExecutor.Execute(cxxExecutable, ccArgs);
	if (!ccResult.m_result)
	{
		std::cerr << ccResult.m_output;
		return 1;
	}
	// не учтен вариант, что есть стандартный вывод, но результат успешен.

	return 0;
}

#pragma once

#include <string>
#include <vector>
#include <algorithm>

using StringVector = std::vector<std::string>;

class InvocationRewriter
{
public:
	bool SplitInvocation(const StringVector & original,
						 StringVector & preprocessor,
						 StringVector & compilation)
	{
		// Найдем сперва позиции аргументов -c и  -o.
		// Будем считать, что после -c всегда идет имя входного файла, хоть это и не так.
		const auto cIter = std::find(original.cbegin(), original.cend(), "-c");
		const auto oIter = std::find(original.cbegin(), original.cend(), "-o");
		if (cIter == original.cend() || oIter == original.cend())
			return false;

		const auto cIndex = cIter - original.cbegin();
		const auto oIndex = oIter - original.cbegin();
		preprocessor = compilation = original;

		const std::string & inputFilename = original[cIndex + 1];
		preprocessor[oIndex + 1] = "pp_" + inputFilename; // абсолютные имена не поддерживаются
		preprocessor[cIndex] = "-E"; // вместо компиляции - препроцессинг.

		compilation[cIndex + 1] = "pp_" + inputFilename;
		return true;
	}
};

#pragma once

#include <string>
#include <vector>
#include <algorithm>

#include <stdio.h>

using StringVector = std::vector<std::string>;

class LocalExecutor
{
public:
	/// Результат выполнения команды: стандартный вывод + результат
	struct ExecutorResult
	{
		std::string m_output;
		bool m_result = false;
		ExecutorResult(const std::string & output = "", bool result = false)
			: m_output(output), m_result(result) {}
	};

	/// выполняет команду с помощью popen.
	ExecutorResult Execute(const std::string & executable, const StringVector & args)
	{
		std::string cmd = executable;
		for (const auto & arg : args)
			cmd += " " + arg;
		cmd += " 2>&1"; // объединим sterr и stdout.

		FILE * process = popen(cmd.c_str(), "r");
		if (!process)
			return ExecutorResult("Failed to execute:" + cmd);

		ExecutorResult result;
		char buffer[1024];
		while (fgets(buffer, sizeof(buffer)-1, process) != nullptr)
			result.m_output += std::string(buffer);

		result.m_result = pclose(process) == 0;
		return result;
	}
};

• Учитывать и абсолютные имена файлов;
• Использовать одну из библиотек для работы с процессами: Boost.Process, QProcess, или Ninja Subprocess;
• Реализовать поддержку разделения команд для MSVC;
• Сделать API для выполнения команд асинхронным, а выполнение вынести в отдельный поток.

Этап 2. Сетевая подсистема

• Создать сокет нужного типа (TCP) с помощью функции socket();
• После создания, выставить нужные флаги, например неблокирующий режим с помощью setsockopt();
• Получить адрес в нужном формате для BSD сокетов с помощью getaddrinfo();
• Подключиться к TCP-хосту с помощью функции connect(), передав туда подготовленный адрес;
• Вызывать функции read/send для чтения и записи;
• После окончания работы — вызвать close().

• Создаем сокет с помощью функции socket();
• Выставляем опции;
• Вызываем bind() для того, чтобы привязать сокет к определённому адресу (полученному через getaddrinfo)
• Начинаем прослушку порта с помощью вызова listen();
• Входящие соединения примаем функцией accept() — он возвращает нам новый сокет;
• С полученным сокетом выполняем операции read/write;
• Закрываем сокет соединения и сокет прослушки через close().

/// Интерфейс сокета
class IDataSocket
{
public:
	using Ptr = std::shared_ptr<IDataSocket>;
	/// Результаты чтения и запаси. Success- Успешно, TryAgain - данные не были прочитаны либо записаны, Fail - сокет был закрыт.
	enum class WriteState { Success, TryAgain, Fail };
	enum class ReadState { Success, TryAgain, Fail };

public:
	virtual ~IDataSocket() = default;

	/// Подключение к удаленному хосту
	virtual bool Connect () = 0;

	/// Закрытие соединение
	virtual void Disconnect () = 0;

	/// Проверки статуса соединения - подключен; сейчас подключается
	virtual bool IsConnected () const = 0;
	virtual bool IsPending() const = 0;

	/// Читаем данные из сокета в буфер
	virtual ReadState Read(ByteArrayHolder & buffer) = 0;

	/// Пишем данные в сокет.
	virtual WriteState Write(const ByteArrayHolder & buffer, size_t maxBytes = size_t(-1)) = 0;
};

/// интерфейс "слушателя". Он может создавать сокеты при подключении.
class IDataListener
{
public:
	using Ptr = std::shared_ptr<IDataListener>;
	virtual ~IDataListener() = default;

	/// Получение следующего соединения
	virtual IDataSocket::Ptr GetPendingConnection() = 0;

	/// Начало прослушивания порта:
	virtual bool StartListen() = 0;
};

#include <TcpListener.h>

#include <algorithm>
#include <iostream>

#include "LocalExecutor.hpp"

int main()
{
	// Создадим настройки для подключения.
	TcpConnectionParams tcpParams;
	tcpParams.SetPoint(6666, "localhost");

	// Создадим прослушку на порту 6666;
	auto listener = TcpListener::Create(tcpParams);
	IDataSocket::Ptr connection;

	// Дождемся первого входящего соединения;
	while((connection = listener->GetPendingConnection()) == nullptr) ;

	// Подключим соединение и прочитаем все данные.
	connection->Connect();
	ByteArrayHolder incomingBuffer;  //!< просто обертка над std::vector<uint8_t>;
	while (connection->Read(incomingBuffer) == IDataSocket::ReadState::TryAgain) ;

	// Считая, что в качестве данных нам пришла команда, выполним её.
	std::string args((const char*)(incomingBuffer.data()), incomingBuffer.size());
	std::replace(args.begin(), args.end(), '\n', ' ');
	LocalExecutor localExecutor;
	const auto result = localExecutor.Execute("/usr/bin/g++", StringVector(1, args));
	std::string stdOutput = result.m_output;
	if (stdOutput.empty())
		stdOutput = "OK\n"; // небольшой хак - если результат выполнения пустой, отправим хотя бы OK.

	// запишем в подключившийся сокет результат выполнения команды.
	ByteArrayHolder outgoingBuffer;
	std::copy(stdOutput.cbegin(), stdOutput.cend(), std::back_inserter(outgoingBuffer.ref()));
	connection->Write(outgoingBuffer);

	connection->Disconnect();

	// Можно не выходить здесь, а вынести обработку соединений в отдельный поток.
	// А потом и обработку чтения/записи из каждого подключения в отдельный поток.
	return 0;
}

#include <iostream>

#include <TcpSocket.h>

#include "InvocationRewriter.hpp"
#include "LocalExecutor.hpp"

int main(int argc, char ** argv)
{
	StringVector args;
	for (int i = 1; i < argc; ++i)
		args.emplace_back(argv[i]);

	InvocationRewriter rewriter;
	StringVector ppArgs, ccArgs; // аргументы для препроцессинга и компиляции соотвественно.
	if (!rewriter.SplitInvocation(args, ppArgs, ccArgs))
	{
		std::cerr << "Usage: -c <filename> -o <filename> \n";
		return 1;
	}

	LocalExecutor localExecutor;
	const std::string cxxExecutable = "/usr/bin/g++"; // предполагаем, что мы работаем под GNU/Linux.
	const auto ppResult = localExecutor.Execute(cxxExecutable, ppArgs);
	if (!ppResult.m_result)
	{
		std::cerr << ppResult.m_output;
		return 1;
	}

	// Подключимся к серверу на порт 6666
	TcpConnectionParams tcpParams;
	tcpParams.SetPoint(6666, "localhost");
	auto connection = TcpSocket::Create(tcpParams);
	connection->Connect();

	ByteArrayHolder outgoingBuffer;
	for (auto arg : ccArgs)
	{
		arg += " "; // разделим аргументы пробелом и вставим в буфер.
		std::copy(arg.cbegin(), arg.cend(), std::back_inserter(outgoingBuffer.ref()));
	}

	connection->Write(outgoingBuffer);

	ByteArrayHolder incomingBuffer;
	while (connection->Read(incomingBuffer) == IDataSocket::ReadState::TryAgain) ;

	std::string response((const char*)(incomingBuffer.data()), incomingBuffer.size());
	if (response != "OK\n")
	{
		std::cerr << response;
		return 1;
	}

	return 0;
}

• Настоятельно рекомендую использовать одну из существующих сетевых библиотек — Boost.Asio, QTcpSocket (QtNetwork), так же подумать над сериализацией с помощью Protobuf или других подобных
• Реализовать передачу файлов по сети. Скорее всего, придется их разбивать на фрагменты, но будет зависеть от выбранной вами библиотеки.
• Необходимо задуматься об асинхронном API отправки и приема сообщений. Кроме того, желательно его сделать абстрактным и не привязанным к сокетам вообще.

Этап 3. Интеграция с Ninja

• Узел (Node) — это просто файл. Входной (исходники), выходной (объектные файлы) — это все узлы или вершины графа.
• Правило (Rule) — по сути это просто команда с шаблоном аргументов. Например, вызов gcc — правило, а его аргументы — $FLAGS $INCLUDES $DEFINES и еще какие-то общие аргументы.
• Ребро (Edge). Для меня было немного удивительно, но ребро соединяет не два узла, а несколько входных узлов и один выходной, посредством Правила. Вся система сборки основана на том, что последовательно обходит граф, выполняя команды для ребер. Как только все ребра обработаны, проект собран.
• Состояние (State) — это контейнер со всем вышеперечисленным, который система сборки и использует.

 // Limit number of rebuilds, to prevent infinite loops.
  const int kCycleLimit = 100;
  for (int cycle = 1; cycle <= kCycleLimit; ++cycle) {
	NinjaMain ninja(ninja_command, config);

	ManifestParser parser(&ninja.state_, &ninja.disk_interface_,
						  options.dupe_edges_should_err
							  ? kDupeEdgeActionError
							  : kDupeEdgeActionWarn);
	string err;
	if (!parser.Load(options.input_file, &err)) {
	  Error("%s", err.c_str());
	  return 1;
	}
        // граф сборки уже загружен, теперь модифицируем его:
	RewriteStateRules(&ninja.state_); // вот этот вызов

#include "InvocationRewriter.hpp"

// Структура, которая описывает замену правил Ninja.
struct RuleReplace
{
	const Rule* pp;
	const Rule* cc;
	std::string toolId;
	RuleReplace() = default;
	RuleReplace(const Rule* pp_, const Rule* cc_, std::string id) : pp(pp_), cc(cc_), toolId(id) {}
};

void RewriteStateRules(State *state)
{
	// скопируем текущие правила, т.к. будет не очень хорошо, когда мы будем модифицировать этот контейнер.
	const auto rules = state->bindings_.GetRules();
	std::map<const Rule*,  RuleReplace> ruleReplacement;
	InvocationRewriter rewriter;
	// пройдем по всем существующим правилам
	for (const auto & ruleIt : rules)
	{
		const Rule * rule = ruleIt.second;
		const EvalString* command = rule->GetBinding("command");
		if (!command) continue;
		// сформируем команду для нашего rewriter-а.
		std::vector<std::string> originalRule;
		for (const auto & strPair : command->parsed_)
		{
			std::string str = strPair.first;
			if (strPair.second == EvalString::SPECIAL)
				str = '$' + str;
			originalRule.push_back(str);
		}
		// попробуем разделить команду:
		std::vector<std::string> preprocessRule, compileRule;
		if (rewriter.SplitInvocation(originalRule, preprocessRule, compileRule))
		{
			// создадим 2 копии rule - rulePP и ruleCC, заменим их bindings_ на новые команды.
			// занесем новые правила в ruleReplacement (ruleReplacement[rule] = ...)
		}
	}

	const auto paths = state->paths_;
	std::set<Edge*> erasedEdges;
	// пройдем по всем узлам графа
	for (const auto & iter : paths)
	{
		Node* node = iter.second;
		Edge* in_egde = node->in_edge();
		if (!in_egde)
			continue;
		// получим входное ребро и соответствующее ему правило.
		// если правило необходимо разбить на два, сделаем это:
		const Rule * in_rule = &(in_egde->rule());
		auto replacementIt = ruleReplacement.find(in_rule);
		if (replacementIt != ruleReplacement.end())
		{
			RuleReplace replacement = replacementIt->second;
			const std::string objectPath = node->path();
			const std::string sourcePath = in_egde->inputs_[0]->path();
			const std::string ppPath = sourcePath + ".pp"; // лучше сделать отдельный метод для имен файлов.
			Node *pp_node = state->GetNode(ppPath, node->slash_bits());

			// Создадим два новых ребра
			Edge* edge_pp = state->AddEdge(replacement.pp);
			Edge* edge_cc = state->AddEdge(replacement.cc);

			// ... код пропущен ...
			// поместим входы исходного ребра во входы edge_pp;
			// выходы исходного ребра в выходы edge_cc
			// а в середине вставим pp_node.

			// кроме того, особо отметим edge_cc, что может выполняться удалённо -
			// например, добавив поле:
			edge_cc->is_remote_ = true;

			// после всех манипуляций, очистим исходное ребро.
			in_egde->outputs_.clear();
			in_egde->inputs_.clear();
			in_egde->env_ = nullptr;
			erasedEdges.insert(in_egde);
		}
	}
	// удалим лишние ребра.
	vector<Edge*> newEdges;
	for (auto * edge : state->edges_)
	{
		if (erasedEdges.find(edge) == erasedEdges.end())
			newEdges.push_back(edge);
	}
	state->edges_ = newEdges;
}

• Скорее всего, первый вариант InvocationRewriter не заработает, нужно будет учитывать много вещей — например, то что аргумент компиляции "-c" может быть задан " -c ", ну и я уже молчу про то что он не обязательно предваряет исходный файл.
• Может быть много дополнительных флагов, которые отмечают какие-то файлы, так что не всё то, что «не флаг» — это файл.
• После создания разделённого графа, если он успешно собирается в две фазы «препроцессинг и компиляция» — нужно будет проинтегрировать удаленное выполнение по сети с нашим сетевым слоем. Собственно цикл сборки в Ninja находится в build.cc в функции Builder::Build. В нее можно добавить по аналогии с
  «if (failures_allowed && command_runner_->CanRunMore())» и «if (pending_commands)» свои этапы для распределённой сборки.

Этап X. Что дальше?

• Конфигурирование всех модулей — как сетевой подсистемы, так и InvocationRewriter-а;
• Поддержка любых комбинаций опций под разными компиляторами;
• Поддержка сжатия при передаче файлов;
• Разнообразная диагностика в виде логов;
• Написание координатора, который сможет обслуживать подключение к нескольким серверам сборки;
• Написание балансировщика, который будет учитывать то, что серверами пользуется сразу несколько клиентов (и не перегружать их сверх меры);
• Написать интеграцию с другими системами сборки, не только Ninja.

• Распределённая сборка с возможностью кросс-компиляции (Clang) под Win, Mac, Linux;
• Интеграция с Ninja и Make.