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

working on scheduler and fix issue #17

Browse source
This commit is contained in:
Rongsong Shen 2025-12-25 16:24:56 +08:00
parent 4034fd7556
commit a46ea822c1
10 changed files with 159 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

43
src/controller/job.c
View file

@ -418,6 +418,13 @@ int handle_jobdispatch(cluster_t* c, endpoint_t* ep, xce_request_t* request)
return rc;
}
static int xlist_deref_job_item_(job_t* j, void* arg)
{
obj_deref(j);
return 0;
}
int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
uint64_t jobid;
@ -425,6 +432,7 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xce_reply_t reply;
xlist_t jlist;
job_t* j = NULL;
int i;
submit = request->body.jobsubmsg;
@ -455,26 +463,31 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xce_reply(ep, c->myhost, &reply);
/* Run hooks to tell scheduler */
j = jobmgr_lockjob(&(c->jobmgr), jobid, OBJREF_READ);
if (j == NULL) {
char idstring[128];
xlist_init(&jlist, submit->array_size);
fmt_jobid(idstring, sizeof(idstring) - 1, jobid);
ERROR("%s fail to lock job %s (%lu)", __func__, idstring, jobid);
} else {
xlist_init(&jlist, 2);
if (xlist_store(&jlist, j) >= 0) {
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB,
HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_NEW, c, &jlist);
for (i = 0; i < submit->array_size; i++) {
/* Run hooks to tell scheduler */
j = jobmgr_lockjob(&(c->jobmgr), jobid + i, OBJREF_READ);
if (j == NULL) {
char idstring[128];
fmt_jobid(idstring, sizeof(idstring) - 1, jobid);
ERROR("%s fail to lock job %s (%lu)", __func__, idstring, jobid);
} else {
PERROR("xlist_store");
if (xlist_store(&jlist, j) < 0) {
PERROR("xlist_store");
obj_deref(j);
}
}
obj_deref(j);
xlist_fini(&jlist);
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_NEW, c, &jlist);
xlist_iter(&jlist, (xlist_iter_fn_t)xlist_deref_job_item_, NULL);
xlist_fini(&jlist);
return 0;
}

7
src/include/utils.h
View file

@ -121,10 +121,9 @@ extern xce_timewindow_config_t* dup_timewindow_config(xce_timewindow_config_t*);
extern void free_resource_thresholds(xce_res_threshold_t**, int);
extern xce_res_threshold_t** dup_resource_thresholds(xce_res_threshold_t**,
int);
extern void free_resval_with_idx_list(xce_val_with_idx_t***, int);
extern void free_strlist(xce_strlist_t*);
extern int fmt_jobid_ex(char*, size_t, uint64_t, char, char);
extern void free_resval_with_idx_list(xce_val_with_idx_t***, int);
extern void free_strlist(xce_strlist_t*);
extern int fmt_jobid_ex(char*, size_t, uint64_t, char*, char*);
extern int fmt_jobid(char*, size_t, uint64_t);
extern int parse_jobid(uint64_t*, char*);
extern uint64_t array_jobid(uint64_t);

6
src/lib/utils.c
View file

