VILLASnode/lib/node.cpp

/** Nodes.
 *
 * @author Steffen Vogel <post@steffenvogel.de>
 * @copyright 2014-2022, Institute for Automation of Complex Power Systems, EONERC
 * @license Apache 2.0
 *********************************************************************************/

#include <regex>
#include <cstring>
#include <cctype>
#include <openssl/md5.h>

#ifdef __linux__
  #include <sys/types.h>
  #include <sys/socket.h>
#endif

#include <villas/node/config.hpp>
#include <villas/hook.hpp>
#include <villas/hook_list.hpp>
#include <villas/sample.hpp>
#include <villas/node.hpp>
#include <villas/node_list.hpp>
#include <villas/path.hpp>
#include <villas/utils.hpp>
#include <villas/uuid.hpp>
#include <villas/colors.hpp>
#include <villas/mapping.hpp>
#include <villas/timing.hpp>
#include <villas/signal.hpp>
#include <villas/node/memory.hpp>

#ifdef WITH_NETEM
  #include <villas/kernel/if.hpp>
  #include <villas/kernel/nl.hpp>
  #include <villas/kernel/tc.hpp>
  #include <villas/kernel/tc_netem.hpp>
#endif /* WITH_NETEM */

using namespace villas;
using namespace villas::node;
using namespace villas::utils;

Node::Node(const uuid_t &id, const std::string &name) :
	logger(logging.get("node")),
	sequence_init(0),
	sequence(0),
	in(NodeDirection::Direction::IN, this),
	out(NodeDirection::Direction::OUT, this),
#ifdef __linux__
	fwmark(-1),
#endif /* __linux__ */
#ifdef WITH_NETEM
	tc_qdisc(nullptr),
	tc_classifier(nullptr),
#endif /* WITH_NETEM */
	state(State::INITIALIZED),
	enabled(true),
	config(nullptr),
	name_short(name),
	affinity(-1), /* all cores */
	factory(nullptr)
{
	if (uuid_is_null(id)) {
		uuid_generate_random(uuid);
	} else {
		uuid_copy(uuid, id);
	}

	if (!name_short.empty()) {
		name_long = fmt::format(CLR_RED("{}"), name_short);
	}
	else if (name_short.empty()) {
		name_short = "<none>";
		name_long = CLR_RED("<none>");
	}
}

Node::~Node()
{
#ifdef WITH_NETEM
	rtnl_qdisc_put(tc_qdisc);
	rtnl_cls_put(tc_classifier);
#endif /* WITH_NETEM */

	factory->instances.remove(this);
}

int Node::prepare()
{
	int ret;

	ret = in.prepare();
	if (ret)
		return ret;

	ret = out.prepare();
	if (ret)
		return ret;

	state = State::PREPARED;

	return 0;
}

int Node::parse(json_t *json)
{
	assert(state == State::INITIALIZED ||
	       state == State::PARSED ||
	       state == State::CHECKED);

	int ret, en = enabled, init_seq = -1;

	json_error_t err;
	json_t *json_netem = nullptr;

	ret = json_unpack_ex(json, &err, 0, "{ s?: b, s?: i }",
		"enabled", &en,
		"initial_sequenceno", &init_seq
	);
	if (ret)
		return ret;

	if (init_seq >= 0)
		sequence_init = init_seq;

#ifdef __linux__
	ret = json_unpack_ex(json, &err, 0, "{ s?: { s?: o, s?: i } }",
		"out",
			"netem", &json_netem,
			"fwmark", &fwmark
	);
	if (ret)
		return ret;
#endif /* __linux__ */

	enabled = en;

	if (json_netem) {
#ifdef WITH_NETEM
		int enabled = 1;

		ret = json_unpack_ex(json_netem, &err, 0, "{ s?: b }",  "enabled", &enabled);
		if (ret)
			return ret;

		if (enabled)
			kernel::tc::netem_parse(&tc_qdisc, json_netem);
		else
			tc_qdisc = nullptr;
#endif /* WITH_NETEM */
	}

	struct {
		const char *str;
		struct NodeDirection *dir;
	} dirs[] = {
		{ "in", &in },
		{ "out", &out }
	};

	const char *fields[] = { "signals", "builtin", "vectorize", "hooks" };

	for (unsigned j = 0; j < ARRAY_LEN(dirs); j++) {
		json_t *json_dir = json_object_get(json, dirs[j].str);

		/* Skip if direction is unused */
		if (!json_dir) {
			json_dir = json_pack("{ s: b }", "enabled", 0);
		}

		/* Copy missing fields from main node config to direction config */
		for (unsigned i = 0; i < ARRAY_LEN(fields); i++) {
			json_t *json_field_dir  = json_object_get(json_dir, fields[i]);
			json_t *json_field_node = json_object_get(json, fields[i]);

			if (json_field_node && !json_field_dir)
				json_object_set(json_dir, fields[i], json_field_node);
		}

		ret = dirs[j].dir->parse(json_dir);
		if (ret)
			return ret;
	}

	config = json;

	return 0;
}

