From a46ea822c1d633ba54e674059fcba9facaf1e241 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 25 Dec 2025 16:24:56 +0800 Subject: [PATCH] working on scheduler and fix issue #17 --- src/controller/job.c | 43 +++++++++++++------- src/include/utils.h | 7 ++-- src/lib/utils.c | 6 +-- src/scheduler/include/plugin.h | 14 ++++--- src/scheduler/include/scheduler.h | 5 +++ src/scheduler/src/handler.c | 39 ++++++++++++++---- src/scheduler/src/plugin.c | 41 +++++++++++++++++++ src/scheduler/src/plugins/simple_allocator.c | 21 ++++++++-- src/scheduler/src/scheduler.c | 21 ++++++++++ src/worker/src/xcecd_worker_utils.c | 2 +- 10 files changed, 159 insertions(+), 40 deletions(-) diff --git a/src/controller/job.c b/src/controller/job.c index 6c757201..7ab28d2f 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -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; } diff --git a/src/include/utils.h b/src/include/utils.h index 61203934..5ef9929a 100644 --- a/src/include/utils.h +++ b/src/include/utils.h @@ -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); diff --git a/src/lib/utils.c b/src/lib/utils.c index 4a147fb3..c1946142 100644 --- a/src/lib/utils.c +++ b/src/lib/utils.c @@ -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) diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 89b24120..b5e9c8fd 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -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*); diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index e0cbd8b2..4baca3e1 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -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; diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 85b2c499..aa8a9faf 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -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: diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index dc7ea8fd..0e4fe2af 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -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; +} diff --git a/src/scheduler/src/plugins/simple_allocator.c b/src/scheduler/src/plugins/simple_allocator.c index 7dcb806e..dcf7ad02 100644 --- a/src/scheduler/src/plugins/simple_allocator.c +++ b/src/scheduler/src/plugins/simple_allocator.c @@ -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, diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index ec21266d..11ed20da 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -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; diff --git a/src/worker/src/xcecd_worker_utils.c b/src/worker/src/xcecd_worker_utils.c index 47b99e91..2d5dc785 100644 --- a/src/worker/src/xcecd_worker_utils.c +++ b/src/worker/src/xcecd_worker_utils.c @@ -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); }