From 4034fd755614560b4fd18e85af5f43b5a222e1ef Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 24 Dec 2025 17:56:45 +0800 Subject: [PATCH] fix pkt size problem and fmt_jobid --- src/flake.lock | 12 +++--- src/include/netio.h | 2 +- src/include/utils.h | 3 +- src/include/xce/types.h | 6 +++ src/lib/xcealloc.c | 2 +- src/nixSupport/xce.nix | 8 +++- src/scheduler/include/plugin.h | 10 ++--- src/scheduler/include/scheduler.h | 31 +++++++------- src/scheduler/meson.build | 2 +- src/scheduler/src/handler.c | 41 +++++++++++++++++++ src/scheduler/src/plugin.c | 9 ++-- src/scheduler/src/plugins/fcfs.c | 4 +- ...default_allocator.c => simple_allocator.c} | 35 ++++++++-------- src/scheduler/src/schedobj.c | 7 +--- src/scheduler/src/scheduler.c | 22 ++++++++-- 15 files changed, 130 insertions(+), 64 deletions(-) rename src/scheduler/src/plugins/{default_allocator.c => simple_allocator.c} (79%) diff --git a/src/flake.lock b/src/flake.lock index 990c2c4..a429c39 100644 --- a/src/flake.lock +++ b/src/flake.lock @@ -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": { diff --git a/src/include/netio.h b/src/include/netio.h index 552192d..32bcaf9 100644 --- a/src/include/netio.h +++ b/src/include/netio.h @@ -26,7 +26,7 @@ typedef struct { typedef struct { data_item_t link; - uint16_t sz; + uint64_t sz; char* data; } io_packet_t; diff --git a/src/include/utils.h b/src/include/utils.h index 6c2c519..6120393 100644 --- a/src/include/utils.h +++ b/src/include/utils.h @@ -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)) diff --git a/src/include/xce/types.h b/src/include/xce/types.h index fa5663e..d8840e2 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -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, diff --git a/src/lib/xcealloc.c b/src/lib/xcealloc.c index d928b80..ed127d3 100644 --- a/src/lib/xcealloc.c +++ b/src/lib/xcealloc.c @@ -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; } diff --git a/src/nixSupport/xce.nix b/src/nixSupport/xce.nix index 2fc6663..7c4a5c3 100644 --- a/src/nixSupport/xce.nix +++ b/src/nixSupport/xce.nix @@ -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 ]; diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index ba6eb12..89b2412 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -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 diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index bb1badf..e0cbd8b 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -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*, diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 7cb0f48..f704b76 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -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', diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 3ae19db..85b2c49 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -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; diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 674327b..dc7ea8f 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -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); diff --git a/src/scheduler/src/plugins/fcfs.c b/src/scheduler/src/plugins/fcfs.c index 07ffd1d..63b62ae 100644 --- a/src/scheduler/src/plugins/fcfs.c +++ b/src/scheduler/src/plugins/fcfs.c @@ -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; diff --git a/src/scheduler/src/plugins/default_allocator.c b/src/scheduler/src/plugins/simple_allocator.c similarity index 79% rename from src/scheduler/src/plugins/default_allocator.c rename to src/scheduler/src/plugins/simple_allocator.c index c5def9b..7dcb806 100644 --- a/src/scheduler/src/plugins/default_allocator.c +++ b/src/scheduler/src/plugins/simple_allocator.c @@ -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); } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index fc996d4..ceec602 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -1,3 +1,4 @@ +#include #include #include @@ -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; } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 7f3df40..ec21266 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -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;