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

improve api server support

Browse source
This commit is contained in:
Rongsong Shen 2026-01-23 17:33:10 +08:00
parent 3b728e5135
commit 9e07a1b227
7 changed files with 141 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

8
src/apiserver/cmd/main.go
View file

@ -127,16 +127,16 @@ func (apiserver *ApiServer) handle_worker_push(workers []*C.xce_worker_t) C.int
func (apiserver *ApiServer)handle_dataset(dataset *C.dataset_t) C.int {
fmt.Printf("handle dataset table %d\n", dataset.table)
fmt.Printf("handle dataset table `%s'\n", C.GoString(C.xce_dataset_table(dataset.table)))
switch(dataset.table) {
case C.Data_data_schedjobs:
case C.Data_data_jobs:
fmt.Printf("%d jobs arrived\n", dataset.nrows)
var theJobs **C.xce_job_t = C.dataset_jobs(dataset)
jobs := unsafe.Slice(theJobs, dataset.nrows)
return apiserver.handle_job_push(jobs)
case C.Data_data_schedworkers:
case C.Data_data_workers:
fmt.Printf("%d workers arrived\n", dataset.nrows)
var theWorkers **C.xce_worker_t = C.dataset_workers(dataset)
@ -153,7 +153,7 @@ func go_req_handler(request *C.xce_request_t, ctx *C.apiserver_ctx_t) C.int {
var apiserver = (*ApiServer)(unsafe.Pointer(ctx.go_apiserver))
fmt.Println("api server handler called")
fmt.Printf("op `%s' body_which %d from %s\n", C.GoString(C.xce_request_op_name(request)), body_which, from)
fmt.Printf("op `%s' msg `%s' from %s\n", C.GoString(C.xce_request_op_name(request)), C.GoString(C.xce_request_msg_name(request)), from)
switch (request.opcode) {
case C.OPCode_push:

43
src/controller/apiserver.c
View file

@ -147,9 +147,8 @@ static int apiserver_job_start_fun(int type, int situation, void* data,
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 } };
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);
@ -461,7 +460,7 @@ static int apiserver_resource_hook_fun(int type, int situation, int detail,
= { .event = NotifyData_data_schedUpdateResvals,
.data = { .sched_update_resvals = resvals } };
rc = notify(c, c->sched.ep, &xdata);
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);
@ -509,27 +508,33 @@ static int apiserver_queue_hook_fun(int type, int situation, int detail,
return 0;
}
int apiserver_init(cluster_t* c)
{
int i;
for(i = 0; i < N_APISERVER_MAX; i ++) {
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)
{
/* FIXME: use lock to protect this later */
static int hook_installed = 0;
xce_apiserver_push_configs(c, ep);
xce_apiserver_push_workers(c, ep);
xce_apiserver_push_jobs(c, ep);
if (!hook_installed) {
hook_installed = 1;
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;
}

2
src/controller/cluster.c
View file

@ -542,6 +542,8 @@ int cluster_init(cluster_t* c)
delaymgr_init(&(c->delaymgr), 2, run_delay_tasks);
apiserver_init(c);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT,
HOOK_DETAIL_CLUSTER_INIT, c);

1
src/controller/include/cluster.h
View file

@ -138,5 +138,6 @@ extern int do_set_preason(cluster_t*, endpoint_t*, xce_request_t*);
extern int do_clr_preason(cluster_t*, endpoint_t*, xce_request_t*);
extern int handle_apiserver_leave(cluster_t*, endpoint_t*);
extern int add_apiserver(cluster_t*, char*, endpoint_t*);
extern int apiserver_init(cluster_t*);
#endif

2
src/include/helper.h
View file

@ -9,6 +9,7 @@
extern void* xce_request_body(xce_request_t*);
extern char* xce_request_op_name(xce_request_t*);
extern char* xce_request_msg_name(xce_request_t*);
extern void* xce_val_ptr(xce_val_t*);
extern void* xce_val_ptr(xce_val_t*);
extern void* xce_configdata_ptr(xce_configdata_t*);
@ -17,5 +18,6 @@ extern void* xce_update_data_ptr(xce_update_data_t*);
extern void* xce_notify_data_ptr(xce_notify_data_t*);
extern void* xce_jobctrl_data_ptr(xce_jobctrl_t*);
extern void* xce_reply_data_ptr(xce_reply_t*);
extern char* xce_dataset_table(int);
#endif

117
src/lib/helper.c
View file

@ -52,6 +52,52 @@ void* xce_request_body(xce_request_t* req)
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_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) {
@ -104,8 +150,9 @@ char* xce_request_op_name(xce_request_t* req)
return "unknown";
}
void *xce_val_ptr(xce_val_t *v) {
switch(v->which) {
void* xce_val_ptr(xce_val_t* v)
{
switch (v->which) {
case Value_u_ivalue32:
return &(v->u.ivalue32);
case Value_u_uvalue32:
@ -127,8 +174,9 @@ void *xce_val_ptr(xce_val_t *v) {
return NULL;
}
void* xce_configdata_ptr(xce_configdata_t *d) {
switch(d->data_which) {
void* xce_configdata_ptr(xce_configdata_t* d)
{
switch (d->data_which) {
case ConfigData_data_schedconfig:
return d->data.conf4sched;
case ConfigData_data_workerconfig:
@ -142,8 +190,9 @@ void* xce_configdata_ptr(xce_configdata_t *d) {
return NULL;
}
void* xce_data_ptr(xce_data_t *d) {
switch(d->table) {
void* xce_data_ptr(xce_data_t* d)
{
switch (d->table) {
case Data_data_schedjobs:
return d->data.jobs;
case Data_data_schedworkers:
@ -162,8 +211,9 @@ void* xce_data_ptr(xce_data_t *d) {
return NULL;
}
void* xce_update_data_ptr(xce_update_data_t *d) {
switch(d->table) {
void* xce_update_data_ptr(xce_update_data_t* d)
{
switch (d->table) {
case Update_data_nodata:
break;
case Update_data_resvals:
@ -176,8 +226,9 @@ void* xce_update_data_ptr(xce_update_data_t *d) {
return NULL;
}
void* xce_notify_data_ptr(xce_notify_data_t *d) {
switch(d->event) {
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:
@ -187,19 +238,19 @@ void* xce_notify_data_ptr(xce_notify_data_t *d) {
case NotifyData_data_jobFinishInfo:
return d->data.finishinfo;
case NotifyData_data_schedStartJob:
return (void *)d->data.sched_start_job;
return (void*)d->data.sched_start_job;
case NotifyData_data_schedExitJob:
return (void *)d->data.sched_exit_job;
return (void*)d->data.sched_exit_job;
case NotifyData_data_schedFailJob:
return (void *)d->data.sched_fail_job;
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;
return (void*)d->data.stopped_job;
case NotifyData_data_resumedJob:
return (void *)d->data.resumed_job;
return (void*)d->data.resumed_job;
case NotifyData_data_ckptinfo:
return d->data.ckptinfo;
case NotifyData_data_schedUpdateQueue:
@ -211,7 +262,7 @@ void* xce_notify_data_ptr(xce_notify_data_t *d) {
case NotifyData_data_taskFinish:
return d->data.taskfinish;
case NotifyData_data_postDoneJob:
return (void *)d->data.pdone_job;
return (void*)d->data.pdone_job;
case NotifyData_data_schedJobRusages:
return d->data.sched_job_rusages;
default:
@ -220,8 +271,9 @@ void* xce_notify_data_ptr(xce_notify_data_t *d) {
return NULL;
}
void* xce_jobctrl_data_ptr(xce_jobctrl_t *d) {
switch(d->action) {
void* xce_jobctrl_data_ptr(xce_jobctrl_t* d)
{
switch (d->action) {
case JobControl_u_siginfo:
return d->u.siginfo;
case JobControl_u_ckptoption:
@ -232,8 +284,9 @@ void* xce_jobctrl_data_ptr(xce_jobctrl_t *d) {
return NULL;
}
void* xce_reply_data_ptr(xce_reply_t *reply) {
switch(reply->data_which) {
void* xce_reply_data_ptr(xce_reply_t* reply)
{
switch (reply->data_which) {
case RequestReply_data_nodata:
break;
case RequestReply_data_debugmsg:
@ -241,7 +294,7 @@ void* xce_reply_data_ptr(xce_reply_t *reply) {
case RequestReply_data_getdata:
return reply->data.getdata;
case RequestReply_data_jobid:
return (void *)reply->data.jobid;
return (void*)reply->data.jobid;
case RequestReply_data_clusterconf:
return reply->data.clusterconf;
default:
@ -249,3 +302,25 @@ void* xce_reply_data_ptr(xce_reply_t *reply) {
}
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";
}

12
src/lib/request.c
View file

@ -281,6 +281,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data,
data1.table = data->table;
switch (data->table) {
case Data_data_schedjobs:
case Data_data_jobs:
nrows = data->data.jobs->n_total;
data1.data.jobs = (xce_jobs_data_t*)calloc(1, sizeof(xce_jobs_data_t));
if (data1.data.jobs == NULL) {
@ -288,6 +289,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data,
}
break;
case Data_data_schedworkers:
case Data_data_workers:
nrows = data->data.workers->n_total;
data1.data.workers
= (xce_workers_data_t*)calloc(1, sizeof(xce_workers_data_t));
@ -295,6 +297,10 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data,
rc = -ENOMEM;
}
break;
case Data_data_resources:
break;
case Data_data_resvals:
break;
default:
rc = -EINVAL;
break;
@ -324,6 +330,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data,
switch (data->table) {
case Data_data_schedjobs:
case Data_data_jobs:
data1.data.jobs->n_total = data->data.jobs->n_total;
data1.data.jobs->pos = pos;
data1.data.jobs->count = nc;
@ -334,6 +341,7 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data,
}
break;
case Data_data_schedworkers:
case Data_data_workers:
data1.data.workers->n_total = data->data.workers->n_total;
data1.data.workers->pos = pos;
data1.data.workers->count = nc;
@ -344,6 +352,10 @@ int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data,
= &(data->data.workers->workers[pos]);
}
break;
case Data_data_resources:
break;
case Data_data_resvals:
break;
default:
break;
}