From 93ed282ef79e8416003b25fc0288920764b4a191 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 20 May 2026 08:24:26 +0000 Subject: [PATCH] feat(worker-join-refactor): implement apply_join_state, init_worker, rejoin_worker Replace monolithic add_worker with three focused functions: - apply_join_state (static under production, exported under XCE_TESTING): shared status/resource update called by both create and rejoin paths; clears XCE_WORKER_UNAVAIL, applies hconfig overrides, updates resources - init_worker: first-time join path via makeWorker + apply_join_state - rejoin_worker: reconnect path with explicit admin-bit preservation (XCE_WORKER_ADMIN_MASK = LOCKEDU|CLOSED|HIDDEN|CLOSED_BY_ADMIN) Also makes find_matched_hostconfig non-static so join.c can call it via the extern declaration already in worker.h. Removes add_worker (no longer declared in worker.h, no callers). All 7 tests in test_worker_join pass: apply_join_state (clears UNAVAIL, hosttype override, resource values), init_worker (creates entry, returns correct idx), rejoin_worker (updates attrs/ep, preserves admin bits). --- .kiro/specs/worker-join-refactor/spec.json | 22 ++++ .kiro/specs/worker-join-refactor/tasks.md | 88 +++++++++++++ src/controller/worker.c | 139 +++++++++++++-------- 3 files changed, 194 insertions(+), 55 deletions(-) create mode 100644 .kiro/specs/worker-join-refactor/spec.json create mode 100644 .kiro/specs/worker-join-refactor/tasks.md diff --git a/.kiro/specs/worker-join-refactor/spec.json b/.kiro/specs/worker-join-refactor/spec.json new file mode 100644 index 00000000..b5a8554d --- /dev/null +++ b/.kiro/specs/worker-join-refactor/spec.json @@ -0,0 +1,22 @@ +{ + "feature_name": "worker-join-refactor", + "created_at": "2026-05-15T00:00:00Z", + "updated_at": "2026-05-15T03:20:00Z", + "language": "en", + "phase": "tasks-generated", + "approvals": { + "requirements": { + "generated": true, + "approved": true + }, + "design": { + "generated": true, + "approved": true + }, + "tasks": { + "generated": true, + "approved": true + } + }, + "ready_for_implementation": true +} diff --git a/.kiro/specs/worker-join-refactor/tasks.md b/.kiro/specs/worker-join-refactor/tasks.md new file mode 100644 index 00000000..d3437e20 --- /dev/null +++ b/.kiro/specs/worker-join-refactor/tasks.md @@ -0,0 +1,88 @@ +# Implementation Plan + +## Task Format + +- [ ] Major task number and description +- [ ] Sub-task number and description + - Detail items + - Observable completion condition + - _Requirements: X.X, Y.Y_ + +--- + +- [ ] 1. Foundation: write failing tests for all new worker join behaviors (TDD Red) + +- [x] 1.1 Write failing tests for the shared status and resource update step + - Add a new test file (e.g., `test_worker_join.c`) to `src/controller/testsuite/` and register it in `controller_test.c` + - Tests assert: after the helper runs, `XCE_WORKER_UNAVAIL` is cleared on the worker's status, hosttype/model match the hconfig override values, slot count equals the hconfig mxj value, and host-attribute resources (type, model, cpuf, maxcpus, maxmem, maxswap, maxtmp, ndisks) hold the expected values + - Tests compile but fail because the helper does not yet exist + - _Requirements: 4.1, 4.2, 4.3_ + +- [x] 1.2 Write failing tests for the first-time worker creation path + - Tests assert: calling the creation function with a previously-unknown hostname produces a non-negative index; `key_find` on `workermgr` with that name succeeds after the call + - Tests assert: when the creation function returns an error, the error code is negative and no `on_worker_join` side effect occurs + - Tests compile but fail because the function does not yet exist + - _Requirements: 2.1, 2.2, 2.3_ + +- [x] 1.3 Write failing tests for the rejoin update path with admin-bit preservation + - Tests assert: after the update function runs, the worker's `host.attrs` reflects the new values and `w->ep` matches the new endpoint + - Tests assert: admin status bits (`XCE_WORKER_LOCKEDU`, `XCE_WORKER_CLOSED`, `XCE_WORKER_HIDDEN`, `XCE_WORKER_CLOSED_BY_ADMIN`) present before the call are still set after it + - Tests compile but fail because the function does not yet exist + - _Requirements: 3.1, 3.2, 3.3_ + +--- + +- [x] 2. Core: extract shared status and resource update into a standalone helper (TDD Green — 1.1) + - Move the status-clear + hconfig-override + resource-update block out of `add_worker` in `worker.c` into a new `static` helper function + - The helper covers: clearing `XCE_WORKER_UNAVAIL`, assigning `w->ep`, overriding hosttype/model strings from hconfig (with `dereference`/`makeString`), calling `add_hostattrs_resvals`, computing mxj slots via `set_host_slots`, replacing `dispatch_window_config` and `thresholds` from hconfig, and looping over `hconfig->nresources` with `mark_host_res` + - Must be called under the caller-held per-worker write lock; document this precondition with a comment + - Tests from 1.1 turn green; `add_worker` still compiles by delegating to the new helper + - _Requirements: 4.1, 4.2, 4.3_ + +--- + +- [x] 3. Core: implement the first-time creation and rejoin update helpers + +- [x] 3.1 Implement the first-time worker creation function (TDD Green — 1.2) + - Build a temporary `xce_worker_t` from the incoming hostname and attrs, then call `makeWorker` to register it in `workermgr` (makeWorker manages its own write lock) + - On a `NULL` return from `makeWorker`, log the failure and return `-errno` to the caller without touching `on_worker_join` + - On success, acquire the per-worker write lock, call the shared helper from Task 2, release the lock, and return `w->idx` + - Tests from 1.2 turn green + - _Requirements: 2.1, 2.2, 2.3_ + +- [x] 3.2 Implement the rejoin worker update function (TDD Green — 1.3) + - Acquire the per-worker write lock at the start; save `admin_status = w->host.status & XCE_WORKER_ADMIN_MASK` before any modification + - Free and replace the existing `hosttype` and `hostmodel` strings; overwrite numeric attrs fields (`cpufactor`, `maxcpus`, `maxmem`, `maxswap`, `maxtmp`, `ndisks`) from the incoming attrs argument + - Call the shared helper from Task 2 (status/resource update, ep assignment), then restore admin bits via `w->host.status |= admin_status`; release the lock + - Return `w->idx`; tests from 1.3 turn green and admin bits are unchanged after a full round-trip + - _Requirements: 3.1, 3.2, 3.3_ + +--- + +- [x] 4. Integration: wire the explicit existence check and new helpers into the join handler + +- [x] 4.1 Update exported worker function declarations + - Remove the `add_worker` declaration from `src/controller/include/worker.h` + - Add declarations for the first-time creation function and the rejoin update function with their full parameter lists (including the `xce_hostconfig_t*` parameter) + - Verify that the only direct caller of `add_worker` was `handle_worker_join`; if any other callers exist, update them to use the appropriate new function + - `worker.h` compiles cleanly with no remaining reference to `add_worker` + - _Requirements: 5.4_ + +- [x] 4.2 Rewrite the worker join handler with explicit existence check and branching + - Remove the outer `lock(c->workermgr, RTTI_LOCK_WRITE)` / `unlock` pair that currently wraps the `add_worker` call (eliminates the double write-lock deadlock) + - Move `find_matched_hostconfig` to the top of the handler, before any lock; reject the join with the appropriate error code and set `reply.status` if it returns `NULL` + - Acquire a brief workermgr read lock, call `key_find`, release the read lock immediately + - On `-ENOENT` call the first-time creation function; on a found entry call the rejoin update function; on any other error from `key_find` reject the join and return that error code + - Fire `BEFORE_EVENT / HOOK_DETAIL_WORKER_JOIN` before the `key_find` step; fire `AFTER_EVENT` after the create-or-update completes; call `xce_reply` and `on_worker_join` outside any lock + - Set `priv->u.idx` before calling `on_worker_join`; if create or update returned an error, set `reply.status`, call `xce_reply`, and return without calling `on_worker_join` + - The join handler compiles and the sequence check → create-or-update → status/resource-update is visible at the call site with no hidden branching inside `makeWorker` + - _Requirements: 1.1, 1.2, 1.3, 2.2, 2.3, 4.4, 5.1, 5.2, 5.3_ + +--- + +- [x] 5. Validation: run full test suite and confirm behavioral preservation + - Build the controller test binary and run all suites in `src/controller/testsuite/` + - Confirm `test_e2e_hooks` records `BEFORE_EVENT` before any workermgr mutation and `AFTER_EVENT` after it, matching the pre-refactor ordering + - If any test fixture in the suite constructed workers by calling `add_worker` directly, update the fixture to call the creation function with the added `xce_hostconfig_t*` argument + - `make test` exits 0; no previously-passing test is broken by the refactor + - _Requirements: 5.1, 5.2, 5.3, 5.4_ diff --git a/src/controller/worker.c b/src/controller/worker.c index 80fb12ff..7a19871b 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -22,7 +22,7 @@ static int hostmatched(const char* pattern, const char* name) return 0; } -static xce_hostconfig_t* find_matched_hostconfig(cluster_t* c, const char* name) +xce_hostconfig_t* find_matched_hostconfig(cluster_t* c, const char* name) { xce_hostconfig_t* defaultconfig = NULL; xce_hostconfig_t** hconfigs = NULL; @@ -271,104 +271,133 @@ static int set_host_slots(cluster_t* c, worker_t* w, int nslots) return update_resvals(c, w, 1, &(resvals[0])); } -int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs) +/* Bits set by admin commands that must survive a worker rejoin */ +#define XCE_WORKER_ADMIN_MASK \ + (XCE_WORKER_LOCKEDU | XCE_WORKER_CLOSED | XCE_WORKER_HIDDEN \ + | XCE_WORKER_CLOSED_BY_ADMIN) + +/* + * apply_join_state — shared status/resource update for both create and rejoin. + * Precondition: caller holds per-worker WRITE lock. + */ +#ifdef XCE_TESTING +int apply_join_state(cluster_t* c, worker_t* w, endpoint_t* ep, + xce_hostconfig_t* hconfig) +#else +static int apply_join_state(cluster_t* c, worker_t* w, endpoint_t* ep, + xce_hostconfig_t* hconfig) +#endif { - cluster_t* c = (cluster_t*)cls; - /* clang-format off */ - worker_t worker = { - .host = (xce_worker_t){ - .name = (char*)host, - } - }; - /* clang-format on */ - worker_t* w = NULL; - int idx, i; - xce_hostconfig_t* hconfig = NULL; - int nslots; + int nslots; + int i; - if ((cls == NULL) || (host == NULL)) { - return -EINVAL; - } - - w = calloc(1, sizeof(*w)); - if (w == NULL) { - return -ENOMEM; - } - - hconfig = find_matched_hostconfig(c, host); - if (hconfig == NULL) { - return -errno; - } - - worker.host.attrs = attrs; - - w = makeWorker(c->workermgr, &(worker.host)); - if (w == NULL) { - return -errno; - } - - /* we should hold the update lock */ - /* If admin makes the host in locked or closed, we need keep this */ - - lock(w, RTTI_LOCK_WRITE); - - w->host.status &= ~XCE_WORKER_UNAVAIL; /* status OK */ + w->host.status &= ~XCE_WORKER_UNAVAIL; w->ep = ep; - /* override host type? */ if (strcmp(hconfig->hosttype, "!") != 0) { dereference(w->host.attrs->hosttype); w->host.attrs->hosttype = makeString(hconfig->hosttype); } - /* override host model */ if (strcmp(hconfig->hostmodel, "!") != 0) { dereference(w->host.attrs->hostmodel); w->host.attrs->hostmodel = makeString(hconfig->hostmodel); } + add_hostattrs_resvals(c, w); - idx = w->idx; - /* using auto detected or pre configured MXJ? */ nslots = hconfig->mxj; - if (nslots < 0) { + if (nslots < 0) nslots = w->host.attrs->maxcpus; - } - set_host_slots(c, w, nslots); if (w->host.dispatch_window_config != NULL) { free_timewindow_config(w->host.dispatch_window_config); w->host.dispatch_window_config = NULL; } - - if (hconfig->dispatchwindow != NULL) { + if (hconfig->dispatchwindow != NULL) w->host.dispatch_window_config = dup_timewindow_config(hconfig->dispatchwindow); - } if (w->host.thresholds != NULL) { free_resource_thresholds(w->host.thresholds, w->host.nthresholds); w->host.nthresholds = 0; w->host.thresholds = NULL; } - if (hconfig->thresholds != NULL) { w->host.thresholds = dup_resource_thresholds(hconfig->thresholds, hconfig->nthresholds); w->host.nthresholds = hconfig->nthresholds; } - /* add bool resources to host */ - for (i = 0; i < hconfig->nresources; i++) { + for (i = 0; i < hconfig->nresources; i++) mark_host_res(c, w, hconfig->resources[i]); + + return 0; +} + +int init_worker(cluster_t* c, endpoint_t* ep, const char* host, + hostattrs_t* attrs, xce_hostconfig_t* hconfig) +{ + xce_worker_t tmp; + worker_t* w; + int rc; + + memset(&tmp, 0, sizeof(tmp)); + tmp.name = (char*)host; + tmp.attrs = attrs; + + w = makeWorker(c->workermgr, &tmp); + if (w == NULL) { + ERROR("%s makeWorker failed for %s", __func__, host); + return -errno; } + lock(w, RTTI_LOCK_WRITE); + rc = apply_join_state(c, w, ep, hconfig); + unlock(w); + + if (rc < 0) + return rc; + + return w->idx; +} + +int rejoin_worker(cluster_t* c, worker_t* w, endpoint_t* ep, + hostattrs_t* attrs, xce_hostconfig_t* hconfig) +{ + int admin_status; + int rc; + + lock(w, RTTI_LOCK_WRITE); + + admin_status = w->host.status & XCE_WORKER_ADMIN_MASK; + + dereference(w->host.attrs->hosttype); + w->host.attrs->hosttype = makeString(attrs->hosttype); + dereference(w->host.attrs->hostmodel); + w->host.attrs->hostmodel = makeString(attrs->hostmodel); + + w->host.attrs->cpufactor = attrs->cpufactor; + w->host.attrs->maxcpus = attrs->maxcpus; + w->host.attrs->maxmem = attrs->maxmem; + w->host.attrs->maxswap = attrs->maxswap; + w->host.attrs->maxtmp = attrs->maxtmp; + w->host.attrs->ndisks = attrs->ndisks; + + rc = apply_join_state(c, w, ep, hconfig); + + w->host.status |= admin_status; + unlock(w); - return idx; + if (rc < 0) + return rc; + + return w->idx; } + static void reset_hostattrs_unknown(hostattrs_t* a) { dereference(a->hosttype);