@ -436,7 +436,7 @@ void free_strlist(xce_strlist_t* l)
FREEUP(l->data);
}
int fmt_jobid_ex(char* buf, size_t sz, uint64_t jobid, char lc, char rc)
int fmt_jobid_ex(char* buf, size_t sz, uint64_t jobid, char* lc, char* rc)
{
uint32_t tag, id_cluster, id_apiserver, id_timestamp;
char data[128];
@ -455,7 +455,7 @@ int fmt_jobid_ex(char* buf, size_t sz, uint64_t jobid, char lc, char rc)
id_job = EXTRACT_ARRAY_JOB(jobid);
id_idx = EXTRACT_ARRAY_IDX(jobid);
snprintf(data, sizeof(data) - 1, "%d-%d-%d-%d%c%d%c", id_cluster,
snprintf(data, sizeof(data) - 1, "%d-%d-%d-%d%s%d%s", id_cluster,
id_apiserver, id_timestamp, id_job, lc, id_idx, rc);
} else {
uint32_t id_seq;
@ -476,7 +476,7 @@ int fmt_jobid_ex(char* buf, size_t sz, uint64_t jobid, char lc, char rc)
int fmt_jobid(char* buf, size_t sz, uint64_t jobid)
{
return fmt_jobid_ex(buf, sz, jobid, '[', ']');
return fmt_jobid_ex(buf, sz, jobid, "[", "]");
}
int parse_jobid(uint64_t* jobid, char* buf)

14
src/scheduler/include/plugin.h
View file

@ -19,6 +19,8 @@ typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*,
xce_sched_job_t*);
typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*,
xce_sched_worker_t*);
typedef int (*plugin_job_state_change_hook_t)(void*, xce_sched_job_t*,
uint32_t);
typedef int (*plugin_commit_decision_hook_t)(void*, xce_decision_t*);
/* initialize function of plugin */
@ -37,11 +39,12 @@ typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t,
xce_pendreason_t**);
typedef struct {
plugin_update_config_hook_t config_hook;
plugin_update_queue_hook_t queue_hook;
plugin_update_job_hook_t job_hook;
plugin_update_worker_hook_t worker_hook;
plugin_commit_decision_hook_t commit_hook;
plugin_update_config_hook_t config_hook;
plugin_update_queue_hook_t queue_hook;
plugin_update_job_hook_t job_hook;
plugin_job_state_change_hook_t jstate_hook;
plugin_update_worker_hook_t worker_hook;
plugin_commit_decision_hook_t commit_hook;
} plugin_hooks_t;
typedef struct {
@ -64,6 +67,7 @@ extern int plugin_queue_update(void*, xce_sched_queue_t*, xce_sched_queue_t*);
extern int plugin_job_update(void*, xce_sched_job_t*, xce_sched_job_t*);
extern int plugin_worker_update(void*, xce_sched_worker_t*,
xce_sched_worker_t*);
extern int plugin_job_state_change(void*, xce_sched_job_t*, uint32_t);
extern int plugin_begin_session(void*);
extern int plugin_end_session(void*);
extern int plugin_commit_decision(void*, xce_decision_t*);

5
src/scheduler/include/scheduler.h
View file

@ -19,6 +19,7 @@ typedef enum {
XCE_SCHED_OP_NONE = 0,
XCE_SCHED_OP_UPDATE_CONFIG,
XCE_SCHED_OP_UPDATE_JOB,
XCE_SCHED_OP_JSTATE_CHANGE,
XCE_SCHED_OP_UPDATE_WORKER
} xce_sched_op_t;
@ -35,6 +36,10 @@ typedef struct {
int nworkers;
xce_worker_t** workers;
} workerupdate;
struct {
uint64_t jobid;
uint32_t state;
} job_state_change;
} u;
} xce_sched_update_request_t;

39
src/scheduler/src/handler.c
View file

@ -124,26 +124,49 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset)
return rc;
}
static int sched_job_state_change(xce_scheduler_t* sched, uint64_t jobid,
uint32_t state)
{
char jobidString[128];
xce_sched_update_request_t *req = NULL;
fmt_jobid(jobidString, sizeof(jobidString) - 1, jobid);
TRACE("%s %s state %u ", __func__, jobidString, state);
req = (xce_sched_update_request_t *)calloc(1, sizeof(*req));
if (req == NULL) {
return -ENOMEM;
}
req->op = XCE_SCHED_OP_JSTATE_CHANGE;
req->u.job_state_change.jobid = jobid;
req->u.job_state_change.state = state;
enq_pend_update(sched, req);
return 0;
}
int sched_handle_notify(xce_scheduler_t* sched, xce_notify_data_t* notify)
{
int rc = 0;
int rc = 0;
char jobidString[128];
switch (notify->event) {
case NotifyData_data_schedStartJob:
fmt_jobid(jobidString, sizeof(jobidString)-1, notify->data.sched_start_job);
TRACE("%s %s started", __func__, jobidString);
rc = sched_job_state_change(sched, notify->data.sched_start_job,
JOB_STAT_RUN);
break;
case NotifyData_data_schedExitJob:
fmt_jobid(jobidString, sizeof(jobidString) - 1, notify->data.sched_exit_job);
TRACE("%s %s exit", __func__, jobidString);
rc = sched_job_state_change(sched, notify->data.sched_exit_job,
JOB_STAT_EXIT);
break;
case NotifyData_data_schedFailJob:
fmt_jobid(jobidString, sizeof(jobidString) - 1, notify->data.sched_fail_job);
TRACE("%s %s fail", __func__, jobidString);
rc = sched_job_state_change(sched, notify->data.sched_fail_job,
JOB_STAT_EXIT);
break;
case NotifyData_data_schedUpdateJob:
fmt_jobid(jobidString, sizeof(jobidString) - 1, notify->data.sched_update_job->jobid);
fmt_jobid(jobidString, sizeof(jobidString) - 1,
notify->data.sched_update_job->jobid);
TRACE("%s %s updated", __func__, jobidString);
break;
case NotifyData_data_schedUpdateQueue:

41
src/scheduler/src/plugin.c
View file

@ -304,3 +304,44 @@ int plugin_commit_decision(void* arg1, xce_decision_t* decision)
return rc;
}
typedef struct {
xce_sched_job_t* job;
uint32_t state;
} job_state_change_t;
static int plugin_job_state_change_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
job_state_change_t* args = (job_state_change_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.jstate_hook != NULL) {
return plugin->hooks.jstate_hook(plugin, args->job, args->state);
}
return 0;
}
int plugin_job_state_change(void* arg, xce_sched_job_t* job, uint32_t state)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
int rc;
job_state_change_t args = { .job = job, .state = state };
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_job_state_change_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
xlist_fini(&lst);
return rc;
}

