From f4b1ba4cd77fa2fd4481ca9789edf4d7b2ac17b7 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 2 Mar 2026 12:14:30 +0800 Subject: [PATCH] save works --- src/controller/apiserver.c | 16 ++-- src/controller/cluster.c | 26 +++---- src/controller/job.c | 143 ++++++++++++++++------------------ src/controller/opget.c | 2 +- src/controller/queue.c | 31 ++++---- src/controller/sched.c | 16 ++-- src/controller/task.c | 4 +- src/controller/worker.c | 90 ++++++++++----------- src/include/jobj.h | 4 +- src/include/qobj.h | 4 +- src/include/wobj.h | 4 +- src/include/xce/types.h | 14 ++-- src/lib/jobj.c | 61 ++++++++------- src/lib/qobj.c | 73 ++++++++--------- src/lib/testsuite/test_jobj.c | 50 +++++++----- src/lib/testsuite/test_qobj.c | 20 ++++- src/lib/testsuite/test_wobj.c | 49 +++++++----- src/lib/wobj.c | 61 ++++++++------- src/meson.build | 9 +++ src/nixSupport/xce.nix | 6 ++ src/tools/appimage/entry.sh | 14 ++++ 21 files changed, 373 insertions(+), 324 deletions(-) create mode 100644 src/tools/appimage/entry.sh diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c index 44d4f2d..65dd8f0 100644 --- a/src/controller/apiserver.c +++ b/src/controller/apiserver.c @@ -256,7 +256,7 @@ static int apiserver_job_update_fun(int type, int situation, void* data, int rc; xce_notify_data_t xdata = { .event = NotifyData_data_updateJob, - .data = { .update_job = job->job } }; + .data = { .update_job = &(job->job) } }; rc = notify(c, c->apiservers[i].ep, &xdata); if (rc < 0) { @@ -351,11 +351,11 @@ static int apiserver_worker_change(int type, int situation, void* data, { int i; - TRACE("%s worker %s idx %d situation %d", __func__, w->host->name, w->idx, + TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx, situation); - collect_resval_with_idx(&(w->host->resvals), &(w->host->nresvals_to_sched), - &(w->host->resvals_to_sched)); + collect_resval_with_idx(&(w->host.resvals), &(w->host.nresvals_to_sched), + &(w->host.resvals_to_sched)); switch (situation) { case HOOK_AFTER_EVENT: @@ -383,8 +383,8 @@ static int apiserver_worker_change(int type, int situation, void* data, break; } - free_resval_with_idx_list(&(w->host->resvals_to_sched), - w->host->nresvals_to_sched); + free_resval_with_idx_list(&(w->host.resvals_to_sched), + w->host.nresvals_to_sched); return 0; } @@ -402,7 +402,7 @@ static int apiserver_worker_join(int type, int situation, void* data, return 0; } - rc = key_find(c->workermgr, (void *)name, (void **)&w); + rc = key_find(c->workermgr, (void*)name, (void**)&w); if (rc < 0) { return rc; } @@ -514,7 +514,7 @@ static int apiserver_queue_hook_fun(int type, int situation, int detail, int rc; xce_notify_data_t xdata = { .event = NotifyData_data_schedUpdateQueue, - .data = { .sched_update_queue = q->queue } }; + .data = { .sched_update_queue = &(q->queue) } }; rc = notify(c, c->apiservers[i].ep, &xdata); if (rc < 0) { diff --git a/src/controller/cluster.c b/src/controller/cluster.c index dfbad8c..aa0e232 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -515,26 +515,18 @@ int cluster_init(cluster_t* c) kvmgr_params_t resmgr_params = { .key_type = KVMGR_KEY_STRING, .dynamic = 1, - .nslots = c->myconf->cluster_config->resmgr->nhash_slots, - .delta = c->myconf->cluster_config->resmgr->delta, .datakiller = (destroy_fn_t)res_obj_killer }; kvmgr_params_t jobmgr_params = { .key_type = KVMGR_KEY_INT, .dynamic = 1, - .nslots = c->myconf->cluster_config->jobmgr->nhash_slots, - .delta = c->myconf->cluster_config->jobmgr->delta, .datakiller = (destroy_fn_t)job_obj_killer }; kvmgr_params_t workermgr_params = { .key_type = KVMGR_KEY_STRING, .dynamic = 1, - .nslots = c->myconf->cluster_config->workermgr->nhash_slots, - .delta = c->myconf->cluster_config->workermgr->delta, .datakiller = (destroy_fn_t)worker_obj_killer }; kvmgr_params_t qmgr_params = { .key_type = KVMGR_KEY_STRING, .dynamic = 0, - .nslots = c->myconf->cluster_config->qmgr->nhash_slots, - .delta = c->myconf->cluster_config->qmgr->delta, .datakiller = (destroy_fn_t)queue_obj_killer }; if (c == NULL) { @@ -565,6 +557,15 @@ int cluster_init(cluster_t* c) return -errno; } + resmgr_params.nslots = c->myconf->cluster_config->resmgr->nhash_slots; + resmgr_params.delta = c->myconf->cluster_config->resmgr->delta; + jobmgr_params.nslots = c->myconf->cluster_config->jobmgr->nhash_slots; + jobmgr_params.delta = c->myconf->cluster_config->jobmgr->delta; + workermgr_params.nslots = c->myconf->cluster_config->workermgr->nhash_slots; + workermgr_params.delta = c->myconf->cluster_config->workermgr->delta; + qmgr_params.nslots = c->myconf->cluster_config->qmgr->nhash_slots; + qmgr_params.delta = c->myconf->cluster_config->qmgr->delta; + c->configmgr = makeInstance(&xkvmgr_type, &config_mgr_params); objmgr_init(&(c->objmgr)); @@ -644,13 +645,13 @@ static int check_worker_timeout_(cluster_t* c, worker_t* w) if (now - ep->ts_last_in > timeout) { TRACE("%s worker %s becomes UNAVAIL as heartbeat timeout", __func__, - w->host->name); + w->host.name); mark_worker_unavail(c, w); return 0; } else { - if (IS_WORKER_UNAVAIL(w->host->status)) { + if (IS_WORKER_UNAVAIL(w->host.status)) { mark_worker_back(c, w); } } @@ -659,13 +660,12 @@ static int check_worker_timeout_(cluster_t* c, worker_t* w) if (DATAQ_EMPTY(&(ep->outq))) { int rc = 0; - TRACE("%s send heartbeat to %s", __func__, w->host->name); + TRACE("%s send heartbeat to %s", __func__, w->host.name); rc = make_heartbeat(c, ep); if (rc < 0) { - ERROR("%s make_heartbeat to %s failed", __func__, - w->host->name); + ERROR("%s make_heartbeat to %s failed", __func__, w->host.name); return rc; } diff --git a/src/controller/job.c b/src/controller/job.c index 35bcf7c..b4826ec 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -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); diff --git a/src/controller/opget.c b/src/controller/opget.c index 1408abc..0cb9f2a 100644 --- a/src/controller/opget.c +++ b/src/controller/opget.c @@ -157,7 +157,7 @@ static int do_reply_joblist(cluster_t* c, endpoint_t* ep, if (rc > 0) { jobdata.jobs = (xce_job_t**)calloc(1, sizeof(xce_job_t*)); if (jobdata.jobs != NULL) { - jobdata.jobs[0] = j->job; + jobdata.jobs[0] = &(j->job); jobdata.n_total = 1; } } diff --git a/src/controller/queue.c b/src/controller/queue.c index 7021484..edfd000 100644 --- a/src/controller/queue.c +++ b/src/controller/queue.c @@ -19,12 +19,7 @@ queue_t* create_queue(cluster_t* c, xce_queue_t* p) { queue_t* q; - q = calloc(1, sizeof(*q)); - if (q == NULL) { - return NULL; - } - - q->queue = makeQueue(p); + q = makeQueue(c->qmgr, p); return q; } @@ -72,28 +67,28 @@ int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl) switch (ctrl->action) { case QUEUE_OPEN: - if (!(q->queue->status & QUEUE_STAT_CLOSED)) { + if (!(q->queue.status & QUEUE_STAT_CLOSED)) { goto clean; } - q->queue->status &= ~QUEUE_STAT_CLOSED; + q->queue.status &= ~QUEUE_STAT_CLOSED; break; case QUEUE_CLOSE: - if (q->queue->status & QUEUE_STAT_CLOSED) { + if (q->queue.status & QUEUE_STAT_CLOSED) { goto clean; } - q->queue->status |= QUEUE_STAT_CLOSED; + q->queue.status |= QUEUE_STAT_CLOSED; break; case QUEUE_INACTIVE: - if (q->queue->status & QUEUE_STAT_INACT_ADMIN) { + if (q->queue.status & QUEUE_STAT_INACT_ADMIN) { goto clean; } - q->queue->status |= QUEUE_STAT_INACT_ADMIN; + q->queue.status |= QUEUE_STAT_INACT_ADMIN; break; case QUEUE_ACTIVE: - if (!(q->queue->status & QUEUE_STAT_INACT_ADMIN)) { + if (!(q->queue.status & QUEUE_STAT_INACT_ADMIN)) { goto clean; } - q->queue->status &= ~QUEUE_STAT_INACT_ADMIN; + q->queue.status &= ~QUEUE_STAT_INACT_ADMIN; break; default: rc = -1; @@ -147,14 +142,14 @@ int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec) return -ENOENT; } - if (q->queue->pre_exec != NULL) { - *pre_exec = makeString(q->queue->pre_exec); + if (q->queue.pre_exec != NULL) { + *pre_exec = makeString(q->queue.pre_exec); } else { *pre_exec = NULL; } - if (q->queue->post_exec != NULL) { - *post_exec = makeString(q->queue->post_exec); + if (q->queue.post_exec != NULL) { + *post_exec = makeString(q->queue.post_exec); } else { *post_exec = NULL; } diff --git a/src/controller/sched.c b/src/controller/sched.c index 9a77c01..346af2a 100644 --- a/src/controller/sched.c +++ b/src/controller/sched.c @@ -341,7 +341,7 @@ static int sched_job_update_fun(int type, int situation, void* data, va_list ap) xce_notify_data_t xdata = { .event = NotifyData_data_schedUpdateJob, .data = { - .sched_update_job = job->job + .sched_update_job = &(job->job) } }; /* clang-format on */ @@ -419,11 +419,11 @@ static int sched_worker_change(int type, int situation, void* data, int rc = 0; xlist_t wlist; - TRACE("%s worker %s idx %d situation %d", __func__, w->host->name, w->idx, + TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx, situation); - collect_resval_with_idx(&(w->host->resvals), &(w->host->nresvals_to_sched), - &(w->host->resvals_to_sched)); + collect_resval_with_idx(&(w->host.resvals), &(w->host.nresvals_to_sched), + &(w->host.resvals_to_sched)); mylock_lock(&(c->sched.lock)); if (c->sched.ep != NULL) { @@ -440,8 +440,8 @@ static int sched_worker_change(int type, int situation, void* data, } mylock_unlock(&(c->sched.lock)); - free_resval_with_idx_list(&(w->host->resvals_to_sched), - w->host->nresvals_to_sched); + free_resval_with_idx_list(&(w->host.resvals_to_sched), + w->host.nresvals_to_sched); return rc; } @@ -458,7 +458,7 @@ static int sched_worker_join(int type, int situation, void* data, va_list ap) return 0; } - rc = key_find(c->workermgr, (void *)name, (void **)&w); + rc = key_find(c->workermgr, (void*)name, (void**)&w); if (rc < 0) { return rc; } @@ -558,7 +558,7 @@ static int sched_queue_update(int type, xce_notify_data_t xdata = { .event = NotifyData_data_schedUpdateQueue, .data = { - .sched_update_queue = q->queue + .sched_update_queue = &(q->queue) } }; /* clang-format on */ diff --git a/src/controller/task.c b/src/controller/task.c index 8b78234..84744ab 100644 --- a/src/controller/task.c +++ b/src/controller/task.c @@ -26,7 +26,7 @@ static int run_tasks_on_host(cluster_t* c, endpoint_t* ep, uint64_t jobid, return -ENOENT; } - if ((w->ep == NULL) || (IS_WORKER_UNAVAIL(w->host->status))) { + if ((w->ep == NULL) || (IS_WORKER_UNAVAIL(w->host.status))) { obj_deref(w); return -EINVAL; } @@ -59,7 +59,7 @@ static int check_task_hosts(cluster_t* c, xce_taskrun_t* runmsg) return -ENOENT; } - if ((w->ep == NULL) || IS_WORKER_UNAVAIL(w->host->status)) { + if ((w->ep == NULL) || IS_WORKER_UNAVAIL(w->host.status)) { return -ENOENT; } } diff --git a/src/controller/worker.c b/src/controller/worker.c index 5c01557..fd7c8f6 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -70,7 +70,7 @@ static int update_val_res(cluster_t* c, worker_t* w, xce_resource_t* r, array = &(c->sharedinst); } else { /* update host based resource */ - array = &(w->host->resvals); + array = &(w->host.resvals); } if (idx < array->caps) { @@ -213,17 +213,16 @@ static int add_hostattrs_resvals(cluster_t* c, worker_t* w) { #define N_HOST_ATTRS (8) xce_val_t values[N_HOST_ATTRS] = { + { .which = Value_u_svalue, .u = { .svalue = w->host.attrs->hosttype } }, { .which = Value_u_svalue, - .u = { .svalue = w->host->attrs->hosttype } }, - { .which = Value_u_svalue, - .u = { .svalue = w->host->attrs->hostmodel } }, + .u = { .svalue = w->host.attrs->hostmodel } }, { .which = Value_u_fvalue, - .u = { .fvalue = w->host->attrs->cpufactor } }, - { .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->maxcpus } }, - { .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->maxmem } }, - { .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->maxswap } }, - { .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->maxtmp } }, - { .which = Value_u_fvalue, .u = { .fvalue = w->host->attrs->ndisks } } + .u = { .fvalue = w->host.attrs->cpufactor } }, + { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxcpus } }, + { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxmem } }, + { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxswap } }, + { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxtmp } }, + { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->ndisks } } }; xce_resval_t resvals[N_HOST_ATTRS] = { { .name = "type", .val = &values[0] }, @@ -266,7 +265,7 @@ int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs) cluster_t* c = (cluster_t*)cls; /* clang-format off */ worker_t worker = { - .host = &(xce_worker_t){ + .host = (xce_worker_t){ .name = (char*)host, } }; @@ -290,28 +289,28 @@ int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs) return -errno; } - worker.host->attrs = attrs; + worker.host.attrs = attrs; - w->host = makeWorker(worker.host); - if (w->host == NULL) { + w = makeWorker(c->workermgr, &(worker.host)); + if (w == NULL) { return -errno; } /* we should hold the update lock */ /* If admin makes the host in locked or closed, we need keep this */ - w->host->status &= ~XCE_WORKER_UNAVAIL; /* status OK */ + w->host.status &= ~XCE_WORKER_UNAVAIL; /* status OK */ w->ep = ep; /* override host type? */ if (strcmp(hconfig->hosttype, "!") != 0) { - dereference(w->host->attrs->hosttype); - w->host->attrs->hosttype = makeString(hconfig->hosttype); + dereference(w->host.attrs->hosttype); + w->host.attrs->hosttype = makeString(hconfig->hosttype); } /* override host model */ if (strcmp(hconfig->hostmodel, "!") != 0) { - dereference(w->host->attrs->hostmodel); - w->host->attrs->hostmodel = makeString(hconfig->hostmodel); + dereference(w->host.attrs->hostmodel); + w->host.attrs->hostmodel = makeString(hconfig->hostmodel); } add_hostattrs_resvals(c, w); idx = w->idx; @@ -319,31 +318,31 @@ int add_worker(void* cls, endpoint_t* ep, const char* host, hostattrs_t* attrs) /* using auto detected or pre configured MXJ? */ nslots = hconfig->mxj; if (nslots < 0) { - nslots = w->host->attrs->maxcpus; + nslots = w->host.attrs->maxcpus; } set_host_slots(c, w, nslots); - if (w->host->dispatch_window_config != NULL) { - free_timewindow_config(w->host->dispatch_window_config); - w->host->dispatch_window_config = NULL; + if (w->host.dispatch_window_config != NULL) { + free_timewindow_config(w->host.dispatch_window_config); + w->host.dispatch_window_config = NULL; } if (hconfig->dispatchwindow != NULL) { - w->host->dispatch_window_config + w->host.dispatch_window_config = dup_timewindow_config(hconfig->dispatchwindow); } - if (w->host->thresholds != NULL) { - free_resource_thresholds(w->host->thresholds, w->host->nthresholds); - w->host->nthresholds = 0; - w->host->thresholds = NULL; + if (w->host.thresholds != NULL) { + free_resource_thresholds(w->host.thresholds, w->host.nthresholds); + w->host.nthresholds = 0; + w->host.thresholds = NULL; } if (hconfig->thresholds != NULL) { - w->host->thresholds = dup_resource_thresholds(hconfig->thresholds, - hconfig->nthresholds); - w->host->nthresholds = hconfig->nthresholds; + w->host.thresholds = dup_resource_thresholds(hconfig->thresholds, + hconfig->nthresholds); + w->host.nthresholds = hconfig->nthresholds; } /* add bool resources to host */ @@ -361,27 +360,18 @@ int restore_worker(void* cls, xce_worker_t host) cluster_t* c = (cluster_t*)cls; int idx = 0; worker_t* w = NULL; - /* do work in add_worker() doing for normal worker join */ - /* clang-format off */ - worker_t worker = { - .host = &(xce_worker_t){ - .name = (char*)host.name, - .status = host.status, - }, - }; - /* clang-format on */ w = calloc(1, sizeof(*w)); if (w == NULL) { return -ENOMEM; } - w->host = makeWorker(worker.host); - if (w->host == NULL) { + w = makeWorker(c->workermgr, &host); + if (w == NULL) { return -errno; } add_hostattrs_resvals(c, w); idx = w->idx; - w->host->status |= XCE_WORKER_UNAVAIL; + w->host.status |= XCE_WORKER_UNAVAIL; DEREF_OBJ(w); @@ -476,7 +466,7 @@ int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val) return -ENOENT; } - array = &(w->host->resvals); + array = &(w->host.resvals); } else { array = &(c->sharedinst); } @@ -497,7 +487,7 @@ int mark_worker_unavail(cluster_t* c, worker_t* w) { xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT, HOOK_DETAIL_WORKER_DEAD, c, w); - w->host->status |= XCE_WORKER_UNAVAIL; + w->host.status |= XCE_WORKER_UNAVAIL; xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_DEAD, c, w); @@ -508,7 +498,7 @@ int mark_worker_back(cluster_t* c, worker_t* w) { xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w); - w->host->status &= ~XCE_WORKER_UNAVAIL; + w->host.status &= ~XCE_WORKER_UNAVAIL; xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w); @@ -574,16 +564,16 @@ int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl) switch (ctrl->action) { case HOST_OPEN: - if (!(w->host->status & XCE_WORKER_CLOSED)) { + if (!(w->host.status & XCE_WORKER_CLOSED)) { goto clean; } - w->host->status &= ~XCE_WORKER_CLOSED; + w->host.status &= ~XCE_WORKER_CLOSED; break; case HOST_CLOSE: - if (w->host->status & XCE_WORKER_CLOSED) { + if (w->host.status & XCE_WORKER_CLOSED) { goto clean; } - w->host->status |= XCE_WORKER_CLOSED; + w->host.status |= XCE_WORKER_CLOSED; break; default: rc = -1; diff --git a/src/include/jobj.h b/src/include/jobj.h index f4280c4..452bbf8 100644 --- a/src/include/jobj.h +++ b/src/include/jobj.h @@ -2,11 +2,13 @@ #define _JOBJ_H_ 1 +#include + #include "xce/types.h" extern job_submit_data_t* makeJobsub(job_submit_data_t*); extern job_startinfo_t* makeJobstartinfo(job_startinfo_t*); -extern xce_job_t* makeJob(xce_job_t*); +extern job_t* makeJob(var, xce_job_t*); extern int job_obj_init(int, int); extern void job_obj_fini(); diff --git a/src/include/qobj.h b/src/include/qobj.h index df69a1f..f3c8366 100644 --- a/src/include/qobj.h +++ b/src/include/qobj.h @@ -2,8 +2,10 @@ #define _QOBJ_H_ 1 +#include + #include "xce/types.h" -extern xce_queue_t *makeQueue(xce_queue_t*); +extern queue_t* makeQueue(var, xce_queue_t*); #endif diff --git a/src/include/wobj.h b/src/include/wobj.h index af72459..106be53 100644 --- a/src/include/wobj.h +++ b/src/include/wobj.h @@ -2,8 +2,10 @@ #define _WOBJ_H_ 1 +#include + #include "xce/types.h" -extern xce_worker_t* makeWorker(xce_worker_t*); +extern worker_t* makeWorker(var, xce_worker_t*); #endif diff --git a/src/include/xce/types.h b/src/include/xce/types.h index a853d64..8611a9e 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -274,11 +274,11 @@ typedef struct { } xce_res_account_t; typedef struct { - xce_worker_t* host; - char hostkey[XCE_MAXKEY_LEN]; - endpoint_t* ep; - int idx; - zarray_t usages; + xce_worker_t host; + char hostkey[XCE_MAXKEY_LEN]; + endpoint_t* ep; + int idx; + zarray_t usages; } worker_t; typedef struct { @@ -470,7 +470,7 @@ typedef struct { } xce_job_t; typedef struct { - xce_job_t* job; + xce_job_t job; } job_t; typedef struct { @@ -517,7 +517,7 @@ typedef struct { } xce_queue_t; typedef struct { - xce_queue_t* queue; + xce_queue_t queue; } queue_t; typedef struct { diff --git a/src/lib/jobj.c b/src/lib/jobj.c index 621a89e..dc38b47 100644 --- a/src/lib/jobj.c +++ b/src/lib/jobj.c @@ -537,14 +537,14 @@ typedef struct { static size_t job_datasize(va_list ap) { - size_t sz = sizeof(xce_job_t); + size_t sz = sizeof(job_t); return sz; } static int job_init(meta_obj_t* obj, va_list ap) { - xce_job_t* dst = RTTI_OBJ_TO_DATA(xce_job_t, obj); + job_t* dst = RTTI_OBJ_TO_DATA(job_t, obj); job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); xce_job_t* src = va_arg(ap, xce_job_t*); @@ -552,22 +552,22 @@ static int job_init(meta_obj_t* obj, va_list ap) return -EINVAL; } - dst->jobid = src->jobid; - dst->submit = makeJobsub(src->submit); - dst->sub_time = src->sub_time; - dst->queue_time = src->queue_time; - dst->start_time = src->start_time; - dst->finish_time = src->finish_time; - MOVE_PTR(dst->pidinfo, src->pidinfo); - dst->state = src->state; - MOVE_PTR(dst->allocs, src->allocs); - MOVE_PTR(dst->accounts, src->accounts); - dst->run_time = src->run_time; - MOVE_PTR(dst->run_usage, src->run_usage); - MOVE_PTR(dst->exit_info, src->exit_info); - dst->ckptinfo = makeString(src->ckptinfo); - dst->pending_ctrl = src->pending_ctrl; - MOVE_PTR(dst->pend_reasons, src->pend_reasons); + dst->job.jobid = src->jobid; + dst->job.submit = makeJobsub(src->submit); + dst->job.sub_time = src->sub_time; + dst->job.queue_time = src->queue_time; + dst->job.start_time = src->start_time; + dst->job.finish_time = src->finish_time; + MOVE_PTR(dst->job.pidinfo, src->pidinfo); + dst->job.state = src->state; + MOVE_PTR(dst->job.allocs, src->allocs); + MOVE_PTR(dst->job.accounts, src->accounts); + dst->job.run_time = src->run_time; + MOVE_PTR(dst->job.run_usage, src->run_usage); + MOVE_PTR(dst->job.exit_info, src->exit_info); + dst->job.ckptinfo = makeString(src->ckptinfo); + dst->job.pending_ctrl = src->pending_ctrl; + MOVE_PTR(dst->job.pend_reasons, src->pend_reasons); my_rwlock_init(&(priv->lock)); priv->refcount = 1; @@ -577,24 +577,24 @@ static int job_init(meta_obj_t* obj, va_list ap) static void job_destroy(meta_obj_t* obj) { - xce_job_t* j = RTTI_OBJ_TO_DATA(xce_job_t, obj); + job_t* j = RTTI_OBJ_TO_DATA(job_t, obj); job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); - dereference(j->submit); - dereference(j->ckptinfo); + dereference(j->job.submit); + dereference(j->job.ckptinfo); /* FIXME: */ - if (j->pidinfo != NULL) { } + if (j->job.pidinfo != NULL) { } - if (j->allocs != NULL) { } + if (j->job.allocs != NULL) { } - if (j->accounts != NULL) { } + if (j->job.accounts != NULL) { } - if (j->run_usage != NULL) { } + if (j->job.run_usage != NULL) { } - if (j->exit_info != NULL) { } + if (j->job.exit_info != NULL) { } - if (j->pend_reasons != NULL) { } + if (j->job.pend_reasons != NULL) { } my_rwlock_fini(&(priv->lock)); memset(priv, 0x0, sizeof(*priv)); @@ -616,7 +616,7 @@ static int job_reference(meta_obj_t* obj) static void job_dereference(meta_obj_t* obj) { job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); - xce_job_t* j = RTTI_OBJ_TO_DATA(xce_job_t, obj); + job_t* j = RTTI_OBJ_TO_DATA(job_t, obj); uint64_t count; my_rwlock_wlock(&(priv->lock)); @@ -674,9 +674,10 @@ static typeinfo_t job_type = { .unlock = (unlock_fn_t)job_unlock } } }; -xce_job_t* makeJob(xce_job_t* j) +job_t* makeJob(var jobmgr, xce_job_t* j) { - xce_job_t* job = makeInstance(&job_type, j); + job_t* job = makeInstance(&job_type, j); + /* FIXME */ return job; } diff --git a/src/lib/qobj.c b/src/lib/qobj.c index 79b9dc0..cad7a69 100644 --- a/src/lib/qobj.c +++ b/src/lib/qobj.c @@ -16,14 +16,14 @@ typedef struct { static size_t queue_datasize(va_list ap) { - size_t sz = sizeof(xce_queue_t); + size_t sz = sizeof(queue_t); return sz; } static int queue_init(meta_obj_t* obj, va_list ap) { - xce_queue_t* dst = RTTI_OBJ_TO_DATA(xce_queue_t, obj); + queue_t* dst = RTTI_OBJ_TO_DATA(queue_t, obj); queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj); xce_queue_t* src = va_arg(ap, xce_queue_t*); @@ -31,26 +31,26 @@ static int queue_init(meta_obj_t* obj, va_list ap) return -EINVAL; } - dst->queue = makeString(src->queue); - dst->attrs = src->attrs; - dst->status = src->status; - dst->priority = src->priority; - dst->accept_interval = src->accept_interval; - dst->sched_delay = src->sched_delay; - dst->resreq = makeString(src->resreq); - MOVE_PTR(dst->users, src->users); - MOVE_PTR(dst->hosts, src->hosts); - dst->nthresholds = src->nthresholds; - MOVE_PTR(dst->thresholds, src->thresholds); + dst->queue.queue = makeString(src->queue); + dst->queue.attrs = src->attrs; + dst->queue.status = src->status; + dst->queue.priority = src->priority; + dst->queue.accept_interval = src->accept_interval; + dst->queue.sched_delay = src->sched_delay; + dst->queue.resreq = makeString(src->resreq); + MOVE_PTR(dst->queue.users, src->users); + MOVE_PTR(dst->queue.hosts, src->hosts); + dst->queue.nthresholds = src->nthresholds; + MOVE_PTR(dst->queue.thresholds, src->thresholds); - MOVE_PTR(dst->runwindow, src->runwindow); - MOVE_PTR(dst->dispatchwindow, src->dispatchwindow); - MOVE_PTR(dst->runwindow_config, src->runwindow_config); - MOVE_PTR(dst->dispatchwindow_config, src->dispatchwindow_config); - dst->pending_ctrl = src->pending_ctrl; - dst->pre_exec = makeString(src->pre_exec); - dst->post_exec = makeString(src->post_exec); - dst->autopscaler = makeString(src->autopscaler); + MOVE_PTR(dst->queue.runwindow, src->runwindow); + MOVE_PTR(dst->queue.dispatchwindow, src->dispatchwindow); + MOVE_PTR(dst->queue.runwindow_config, src->runwindow_config); + MOVE_PTR(dst->queue.dispatchwindow_config, src->dispatchwindow_config); + dst->queue.pending_ctrl = src->pending_ctrl; + dst->queue.pre_exec = makeString(src->pre_exec); + dst->queue.post_exec = makeString(src->post_exec); + dst->queue.autopscaler = makeString(src->autopscaler); my_rwlock_init(&(priv->lock)); priv->refcount = 1; @@ -60,30 +60,30 @@ static int queue_init(meta_obj_t* obj, va_list ap) static void queue_destroy(meta_obj_t* obj) { - xce_queue_t* q = RTTI_OBJ_TO_DATA(xce_queue_t, obj); + queue_t* q = RTTI_OBJ_TO_DATA(queue_t, obj); queue_private_t* priv = (queue_private_t*)RTTI_OBJ_TO_PRIV(obj); - dereference(q->queue); - dereference(q->resreq); + dereference(q->queue.queue); + dereference(q->queue.resreq); /* FIXME: */ - if (q->users != NULL) { } + if (q->queue.users != NULL) { } - if (q->hosts != NULL) { } + if (q->queue.hosts != NULL) { } - if (q->thresholds != NULL) { } + if (q->queue.thresholds != NULL) { } - if (q->runwindow != NULL) { } + if (q->queue.runwindow != NULL) { } - if (q->dispatchwindow != NULL) { } + if (q->queue.dispatchwindow != NULL) { } - if (q->runwindow_config != NULL) { } + if (q->queue.runwindow_config != NULL) { } - if (q->dispatchwindow_config != NULL) { } + if (q->queue.dispatchwindow_config != NULL) { } - dereference(q->pre_exec); - dereference(q->post_exec); - dereference(q->autopscaler); + dereference(q->queue.pre_exec); + dereference(q->queue.post_exec); + dereference(q->queue.autopscaler); my_rwlock_fini(&(priv->lock)); memset(priv, 0x0, sizeof(*priv)); @@ -163,8 +163,9 @@ static typeinfo_t queue_type = { .unlock = (unlock_fn_t)queue_unlock } } }; -xce_queue_t* makeQueue(xce_queue_t* w) +queue_t* makeQueue(var qmgr, xce_queue_t* w) { - xce_queue_t* queue = makeInstance(&queue_type, w); + /* FIMXE: */ + queue_t* queue = makeInstance(&queue_type, w); return queue; } diff --git a/src/lib/testsuite/test_jobj.c b/src/lib/testsuite/test_jobj.c index 79ebbf2..a50f195 100644 --- a/src/lib/testsuite/test_jobj.c +++ b/src/lib/testsuite/test_jobj.c @@ -6,6 +6,7 @@ #include #include "jobj.h" +#include "kvmgr.h" #include "utils.h" #include "xstring.h" @@ -206,6 +207,12 @@ static void test_jobsub_share(void) static void test_job_create(void) { int rc; + kvmgr_params_t jobmgr_params = { .key_type = KVMGR_KEY_INT, + .dynamic = 1, + .nslots = 32, + .delta = 32, + .datakiller = destroy }; + var jobmgr; job_startinfo_t s1 = { .exechome = "/homes/rshen", .execcwd = "/tmp", .execuser = "rshen", @@ -226,7 +233,7 @@ static void test_job_create(void) .job_description = "no description", .resreq = "type == any" }; xce_job_t job1 = { .submit = &jobsub1 }; - xce_job_t* j2 = NULL; + job_t* j2 = NULL; time_t now = time(NULL); rc = xstring_table_init(32); @@ -249,31 +256,36 @@ static void test_job_create(void) job1.pending_ctrl = 0; job1.jobid = 100; - j2 = makeJob(&job1); + jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params); + TEST_ASSERT_TRUE(jobmgr != NULL); + + j2 = makeJob(jobmgr, &job1); TEST_ASSERT_TRUE(j2 != NULL); TEST_ASSERT_TRUE( - EQUAL_STRING(s1.exechome, j2->submit->start_info->exechome) - && EQUAL_STRING(s1.execcwd, j2->submit->start_info->execcwd) - && EQUAL_STRING(s1.execuser, j2->submit->start_info->execuser) - && EQUAL_STRING(s1.cmdline, j2->submit->start_info->cmdline) - && EQUAL_STRING(s1.exdata, j2->submit->start_info->exdata) - && EQUAL_STRING(s1.q_pre_exec, j2->submit->start_info->q_pre_exec) - && EQUAL_STRING(s1.q_post_exec, j2->submit->start_info->q_post_exec) - && EQUAL_STRING(s1.job_pre_exec, j2->submit->start_info->job_pre_exec) - && EQUAL_STRING(s1.job_post_exec, j2->submit->start_info->job_post_exec) - && EQUAL_STRING(j2->submit->queue, "xce") - && EQUAL_STRING(j2->submit->user, "rshen") - && EQUAL_STRING(j2->submit->job_name, "J1") - && EQUAL_STRING(j2->submit->project_name, "default") - && EQUAL_STRING(j2->submit->job_description, "no description") - && EQUAL_STRING(j2->submit->resreq, "type == any") - && (j2->submit->mintasks == 1) && (j2->submit->maxtasks == 5) - && (j2->submit->user_group == NULL)); + EQUAL_STRING(s1.exechome, j2->job.submit->start_info->exechome) + && EQUAL_STRING(s1.execcwd, j2->job.submit->start_info->execcwd) + && EQUAL_STRING(s1.execuser, j2->job.submit->start_info->execuser) + && EQUAL_STRING(s1.cmdline, j2->job.submit->start_info->cmdline) + && EQUAL_STRING(s1.exdata, j2->job.submit->start_info->exdata) + && EQUAL_STRING(s1.q_pre_exec, j2->job.submit->start_info->q_pre_exec) + && EQUAL_STRING(s1.q_post_exec, j2->job.submit->start_info->q_post_exec) + && EQUAL_STRING(s1.job_pre_exec, j2->job.submit->start_info->job_pre_exec) + && EQUAL_STRING(s1.job_post_exec, j2->job.submit->start_info->job_post_exec) + && EQUAL_STRING(j2->job.submit->queue, "xce") + && EQUAL_STRING(j2->job.submit->user, "rshen") + && EQUAL_STRING(j2->job.submit->job_name, "J1") + && EQUAL_STRING(j2->job.submit->project_name, "default") + && EQUAL_STRING(j2->job.submit->job_description, "no description") + && EQUAL_STRING(j2->job.submit->resreq, "type == any") + && (j2->job.submit->mintasks == 1) && (j2->job.submit->maxtasks == 5) + && (j2->job.submit->user_group == NULL)); dereference(j2); + destroy(jobmgr); + job_obj_fini(); xstring_table_fini(); } diff --git a/src/lib/testsuite/test_qobj.c b/src/lib/testsuite/test_qobj.c index bf342a0..9c76032 100644 --- a/src/lib/testsuite/test_qobj.c +++ b/src/lib/testsuite/test_qobj.c @@ -2,28 +2,40 @@ #include #include #include +#include #include #include "qobj.h" #include "utils.h" #include "xstring.h" +#include "kvmgr.h" static void test_queue_create(void) { - int rc; - xce_queue_t q1 = { 0 }; - xce_queue_t* q = NULL; + int rc; + kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING, + .nslots = 32, + .delta = 32, + .datakiller = destroy }; + var qmgr; + + xce_queue_t q1 = { 0 }; + queue_t* q = NULL; rc = xstring_table_init(32); TEST_ASSERT_TRUE(rc >= 0); - q = makeQueue(&q1); + qmgr = makeInstance(&xkvmgr_type, ¶ms); + TEST_ASSERT_TRUE(qmgr != NULL); + + q = makeQueue(qmgr, &q1); TEST_ASSERT_TRUE(q != NULL); destroy(q); + destroy(qmgr); xstring_table_fini(); } diff --git a/src/lib/testsuite/test_wobj.c b/src/lib/testsuite/test_wobj.c index eadb521..1bc5130 100644 --- a/src/lib/testsuite/test_wobj.c +++ b/src/lib/testsuite/test_wobj.c @@ -5,41 +5,52 @@ #include +#include "kvmgr.h" #include "utils.h" #include "wobj.h" #include "xstring.h" static void test_worker_create(void) { - int rc; - hostattrs_t attrs1 = { .hosttype = "LINUX64", - .hostmodel = "linux-gnu-x86_64", - .cpufactor = 100.0, - .maxcpus = 5, - .maxmem = 65536.0, - .maxswap = 65536.0, - .maxtmp = 0.0, - .ndisks = 2 }; - xce_worker_t w1 = { - .name = "worker1", - .attrs = &attrs1, + int rc; + kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING, + .nslots = 32, + .delta = 32, + .datakiller = destroy }; + var workermgr; + hostattrs_t attrs1 = { .hosttype = "LINUX64", + .hostmodel = "linux-gnu-x86_64", + .cpufactor = 100.0, + .maxcpus = 5, + .maxmem = 65536.0, + .maxswap = 65536.0, + .maxtmp = 0.0, + .ndisks = 2 }; + xce_worker_t w1 = { + .name = "worker1", + .attrs = &attrs1, }; - xce_worker_t* w = NULL; + worker_t* w = NULL; rc = xstring_table_init(32); TEST_ASSERT_TRUE(rc >= 0); - w = makeWorker(&w1); + workermgr = makeInstance(&xkvmgr_type, ¶ms); + TEST_ASSERT_TRUE(workermgr != NULL); + + w = makeWorker(workermgr, &w1); TEST_ASSERT_TRUE(w != NULL); - TEST_ASSERT_TRUE(EQUAL_STRING(w->name, w1.name) - && EQUAL_STRING(w->attrs->hosttype, attrs1.hosttype) - && EQUAL_STRING(w->attrs->hostmodel, attrs1.hostmodel) - && (w->attrs->cpufactor == attrs1.cpufactor) - && (w->attrs->maxcpus == attrs1.maxcpus)); + TEST_ASSERT_TRUE(EQUAL_STRING(w->host.name, w1.name) + && EQUAL_STRING(w->host.attrs->hosttype, attrs1.hosttype) + && EQUAL_STRING(w->host.attrs->hostmodel, attrs1.hostmodel) + && (w->host.attrs->cpufactor == attrs1.cpufactor) + && (w->host.attrs->maxcpus == attrs1.maxcpus)); dereference(w); + destroy(workermgr); + xstring_table_fini(); } diff --git a/src/lib/wobj.c b/src/lib/wobj.c index 4e62b06..16aeac1 100644 --- a/src/lib/wobj.c +++ b/src/lib/wobj.c @@ -16,14 +16,14 @@ typedef struct { static size_t worker_datasize(va_list ap) { - size_t sz = sizeof(xce_worker_t); + size_t sz = sizeof(worker_t); return sz; } static int worker_init(meta_obj_t* obj, va_list ap) { - xce_worker_t* dst = RTTI_OBJ_TO_DATA(xce_worker_t, obj); + worker_t* dst = RTTI_OBJ_TO_DATA(worker_t, obj); worker_private_t* priv = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj); xce_worker_t* src = va_arg(ap, xce_worker_t*); @@ -31,28 +31,28 @@ static int worker_init(meta_obj_t* obj, va_list ap) return -EINVAL; } - dst->name = makeString(src->name); - dst->attrs = (hostattrs_t*)calloc(1, sizeof(hostattrs_t)); - if (dst->attrs == NULL) { + dst->host.name = makeString(src->name); + dst->host.attrs = (hostattrs_t*)calloc(1, sizeof(hostattrs_t)); + if (dst->host.attrs == NULL) { return -ENOMEM; } - memcpy(dst->attrs, src->attrs, sizeof(*(dst->attrs))); + memcpy(dst->host.attrs, src->attrs, sizeof(*(dst->host.attrs))); - dst->attrs->hosttype = makeString(src->attrs->hosttype); - dst->attrs->hostmodel = makeString(src->attrs->hostmodel); - dst->status = src->status; + dst->host.attrs->hosttype = makeString(src->attrs->hosttype); + dst->host.attrs->hostmodel = makeString(src->attrs->hostmodel); + dst->host.status = src->status; /* FIXME: */ - zarray_init(&(dst->resvals), 64, NULL); + zarray_init(&(dst->host.resvals), 64, NULL); - dst->dead_since = src->dead_since; - dst->pending_ctrl = src->pending_ctrl; - MOVE_PTR(dst->dispatch_window_config, src->dispatch_window_config); - dst->nthresholds = src->nthresholds; - MOVE_PTR(dst->thresholds, src->thresholds); - dst->nresvals_to_sched = 0; - dst->resvals_to_sched = NULL; + dst->host.dead_since = src->dead_since; + dst->host.pending_ctrl = src->pending_ctrl; + MOVE_PTR(dst->host.dispatch_window_config, src->dispatch_window_config); + dst->host.nthresholds = src->nthresholds; + MOVE_PTR(dst->host.thresholds, src->thresholds); + dst->host.nresvals_to_sched = 0; + dst->host.resvals_to_sched = NULL; my_rwlock_init(&(priv->lock)); priv->refcount = 1; @@ -62,24 +62,24 @@ static int worker_init(meta_obj_t* obj, va_list ap) static void worker_destroy(meta_obj_t* obj) { - xce_worker_t* w = RTTI_OBJ_TO_DATA(xce_worker_t, obj); + worker_t* w = RTTI_OBJ_TO_DATA(worker_t, obj); worker_private_t* priv = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj); - dereference(w->name); + dereference(w->host.name); - if (w->attrs != NULL) { - dereference(w->attrs->hosttype); - dereference(w->attrs->hostmodel); - FREEUP(w->attrs); + if (w->host.attrs != NULL) { + dereference(w->host.attrs->hosttype); + dereference(w->host.attrs->hostmodel); + FREEUP(w->host.attrs); } - if (w->dispatch_window_config != NULL) { } + if (w->host.dispatch_window_config != NULL) { } - if (w->thresholds != NULL) { } + if (w->host.thresholds != NULL) { } - if (w->resvals_to_sched != NULL) { } + if (w->host.resvals_to_sched != NULL) { } - zarray_fini(&(w->resvals)); + zarray_fini(&(w->host.resvals)); my_rwlock_fini(&(priv->lock)); memset(priv, 0x0, sizeof(*priv)); @@ -101,7 +101,7 @@ static int worker_reference(meta_obj_t* obj) static void worker_dereference(meta_obj_t* obj) { worker_private_t* priv = (worker_private_t*)RTTI_OBJ_TO_PRIV(obj); - xce_worker_t* w = RTTI_OBJ_TO_DATA(xce_worker_t, obj); + worker_t* w = RTTI_OBJ_TO_DATA(worker_t, obj); uint64_t count = 0; my_rwlock_wlock(&(priv->lock)); @@ -159,8 +159,9 @@ static typeinfo_t worker_type = { .unlock = (unlock_fn_t)worker_unlock } } }; -xce_worker_t* makeWorker(xce_worker_t* w) +worker_t* makeWorker(var workermgr, xce_worker_t* w) { - xce_worker_t* worker = makeInstance(&worker_type, w); + worker_t* worker = makeInstance(&worker_type, w); + /* fIXME */ return worker; } diff --git a/src/meson.build b/src/meson.build index 7a483dc..6e3ceee 100644 --- a/src/meson.build +++ b/src/meson.build @@ -60,3 +60,12 @@ subdir('worker') subdir('autoscaler') subdir('scheduler') subdir('apiserver') + +configure_file( + input: 'tools/appimage/entry.sh', + output: 'xcerun', + copy: true, + install: true, + install_dir: 'bin', + install_mode: 'rwxr-xr-x' +) diff --git a/src/nixSupport/xce.nix b/src/nixSupport/xce.nix index d9187f9..2df0cad 100644 --- a/src/nixSupport/xce.nix +++ b/src/nixSupport/xce.nix @@ -60,6 +60,8 @@ stdenv.mkDerivation { separateDebugInfo = true; })) packcc + cowsay + fortune ]; build-system = [ pkgs.meson ]; @@ -75,4 +77,8 @@ stdenv.mkDerivation { ("-Dcommit_id=" + myCommitId) # (lib.optional (!with_debug) "-Dbuildtype=release") ]; + + meta = { + mainProgram = "xcerun"; + }; } diff --git a/src/tools/appimage/entry.sh b/src/tools/appimage/entry.sh new file mode 100644 index 0000000..a8a4625 --- /dev/null +++ b/src/tools/appimage/entry.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +case "$1" in + hello) + shift + cowsay "Hello, $*" + ;; + fortune) + fortune + ;; + *) + echo "Usage: $0 {hello |fortune}" + exit 1 + ;; +esac