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

feat: filter lost_and_found sentinel from pushes; reset unavail host attrs

Browse source 
- create_lost_and_found(): provide hostattrs_t so makeWorker succeeds
- collect_queues/collect_workers (dataex.c): skip lost_and_found sentinel
  so it is never pushed to apiserver or scheduler
- mark_worker_unavail(): reset hosttype/hostmodel to "unknown" and zero
  all numeric attrs, then update resvals and slots — prevents stale
  capacity from being visible while the host is down
- restore_worker(): apply the same attr reset on WAL recovery so
  restored-but-unavailable workers advertise no capacity until reconnect;
  also remove dead calloc() that was immediately overwritten by makeWorker
- Extract reset_hostattrs_unknown() helper to eliminate copy-paste
- Define XCE_LOST_AND_FOUND_NAME constant in config_repl.h; replace all
  six raw "lost_and_found" string literals across config_repl.c and dataex.c
This commit is contained in:
Rongsong Shen 2026-05-14 08:44:55 +00:00
parent 2e2aaea78c
commit 1599d06bc4
6 changed files with 79 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

76
src/controller/config_repl.c
View file

@ -498,14 +498,14 @@ int config_contains_reserved_name(cluster_config_t *cfg)
for (int i = 0; i < cfg->nqueues; i++) {
if (cfg->queues[i] &&
cfg->queues[i]->queue &&
strcmp(cfg->queues[i]->queue, "lost_and_found") == 0)
strcmp(cfg->queues[i]->queue, XCE_LOST_AND_FOUND_NAME) == 0)
return 1;
}
/* Check host config names — xce_hostconfig_t.name */
for (int i = 0; i < cfg->nhconfigs; i++) {
if (cfg->hconfigs[i] &&
cfg->hconfigs[i]->name &&
strcmp(cfg->hconfigs[i]->name, "lost_and_found") == 0)
strcmp(cfg->hconfigs[i]->name, XCE_LOST_AND_FOUND_NAME) == 0)
return 1;
}
return 0;
@ -551,9 +551,9 @@ void create_lost_and_found(cluster_t *c)
* unknown queue IDs here instead of failing. */
{
void *existing = NULL;
if (key_find(c->qmgr, (void *)"lost_and_found", &existing) <= 0) {
if (key_find(c->qmgr, (void *)XCE_LOST_AND_FOUND_NAME, &existing) <= 0) {
xce_queue_t lf = {0};
lf.queue = "lost_and_found";
lf.queue = XCE_LOST_AND_FOUND_NAME;
queue_t *q = create_queue(c, &lf);
if (q)
INFO("Created synthetic 'lost_and_found' queue");
@ -567,9 +567,11 @@ void create_lost_and_found(cluster_t *c)
* for this sentinel host. */
{
void *existing = NULL;
if (key_find(c->workermgr, (void *)"lost_and_found", &existing) <= 0) {
xce_worker_t lf_host = {0};
lf_host.name = "lost_and_found";
if (key_find(c->workermgr, (void *)XCE_LOST_AND_FOUND_NAME, &existing) <= 0) {
hostattrs_t lf_attrs = { .hosttype = XCE_LOST_AND_FOUND_NAME,
.hostmodel = XCE_LOST_AND_FOUND_NAME };
xce_worker_t lf_host = { .name = XCE_LOST_AND_FOUND_NAME,
.attrs = &lf_attrs };
worker_t *w = makeWorker(c->workermgr, &lf_host);
if (w)
INFO("Created synthetic 'lost_and_found' host");
@ -780,15 +782,13 @@ typedef struct {
const uint8_t *bytes;
size_t blen;
const uint8_t *sha256;
atomic_int *stop;
int acked;
int started;
} push_ctx_t;
static void *push_thread(void *arg)
{
push_ctx_t *ctx = (push_ctx_t *)arg;
if (atomic_load(ctx->stop))
return NULL;
int rc = xce_replconfigpush_req_raw(
ctx->ep, ctx->from,
@ -810,71 +810,51 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
if (!c->ha.standby_endpoints)
return 0;
/* Snapshot connected standbys under lock */
pthread_mutex_lock(&c->ha.lock);
int n = c->ha.standby_endpoints->count;
if (n == 0) {
pthread_mutex_unlock(&c->ha.lock);
return 0;
}
endpoint_t **eps = malloc((size_t)n * sizeof(endpoint_t *));
if (!eps) {
push_ctx_t *ctxs = calloc((size_t)n, sizeof(push_ctx_t));
if (!ctxs) {
pthread_mutex_unlock(&c->ha.lock);
return 0;
}
for (int i = 0; i < n; i++)
eps[i] = (endpoint_t *)zarray_get(c->ha.standby_endpoints, i);
pthread_mutex_unlock(&c->ha.lock);
atomic_int stop = 0;
push_ctx_t *ctxs = calloc((size_t)n, sizeof(push_ctx_t));
pthread_t *threads = malloc((size_t)n * sizeof(pthread_t));
if (!ctxs || !threads) {
free(ctxs); free(threads); free(eps);
return 0;
}
for (int i = 0; i < n; i++) {
ctxs[i].ep = eps[i];
ctxs[i].ep = (endpoint_t *)zarray_get(c->ha.standby_endpoints, i);
ctxs[i].from = c->myhost;
ctxs[i].new_tag = new_tag;
ctxs[i].bytes = bytes;
ctxs[i].blen = blen;
ctxs[i].sha256 = sha256;
ctxs[i].stop = &stop;
ctxs[i].acked = 0;
pthread_create(&threads[i], NULL, push_thread, &ctxs[i]);
}
pthread_mutex_unlock(&c->ha.lock);
pthread_t *threads = malloc((size_t)n * sizeof(pthread_t));
if (!threads) {
free(ctxs);
return 0;
}
/* Aggregate wait: CONF_PUSH_TIMEOUT_MS total, not per-standby */
struct timespec deadline;
clock_gettime(CLOCK_MONOTONIC, &deadline);
deadline.tv_sec += CONF_PUSH_TIMEOUT_MS / 1000;
deadline.tv_nsec += (CONF_PUSH_TIMEOUT_MS % 1000) * 1000000L;
if (deadline.tv_nsec >= 1000000000L) {
deadline.tv_sec++;
deadline.tv_nsec -= 1000000000L;
}
while (1) {
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
if (now.tv_sec > deadline.tv_sec ||
(now.tv_sec == deadline.tv_sec && now.tv_nsec >= deadline.tv_nsec))
break;
struct timespec sl = { .tv_sec = 0, .tv_nsec = 5 * 1000000L };
nanosleep(&sl, NULL);
for (int i = 0; i < n; i++) {
ctxs[i].started = (pthread_create(&threads[i], NULL, push_thread, &ctxs[i]) == 0);
if (!ctxs[i].started)
WARN("push_config_to_standbys: pthread_create failed for standby %d", i);
}
atomic_store(&stop, 1);
int acked = 0;
for (int i = 0; i < n; i++) {
if (!ctxs[i].started)
continue;
pthread_join(threads[i], NULL);
if (ctxs[i].acked) acked++;
}
INFO("Config pushed to %d/%d standbys", acked, n);
free(ctxs); free(threads); free(eps);
free(ctxs);
free(threads);
return acked;
}