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

pass pend reason back to controller

Browse source
This commit is contained in:
Rongsong Shen 2025-12-23 15:29:26 +08:00
parent cd0bca3138
commit 5560ae91da
16 changed files with 349 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/scheduler/include/decision.h
View file

@ -14,13 +14,13 @@ typedef enum {
typedef struct {
xce_decision_op_t op;
uint64_t jobid;
union {
xce_job_alloc_t* alloc;
xce_job_alloc_t* alloc;
xce_pendreason_t* reason;
} u;
} xce_decision_t;
extern xce_decision_t* make_pend_decision(uint64_t);
extern xce_decision_t* make_pend_decision(xce_pendreason_t*);
extern xce_decision_t* make_dispatch_decision(xce_job_alloc_t*);
#endif

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

@ -29,7 +29,8 @@ typedef int (*plugin_begin_session_fn_t)(void*);
typedef int (*plugin_end_session_fn_t)(void*);
/* allocation function of job */
typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t);
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;
@ -60,6 +61,7 @@ extern int plugin_job_update(void*, xce_sched_job_t*);
extern int plugin_worker_update(void*, xce_sched_worker_t*);
extern int plugin_begin_session(void*);
extern int plugin_end_session(void*);
extern int plugin_commit_decision(void*, xce_decision_t*);
extern int register_fcfs(void*);
extern int register_default_allocator(void*);

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

@ -81,5 +81,6 @@ extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);
extern int sched_job_is_ready(xce_sched_job_t*);
extern int sched_queue_is_ready(xce_sched_queue_t*);
extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*);
extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
#endif

4
src/scheduler/meson.build
View file

@ -5,8 +5,8 @@ scheduler_srcs = [
'src/decision.c',
'src/schedobj.c',
'src/plugin.c',
'src/plugin_fcfs.c',
'src/plugin_default_allocator.c',
'src/plugins/fcfs.c',
'src/plugins/default_allocator.c',
]
incdir = include_directories('src',

7
src/scheduler/src/decision.c
View file

@ -3,7 +3,7 @@
#include "decision.h"
xce_decision_t* make_pend_decision(uint64_t jobid)
xce_decision_t* make_pend_decision(xce_pendreason_t* reason)
{
xce_decision_t* decision;
@ -12,8 +12,8 @@ xce_decision_t* make_pend_decision(uint64_t jobid)
return NULL;
}
decision->op = XCE_DECISION_PEND;
decision->jobid = jobid;
decision->op = XCE_DECISION_PEND;
decision->u.reason = reason;
return decision;
}
@ -28,7 +28,6 @@ xce_decision_t* make_dispatch_decision(xce_job_alloc_t* alloc)
decision->op = XCE_DECISION_DISPATCH;
decision->u.alloc = alloc;
decision->jobid = alloc->jobid;
return decision;
}

10
src/scheduler/src/main.c
View file

@ -123,6 +123,12 @@ int main(int argc, char* argv[])
while (1) {
xce_decision_t* decision = NULL;
int rc = 0;
xce_pendreason_t *preasons[1];
xce_preason_set_t reasons = {
.nreasons = 1,
.reasons = preasons
};
cbuf_get(&(sched.decisions), &decision);
@ -141,10 +147,14 @@ int main(int argc, char* argv[])
break;
case XCE_DECISION_PEND:
/* set pend reason for jobs */
preasons[ 0] = decision->u.reason;
rc = sched_publish_pend_reason(&sched, &reasons);
break;
default:
break;
}
TRACE("%s rc %d", __func__, rc);
}
sched_end_session(&sched);

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

@ -252,3 +252,35 @@ int plugin_end_session(void* arg)
return rc;
}
static int plugin_commit_decision_(void* arg1, void* arg2) {
xce_sched_plugin_t *plugin = (xce_sched_plugin_t *)arg2;
if (plugin == NULL) {
return 0;
}
if (plugin->hooks.commit_hook != NULL) {
return plugin->hooks.commit_hook(plugin, arg1);
}
return 0;
}
int plugin_commit_decision(void* arg1, xce_decision_t* decision)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg1;
xlist_t lst;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_commit_decision_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, decision);
xlist_fini(&lst);
return rc;
}

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

