From c329db504dd7d0c9493f9ee2b3e5117a554ca9b3 Mon Sep 17 00:00:00 2001 From: WHC Date: Thu, 29 Oct 2026 18:16:46 +0800 Subject: [PATCH] perf: accept thread + dial backoff; join/peers.json (0.9.23) Run inbound peer handling on its own thread so outbound dials cannot starve :9100. Failed peers cool down exponentially. Also land --join (member seat) and realm/peers.json seed discovery for Tailscale meshes. Co-authored-by: Cursor --- docs/parliament.md | 14 ++- include/cst_parliament_cli.h | 4 + include/cst_version.h | 2 +- src/discover.c | 55 +++++++++++ src/main.c | 22 ++++- src/parliament_cli.c | 34 +++++++ src/parliament_daemon.c | 179 ++++++++++++++++++++++++++++------- 7 files changed, 273 insertions(+), 37 deletions(-) diff --git a/docs/parliament.md b/docs/parliament.md index bb062e4..a12a8d7 100644 --- a/docs/parliament.md +++ b/docs/parliament.md @@ -17,15 +17,24 @@ cst-agent --realm-password-init ```bash export CST_INSTANCE_ID=laptop-a # 可选；默认用本机 hostname cst-agent --login --realm-label "thinkpad" +# 其他机器（同 salt/公钥 + 密码）： +cst-agent --join --instance phone-b --realm-label "pixel" # TUI: /realm login（夺主席）或 /realm join（不抢主席） ``` +跨网段（如 Tailscale）多播发现不可用时，在 Realm 目录放 `peers.json`： + +```json +{"peers":[{"host":"100.64.0.13","port":9100,"iid":"labpc","name":"labpc"}]} +``` + 效果： 1. 用密码解锁 Realm；`--login` / `/realm login` 使本机成为**唯一主席** -2. `/realm join` 只登记席位票：主席空缺时才夺椅，否则以 **member** 身份加入 +2. `--join` / `/realm join` 只登记席位票：主席空缺时才夺椅，否则以 **member** 身份加入 3. 写入本地登录票（24h；成员票在主席易主后仍有效，除非本席被剥夺主席） -4. 拉起后台议会服务（发现 + 密封通道按 revision 同步 `parliament.context`） +4. 拉起后台议会服务（发现 + 密封通道按 revision 同步 `parliament.context`； + Unix 下 accept 独立线程，dial 失败会指数退避，避免饿死入站） 查看状态： @@ -94,6 +103,7 @@ Alpine 内构建，chair↔member 握手 + 密封通道同步（`sealed=1` / `pe |------|------| | `--realm-password-init` | 创建密码 Realm | | `--login` | 登录 + 夺主席 + 启服务 | +| `--join` | 加入席位（不抢主席）+ 启服务 | | `--service` / `--service-fg` | 仅启服务 | | `--service-stop` | 停服务 | | `--new-project ID` | 清场换项目 | diff --git a/include/cst_parliament_cli.h b/include/cst_parliament_cli.h index 17144b5..b2a98c5 100644 --- a/include/cst_parliament_cli.h +++ b/include/cst_parliament_cli.h @@ -15,6 +15,10 @@ int cst_parliament_cli_login(const char *password, const char *instance_id, int cst_parliament_cli_login_and_serve(const char *password, const char *instance_id, const char *device_label, int peer_port); +/* Join as member seat (no chair steal) then start background service. */ +int cst_parliament_cli_join_and_serve(const char *password, + const char *instance_id, + const char *device_label, int peer_port); int cst_parliament_cli_service(const char *password, const char *instance_id, int peer_port, int foreground); int cst_parliament_cli_service_stop(void); diff --git a/include/cst_version.h b/include/cst_version.h index 68ae0e5..decc8dc 100644 --- a/include/cst_version.h +++ b/include/cst_version.h @@ -3,7 +3,7 @@ #define CST_VERSION_MAJOR 0 #define CST_VERSION_MINOR 9 -#define CST_VERSION_PATCH 22 +#define CST_VERSION_PATCH 23 #define CST_VERSION_STR_(x) #x #define CST_VERSION_STR(x) CST_VERSION_STR_(x) diff --git a/src/discover.c b/src/discover.c index 172c5cc..9e4f1e1 100644 --- a/src/discover.c +++ b/src/discover.c @@ -250,6 +250,60 @@ static void ingest_local(cst_discover_t *d) { remember(d, &list[i]); } +/* Optional realm/peers.json for meshes without multicast (e.g. Tailscale): + * {"peers":[{"host":"100.x.y.z","port":9100,"iid":"seat-a","name":"a"}]} */ +static void ingest_seed_peers(cst_discover_t *d) { + char dir[OS_PATH_MAX], path[OS_PATH_MAX]; + if (cst_realm_dir(dir, sizeof(dir)) != 0) + return; + if (os_path_join(path, sizeof(path), dir, "peers.json") != 0) + return; + FILE *f = fopen(path, "rb"); + if (!f) + return; + cst_buf_t b = {0}; + char chunk[4096]; + size_t n; + while ((n = fread(chunk, 1, sizeof(chunk), f)) && b.len < 64 * 1024) + cst_buf_append(&b, chunk, n); + fclose(f); + cJSON *root = cJSON_Parse(b.buf ? b.buf : "{}"); + cst_buf_free(&b); + if (!root) + return; + cJSON *arr = cJSON_GetObjectItemCaseSensitive(root, "peers"); + if (!cJSON_IsArray(arr)) { + cJSON_Delete(root); + return; + } + const cJSON *it; + cJSON_ArrayForEach(it, arr) { + const cJSON *host = cJSON_GetObjectItemCaseSensitive(it, "host"); + const cJSON *port = cJSON_GetObjectItemCaseSensitive(it, "port"); + const cJSON *iid = cJSON_GetObjectItemCaseSensitive(it, "iid"); + const cJSON *name = cJSON_GetObjectItemCaseSensitive(it, "name"); + const cJSON *rid = cJSON_GetObjectItemCaseSensitive(it, "rid"); + if (!cJSON_IsString(host) || !host->valuestring[0] || !cJSON_IsNumber(port) || + port->valueint <= 0 || !cJSON_IsString(iid) || !iid->valuestring[0]) + continue; + if (cJSON_IsString(rid) && rid->valuestring[0] && + strcmp(rid->valuestring, d->realm_fp) != 0) + continue; + cst_peer_info_t p; + memset(&p, 0, sizeof(p)); + snprintf(p.realm_fp, sizeof(p.realm_fp), "%s", d->realm_fp); + snprintf(p.instance_id, sizeof(p.instance_id), "%s", iid->valuestring); + snprintf(p.host, sizeof(p.host), "%s", host->valuestring); + if (cJSON_IsString(name) && name->valuestring[0]) + snprintf(p.name, sizeof(p.name), "%s", name->valuestring); + else + snprintf(p.name, sizeof(p.name), "%s", iid->valuestring); + p.port = port->valueint; + remember(d, &p); + } + cJSON_Delete(root); +} + static int send_beacon(cst_discover_t *d) { if (d->sock == DISC_BAD) return -1; @@ -310,6 +364,7 @@ static void *discover_main(void *arg) { (void)cst_discover_local_put(d->realm_fp, d->instance_id, d->name, d->peer_port); ingest_local(d); + ingest_seed_peers(d); } ticks++; if (d->sock == DISC_BAD) { diff --git a/src/main.c b/src/main.c index 1fa6542..115fd69 100644 --- a/src/main.c +++ b/src/main.c @@ -68,6 +68,7 @@ static void usage(const char *argv0) { " --realm-new create Realm file keys (legacy/dev)\n" " --realm-password-init create password-derived Realm (no sk on disk)\n" " --login unlock (CST_REALM_PASSWORD or prompt); claim chair\n" + " --join unlock; join as member seat (no chair steal)\n" " --service run parliament service (discover+context sync)\n" " --service-fg same as --service, stay in foreground\n" " --service-stop stop the parliament background service\n" @@ -393,8 +394,8 @@ int main(int argc, char **argv) { int new_session = 0, do_list = 0, do_migrate = 0; const char *resume_id = NULL; int do_realm_new = 0, do_realm_enroll = 0, do_peer_listen = 0, do_peer_dial = 0, - do_discover = 0, do_realm_pw_init = 0, do_login = 0, do_new_project = 0, - do_service = 0, do_service_fg = 0, do_service_stop = 0; + do_discover = 0, do_realm_pw_init = 0, do_login = 0, do_join = 0, + do_new_project = 0, do_service = 0, do_service_fg = 0, do_service_stop = 0; const char *realm_enroll_id = NULL, *realm_label = NULL; const char *peer_dial_spec = NULL, *parliament_instance = NULL; const char *new_project_id = NULL; @@ -462,6 +463,7 @@ int main(int argc, char **argv) { {"service", no_argument, 0, 0x11b}, {"service-fg", no_argument, 0, 0x11c}, {"service-stop", no_argument, 0, 0x11d}, + {"join", no_argument, 0, 0x11e}, {0, 0, 0, 0}}; int c; while ((c = getopt_long(argc, argv, "w:s:b:k:m:P:r:n:T:Hp:dahVNcR:LyA:C:e:M:", @@ -524,6 +526,9 @@ int main(int argc, char **argv) { case 0x11d: do_service_stop = 1; break; + case 0x11e: + do_join = 1; + break; case 'w': snprintf(cfg.workspace_root, sizeof(cfg.workspace_root), "%s", optarg); break; @@ -705,6 +710,19 @@ int main(int argc, char **argv) { cst_curl_cleanup(); return rc; } + if (do_join) { + const char *pw = getenv("CST_REALM_PASSWORD"); + char iidbuf[64]; + const char *iid = parliament_instance; + if (!iid || !iid[0]) { + cst_parliament_default_instance(iidbuf, sizeof(iidbuf)); + iid = iidbuf; + } + int rc = cst_parliament_cli_join_and_serve(pw, iid, realm_label, + discover_peer_port); + cst_curl_cleanup(); + return rc; + } if (do_new_project) { int rc = cst_parliament_cli_new_project(new_project_id); cst_curl_cleanup(); diff --git a/src/parliament_cli.c b/src/parliament_cli.c index 5c50b1f..ed49162 100644 --- a/src/parliament_cli.c +++ b/src/parliament_cli.c @@ -125,6 +125,40 @@ int cst_parliament_cli_login_and_serve(const char *password, return 0; } +int cst_parliament_cli_join_and_serve(const char *password, + const char *instance_id, + const char *device_label, int peer_port) { + char err[256], pwbuf[256]; + const char *pw = password; + if ((!pw || !pw[0]) && resolve_password(pwbuf, sizeof(pwbuf), &pw) != 0) + return 1; + char *hand = + cst_parliament_seat_join(pw, instance_id, device_label, err, sizeof(err)); + if (!hand) { + fprintf(stderr, "%s\n", err[0] ? err : "join failed"); + if (pw == pwbuf) + memset(pwbuf, 0, sizeof(pwbuf)); + return 1; + } + fputs(hand, stdout); + free(hand); + if (cst_parliament_daemon_start(pw, instance_id, + peer_port > 0 ? peer_port : 9100, err, + sizeof(err)) != 0) { + fprintf(stderr, "warning: service: %s (seat join still valid)\n", + err[0] ? err : "start failed"); + printf("service=failed\n"); + if (pw == pwbuf) + memset(pwbuf, 0, sizeof(pwbuf)); + return 0; + } + if (cst_parliament_daemon_running()) + printf("service=running\n"); + if (pw == pwbuf) + memset(pwbuf, 0, sizeof(pwbuf)); + return 0; +} + int cst_parliament_cli_service(const char *password, const char *instance_id, int peer_port, int foreground) { char err[256], pwbuf[256]; diff --git a/src/parliament_daemon.c b/src/parliament_daemon.c index 77bb237..78f747b 100644 --- a/src/parliament_daemon.c +++ b/src/parliament_daemon.c @@ -20,8 +20,10 @@ #include #else #include +#include #include #include +#include #include #endif @@ -72,7 +74,6 @@ static int read_meta(daemon_meta_t *m) { } else if (!strncmp(line, "port=", 5)) m->port = atoi(line + 5); else if (m->pid <= 0) { - /* Legacy single-line pidfile. */ int pid = atoi(line); if (pid > 0) m->pid = pid; @@ -165,6 +166,20 @@ static void sleep_ms(int ms) { #endif } +static int64_t now_ms(void) { +#ifdef _WIN32 + return (int64_t)GetTickCount64(); +#else + struct timeval tv; + gettimeofday(&tv, NULL); + return (int64_t)tv.tv_sec * 1000 + (int64_t)tv.tv_usec / 1000; +#endif +} + +#ifndef _WIN32 +static pthread_mutex_t g_sync_mu = PTHREAD_MUTEX_INITIALIZER; +#endif + static int load_local_bundle(cst_peer_sync_bundle_t *b) { memset(b, 0, sizeof(*b)); cst_parliament_meta_t m; @@ -186,16 +201,27 @@ static int apply_if_newer(cst_peer_sync_bundle_t *b, unsigned long before) { b->chair_instance, b->context); } +/* Short sync budget: accept thread must not stall the listen backlog. */ static int sync_on_session(cst_peer_session_t *sess) { cst_peer_sync_bundle_t local; - if (load_local_bundle(&local) != 0) +#ifndef _WIN32 + pthread_mutex_lock(&g_sync_mu); +#endif + if (load_local_bundle(&local) != 0) { +#ifndef _WIN32 + pthread_mutex_unlock(&g_sync_mu); +#endif return -1; + } unsigned long before = local.revision; char err[128]; - int rc = cst_peer_sync(sess, &local, 15000, err, sizeof(err)); + int rc = cst_peer_sync(sess, &local, 3000, err, sizeof(err)); if (rc == 0) apply_if_newer(&local, before); free(local.context); +#ifndef _WIN32 + pthread_mutex_unlock(&g_sync_mu); +#endif return rc; } @@ -208,6 +234,72 @@ static int unlock_realm(const char *password, cst_realm_keys_t *keys) { return cst_realm_load(dir, keys); } +typedef struct { + os_socket_t ls; + cst_realm_cert_t cert; + uint8_t sk[64]; + uint8_t realm_pk[32]; +} accept_ctx_t; + +#ifndef _WIN32 +static void *accept_main(void *arg) { + accept_ctx_t *ac = arg; + while (!g_stop) { + cst_peer_session_t sess; + char herr[128]; + if (cst_peer_accept((intptr_t)ac->ls, &ac->cert, ac->sk, ac->realm_pk, 400, + &sess, herr, sizeof(herr)) != 0) + continue; + sync_on_session(&sess); + cst_peer_close(&sess); + } + return NULL; +} +#endif + +/* Per-peer dial cooldown so stuck remotes do not dominate the dial loop. */ +typedef struct { + char iid[64]; + int fails; + int64_t cool_until_ms; +} dial_cool_t; + +static dial_cool_t *cool_find(dial_cool_t *tab, int *n, int cap, + const char *iid) { + for (int i = 0; i < *n; i++) { + if (!strcmp(tab[i].iid, iid)) + return &tab[i]; + } + if (*n >= cap) + return NULL; + dial_cool_t *c = &tab[(*n)++]; + memset(c, 0, sizeof(*c)); + snprintf(c->iid, sizeof(c->iid), "%s", iid); + return c; +} + +static int cool_blocked(dial_cool_t *c, int64_t now) { + return c && c->cool_until_ms > now; +} + +static void cool_fail(dial_cool_t *c, int64_t now) { + if (!c) + return; + c->fails = c->fails < 8 ? c->fails + 1 : 8; + /* 4s, 8s, 16s, … capped at 120s */ + int64_t wait = 4000LL << (c->fails - 1); + if (wait > 120000) + wait = 120000; + c->cool_until_ms = now + wait; +} + +static void cool_ok(dial_cool_t *c) { + if (!c) + return; + c->fails = 0; + c->cool_until_ms = 0; +} + int cst_parliament_daemon_run(const char *password, const char *instance_id, int peer_port, char *err, size_t err_sz) { if (err && err_sz) @@ -272,65 +364,93 @@ int cst_parliament_daemon_run(const char *password, const char *instance_id, signal(SIGHUP, on_sig); #endif g_stop = 0; - int tick = 0, idle = 0, dial_rr = 0; + + accept_ctx_t ac; + memset(&ac, 0, sizeof(ac)); + ac.ls = ls; + ac.cert = cert; + memcpy(ac.sk, sk, 64); + memcpy(ac.realm_pk, realm.pk, 32); + +#ifndef _WIN32 + pthread_t ath; + int have_ath = pthread_create(&ath, NULL, accept_main, &ac) == 0; +#else + int have_ath = 0; +#endif + + int tick = 0, dial_rr = 0; + dial_cool_t cool[24]; + int ncool = 0; + while (!g_stop) { cst_peer_session_t sess; char herr[128]; - /* Prefer accept: a blocking outbound dial used to starve inbound peers - * (CLOSE-WAIT + unread hello ~385B on :9100). Drain the backlog first. */ - int accepted = 0; - for (;;) { - int accept_ms = accepted > 0 ? 50 : (idle > 20 ? 800 : idle > 5 ? 400 : 200); - if (cst_peer_accept((intptr_t)ls, &cert, sk, realm.pk, accept_ms, &sess, - herr, sizeof(herr)) != 0) + +#ifdef _WIN32 + /* No accept thread: keep prefer-accept single-threaded path. */ + for (int a = 0; a < 4; a++) { + if (cst_peer_accept((intptr_t)ls, &cert, sk, realm.pk, 200, &sess, herr, + sizeof(herr)) != 0) break; sync_on_session(&sess); cst_peer_close(&sess); - accepted++; - idle = 0; - if (accepted >= 8) - break; } - if (accepted == 0) - idle++; +#endif int push = cst_parliament_sync_pending(); tick++; - /* At most one outbound dial per loop turn, short wall budget, so accept - * keeps pace. Lexicographic rule avoids A↔B mutual-dial deadlock unless - * we have a local push to deliver. */ - int period = idle > 20 ? 30 : idle > 5 ? 12 : 6; + int period = push ? 1 : 8; if ((push || tick % period == 0) && disc) { cst_peer_info_t peers[16]; int n = cst_discover_peers(disc, peers, 16); + int64_t now = now_ms(); int tried = 0; for (int k = 0; k < n; k++) { int i = (dial_rr + k) % n; if (peers[i].port <= 0) continue; if (!strcmp(peers[i].instance_id, instance_id)) - continue; /* never dial ourselves */ + continue; if (!push && strcmp(instance_id, peers[i].instance_id) > 0) - continue; /* lower id initiates idle polls */ + continue; + dial_cool_t *c = cool_find(cool, &ncool, 24, peers[i].instance_id); + if (!push && cool_blocked(c, now)) + continue; dial_rr = (i + 1) % n; - if (cst_peer_dial(peers[i].host, peers[i].port, &cert, sk, realm.pk, - 2500, &sess, herr, sizeof(herr)) == 0) { + /* Dial budget stays short; accept thread covers inbound meanwhile. */ + int ok = cst_peer_dial(peers[i].host, peers[i].port, &cert, sk, + realm.pk, 2000, &sess, herr, sizeof(herr)) == 0; + if (ok) { sync_on_session(&sess); cst_peer_close(&sess); - idle = 0; + cool_ok(c); + } else { + cool_fail(c, now); } tried = 1; - break; /* one dial then go back to accept */ + break; } if (push && (tried || n == 0)) cst_parliament_sync_ack(); } + /* Pace dial loop; accept runs concurrently on Unix. */ + sleep_ms(have_ath ? 150 : 50); } + +#ifndef _WIN32 + if (have_ath) { + /* Wake accept's poll by shutting down listen sock after stop. */ + os_net_shutdown(ls); + pthread_join(ath, NULL); + } +#endif if (disc) cst_discover_stop(disc); os_net_close(ls); clear_pid(); crypto_wipe(sk, sizeof(sk)); + crypto_wipe(ac.sk, sizeof(ac.sk)); crypto_wipe(realm.seed, sizeof(realm.seed)); crypto_wipe(realm.sk, sizeof(realm.sk)); return 0; @@ -380,13 +500,10 @@ int cst_parliament_daemon_start(const char *password, const char *instance_id, (cur_port <= 0 || cur_port == want_port); if (same) return 0; - /* Different seat or port: take over. */ (void)cst_parliament_daemon_stop(); } } #ifdef _WIN32 - /* Windows: run in-process foreground loop under a detached console is - * deferred; callers should use --service in a service manager. */ if (err && err_sz) snprintf(err, err_sz, "background daemon requires --service on Windows"); return -1; @@ -403,7 +520,6 @@ int cst_parliament_daemon_start(const char *password, const char *instance_id, return -1; } if (pid > 0) { - /* Parent: wait briefly for matching pidfile. */ for (int i = 0; i < 50; i++) { int cur_pid = 0, cur_port = 0; char cur_id[64] = {0}; @@ -417,7 +533,6 @@ int cst_parliament_daemon_start(const char *password, const char *instance_id, snprintf(err, err_sz, "service did not start"); return -1; } - /* Child: detach fully so SSH/nohup callers are not held by open stdio. */ setsid(); int fd = open("/dev/null", O_RDWR); if (fd >= 0) {