summaryrefslogtreecommitdiff
path: root/src/core/circuit_balance.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/circuit_balance.c')
-rw-r--r--src/core/circuit_balance.c545
1 files changed, 545 insertions, 0 deletions
diff --git a/src/core/circuit_balance.c b/src/core/circuit_balance.c
new file mode 100644
index 0000000..5571e9d
--- /dev/null
+++ b/src/core/circuit_balance.c
@@ -0,0 +1,545 @@
+#include "hybbx/circuit_balance.h"
+#include "hybbx/util.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+typedef struct balance_frame {
+ uint8_t data[HYBBX_CIRCUIT_MAX_FRAME];
+ size_t len;
+} balance_frame_t;
+
+struct hybbx_circuit_balance {
+ hybbx_circuit_balance_config_t cfg;
+ hybbx_circuit_link_profile_t profile;
+ balance_frame_t queue[HYBBX_CIRCUIT_BALANCE_QUEUE_MAX];
+ size_t q_head;
+ size_t q_count;
+ size_t queue_bytes;
+ hybbx_circuit_balance_action_t action;
+ hybbx_circuit_balance_action_t last_sent_action;
+ double drain_accum;
+ time_t oldest_enqueue;
+ int link_low_bw;
+};
+
+static int str_ieq_local(const char *a, const char *b)
+{
+ if (a == NULL || b == NULL) {
+ return 0;
+ }
+
+ while (*a != '\0' && *b != '\0') {
+ char ca = *a;
+ char cb = *b;
+
+ if (ca >= 'A' && ca <= 'Z') {
+ ca = (char)(ca - 'A' + 'a');
+ }
+ if (cb >= 'A' && cb <= 'Z') {
+ cb = (char)(cb - 'A' + 'a');
+ }
+ if (ca != cb) {
+ return 0;
+ }
+ a++;
+ b++;
+ }
+
+ return *a == '\0' && *b == '\0';
+}
+
+static const char *find_kv_line(const char *payload, size_t len,
+ const char *key,
+ char *scratch, size_t scratch_len)
+{
+ const char *cursor = payload;
+ const char *end = payload + len;
+ size_t key_len = strlen(key);
+
+ while (cursor < end) {
+ const char *line_end = memchr(cursor, '\n', (size_t)(end - cursor));
+ const char *line_stop = line_end != NULL ? line_end : end;
+ const char *eq = memchr(cursor, '=', (size_t)(line_stop - cursor));
+
+ if (eq != NULL && (size_t)(eq - cursor) == key_len &&
+ memcmp(cursor, key, key_len) == 0) {
+ const char *value = eq + 1;
+ size_t value_len = (size_t)(line_stop - value);
+
+ while (value_len > 0 &&
+ (value[value_len - 1] == '\r' ||
+ value[value_len - 1] == '\n' ||
+ value[value_len - 1] == ' ')) {
+ value_len--;
+ }
+ if (value_len >= scratch_len) {
+ value_len = scratch_len - 1;
+ }
+ memcpy(scratch, value, value_len);
+ scratch[value_len] = '\0';
+ return scratch;
+ }
+
+ if (line_end == NULL) {
+ break;
+ }
+ cursor = line_end + 1;
+ }
+
+ return NULL;
+}
+
+void hybbx_circuit_balance_config_defaults(hybbx_circuit_balance_config_t *cfg)
+{
+ if (cfg == NULL) {
+ return;
+ }
+
+ memset(cfg, 0, sizeof(*cfg));
+ cfg->enabled = 1;
+ cfg->lag_sec = 5u;
+ cfg->queue_pause = 4096u;
+ cfg->queue_break = 16384u;
+ cfg->queue_cancel = 65536u;
+}
+
+void hybbx_circuit_link_profile_from_auth(const hybbx_link_auth_t *auth,
+ hybbx_circuit_link_profile_t *out)
+{
+ unsigned baud;
+
+ if (out == NULL) {
+ return;
+ }
+
+ memset(out, 0, sizeof(*out));
+ out->bandwidth = HYBBX_CIRCUIT_BW_HIGH;
+ out->duplex = HYBBX_CIRCUIT_DUPLEX_FULL;
+ out->baud = HYBBX_CIRCUIT_BALANCE_DEFAULT_BAUD;
+
+ if (auth == NULL) {
+ return;
+ }
+
+ baud = auth->baud;
+ out->duplex = (hybbx_circuit_duplex_t)auth->duplex;
+
+ if (auth->bandwidth[0] != '\0') {
+ if (str_ieq_local(auth->bandwidth, "low")) {
+ out->bandwidth = HYBBX_CIRCUIT_BW_LOW;
+ } else if (str_ieq_local(auth->bandwidth, "high")) {
+ out->bandwidth = HYBBX_CIRCUIT_BW_HIGH;
+ }
+ } else if (baud > 0 && baud <= HYBBX_CIRCUIT_BALANCE_LOW_BAUD_THRESHOLD) {
+ out->bandwidth = HYBBX_CIRCUIT_BW_LOW;
+ }
+
+ if (baud > 0) {
+ out->baud = baud;
+ } else if (out->bandwidth == HYBBX_CIRCUIT_BW_LOW) {
+ out->baud = 1200u;
+ }
+
+ if (out->duplex == HYBBX_CIRCUIT_DUPLEX_UNSET) {
+ out->duplex = out->bandwidth == HYBBX_CIRCUIT_BW_LOW ?
+ HYBBX_CIRCUIT_DUPLEX_HALF : HYBBX_CIRCUIT_DUPLEX_FULL;
+ }
+
+ if (auth->frequency_mhz[0] != '\0') {
+ out->frequency_mhz = strtod(auth->frequency_mhz, NULL);
+ }
+}
+
+const char *hybbx_circuit_balance_action_name(hybbx_circuit_balance_action_t action)
+{
+ switch (action) {
+ case HYBBX_CIRCUIT_BAL_PAUSE:
+ return "pause";
+ case HYBBX_CIRCUIT_BAL_BREAK:
+ return "break";
+ case HYBBX_CIRCUIT_BAL_CANCEL:
+ return "cancel";
+ case HYBBX_CIRCUIT_BAL_RESUME:
+ return "resume";
+ case HYBBX_CIRCUIT_BAL_NONE:
+ default:
+ return "none";
+ }
+}
+
+hybbx_result_t hybbx_circuit_flow_ctrl_parse(const char *payload, size_t len,
+ hybbx_circuit_balance_action_t *action_out,
+ char *reason_out, size_t reason_cap)
+{
+ char scratch[128];
+ const char *action_str;
+
+ if (payload == NULL || action_out == NULL) {
+ return HYBBX_ERR_INVALID;
+ }
+
+ action_str = find_kv_line(payload, len, "action", scratch, sizeof(scratch));
+ if (action_str == NULL) {
+ return HYBBX_ERR_INVALID;
+ }
+
+ if (str_ieq_local(action_str, "pause")) {
+ *action_out = HYBBX_CIRCUIT_BAL_PAUSE;
+ } else if (str_ieq_local(action_str, "break")) {
+ *action_out = HYBBX_CIRCUIT_BAL_BREAK;
+ } else if (str_ieq_local(action_str, "cancel")) {
+ *action_out = HYBBX_CIRCUIT_BAL_CANCEL;
+ } else if (str_ieq_local(action_str, "resume")) {
+ *action_out = HYBBX_CIRCUIT_BAL_RESUME;
+ } else {
+ return HYBBX_ERR_INVALID;
+ }
+
+ if (reason_out != NULL && reason_cap > 0) {
+ reason_out[0] = '\0';
+ if (find_kv_line(payload, len, "reason", scratch, sizeof(scratch)) != NULL) {
+ hybbx_strlcpy(reason_out, scratch, reason_cap);
+ }
+ }
+
+ return HYBBX_OK;
+}
+
+size_t hybbx_circuit_flow_ctrl_format(hybbx_circuit_balance_action_t action,
+ const char *reason,
+ char *out, size_t out_cap)
+{
+ int n;
+ const char *reason_str = reason != NULL && reason[0] != '\0' ? reason : "-";
+
+ if (out == NULL || out_cap == 0) {
+ return 0;
+ }
+
+ n = snprintf(out, out_cap, "action=%s\nreason=%s\n",
+ hybbx_circuit_balance_action_name(action), reason_str);
+ if (n < 0 || (size_t)n >= out_cap) {
+ return 0;
+ }
+
+ return (size_t)n;
+}
+
+hybbx_circuit_balance_t *hybbx_circuit_balance_create(
+ const hybbx_circuit_balance_config_t *cfg)
+{
+ hybbx_circuit_balance_t *bal;
+
+ bal = calloc(1, sizeof(*bal));
+ if (bal == NULL) {
+ return NULL;
+ }
+
+ if (cfg != NULL) {
+ bal->cfg = *cfg;
+ } else {
+ hybbx_circuit_balance_config_defaults(&bal->cfg);
+ }
+
+ hybbx_circuit_link_profile_from_auth(NULL, &bal->profile);
+ bal->link_low_bw = 0;
+ return bal;
+}
+
+void hybbx_circuit_balance_destroy(hybbx_circuit_balance_t *bal)
+{
+ free(bal);
+}
+
+void hybbx_circuit_balance_set_profile(hybbx_circuit_balance_t *bal,
+ const hybbx_circuit_link_profile_t *profile)
+{
+ if (bal == NULL || profile == NULL) {
+ return;
+ }
+
+ bal->profile = *profile;
+ bal->link_low_bw = profile->bandwidth == HYBBX_CIRCUIT_BW_LOW;
+}
+
+hybbx_circuit_balance_action_t hybbx_circuit_balance_action(
+ const hybbx_circuit_balance_t *bal)
+{
+ if (bal == NULL) {
+ return HYBBX_CIRCUIT_BAL_NONE;
+ }
+
+ return bal->action;
+}
+
+size_t hybbx_circuit_balance_queued_bytes(const hybbx_circuit_balance_t *bal)
+{
+ if (bal == NULL) {
+ return 0;
+ }
+
+ return bal->queue_bytes;
+}
+
+void hybbx_circuit_balance_spared_cancel(hybbx_circuit_balance_t *bal)
+{
+ if (bal == NULL) {
+ return;
+ }
+
+ if (bal->action == HYBBX_CIRCUIT_BAL_CANCEL) {
+ bal->action = HYBBX_CIRCUIT_BAL_PAUSE;
+ bal->last_sent_action = HYBBX_CIRCUIT_BAL_NONE;
+ }
+}
+
+static unsigned balance_effective_baud(const hybbx_circuit_balance_t *bal)
+{
+ unsigned baud;
+
+ if (bal == NULL) {
+ return HYBBX_CIRCUIT_BALANCE_DEFAULT_BAUD;
+ }
+
+ baud = bal->profile.baud;
+ if (baud == 0) {
+ baud = HYBBX_CIRCUIT_BALANCE_DEFAULT_BAUD;
+ }
+ return baud;
+}
+
+static unsigned balance_estimated_lag_sec(const hybbx_circuit_balance_t *bal)
+{
+ unsigned baud;
+ unsigned lag;
+
+ if (bal == NULL || bal->queue_bytes == 0) {
+ return 0;
+ }
+
+ baud = balance_effective_baud(bal);
+ if (baud == 0) {
+ return 0;
+ }
+
+ lag = (unsigned)((bal->queue_bytes * 8u) / baud);
+ if (bal->oldest_enqueue > 0) {
+ time_t now = time(NULL);
+ if (now > bal->oldest_enqueue) {
+ unsigned wall = (unsigned)(now - bal->oldest_enqueue);
+ if (wall > lag) {
+ lag = wall;
+ }
+ }
+ }
+ return lag;
+}
+
+static void balance_send_flow(hybbx_circuit_balance_t *bal,
+ hybbx_circuit_balance_action_t action,
+ const char *reason,
+ hybbx_circuit_balance_flow_fn flow_fn,
+ void *flow_ctx)
+{
+ if (bal == NULL || flow_fn == NULL) {
+ return;
+ }
+ if (action == bal->last_sent_action) {
+ return;
+ }
+
+ flow_fn(flow_ctx, action, reason);
+ bal->last_sent_action = action;
+}
+
+static void balance_clear_queue(hybbx_circuit_balance_t *bal)
+{
+ if (bal == NULL) {
+ return;
+ }
+
+ bal->q_head = 0;
+ bal->q_count = 0;
+ bal->queue_bytes = 0;
+ bal->oldest_enqueue = 0;
+ bal->drain_accum = 0.0;
+}
+
+static void balance_apply_escalation(hybbx_circuit_balance_t *bal,
+ hybbx_circuit_balance_flow_fn flow_fn,
+ void *flow_ctx)
+{
+ unsigned lag;
+ const hybbx_circuit_balance_config_t *cfg;
+
+ if (bal == NULL || !bal->cfg.enabled) {
+ return;
+ }
+
+ cfg = &bal->cfg;
+ lag = balance_estimated_lag_sec(bal);
+
+ if (bal->link_low_bw &&
+ (bal->queue_bytes >= cfg->queue_cancel ||
+ lag >= cfg->lag_sec * 2u)) {
+ if (bal->action != HYBBX_CIRCUIT_BAL_CANCEL) {
+ bal->action = HYBBX_CIRCUIT_BAL_CANCEL;
+ balance_clear_queue(bal);
+ balance_send_flow(bal, HYBBX_CIRCUIT_BAL_CANCEL, "queue_overload",
+ flow_fn, flow_ctx);
+ }
+ return;
+ }
+
+ if (bal->queue_bytes >= cfg->queue_break || lag >= cfg->lag_sec) {
+ if (bal->action != HYBBX_CIRCUIT_BAL_BREAK &&
+ bal->action != HYBBX_CIRCUIT_BAL_CANCEL) {
+ bal->action = HYBBX_CIRCUIT_BAL_BREAK;
+ balance_clear_queue(bal);
+ balance_send_flow(bal, HYBBX_CIRCUIT_BAL_BREAK, "queue_lag",
+ flow_fn, flow_ctx);
+ bal->action = HYBBX_CIRCUIT_BAL_PAUSE;
+ balance_send_flow(bal, HYBBX_CIRCUIT_BAL_PAUSE, "stabilize",
+ flow_fn, flow_ctx);
+ }
+ return;
+ }
+
+ if (bal->queue_bytes >= cfg->queue_pause ||
+ lag >= (cfg->lag_sec > 0 ? cfg->lag_sec / 2u : 0u)) {
+ if (bal->action == HYBBX_CIRCUIT_BAL_NONE) {
+ bal->action = HYBBX_CIRCUIT_BAL_PAUSE;
+ balance_send_flow(bal, HYBBX_CIRCUIT_BAL_PAUSE, "queue_pressure",
+ flow_fn, flow_ctx);
+ }
+ return;
+ }
+
+ if (bal->action == HYBBX_CIRCUIT_BAL_PAUSE &&
+ bal->queue_bytes < cfg->queue_pause / 2u &&
+ lag < (cfg->lag_sec > 0 ? cfg->lag_sec / 2u : 0u)) {
+ bal->action = HYBBX_CIRCUIT_BAL_NONE;
+ balance_send_flow(bal, HYBBX_CIRCUIT_BAL_RESUME, "stabilized",
+ flow_fn, flow_ctx);
+ }
+}
+
+hybbx_result_t hybbx_circuit_balance_submit(hybbx_circuit_balance_t *bal,
+ const uint8_t *frame, size_t len,
+ hybbx_circuit_balance_send_fn send_fn,
+ void *send_ctx,
+ hybbx_circuit_balance_flow_fn flow_fn,
+ void *flow_ctx)
+{
+ size_t slot;
+ balance_frame_t *entry;
+
+ if (bal == NULL || frame == NULL || len == 0 || send_fn == NULL) {
+ return HYBBX_ERR_INVALID;
+ }
+
+ if (!bal->cfg.enabled ||
+ bal->profile.bandwidth == HYBBX_CIRCUIT_BW_HIGH) {
+ return send_fn(send_ctx, frame, len);
+ }
+
+ if (bal->action == HYBBX_CIRCUIT_BAL_CANCEL) {
+ return HYBBX_ERR_BUSY;
+ }
+
+ if (bal->q_count >= HYBBX_CIRCUIT_BALANCE_QUEUE_MAX) {
+ balance_apply_escalation(bal, flow_fn, flow_ctx);
+ if (bal->action == HYBBX_CIRCUIT_BAL_CANCEL) {
+ return HYBBX_ERR_BUSY;
+ }
+ return HYBBX_ERR_NOMEM;
+ }
+
+ if (len > sizeof(entry->data)) {
+ return HYBBX_ERR_INVALID;
+ }
+
+ slot = (bal->q_head + bal->q_count) % HYBBX_CIRCUIT_BALANCE_QUEUE_MAX;
+ entry = &bal->queue[slot];
+ memcpy(entry->data, frame, len);
+ entry->len = len;
+ bal->q_count++;
+ bal->queue_bytes += len;
+ if (bal->oldest_enqueue == 0) {
+ bal->oldest_enqueue = time(NULL);
+ }
+
+ balance_apply_escalation(bal, flow_fn, flow_ctx);
+ return HYBBX_OK;
+}
+
+hybbx_circuit_balance_tick_result_t hybbx_circuit_balance_tick(
+ hybbx_circuit_balance_t *bal, unsigned poll_ms,
+ hybbx_circuit_balance_send_fn send_fn, void *send_ctx,
+ hybbx_circuit_balance_flow_fn flow_fn, void *flow_ctx)
+{
+ unsigned baud;
+ double budget;
+ size_t sent_budget;
+
+ if (bal == NULL || send_fn == NULL) {
+ return HYBBX_CIRCUIT_BAL_TICK_OK;
+ }
+
+ if (bal->action == HYBBX_CIRCUIT_BAL_CANCEL) {
+ return HYBBX_CIRCUIT_BAL_TICK_CANCEL_LINK;
+ }
+
+ if (!bal->cfg.enabled || bal->q_count == 0) {
+ balance_apply_escalation(bal, flow_fn, flow_ctx);
+ return HYBBX_CIRCUIT_BAL_TICK_OK;
+ }
+
+ if (bal->action == HYBBX_CIRCUIT_BAL_PAUSE) {
+ balance_apply_escalation(bal, flow_fn, flow_ctx);
+ if (bal->action == HYBBX_CIRCUIT_BAL_CANCEL) {
+ return HYBBX_CIRCUIT_BAL_TICK_CANCEL_LINK;
+ }
+ return HYBBX_CIRCUIT_BAL_TICK_OK;
+ }
+
+ baud = balance_effective_baud(bal);
+ if (poll_ms == 0) {
+ poll_ms = 1;
+ }
+
+ budget = ((double)baud * (double)poll_ms) / 8000.0;
+ bal->drain_accum += budget;
+ sent_budget = (size_t)bal->drain_accum;
+ bal->drain_accum -= (double)sent_budget;
+
+ while (sent_budget > 0 && bal->q_count > 0) {
+ balance_frame_t *entry = &bal->queue[bal->q_head];
+
+ if (entry->len > sent_budget) {
+ break;
+ }
+
+ if (send_fn(send_ctx, entry->data, entry->len) != HYBBX_OK) {
+ break;
+ }
+
+ sent_budget -= entry->len;
+ bal->queue_bytes -= entry->len;
+ bal->q_head = (bal->q_head + 1u) % HYBBX_CIRCUIT_BALANCE_QUEUE_MAX;
+ bal->q_count--;
+ }
+
+ if (bal->q_count == 0) {
+ bal->oldest_enqueue = 0;
+ }
+
+ balance_apply_escalation(bal, flow_fn, flow_ctx);
+ if (bal->action == HYBBX_CIRCUIT_BAL_CANCEL) {
+ return HYBBX_CIRCUIT_BAL_TICK_CANCEL_LINK;
+ }
+
+ return HYBBX_CIRCUIT_BAL_TICK_OK;
+}
git clone -b <branch> https://cgit.mode42.com/<repo>.git
git clone -b <branch> git://cgit.mode42.com/<repo>.git


info@mode42.com