wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(scheduler-controller-msg-gap): route notify job-update to in-place sched_update_existing_job (task 5.6, bug-765 G-8+G-4)

Browse source 
The notify schedUpdateJob path shared XCE_SCHED_OP_UPDATE_JOB with the
dataset bulk-load, so the dequeue routed both to sched_update_jobs()
(create/replace) — a notify update to an existing job corrupted/duplicated
live job state and never returned -ENOENT (Req 3 UB-3.1/UB-3.2 unmet;
sched_update_existing_job was dead code copying only state).

- Add distinct op XCE_SCHED_OP_UPDATE_JOB_FIELDS (+ jobfieldsupdate union),
  enqueued only by the notify handler, dequeued to sched_update_existing_job.
- Move sched_update_existing_job to schedobj.c; update all mutable fields
  (state, priority, queue, resreq) in place; -ENOENT unknown, -EINVAL NULL.
- Dataset bulk-create path unchanged. Killer frees the new payload.
- New white-box test sched_job_update_test (3/3); regression sched_test 15/15.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-06-19 03:17:40 +00:00
parent d9b43c2d2f
commit 5c1f715ccf
7 changed files with 344 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -32,7 +32,11 @@ typedef enum {
XCE_SCHED_OP_UPDATE_RESOURCES,
XCE_SCHED_OP_UPDATE_RESVALS_FULL,
XCE_SCHED_OP_UPDATE_RESVALS,
XCE_SCHED_OP_JOB_RUSAGE
XCE_SCHED_OP_JOB_RUSAGE,
/* Notify-only single-job in-place field update. Distinct from
* XCE_SCHED_OP_UPDATE_JOB (the dataset bulk-create path) so a notify
* update to an existing job mutates it in place and never creates a job. */
XCE_SCHED_OP_UPDATE_JOB_FIELDS
} xce_sched_op_t;
typedef struct {
@ -44,6 +48,9 @@ typedef struct {
int njobs;
xce_job_t** jobs;
} jobupdate;
struct {
xce_job_t* job;
} jobfieldsupdate;
struct {
int nworkers;
xce_worker_t** workers;
@ -131,6 +138,7 @@ extern xce_sched_project_t* sched_get_project(xce_scheduler_t*, char*);
extern int sched_add_job(xce_scheduler_t*, xce_sched_job_t*);
extern int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*);
extern int sched_update_existing_queue(xce_scheduler_t*, xce_queue_t*);
extern int sched_update_existing_job(xce_scheduler_t*, xce_job_t*);
extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*);
extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*);
extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);

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

