lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(worker-join-refactor): implement apply_join_state, init_worker, rejoin_worker

Browse source 
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).
This commit is contained in:
Rongsong Shen 2026-05-20 08:24:26 +00:00
parent 1599d06bc4
commit 93ed282ef7
3 changed files with 194 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

139
src/controller/worker.c
View file

@ -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);