int Node::check()
{
	assert(state == State::CHECKED ||
	       state == State::PARSED ||
	       state == State::INITIALIZED);

	in.check();
	out.check();

	state = State::CHECKED;

	return 0;
}

int Node::start()
{
	int ret;

	assert(state == State::PREPARED);

	logger->info("Starting node {}", getNameFull());

	ret = in.start();
	if (ret)
		return ret;

	ret = out.start();
	if (ret)
		return ret;

#ifdef __linux__
	/* Set fwmark for outgoing packets if netem is enabled for this node */
	if (fwmark >= 0) {
		for (int fd : getNetemFDs()) {
			ret = setsockopt(fd, SOL_SOCKET, SO_MARK, &fwmark, sizeof(fwmark));
			if (ret)
				throw RuntimeError("Failed to set FW mark for outgoing packets");
			else
				logger->debug("Set FW mark for socket (sd={}) to {}", fd, fwmark);
		}
	}
#endif /* __linux__ */

	state = State::STARTED;
	sequence = sequence_init;

	return 0;
}

int Node::stop()
{
	int ret;

	if (state != State::STOPPING &&
	    state != State::STARTED &&
	    state != State::CONNECTED &&
	    state != State::PENDING_CONNECT)
		return 0;

	logger->info("Stopping node");
	setState(State::STOPPING);

	ret = in.stop();
	if (ret)
		return ret;

	ret = out.stop();
	if (ret)
		return ret;

	return 0;
}

int Node::restart()
{
	int ret;

	assert(state == State::STARTED);

	logger->info("Restarting node");

	ret = stop();
	if (ret)
		return ret;

	ret = start();
	if (ret)
		return ret;

	return 0;
}

int Node::read(struct Sample * smps[], unsigned cnt)
{
	int toread, readd, nread = 0;
	unsigned vect;

	if (state == State::PAUSED || state == State::PENDING_CONNECT)
		return 0;
	else if (state != State::STARTED && state != State::CONNECTED)
		return -1;

	vect = factory->getVectorize();
	if (!vect)
		vect = cnt;

	while (cnt - nread > 0) {
		toread = MIN(cnt - nread, vect);
		readd = _read(&smps[nread], toread);
		if (readd < 0)
			return readd;

		nread += readd;
	}

#ifdef WITH_HOOKS
	/* Run read hooks */
	int rread = in.hooks.process(smps, nread);
	if (rread < 0)
		return rread;

	int skipped = nread - rread;
	if (skipped > 0) {
		if (stats != nullptr)
			stats->update(Stats::Metric::SMPS_SKIPPED, skipped);

		logger->debug("Received {} samples of which {} have been skipped", nread, skipped);
	}
	else
		logger->debug("Received {} samples", nread);

	return rread;
#else
	logger->debug("Received {} samples", nread);

	return nread;
#endif /* WITH_HOOKS */
}

int Node::write(struct Sample * smps[], unsigned cnt)
{
	int tosend, sent, nsent = 0;
	unsigned vect;

	if (state == State::PAUSED || state == State::PENDING_CONNECT)
		return 0;
	else if (state != State::STARTED && state != State::CONNECTED)
		return -1;

#ifdef WITH_HOOKS
	/* Run write hooks */
	cnt = out.hooks.process(smps, cnt);
	if (cnt <= 0)
		return cnt;
#endif /* WITH_HOOKS */

	vect = getFactory()->getVectorize();
	if (!vect)
		vect = cnt;

	while (cnt - nsent > 0) {
		tosend = MIN(cnt - nsent, vect);
		sent = _write(&smps[nsent], tosend);
		if (sent < 0)
			return sent;

		nsent += sent;
		logger->debug("Sent {} samples", sent);
	}

	return nsent;
}

