From e497dd1b693ec31bcf3d97c13b0f2be22fd56651 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 28 Nov 2025 17:24:31 +0800 Subject: [PATCH] job submit and support run in worker --- src/controller/job.c | 84 ++++++++++------ src/controller/meson.build | 8 +- src/controller/testsuite/test.c | 146 +++++++++++++++++++++++----- src/controller/xceobjs.c | 144 ++++++++++++++------------- src/include/utils.h | 11 ++- src/include/xce/types.h | 59 ++++++----- src/lib/utils.c | 30 +++++- src/schema/controller.capnp | 81 +++++++-------- src/worker/include/xcecd_utils.h | 15 ++- src/worker/src/main.c | 45 +++++++++ src/worker/src/process_manager.c | 11 ++- src/worker/src/uv_proc.c | 27 ++++- src/worker/src/xcecd.c | 8 +- src/worker/src/xcecd_cgroups.c | 9 +- src/worker/src/xcecd_dbus_manager.c | 20 ++-- src/worker/src/xcecd_exdata.c | 20 +++- src/worker/src/xcecd_worker.c | 4 +- src/worker/src/xcecd_worker_utils.c | 16 ++- 18 files changed, 502 insertions(+), 236 deletions(-) diff --git a/src/controller/job.c b/src/controller/job.c index bcbd10c..d74659a 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -16,32 +16,61 @@ static int mark_job_update(cluster_t*, job_t*, int); uint64_t jobid = 100; -static job_t* create_job(cluster_t* c, uint64_t jobid, char* queue, char* user, - char* exechome, char* execcwd, char* execuser, - char* cmdline, char* exdata, uint32_t mintasks, - uint32_t maxtasks) +static job_t* create_job(cluster_t* c, xce_job_t* j) { - job_t* job = NULL; + job_t* job = NULL; + static int typeid = -1; + + if (typeid < 0) { + typeid = typeid_of_name(&(c->objmgr), "xjob"); + TRACE("%s type %s id %d", __func__, "xjob", typeid); + } + + if (typeid < 0) { + SET_ERROR(-typeid); + return NULL; + } + + job = create_obj_by_typeid(&(c->objmgr), typeid, j); + return job; } -uint64_t job_submit(void* c, char* user, char* queue, char* resreq, - char* cmdline, char* exdata, uint32_t mintasks, - uint32_t maxtasks) +uint64_t job_submit(void* c, xce_job_submit_t* submit) { - job_t* job = NULL; + job_t* job = NULL; + uint64_t base_id = submit->jobid; + xce_job_t j = { .submit = submit->submit }; + time_t now = time(NULL); + int i; - job = create_job(c, jobid++, queue, user, "/tmp", "/tmp", user, cmdline, - exdata, mintasks, maxtasks); - if (job == NULL) { - return -errno; + j.sub_time = now; + j.queue_time = now; + j.start_time = 0; + j.finish_time = 0; + j.pidinfo = NULL; + j.state = JOB_STAT_PEND; + j.allocs = NULL; + j.run_time = 0; + j.run_usage = NULL; + j.exit_info = NULL; + j.ckptinfo = NULL; + j.pending_ctrl = 0; + + for (i = 0; i < submit->array_size; i++) { + j.jobid = base_id + i; + job = create_job(c, &j); + if (job == NULL) { + return -errno; + } } - return job->job.jobid; + return base_id; } int load_jobs(void* c) { +#if 0 job_t* job; int i; char cmdline[1024]; @@ -62,6 +91,7 @@ int load_jobs(void* c) return -errno; } } +#endif return 0; } @@ -181,27 +211,27 @@ 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.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.start_info->q_pre_exec != NULL) { - obj_deref(j->job.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.start_info->q_post_exec != NULL) { - obj_deref(j->job.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.start_info->q_pre_exec = q_pre_exec; - j->job.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.start_info; + einfo.start_info = j->job.submit->start_info; einfo.launchinfo = &launchinfo; rc = xce_startjob_req(w->ep, &einfo); @@ -398,18 +428,16 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) init_reply_by_req(&reply, request); - if (NULLSTR(submit->user) || NULLSTR(submit->queue) - || NULLSTR(submit->cmdline)) { + if (NULLSTR(submit->submit->user) || NULLSTR(submit->submit->queue) + || (submit->submit->start_info == NULL) + || NULLSTR(submit->submit->start_info->cmdline)) { jobid = 0; errno = EINVAL; } else { xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_NEW, c, submit); - jobid = job_submit(c, (char*)submit->user, (char*)submit->queue, - (char*)submit->resreq, (char*)submit->cmdline, - (char*)submit->exdata, submit->mintasks, - submit->maxtasks); + jobid = job_submit(c, submit); } if (jobid == 0) { reply.status = -errno; diff --git a/src/controller/meson.build b/src/controller/meson.build index c722920..6c56b66 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -43,7 +43,13 @@ readline_dep = dependency('readline') ncurses_dep = dependency('ncurses') history_dep = dependency('history') -xce_test_deps = [libxce_dep, readline_dep, history_dep, ncurses_dep] +xce_test_deps = [ + libxce_dep, + libcjson_dep, + readline_dep, + history_dep, + ncurses_dep +] xce_test = executable('xce_test', xce_test_src, include_directories: [incdir], diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index 059fe45..3d9d17c 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -12,8 +12,11 @@ #include #include +#include + #include "client.h" #include "controller.capnp.h" +#include "idgen.h" #include "xce/types.h" typedef int (*cmd_func_t)(int, char*[]); @@ -97,7 +100,8 @@ static void handler(int, siginfo_t*, void*); #define CLOCKID CLOCK_REALTIME #define SIG SIGRTMIN -static timer_t timerid; +static timer_t timerid; +static idgen_ctx_t idctx; static setting_t settings[] = { { (char*)"host", (char*)NULL, (char*)"localhost" }, @@ -562,12 +566,12 @@ int cmd_workerlist(int argc, char* argv[]) return rc; } - printf("%-20s%-8s%-15s%-15s%12s%12s%12s%12s%12s%12s\n", "HostName", + printf("%-20s%-8s%-25s%-25s%12s%12s%12s%12s%12s%12s\n", "HostName", "Status", "HostType", "HostModel", "CPUfactor", "MaxCPU", "MaxMem", "MaxSwap", "MaxTmp", "NDisks"); for (i = 0; i < rc; i++) { - printf("%-20s%-8s%-15s%-15s%12f%12d%12f%12f%12f%12d\n", + printf("%-20s%-8s%-25s%-25s%12f%12d%12f%12f%12f%12d\n", workers[i]->name, worker_status_str(workers[i]->status), workers[i]->attrs->hosttype, workers[i]->attrs->hostmodel, workers[i]->attrs->cpufactor, workers[i]->attrs->maxcpus, @@ -619,16 +623,21 @@ int cmd_joblist(int argc, char* argv[]) return rc; } - printf("%-16s%-10s%-10s%-8s%-16s%-16s%-16s%10s%30s\n", "JobId", "User", + printf("%-26s%-10s%-10s%-8s%-16s%-16s%-16s%10s%30s\n", "JobId", "User", "Queue", "State", "SubmitTime", "StartTime", "FinishTime", "NTasks", "Command"); for (i = 0; i < rc; i++) { - printf("%-16ld%-10s%-10s%-8s%-16ld%-16ld%-16ld%4d,%-5d%30s\n", - jobs[i]->jobid, jobs[i]->user, jobs[i]->queue, + char buf[128]; + + fmt_jobid(buf, sizeof(buf) - 1, jobs[i]->jobid); + + printf("%-26s%-10s%-10s%-8s%-16ld%-16ld%-16ld%4d,%-5d%30s\n", buf, + jobs[i]->submit->user, jobs[i]->submit->queue, job_state_str(jobs[i]->state), jobs[i]->queue_time, - jobs[i]->start_time, jobs[i]->finish_time, jobs[i]->mintasks, - jobs[i]->maxtasks, jobs[i]->start_info->cmdline); + jobs[i]->start_time, jobs[i]->finish_time, + jobs[i]->submit->mintasks, jobs[i]->submit->maxtasks, + jobs[i]->submit->start_info->cmdline); } free(jobs); @@ -723,7 +732,7 @@ int cmd_jobrun(int argc, char* argv[]) nhosts++; break; case 'j': - jobid = atol(optarg); + parse_jobid(&jobid, optarg); if (jobid == 0) { printf("you need to specify a positive jobid with -j"); return -1; @@ -756,19 +765,76 @@ int cmd_jobrun(int argc, char* argv[]) return 0; } +static int build_job_envs(xce_job_submit_t* submit) +{ + char* names[] = { "PATH", "LSAN_OPTIONS", NULL }; + int i; + cJSON* top; + cJSON* envars; + char* exdata; + + top = cJSON_CreateObject(); + if (top == NULL) { + return -1; + } + + envars = cJSON_CreateObject(); + if (envars == NULL) { + return -1; + } + + cJSON_AddItemToObject(top, "sub_env_vars", envars); + + for (i = 0; names[i] != NULL; i++) { + cJSON* val; + char* s = getenv(names[i]); + + if (s == NULL) { + s = ""; + } + + val = cJSON_CreateString(s); + cJSON_AddItemToObject(envars, names[i], val); + } + + exdata = cJSON_Print(top); + + submit->submit->start_info->exdata = exdata; + + cJSON_Delete(top); + + return 0; +} + +static int build_job_exdata(xce_job_submit_t* submit) +{ + build_job_envs(submit); + + return 0; +} + int cmd_jobsub(int argc, char* argv[]) { - int rc = 0; - xce_job_submit_t submit; - int opt; - uint64_t jobid; - uint32_t min = 1, max = 1; + int rc = 0; + job_submit_data_t submit; + xce_job_submit_t jobsubmit = { .submit = &submit }; + int opt; + uint64_t jobid; + uint32_t min = 1, max = 1; + uint32_t arsize = 1; + char jobidStr[128]; optind = 1; memset(&submit, 0x0, sizeof(submit)); + submit.start_info + = (job_startinfo_t*)calloc(1, sizeof(*(submit.start_info))); + if (submit.start_info == NULL) { + perror("calloc"); + return -ENOMEM; + } - while ((opt = getopt(argc, argv, "u:q:R:c:x:n:")) != -1) { + while ((opt = getopt(argc, argv, "u:q:R:c:x:n:a:")) != -1) { switch (opt) { case 'u': submit.user = strdup(optarg); @@ -780,10 +846,10 @@ int cmd_jobsub(int argc, char* argv[]) submit.resreq = strdup(optarg); break; case 'c': - submit.cmdline = strdup(optarg); + submit.start_info->cmdline = strdup(optarg); break; case 'x': - submit.exdata = strdup(optarg); + submit.start_info->exdata = strdup(optarg); break; case 'n': if (strchr(optarg, (int)',') != NULL) { @@ -805,6 +871,12 @@ int cmd_jobsub(int argc, char* argv[]) max = min; } break; + case 'a': + arsize = atoi(optarg); + if (arsize <= 0) { + arsize = 1; + } + break; default: printf( "jobsub -u -q -R -c \n"); @@ -827,7 +899,7 @@ int cmd_jobsub(int argc, char* argv[]) return -EINVAL; } - if (NULLSTR(submit.cmdline)) { + if (NULLSTR(submit.start_info->cmdline)) { printf("you need to specify job command with -c\n"); return -EINVAL; } @@ -839,7 +911,24 @@ int cmd_jobsub(int argc, char* argv[]) submit.mintasks = min; submit.maxtasks = max; - rc = xcecli_jobsubmit(active_ctx->ctx, &submit, &jobid); + idgen(&idctx, 0, 0, arsize, &(jobsubmit.jobid)); + + if (NULLSTR(submit.start_info->execuser)) { + submit.start_info->execuser = strdup(submit.user); + } + + if (NULLSTR(submit.start_info->exechome)) { + submit.start_info->exechome = strdup("/tmp"); + } + + if (NULLSTR(submit.start_info->execcwd)) { + submit.start_info->execcwd = strdup("/tmp"); + } + + build_job_exdata(&jobsubmit); + + jobsubmit.array_size = arsize; + rc = xcecli_jobsubmit(active_ctx->ctx, &jobsubmit, &jobid); if (rc < 0) { SET_ERROR(-rc); perror("xcecli_jobsubmit"); @@ -849,10 +938,13 @@ int cmd_jobsub(int argc, char* argv[]) FREEUP(submit.user); FREEUP(submit.queue); FREEUP(submit.resreq); - FREEUP(submit.cmdline); - FREEUP(submit.exdata); + FREEUP(submit.start_info->cmdline); + FREEUP(submit.start_info->exdata); + FREEUP(submit.start_info); - printf("jobsubmit ok: JobId: %lu\n", jobid); + fmt_jobid(jobidStr, sizeof(jobidStr) - 1, array_jobid(jobid)); + + printf("jobsubmit ok: JobId: %s\n", jobidStr); return 0; } @@ -940,7 +1032,7 @@ int cmd_jobkill(int argc, char* argv[]) signo = atoi(optarg); break; case 'j': - jobid = atol(optarg); + parse_jobid(&jobid, optarg); break; default: break; @@ -986,7 +1078,7 @@ int cmd_jobckpt(int argc, char* argv[]) ckptoption.kill = 1; break; case 'j': - jobid = atol(optarg); + parse_jobid(&jobid, optarg); break; case 'm': ckptoption.method = optarg; @@ -1146,7 +1238,7 @@ int cmd_runtasks(int argc, char* argv[]) dir = strdup(optarg); break; case 'j': - jobid = atol(optarg); + parse_jobid(&jobid, optarg); break; case 't': launch_type = strdup(optarg); @@ -1268,7 +1360,7 @@ int cmd_jobfinish(int argc, char* argv[]) while ((opt = getopt(argc, argv, "j:c:")) != -1) { switch (opt) { case 'j': - jobid = atol(optarg); + parse_jobid(&jobid, optarg); break; case 'c': code = atoi(optarg); @@ -2242,6 +2334,8 @@ int main(int argc, char* argv[]) memset(&run_jobs, 0x0, sizeof(run_jobs)); memset(&run_tasks, 0x0, sizeof(run_tasks)); + idgen_ctx_init(&idctx); + while (!done) { char* s; int rc; diff --git a/src/controller/xceobjs.c b/src/controller/xceobjs.c index 94bf45a..4bcc9a6 100644 --- a/src/controller/xceobjs.c +++ b/src/controller/xceobjs.c @@ -114,30 +114,43 @@ static void* xjob_create(objmgr_t* mgr, objtype_t* type, va_list ap) job = MYOBJ_TO_OBJ(job_t*, obj); /* fill job data */ - job->job.jobid = submit->jobid; - job->job.queue = create_obj_by_typename(mgr, "xstring", submit->queue); - job->job.sub_time = submit->sub_time; - job->job.queue_time = submit->queue_time; + job->job.jobid = submit->jobid; + job->job.submit = (job_submit_data_t*)calloc(1, sizeof(*(job->job.submit))); + if (job->job.submit == NULL) { + return NULL; + } + job->job.submit->queue + = create_obj_by_typename(mgr, "xstring", submit->submit->queue); + /* FIXME: */ + if (1) { + time_t now = time(NULL); + + job->job.sub_time = now; + job->job.queue_time = now; + } + if (job->job.queue_time == 0) { job->job.queue_time = job->job.sub_time; } - job->job.state = submit->state; - job->job.user = create_obj_by_typename(mgr, "xstring", submit->user); + job->job.state = submit->state; + job->job.submit->user + = create_obj_by_typename(mgr, "xstring", submit->submit->user); job->job.start_time = submit->start_time; job->job.finish_time = submit->finish_time; - job->job.job_name - = create_obj_by_typename(mgr, "xstring", submit->job_name); - job->job.project_name - = create_obj_by_typename(mgr, "xstring", submit->project_name); - job->job.user_group - = create_obj_by_typename(mgr, "xstring", submit->user_group); - job->job.job_description - = create_obj_by_typename(mgr, "xstring", submit->job_description); - job->job.resreq = create_obj_by_typename(mgr, "xstring", submit->resreq); - job->job.run_time = submit->run_time; - job->job.mintasks = submit->mintasks; - job->job.maxtasks = submit->maxtasks; - job->job.pending_ctrl = submit->pending_ctrl; + job->job.submit->job_name + = create_obj_by_typename(mgr, "xstring", submit->submit->job_name); + job->job.submit->project_name + = create_obj_by_typename(mgr, "xstring", submit->submit->project_name); + job->job.submit->user_group + = create_obj_by_typename(mgr, "xstring", submit->submit->user_group); + job->job.submit->job_description = create_obj_by_typename( + mgr, "xstring", submit->submit->job_description); + job->job.submit->resreq + = create_obj_by_typename(mgr, "xstring", submit->submit->resreq); + job->job.run_time = submit->run_time; + job->job.submit->mintasks = submit->submit->mintasks; + job->job.submit->maxtasks = submit->submit->maxtasks; + job->job.pending_ctrl = submit->pending_ctrl; job->job.pidinfo = NULL; @@ -159,35 +172,35 @@ static void* xjob_create(objmgr_t* mgr, objtype_t* type, va_list ap) } } - job->job.start_info = NULL; + job->job.submit->start_info = NULL; - if (submit->start_info != NULL) { - job->job.start_info + if (submit->submit->start_info != NULL) { + job->job.submit->start_info = (job_startinfo_t*)calloc(1, sizeof(job_startinfo_t)); - if (job->job.start_info == NULL) { + if (job->job.submit->start_info == NULL) { SET_ERROR(ENOMEM); goto clean_job; } - job->job.start_info->cmdline = create_obj_by_typename( - mgr, "xstring", submit->start_info->cmdline); - job->job.start_info->exdata = create_obj_by_typename( - mgr, "xstring", submit->start_info->exdata); - job->job.start_info->execcwd = create_obj_by_typename( - mgr, "xstring", submit->start_info->execcwd); - job->job.start_info->exechome = create_obj_by_typename( - mgr, "xstring", submit->start_info->exechome); - job->job.start_info->execuser = create_obj_by_typename( - mgr, "xstring", submit->start_info->execuser); - job->job.start_info->launch_type = create_obj_by_typename( - mgr, "xstring", submit->start_info->launch_type); - if (submit->start_info->job_pre_exec != NULL) { - job->job.start_info->job_pre_exec = create_obj_by_typename( - mgr, "xstring", submit->start_info->job_pre_exec); + job->job.submit->start_info->cmdline = create_obj_by_typename( + mgr, "xstring", submit->submit->start_info->cmdline); + job->job.submit->start_info->exdata = create_obj_by_typename( + mgr, "xstring", submit->submit->start_info->exdata); + job->job.submit->start_info->execcwd = create_obj_by_typename( + mgr, "xstring", submit->submit->start_info->execcwd); + job->job.submit->start_info->exechome = create_obj_by_typename( + mgr, "xstring", submit->submit->start_info->exechome); + job->job.submit->start_info->execuser = create_obj_by_typename( + mgr, "xstring", submit->submit->start_info->execuser); + job->job.submit->start_info->launch_type = create_obj_by_typename( + mgr, "xstring", submit->submit->start_info->launch_type); + if (submit->submit->start_info->job_pre_exec != NULL) { + job->job.submit->start_info->job_pre_exec = create_obj_by_typename( + mgr, "xstring", submit->submit->start_info->job_pre_exec); } - if (submit->start_info->job_post_exec != NULL) { - job->job.start_info->job_post_exec = create_obj_by_typename( - mgr, "xstring", submit->start_info->job_post_exec); + if (submit->submit->start_info->job_post_exec != NULL) { + job->job.submit->start_info->job_post_exec = create_obj_by_typename( + mgr, "xstring", submit->submit->start_info->job_post_exec); } } @@ -203,17 +216,14 @@ static void* xjob_create(objmgr_t* mgr, objtype_t* type, va_list ap) MOVE_PTR(job->job.run_usage, submit->run_usage); } - /* clang-format off */ - if ((job->job.queue == NULL) || - (job->job.user == NULL) || - (job->job.job_name == NULL) || - (job->job.start_info->cmdline == NULL) || - (job->job.project_name == NULL) || - (job->job.job_description == NULL) || - (job->job.resreq == NULL)) { - /* clang-format on */ + if ((job->job.submit->queue == NULL) || (job->job.submit->user == NULL) + || (job->job.submit->job_name == NULL) + || (job->job.submit->start_info->cmdline == NULL) + || (job->job.submit->project_name == NULL) + || (job->job.submit->job_description == NULL) + || (job->job.submit->resreq == NULL)) { - FREEUP(job->job.start_info); + FREEUP(job->job.submit->start_info); job_obj_killer(job); @@ -240,12 +250,12 @@ void job_obj_killer(job_t* job) } obj = TO_MYOBJ(job); - DEREF_OBJ(job->job.queue); - DEREF_OBJ(job->job.user); - DEREF_OBJ(job->job.job_name); - DEREF_OBJ(job->job.project_name); - DEREF_OBJ(job->job.job_description); - DEREF_OBJ(job->job.resreq); + DEREF_OBJ(job->job.submit->queue); + DEREF_OBJ(job->job.submit->user); + DEREF_OBJ(job->job.submit->job_name); + DEREF_OBJ(job->job.submit->project_name); + DEREF_OBJ(job->job.submit->job_description); + DEREF_OBJ(job->job.submit->resreq); if (job->job.pidinfo != NULL) { FREEUP(job->job.pidinfo->pids); @@ -285,16 +295,16 @@ void job_obj_killer(job_t* job) FREEUP(job->job.run_usage); } - if (job->job.start_info != NULL) { - DEREF_OBJ(job->job.start_info->cmdline); - DEREF_OBJ(job->job.start_info->exechome); - DEREF_OBJ(job->job.start_info->execcwd); - DEREF_OBJ(job->job.start_info->execuser); - DEREF_OBJ(job->job.start_info->exdata); - DEREF_OBJ(job->job.start_info->launch_type); - DEREF_OBJ(job->job.start_info->job_pre_exec); - DEREF_OBJ(job->job.start_info->job_post_exec); - FREEUP(job->job.start_info); + if (job->job.submit->start_info != NULL) { + DEREF_OBJ(job->job.submit->start_info->cmdline); + DEREF_OBJ(job->job.submit->start_info->exechome); + DEREF_OBJ(job->job.submit->start_info->execcwd); + DEREF_OBJ(job->job.submit->start_info->execuser); + DEREF_OBJ(job->job.submit->start_info->exdata); + DEREF_OBJ(job->job.submit->start_info->launch_type); + DEREF_OBJ(job->job.submit->start_info->job_pre_exec); + DEREF_OBJ(job->job.submit->start_info->job_post_exec); + FREEUP(job->job.submit->start_info); } FREEUP(job->job.exit_info); diff --git a/src/include/utils.h b/src/include/utils.h index 9f6dfa7..2443748 100644 --- a/src/include/utils.h +++ b/src/include/utils.h @@ -40,6 +40,10 @@ & UMAX_BITS(N_SEQUENCE_BITS - SEQUENCE_HIGH_SHIFT)) \ << SEQUENCE_HIGH_SHIFT) +#define EXTRACT_ARRAY_BASEID(id) \ + MAKE_JOBID(EXTRACT_ID_CLUSTER(id), EXTRACT_ID_APISERVER(id), \ + EXTRACT_ID_TIMESTAMP(id), 0x0, EXTRACT_ARRAY_JOB(id)) + #define EXTRACT_ARRAY_IDX(id) \ EXTRACT_BITS(id, 0, (N_SEQUENCE_BITS - SEQUENCE_HIGH_SHIFT)) @@ -120,7 +124,10 @@ extern xce_res_threshold_t** dup_resource_thresholds(xce_res_threshold_t**, extern void free_resval_with_idx_list(xce_val_with_idx_t***, int); extern void free_strlist(xce_strlist_t*); -extern int fmt_jobid(char*, size_t, uint64_t); -extern int parse_jobid(uint64_t*, char*); +extern int fmt_jobid_ex(char*, size_t, uint64_t, char, char); +extern int fmt_jobid(char*, size_t, uint64_t); +extern int parse_jobid(uint64_t*, char*); +extern uint64_t array_jobid(uint64_t); +extern uint64_t array_idx(uint64_t); #endif diff --git a/src/include/xce/types.h b/src/include/xce/types.h index 5d8587d..00dabe6 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -5,6 +5,7 @@ #include "config.h" #include +#include #include #include "xlist.h" @@ -229,16 +230,16 @@ typedef struct { } xce_launch_t; typedef struct { - char* exechome; - char* execcwd; - char* execuser; - char* cmdline; - char* exdata; - char* launch_type; - char* q_pre_exec; - char* q_post_exec; - char* job_pre_exec; - char* job_post_exec; + char* exechome; + char* execcwd; + char* execuser; + char* cmdline; + char* exdata; + char* launch_type; + char* q_pre_exec; + char* q_post_exec; + char* job_pre_exec; + char* job_post_exec; } job_startinfo_t; typedef struct { @@ -284,15 +285,22 @@ typedef struct { } xce_job_dispatch_t; typedef struct { - char* queue; - char* user; - char* cmdline; - char* resreq; - char* exdata; - char* pre_exec; - char* post_exec; - uint32_t mintasks; - uint32_t maxtasks; + char* queue; + char* user; + uint32_t mintasks; + uint32_t maxtasks; + job_startinfo_t* start_info; + char* job_name; + char* project_name; + char* user_group; + char* job_description; + char* resreq; +} job_submit_data_t; + +typedef struct { + uint64_t jobid; + uint64_t array_size; + job_submit_data_t* submit; } xce_job_submit_t; typedef struct { @@ -383,25 +391,14 @@ typedef struct { typedef struct { uint64_t jobid; - uint64_t array_base_id; - int array_idx; - char* queue; - char* user; + job_submit_data_t* submit; int64_t idx; time_t sub_time; time_t queue_time; time_t start_time; time_t finish_time; - uint32_t mintasks; - uint32_t maxtasks; job_pidinfo_t* pidinfo; - job_startinfo_t* start_info; unsigned int state; - char* job_name; - char* project_name; - char* user_group; - char* job_description; - char* resreq; xce_job_alloc_t* allocs; int run_time; xce_job_runusage_t* run_usage; diff --git a/src/lib/utils.c b/src/lib/utils.c index e1deef0..17805ea 100644 --- a/src/lib/utils.c +++ b/src/lib/utils.c @@ -411,7 +411,7 @@ void free_strlist(xce_strlist_t* l) FREEUP(l->data); } -int fmt_jobid(char* buf, size_t sz, uint64_t jobid) +int fmt_jobid_ex(char* buf, size_t sz, uint64_t jobid, char lc, char rc) { uint32_t tag, id_cluster, id_apiserver, id_timestamp; char data[128]; @@ -430,8 +430,8 @@ int fmt_jobid(char* buf, size_t sz, uint64_t jobid) id_job = EXTRACT_ARRAY_JOB(jobid); id_idx = EXTRACT_ARRAY_IDX(jobid); - snprintf(data, sizeof(data) - 1, "%d-%d-%d-%d[%d]", id_cluster, - id_apiserver, id_timestamp, id_job, id_idx); + snprintf(data, sizeof(data) - 1, "%d-%d-%d-%d%c%d%c", id_cluster, + id_apiserver, id_timestamp, id_job, lc,id_idx, rc); } else { uint32_t id_seq; @@ -449,6 +449,10 @@ int fmt_jobid(char* buf, size_t sz, uint64_t jobid) return len; } +int fmt_jobid(char* buf, size_t sz, uint64_t jobid) { + return fmt_jobid_ex(buf, sz, jobid, '[', ']'); +} + int parse_jobid(uint64_t* jobid, char* buf) { regex_t regex; @@ -517,3 +521,23 @@ int parse_jobid(uint64_t* jobid, char* buf) regfree(®ex); return -EINVAL; } + +uint64_t array_jobid(uint64_t jobid) { + uint32_t tag = EXTRACT_ID_TAG(jobid); + + if (tag == 0x00) { + return jobid; + } + + return EXTRACT_ARRAY_BASEID(jobid); +} + +uint64_t array_idx(uint64_t jobid) { + uint32_t tag = EXTRACT_ID_TAG(jobid); + + if (tag == 0x00) { + return 0; + } + + return EXTRACT_ARRAY_IDX(jobid); +} diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index e0c3a37..3a8de61 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -231,46 +231,30 @@ struct JobStartInfo $C.mapname("job_startinfo_t"){ } struct Job $C.mapname("xce_job_t"){ - jobid @0: UInt64; - queue @1: Text; - user @2: Text; - idx @3: UInt64; - subtime @4: UInt64 $C.mapname("sub_time"); - queuetime @5: UInt64 $C.mapname("queue_time"); - starttime @6: UInt64 $C.mapname("start_time"); - finishtime @7: UInt64 $C.mapname("finish_time"); - state @8: UInt32; - jobname @9: Text $C.mapname("job_name"); - projectname @10: Text $C.mapname("project_name"); - usergroup @11: Text $C.mapname("user_group"); - desc @12: Text $C.mapname("job_description"); - resreq @13: Text; - allocs @14: JobAlloc $C.mapname("allocs"); - runtime @15: UInt32 $C.mapname("run_time"); - pidinfo @16: JobPIDInfo; - startinfo @17: JobStartInfo $C.mapname("start_info"); - mintasks @18: UInt32 $C.mapname("mintasks"); - maxtasks @19: UInt32 $C.mapname("maxtasks"); + jobid @0: UInt64; + submitData @1: JobSubmitData $C.mapname("submit"); + idx @2: UInt64; + subtime @3: UInt64 $C.mapname("sub_time"); + queuetime @4: UInt64 $C.mapname("queue_time"); + starttime @5: UInt64 $C.mapname("start_time"); + finishtime @6: UInt64 $C.mapname("finish_time"); + state @7: UInt32; + allocs @8: JobAlloc $C.mapname("allocs"); + runtime @9: UInt32 $C.mapname("run_time"); + pidinfo @10: JobPIDInfo; } struct SchedJob $C.mapname("xce_job_t"){ - jobid @0: UInt64; - queue @1: Text; - user @2: Text; - idx @3: UInt64; - subtime @4: UInt64 $C.mapname("sub_time"); - queuetime @5: UInt64 $C.mapname("queue_time"); - starttime @6: UInt64 $C.mapname("start_time"); - finishtime @7: UInt64 $C.mapname("finish_time"); - state @8: UInt32; - jobname @9: Text $C.mapname("job_name"); - projectname @10: Text $C.mapname("project_name"); - usergroup @11: Text $C.mapname("user_group"); - resreq @12: Text; - allocs @13: JobAlloc $C.mapname("allocs"); - runtime @14: UInt32 $C.mapname("run_time"); - mintasks @15: UInt32 $C.mapname("mintasks"); - maxtasks @16: UInt32 $C.mapname("maxtasks"); + jobid @0: UInt64; + submitData @1: JobSubmitData $C.mapname("submit"); + idx @2: UInt64; + subtime @3: UInt64 $C.mapname("sub_time"); + queuetime @4: UInt64 $C.mapname("queue_time"); + starttime @5: UInt64 $C.mapname("start_time"); + finishtime @6: UInt64 $C.mapname("finish_time"); + state @7: UInt32; + allocs @8: JobAlloc $C.mapname("allocs"); + runtime @9: UInt32 $C.mapname("run_time"); } struct JobData $C.mapname("xce_jobs_data_t") { @@ -448,14 +432,23 @@ struct ClusterConf $C.mapname("xce_clusterconf_t") { workers @3: List(Worker) $C.mapname("wlist.workers") $C.maplistcount("wlist.count"); } +struct JobSubmitData $C.mapname("job_submit_data_t") { + queue @0: Text; + user @1: Text; + mintasks @2: UInt32; + maxtasks @3: UInt32; + jobname @4: Text $C.mapname("job_name"); + project @5: Text $C.mapname("project_name"); + ugroup @6: Text $C.mapname("user_group"); + description @7: Text $C.mapname("job_description"); + resreq @8: Text; + startinfo @9: JobStartInfo $C.mapname("start_info"); +} + struct JobSubmit $C.mapname("xce_job_submit_t") { - queue @0: Text; - user @1: Text; - cmdline @2: Text; - resreq @3: Text; - exdata @4: Text; - mintasks @5: UInt32; - maxtasks @6: UInt32; + jobid @0: UInt64; + arraySize @1: UInt64 $C.mapname("array_size"); + submit @2: JobSubmitData; } struct JobSignal $C.mapname("xce_signal_t") { diff --git a/src/worker/include/xcecd_utils.h b/src/worker/include/xcecd_utils.h index d6cbbd0..2420eae 100644 --- a/src/worker/include/xcecd_utils.h +++ b/src/worker/include/xcecd_utils.h @@ -8,9 +8,16 @@ /* This header file has some utilities and structure for compatibility with * other scheduler systems. */ -#define XCE_DEFAULT_SOCKS 15 -#define MAXPATHLEN 1024 -#define MAXHOSTNAMELEN 256 +#define XCE_DEFAULT_SOCKS 15 + +#if !defined(MAXPATHLEN) +#define MAXPATHLEN 1024 +#endif + +#if !defined(MAXHOSTNAMELEN) +#define MAXHOSTNAMELEN 256 +#endif + #define MAXLINELEN 4096 #define MAXDATASIZE (16 * 1024) #define MAXXCENAMELEN 40 @@ -133,4 +140,6 @@ typedef struct { void block_signal(int sig); void unblock_signal(int sig); +extern int make_job_sd_unit_name(char*, size_t, const char*, uint64_t); + #endif diff --git a/src/worker/src/main.c b/src/worker/src/main.c index 1481a1d..1c13168 100644 --- a/src/worker/src/main.c +++ b/src/worker/src/main.c @@ -1,6 +1,9 @@ #include "sqlite3.h" #include "xcecd.h" + #include +#include +#include #include #include #include @@ -216,6 +219,44 @@ static void master_ckpt_result_callback(uint64_t job_id, int status, xcecd_job_ckpted(job_id, status, restore_cmd); } +static int check_xce_server_dir(char* myfile) +{ + char* xce_server_dir = getenv("XCE_SERVER_DIR"); + char myfullpath[XCE_MAXPATH_LEN + 1]; + + if (xce_server_dir != NULL) { + return 0; + } + + if (realpath(myfile, myfullpath) != NULL) { + char* mydir = dirname(myfullpath); + char workerpath[XCE_MAXPATH_LEN + 1]; + + snprintf(workerpath, sizeof(workerpath) - 1, "%s/xcecd_worker", mydir); + if (access(workerpath, X_OK) == 0) { + /* setup XCE_SERVER_DIR */ + XCECD_LOG(XCE_LOG_INFO, "XCE_SERVER_DIR set to `%s'", mydir); + setenv("XCE_SERVER_DIR", mydir, 0); + return 1; + } + } + + XCECD_LOG(XCE_LOG_INFO, "XCE_SERVER_DIR set to `/opt/xce/bin'"); + + setenv("XCE_SERVER_DIR", "/opt/xce/bin", 0); + return 1; +} + +static int check_user(void) +{ + if (geteuid() != 0) { + fprintf(stderr, "FATAL: xcecd should be run as root"); + exit(-1); + } + + return 1; +} + int main(int argc, char* argv[]) { int opt; @@ -225,6 +266,8 @@ int main(int argc, char* argv[]) char* custom_config_path = NULL; __attribute__((unused)) char worker_log[1024] = { 0 }; + check_user(); + signal(SIGPIPE, SIG_IGN); struct sigaction sa; @@ -392,6 +435,8 @@ int main(int argc, char* argv[]) } config.log_file = logfile; + check_xce_server_dir(argv[0]); + /* array to store exit worker pids */ init_static_array(&pids, sizeof(int)); diff --git a/src/worker/src/process_manager.c b/src/worker/src/process_manager.c index 59a2c14..9d5f6e0 100644 --- a/src/worker/src/process_manager.c +++ b/src/worker/src/process_manager.c @@ -20,6 +20,8 @@ #include #endif +#include "utils.h" + #define MAX_PROCESSES 1024 #define BUFFER_SIZE 4096 @@ -136,8 +138,11 @@ int process_manager_start_process(process_manager_t* manager, time(&now); struct tm* timeinfo = localtime(&now); if (jobid != 0) { - snprintf(log_path, sizeof(log_path), "/tmp/job_%lu_%lu.out", - (unsigned long)now, jobid); + char idstring[128]; + + fmt_jobid(idstring, sizeof(idstring) - 1, jobid); + snprintf(log_path, sizeof(log_path), "/tmp/job_%lu_%s.out", + (unsigned long)now, idstring); XCECD_WORKER_LOG(XCE_LOG_INFO, "Job output file path is: %s", log_path); } else { snprintf(log_path, sizeof(log_path), @@ -606,4 +611,4 @@ void process_manager_free(process_manager_t* manager) FREEUP(manager->processes); FREEUP(manager); XCECD_LOG(XCE_LOG_DEBUG, "Process manager freed successfully"); -} \ No newline at end of file +} diff --git a/src/worker/src/uv_proc.c b/src/worker/src/uv_proc.c index fc2e155..be3a607 100644 --- a/src/worker/src/uv_proc.c +++ b/src/worker/src/uv_proc.c @@ -1,5 +1,8 @@ #include "uv_proc.h" +#include "utils.h" +#include "xcecd_utils.h" #include "xcelog.h" + #include #include #include @@ -8,6 +11,23 @@ #include #include +static int make_job_output_path(char* buf, size_t bufsize, const char* dir, + uint64_t jobid, int task) +{ + char idstring[128]; + + if (buf == NULL) { + return -EINVAL; + } + + if (dir == NULL) { + dir = "/tmp"; + } + + fmt_jobid(idstring, sizeof(idstring) - 1, jobid); + return snprintf(buf, bufsize, "%s/job_%s_task%d.out", dir, idstring, task); +} + static void on_process_exit(uv_process_t* process, int64_t exit_status, int term_signal) { @@ -241,8 +261,9 @@ int uv_proc_exec(uv_proc_manager_t* manager, const char* command, } char job_output[256]; - snprintf(job_output, sizeof(job_output), "/tmp/job_%lu_task%d.out", - manager->job_id, i); + + make_job_output_path(job_output, sizeof(job_output) - 1, "/tmp", + manager->job_id, i); uv_stdio_container_t stdio[3]; uv_fs_t file_req; @@ -491,4 +512,4 @@ int uv_proc_exec_single(uv_proc_manager_t* manager, const char* command, void* user_data) { return uv_proc_exec(manager, command, args, exit_cb, user_data, 1); -} \ No newline at end of file +} diff --git a/src/worker/src/xcecd.c b/src/worker/src/xcecd.c index e114ff0..20414c6 100644 --- a/src/worker/src/xcecd.c +++ b/src/worker/src/xcecd.c @@ -13,6 +13,7 @@ #include #include "process_worker_manager.h" +#include "utils.h" xcecd_context_t xcecd_ctx = { 0 }; hostattrs_t host_attrs = { 0 }; @@ -31,7 +32,7 @@ static int virtual_pid = 1000; void xcecd_process_command(void* arg) { pid_t pid = 0; - char unit_name[256]; + char unit_name[256]; fork_task_t* ctx = (fork_task_t*)arg; if (!ctx || !ctx->startinfo || !ctx->startinfo->cmdline) { @@ -126,9 +127,8 @@ void xcecd_process_command(void* arg) return; } - snprintf(unit_name, sizeof(unit_name), "xcecd-worker-%lu.service", - ctx->xce_jobid); - + make_job_sd_unit_name(unit_name, sizeof(unit_name) - 1, "xcecd-worker", + ctx->xce_jobid); /* get worker pid */ pid = dbus_get_worker_pid(unit_name); #else diff --git a/src/worker/src/xcecd_cgroups.c b/src/worker/src/xcecd_cgroups.c index e160bdb..55b7d20 100644 --- a/src/worker/src/xcecd_cgroups.c +++ b/src/worker/src/xcecd_cgroups.c @@ -27,10 +27,9 @@ char* start_systemd_job(const char* command, const res_limits_t* limits, char service_name[256]; // 生成唯一的unit名 - const char* prefix = unit_name_prefix ? unit_name_prefix : "xcecd-worker"; - snprintf(service_name, sizeof(service_name), "%s-%lu.service", prefix, - job_id); - + make_job_sd_unit_name(service_name, sizeof(service_name) - 1, + unit_name_prefix ? unit_name_prefix : "xcecd-worker", + job_id); // 连接到system bus int r = sd_bus_open_system(&bus); if (r < 0) { @@ -1225,4 +1224,4 @@ char* get_systemd_job_status_text(const char* unit_name) sd_bus_unref(bus); return status_text; -} \ No newline at end of file +} diff --git a/src/worker/src/xcecd_dbus_manager.c b/src/worker/src/xcecd_dbus_manager.c index 38078e2..4e0c028 100644 --- a/src/worker/src/xcecd_dbus_manager.c +++ b/src/worker/src/xcecd_dbus_manager.c @@ -3,6 +3,7 @@ * @brief This file contains the implementation of the D-Bus manager functions. */ #include "xcecd_dbus_manager.h" +#include "utils.h" #include "xcecd_utils.h" #include "xcelog.h" @@ -62,7 +63,8 @@ bool dbus_start_worker(const res_limits_t* limits, uint64_t job_id, } char myname[256]; - snprintf(myname, sizeof(myname), "xcecd-worker-%lu.service", job_id); + + make_job_sd_unit_name(myname, sizeof(myname) - 1, "xcecd-worker", job_id); /* checking if service name is exist before start service and it's a * negative case */ dbus_stop_worker(myname); @@ -207,8 +209,9 @@ int dbus_workers_update(xcecd_worker_t** worker_list, xcecd_job_t* job, } char unit_name[256]; - snprintf(unit_name, sizeof(unit_name), "xcecd-worker-%lu.service", - worker->job_info.jobid); + + make_job_sd_unit_name(unit_name, sizeof(unit_name) - 1, "xcecd-worker", + worker->job_info.jobid); if (!dbus_is_worker_active(unit_name)) { #ifndef VIRTUAL_CLUSTER @@ -350,8 +353,9 @@ bool dbus_worker_clean(xcecd_worker_t* worker, xcecd_job_t* job) } char unit_name[256]; - snprintf(unit_name, sizeof(unit_name), "xcecd-worker-%lu.service", - worker->job_info.jobid); + + make_job_sd_unit_name(unit_name, sizeof(unit_name) - 1, "xcecd-worker", + worker->job_info.jobid); if (dbus_is_worker_active(unit_name)) { stop_systemd_job(unit_name); @@ -380,9 +384,9 @@ void dbus_stop_all_workers(xcecd_worker_t** worker_list, sqlite3* db) xcecd_worker_t* worker = *worker_list; while (worker) { char unit_name[256]; - snprintf(unit_name, sizeof(unit_name), "xcecd-worker-%lu.service", - worker->job_info.jobid); + make_job_sd_unit_name(unit_name, sizeof(unit_name) - 1, "xcecd-worker", + worker->job_info.jobid); job.job_id = worker->job_info.jobid; job.job_pid = worker->worker_pid; job.job_status = JOB_EXITED; @@ -395,4 +399,4 @@ void dbus_stop_all_workers(xcecd_worker_t** worker_list, sqlite3* db) worker = worker->next; } -} \ No newline at end of file +} diff --git a/src/worker/src/xcecd_exdata.c b/src/worker/src/xcecd_exdata.c index b07d98b..da5b780 100644 --- a/src/worker/src/xcecd_exdata.c +++ b/src/worker/src/xcecd_exdata.c @@ -9,6 +9,9 @@ #include "xcelog.h" #include #include +#include + +#include "utils.h" /** * @brief Parses the JSON string, populates an environment variable linked list, @@ -279,8 +282,15 @@ bool parse_limits(const char* exdata, res_limits_t* limits) bool success = false; cJSON* limits_obj = cJSON_GetObjectItemCaseSensitive(root, "limits"); - if (!limits_obj || !cJSON_IsObject(limits_obj)) { - // limits not exist + if (limits_obj == NULL) { + success = true; + + memset(limits, 0x0, sizeof(*limits)); + goto cleanup; + } + + if (!cJSON_IsObject(limits_obj)) { + // invalidate limits goto cleanup; } @@ -404,7 +414,7 @@ void apply_env_vars(env_var_t** env_vars, uint64_t jobid) // Set some default XCE environment variables char jobid_str[32]; - snprintf(jobid_str, sizeof(jobid_str), "%lu", jobid); + snprintf(jobid_str, sizeof(jobid_str), "%lu", array_jobid(jobid)); if (setenv("XCE_JOBID", jobid_str, 1) != 0) { XCECD_WORKER_LOG(XCE_LOG_ERROR, "Error setting environment variable: %s", @@ -416,7 +426,7 @@ void apply_env_vars(env_var_t** env_vars, uint64_t jobid) } char jobindex_str[8]; - snprintf(jobindex_str, sizeof(jobindex_str), "%d", 0); + snprintf(jobindex_str, sizeof(jobindex_str), "%lu", array_idx(jobid)); if (setenv("XCE_JOBINDEX", jobindex_str, 1) != 0) { XCECD_WORKER_LOG(XCE_LOG_ERROR, "Error setting environment variable: %s", @@ -491,4 +501,4 @@ char* generate_exdata(const start_exdata_t* start_info) } return exdata_str; -} \ No newline at end of file +} diff --git a/src/worker/src/xcecd_worker.c b/src/worker/src/xcecd_worker.c index fe63ca5..e52f7ef 100644 --- a/src/worker/src/xcecd_worker.c +++ b/src/worker/src/xcecd_worker.c @@ -1168,8 +1168,8 @@ static void do_criu_ckpt(uint64_t job_id, int kill) XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Checkpoint directory created: %s", ckpt_dir); - char criu_cmd[MAXLINELEN]; - char info[MAXLINELEN]; + char criu_cmd[MAXLINELEN*2]; + char info[MAXLINELEN*2]; if (kill) { snprintf(criu_cmd, sizeof(criu_cmd), "cd %s; %s dump -t %d -vvv -o dump.log", ckpt_dir, path, diff --git a/src/worker/src/xcecd_worker_utils.c b/src/worker/src/xcecd_worker_utils.c index 27fa1eb..8b0d64f 100644 --- a/src/worker/src/xcecd_worker_utils.c +++ b/src/worker/src/xcecd_worker_utils.c @@ -1,4 +1,5 @@ #include "xcecd_worker_utils.h" +#include "utils.h" #include "uv_ipc.h" #include "xcecd_dbus_manager.h" #include "xcecd_utils.h" @@ -512,4 +513,17 @@ bool check_worker_sleep_time_elapsed(xcecd_worker_t* worker) return false; } } -#endif \ No newline at end of file +#endif + +int make_job_sd_unit_name(char* buf, size_t bufsize, const char* prefix, + uint64_t jobid) +{ + char idstring[128]; + + if (buf == NULL) { + return -EINVAL; + } + + fmt_jobid_ex(idstring, sizeof(idstring) - 1, jobid, '_', '_'); + return snprintf(buf, bufsize, "%s-%s.service", prefix, idstring); +}