lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/worker.c
Rongsong Shen 7547a8f14b fix(worker): replicate admin host-control status updates to standbys (bug-1472) 
worker_from_capnp()'s idempotency guard returned -EEXIST unconditionally
whenever the target worker already existed, discarding the decoded WAL
payload entirely. Correct for a duplicate WAL_WORKER_CTRL/JOIN replay,
but wrong for HOOK_DETAIL_WORKER_UPDATE (hostctrl -c/-o admin close/open):
the target existing is the normal case there, yet the status change was
silently dropped on every member other than the one that processed the
RPC directly -- a permanent, unconditional, cluster-wide replication gap
for host admin-status, confirmed via byte-precise checkpoint divergence
on a live 3-node soak.

Now decodes the entry's hook_detail before applying the existing-worker
guard: JOIN keeps its no-op short-circuit; UPDATE replaces (not ORs,
since HOST_OPEN must clear closed bits) the existing worker's
XCE_WORKER_ADMIN_MASK bits with the decoded payload's, in place, under
the worker's own write lock. HOOK_DETAIL_WORKER_DEAD is deliberately
left unfixed -- its replication would need fields (resvals) that never
round-trip through the Worker capnp wire schema at all, and this
codebase already treats liveness state as locally-derived rather than
WAL-replicated elsewhere -- filed separately as bug-1472-remainder.

Independently reviewed: RED->GREEN cycle re-reproduced from scratch,
replace-vs-OR semantics and lock discipline verified directly, DEAD
deferral corroborated against an independent design-invariant comment
this fix didn't originally cite. Suite: 1030 -> 1033 tests, 0 failures.
2026-07-26 04:06:42 +00:00

899 lines
25 KiB
C
Raw Blame History