@ -1,41 +0,0 @@
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "plugin.h"
#include "scheduler.h"
#include "utils.h"
static int default_allocator_init(void* arg) { return 1; }
static void default_allocator_fini(void* arg) { return; }
static int my_update_job(xce_sched_plugin_t *plugin, xce_sched_job_t *j) {
/* FIXME: we need to add logic to check whether the job should be handled by me */
j->allocator = plugin;
return 1;
}
static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid)
{
return NULL;
}
static xce_sched_plugin_t default_allocator_plugin = {
.name = "xce_sched_default_allocator",
.hooks = { .config_hook = NULL, .job_hook = (plugin_update_job_hook_t)my_update_job },
.init_fn = (plugin_init_fn_t)default_allocator_init,
.fini_fn = (plugin_fini_fn_t)default_allocator_fini,
.begin_session_fn = NULL,
.end_session_fn = NULL,
.job_elect_fn = NULL,
.job_alloc_fn = (plugin_job_allocate_fn_t)my_alloc,
.data = NULL
};
int register_default_allocator(void* sched)
{
return plugin_register(sched, &default_allocator_plugin);
}

67
src/scheduler/src/plugins/default_allocator.c Normal file
View file

@ -0,0 +1,67 @@
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "decision.h"
#include "plugin.h"
#include "preason.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
static int default_allocator_init(xce_sched_plugin_t* plugin) { return 1; }
static void default_allocator_fini(xce_sched_plugin_t* plugin) {
if (plugin == NULL) {
return;
}
FREEUP(plugin);
}
static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j)
{
/* FIXME: we need to add logic to check whether the job should be handled by
* me */
j->allocator = plugin;
return 1;
}
static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid,
xce_pendreason_t** preason)
{
xce_pendreason_t* reason = make_pendreason(jobid);
add_job_preason(reason->reasons, PREASON_BAD_RESREQ, NULL, NULL);
*preason = reason;
return NULL;
}
int register_default_allocator(void* sched)
{
static xce_sched_plugin_t default_allocator_plugin = {
.name = "xce_sched_default_allocator",
.hooks = { .config_hook = NULL,
.job_hook = (plugin_update_job_hook_t)my_update_job },
.init_fn = (plugin_init_fn_t)default_allocator_init,
.fini_fn = (plugin_fini_fn_t)default_allocator_fini,
.begin_session_fn = NULL,
.end_session_fn = NULL,
.job_elect_fn = NULL,
.job_alloc_fn = (plugin_job_allocate_fn_t)my_alloc,
.data = NULL
};
xce_sched_plugin_t *plugin = NULL;
plugin= (xce_sched_plugin_t *)calloc(1, sizeof(*plugin));
if (plugin == NULL) {
return -ENOMEM;
}
memcpy(plugin, &default_allocator_plugin, sizeof(*plugin));
return plugin_register(sched, plugin);
}

143
src/scheduler/src/plugin_fcfs.c → src/scheduler/src/plugins/fcfs.c
View file

