#include #include #include #include #include #include #include "jobj.h" #include "kv.h" #include "misc.h" #include "mylock.h" #include "preason.h" #include "utils.h" #include "xce/types.h" #include "xstring.h" static struct { mylock_t lock; var sub_table; } jobsub_table; static struct { mylock_t lock; var startinfo_table; } startinfo_table; static typeinfo_t job_submit_type; static uint32_t submit_data_key_hash(job_submit_data_t* key) { return hash_block((char*)key, sizeof(*key)); } static int submit_data_key_assign(job_submit_data_t** dst, job_submit_data_t* src) { job_submit_data_t* p = calloc(1, sizeof(*p)); if (p == NULL) { return -ENOMEM; } memcpy(p, src, sizeof(*p)); *dst = p; return 1; } static int submit_data_key_cmp(job_submit_data_t* key1, job_submit_data_t* key2) { return memcmp((void*)key1, (void*)key2, sizeof(*key1)); } static void submit_data_key_destroy(job_submit_data_t* key) { if (key == NULL) { return; } FREEUP(key); } static kv_key_ops_t submit_data_key_ops = { .hash = (kv_key_hash_fn_t)submit_data_key_hash, .assign = (kv_key_assign_fn_t)submit_data_key_assign, .cmp = (kv_key_cmp_fn_t)submit_data_key_cmp, .destroy = (kv_key_destroy_fn_t)submit_data_key_destroy }; static void jobsub_destroy(meta_obj_t*); static void jobsub_killer(var s) { if (s == NULL) { return; } destroy_(s, (obj_destroy_fn_t)jobsub_destroy); } static int jobsub_table_init(int nslots) { kv_params_t params; if (nslots <= 1) { return -EINVAL; } params.nslots = nslots; params.keyops = &submit_data_key_ops; params.datakiller = jobsub_killer; mylock_init(&(jobsub_table.lock)); jobsub_table.sub_table = makeInstance(&kvtab_type, ¶ms); if (jobsub_table.sub_table == NULL) { return -errno; } return 1; } static void jobsub_table_fini() { destroy(jobsub_table.sub_table); /* Leave no dangling pointer behind: makeJobsub() checks this for * NULL (see below) to detect a not-(yet-)initialized table instead * of dereferencing freed memory (Requirement 10.2 hardening). */ jobsub_table.sub_table = NULL; mylock_fini(&(jobsub_table.lock)); } typedef struct { atomic_ulong refcount; } job_submit_private_t; static size_t jobsub_datasize(va_list ap) { size_t sz = sizeof(job_submit_data_t); return sz; } static int jobsub_init(meta_obj_t* obj, va_list ap) { job_submit_data_t* src = va_arg(ap, job_submit_data_t*); job_submit_data_t* dst = RTTI_OBJ_TO_DATA(job_submit_data_t, obj); job_submit_private_t* priv = (job_submit_private_t*)RTTI_OBJ_TO_PRIV(obj); if (src == NULL) { return -EINVAL; } dst->queue = makeString(src->queue); dst->user = makeString(src->user); dst->mintasks = src->mintasks; dst->maxtasks = src->maxtasks; if (src->start_info != NULL) { dst->start_info = makeJobstartinfo(src->start_info); if (dst->start_info == NULL) { jobsub_destroy(obj); return -ENOMEM; } } else { dst->start_info = NULL; } dst->job_name = makeString(src->job_name); dst->project_name = makeString(src->project_name); dst->user_group = makeString(src->user_group); dst->job_description = makeString(src->job_description); dst->resreq = makeString(src->resreq); if (!(EQUAL_STRING(dst->queue, src->queue) && EQUAL_STRING(dst->user, src->user) && EQUAL_STRING(dst->job_name, src->job_name) && EQUAL_STRING(dst->project_name, src->project_name) && EQUAL_STRING(dst->user_group, src->user_group) && EQUAL_STRING(dst->job_description, src->job_description) && EQUAL_STRING(dst->resreq, src->resreq))) { jobsub_destroy(obj); return -ENOMEM; } priv->refcount = 1; return 1; } static void jobsub_destroy(meta_obj_t* obj) { job_submit_data_t* jobsub = RTTI_OBJ_TO_DATA(job_submit_data_t, obj); job_submit_private_t* priv = (job_submit_private_t*)RTTI_OBJ_TO_PRIV(obj); dereference(jobsub->queue); dereference(jobsub->user); dereference(jobsub->job_name); dereference(jobsub->project_name); dereference(jobsub->user_group); dereference(jobsub->job_description); dereference(jobsub->resreq); if (jobsub->start_info != NULL) { dereference(jobsub->start_info); } memset(priv, 0x0, sizeof(*priv)); } static int jobsub_reference(meta_obj_t* obj) { job_submit_private_t* priv = (job_submit_private_t*)RTTI_OBJ_TO_PRIV(obj); atomic_fetch_add(&(priv->refcount), 1); return 1; } static void jobsub_dereference(meta_obj_t* obj) { job_submit_private_t* priv = (job_submit_private_t*)RTTI_OBJ_TO_PRIV(obj); job_submit_data_t* data = RTTI_OBJ_TO_DATA(job_submit_data_t, obj); uint64_t count; count = atomic_fetch_sub(&(priv->refcount), 1); if (count == 1) { mylock_lock(&(jobsub_table.lock)); key_del(jobsub_table.sub_table, data); mylock_unlock(&(jobsub_table.lock)); } return; } static typeinfo_t job_submit_type = { .name = "job submit", .privateSize = sizeof(job_submit_private_t), .dataSize = (datasize_fn_t)jobsub_datasize, .init = (obj_init_fn_t)jobsub_init, .destroy = (obj_destroy_fn_t)NULL, /* disable destroy() */ .interfaces = { .reference = &(intf_reference_t) { .reference = (reference_fn_t)jobsub_reference, .dereference = (dereference_fn_t)jobsub_dereference } } }; job_submit_data_t* makeJobsub(job_submit_data_t* subdata) { var jobsub = NULL; var jobsub1 = NULL; int rc = 0; if (subdata == NULL) { errno = EINVAL; return NULL; } if (jobsub_table.sub_table == NULL) { /* job_obj_init() has not (yet) completed for this generation of * the module -- dispatching key_find/key_add below against an * uninitialized/torn-down table is unsafe, not just wrong. * Root cause of the previously observed makeJobsub() * NULL-with-EINVAL-for-a-valid-src failure (Requirement 10.2): * that generic EINVAL was indistinguishable from "subdata was * invalid" above. Fail loudly with a distinct errno instead, so * an ordering bug can never again masquerade as a bad-argument * error. */ errno = ENODEV; return NULL; } jobsub = makeInstance(&job_submit_type, subdata); if (jobsub == NULL) { return NULL; } mylock_lock(&(jobsub_table.lock)); rc = key_find(jobsub_table.sub_table, (void*)jobsub, (void**)&jobsub1); if (rc > 0) { reference(jobsub1); } mylock_unlock(&(jobsub_table.lock)); if (rc > 0) { destroy_(jobsub, (obj_destroy_fn_t)jobsub_destroy); return (job_submit_data_t*)jobsub1; } mylock_lock(&(jobsub_table.lock)); rc = key_add(jobsub_table.sub_table, (void*)jobsub, jobsub); mylock_unlock(&(jobsub_table.lock)); if (rc < 0) { destroy_(jobsub, (obj_destroy_fn_t)jobsub_destroy); jobsub = NULL; errno = -rc; } else { errno = 0; } return jobsub; } typedef struct { atomic_ulong refcount; } startinfo_private_t; static void startinfo_destroy(meta_obj_t* obj) { job_startinfo_t* startinfo = RTTI_OBJ_TO_DATA(job_startinfo_t, obj); startinfo_private_t* priv = (startinfo_private_t*)RTTI_OBJ_TO_PRIV(obj); dereference(startinfo->exechome); dereference(startinfo->execcwd); dereference(startinfo->execuser); dereference(startinfo->cmdline); if (startinfo->exdata != NULL) { dereference(startinfo->exdata); } if (startinfo->launch_type != NULL) { dereference(startinfo->launch_type); } if (startinfo->q_pre_exec != NULL) { dereference(startinfo->q_pre_exec); } if (startinfo->q_post_exec != NULL) { dereference(startinfo->q_post_exec); } if (startinfo->job_pre_exec != NULL) { dereference(startinfo->job_pre_exec); } if (startinfo->job_post_exec != NULL) { dereference(startinfo->job_post_exec); } memset(priv, 0x0, sizeof(*priv)); return; } static size_t startinfo_datasize(va_list ap) { size_t sz = sizeof(job_startinfo_t); return sz; } static int startinfo_init(meta_obj_t* obj, va_list ap) { job_startinfo_t* src = va_arg(ap, job_startinfo_t*); job_startinfo_t* dst = RTTI_OBJ_TO_DATA(job_startinfo_t, obj); startinfo_private_t* priv = RTTI_OBJ_TO_PRIV(obj); if (src == NULL) { return -EINVAL; } priv->refcount = 1; dst->exechome = makeString(src->exechome); dst->execcwd = makeString(src->execcwd); dst->execuser = makeString(src->execuser); dst->cmdline = makeString(src->cmdline); dst->exdata = NULL; if (src->exdata != NULL) { dst->exdata = makeString(src->exdata); } dst->launch_type = NULL; if (src->launch_type != NULL) { dst->launch_type = makeString(src->launch_type); } dst->q_pre_exec = NULL; if (src->q_pre_exec != NULL) { dst->q_pre_exec = makeString(src->q_pre_exec); } dst->q_post_exec = NULL; if (src->q_post_exec != NULL) { dst->q_post_exec = makeString(src->q_post_exec); } dst->job_pre_exec = NULL; if (src->job_pre_exec != NULL) { dst->job_pre_exec = makeString(src->job_pre_exec); } dst->job_post_exec = NULL; if (src->job_post_exec != NULL) { dst->job_post_exec = makeString(src->job_post_exec); } return 1; } static int startinfo_reference(meta_obj_t* obj) { startinfo_private_t* priv = (startinfo_private_t*)RTTI_OBJ_TO_PRIV(obj); atomic_fetch_add(&(priv->refcount), 1); return 1; } static void startinfo_dereference(meta_obj_t* obj) { job_startinfo_t* startinfo = RTTI_OBJ_TO_DATA(job_startinfo_t, obj); startinfo_private_t* priv = (startinfo_private_t*)RTTI_OBJ_TO_PRIV(obj); uint64_t count; count = atomic_fetch_sub(&(priv->refcount), 1); if (count == 1) { mylock_lock(&(startinfo_table.lock)); key_del(startinfo_table.startinfo_table, startinfo); mylock_unlock(&(startinfo_table.lock)); } return; } static typeinfo_t job_startinfo_type = { .name = "job start info", .privateSize = sizeof(startinfo_private_t), .dataSize = (datasize_fn_t)startinfo_datasize, .init = (obj_init_fn_t)startinfo_init, .destroy = NULL, .interfaces = { .reference = &(intf_reference_t) { .reference = (reference_fn_t)startinfo_reference, .dereference = (dereference_fn_t)startinfo_dereference } } }; job_startinfo_t* makeJobstartinfo(job_startinfo_t* startinfo) { var startinfo1 = NULL; var startinfo2 = NULL; int rc = 0; if (startinfo == NULL) { errno = EINVAL; return NULL; } if (startinfo_table.startinfo_table == NULL) { /* Mirrors makeJobsub()'s guard above -- see its comment. */ errno = ENODEV; return NULL; } startinfo1 = makeInstance(&(job_startinfo_type), startinfo); if (startinfo1 == NULL) { return NULL; } mylock_lock(&(startinfo_table.lock)); rc = key_find(startinfo_table.startinfo_table, (void*)startinfo1, (void**)&startinfo2); if (rc > 0) { reference(startinfo2); } mylock_unlock(&(startinfo_table.lock)); if (rc > 0) { destroy_(startinfo1, (obj_destroy_fn_t)startinfo_destroy); return (job_startinfo_t*)startinfo2; } mylock_lock(&(startinfo_table.lock)); rc = key_add(startinfo_table.startinfo_table, (void*)startinfo1, startinfo1); mylock_unlock(&(startinfo_table.lock)); if (rc < 0) { destroy_(startinfo1, (obj_destroy_fn_t)startinfo_destroy); startinfo1 = NULL; errno = -rc; } else { errno = 0; } return startinfo1; } static void startinfo_killer(var s) { if (s == NULL) { return; } destroy_(s, (obj_destroy_fn_t)startinfo_destroy); } static uint32_t startinfo_key_hash(job_startinfo_t* key) { return hash_block((void*)key, sizeof(*key)); } static int startinfo_key_assign(job_startinfo_t** dst, job_startinfo_t* src) { job_startinfo_t* p = calloc(1, sizeof(*p)); if (p == NULL) { return -ENOMEM; } memcpy(p, src, sizeof(*p)); *dst = p; return 1; } static int startinfo_key_cmp(job_startinfo_t* k1, job_startinfo_t* k2) { return memcmp((void*)k1, (void*)k2, sizeof(*k1)); } static void startinfo_key_destroy(job_startinfo_t* key) { if (key == NULL) { return; } FREEUP(key); } static kv_key_ops_t startinfo_key_ops = { .hash = (kv_key_hash_fn_t)startinfo_key_hash, .assign = (kv_key_assign_fn_t)startinfo_key_assign, .cmp = (kv_key_cmp_fn_t)startinfo_key_cmp, .destroy = (kv_key_destroy_fn_t)startinfo_key_destroy }; static int startinfo_table_init(int nslots) { kv_params_t params; if (nslots <= 1) { return -EINVAL; } params.nslots = nslots; params.keyops = &startinfo_key_ops; params.datakiller = startinfo_killer; mylock_init(&(startinfo_table.lock)); startinfo_table.startinfo_table = makeInstance(&kvtab_type, ¶ms); if (startinfo_table.startinfo_table == NULL) { return -errno; } return 0; } static void startinfo_table_fini() { destroy(startinfo_table.startinfo_table); /* See jobsub_table_fini(): avoid leaving a dangling pointer that * makeJobstartinfo()'s not-(yet-)initialized guard would otherwise * have to distinguish from a live table. */ startinfo_table.startinfo_table = NULL; mylock_fini(&(startinfo_table.lock)); return; } typedef struct { my_rwlock_t lock; atomic_ulong refcount; } job_private_t; static size_t job_datasize(va_list ap) { size_t sz = sizeof(job_t); return sz; } static int job_init(meta_obj_t* obj, va_list ap) { 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*); job_submit_data_t* submit = NULL; char* ckptinfo = NULL; int saved_errno; if (src->submit == NULL) { return -EINVAL; } /* Resolve every fallible field BEFORE any MOVE_PTR runs below, so a * construction failure here never leaves src half-moved: callers * (e.g. job.c's restore_job()) rely on src being fully intact so * free_Job() can release everything on a create_job()/makeJob() * failure. Requirement 10.1: check makeJobsub()'s (and * makeString(ckptinfo)'s) return and propagate failure instead of * silently storing NULL and reporting success. */ submit = makeJobsub(src->submit); if (submit == NULL) { saved_errno = (errno != 0) ? errno : ENOMEM; return -saved_errno; } if (src->ckptinfo != NULL) { ckptinfo = makeString(src->ckptinfo); if (ckptinfo == NULL) { saved_errno = (errno != 0) ? errno : ENOMEM; dereference(submit); return -saved_errno; } } dst->job.jobid = src->jobid; dst->job.submit = 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 = 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; return 1; } static void job_destroy(meta_obj_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->job.submit); dereference(j->job.ckptinfo); if (j->job.pidinfo != NULL) { FREEUP(j->job.pidinfo->pids); FREEUP(j->job.pidinfo->pgids); FREEUP(j->job.pidinfo->exdata); FREEUP(j->job.pidinfo); } if (j->job.allocs != NULL) { free_JobAlloc_ptr(&(j->job.allocs)); } if (j->job.accounts != NULL) { /* Requirement 10.4: accounts is MOVE_PTR'd into the job during * job_init but was never freed here -- a genuine leak, multiplied * by array element count once array submission lands. Mirrors * the free_job_account() usage already established for this * exact type in sched.c's job teardown path (misc.c owns the * free function since xce_job_account_t has no capnp-generated * free_*_ptr counterpart -- it is not a capnp-mapped struct). */ free_job_account(j->job.accounts); j->job.accounts = NULL; } if (j->job.run_usage != NULL) { int i; for (i = 0; i < j->job.run_usage->num_hosts; i++) { if (j->job.run_usage->host_runusage[i] != NULL) { FREEUP(j->job.run_usage->host_runusage[i]->host_name); } FREEUP(j->job.run_usage->host_runusage[i]); } FREEUP(j->job.run_usage->host_runusage); FREEUP(j->job.run_usage); } if (j->job.exit_info != NULL) { FREEUP(j->job.exit_info); } if (j->job.pend_reasons != NULL) { erase_job_preason(j->job.pend_reasons); } my_rwlock_fini(&(priv->lock)); memset(priv, 0x0, sizeof(*priv)); return; } static int job_reference(meta_obj_t* obj) { job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); atomic_fetch_add(&(priv->refcount), 1); return 1; } static void job_dereference(meta_obj_t* obj) { job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); job_t* j = RTTI_OBJ_TO_DATA(job_t, obj); uint64_t count; count = atomic_fetch_sub(&(priv->refcount), 1); if (count == 1) { destroy_(j, (obj_destroy_fn_t)job_destroy); } return; } static int job_lock(meta_obj_t* obj, uint64_t option) { job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); int rc = -EINVAL; switch (option) { case RTTI_LOCK_READ: rc = my_rwlock_rlock(&(priv->lock)); break; case RTTI_LOCK_WRITE: rc = my_rwlock_wlock(&(priv->lock)); break; default: break; } return rc; } static void job_unlock(meta_obj_t* obj) { job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); my_rwlock_unlock(&(priv->lock)); return; } static typeinfo_t job_type = { .name = "xce job", .privateSize = sizeof(job_private_t), .dataSize = (datasize_fn_t)job_datasize, .init = (obj_init_fn_t)job_init, .destroy = NULL, .interfaces = { .reference = &(intf_reference_t) { .reference = (reference_fn_t)job_reference, .dereference = (dereference_fn_t)job_dereference }, .lock = &(intf_lock_t) { .lock = (lock_fn_t)job_lock, .unlock = (unlock_fn_t)job_unlock } } }; static job_t* makeJob_(var jobmgr, xce_job_t* j) { job_t* job = NULL; int rc = 0; rc = key_find(jobmgr, (void*)j->jobid, (void**)&job); if ((rc < 0) && (rc != -ENOENT)) { errno = -rc; return NULL; } if (rc > 0) { errno = EEXIST; return NULL; } job = makeInstance(&job_type, j); if (job != NULL) { rc = key_add(jobmgr, (void*)job->job.jobid, (void**)job); if (rc < 0) { destroy(job); errno = -rc; return NULL; } } return job; } job_t* makeJob(var jobmgr, xce_job_t* j) { job_t* job = NULL; int rc = 0; lock(jobmgr, RTTI_LOCK_WRITE); job = makeJob_(jobmgr, j); rc = errno; unlock(jobmgr); errno = rc; return job; } /* Requirement 10.2: job_obj_init() must be safe to call more than once * within a process without an intervening job_obj_fini(). Root cause of * the previously observed jobsub_table ordering hazard: every controller * unit-test file that constructs jobs calls job_obj_init() from its own * setUp() (confirmed via src/controller/meson.build -- all such test * files are linked into the single controller_unit_test executable and * run in one process), each guarded only by its own file-local static * flag, none of which coordinate with each other. Across a full test run * job_obj_init() was therefore re-entered many times with no intervening * job_obj_fini(), and jobsub_table_init()/startinfo_table_init() * unconditionally re-ran mylock_init() on the already-live lock * (undefined behavior on an in-use mutex, per POSIX) and unconditionally * overwrote the table pointer, silently orphaning the previous table and * corrupting the refcount bookkeeping of every entry still referencing * it. This guard makes every call after the first a no-op success until * job_obj_fini() runs, matching what every existing call site already * expects (EQUAL_INT(0, job_obj_init(...))). The production controller * boot path (cluster_init_managers(), called once from * cluster_init_phase1()) was never actually at risk -- this hazard was * confined to the combined test binary. */ static atomic_flag job_obj_ready = ATOMIC_FLAG_INIT; int job_obj_init(int n_sub_table_slots, int n_startinfo_table_slots) { int rc = 0; if (atomic_flag_test_and_set(&job_obj_ready)) { return 0; } rc = jobsub_table_init(n_sub_table_slots); if (rc < 0) { atomic_flag_clear(&job_obj_ready); return rc; } rc = startinfo_table_init(n_startinfo_table_slots); if (rc < 0) { atomic_flag_clear(&job_obj_ready); return rc; } return 0; } void job_obj_fini(void) { jobsub_table_fini(); startinfo_table_fini(); atomic_flag_clear(&job_obj_ready); }