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

fix pkt size problem and fmt_jobid

Browse source
This commit is contained in:
Rongsong Shen 2025-12-24 17:56:45 +08:00
parent 088afe4ea0
commit 4034fd7556
15 changed files with 130 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

12
src/flake.lock generated
View file

@ -92,11 +92,11 @@
},
"nixpkgs_2": {
"locked": {
"lastModified": 1764667669,
"narHash": "sha256-7WUCZfmqLAssbDqwg9cUDAXrSoXN79eEEq17qhTNM/Y=",
"lastModified": 1766309749,
"narHash": "sha256-3xY8CZ4rSnQ0NqGhMKAy5vgC+2IVK0NoVEzDoOh4DA4=",
"owner": "nixos",
"repo": "nixpkgs",
"rev": "418468ac9527e799809c900eda37cbff999199b6",
"rev": "a6531044f6d0bef691ea18d4d4ce44d0daa6e816",
"type": "github"
},
"original": {
@ -181,11 +181,11 @@
"nixpkgs": "nixpkgs_3"
},
"locked": {
"lastModified": 1764926264,
"narHash": "sha256-1N8BKVwxESYUjU4SwDFhb7moCwT9lvkqiq0kmdrjeg4=",
"lastModified": 1766562395,
"narHash": "sha256-bbnbjJ2sru/2YqHN1jzdYs+khOhrjngaRLBJFd5nQ4Y=",
"owner": "shen390s",
"repo": "my-nix-channel",
"rev": "597aa03bdb99a95853744acd349f113a9cd2b3cd",
"rev": "828497661dbef0721ca1aceb0b76fd71086a720b",
"type": "github"
},
"original": {

2
src/include/netio.h
View file

@ -26,7 +26,7 @@ typedef struct {
typedef struct {
data_item_t link;
uint16_t sz;
uint64_t sz;
char* data;
} io_packet_t;

3
src/include/utils.h
View file

@ -44,8 +44,7 @@
MAKE_JOBID(EXTRACT_ID_CLUSTER(id), EXTRACT_ID_APISERVER(id), \
EXTRACT_ID_TIMESTAMP(id), 0x0, EXTRACT_ARRAY_JOB(id))
#define EXTRACT_ARRAY_IDX(id) \
EXTRACT_BITS(id, 0, (N_SEQUENCE_BITS - SEQUENCE_HIGH_SHIFT))
#define EXTRACT_ARRAY_IDX(id) EXTRACT_BITS(id, 0, N_ARRAY_IDX_BITS)
#define ROUNDUP(v, bits) (((v) + UMAX_BITS(bits)) & ~UMAX_BITS(bits))

6
src/include/xce/types.h
View file

@ -357,6 +357,12 @@ typedef struct {
#define QUEUE_STAT_FAIRSHARE_DEFAULT \
0x20 /**< Fairshare queue with null user share defined*/
/* FIXME: improve following later */
#define JOB_IS_PEND(j) ((j->state & JOB_STAT_PEND) != 0)
#define JOB_IS_RUN(j) ((j->state & JOB_STAT_RUN) != 0)
#define JOB_IS_EXIT(j) ((j->state & JOB_STAT_EXIT) != 0)
#define JOB_IS_DONE(j) ((j->state & JOB_STAT_DONE) != 0)
typedef enum {
NORMAL = 0,
TTY,

2
src/lib/xcealloc.c
View file

@ -80,7 +80,7 @@ xce_host_alloc_t** make_host_allocs(int nhosts, ...)
int i;
va_list ap;
if (nhosts == NULL) {
if (nhosts <= 0) {
return NULL;
}

8
src/nixSupport/xce.nix
View file

@ -29,7 +29,9 @@ stdenv.mkDerivation {
readline
ncurses
struct2json
capnpc
(capnpc.overrideAttrs (oldAttrs: {
separateDebugInfo = true;
}))
which
findutils
(libuv.overrideAttrs (oldAttrs: {
@ -50,7 +52,9 @@ stdenv.mkDerivation {
libcyaml
libyaml
libcgroup
extralib
(extralib.overrideAttrs (oldAttrs: {
separateDebugInfo = true;
}))
];
build-system = [ pkgs.meson ];

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

@ -11,7 +11,8 @@
typedef void* plugin_data_t;
/* hooker of plugin, the first parameter should be xce_sched_plugin_t */
typedef int (*plugin_update_config_hook_t)(void*, xce_conf4sched_t*);
typedef int (*plugin_update_config_hook_t)(void*, xce_conf4sched_t*,
xce_conf4sched_t*);
typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*,
xce_sched_queue_t*);
typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*,
@ -58,7 +59,7 @@ typedef struct {
extern int plugin_register(void*, void*);
extern void plugin_unregister(void*);
extern int plugin_config_update(void*, xce_conf4sched_t*);
extern int plugin_config_update(void*, xce_conf4sched_t*, xce_conf4sched_t*);
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*,
@ -66,8 +67,7 @@ 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*);
extern int register_fcfs(void*);
extern int register_simple_allocator(void*);
#endif

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

@ -44,21 +44,22 @@ typedef struct {
link)))
typedef struct {
mylock_t lock;
int stop;
xcecli_ctx_t apictx;
kvmgr_t plugins;
dataq_t* pend_updates[2];
kvmgr_t jobs;
kvmgr_t queues;
kvmgr_t workers;
kvmgr_t users;
kvmgr_t projects;
pthread_t jobpicker;
pthread_t allocator;
cbuf_t wbuf; /* action to trigger scheduling */
cbuf_t jbuf; /* job picked for allocation */
cbuf_t decisions; /* scheduler decisions */
mylock_t lock;
int stop;
xcecli_ctx_t apictx;
xce_conf4sched_t* config;
kvmgr_t plugins;
dataq_t* pend_updates[2];
kvmgr_t jobs;
kvmgr_t queues;
kvmgr_t workers;
kvmgr_t users;
kvmgr_t projects;
pthread_t jobpicker;
pthread_t allocator;
cbuf_t wbuf; /* action to trigger scheduling */
cbuf_t jbuf; /* job picked for allocation */
cbuf_t decisions; /* scheduler decisions */
} xce_scheduler_t;
extern int sched_init(xce_scheduler_t*, xce_client_config_t*,

2
src/scheduler/meson.build
View file

@ -6,7 +6,7 @@ scheduler_srcs = [
'src/schedobj.c',
'src/plugin.c',
'src/plugins/fcfs.c',
'src/plugins/default_allocator.c',
'src/plugins/simple_allocator.c',
]
incdir = include_directories('src',

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

@ -124,6 +124,41 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset)
return rc;
}
int sched_handle_notify(xce_scheduler_t* sched, xce_notify_data_t* notify)
{
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);
break;
case NotifyData_data_schedExitJob:
fmt_jobid(jobidString, sizeof(jobidString) - 1, notify->data.sched_exit_job);
TRACE("%s %s exit", __func__, jobidString);
break;
case NotifyData_data_schedFailJob:
fmt_jobid(jobidString, sizeof(jobidString) - 1, notify->data.sched_fail_job);
TRACE("%s %s fail", __func__, jobidString);
break;
case NotifyData_data_schedUpdateJob:
fmt_jobid(jobidString, sizeof(jobidString) - 1, notify->data.sched_update_job->jobid);
TRACE("%s %s updated", __func__, jobidString);
break;
case NotifyData_data_schedUpdateQueue:
break;
case NotifyData_data_schedUpdateResvals:
break;
case NotifyData_data_schedJobRusages:
break;
default:
rc = -EINVAL;
break;
}
return rc;
}
int scheduler_req_handler(xce_request_t* request, void* args)
{
int rc = 0;
@ -158,11 +193,17 @@ int scheduler_req_handler(xce_request_t* request, void* args)
case Request_body_datamsg:
rc = sched_handle_dataset(sched, request->body.datamsg);
break;
case Request_body_notifymsg:
rc = sched_handle_notify(sched, request->body.notifymsg);
break;
default:
rc = -EINVAL;
break;
}
break;
case OPCode_notify:
rc = sched_handle_notify(sched, request->body.notifymsg);
break;
default:
rc = -EINVAL;
break;

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

@ -59,23 +59,26 @@ void plugin_unregister(void* arg)
static int plugin_config_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
update_args_t* args = (update_args_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.config_hook != NULL) {
return plugin->hooks.config_hook(plugin, arg1);
return plugin->hooks.config_hook(plugin, args->old, args->new);
}
return 0;
}
int plugin_config_update(void* arg, xce_conf4sched_t* config)
int plugin_config_update(void* arg, xce_conf4sched_t* old,
xce_conf4sched_t* config)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
int rc;
update_args_t args = { .old = old, .new = config };
if (sched == NULL) {
return -EINVAL;
@ -83,7 +86,7 @@ int plugin_config_update(void* arg, xce_conf4sched_t* config)
xlist_init(&lst, 3);
xlist_store(&lst, plugin_config_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, config);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
xlist_fini(&lst);

4
src/scheduler/src/plugins/fcfs.c
View file

@ -86,8 +86,8 @@ static int fcfs_priv_data_init(fcfs_plugin_data_t* data)
return 0;
}
static int fcfs_update_config(xce_sched_plugin_t* plugin,
xce_conf4sched_t* config)
static int fcfs_update_config(xce_sched_plugin_t* plugin, xce_conf4sched_t* old,
xce_conf4sched_t* config)
{
fcfs_plugin_data_t* privdata = NULL;

35
src/scheduler/src/plugins/default_allocator.c → src/scheduler/src/plugins/simple_allocator.c
View file

@ -15,9 +15,9 @@
static uint32_t maxslots = 8;
static uint32_t slotsInUse = 0;
static int default_allocator_init(xce_sched_plugin_t* plugin) { return 1; }
static int simple_allocator_init(xce_sched_plugin_t* plugin) { return 1; }
static void default_allocator_fini(xce_sched_plugin_t* plugin)
static void simple_allocator_fini(xce_sched_plugin_t* plugin)
{
if (plugin == NULL) {
return;
@ -31,16 +31,17 @@ static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold,
/* FIXME: we need to add logic to check whether the job should be handled by
* me */
j->allocator = plugin;
switch (j->state) {
case JOB_STAT_PEND:
break;
case JOB_STAT_RUN:
break;
case JOB_STAT_DONE:
break;
default:
break;
if (jold == NULL) {
if (JOB_IS_RUN(j)) {
slotsInUse++;
}
} else {
if (JOB_IS_RUN(jold) && (JOB_IS_EXIT(j) || (JOB_IS_DONE(j)))) {
slotsInUse--;
}
}
return 1;
}
@ -99,15 +100,15 @@ static int my_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision)
return rc;
}
int register_default_allocator(void* sched)
int register_simple_allocator(void* sched)
{
static xce_sched_plugin_t default_allocator_plugin = {
.name = "xce_sched_default_allocator",
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 },
.init_fn = (plugin_init_fn_t)default_allocator_init,
.fini_fn = (plugin_fini_fn_t)default_allocator_fini,
.init_fn = (plugin_init_fn_t)simple_allocator_init,
.fini_fn = (plugin_fini_fn_t)simple_allocator_fini,
.begin_session_fn = NULL,
.end_session_fn = NULL,
.job_elect_fn = NULL,
@ -122,7 +123,7 @@ int register_default_allocator(void* sched)
return -ENOMEM;
}
memcpy(plugin, &default_allocator_plugin, sizeof(*plugin));
memcpy(plugin, &simple_allocator_plugin, sizeof(*plugin));
return plugin_register(sched, plugin);
}

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

@ -1,3 +1,4 @@
#include <errno.h>
#include <stdlib.h>
#include <string.h>
@ -33,8 +34,6 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
job->maxtasks = j->submit->maxtasks;
job->priority = 0; /* FIXME: support job level priority */
sched_add_job(sched, job);
return job;
}
@ -66,8 +65,6 @@ xce_sched_queue_t* make_sched_queue(void* arg, xce_queue_t* q)
queue->accept_interval = q->accept_interval;
queue->sched_delay = q->sched_delay;
sched_add_queue(sched, queue);
return queue;
}
@ -95,8 +92,6 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
worker->name = strdup(w->name);
/* FIXME: */
sched_add_worker(sched, worker);
return worker;
}

22
src/scheduler/src/scheduler.c
View file

@ -133,7 +133,7 @@ static void sched_pend_update_killer(data_item_t* it)
static int sched_register_plugins(xce_scheduler_t* sched)
{
register_fcfs(sched);
register_default_allocator(sched);
register_simple_allocator(sched);
return 0;
}
@ -241,7 +241,7 @@ static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
{
int i;
plugin_config_update(sched, config);
plugin_config_update(sched, sched->config, config);
for (i = 0; i < config->n_queues; i++) {
xce_sched_queue_t* qold
@ -249,6 +249,12 @@ static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]);
plugin_queue_update(sched, qold, q);
if (qold == NULL) {
sched_add_queue(sched, q);
} else {
/* FIXME: overwrite qold with q */
}
}
return 0;
@ -264,6 +270,12 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs,
xce_sched_job_t* j = make_sched_job(sched, jobs[i]);
plugin_job_update(sched, jold, j);
if (jold == NULL) {
sched_add_job(sched, j);
} else {
/**/
}
}
return 0;
@ -278,7 +290,11 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
xce_sched_worker_t* wold = sched_get_worker(sched, workers[i]->name);
xce_sched_worker_t* w = make_sched_worker(sched, workers[i]);
plugin_worker_update(sched, wold,w);
plugin_worker_update(sched, wold, w);
if (wold == NULL) {
sched_add_worker(sched, w);
} else {
}
}
return 0;