/** Rate-limiting hook. * * @author Steffen Vogel * @copyright 2017, Institute for Automation of Complex Power Systems, EONERC * @license GNU General Public License (version 3) * * VILLASnode * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * any later version. * * This program 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 General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program. If not, see . *********************************************************************************/ /** @addtogroup hooks Hook functions * @{ */ #include #include "villas/hook.h" #include "villas/plugin.h" #include "villas/timing.h" #include "villas/sample.h" struct limit_rate { enum { LIMIT_RATE_LOCAL, LIMIT_RATE_RECEIVED, LIMIT_RATE_ORIGIN } mode; /**< The timestamp which should be used for limiting. */ double deadtime; struct timespec last; }; static int limit_rate_init(struct hook *h) { struct limit_rate *p = (struct limit_rate *) h->_vd; p->last = (struct timespec) { 0 }; /* Default values */ p->mode = LIMIT_RATE_LOCAL; return 0; } static int limit_rate_parse(struct hook *h, json_t *cfg) { struct limit_rate *p = (struct limit_rate *) h->_vd; int ret; json_error_t err; double rate; const char *mode = NULL; ret = json_unpack_ex(cfg, &err, 0, "{ s: F, s?: s }", "rate", &rate, "mode", &mode ); if (ret) jerror(&err, "Failed to parse configuration of hook '%s'", plugin_name(h->_vt)); if (mode) { if (!strcmp(mode, "origin")) p->mode = LIMIT_RATE_ORIGIN; else if (!strcmp(mode, "received")) p->mode = LIMIT_RATE_RECEIVED; else if (!strcmp(mode, "local")) p->mode = LIMIT_RATE_LOCAL; else error("Invalid value '%s' for setting 'mode' in limit_rate hook", mode); } p->deadtime = 1.0 / rate; return 0; } static int limit_rate_write(struct hook *h, struct sample *smps[], unsigned *cnt) { struct limit_rate *p = (struct limit_rate *) h->_vd; struct timespec next; unsigned ret = 0; for (unsigned i = 0; i < *cnt; i++) { switch (p->mode) { case LIMIT_RATE_LOCAL: next = time_now(); break; case LIMIT_RATE_ORIGIN: next = smps[i]->ts.origin; break; case LIMIT_RATE_RECEIVED: next = smps[i]->ts.received; break; } if (time_delta(&p->last, &next) < p->deadtime) continue; /* Drop this sample */ p->last = next; smps[ret++] = smps[i]; } *cnt = ret; return 0; } static struct plugin p = { .name = "limit_rate", .description = "Limit sending rate", .type = PLUGIN_TYPE_HOOK, .hook = { .flags = HOOK_NODE | HOOK_PATH, .priority = 99, .init = limit_rate_init, .parse = limit_rate_parse, .read = limit_rate_write, .write = limit_rate_write, .process= limit_rate_write, .size = sizeof(struct limit_rate) } }; REGISTER_PLUGIN(&p) /** @} */