myxce/src/lib/helper.c
Rongsong Shen 3e652d2d7e feat: fill scheduler controller message gaps
Wire up 6 previously-stubbed message handlers in the scheduler:
resource dataset, resval dataset, job update notify, queue update
notify, resval update notify, and job rusage notify. Remove the
unreachable pushmsg branch (vtask_entry rewrites all pushmsg to
datamsg before the handler fires).

Changes by component:
- helper.h/helper.c: add xce_notify_event_name() for symbolic log names
- scheduler.h: add 5 new xce_sched_op_t values and matching union fields
- plugin.h/plugin.c: add plugin_job_rusage_hook_t and plugin_job_rusage_update()
- schedobj.c: add sched_update_existing_queue() with NULL/-ENOENT guards
- scheduler.c: wire 5 new dequeue cases; add sched_update_existing_job()
  and static sched_op_name(); emit WARN log on handler failure
- handler.c: implement sched_handle_resources/resvals (dataset) and
  sched_handle_sched_update_{job,queue,resvals}/job_rusage (notify);
  remove dead sched_handle_jobpush/workerpush and pushmsg branch;
  update TRACE logs to use symbolic name helpers
- test.c: 5 new unit tests covering notify event names, enum values,
  and queue update NULL guard (15/15 pass)
2026-05-12 07:01:52 +00:00

