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

save works

Browse source
This commit is contained in:
Rongsong Shen 2026-03-02 12:14:30 +08:00
parent b748e30d20
commit f4b1ba4cd7
21 changed files with 373 additions and 324 deletions
Download patch file Download diff file Expand all files Collapse all files

143
src/controller/job.c
View file

@ -23,13 +23,7 @@ static job_t* create_job(cluster_t* c, xce_job_t* j)
{
job_t* job = NULL;
job = calloc(1, sizeof(*job));
if (job == NULL) {
return NULL;
}
/* FIXME: */
job->job = makeJob(j);
job = makeJob(c->jobmgr, j);
return job;
}
@ -198,9 +192,9 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
return -errno;
}
if ((w->ep == NULL) || (w->host->status != 0)) {
if ((w->ep == NULL) || (w->host.status != 0)) {
ERROR("worker %s is not in correct state, status %x ep %p",
w->host->name, w->host->status, w->ep);
w->host.name, w->host.status, w->ep);
free_launchinfo(&launchinfo);
obj_deref(w);
@ -215,41 +209,40 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
char* q_pre_exec;
char* q_post_exec;
rc = get_queue_exec(c, j->job->submit->queue, &q_pre_exec,
&q_post_exec);
rc = get_queue_exec(c, j->job.submit->queue, &q_pre_exec, &q_post_exec);
if (rc < 0) {
obj_deref(j);
obj_deref(w);
return rc;
}
if (j->job->submit->start_info->q_pre_exec != NULL) {
obj_deref(j->job->submit->start_info->q_pre_exec);
if (j->job.submit->start_info->q_pre_exec != NULL) {
obj_deref(j->job.submit->start_info->q_pre_exec);
}
if (j->job->submit->start_info->q_post_exec != NULL) {
obj_deref(j->job->submit->start_info->q_post_exec);
if (j->job.submit->start_info->q_post_exec != NULL) {
obj_deref(j->job.submit->start_info->q_post_exec);
}
j->job->submit->start_info->q_pre_exec = q_pre_exec;
j->job->submit->start_info->q_post_exec = q_post_exec;
j->job.submit->start_info->q_pre_exec = q_pre_exec;
j->job.submit->start_info->q_post_exec = q_post_exec;
}
einfo.jobid = alloc->jobid;
einfo.start_info = j->job->submit->start_info;
einfo.start_info = j->job.submit->start_info;
einfo.launchinfo = &launchinfo;
rc = xce_startjob_req(w->ep, &einfo);
free_launchinfo(&launchinfo);
if (j->job->allocs != NULL) {
free_JobAlloc_ptr(&(j->job->allocs));
if (j->job.allocs != NULL) {
free_JobAlloc_ptr(&(j->job.allocs));
}
j->job->allocs = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t));
j->job->allocs->jobid = alloc->jobid;
j->job->allocs->nhosts = nc;
MOVE_PTR(j->job->allocs->allocs, allocs);
j->job.allocs = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t));
j->job.allocs->jobid = alloc->jobid;
j->job.allocs->nhosts = nc;
MOVE_PTR(j->job.allocs->allocs, allocs);
alloc->allocs = NULL;
DEREF_OBJ(w);
@ -282,8 +275,8 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
return -ENOENT;
}
if (j->job->allocs != NULL) {
xce_host_alloc_t** allocs = j->job->allocs->allocs;
if (j->job.allocs != NULL) {
xce_host_alloc_t** allocs = j->job.allocs->allocs;
worker_t* w = NULL;
/* job control command will be sent to first node
@ -295,15 +288,15 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
}
if (w != NULL) {
/* FIXME: better method to check host status */
if ((w->ep != NULL) && (w->host->status == 0)) {
if ((w->ep != NULL) && (w->host.status == 0)) {
rc = xce_jobctrl_req(w->ep, ctrinfo);
if (rc < 0) {
ERROR("job ctrl %ld on host %s returns %d", jobid,
w->host->name, rc);
w->host.name, rc);
}
} else {
ERROR("worker %s is not in correct state, status %x ep %p",
w->host->name, w->host->status, w->ep);
w->host.name, w->host.status, w->ep);
rc = -errno;
}
obj_deref(w);
@ -322,9 +315,9 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
case JobControl_u_siginfo:
switch (ctrinfo->u.siginfo->signo) {
case SIGCONT:
if (j->job->state
if (j->job.state
& (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) {
int state = j->job->state;
int state = j->job.state;
/* resume suspend job? */
state &= ~(JOB_STAT_PSUSP | JOB_STAT_USUSP);
@ -333,9 +326,9 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
break;
case SIGSTOP:
case SIGTSTP:
if (!(j->job->state
if (!(j->job.state
& (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP))) {
int state = j->job->state;
int state = j->job.state;
state |= JOB_STAT_PSUSP;
mark_job_update(c, j, state);
@ -344,7 +337,7 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
case SIGKILL:
case SIGTERM:
default:
if (j->job->state
if (j->job.state
& (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
break;
}
@ -357,8 +350,7 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
}
break;
case JobControl_u_ckptoption:
if (j->job->state
& (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
if (j->job.state & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
ERROR("job <%ld> has been finished, no checkpoint can be done.",
jobid);
} else {
@ -509,20 +501,20 @@ static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_START, c, info, job);
/* update job information */
job->job->state = JOB_STAT_RUN;
job->job->start_time = time(NULL);
if (job->job->pidinfo == NULL) {
job->job->pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t));
job->job.state = JOB_STAT_RUN;
job->job.start_time = time(NULL);
if (job->job.pidinfo == NULL) {
job->job.pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t));
}
job->job->pidinfo->npids = copy_int_list((int**)&(job->job->pidinfo->pids),
(int*)info->pids, info->npids);
job->job->pidinfo->npgids = copy_int_list(
(int**)&(job->job->pidinfo->pgids), (int*)info->pgids, info->npgids);
job->job.pidinfo->npids = copy_int_list((int**)&(job->job.pidinfo->pids),
(int*)info->pids, info->npids);
job->job.pidinfo->npgids = copy_int_list((int**)&(job->job.pidinfo->pgids),
(int*)info->pgids, info->npgids);
if (info->exdata != NULL) {
job->job->pidinfo->exdata = makeString(info->exdata);
job->job.pidinfo->exdata = makeString(info->exdata);
} else {
job->job->pidinfo->exdata = NULL;
job->job.pidinfo->exdata = NULL;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
@ -556,24 +548,23 @@ static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
HOOK_DETAIL_JOB_EXIT, c, finish, job);
/* update job information */
if (finish->exit_code == 0) {
job->job->state = JOB_STAT_DONE;
job->job.state = JOB_STAT_DONE;
} else {
job->job->state = JOB_STAT_EXIT;
job->job.state = JOB_STAT_EXIT;
}
job->job->finish_time = time(NULL);
job->job.finish_time = time(NULL);
/* fill job run time/rusage information */
if (job->job->exit_info == NULL) {
job->job->exit_info
= (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
if (job->job.exit_info == NULL) {
job->job.exit_info = (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
}
if (job->job->exit_info != NULL) {
job->job->exit_info->exitcode = finish->exit_code;
if (job->job.exit_info != NULL) {
job->job.exit_info->exitcode = finish->exit_code;
}
job->job->run_time = job->job->finish_time - job->job->start_time;
job->job.run_time = job->job.finish_time - job->job.start_time;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_EXIT, c, finish, job);
@ -609,20 +600,20 @@ static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
switch (failinfo->stage) {
case JOB_EXEC_STAGE_Q_POST_EXEC:
case JOB_EXEC_STAGE_POST_EXEC:
if ((job->job->state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
if ((job->job.state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
WARN("job %ld got post exec fail notification but job's exit "
"status has not been set",
job->job->jobid);
job->job->state |= JOB_STAT_EXIT;
job->job.jobid);
job->job.state |= JOB_STAT_EXIT;
}
job->job->state |= JOB_STAT_PERR;
job->job.state |= JOB_STAT_PERR;
break;
default:
job->job->state = JOB_STAT_EXIT;
job->job.state = JOB_STAT_EXIT;
break;
}
job->job->finish_time = time(NULL);
job->job.finish_time = time(NULL);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
@ -635,7 +626,7 @@ static int mark_job_update(cluster_t* c, job_t* job, int state)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);
/* update job state */
job->job->state = state;
job->job.state = state;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);
@ -674,7 +665,7 @@ int on_job_stop(cluster_t* c, uint64_t jobid)
return -errno;
}
state = job->job->state;
state = job->job.state;
state |= JOB_STAT_USUSP;
mark_job_update(c, job, state);
@ -695,7 +686,7 @@ int on_job_resume(cluster_t* c, uint64_t jobid)
return -errno;
}
state = job->job->state;
state = job->job.state;
state &= ~JOB_STAT_USUSP;
mark_job_update(c, job, state);
@ -715,14 +706,14 @@ int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid)
return -errno;
}
state = job->job->state;
state = job->job.state;
if ((state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
WARN("job %ld got post exec fail notification but job's exit status "
"has not been set",
job->job->jobid);
job->job.jobid);
state |= JOB_STAT_EXIT;
/* ensure that job finish time has been set */
job->job->finish_time = time(NULL);
job->job.finish_time = time(NULL);
}
state |= JOB_STAT_PDONE;
@ -756,11 +747,11 @@ int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_CKPT, c, j);
if (j->job->ckptinfo != NULL) {
obj_deref(j->job->ckptinfo);
if (j->job.ckptinfo != NULL) {
obj_deref(j->job.ckptinfo);
}
j->job->ckptinfo = makeString(info);
j->job.ckptinfo = makeString(info);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_CKPT, c, j);
@ -809,7 +800,7 @@ int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request)
return errno;
}
host = makeString(w->host->name);
host = makeString(w->host.name);
obj_deref(w);
if (host == NULL) {
@ -821,12 +812,12 @@ int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, j);
update_host_runusage(&(j->job->run_usage), host, rusages->count,
update_host_runusage(&(j->job.run_usage), host, rusages->count,
rusages->usages);
update_pids(&(j->job->pidinfo), rusages->npids, rusages->pids);
update_pids(&(j->job.pidinfo), rusages->npids, rusages->pids);
update_pgids(&(j->job->pidinfo), rusages->npgids, rusages->pgids);
update_pgids(&(j->job.pidinfo), rusages->npgids, rusages->pgids);
/* FIXME
* update other fields of job rusage later
@ -856,7 +847,7 @@ static int set_preason(cluster_t* c, xce_pendreason_t* reasons)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_SETPREASON, c, j);
job = j->job;
job = &(j->job);
if (job->pend_reasons != NULL) {
erase_job_preason(job->pend_reasons);
FREEUP(job->pend_reasons);
@ -923,7 +914,7 @@ int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_CLRPREASON, c, j);
job = j->job;
job = &(j->job);
if (job->pend_reasons != NULL) {
erase_job_preason(job->pend_reasons);
FREEUP(job->pend_reasons);