wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity
myxce/src/controller/apiserver.c
Rongsong Shen e53c086ec5 fix(schema): rename struct Data -> XceData to unshadow capnp builtin blob type 
controller.capnp declared `struct Data $C.mapname("xce_data_t")` (an
OPCode_get query-result tagged union) at file scope, which shadows
Cap'n Proto's builtin Data blob type for every later `: Data` field in
the same file. This silently miscompiled 7 fields across 6 structs as
a struct-pointer type instead of a raw byte blob: UpgradeConfigRequest
.sha256/.bytes, ReplConfigReply.sha256/.bytes, ReplConfigPushRequest
.sha256/.bytes, ReplEnvelope.payload, SnapshotTransfer's chunk
payload, WalCatchupEntry.payload/.batchPayloads, and CtrlHelloRequest
.key. Production C code had already routed around it for years via
hand-rolled raw capn_ptr manipulation (see request.c's own comments);
nothing before now needed the generated codec/accessors to actually
work for these fields.

Renamed the local struct to XceData (capnp-level name only; the
$C.mapname("xce_data_t") C typedef is unchanged) and explicitly
repointed getdata/pushmsg at XceData by name, since those two fields
correctly want the local tagged union. Updated every derived
generated-identifier reference across 19 hand-written files spanning
controller, lib, scheduler, and the Go apiserver/xcecli CGo bindings.
Fixed 5 call sites in lib/request.c that had been typed against the
old workaround alias but now legitimately receive the correct
capn_data type (a pure type-name swap -- Data_ptr/XceData_ptr and
capn_data share the identical {capn_ptr p} layout).

Verified via full rebuild+test: lib (1/1), controller (3/3, incl. the
job-array push regression tests), scheduler (15/15), Go apiserver +
xcecli (all packages green), and a direct pycapnp round-trip proving
UpgradeConfigRequest.sha256/.bytes now encode/decode correctly.

Bug: bug-1461 (.wolf/buglog.json). Unblocks admin-config-tool task 3.3.
2026-07-22 09:22:04 +00:00

702 lines
22 KiB
C
Raw Blame History

#include <errno.h>
#include <stdlib.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "request.h"
#include "xce/types.h"
#include "xstring.h"
static int xce_apiserver_push_configs(cluster_t* c, endpoint_t* ep)
{
xce_configdata_t configdata;
int rc = 0;
memset(&configdata, 0x0, sizeof(configdata));
configdata.data.clusterconfig = c->myconf->cluster_config;
configdata.data_which = ConfigData_data_clusterconfig;
rc = xce_pushconfig_req(ep, c->myhost, &configdata);
return rc;
}
static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep)
{
xce_data_t data;
xce_workers_data_t workers
= { .n_total = 0, .pos = 0, .count = 0, .workers = NULL };
int i, rc = 0;
xlist_t wlist;
memset(&data, 0x0, sizeof(data));
xlist_init(&wlist, 8192);
data.table = XceData_data_workers;
data.data.workers = &workers;
collect_workers(c, (collect_check_fn_t)check_worker_, &wlist);
workers.n_total = wlist.count;
workers.workers = (xce_worker_t**)wlist.data;
workers.count = workers.n_total;
rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);
/* free up collected resource value which used
passed to api server
*/
for (i = 0; i < workers.n_total; i++) {
xce_worker_t* w = workers.workers[i];
free_name_val_list(&(w->resvals_to_sched), &(w->nresvals_to_sched));
}
unlock_collect_workers(&wlist);
FREEUP(workers.workers);
return rc;
}
static int check_job_(xce_job_t* j)
{
if (j == NULL) {
return 0;
}
return 1;
}
static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep)
{
xce_data_t data;
xce_jobs_data_t jobs = { .n_total = 0, .pos = 0, .count = 0, .jobs = NULL };
int rc = 0;
xlist_t jlist;
memset(&data, 0x0, sizeof(data));
xlist_init(&jlist, 1024 * 1024);
data.table = XceData_data_jobs;
data.data.jobs = &jobs;
collect_jobs(c, (collect_check_fn_t)check_job_, &jlist);
jobs.n_total = jlist.count;
jobs.jobs = (xce_job_t**)jlist.data;
jobs.count = jobs.n_total;
rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);
unlock_collect_jobs(&jlist);
FREEUP(jobs.jobs);
return rc;
}
static int apiserver_push_joblist(cluster_t* c, xlist_t* l)
{
int i;
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) {
int rc;
xce_data_t data;
xce_jobs_data_t jobs = { .n_total = l->count,
.pos = 0,
.count = l->count,
.jobs = (xce_job_t**)l->data };
memset(&data, 0x0, sizeof(data));
data.table = XceData_data_jobs;
data.data.jobs = &jobs;
rc = xce_push_data_req(c->apiservers[i].ep, c->myhost, &data,
check_controller, c, 100);
if (rc < 0) {
WARN("%s push job list to api server %s has problem, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
return 0;
}
/* job-array-single-wal-record task 5.2 (design.md "Compact Scheduler/
* API-Server Push", File Structure Plan apiserver.c bullet; Req 6.2, 6.4):
* encodes ONE compact (base jobid, array size, one shared spec) Data.jobArray
* push per connected API server -- reusing the JobSubmit capnp struct
* (already added to the Data union in task 1.2, controller.capnp), never N
* individually-encoded Job structs. (base_jobid, array_size) are derived
* from the already-constructed xlist_t the HOOK_DETAIL_JOB_NEW fire carries
* (apply_committed_entry's WAL_JOB_ARRAY_SUBMIT case, task 3.2): the first
* element's jobid is the array base and l->count is the array size; every
* element shares the same submit spec by construction (restore_job_array's
* own construction discipline, job.c), so any one element's submit pointer
* is the one shared spec to encode.
*
* Sent via xce_request() directly (OPCode_push / Request_body_pushmsg)
* rather than xce_push_data_req(), mirroring sched.c's xcesched_push_
* jobarray() (task 5.1) for the identical reason: that helper's dispatch
* switch has explicit cases only for jobs/workers/resources/resvals and
* hard-rejects (-EINVAL) anything else via its default case -- it cannot
* carry a single non-list Data.jobArray descriptor. This mirrors xce_push_
* req()'s (request.c, static) exact shape for a single non-list xce_data_t
* payload.
*
* Fan-out mirrors apiserver_push_joblist()'s existing per-connected-server
* loop exactly (same lock discipline, same name!=NULL && ep!=NULL guard) --
* one push sent per connected server, not one push shared/broadcast. */
static int apiserver_push_jobarray(cluster_t* c, xlist_t* l)
{
xce_job_t* first;
xce_job_submit_t job_array;
int i;
if ((c == NULL) || (l == NULL) || (l->count <= 0)) {
return -EINVAL;
}
first = (xce_job_t*)xlist_get(l, 0);
if ((first == NULL) || (first->submit == NULL)) {
return -EINVAL;
}
memset(&job_array, 0x0, sizeof(job_array));
job_array.jobid = first->jobid;
job_array.array_size = (uint64_t)l->count;
job_array.submit = first->submit;
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) {
xce_data_t data;
xce_request_t request;
int rc;
memset(&data, 0x0, sizeof(data));
data.table = XceData_data_jobArray;
data.data.job_array = &job_array;
memset(&request, 0x0, sizeof(request));
request.opcode = OPCode_push;
request.body_which = Request_body_pushmsg;
request.body.pushmsg = &data;
rc = xce_request(c->apiservers[i].ep, &request);
if (rc < 0) {
WARN("%s push job array to api server %s has problem, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
return 0;
}
static int apiserver_job_new_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
xlist_t* l = va_arg(ap, xlist_t*);
int rc = 0;
switch (situation) {
case HOOK_AFTER_EVENT:
rc = -EINVAL;
if (l != NULL) {
/* job-array-single-wal-record task 5.2 (Req 6.2, 6.4):
* apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT case is the
* ONLY way this hook is ever fired with more than one element
* (Req 3.4 -- one HOOK_DETAIL_JOB_NEW fire per committed array
* record, carrying every constructed element). Route that case
* through the compact descriptor push instead of the per-job
* push. The ordinary single-job branch (l->count == 1) below is
* the exact, unmodified call this function always made
* (Req 8.1). */
rc = (l->count > 1) ? apiserver_push_jobarray(c, l)
: apiserver_push_joblist(c, l);
}
break;
default:
break;
}
return rc;
}
static int apiserver_job_start_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
job_pidinfo_t* info = va_arg(ap, job_pidinfo_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata = { .event = NotifyData_data_jobPidInfo,
.data = { .pidinfo = info } };
/* FIXME: add job pid to API server? */
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify job start to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_exit_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
xce_jobfinish_t* info = va_arg(ap, xce_jobfinish_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_jobFinishInfo,
.data = { .finishinfo = info } };
/* FIXME: we need to send exit code of job to API server */
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify job exit to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_fail_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
job_failinfo_t* info = va_arg(ap, job_failinfo_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_jobFailInfo,
.data = { .failinfo = info } };
/* FIXME: we need to add fail reason of job to api server */
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify job fail to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_update_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
job_t* job = va_arg(ap, job_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_updateJob,
.data = { .update_job = &(job->job) } };
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify job update to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_rusage_update_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
xce_job_rusages_t* rusages = va_arg(ap, xce_job_rusages_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedJobRusages,
.data = { .sched_job_rusages = rusages } };
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify rusage update to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_hook_fun(int type, int situation, int detail,
void* data, va_list ap)
{
switch (detail) {
case HOOK_DETAIL_JOB_NEW:
return apiserver_job_new_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_START:
return apiserver_job_start_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_EXIT:
return apiserver_job_exit_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_UPDATE:
return apiserver_job_update_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_FAIL:
return apiserver_job_fail_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_RUSAGE_UPDATE:
return apiserver_job_rusage_update_fun(type, situation, data, ap);
}
return 0;
}
static int apiserver_push_workerlist(cluster_t* c, endpoint_t* ep, xlist_t* l)
{
xce_data_t data;
xce_workers_data_t workers = { .n_total = l->count,
.pos = 0,
.count = l->count,
.workers = (xce_worker_t**)l->data };
int rc = 0;
memset(&data, 0x0, sizeof(data));
data.table = XceData_data_workers;
data.data.workers = &workers;
rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);
return rc;
}
static int apiserver_worker_change(int type, int situation, void* data,
cluster_t* c, worker_t* w)
{
int i;
xlist_t resvalist;
TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx,
situation);
xlist_init(&resvalist, 256);
collect_resval(c->resmgr, &(w->host.resvals), &resvalist);
w->host.nresvals_to_sched = resvalist.count;
w->host.resvals_to_sched = (xce_name_val_t**)resvalist.data;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
xlist_t wlist;
int rc;
xlist_init(&wlist, 2);
xlist_store(&wlist, w);
rc = apiserver_push_workerlist(c, c->apiservers[i].ep, &wlist);
if (rc < 0) {
WARN("%s notify worker change failed to api server %s, rc "
"%d",
__func__, c->apiservers[i].name, rc);
}
xlist_fini(&wlist);
}
mylock_unlock(&(c->apiservers[i].lock));
}
break;
default:
break;
}
w->host.nresvals_to_sched = 0;
w->host.resvals_to_sched = NULL;
xlist_fini(&resvalist);
return 0;
}
static int apiserver_worker_join(int type, int situation, void* data,
va_list ap)
{
/* Unified JOIN signature: (cluster_t*, char* name, worker_t*)
* w is NULL for BEFORE_EVENT (worker not yet created). */
cluster_t* c = va_arg(ap, cluster_t*);
char* name = va_arg(ap, char*);
worker_t* w = va_arg(ap, worker_t*);
(void)name;
if (w == NULL)
return 0;
return apiserver_worker_change(type, situation, data, c, w);
}
static int apiserver_worker_dead(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
worker_t* w = va_arg(ap, worker_t*);
/* FIXME: need to handle job status in this worker */
if (situation == HOOK_BEFORE_EVENT) {
return 0;
}
return apiserver_worker_change(type, situation, data, c, w);
}
static int apiserver_worker_update(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
worker_t* w = va_arg(ap, worker_t*);
if (situation == HOOK_BEFORE_EVENT) {
return 0;
}
return apiserver_worker_change(type, situation, data, c, w);
}
static int apiserver_worker_hook_fun(int type, int situation, int detail,
void* data, va_list ap)
{
switch (detail) {
case HOOK_DETAIL_WORKER_JOIN:
return apiserver_worker_join(type, situation, data, ap);
case HOOK_DETAIL_WORKER_DEAD:
return apiserver_worker_dead(type, situation, data, ap);
case HOOK_DETAIL_WORKER_UPDATE:
return apiserver_worker_update(type, situation, data, ap);
}
return 0;
}
static int apiserver_resource_hook_fun(int type, int situation, int detail,
void* data, va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
xce_resval_list_t* resvals = va_arg(ap, xce_resval_list_t*);
int i;
if (detail != HOOK_DETAIL_RESOURCE_UPDATE) {
return 0;
}
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedUpdateResvals,
.data = { .sched_update_resvals = resvals } };
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify api server %s resource update fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
break;
default:
break;
}
return 0;
}
static int apiserver_queue_hook_fun(int type, int situation, int detail,
void* data, va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
queue_t* q = va_arg(ap, queue_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_lock(&(c->apiservers[i].lock));
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedUpdateQueue,
.data = { .sched_update_queue = &(q->queue) } };
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN(
"%s notify queue change to api server %s failed, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
mylock_unlock(&(c->apiservers[i].lock));
}
break;
default:
break;
}
return 0;
}
int apiserver_init(cluster_t* c)
{
int i;
for (i = 0; i < N_APISERVER_MAX; i++) {
mylock_init(&(c->apiservers[i].lock));
c->apiservers[i].name = NULL;
c->apiservers[i].ep = NULL;
}
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB,
(hook_fn_t)apiserver_job_hook_fun, NULL);
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER,
(hook_fn_t)apiserver_worker_hook_fun, NULL);
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_RESOURCE,
(hook_fn_t)apiserver_resource_hook_fun, NULL);
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE,
(hook_fn_t)apiserver_queue_hook_fun, NULL);
return 0;
}
int on_apiserver_join(cluster_t* c, endpoint_t* ep)
{
xce_apiserver_push_configs(c, ep);
xce_apiserver_push_workers(c, ep);
xce_apiserver_push_jobs(c, ep);
return 0;
}
int handle_apiserver_leave(cluster_t* c, endpoint_t* ep)
{
ep_private_t* priv;
int i;
if (ep == NULL) {
return -EINVAL;
}
priv = (ep_private_t*)ep->data;
if (priv == NULL) {
return -EINVAL;
}
i = priv->u.idx;
if (i < 0) {
SET_ERROR(EINVAL);
return -errno;
}
if (c->apiservers[i].name != NULL) {
c->apiservers[i].ep = NULL;
}
/* FIXME: kill ep? */
return 0;
}
int add_apiserver(cluster_t* c, char* name, endpoint_t* ep)
{
int i;
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (strcmp(c->apiservers[i].name, name) == 0)) {
if (c->apiservers[i].ep != NULL) {
/* FIXME: we need to delete 1 */
}
c->apiservers[i].ep = ep;
return i;
}
}
for (i = 0; i < N_APISERVER_MAX; i++) {
if (c->apiservers[i].name == NULL) {
c->apiservers[i].name = makeString(name);
c->apiservers[i].ep = ep;
return i;
}
}
/* better error code */
return -1;
}
View git blame Copy permalink