forked from lei.chen/myxce
6928f74702
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>
173 lines
6.9 KiB
C
173 lines
6.9 KiB
C
#if !defined(_SCHEDULER_H_)
|
|
|
|
#define _SCHEDULER_H_ 1
|
|
|
|
#include <pthread.h>
|
|
#include <rtti.h>
|
|
#include <stdatomic.h>
|
|
#include <stdint.h>
|
|
#include <time.h>
|
|
|
|
#include "client.h"
|
|
#include "dataq.h"
|
|
#include "decision.h"
|
|
#include "group.h"
|
|
#include "kvmgr.h"
|
|
#include "mylock.h"
|
|
#include "plugin.h"
|
|
#include "xce/cluster.h"
|
|
#include "xce/types.h"
|
|
#include "xlist.h"
|
|
#include "xvec.h"
|
|
|
|
typedef void* (*sched_group_worker_fn_t)(void*, xce_sched_worker_t*);
|
|
|
|
typedef enum {
|
|
XCE_SCHED_OP_NONE = 0,
|
|
XCE_SCHED_OP_UPDATE_CONFIG,
|
|
XCE_SCHED_OP_UPDATE_JOB,
|
|
XCE_SCHED_OP_JSTATE_CHANGE,
|
|
XCE_SCHED_OP_UPDATE_WORKER,
|
|
XCE_SCHED_OP_UPDATE_QUEUE,
|
|
XCE_SCHED_OP_UPDATE_RESOURCES,
|
|
XCE_SCHED_OP_UPDATE_RESVALS_FULL,
|
|
XCE_SCHED_OP_UPDATE_RESVALS,
|
|
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 {
|
|
data_item_t link;
|
|
xce_sched_op_t op;
|
|
union {
|
|
xce_conf4sched_t* config;
|
|
struct {
|
|
int njobs;
|
|
xce_job_t** jobs;
|
|
} jobupdate;
|
|
struct {
|
|
xce_job_t* job;
|
|
} jobfieldsupdate;
|
|
struct {
|
|
int nworkers;
|
|
xce_worker_t** workers;
|
|
} workerupdate;
|
|
struct {
|
|
uint64_t jobid;
|
|
uint32_t state;
|
|
} job_state_change;
|
|
struct {
|
|
xce_queue_t* queue;
|
|
} queueupdate;
|
|
struct {
|
|
int nresources;
|
|
xce_resource_t** resources;
|
|
} resourceupdate;
|
|
struct {
|
|
int nvals;
|
|
xce_resval_t** vals;
|
|
} resvalfullupdate;
|
|
struct {
|
|
xce_resval_list_t* resvals;
|
|
} resvalupdate;
|
|
struct {
|
|
xce_job_rusages_t* rusages;
|
|
} job_rusage;
|
|
} u;
|
|
} xce_sched_update_request_t;
|
|
|
|
#define ITPTR_TO_UPDATE_REQ(p) \
|
|
((xce_sched_update_request_t*)(((char*)p) \
|
|
- offsetof(xce_sched_update_request_t, \
|
|
link)))
|
|
|
|
typedef struct {
|
|
mylock_t lock;
|
|
int stop;
|
|
xcecli_ctx_t apictx;
|
|
uint64_t nextid;
|
|
xce_conf4sched_t* config;
|
|
uint64_t configver;
|
|
var plugins;
|
|
void* seqjob_allocator;
|
|
void* parjob_allocator;
|
|
void* affinity_allocator;
|
|
void* gpu_allocator;
|
|
dataq_t* pend_updates[2];
|
|
var resmgr;
|
|
var ugmgr;
|
|
var qgmgr;
|
|
var pgmgr;
|
|
var hgmgr;
|
|
zarray_t sharedInst;
|
|
zarray_t sharedInUse;
|
|
var jobs;
|
|
var queues;
|
|
var workers;
|
|
var users;
|
|
var 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 */
|
|
_Atomic int session_active; /* 1 during scheduling session, 0 between */
|
|
_Atomic int eval_in_progress; /* 1 while eval_select loop runs */
|
|
_Atomic int staging_merge_complete; /* 1 after staging merge completes */
|
|
select_eval_ctx_t* sel_eval_ctx; /* session-level memoisation context; NULL between sessions */
|
|
xce_cluster_config_t cluster_cfg; /* HA candidate list — empty means single-host mode */
|
|
} xce_scheduler_t;
|
|
|
|
extern int sched_init(xce_scheduler_t*, xce_client_config_t*,
|
|
const xce_cluster_config_t*, xce_request_handler_t);
|
|
extern void sched_fini(xce_scheduler_t*);
|
|
extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*);
|
|
extern int sched_update(xce_scheduler_t*);
|
|
extern void sched_go(xce_scheduler_t*);
|
|
extern void sched_begin_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 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_worker_t* sched_get_worker(xce_scheduler_t*, char*);
|
|
extern xce_sched_user_t* sched_get_user(xce_scheduler_t*, char*);
|
|
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_apply_job_rusage(xce_scheduler_t*, xce_job_rusages_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*);
|
|
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*);
|
|
extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);
|
|
extern void sched_worker_release_host_ctx(xce_sched_worker_t*);
|
|
extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*,
|
|
xce_select_t*);
|
|
extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_t*, xvec_idval_t*,
|
|
xvec_idval_t*);
|
|
extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*,
|
|
xvec_t*, xvec_idval_t*, xvec_idval_t*);
|
|
extern int sched_collect_workers(xce_scheduler_t*, xlist_t*,
|
|
zarray_iter_skip_fn_t,
|
|
sched_group_worker_fn_t);
|
|
extern int sched_charge_account(xce_scheduler_t*, xce_job_account_t*);
|
|
extern int sched_uncharge_account(xce_scheduler_t*, xce_job_account_t*);
|
|
extern int sched_commit_job_alloc(xce_scheduler_t*, xce_sched_job_t*);
|
|
extern uint64_t sched_get_id(xce_scheduler_t*);
|
|
extern xce_alloc_t** sched_make_allocs(xce_scheduler_t*, char*, int, int*,
|
|
float*, int*);
|
|
extern int sched_try_alloc_on_hosts(xce_worker_group_t*, va_list);
|
|
extern int get_resvals(zarray_t*, int, int*, float*);
|
|
extern int put_resvals(zarray_t*, int, int*, float*);
|
|
extern request_res_t* make_resource_request(char*, float);
|
|
extern int sched_fill_limit_requests(xce_scheduler_t*, xvec_idval_t*, xlist_t*);
|
|
|
|
#endif
|