21
src/scheduler/src/plugins/simple_allocator.c
View file

@ -100,13 +100,26 @@ static int my_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision)
return rc;
}
static int my_jstate_changed(xce_sched_plugin_t* plugin, xce_sched_job_t* job,
uint32_t state)
{
if (JOB_IS_RUN(job)
&& ((state == JOB_STAT_EXIT) || (state == JOB_STAT_DONE))) {
slotsInUse--;
}
return 0;
}
int register_simple_allocator(void* sched)
{
static xce_sched_plugin_t simple_allocator_plugin = {
.name = "xce_sched_simple_allocator",
.hooks = { .config_hook = NULL,
.job_hook = (plugin_update_job_hook_t)my_update_job,
.commit_hook = (plugin_commit_decision_hook_t)my_commit },
.name = "xce_sched_simple_allocator",
.hooks
= { .config_hook = NULL,
.job_hook = (plugin_update_job_hook_t)my_update_job,
.jstate_hook = (plugin_job_state_change_hook_t)my_jstate_changed,
.commit_hook = (plugin_commit_decision_hook_t)my_commit },
.init_fn = (plugin_init_fn_t)simple_allocator_init,
.fini_fn = (plugin_fini_fn_t)simple_allocator_fini,
.begin_session_fn = NULL,

21
src/scheduler/src/scheduler.c
View file

@ -123,6 +123,8 @@ static void sched_pend_update_killer(data_item_t* it)
free(req->u.workerupdate.workers);
break;
case XCE_SCHED_OP_JSTATE_CHANGE:
break;
default:
break;
}
@ -281,6 +283,21 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs,
return 0;
}
static int sched_job_state_change(xce_scheduler_t* sched, uint64_t jobid,
uint32_t state)
{
xce_sched_job_t* job = sched_get_job(sched, jobid);
int rc = 0;
if (job == NULL) {
return -ENOENT;
}
rc = plugin_job_state_change(sched, job, state);
job->state = state;
return rc;
}
static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
xce_worker_t** workers)
{
@ -334,6 +351,10 @@ int sched_update(xce_scheduler_t* sched)
rc = sched_update_workers(sched, req->u.workerupdate.nworkers,
req->u.workerupdate.workers);
break;
case XCE_SCHED_OP_JSTATE_CHANGE:
rc = sched_job_state_change(sched, req->u.job_state_change.jobid,
req->u.job_state_change.state);
break;
default:
rc = -EINVAL;
break;

2
src/worker/src/xcecd_worker_utils.c
View file

@ -523,7 +523,7 @@ int make_job_sd_unit_name(char* buf, size_t bufsize, const char* prefix,
return -EINVAL;
}
fmt_jobid_ex(idstring, sizeof(idstring) - 1, jobid, '_', '_');
fmt_jobid_ex(idstring, sizeof(idstring) - 1, jobid, "A", "");
return snprintf(buf, bufsize, "%s-%s.service", prefix, idstring);
}