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

1st workable scheduler with resource from controller

Browse source
This commit is contained in:
Rongsong Shen 2025-12-31 15:48:33 +08:00
parent 780e9fb313
commit bef32ace38
8 changed files with 238 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -106,6 +106,8 @@ extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*, int,
int*, float*, int*, int*, float*);
extern int sched_collect_workers(xce_scheduler_t*, xlist_t*,
zarray_iter_skip_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*,

37
src/scheduler/src/resource.c
View file

@ -3,12 +3,12 @@
#include <stdlib.h>
#include "kvmgr.h"
#include "objmgr.h"
#include "sched/types.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
#include "zarray.h"
#include "objmgr.h"
static int get_resvals(zarray_t* array, int nres, int* idxs, float* vals)
{
@ -187,9 +187,9 @@ int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w,
return 1;
}
static int sched_charge_res_accounts(xce_scheduler_t* sched, zarray_t* array,
static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array,
int naccounts,
xce_res_account_t** accounts)
xce_res_account_t** accounts, int charge)
{
int i;
int idxs[naccounts + 1];
@ -223,7 +223,12 @@ static int sched_charge_res_accounts(xce_scheduler_t* sched, zarray_t* array,
get_resvals(array, naccounts, idxs, vals);
for (i = 0; i < naccounts; i++) {
vals[i] += (accounts[i]->used + accounts[i]->reserved);
float val = accounts[i]->used + accounts[i]->reserved;
if (charge) {
vals[i] += val;
} else {
vals[i] -= val;
}
}
put_resvals(array, naccounts, idxs, vals);
@ -231,13 +236,14 @@ static int sched_charge_res_accounts(xce_scheduler_t* sched, zarray_t* array,
return 0;
}
static int sched_charge_account(xce_scheduler_t* sched,
xce_job_account_t* account)
static int sched_update_account(xce_scheduler_t* sched,
xce_job_account_t* account, int charge)
{
int i;
sched_charge_res_accounts(sched, &(sched->sharedInUse),
account->nres_shared, account->accounts_shared);
sched_update_res_accounts(sched, &(sched->sharedInUse),
account->nres_shared, account->accounts_shared,
charge);
for (i = 0; i < account->nhosts; i++) {
xce_host_account_t* haccount = account->accounts[i];
xce_sched_worker_t* w = NULL;
@ -251,11 +257,22 @@ static int sched_charge_account(xce_scheduler_t* sched,
continue;
}
sched_charge_res_accounts(sched, &(w->resInUse), haccount->naccount,
haccount->accounts);
sched_update_res_accounts(sched, &(w->resInUse), haccount->naccount,
haccount->accounts, charge);
}
return 0;
}
int sched_charge_account(xce_scheduler_t* sched, xce_job_account_t* acct)
{
return sched_update_account(sched, acct, 1);
}
int sched_uncharge_account(xce_scheduler_t* sched, xce_job_account_t* acct)
{
return sched_update_account(sched, acct, 0);
}
int sched_commit_job_alloc(xce_scheduler_t* sched, xce_sched_job_t* job)
{
if (job->accounts == NULL) {

88
src/scheduler/src/scheduler.c
View file

@ -9,6 +9,7 @@
#include "controller.capnp.h"
#include "decision.h"
#include "kvmgr.h"
#include "misc.h"
#include "objmgr.h"
#include "plugin.h"
#include "preason.h"
@ -333,8 +334,6 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs,
if (jold == NULL) {
sched_add_job(sched, j);
} else {
/**/
}
}
@ -352,6 +351,17 @@ static int sched_job_state_change(xce_scheduler_t* sched, uint64_t jobid,
rc = plugin_job_state_change(sched, job, state);
if (!JOB_IS_RUN(state)) {
if (job->accounts != NULL) {
free_job_account(job->accounts);
job->accounts = NULL;
}
if (job->allocs != NULL) {
free_JobAlloc_ptr(&(job->allocs));
}
}
job->state = state;
return rc;
@ -368,7 +378,6 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
plugin_worker_update(sched, wold, w);
if (wold == NULL) {
sched_add_worker(sched, w);
} else {
}
}
@ -439,10 +448,83 @@ void sched_go(xce_scheduler_t* sched)
return;
}
static int my_reset_resval_(xce_scheduler_t* sched, xce_val_t* val)
{
return reset_resval(val);
}
static int sched_reset_shared_resval(xce_scheduler_t* sched)
{
return zarray_iter(&(sched->sharedInUse),
(zarray_iter_fn_t)my_reset_resval_, sched);
}
static int sched_reset_worker_(xce_scheduler_t* sched, xce_sched_worker_t* w)
{
if (w == NULL) {
return 0;
}
if (IS_WORKER_UNAVAIL(w->status)) {
return 0;
}
return zarray_iter(&(w->resInUse), (zarray_iter_fn_t)my_reset_resval_,
sched);
}
static int sched_reset_workers(xce_scheduler_t* sched)
{
xlist_t lst;
int rc = 0;
xlist_init(&lst, 2);
xlist_store(&lst, sched_reset_worker_);
rc = kvmgr_iter(&(sched->workers), NULL, &lst, sched);
xlist_fini(&lst);
return rc;
}
static int charge_job_(xce_scheduler_t* sched, xce_sched_job_t* j)
{
if (j == NULL) {
return 0;
}
if (!JOB_IS_RUN(j->state)) {
return 0;
}
if (j->accounts == NULL) {
return 0;
}
return sched_charge_account(sched, j->accounts);
}
static int sched_account_jobs(xce_scheduler_t* sched)
{
xlist_t lst;
int rc = 0;
xlist_init(&lst, 2);
xlist_store(&lst, charge_job_);
rc = kvmgr_iter(&(sched->jobs), NULL, &lst, sched);
xlist_fini(&lst);
return rc;
}
void sched_begin_session(xce_scheduler_t* sched)
{
plugin_begin_session(sched);
sched_reset_shared_resval(sched);
sched_reset_workers(sched);
sched_account_jobs(sched);
return;
}