fix(scheduler): re-queue job as PEND on dispatch failure (bug-1494 class)
The controller rejects a dispatch when the chosen worker is not in a dispatchable state (UNAVAIL after a worker death, or still joining) and replies with a jobDispatchFail notify. The scheduler had no handler for that event, so a job it had just committed (state=RUN) was never returned to PEND — it was never re-elected and stayed PEND cluster-wide even after the worker recovered. Handle NotifyData_data_jobDispatchFail: log the failure and call the new sched_requeue_pend_job(), which frees the dispatch bookkeeping, resets the job to PEND, and re-enqueues it in the fcfs pendJobs queue so the next election session re-dispatches it. White-box test TUs that #include handler.c get a sched_requeue_pend_job stub. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
5657fca7f7
commit
6928f74702
6 changed files with 94 additions and 0 deletions
|
|
@ -130,6 +130,7 @@ extern void sched_go(xce_scheduler_t*);
|
||||||
extern void sched_begin_session(xce_scheduler_t*);
|
extern void sched_begin_session(xce_scheduler_t*);
|
||||||
extern void sched_end_session(xce_scheduler_t*);
|
extern void sched_end_session(xce_scheduler_t*);
|
||||||
extern xce_sched_job_t* sched_get_job(xce_scheduler_t*, uint64_t);
|
extern xce_sched_job_t* sched_get_job(xce_scheduler_t*, uint64_t);
|
||||||
|
extern int sched_requeue_pend_job(xce_scheduler_t*, uint64_t);
|
||||||
extern xce_sched_queue_t* sched_get_queue(xce_scheduler_t*, char*);
|
extern xce_sched_queue_t* sched_get_queue(xce_scheduler_t*, char*);
|
||||||
extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*);
|
extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*);
|
||||||
extern xce_sched_user_t* sched_get_user(xce_scheduler_t*, char*);
|
extern xce_sched_user_t* sched_get_user(xce_scheduler_t*, char*);
|
||||||
|
|
|
||||||
|
|
@ -527,6 +527,28 @@ int sched_handle_notify(xce_scheduler_t* sched, xce_notify_data_t* notify)
|
||||||
if (rc >= 0)
|
if (rc >= 0)
|
||||||
notify->data.sched_update_queue = NULL;
|
notify->data.sched_update_queue = NULL;
|
||||||
break;
|
break;
|
||||||
|
case NotifyData_data_jobDispatchFail: {
|
||||||
|
/* The controller rejected a dispatch the scheduler committed (e.g.
|
||||||
|
* the chosen worker went UNAVAIL between election and dispatch).
|
||||||
|
* Return the job to PEND so it is re-elected next session — without
|
||||||
|
* this the job stays RUN in the scheduler's view and is never
|
||||||
|
* dispatched again (job stuck PEND cluster-wide). */
|
||||||
|
job_failinfo_t* fail = notify->data.dispatchfail;
|
||||||
|
char jobidString[128];
|
||||||
|
|
||||||
|
if (fail == NULL) {
|
||||||
|
rc = -EINVAL;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
fmt_jobid(jobidString, sizeof(jobidString), fail->jobid);
|
||||||
|
WARN("%s: job %s dispatch failed (code %d) — returning to PEND",
|
||||||
|
__func__, jobidString, fail->code);
|
||||||
|
rc = sched_requeue_pend_job(sched, fail->jobid);
|
||||||
|
if (rc >= 0)
|
||||||
|
notify->data.dispatchfail = NULL;
|
||||||
|
break;
|
||||||
|
}
|
||||||
case NotifyData_data_schedUpdateResvals:
|
case NotifyData_data_schedUpdateResvals:
|
||||||
rc = sched_handle_sched_update_resvals(
|
rc = sched_handle_sched_update_resvals(
|
||||||
sched, notify->data.sched_update_resvals);
|
sched, notify->data.sched_update_resvals);
|
||||||
|
|
|
||||||
|
|
@ -700,6 +700,39 @@ static int sched_job_state_change(xce_scheduler_t* sched, uint64_t jobid,
|
||||||
|
|
||||||
return rc;
|
return rc;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
|
* sched_requeue_pend_job — return a job to PEND after its dispatch failed at
|
||||||
|
* the controller (jobDispatchFail notify). The dispatch decision set the
|
||||||
|
* job's state to RUN and attached allocs/accounts; the controller rejected
|
||||||
|
* the dispatch (e.g. the chosen worker went UNAVAIL/down between the
|
||||||
|
* scheduler's election and the controller's dispatch_a_job_ check), and
|
||||||
|
* without this the job stays RUN in the scheduler's view forever — it is
|
||||||
|
* never re-elected and is stuck PEND cluster-wide even after the worker
|
||||||
|
* recovers (scheduler-side of bug-1494 class). Free the dispatch
|
||||||
|
* bookkeeping, reset to PEND, and re-enqueue so the next election
|
||||||
|
* re-dispatches it.
|
||||||
|
*/
|
||||||
|
int sched_requeue_pend_job(xce_scheduler_t* sched, uint64_t jobid)
|
||||||
|
{
|
||||||
|
xce_sched_job_t* job = sched_get_job(sched, jobid);
|
||||||
|
|
||||||
|
if (job == NULL) {
|
||||||
|
return -ENOENT;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (job->accounts != NULL) {
|
||||||
|
free_job_account(job->accounts);
|
||||||
|
job->accounts = NULL;
|
||||||
|
}
|
||||||
|
if (job->allocs != NULL) {
|
||||||
|
free_JobAlloc_ptr(&(job->allocs));
|
||||||
|
}
|
||||||
|
job->state = JOB_STAT_PEND;
|
||||||
|
|
||||||
|
/* fcfs_update_job (job_hook) enqueues only PEND jobs, with dedup. */
|
||||||
|
return plugin_job_update(sched, job);
|
||||||
|
}
|
||||||
static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
|
static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
|
||||||
xce_worker_t** workers)
|
xce_worker_t** workers)
|
||||||
{
|
{
|
||||||
|
|
|
||||||
|
|
@ -35,6 +35,20 @@ xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
|
||||||
(void)sched; (void)jobid; return NULL;
|
(void)sched; (void)jobid; return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* White-box stub for sched_requeue_pend_job (handler.c's jobDispatchFail
|
||||||
|
* case). sched_get_job is stubbed to NULL here, so this just returns
|
||||||
|
* -ENOENT like the real function does for an unknown jobid. */
|
||||||
|
int sched_requeue_pend_job(xce_scheduler_t* sched, uint64_t jobid)
|
||||||
|
{
|
||||||
|
(void)sched;
|
||||||
|
xce_sched_job_t* job = sched_get_job(sched, jobid);
|
||||||
|
if (job == NULL) {
|
||||||
|
return -ENOENT;
|
||||||
|
}
|
||||||
|
job->state = JOB_STAT_PEND;
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
#include "../src/handler.c"
|
#include "../src/handler.c"
|
||||||
|
|
||||||
void setUp(void) { return; }
|
void setUp(void) { return; }
|
||||||
|
|
|
||||||
|
|
@ -56,6 +56,18 @@ xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
|
||||||
return job;
|
return job;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* White-box stub for sched_requeue_pend_job (handler.c's jobDispatchFail
|
||||||
|
* case), backed by this TU's real kvmgr sched_get_job. */
|
||||||
|
int sched_requeue_pend_job(xce_scheduler_t* sched, uint64_t jobid)
|
||||||
|
{
|
||||||
|
xce_sched_job_t* job = sched_get_job(sched, jobid);
|
||||||
|
if (job == NULL) {
|
||||||
|
return -ENOENT;
|
||||||
|
}
|
||||||
|
job->state = JOB_STAT_PEND;
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
void setUp(void) { return; }
|
void setUp(void) { return; }
|
||||||
void tearDown(void)
|
void tearDown(void)
|
||||||
{
|
{
|
||||||
|
|
|
||||||
|
|
@ -107,6 +107,18 @@ xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
|
||||||
return job;
|
return job;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* White-box stub for sched_requeue_pend_job (handler.c's jobDispatchFail
|
||||||
|
* case), backed by this TU's real kvmgr sched_get_job. */
|
||||||
|
int sched_requeue_pend_job(xce_scheduler_t* sched, uint64_t jobid)
|
||||||
|
{
|
||||||
|
xce_sched_job_t* job = sched_get_job(sched, jobid);
|
||||||
|
if (job == NULL) {
|
||||||
|
return -ENOENT;
|
||||||
|
}
|
||||||
|
job->state = JOB_STAT_PEND;
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
void setUp(void) { return; }
|
void setUp(void) { return; }
|
||||||
void tearDown(void) { return; }
|
void tearDown(void) { return; }
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue