/* Node-type for loopback connections. * * Author: Steffen Vogel * SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University * SPDX-License-Identifier: Apache-2.0 */ #include #include #include #include #include #include #include using namespace villas; using namespace villas::node; using namespace villas::utils; LoopbackNode::LoopbackNode(const uuid_t &id, const std::string &name) : Node(id, name), queuelen(DEFAULT_QUEUE_LENGTH), mode(QueueSignalledMode::AUTO) { queue.queue.state = State::DESTROYED; } LoopbackNode::~LoopbackNode() { int ret __attribute__((unused)); ret = queue_signalled_destroy(&queue); } int LoopbackNode::prepare() { assert(state == State::CHECKED); int ret = queue_signalled_init(&queue, queuelen, &memory::mmap, mode); if (ret) throw RuntimeError("Failed to initialize queue"); return Node::prepare(); } int LoopbackNode::stop() { int ret; ret = queue_signalled_close(&queue); if (ret) return ret; return 0; } int LoopbackNode::_read(struct Sample *smps[], unsigned cnt) { int avail; struct Sample *cpys[cnt]; avail = queue_signalled_pull_many(&queue, (void **)cpys, cnt); sample_copy_many(smps, cpys, avail); sample_decref_many(cpys, avail); return avail; } int LoopbackNode::_write(struct Sample *smps[], unsigned cnt) { sample_incref_many(smps, cnt); int pushed = queue_signalled_push_many(&queue, (void **)smps, cnt); if (pushed < 0) { sample_decref_many(smps, cnt); return pushed; } // Released unpushed samples if ((unsigned)pushed < cnt) { sample_decref_many(smps + pushed, cnt - pushed); logger->warn("Queue overrun"); } return pushed; } std::vector LoopbackNode::getPollFDs() { return {queue_signalled_fd(&queue)}; } const std::string &LoopbackNode::getDetails() { if (details.empty()) { details = fmt::format("queuelen={}", queuelen); } return details; } int LoopbackNode::parse(json_t *json) { const char *mode_str = nullptr; json_error_t err; int ret; ret = json_unpack_ex(json, &err, 0, "{ s?: i, s?: s }", "queuelen", &queuelen, "mode", &mode_str); if (ret) throw ConfigError(json, err, "node-config-node-loopback"); if (mode_str) { if (!strcmp(mode_str, "auto")) mode = QueueSignalledMode::AUTO; #ifdef HAVE_EVENTFD else if (!strcmp(mode_str, "eventfd")) mode = QueueSignalledMode::EVENTFD; #endif else if (!strcmp(mode_str, "pthread")) mode = QueueSignalledMode::PTHREAD; else if (!strcmp(mode_str, "polling")) mode = QueueSignalledMode::POLLING; else throw ConfigError(json, "node-config-node-loopback-mode", "Unknown mode '{}'", mode_str); } return Node::parse(json); } // Register node static char n[] = "loopback"; static char d[] = "loopback node-type"; static NodePlugin nf;