@ -86,9 +86,9 @@ static int fcfs_priv_data_init(fcfs_plugin_data_t* data)
return 0;
}
static int fcfs_update_config(void* arg, xce_conf4sched_t* config)
static int fcfs_update_config(xce_sched_plugin_t* plugin,
xce_conf4sched_t* config)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
fcfs_plugin_data_t* privdata = NULL;
TRACE("%s ...", __func__);
@ -100,29 +100,39 @@ static int fcfs_update_config(void* arg, xce_conf4sched_t* config)
return 0;
}
static int skip_not_matched_queue(void* arg1, va_list ap)
static int queue_match(void* arg1, va_list ap)
{
fcfs_queue_t* q = (fcfs_queue_t*)arg1;
char* name = va_arg(ap, char*);
if ((name == NULL) || (q == NULL)) {
return 1;
return 0;
}
return (strcmp(name, q->queue->name) != 0);
return (strcmp(name, q->queue->name) == 0);
}
static int fcfs_update_queue(void* arg, xce_sched_queue_t* q)
static int job_match(void* arg1, va_list ap)
{
fcfs_job_t* j = (fcfs_job_t*)arg1;
uint64_t jobid = va_arg(ap, uint64_t);
if (j == NULL) {
return 0;
}
return (j->job->jobid == jobid);
}
static int fcfs_update_queue(xce_sched_plugin_t* plugin, xce_sched_queue_t* q)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
ordq_item_t* it;
fcfs_queue_t* queue = NULL;
TRACE("%s queue %s...", __func__, q->name);
it = ordq_get(&(privdata->queues), (ordq_skip_fn_t)skip_not_matched_queue,
q->name);
it = ordq_get(&(privdata->queues), (ordq_match_fn_t)queue_match, q->name);
if (it == NULL) {
queue = (fcfs_queue_t*)calloc(1, sizeof(*queue));
if (queue == NULL) {
@ -146,8 +156,7 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name)
{
ordq_item_t* it = NULL;
it = ordq_peek(&(data->queues), (ordq_skip_fn_t)skip_not_matched_queue,
name);
it = ordq_peek(&(data->queues), (ordq_match_fn_t)queue_match, name);
if (it == NULL) {
return NULL;
}
@ -155,9 +164,8 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name)
return (fcfs_queue_t*)it->data;
}
static int fcfs_update_job(void* arg, xce_sched_job_t* j)
static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
fcfs_queue_t* q = NULL;
fcfs_job_t* job = NULL;
@ -198,6 +206,7 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin)
ordq_t* queues = &(privdata->queues);
ordq_item_t* it = NULL;
int go = 0;
uint64_t jobid = 0;
pthread_testcancel();
@ -222,7 +231,7 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin)
pthread_testcancel();
if (sched_job_is_ready(j->job)) {
uint64_t jobid = j->job->jobid;
jobid = j->job->jobid;
TRACE("%s jobid %ld", __func__, jobid);
cbuf_put(&(sched->jbuf), &jobid);
@ -238,15 +247,18 @@ static void* fcfs_elect_job(xce_sched_plugin_t* plugin)
}
}
/* end of scheduling session */
jobid = 0;
cbuf_put(&(sched->jbuf), &jobid);
sleep(1);
}
return NULL;
}
static int fcfs_end_session(void* arg)
static int fcfs_end_session(xce_sched_plugin_t* plugin)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
privdata->new_session = 1;
@ -254,9 +266,8 @@ static int fcfs_end_session(void* arg)
return 0;
}
static int fcfs_init(void* arg)
static int fcfs_init(xce_sched_plugin_t* plugin)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
fcfs_plugin_data_t* privdata = NULL;
privdata = (fcfs_plugin_data_t*)calloc(1, sizeof(*privdata));
@ -270,32 +281,86 @@ static int fcfs_init(void* arg)
return 1;
}
static void fcfs_fini(void* arg)
static void fcfs_fini(xce_sched_plugin_t* plugin)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg;
fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
fcfs_plugin_data_t* privdata = NULL;
if (privdata == NULL) {
if (plugin == NULL) {
return;
}
ordq_fini(&(privdata->queues));
free(privdata);
privdata = (fcfs_plugin_data_t*)plugin->data;
if (privdata != NULL) {
ordq_fini(&(privdata->queues));
free(privdata);
}
FREEUP(plugin);
}
static xce_sched_plugin_t fcfs_plugin = {
.name = "xce_sched_fcfs",
.hooks = { .config_hook = fcfs_update_config,
.queue_hook = fcfs_update_queue,
.job_hook = fcfs_update_job,
.worker_hook = NULL },
.init_fn = (plugin_init_fn_t)fcfs_init,
.fini_fn = (plugin_fini_fn_t)fcfs_fini,
.begin_session_fn = NULL,
.end_session_fn = (plugin_end_session_fn_t)fcfs_end_session,
.job_elect_fn = (plugin_job_election_fn_t)fcfs_elect_job,
.job_alloc_fn = NULL,
.data = NULL
};
static int fcfs_del_job(xce_sched_plugin_t* plugin, uint64_t jobid)
{
fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
xce_sched_job_t* job = NULL;
fcfs_queue_t* q = NULL;
int register_fcfs(void* sched) { return plugin_register(sched, &fcfs_plugin); }
if (privdata == NULL) {
return 0;
}
job = sched_get_job(plugin->sched, jobid);
if (job == NULL) {
return 0;
}
q = fcfs_get_queue(privdata, job->queue->name);
if (q == NULL) {
return 0;
}
return ordq_del(&(q->pendJobs), (ordq_match_fn_t)job_match, jobid);
}
static int fcfs_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision)
{
int rc = 0;
switch (decision->op) {
case XCE_DECISION_DISPATCH:
rc = fcfs_del_job(plugin, decision->u.alloc->jobid);
break;
default:
break;
}
return rc;
}
int register_fcfs(void* sched)
{
static xce_sched_plugin_t fcfs_plugin = {
.name = "xce_sched_fcfs",
.hooks
= { .config_hook = (plugin_update_config_hook_t)fcfs_update_config,
.queue_hook = (plugin_update_queue_hook_t)fcfs_update_queue,
.job_hook = (plugin_update_job_hook_t)fcfs_update_job,
.commit_hook = (plugin_commit_decision_hook_t)fcfs_commit,
.worker_hook = NULL },
.init_fn = (plugin_init_fn_t)fcfs_init,
.fini_fn = (plugin_fini_fn_t)fcfs_fini,
.begin_session_fn = NULL,
.end_session_fn = (plugin_end_session_fn_t)fcfs_end_session,
.job_elect_fn = (plugin_job_election_fn_t)fcfs_elect_job,
.job_alloc_fn = NULL,
.data = NULL
};
xce_sched_plugin_t* plugin
= (xce_sched_plugin_t*)calloc(1, sizeof(*plugin));
if (plugin == NULL) {
return -ENOMEM;
}
memcpy(plugin, &fcfs_plugin, sizeof(*plugin));
return plugin_register(sched, plugin);
}

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