const std::string & Node::getNameFull()
{
	if (name_full.empty()) {
		name_full = fmt::format("{}: uuid={}, #in.signals={}/{}, #in.hooks={}, #out.hooks={}, in.vectorize={}, out.vectorize={}",
			getName(), uuid::toString(uuid).c_str(),
			getInputSignals(false)->size(),
			getInputSignals(true)->size(),
			in.hooks.size(), out.hooks.size(),
			in.vectorize, out.vectorize
		);

#ifdef WITH_NETEM
		name_full += fmt::format(", out.netem={}", tc_qdisc ? "yes" : "no");

		if (tc_qdisc)
			name_full += fmt::format(", fwmark={}", fwmark);
#endif /* WITH_NETEM */

		if (out.path) {
			name_full += fmt::format(", #out.signals={}/{}",
				getOutputSignals(false) ? getOutputSignals(false)->size() : 0,
				getOutputSignals()      ? getOutputSignals()->size()      : 0);

			name_full += fmt::format(", out.path={}", out.path->toString());
		}

		/* Append node-type specific details */
		auto details = getDetails();
		if (!details.empty())
			name_full += fmt::format(", {}", details);
	}

	return name_full;
}

SignalList::Ptr Node::getInputSignals(bool after_hooks) const
{
	return in.getSignals(after_hooks);
}

SignalList::Ptr Node::getOutputSignals(bool after_hooks) const
{
	if (out.path)
		return out.path->getOutputSignals();

	return nullptr;
}

unsigned Node::getInputSignalsMaxCount() const
{
	return in.getSignalsMaxCount();
}

unsigned Node::getOutputSignalsMaxCount() const
{
	if (out.path)
		return out.path->getOutputSignalsMaxCount();

	return 0;
}

bool Node::isValidName(const std::string &name)
{
	std::regex re(RE_NODE_NAME);

	return std::regex_match(name, re);
}

json_t * Node::toJson() const
{
	json_t *json_node;
	json_t *json_signals_in = nullptr;
	json_t *json_signals_out = nullptr;

	json_signals_in = getInputSignals()->toJson();

	auto output_signals = getOutputSignals();
	if (output_signals)
		json_signals_out = output_signals->toJson();

	json_node = json_pack("{ s: s, s: s, s: s, s: i, s: { s: i, s: o? }, s: { s: i, s: o? } }",
		"name",		getNameShort().c_str(),
		"uuid", 	uuid::toString(uuid).c_str(),
		"state",	stateToString(state).c_str(),
		"affinity",	affinity,
		"in",
			"vectorize",	in.vectorize,
			"signals", 	json_signals_in,
		"out",
			"vectorize",	out.vectorize,
			"signals",	json_signals_out
	);

	if (stats)
		json_object_set_new(json_node, "stats", stats->toJson());

	auto *status = _readStatus();
	if (status)
		json_object_set_new(json_node, "status", status);

	/* Add all additional fields of node here.
	 * This can be used for metadata */
	json_object_update(json_node, config);

	return json_node;
}

void Node::swapSignals() {
	SWAP(in.signals, out.signals);
}

Node * NodeFactory::make(json_t *json, const uuid_t &id, const std::string &name)
{
	int ret;
	std::string type;
	Node *n;

	if (json_is_object(json))
		throw ConfigError(json, "node-config-node", "Node configuration must be an object");

	json_t *json_type = json_object_get(json, "type");

	type = json_string_value(json_type);

	n = NodeFactory::make(type, id, name);
	if (!n)
		return nullptr;

	ret = n->parse(json);
	if (ret) {
		delete n;
		return nullptr;
	}

	return n;
}

Node * NodeFactory::make(const std::string &type, const uuid_t &id, const std::string &name)
{
	NodeFactory *nf = plugin::registry->lookup<NodeFactory>(type);
	if (!nf)
		throw RuntimeError("Unknown node-type: {}", type);

	return nf->make(id, name);
}

int NodeFactory::start(SuperNode *sn)
{
	getLogger()->info("Initialized node type which is used by {} nodes", instances.size());

	state = State::STARTED;

	return 0;
}

int NodeFactory::stop()
{
	getLogger()->info("De-initialized node type");

	state = State::STOPPED;

	return 0;
}