#include "hybbx/circuit_balance.h" #include "hybbx/util.h" #include #include #include 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; }