#include #include #include #include #include #include #include "jobj.h" #include "kv.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); 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; } 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; } 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); 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*); if (src->submit == NULL) { return -EINVAL; } 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; 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) { /* FIXME??? */ } 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; } int job_obj_init(int n_sub_table_slots, int n_startinfo_table_slots) { int rc = 0; rc = jobsub_table_init(n_sub_table_slots); if (rc < 0) { return rc; } rc = startinfo_table_init(n_startinfo_table_slots); return rc; } void job_obj_fini(void) { jobsub_table_fini(); startinfo_table_fini(); }