385 lines
9.9 KiB
C

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include "controller.capnp.h"
#include "helper.h"
#include "typemap.h"
#include "xce/types.h"
void* xce_request_body(xce_request_t* req)
{
switch (req->body_which) {
case Request_body_nomsg:
break;
case Request_body_replymsg:
return req->body.replymsg;
case Request_body_hellomsg:
return req->body.hellomsg;
case Request_body_joinmsg:
return req->body.joinmsg;
case Request_body_configpushmsg:
return req->body.configpushmsg;
case Request_body_pushmsg:
return req->body.pushmsg;
case Request_body_getmsg:
return req->body.getmsg;
case Request_body_datamsg:
return req->body.datamsg;
case Request_body_dispatchmsg:
return req->body.dispatchmsg;
case Request_body_jobstartmsg:
return req->body.jobstartmsg;
case Request_body_notifymsg:
return req->body.notifymsg;
case Request_body_jobsubmsg:
return req->body.jobsubmsg;
case Request_body_updatemsg:
return req->body.updatemsg;
case Request_body_jobctrlmsg:
return req->body.jobctrlmsg;
case Request_body_queuectrlmsg:
return req->body.queuectrlmsg;
case Request_body_hostctrlmsg:
return req->body.hostctrlmsg;
case Request_body_runtaskmsg:
return req->body.runtaskmsg;
case Request_body_preasonmsg:
return req->body.preasonmsg;
case Request_body_clrpreasonmsg:
return req->body.clrpreasonmsg;
default:
break;
}
return NULL;
}
char* xce_request_msg_name(xce_request_t* req)
{
switch (req->body_which) {
case Request_body_nomsg:
return "nomsg";
case Request_body_replymsg:
return "replymsg";
case Request_body_hellomsg:
return "hellomsg";
case Request_body_joinmsg:
return "joinmsg";
case Request_body_configpushmsg:
return "configmsg";
case Request_body_pushmsg:
return "pushmsg";
case Request_body_getmsg:
return "getmsg";
case Request_body_datamsg:
return "datamsg";
case Request_body_dispatchmsg:
return "dispatchmsg";
case Request_body_jobstartmsg:
return "jobstartmsg";
case Request_body_notifymsg:
return "notifymsg";
case Request_body_jobsubmsg:
return "jobsubmsg";
case Request_body_updatemsg:
return "updatemsg";
case Request_body_jobctrlmsg:
return "jobctrlmsg";
case Request_body_queuectrlmsg:
return "queuectrlmsg";
case Request_body_hostctrlmsg:
return "hostctrlmsg";
case Request_body_runtaskmsg:
return "runtaskmsg";
case Request_body_preasonmsg:
return "preasonmsg";
case Request_body_clrpreasonmsg:
return "clrpreasonmsg";
default:
break;
}
return "unknown";
}
char* xce_request_op_name(xce_request_t* req)
{
switch (req->opcode) {
case OPCode_nop:
return "nop";
case OPCode_reply:
return "reply";
case OPCode_hello:
return "hello";
case OPCode_join:
return "join";
case OPCode_shutdown:
return "shutdown";
case OPCode_get:
return "get";
case OPCode_push:
return "push";
case OPCode_update:
return "update";
case OPCode_jobdispatch:
return "jobdispatch";
case OPCode_startjob:
return "startjob";
case OPCode_notify:
return "notify";
case OPCode_jobsub:
return "jobsub";
case OPCode_jobstatus:
return "jobstatus";
case OPCode_jobrusage:
return "jobrusage";
case OPCode_hostcontrol:
return "hostcontrol";
case OPCode_jobcontrol:
return "jobcontrol";
case OPCode_clusterconf:
return "clusterconf";
case OPCode_queuecontrol:
return "queuecontrol";
case OPCode_runtasks:
return "runtasks";
case OPCode_setpreason:
return "setpreason";
case OPCode_clrpreason:
return "clrpreason";
default:
break;
}
return "unknown";
}
void* xce_val_ptr(xce_val_t* v)
{
switch (v->which) {
case Value_u_ivalue32:
return &(v->u.ivalue32);
case Value_u_uvalue32:
return &(v->u.uvalue32);
case Value_u_ivalue64:
return &(v->u.ivalue64);
case Value_u_uvalue64:
return &(v->u.uvalue64);
case Value_u_svalue:
return v->u.svalue;
case Value_u_fvalue:
return &(v->u.fvalue);
case Value_u_bvalue:
return &(v->u.bvalue);
default:
break;
}
return NULL;
}
void* xce_configdata_ptr(xce_configdata_t* d)
{
switch (d->data_which) {
case ConfigData_data_schedconfig:
return d->data.conf4sched;
case ConfigData_data_workerconfig:
return d->data.conf4worker;
case ConfigData_data_clusterconfig:
return d->data.clusterconfig;
default:
break;
}
return NULL;
}
void* xce_data_ptr(xce_data_t* d)
{
switch (d->table) {
case Data_data_schedjobs:
return d->data.jobs;
case Data_data_schedworkers:
return d->data.workers;
case Data_data_jobs:
return d->data.jobs;
case Data_data_workers:
return d->data.workers;
case Data_data_resources:
return d->data.resources;
case Data_data_resvals:
return d->data.resvals;
default:
break;
}
return NULL;
}
void* xce_update_data_ptr(xce_update_data_t* d)
{
switch (d->table) {
case Update_data_nodata:
break;
case Update_data_resvals:
return d->data.resvals;
case Update_data_jobrusages:
return d->data.jobrusages;
default:
break;
}
return NULL;
}
void* xce_notify_data_ptr(xce_notify_data_t* d)
{
switch (d->event) {
case NotifyData_data_jobPidInfo:
return d->data.pidinfo;
case NotifyData_data_jobFailInfo:
return d->data.failinfo;
case NotifyData_data_jobDispatchFail:
return d->data.dispatchfail;
case NotifyData_data_jobFinishInfo:
return d->data.finishinfo;
case NotifyData_data_schedStartJob:
return (void*)d->data.sched_start_job;
case NotifyData_data_schedExitJob:
return (void*)d->data.sched_exit_job;
case NotifyData_data_schedFailJob:
return (void*)d->data.sched_fail_job;
case NotifyData_data_schedUpdateJob:
return d->data.sched_update_job;
case NotifyData_data_schedUpdateResvals:
return d->data.sched_update_resvals;
case NotifyData_data_stoppedJob:
return (void*)d->data.stopped_job;
case NotifyData_data_resumedJob:
return (void*)d->data.resumed_job;
case NotifyData_data_ckptinfo:
return d->data.ckptinfo;
case NotifyData_data_schedUpdateQueue:
return d->data.sched_update_queue;
case NotifyData_data_taskPidInfo:
return d->data.taskinfo;
case NotifyData_data_taskFailInfo:
return d->data.taskfailinfo;
case NotifyData_data_taskFinish:
return d->data.taskfinish;
case NotifyData_data_postDoneJob:
return (void*)d->data.pdone_job;
case NotifyData_data_schedJobRusages:
return d->data.sched_job_rusages;
case NotifyData_data_updateJob:
return d->data.update_job;
default:
break;
}
return NULL;
}
void* xce_jobctrl_data_ptr(xce_jobctrl_t* d)
{
switch (d->action) {
case JobControl_u_siginfo:
return d->u.siginfo;
case JobControl_u_ckptoption:
return d->u.ckptoption;
default:
break;
}
return NULL;
}
void* xce_reply_data_ptr(xce_reply_t* reply)
{
switch (reply->data_which) {
case RequestReply_data_nodata:
break;
case RequestReply_data_debugmsg:
return reply->data.debugmsg;
case RequestReply_data_getdata:
return reply->data.getdata;
case RequestReply_data_jobid:
return (void*)reply->data.jobid;
case RequestReply_data_clusterconf:
return reply->data.clusterconf;
default:
break;
}
return NULL;
}
char* xce_dataset_table(int table)
{
switch (table) {
case Data_data_schedjobs:
return "schedjobs";
case Data_data_jobs:
return "jobs";
case Data_data_workers:
return "workers";
case Data_data_schedworkers:
return "schedworkers";
case Data_data_resources:
return "resources";
case Data_data_resvals:
return "resvals";
default:
break;
}
return "unknown";
}
char* xce_notify_event_name(int event)
{
static char buf[32];
switch (event) {
case NotifyData_data_jobPidInfo:
return "jobPidInfo";
case NotifyData_data_jobFailInfo:
return "jobFailInfo";
case NotifyData_data_jobDispatchFail:
return "jobDispatchFail";
case NotifyData_data_jobFinishInfo:
return "jobFinishInfo";
case NotifyData_data_schedStartJob:
return "schedStartJob";
case NotifyData_data_schedExitJob:
return "schedExitJob";
case NotifyData_data_schedFailJob:
return "schedFailJob";
case NotifyData_data_schedUpdateJob:
return "schedUpdateJob";
case NotifyData_data_schedUpdateResvals:
return "schedUpdateResvals";
case NotifyData_data_stoppedJob:
return "stoppedJob";
case NotifyData_data_resumedJob:
return "resumedJob";
case NotifyData_data_ckptinfo:
return "ckptinfo";
case NotifyData_data_schedUpdateQueue:
return "schedUpdateQueue";
case NotifyData_data_taskPidInfo:
return "taskPidInfo";
case NotifyData_data_taskFailInfo:
return "taskFailInfo";
case NotifyData_data_taskFinish:
return "taskFinish";
case NotifyData_data_postDoneJob:
return "postDoneJob";
case NotifyData_data_schedJobRusages:
return "schedJobRusages";
case NotifyData_data_updateJob:
return "updateJob";
default:
break;
}
snprintf(buf, sizeof(buf), "UNKNOWN(%d)", event);
return buf;
}