#include <errno.h>
#include <stdbool.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "persist.h"
#include "wobj.h"
#include "worker.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xstring.h"
static int hostmatched(const char* pattern, const char* name)
{
if (strcmp(pattern, name) == 0) {
return 1;
}
return 0;
}
xce_hostconfig_t* find_matched_hostconfig(cluster_t* c, const char* name)
{
xce_hostconfig_t* defaultconfig = NULL;
xce_hostconfig_t** hconfigs = NULL;
int i, nhconfigs;
if (c->myconf == NULL) {
errno = EINVAL;
return NULL;
}
if (c->myconf->cluster_config == NULL) {
errno = EINVAL;
return NULL;
}
if (c->myconf->cluster_config->hconfigs == NULL) {
return NULL;
}
nhconfigs = c->myconf->cluster_config->nhconfigs;
hconfigs = c->myconf->cluster_config->hconfigs;
for (i = 0; i < nhconfigs; i++) {
xce_hostconfig_t* config = hconfigs[i];
if (strcmp(config->name, "default") == 0) {
defaultconfig = config;
continue;
}
if (hostmatched(config->name, name)) {
return config;
}
}
return defaultconfig;
}
static int update_val_res(cluster_t* c, worker_t* w, xce_resource_t* r,
xce_val_t* val)
{
int idx;
zarray_t* array = NULL;
xce_val_t* v;
idx = r->idx;
if (r->flags & XCE_RESF_SHARED) {
/* update shared resource */
array = &(c->sharedinst);
} else {
/* update host based resource */
array = &(w->host.resvals);
}
if (idx < array->caps) {
v = (xce_val_t*)array->data[idx];
} else {
v = NULL;
}
if (v == NULL) {
int rc;
v = (xce_val_t*)calloc(1, sizeof(*v));
rc = zarray_put(array, idx, v);
if (rc < 0) {
FREEUP(v);
return rc;
}
}
memcpy(v, val, sizeof(*v));
return 0;
}
static int update_resval(cluster_t* c, worker_t* w, xce_resval_t* val)
{
xce_resource_t* r = NULL;
xce_val_t* v = NULL;
int rc;
lock(c->resmgr, RTTI_LOCK_READ);
rc = key_find(c->resmgr, (void*)val->name, (void**)&r);
unlock(c->resmgr);
if (rc < 0) {
return rc;
}
if (r == NULL) {
ERROR("resource `%s' can not be found", val->name);
return -errno;
}
if (r->idx < 0) {
ERROR("invalid resource definition for %s", val->name);
return -EINVAL;
}
v = val->val;
rc = update_val_res(c, w, r, v);
return rc;
}
int update_resvals(cluster_t* c, worker_t* w, int nres, xce_resval_t* vals)
{
int rc = 0;
int i;
for (i = 0; i < nres; i++) {
rc = update_resval(c, w, &(vals[i]));
if (rc < 0) {
PERROR("update_resval");
}
}
return 0;
}
int update_resval_list(cluster_t* c, worker_t* w, xce_resval_list_t* l)
{
int rc = 0;
int i;
for (i = 0; i < l->count; i++) {
rc = update_resval(c, w, l->vals[i]);
if (rc < 0) {
PERROR("update_resval");
}
}
return 0;
}
static int mark_host_res(cluster_t* c, worker_t* w, xce_config_t* config)
{
xce_resource_t* r = NULL;
char* n = config->name;
int rc;
/* clang-format-off */
xce_val_t val;
/* clang-format-on */
lock(c->resmgr, RTTI_LOCK_READ);
rc = key_find(c->resmgr, (void*)n, (void**)&r);
unlock(c->resmgr);
if (rc < 0) {
return rc;
}
if (r == NULL) {
ERROR("resource `%s' can not be found", n);
return -ENOENT;
}
if (r->idx < 0) {
ERROR("invalid resource definition for `%s'", n);
return -EINVAL;
}
if (r->flags & XCE_RESF_SHARED) {
ERROR("resource `%s' should not be a shared resource", n);
return -EINVAL;
}
switch (r->valtype) {
case XCE_VAL_STRING:
val.which = Value_u_svalue;
if (config->value == NULL) {
val.u.svalue = "";
} else {
val.u.svalue = config->value;
}
break;
case XCE_VAL_NUMERIC:
val.which = Value_u_fvalue;
if (config->value == NULL) {
val.u.fvalue = 0.0;
} else {
val.u.fvalue = atof(config->value);
}
break;
case XCE_VAL_BOOLEAN:
val.which = Value_u_bvalue;
val.u.bvalue = 1;
break;
}
rc = update_val_res(c, w, r, &val);
return rc;
}
static int add_hostattrs_resvals(cluster_t* c, worker_t* w)
{
#define N_HOST_ATTRS (8)
xce_val_t values[N_HOST_ATTRS] = {
{ .which = Value_u_svalue, .u = { .svalue = w->host.attrs->hosttype } },
{ .which = Value_u_svalue,
.u = { .svalue = w->host.attrs->hostmodel } },
{ .which = Value_u_fvalue,
.u = { .fvalue = w->host.attrs->cpufactor } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxcpus } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxmem } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxswap } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxtmp } },
{ .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->ndisks } }
};
xce_resval_t resvals[N_HOST_ATTRS] = {
{ .name = "type", .val = &values[0] },
{ .name = "model", .val = &values[1] },
{ .name = "cpuf", .val = &values[2] },
{ .name = "maxcpus", .val = &values[3] },
{ .name = "maxmem", .val = &values[4] },
{ .name = "maxswap", .val = &values[5] },
{ .name = "maxtmp", .val = &values[6] },
{ .name = "ndisks", .val = &values[7] }
};
return update_resvals(c, w, N_HOST_ATTRS, &(resvals[0]));
}
static int set_host_slots(cluster_t* c, worker_t* w, int nslots)
{
/* clang-format off */
xce_val_t values[1] = {
{
.which = Value_u_fvalue,
.u = {
.fvalue = (float)nslots
}
}
};
xce_resval_t resvals[1] = {
{
.name = "slots",
.val = &values[0]
}
};
/* clang-format on */
return update_resvals(c, w, 1, &(resvals[0]));
}
/* 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
{
int nslots;
int i;
w->host.status &= ~XCE_WORKER_UNAVAIL;
w->ep = ep;
if (strcmp(hconfig->hosttype, "!") != 0) {
dereference(w->host.attrs->hosttype);
w->host.attrs->hosttype = makeString(hconfig->hosttype);
}
if (strcmp(hconfig->hostmodel, "!") != 0) {
dereference(w->host.attrs->hostmodel);
w->host.attrs->hostmodel = makeString(hconfig->hostmodel);
}
add_hostattrs_resvals(c, w);
nslots = hconfig->mxj;
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)
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;
}
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);
if (rc < 0)
return rc;
return w->idx;
}
static void reset_hostattrs_unknown(hostattrs_t* a)
{
dereference(a->hosttype);
a->hosttype = makeString("unknown");
dereference(a->hostmodel);
a->hostmodel = makeString("unknown");
a->cpufactor = 0;
a->maxcpus = 0;
a->maxmem = 0;
a->maxswap = 0;
a->maxtmp = 0;
a->ndisks = 0;
}
int restore_worker(void* cls, xce_worker_t host)
{
cluster_t* c = (cluster_t*)cls;
int idx = 0;
worker_t* w = NULL;
w = makeWorker(c->workermgr, &host);
if (w == NULL) {
return -errno;
}
lock(w, RTTI_LOCK_WRITE);
if (w->host.attrs != NULL)
reset_hostattrs_unknown(w->host.attrs);
add_hostattrs_resvals(c, w);
idx = w->idx;
w->host.status |= XCE_WORKER_UNAVAIL;
unlock(w);
return idx;
}
/*
* restore_host_state — restore a host admin-state frame (config-upgrade-
* snapshot Requirements 1.10, 1.11, 5.8).
*
* Unlike restore_queue_state() (queues are config-provisioned, so an
* update-only lookup is always safe), a worker_t is join-provisioned: it
* may not exist yet on a member that hasn't locally replayed this host's
* original join record (e.g. one that received the upgrade snapshot via
* the out-of-band install path). So this restore path is create-if-missing:
*
* - Hit (host already registered -- a real joined worker, or an earlier
* placeholder left by a prior restore attempt): OR the restored admin
* bits into the existing status only. Nothing else is touched, so a
* real worker's live attrs/ep/resource values survive untouched.
* - Miss: create a minimal placeholder via makeWorker(), mirroring the
* synthetic lost_and_found sentinel host precedent exactly
* (config_repl.c:1114-1130) -- the same "makeWorker() without a join"
* mechanism, not a new one. The placeholder is marked
* XCE_WORKER_UNAVAIL (unjoined, dead-until-connected) and carries the
* restored admin bits. Its hostattrs_t is non-NULL with valid sentinel
* strings (copy_worker, wobj.c:53, requires non-NULL attrs; downstream
* apply_join_state()/add_hostattrs_resvals() treat hosttype/hostmodel
* as always-valid strings) so the host's later real join -- which
* join.c routes to rejoin_worker(), unmodified, because key_find() now
* finds this placeholder -- doesn't crash on rejoin_worker's direct
* attrs->hosttype dereference (worker.c:376-379).
*
* Idempotent under retry (5.8): find first via our own key_find, create
* only if truly missing. A second call for the same host (e.g. a retried
* restore after an earlier attempt failed partway through the snapshot)
* finds either the real worker or the already-created placeholder and just
* OR's the bits in again -- never a duplicate-create conflict.
*
* admin_status is masked to XCE_WORKER_ADMIN_MASK defensively (mirrors the
* collect side's own discipline, collect_host_frames only ever encoding
* admin-masked bits) so a caller passing an unmasked status word can never
* leak a non-admin bit (e.g. XCE_WORKER_BUSY) onto the restored host.
*/
int restore_host_state(cluster_t* c, const char* name, uint32_t admin_status)
{
worker_t* w = NULL;
int rc = 0;
if (name == NULL || name[0] == '\0')
return -EINVAL;
if (c == NULL || c->workermgr == NULL)
return -EINVAL;
admin_status &= XCE_WORKER_ADMIN_MASK;
/* No reference() pin across the workermgr-unlock/w-lock gap (unlike
* join.c's live-join path): this is called from checkpoint restore
* (ha_handle_snapshot_chunk -> persist_load_snapshot), which DOES run
* on the live vtask pool, not a quiesced phase, so a real first-time
* join for the same host CAN race this call. That race is still safe:
* a concurrent init_worker/makeWorker on a miss here would itself
* find-or-create via the same key_find/makeWorker machinery under
* workermgr's own lock, and either ordering converges to a worker
* with the admin bits OR'd in and UNAVAIL cleared by the real join's
* own apply_join_state() call -- the same "outer key_find, then
* unconditionally call into makeWorker" shape already used elsewhere
* in this codebase (worker_from_capnp/restore_worker,
* create_lost_and_found). Mirrors restore_queue_state's (queue.c:208)
* lock idiom, not a claim that this path is concurrency-free. */
lock(c->workermgr, RTTI_LOCK_READ);
rc = key_find(c->workermgr, (void*)name, (void**)&w);
unlock(c->workermgr);
if (rc < 0 && rc != -ENOENT)
return rc;
if (w != NULL) {
/* Hit: real joined worker or an earlier placeholder -- only OR the
* admin bits in, never touch anything else. */
lock(w, RTTI_LOCK_WRITE);
w->host.status |= admin_status;
unlock(w);
return 0;
}
/* Miss: create a minimal, valid placeholder -- mirrors the
* lost_and_found sentinel host precedent (config_repl.c:1114-1130). */
{
hostattrs_t placeholder_attrs = { .hosttype = "unknown",
.hostmodel = "unknown" };
xce_worker_t placeholder_host = { .name = (char*)name,
.attrs = &placeholder_attrs,
.status = XCE_WORKER_UNAVAIL
| admin_status };
w = makeWorker(c->workermgr, &placeholder_host);
if (w == NULL)
return -errno;
}
return 0;
}
struct worker_config_publish_args {
cluster_t* c;
endpoint_t* ep;
};
static int check_conf4worker_(xce_config_t* c)
{
if (c != NULL) {
return 1;
}
return 0;
}
static int worker_push_config(cluster_t* c, endpoint_t* ep)
{
xce_configdata_t configdata;
int rc = 0;
xlist_t clist;
memset(&configdata, 0x0, sizeof(configdata));
configdata.data.conf4worker = calloc(1, sizeof(xce_conf4worker_t));
if (configdata.data.conf4worker == NULL) {
return -ENOMEM;
}
xlist_init(&clist, 256);
collect_confvars(c, (collect_check_fn_t)check_conf4worker_, &clist);
configdata.data.conf4worker->n_confvars = clist.count;
configdata.data.conf4worker->confvars = (xce_config_t**)clist.data;
configdata.data_which = ConfigData_data_workerconfig;
configdata.data.conf4worker->clustername
= c->myconf->cluster_config->clustername;
rc = xce_pushconfig_req(ep, c->myhost, &configdata);
FREEUP(configdata.data.conf4worker);
xlist_fini(&clist);
return rc;
}
/* consider to push jobs on hosts to */
int on_worker_join(void* cls, endpoint_t* ep)
{
cluster_t* c = (cluster_t*)cls;
int rc = 0;
rc = worker_push_config(c, ep);
return rc;
}
int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val)
{
xce_resource_t* r = NULL;
zarray_t* array = NULL;
xce_val_t* v = NULL;
worker_t* w = NULL;
int rc = 0;
lock(c->resmgr, RTTI_LOCK_READ);
rc = key_find(c->resmgr, (void*)resname, (void**)&r);
unlock(c->resmgr);
if (rc < 0) {
return rc;
}
if (r == NULL) {
return -ENOENT;
}
if ((r->flags & XCE_RESF_SHARED) == 0) {
if ((host == NULL) || (strlen(host) == 0)) {
return -EINVAL;
}
lock(c->workermgr, RTTI_LOCK_READ);
rc = key_find(c->workermgr, (void*)host, (void**)&w);
if (w != NULL) {
reference(w);
}
unlock(c->workermgr);
if (rc < 0) {
return rc;
}
if (w == NULL) {
return -ENOENT;
}
array = &(w->host.resvals);
} else {
array = &(c->sharedinst);
}
if (r->idx < array->caps) {
v = (xce_val_t*)array->data[r->idx];
}
if (v == NULL) {
if (w != NULL) {
dereference(w);
}
return -ENOENT;
}
memcpy(val, v, sizeof(*val));
if (w != NULL) {
dereference(w);
}
return 1;
}
int mark_worker_unavail(cluster_t* c, worker_t* w)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT,
HOOK_DETAIL_WORKER_DEAD, c, w);
w->host.status |= XCE_WORKER_UNAVAIL;
if (w->host.dead_since == 0)
w->host.dead_since = time(NULL);
if (w->host.attrs != NULL) {
reset_hostattrs_unknown(w->host.attrs);
add_hostattrs_resvals(c, w);
set_host_slots(c, w, 0);
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT,
HOOK_DETAIL_WORKER_DEAD, c, w);
return 0;
}
int mark_worker_back(cluster_t* c, worker_t* w)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT,
HOOK_DETAIL_WORKER_UPDATE, c, w);
w->host.status &= ~XCE_WORKER_UNAVAIL;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT,
HOOK_DETAIL_WORKER_UPDATE, c, w);
return 0;
}
int handle_worker_leave(cluster_t* c, int idx)
{
worker_t* w = NULL;
int rc;
TRACE("%s worker %d broken connection", __func__, idx);
if (c == NULL) {
SET_ERROR(EINVAL);
return -errno;
}
if (idx < 0) {
SET_ERROR(EINVAL);
return -errno;
}
rc = vget(c->workermgr, idx, (void**)&w);
if (rc < 0) {
return rc;
}
if (w == NULL) {
ERROR("fail to find worker by id %d", idx);
SET_ERROR(E_XCE_BAD_WORKER);
return -errno;
}
lock(w, RTTI_LOCK_WRITE);
w->ep = NULL;
mark_worker_unavail(c, w);
unlock(w);
return 0;
}
int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl)
{
worker_t* w = NULL;
int rc = 0;
if (!c || !ctrl || !ctrl->hostname || *ctrl->hostname == '\0') {
TRACE("%s invalid arguments.", __func__);
return -EINVAL;
}
TRACE("%s worker_ctrl worker %s", __func__, ctrl->hostname);
lock(c->workermgr, RTTI_LOCK_READ);
rc = key_find(c->workermgr, (void*)ctrl->hostname, (void**)&w);
if (w != NULL) {
reference(w);
}
unlock(c->workermgr);
if (rc < 0) {
return rc;
}
if (w == NULL) {
PERROR("workermgr_lock");
return -ENOENT;
}
/* bug-447: key_find returns 1 on found (binary librtti, unnormalizable
* upstream). Normalize the found-rc to 0 here so the AFTER_EVENT hook
* gate below fires (WAL append + replication) and the handler's
* commit-barrier pattern sees 0. Real errors are untouched: rc < 0
* returned above, the w == NULL miss returned -ENOENT, and control-
* action failures set rc themselves. */
rc = 0;
lock(w, RTTI_LOCK_WRITE);
switch (ctrl->action) {
case HOST_OPEN:
if (!(w->host.status & XCE_WORKER_CLOSED)) {
goto clean;
}
w->host.status &= ~(XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN);
break;
case HOST_CLOSE:
if (w->host.status & XCE_WORKER_CLOSED) {
goto clean;
}
w->host.status |= (XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN);
break;
default:
rc = -1;
break;
}
if (rc == 0) {
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER,
HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w);
}
clean:
unlock(w);
dereference(w);
return rc;
}
int do_hostcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
int rc = 0;
xce_reply_t reply;
init_reply_by_req(&reply, request);
rc = worker_ctrl(c, request->body.hostctrlmsg);
/* Commit barrier (ha-commit-barrier-rollout Req 1.1-1.4): the success
* reply is sent only after the operation's WAL entry is majority-
* committed; barrier failures (-ETIMEDOUT, ...) become the error
* reply. Errors from worker_ctrl reply immediately — no barrier wait.
* worker_ctrl released all entity locks before returning, so blocking
* here holds no lock. timeout 0 = configured sync_timeout_ms. */
if (rc == 0) {
rc = persist_commit_barrier(&c->persist, 0);
if (rc != 0)
WARN("%s commit barrier failed, rc %d", __func__, rc);
}
reply.status = rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return rc;
}
/* Reconstruct a worker from a WAL_WORKER_CTRL Cap'n Proto payload.
* Payload layout: [uint32_t detail][Cap'n Proto xce_worker_t].
*
* If the worker is already registered:
* - HOOK_DETAIL_WORKER_JOIN: a duplicate JOIN replay is genuinely a
* no-op -- returns -EEXIST, no mutation (unchanged, pre-bug-1472
* behavior).
* - HOOK_DETAIL_WORKER_UPDATE: the target already existing is the
* NORMAL case (worker_ctrl()'s HOST_OPEN/HOST_CLOSE, worker.c:732-794,
* only ever runs against an already-joined host). bug-1472: replace
* (not OR -- worker_ctrl uses assignment-style clear/set, so a
* HOST_OPEN must be able to CLEAR the closed bits) the existing
* worker's admin-masked status bits with the decoded payload's, then
* return 0 so the caller (persist.c's WAL_WORKER_CTRL apply case)
* falls through to fire the HOOK_DETAIL_WORKER_UPDATE notification
* instead of silently discarding the change.
* - HOOK_DETAIL_WORKER_DEAD: left as -EEXIST (unchanged) -- see
* buglog.json bug-1472 for why DEAD's replication needs its own,
* separately-tracked fix: its mutation isn't confined to
* XCE_WORKER_ADMIN_MASK (XCE_WORKER_UNAVAIL + dead_since + hostattrs
* reset), and the live per-host resource-value array
* (xce_worker_t.resvals, a zarray_t) that mark_worker_unavail()
* zeroes via set_host_slots() never round-trips through the Worker
* capnp schema at all (only resvals_to_sched does) -- a naive
* status-replace here would be an incomplete fix. */
int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len)
{
if (!capnp_payload || len <= sizeof(uint32_t))
return -EINVAL;
uint32_t hook_detail;
memcpy(&hook_detail, capnp_payload, sizeof(hook_detail));
/* Skip the 4-byte detail discriminator written by persist_worker_hook_fun */
const uint8_t* data = (const uint8_t*)capnp_payload + sizeof(uint32_t);
size_t data_len = len - sizeof(uint32_t);
struct capn c_ctx;
capn_init_mem(&c_ctx, data, data_len, 0);
Worker_ptr wp;
wp.p = capn_getp(capn_root(&c_ctx), 0, 1);
struct Worker cw;
xce_worker_t host;
memset(&host, 0, sizeof(host));
read_Worker(&cw, wp);
decode_Worker(&host, &cw);
capn_free(&c_ctx);
/* Idempotency guard: an existing worker means either a duplicate JOIN
* replay (no-op) or an UPDATE targeting an already-joined host (the
* normal case -- must be applied, see bug-1472). */
worker_t* existing = NULL;
lock(c->workermgr, RTTI_LOCK_READ);
int found = key_find(c->workermgr, (void*)host.name, (void**)&existing);
unlock(c->workermgr);
if (found > 0 && existing != NULL) {
if (hook_detail == HOOK_DETAIL_WORKER_UPDATE) {
lock(existing, RTTI_LOCK_WRITE);
existing->host.status
= (existing->host.status & ~XCE_WORKER_ADMIN_MASK)
| (host.status & XCE_WORKER_ADMIN_MASK);
unlock(existing);
free_Worker(&host);
return 0;
}
return -EEXIST;
}
return restore_worker(c, host);
}
/* worker_should_retain lives in persist.c —
* it only depends on xce_worker_t fields and persist_config_t,
* and must be accessible from the unit test binary which links persist.c. */
Reference in a new issue View git blame Copy permalink