@ -33,6 +33,7 @@ static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid)
xce_sched_job_t* job = sched_get_job(sched, jobid);
xce_sched_plugin_t* plugin = NULL;
xce_job_alloc_t* alloc = NULL;
xce_pendreason_t* reason = NULL;
if (job == NULL) {
ERROR("%s job %lu can not be found", __func__, jobid);
@ -51,15 +52,16 @@ static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid)
return NULL;
}
alloc = plugin->job_alloc_fn(job->allocator, jobid);
alloc = plugin->job_alloc_fn(job->allocator, jobid, &reason);
if (alloc == NULL) {
return make_pend_decision(jobid);
return make_pend_decision(reason);
}
return make_dispatch_decision(alloc);
}
static int commit_decision(xce_scheduler_t *sched, xce_decision_t *decision) {
static int commit_decision(xce_scheduler_t* sched, xce_decision_t* decision)
{
return sched_commit_decision(sched, decision);
}
@ -86,7 +88,7 @@ static void* sched_allocator(void* arg)
decision = calc_decision(sched, jobid);
commit_decision(sched, decision);
cbuf_put(&(sched->decisions), &decision);
}
@ -401,6 +403,13 @@ int sched_job_is_ready(xce_sched_job_t* j) { return 1; }
int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; }
int sched_commit_decision(xce_scheduler_t *sched, xce_decision_t *decision) {
return 0;
int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision)
{
return plugin_commit_decision(sched, decision);
}
int sched_publish_pend_reason(xce_scheduler_t* sched,
xce_preason_set_t* reasons)
{
return xcecli_set_pend_reason(sched->apictx, reasons);
}