udphole

commit a604588cb3646c5b3f2e8b6af5adc488cbc226f3
parent 3be552addb2b63435c5076bfc53d8abeeca12abb
Author: Robin Bron <robin.bron@yourhosting.nl>
Date:   Mon,  2 Mar 2026 01:08:48 +0100

More cleanup

Diffstat:
M
src/domain/daemon/session.c
 | 
26
+++-----------------------
M
src/domain/daemon/session.h
 | 
4
++++
M
src/interface/api/server.c
 | 
81
-------------------------------------------------------------------------------
M
src/interface/cli/command/cluster.c
 | 
43
++++++++++++++++++++-----------------------
M
src/interface/cli/command/daemon.c
 | 
4
+++-

5 files changed, 30 insertions(+), 128 deletions(-)
diff --git a/src/domain/daemon/session.c b/src/domain/daemon/session.c
@@ -60,7 +60,6 @@ typedef struct session {
 
 static session_t **sessions = NULL;
 static size_t sessions_count = 0;
-static int running = 0;
 
 static session_t *find_session(const char *session_id) {
   for (size_t i = 0; i < sessions_count; i++) {
@@ -71,18 +70,6 @@ static session_t *find_session(const char *session_id) {
   return NULL;
 }
 
-static uint64_t socket_hash(const void *item, uint64_t seed0, uint64_t seed1) {
-  const socket_t *s = item;
-  return hashmap_sip(s->socket_id, strlen(s->socket_id), seed0, seed1);
-}
-
-static int socket_compare(const void *a, const void *b, void *udata) {
-  (void)udata;
-  const socket_t *sa = a;
-  const socket_t *sb = b;
-  return strcmp(sa->socket_id, sb->socket_id);
-}
-
 static socket_t *find_socket(session_t *s, const char *socket_id) {
   if (!s || !s->sockets || !socket_id) return NULL;
   for (size_t i = 0; i < s->sockets_count; i++) {
@@ -392,7 +379,6 @@ static socket_t *find_socket_by_fd(session_t *s, int fd) {
 PT_THREAD(session_pt(struct pt *pt, int64_t timestamp, struct pt_task *task)) {
   session_t *s = task->udata;
 
-  (void)timestamp;
   PT_BEGIN(pt);
 
   char buffer[BUFFER_SIZE];
@@ -993,29 +979,23 @@ resp_object *domain_session_count(const char *cmd, resp_object *args) {
 }
 
 PT_THREAD(session_manager_pt(struct pt *pt, int64_t timestamp, struct pt_task *task)) {
-  (void)timestamp;
+  session_manager_udata_t *udata = task->udata;
   log_trace("session_manager: protothread entry");
   PT_BEGIN(pt);
 
   PT_WAIT_UNTIL(pt, domain_cfg);
 
-  running = 1;
   log_info("udphole: manager started with port range %d-%d", domain_cfg->port_low, domain_cfg->port_high);
 
-  int64_t last_cleanup = 0;
-
   for (;;) {
-    int64_t now = (int64_t)(time(NULL));
-    if (now - last_cleanup >= 1) {
+    if (timestamp - udata->last_cleanup >= 1000) {
       cleanup_expired_sessions();
-      last_cleanup = now;
+      udata->last_cleanup = timestamp;
     }
 
     PT_YIELD(pt);
   }
 
-  running = 0;
-
   for (size_t i = 0; i < sessions_count; i++) {
     if (sessions[i]) {
       sessions[i]->marked_for_deletion = 1;
diff --git a/src/domain/daemon/session.h b/src/domain/daemon/session.h
@@ -6,6 +6,10 @@
 #include "domain/scheduler.h"
 #include "common/resp.h"
 
+typedef struct {
+  int64_t last_cleanup;
+} session_manager_udata_t;
+
 PT_THREAD(session_manager_pt(struct pt *pt, int64_t timestamp, struct pt_task *task));
 
 resp_object *domain_session_create(const char *cmd, resp_object *args);
diff --git a/src/interface/api/server.c b/src/interface/api/server.c
@@ -157,85 +157,6 @@ static void client_flush(api_client_t *c) {
   }
 }
 
-static int parse_inline(const char *line, size_t len, char **args, int max_args) {
-  int nargs = 0;
-  const char *p = line;
-  const char *end = line + len;
-  while (p < end && nargs < max_args) {
-    while (p < end && (*p == ' ' || *p == '\t')) p++;
-    if (p >= end) break;
-    const char *start;
-    const char *tok_end;
-    if (*p == '"' || *p == '\'') {
-      char quote = *p++;
-      start = p;
-      while (p < end && *p != quote) p++;
-      tok_end = p;
-      if (p < end) p++;
-    } else {
-      start = p;
-      while (p < end && *p != ' ' && *p != '\t') p++;
-      tok_end = p;
-    }
-    size_t tlen = (size_t)(tok_end - start);
-    char *arg = malloc(tlen + 1);
-    if (!arg) return -1;
-    memcpy(arg, start, tlen);
-    arg[tlen] = '\0';
-    args[nargs++] = arg;
-  }
-  return nargs;
-}
-
-static int parse_resp_command(api_client_t *c, char **args, int max_args, int *nargs) {
-  *nargs = 0;
-  if (c->rlen == 0) return 0;
-
-  if (c->rbuf[0] != '*') {
-    char *nl = memchr(c->rbuf, '\n', c->rlen);
-    if (!nl) return 0;
-    size_t line_len = (size_t)(nl - c->rbuf);
-    size_t trim = line_len;
-    if (trim > 0 && c->rbuf[trim - 1] == '\r') trim--;
-    int n = parse_inline(c->rbuf, trim, args, max_args);
-    if (n < 0) return -1;
-    *nargs = n;
-    size_t consumed = line_len + 1;
-    c->rlen -= consumed;
-    if (c->rlen > 0) memmove(c->rbuf, c->rbuf + consumed, c->rlen);
-    return n > 0 ? 1 : 0;
-  }
-
-  size_t pos = 0;
-  char *nl = memchr(c->rbuf + pos, '\n', c->rlen - pos);
-  if (!nl) return 0;
-  int count = atoi(c->rbuf + 1);
-  if (count <= 0 || count > max_args) return -1;
-  pos = (size_t)(nl - c->rbuf) + 1;
-
-  for (int i = 0; i < count; i++) {
-    if (pos >= c->rlen) return 0;
-    if (c->rbuf[pos] != '$') return -1;
-    nl = memchr(c->rbuf + pos, '\n', c->rlen - pos);
-    if (!nl) return 0;
-    int blen = atoi(c->rbuf + pos + 1);
-    if (blen < 0) return -1;
-    size_t hdr_end = (size_t)(nl - c->rbuf) + 1;
-    if (hdr_end + (size_t)blen + 2 > c->rlen) return 0;
-    char *arg = malloc((size_t)blen + 1);
-    if (!arg) return -1;
-    memcpy(arg, c->rbuf + hdr_end, (size_t)blen);
-    arg[blen] = '\0';
-    args[i] = arg;
-    pos = hdr_end + (size_t)blen + 2;
-  }
-
-  *nargs = count;
-  c->rlen -= pos;
-  if (c->rlen > 0) memmove(c->rbuf, c->rbuf + pos, c->rlen);
-  return 1;
-}
-
 static bool permit_matches(const char *pattern, const char *cmd) {
   size_t plen = strlen(pattern);
   if (plen == 1 && pattern[0] == '*')
@@ -587,7 +508,6 @@ PT_THREAD(api_server_pt(struct pt *pt, int64_t timestamp, struct pt_task *task))
   }
 
   for (;;) {
-    (void)timestamp;
     if (udata->server_fds && udata->server_fds[0] > 0) {
       PT_WAIT_UNTIL(pt, domain_schedmod_has_data(udata->server_fds, &udata->ready_fds) > 0);
       if (udata->ready_fds && udata->ready_fds[0] > 0) {
@@ -624,7 +544,6 @@ PT_THREAD(api_server_pt(struct pt *pt, int64_t timestamp, struct pt_task *task))
 }
 
 PT_THREAD(api_client_pt(struct pt *pt, int64_t timestamp, struct pt_task *task)) {
-  (void)timestamp;
   api_client_t *state = task->udata;
 
   log_trace("api_client: protothread entry fd=%d", state->fd);
diff --git a/src/interface/cli/command/cluster.c b/src/interface/cli/command/cluster.c
@@ -24,6 +24,8 @@
 static cluster_node_t *cluster_nodes = NULL;
 static const char *cluster_listen = NULL;
 
+PT_THREAD(cluster_manager_pt(struct pt *pt, int64_t timestamp, struct pt_task *task));
+
 typedef struct {
   int64_t last_health_check;
 } cluster_udata_t;
@@ -128,19 +130,6 @@ static void cluster_health_check(void) {
   }
 }
 
-static void cluster_refresh_session_counts(void) {
-  cluster_node_t *node = cluster_nodes;
-  while (node) {
-    if (node->available) {
-      int count = cluster_node_get_session_count(node);
-      if (count >= 0) {
-        node->session_count = count;
-      }
-    }
-    node = node->next;
-  }
-}
-
 static cluster_node_t *cluster_select_node(void) {
   cluster_node_t *best = NULL;
   cluster_node_t *node = cluster_nodes;
@@ -171,15 +160,6 @@ static cluster_node_t *cluster_find_session_node(const char *session_id) {
   return NULL;
 }
 
-static resp_object *cluster_forward_to_node(cluster_node_t *node, const char *cmd, resp_object *args) {
-  if (!node || node->fd < 0) return NULL;
-
-  resp_object *resp = cluster_node_send_command(node, cmd, NULL);
-  if (resp) return resp;
-
-  return NULL;
-}
-
 static resp_object *cluster_handle_command(const char *cmd, resp_object *args) {
   if (!cmd || !args) return NULL;
 
@@ -356,7 +336,6 @@ static resp_object *cluster_handle_command(const char *cmd, resp_object *args) {
 
   if (strcmp(cmd, "system.load") == 0) {
     resp_object *result = resp_array_init();
-    char buf[64];
 
     resp_array_append_bulk(result, "1min");
     resp_array_append_bulk(result, "0.00");
@@ -478,8 +457,26 @@ int cli_cmd_cluster(int argc, const char **argv) {
   udata->last_health_check = 0;
 
   domain_schedmod_pt_create(api_server_pt, NULL);
+  domain_schedmod_pt_create(cluster_manager_pt, udata);
 
   log_info("udphole: cluster daemon started");
 
   return domain_schedmod_main();
 }
+
+PT_THREAD(cluster_manager_pt(struct pt *pt, int64_t timestamp, struct pt_task *task)) {
+  cluster_udata_t *udata = task->udata;
+  PT_BEGIN(pt);
+
+  PT_WAIT_UNTIL(pt, cluster_nodes);
+
+  for (;;) {
+    if (timestamp - udata->last_health_check >= HEALTH_CHECK_INTERVAL * 1000) {
+      cluster_health_check();
+      udata->last_health_check = timestamp;
+    }
+    PT_YIELD(pt);
+  }
+
+  PT_END(pt);
+}
diff --git a/src/interface/cli/command/daemon.c b/src/interface/cli/command/daemon.c
@@ -107,8 +107,10 @@ int cli_cmd_daemon(int argc, const char **argv) {
 
   log_info("udphole: starting daemon");
 
+  session_manager_udata_t *session_udata = calloc(1, sizeof(session_manager_udata_t));
+
   domain_schedmod_pt_create(api_server_pt, NULL);
-  domain_schedmod_pt_create(session_manager_pt, NULL);
+  domain_schedmod_pt_create(session_manager_pt, session_udata);
 
   log_info("udphole: daemon started");