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

feat(scheduler-core): add controller-side OPCode_schedAdmin dispatch (task 10.1)

Browse source 
Adds the spec's one and only src/controller/ change: a new
case OPCode_schedAdmin in myreq_handler() (cluster.c) that mirrors the
existing OPCode_upgradeConfig case -- primary-only gate, explicit reply
on every path, stateless pass-through with no WAL entry.

Forwards to the scheduler via a new xce_schedadmin_req() (async send,
mirrors xce_fetchfile_req) and blocks the vtask thread on a condvar with
a 5s deadline while holding c->sched.lock for the round-trip, matching
this spec's own research-phase design decision (parity with
OPCode_upgradeConfig's no-RBAC posture, blocking wait accepted for
low-frequency admin traffic). A new branch in the existing
OPCode_reply case demuxes the scheduler's SchedAdminReply to a
process-global single-slot pending-waiter mechanism, serialized by
c->sched.lock, matching job.c's existing fetchfile-proxy pattern.

Adversarial review independently traced double-free/cleanup, the
timeout/signal race, and reply-opcode demux correctness against source;
found and a follow-up round closed a narrow-window leak via a carefully
lock-ordered unlock->unregister->relock sequence (avoiding an AB-BA
inversion against schedadmin_signal()'s lock acquisition order).

New test_sched_admin_dispatch.c drives the real myreq_handler via the
existing myreq_handler_for_test seam: not-primary rejection (proven via
a poison-pointer sched.ep that would crash if forwarding were ever
attempted), not-connected rejection, and a genuine two-thread round-trip
test exercising the real signal/demux path.

Controller suite 3/3 (1090/1090 assertions), scheduler suite 26/26
unaffected, clang-format clean.
This commit is contained in:
Rongsong Shen 2026-08-06 03:38:42 +00:00
parent 2d533e652e
commit 045be6b3d6
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
7 changed files with 598 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

252
src/controller/cluster.c
View file

@ -2,9 +2,11 @@
#include <errno.h>
#include <inttypes.h>
#include <pthread.h>
#include <rtti.h>
#include <stdatomic.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include "cluster.h"
@ -235,6 +237,78 @@ static int handle_snapshot_transfer(cluster_t *c, xce_request_t *req)
return 0;
}
/* ---- OPCode_schedAdmin: forward-and-block-for-reply support (task 10.1) --
*
* The controller is a stateless pass-through for OPCode_schedAdmin: it
* forwards a SchedAdminRequest to the scheduler over c->sched.ep and must
* relay the scheduler's SchedAdminReply back to the original caller, which
* means the request-handling vtask thread has to block until that reply
* arrives (or 5s elapses) -- unlike every other controller->scheduler call
* in this file (sched.c's xce_push_*_req / xce_pushconfig_req), which are
* fire-and-forget.
*
* This mirrors on_fetchfile's proxy pattern (job.c): register a waiter
* BEFORE sending (avoids a race with a reply that arrives before the wait
* starts), send async via the normal xce_request() path, then block on a
* condvar with a deadline. The one difference from fetchfile's keyed
* (jobid) multi-waiter list: this needs only a SINGLE slot, because the
* schedAdmin case below holds c->sched.lock for the entire forward+wait
* duration, which already serializes at most one in-flight schedAdmin
* round-trip per controller at a time (an accepted trade-off -- xceadmin
* admin commands are low-frequency, human-paced traffic, not a hot path;
* see design research for task 10.1). */
typedef struct {
pthread_mutex_t mu;
pthread_cond_t cv;
int done;
int status;
sched_admin_reply_t* reply;
} schedadmin_pending_t;
static pthread_mutex_t g_schedadmin_mu = PTHREAD_MUTEX_INITIALIZER;
static schedadmin_pending_t* g_schedadmin_pending = NULL;
static void schedadmin_register(schedadmin_pending_t* p)
{
pthread_mutex_lock(&g_schedadmin_mu);
g_schedadmin_pending = p;
pthread_mutex_unlock(&g_schedadmin_mu);
}
static void schedadmin_unregister(schedadmin_pending_t* p)
{
pthread_mutex_lock(&g_schedadmin_mu);
if (g_schedadmin_pending == p)
g_schedadmin_pending = NULL;
pthread_mutex_unlock(&g_schedadmin_mu);
}
/* Signal the pending schedAdmin waiter (if any) with the scheduler's reply.
* Called from the OPCode_reply case below when a SchedAdminReply arrives on
* c->sched.ep. Returns 1 if a waiter was found and signaled (ownership of
* `sar` transfers to it -- the caller must NULL its own copy so the normal
* xce_free_request path does not also free it), 0 if no waiter was
* registered (e.g. a late reply arriving after this controller already gave
* up and replied "request timeout" to xceadmin -- `sar` stays owned by the
* caller in that case, freed normally). */
static int schedadmin_signal(int status, sched_admin_reply_t* sar)
{
pthread_mutex_lock(&g_schedadmin_mu);
schedadmin_pending_t* p = g_schedadmin_pending;
if (p == NULL) {
pthread_mutex_unlock(&g_schedadmin_mu);
return 0;
}
pthread_mutex_lock(&p->mu);
p->status = status;
p->reply = sar;
p->done = 1;
pthread_cond_signal(&p->cv);
pthread_mutex_unlock(&p->mu);
pthread_mutex_unlock(&g_schedadmin_mu);
return 1;
}
static int myreq_handler(vtask_t* task)
{
vtask_data_t* data = (vtask_data_t*)task->data;
@ -714,8 +788,162 @@ static int myreq_handler(vtask_t* task)
}
break;
#endif
case OPCode_schedAdmin: {
/* Stateless pass-through forward to the scheduler (task 10.1, Req
* 5.10). The controller commits no WAL entry and interprets no
* field beyond the opcode -- every field of the request/reply
* passes through untouched. Structurally mirrors OPCode_upgradeConfig
* above: primary-only gate, explicit reply on every path, never a
* dropped request. See the schedadmin_* helpers above this function
* for the forward-and-block-for-reply mechanism (mirrors
* on_fetchfile's proxy pattern in job.c). */
sched_admin_req_t* areq = request.body.schedadminmsg;
/* Defensive belt-and-suspenders guard -- mirrors OPCode_upgradeConfig's
* identical inner check above. The real gate is ha_dispatch_gate
* (called before this switch, ~line 276): OPCode_schedAdmin is
* absent from the CTRL_STANDBY / CTRL_INIT / CTRL_ELECTING
* allow-lists in ha.c -- exactly like OPCode_upgradeConfig -- so a
* non-primary controller is rejected with no forwarding attempt
* before this switch is ever reached; this check only guards a
* role change racing between that gate and here. */
if (!ha_is_primary(&c->ha)) {
reply.status = -EAGAIN;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
rc = -EAGAIN;
break;
}
if (areq == NULL) {
reply.status = -EINVAL;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
rc = -EINVAL;
break;
}
/* c->sched.lock is held for the FULL forward+wait duration below --
* an accepted trade-off (task 10.1 design notes): xceadmin admin
* commands are low-frequency, human-paced traffic, not a hot path.
* This also naturally serializes at most one in-flight schedAdmin
* round-trip at a time, which the single-slot pending waiter above
* relies on. */
mylock_lock(&(c->sched.lock));
if (c->sched.ep == NULL) {
sched_admin_reply_t nc_reply;
memset(&nc_reply, 0, sizeof(nc_reply));
nc_reply.error = "scheduler not connected";
mylock_unlock(&(c->sched.lock));
reply.status = 0;
reply.data_which = RequestReply_data_schedadminreplymsg;
reply.data.schedadminreplymsg = &nc_reply;
xce_reply(ep, c->myhost, &reply);
rc = 0;
break;
}
schedadmin_pending_t pending = { .mu = PTHREAD_MUTEX_INITIALIZER,
.cv = PTHREAD_COND_INITIALIZER,
.done = 0,
.status = -ETIMEDOUT,
.reply = NULL };
/* Register before sending -- avoid a race with a reply that arrives
* before pthread_cond_timedwait below starts waiting (same
* discipline as fetchfile_register, job.c). */
schedadmin_register(&pending);
rc = xce_schedadmin_req(c->sched.ep, areq);
if (rc < 0) {
schedadmin_unregister(&pending);
mylock_unlock(&(c->sched.lock));
reply.status = rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
break;
}
struct timespec deadline;
clock_gettime(CLOCK_REALTIME, &deadline);
deadline.tv_sec += 5;
pthread_mutex_lock(&pending.mu);
while (!pending.done) {
int wrc
= pthread_cond_timedwait(&pending.cv, &pending.mu, &deadline);
if (wrc == ETIMEDOUT)
break;
}
pthread_mutex_unlock(&pending.mu);
/* Unregister immediately after releasing pending.mu, BEFORE reading
* the outcome below -- closes the window where a late-arriving
* reply could still find this slot registered and hand off `sar`
* ownership into it after we have already stopped looking (the
* abandoned-slot leak this fix addresses).
*
* This must NOT be done while still holding pending.mu:
* schedadmin_unregister() only takes g_schedadmin_mu (confirmed by
* its body above -- it never touches p->mu), but schedadmin_signal()
* takes g_schedadmin_mu THEN p->mu. Calling unregister while holding
* pending.mu would acquire the two locks in the reverse order
* (pending.mu then g_schedadmin_mu) from a concurrent
* schedadmin_signal() (g_schedadmin_mu then pending.mu) -- a classic
* AB-BA lock-ordering inversion that can deadlock. Releasing
* pending.mu first keeps both threads' lock order consistent
* (g_schedadmin_mu is always acquired without holding pending.mu).
*
* Once schedadmin_unregister() returns, no NEW signal can find this
* slot. Re-acquiring pending.mu below picks up the result of any
* signal that was already in flight (had read g_schedadmin_pending
* before we cleared it) and finished its write before we got here --
* mutual exclusion on g_schedadmin_mu guarantees that race resolves
* one way or the other before schedadmin_unregister() returns. */
schedadmin_unregister(&pending);
pthread_mutex_lock(&pending.mu);
int got_done = pending.done;
int got_status = pending.status;
sched_admin_reply_t* got_reply = pending.reply;
pthread_mutex_unlock(&pending.mu);
if (got_done && got_reply != NULL) {
/* Relay the scheduler's SchedAdminReply back to xceadmin
* verbatim -- every field passes through untouched. */
reply.status = 0;
reply.data_which = RequestReply_data_schedadminreplymsg;
reply.data.schedadminreplymsg = got_reply;
xce_reply(ep, c->myhost, &reply);
free_SchedAdminReply_ptr(&got_reply);
} else if (got_done) {
/* Scheduler replied at the RPC level with no SchedAdminReply
* payload (unexpected, but handled loudly, never dropped) --
* relay its RPC status. */
reply.status = got_status;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
} else {
sched_admin_reply_t to_reply;
memset(&to_reply, 0, sizeof(to_reply));
to_reply.error = "request timeout";
reply.status = 0;
reply.data_which = RequestReply_data_schedadminreplymsg;
reply.data.schedadminreplymsg = &to_reply;
xce_reply(ep, c->myhost, &reply);
}
mylock_unlock(&(c->sched.lock));
rc = 0;
break;
}
case OPCode_reply: {
/* Reply arriving on an inbound endpoint. Two independent producers
/* Reply arriving on an inbound endpoint. Three independent producers
* feed this case:
* 1. fetchfile proxy replies from a worker (fetchfile feature) —
* routed to the pending waiter via fetchfile_signal.
@ -726,7 +954,10 @@ static int myreq_handler(vtask_t* task)
* replicator so the configAck barrier can observe it
* (ha-commit-barrier-rollout task 5.2 remediation r1; same
* ingestion as repl_ack_decode_and_dispatch in ha.c — completes
* the unlanded half of ha-config-bootstrap task 8.1). */
* the unlanded half of ha-config-bootstrap task 8.1).
* 3. SchedAdminReply arriving on c->sched.ep in response to the
* OPCode_schedAdmin case's forwarded request — routed to the
* pending waiter via schedadmin_signal (task 10.1). */
xce_reply_t* workreply = request.body.replymsg;
if (workreply != NULL && workreply->opcode == OPCode_fetchfile) {
@ -742,6 +973,23 @@ static int myreq_handler(vtask_t* task)
break;
}
if (workreply != NULL && workreply->opcode == OPCode_schedAdmin) {
sched_admin_reply_t* sar = (workreply->data_which
== RequestReply_data_schedadminreplymsg)
? workreply->data.schedadminreplymsg
: NULL;
if (schedadmin_signal(workreply->status, sar)) {
/* Ownership of sar (if any) transferred to the pending
* waiter — NULL it so xce_free_request below does not also
* free it. If no waiter was found (e.g. this reply arrived
* after the forwarder already gave up at the 5s timeout),
* leave it owned by workreply for the normal free path. */
workreply->data.schedadminreplymsg = NULL;
}
rc = 0;
break;
}
if (request.body_which == Request_body_replymsg
&& workreply != NULL
&& workreply->data_which