@ -175,7 +175,6 @@ static int sched_handle_sched_update_job(xce_scheduler_t* sched,
xce_job_t* job)
{
xce_sched_update_request_t* req;
xce_job_t** job_arr;
char jobidString[128];
if (job == NULL) {
@ -185,21 +184,17 @@ static int sched_handle_sched_update_job(xce_scheduler_t* sched,
fmt_jobid(jobidString, sizeof(jobidString) - 1, job->jobid);
TRACE("%s %s updated", __func__, jobidString);
job_arr = (xce_job_t**)malloc(sizeof(xce_job_t*));
if (job_arr == NULL) {
return -ENOMEM;
}
job_arr[0] = job;
req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
if (req == NULL) {
free(job_arr);
return -ENOMEM;
}
req->op = XCE_SCHED_OP_UPDATE_JOB;
req->u.jobupdate.njobs = 1;
req->u.jobupdate.jobs = job_arr;
/* Notify single-job update takes the in-place field-update path
* (XCE_SCHED_OP_UPDATE_JOB_FIELDS), NOT the dataset bulk-create path
* (XCE_SCHED_OP_UPDATE_JOB). An unknown jobid must yield -ENOENT, never a
* new job. The dataset bulk-create path (sched_on_jobs) is unchanged. */
req->op = XCE_SCHED_OP_UPDATE_JOB_FIELDS;
req->u.jobfieldsupdate.job = job;
enq_pend_update(sched, req);

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

@ -509,3 +509,78 @@ int sched_update_existing_queue(xce_scheduler_t* sched, xce_queue_t* updated)
return 0;
}
/*
* sched_update_existing_job — apply a notify single-job update to an existing
* job IN PLACE. The job must already exist (created via the Data_data_schedjobs
* dataset path); this function never creates a job.
*
* - NULL payload → -EINVAL
* - unknown jobid → -ENOENT (no job created)
* - otherwise → update mutable fields (state, priority, queue,
* resource request) and notify plugins; return 0
*
* Runtime state (schedobj, allocs, accounts) is preserved — the live
* xce_sched_job_t object is mutated, not replaced. Mirrors the field-copy +
* resreq-replace pattern of sched_update_existing_queue().
*/
int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated)
{
xce_sched_job_t* job;
if (updated == NULL) {
return -EINVAL;
}
job = sched_get_job(sched, updated->jobid);
if (job == NULL) {
return -ENOENT;
}
/* state — always carried on xce_job_t */
job->state = updated->state;
/* mutable submit-derived fields */
if (updated->submit != NULL) {
/* queue — re-resolve the queue pointer by name. xce_job_t carries no
* per-job priority, so job priority follows the (possibly new) queue,
* consistent with the queue-derived priority model. */
if (updated->submit->queue != NULL) {
xce_sched_queue_t* q =
sched_get_queue(sched, updated->submit->queue);
if (q != NULL) {
job->queue = q;
job->priority = q->priority;
}
}
/* resource request — re-parse and swap; free the old parsed container.
* Mirrors sched_update_existing_queue's free-old/replace pattern. The
* inner select tree is intern/GC-owned, consistent with make_sched_job. */
if (updated->submit->resreq != NULL) {
xce_resreq_t* newreq =
(xce_resreq_t*)calloc(1, sizeof(*newreq));
if (newreq == NULL) {
return -ENOMEM;
}
char* errmsg = NULL;
int rc = parse_resreq(newreq, updated->submit->resreq,
sched->resmgr, &errmsg);
if (rc < 0) {
ERROR("%s parse_resreq failed: %s", __func__,
errmsg ? errmsg : "(null)");
FREEUP(errmsg);
free(newreq);
return -EINVAL;
}
free(job->resreq);
job->resreq = newreq;
}
}
plugin_job_update(sched, job);
return 0;
}

27
src/scheduler/src/scheduler.c
View file

@ -299,6 +299,9 @@ static void sched_pend_update_killer(data_item_t* it)
free(req->u.jobupdate.jobs);
break;
case XCE_SCHED_OP_UPDATE_JOB_FIELDS:
free_Job_ptr(&(req->u.jobfieldsupdate.job));
break;
case XCE_SCHED_OP_UPDATE_WORKER:
for (i = 0; i < req->u.workerupdate.nworkers; i++) {
free_Worker_ptr(&(req->u.workerupdate.workers[i]));
@ -697,6 +700,7 @@ static char* sched_op_name(xce_sched_op_t op)
case XCE_SCHED_OP_UPDATE_RESVALS_FULL: return "UPDATE_RESVALS_FULL";
case XCE_SCHED_OP_UPDATE_RESVALS: return "UPDATE_RESVALS";
case XCE_SCHED_OP_JOB_RUSAGE: return "JOB_RUSAGE";
case XCE_SCHED_OP_UPDATE_JOB_FIELDS: return "UPDATE_JOB_FIELDS";
default: break;
}
@ -704,26 +708,6 @@ static char* sched_op_name(xce_sched_op_t op)
return buf;
}
static int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated)
{
xce_sched_job_t* job;
if (updated == NULL) {
return -EINVAL;
}
job = sched_get_job(sched, updated->jobid);
if (job == NULL) {
return -ENOENT;
}
job->state = updated->state;
plugin_job_update(sched, job);
return 0;
}
int sched_update(xce_scheduler_t* sched)
{
dataq_t* q = NULL;
@ -756,6 +740,9 @@ int sched_update(xce_scheduler_t* sched)
rc = sched_update_jobs(sched, req->u.jobupdate.njobs,
req->u.jobupdate.jobs);
break;
case XCE_SCHED_OP_UPDATE_JOB_FIELDS:
rc = sched_update_existing_job(sched, req->u.jobfieldsupdate.job);
break;
case XCE_SCHED_OP_UPDATE_WORKER:
rc = sched_update_workers(sched, req->u.workerupdate.nworkers,
req->u.workerupdate.workers);

9
src/scheduler/testsuite/meson.build
View file

@ -11,3 +11,12 @@ scheduler_test = executable('sched_test',
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)
# White-box test for the notify single-job in-place update path
# (sched_update_existing_job). It #includes ../src/schedobj.c directly, so it
# is a standalone TU with its own main() — built as a separate executable.
sched_job_update_test = executable('sched_job_update_test',
['test_job_update.c'],
include_directories:[incdir],
dependencies: [scheduler_test_deps],
install: false)

232
src/scheduler/testsuite/test_job_update.c Normal file
View file

@ -0,0 +1,232 @@
/*
* test_job_update.c - white-box tests for the notify single-job in-place
* update path (sched_update_existing_job), remediation of bug-765 / gap G-8+G-4.
*
* Strategy: pull the REAL sched_update_existing_job into this TU by including
* schedobj.c, then drive it against a REAL sched->jobs kvmgr seeded with a
* REAL xce_sched_job_t. The scheduler-internal externs schedobj.c references
* but which are NOT under test (and live in other .c files not linked here)
* are stubbed below. sched_get_job/sched_add_job semantics are exercised for
* real via the kvmgr key_find/key_add path.
*
* Asserts:
* - existing jobid : fields mutate IN PLACE (same object, runtime state kept)
* - unknown jobid : -ENOENT and NO job created
* - NULL payload : -EINVAL
*/
#define UNIT_TEST 1
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "kvmgr.h"
#include "scheduler.h"
#include "xstring.h"
/* Pull the real implementation under test into this TU. */
#include "../src/schedobj.c"
/* Stubs for scheduler-internal externs schedobj.c references but which are
* defined in other .c files not linked into this test. (make_sched_obj and
* kill_sched_job are static in schedobj.c and come in via the #include.) */
xce_sched_user_t* sched_get_user(xce_scheduler_t* s, char* n)
{
(void)s; (void)n; return NULL;
}
xce_sched_project_t* sched_get_project(xce_scheduler_t* s, char* n)
{
(void)s; (void)n; return NULL;
}
int plugin_job_update(void* s, xce_sched_job_t* j) { (void)s; (void)j; return 0; }
int plugin_queue_update(void* s, xce_sched_queue_t* q) { (void)s; (void)q; return 0; }
xce_job_account_t* make_job_account(xce_job_alloc_t* a) { (void)a; return NULL; }
void free_job_account(xce_job_account_t* a) { (void)a; }
int sched_add_user(xce_scheduler_t* s, xce_sched_user_t* u) { (void)s; (void)u; return 0; }
int sched_add_project(xce_scheduler_t* s, xce_sched_project_t* p) { (void)s; (void)p; return 0; }
void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }
int plugin_prepare_allocation(void* s, xce_sched_ctx_t* c) { (void)s; (void)c; return 0; }
/* sched_get_queue: a tiny by-name lookup over a fixed test queue set. */
static xce_sched_queue_t* g_test_queues[8];
static int g_test_nqueues;
xce_sched_queue_t* sched_get_queue(xce_scheduler_t* s, char* name)
{
int i;
(void)s;
if (name == NULL) {
return NULL;
}
for (i = 0; i < g_test_nqueues; i++) {
if (g_test_queues[i] != NULL && g_test_queues[i]->name != NULL
&& strcmp(g_test_queues[i]->name, name) == 0) {
return g_test_queues[i];
}
}
return NULL;
}
/* REAL kvmgr-backed job add/get - same rtti path the production code uses. */
int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j)
{
int rc;
lock(sched->jobs, RTTI_LOCK_WRITE);
rc = key_add(sched->jobs, (void*)j->jobid, j);
unlock(sched->jobs);
return rc;
}
xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
{
int rc;
xce_sched_job_t* job = NULL;
lock(sched->jobs, RTTI_LOCK_READ);
rc = key_find(sched->jobs, (void*)jobid, (void**)&job);
unlock(sched->jobs);
if (rc <= 0) {
return NULL;
}
return job;
}
void setUp(void) { return; }
void tearDown(void)
{
printf("\n-----------------------------------------------------\n");
return;
}
/* Test fixture helpers */
static void fixture_init(xce_scheduler_t* sched)
{
kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT,
.nslots = 59,
.delta = 16,
.datakiller = kill_sched_job };
kvmgr_params_t res_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 16,
.delta = 8,
.datakiller = NULL };
memset(sched, 0, sizeof(*sched));
resreq_init(1);
xstring_table_init(1024 * 16);
sched->jobs = makeInstance(&xkvmgr_type, &jobs_params);
sched->resmgr = makeInstance(&xkvmgr_type, &res_params);
g_test_nqueues = 0;
memset(g_test_queues, 0, sizeof(g_test_queues));
}
static xce_sched_queue_t* make_test_queue(const char* name, int32_t prio)
{
xce_sched_queue_t* q = (xce_sched_queue_t*)calloc(1, sizeof(*q));
q->name = makeString((char*)name);
q->priority = prio;
g_test_queues[g_test_nqueues++] = q;
return q;
}
/* A seeded existing job carrying live runtime state we must not lose. */
static xce_sched_job_t* seed_job(xce_scheduler_t* sched, uint64_t jobid)
{
xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job));
job->jobid = jobid;
job->state = JOB_STAT_PEND;
job->priority = 1;
job->resreq = (xce_resreq_t*)calloc(1, sizeof(*job->resreq));
/* schedobj is live runtime state that MUST survive an in-place update */
job->schedobj = (xce_sched_obj_t*)calloc(1, sizeof(*job->schedobj));
sched_add_job(sched, job);
return job;
}
static xce_job_t* make_update_payload(uint64_t jobid, unsigned int state,
const char* queue)
{
xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j));
j->jobid = jobid;
j->state = state;
j->submit = (job_submit_data_t*)calloc(1, sizeof(*j->submit));
if (queue != NULL) {
j->submit->queue = makeString((char*)queue);
}
return j;
}
/* Tests */
static void
test_update_existing_job_mutates_in_place(void)
{
xce_scheduler_t sched;
fixture_init(&sched);
make_test_queue("normal", 5);
xce_sched_job_t* job = seed_job(&sched, 42);
xce_sched_obj_t* schedobj = job->schedobj; /* capture live runtime ptr */
xce_job_t* upd = make_update_payload(42, JOB_STAT_RUN, "normal");
int rc = sched_update_existing_job(&sched, upd);
TEST_ASSERT_EQUAL_INT(0, rc);
/* SAME object mutated in place, not replaced */
TEST_ASSERT_EQUAL_PTR(job, sched_get_job(&sched, 42));
/* mutable fields updated */
TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state);
TEST_ASSERT_EQUAL_PTR(g_test_queues[0], job->queue);
TEST_ASSERT_EQUAL_INT(5, job->priority);
/* live runtime state preserved (object identity, not a fresh calloc) */
TEST_ASSERT_EQUAL_PTR(schedobj, job->schedobj);
/* upd->submit->queue is a makeString()-interned string owned by the
* xstring table — not caller-freeable; only free the calloc'd structs. */
free(upd->submit);
free(upd);
}
static void
test_update_unknown_job_returns_enoent_no_create(void)
{
xce_scheduler_t sched;
fixture_init(&sched);
/* No job seeded for jobid 999 */
xce_job_t* upd = make_update_payload(999, JOB_STAT_RUN, NULL);
int rc = sched_update_existing_job(&sched, upd);
TEST_ASSERT_EQUAL_INT(-ENOENT, rc);
/* And NO job was created via the update path */
TEST_ASSERT_NULL(sched_get_job(&sched, 999));
free(upd->submit);
free(upd);
}
static void
test_update_null_payload_returns_einval(void)
{
xce_scheduler_t sched;
fixture_init(&sched);
int rc = sched_update_existing_job(&sched, NULL);
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}
int main(int argc, char* argv[])
{
(void)argc; (void)argv;
UNITY_BEGIN();
RUN_TEST(test_update_existing_job_mutates_in_place, 0);
RUN_TEST(test_update_unknown_job_returns_enoent_no_create, 0);
RUN_TEST(test_update_null_payload_returns_einval, 0);
return UNITY_END();
}