#include #include #include "cluster.h" #include "worker.h" #include "xce/types.h" #include "xlist.h" int add_to_xlist(xlist_t* l, void* data) { return xlist_store(l, data); } int add_worker_to_xlist(xlist_t* l, void* data) { worker_t* w = (worker_t*)data; return xlist_store(l, &(w->host)); } int add_queue_to_xlist(xlist_t* l, void* data) { queue_t* q = (queue_t*)data; return xlist_store(l, &(q->queue)); } int add_job_to_xlist(xlist_t* l, void* data) { job_t* j = (job_t*)data; return xlist_store(l, &(j->job)); } int get_ep_role(endpoint_t* ep) { ep_private_t* priv; if (ep == NULL) { return -EINVAL; } priv = (ep_private_t*)ep->data; if (priv == NULL) { ERROR("priv data of endpoint %p is NULL", ep); return -EINVAL; } return priv->role; } worker_t* get_ep_worker(cluster_t* c, endpoint_t* ep) { int role, idx; worker_t* w = NULL; int rc = 0; role = get_ep_role(ep); if (role != HostRole_worker) { ERROR("endpoint %p is not a worker", ep); return NULL; } idx = ((ep_private_t*)ep->data)->u.idx; if (idx < 0) { ERROR("invalid idx (%d) for endpoint %p", idx, ep); return NULL; } lock(c->workermgr, RTTI_LOCK_READ); rc = vget(c->workermgr, idx, (void**)&w); if (w != NULL) { reference(w); } unlock(c->workermgr); if (rc < 0) { return NULL; } if (w == NULL) { ERROR("can not find worker with id (%d)", idx); } return w; } int update_host_runusage(xce_job_runusage_t** runusage, char* host, int count, xce_rusage_t** usages) { xce_job_runusage_t* rusages = NULL; int i, j; if ((*runusage) == NULL) { *runusage = (xce_job_runusage_t*)calloc(1, sizeof(xce_job_runusage_t)); if ((*runusage) == NULL) { return -ENOMEM; } } rusages = *runusage; j = -1; for (i = 0; i < rusages->num_hosts; i++) { if (strcmp(host, rusages->host_runusage[i]->host_name) == 0) { j = i; break; } } if (j < 0) { int cnt = rusages->num_hosts + 1; xce_host_runusage_t** hru = (xce_host_runusage_t**)realloc( rusages->host_runusage, cnt * sizeof(xce_host_runusage_t*)); if (hru == NULL) { return -ENOMEM; } rusages->host_runusage = hru; j = cnt - 1; hru[j] = (xce_host_runusage_t*)calloc(1, sizeof(xce_host_runusage_t)); if (hru[j] == NULL) { return -ENOMEM; } rusages->num_hosts++; hru[j]->host_name = host; } for (i = 0; i < count; i++) { /* bug-1735: these must match the REAL worker's rusage collector * (collect_job_rusages(), xcecd_cgroups.c res_names[]) -- "cputime"/ * "mem" never matched anything that worker actually sends, so * cpu_time/mem_usage stayed at their calloc()'d zero forever. */ if (strcmp(usages[i]->resname, "cpu_time_sec") == 0) { rusages->host_runusage[j]->cpu_time = usages[i]->value; continue; } if (strcmp(usages[i]->resname, "memory_rss") == 0) { rusages->host_runusage[j]->mem_usage = usages[i]->value; continue; } } return 0; } /* bug-1735: derives xce_job_runusage_t's own job-level max_mem/avg_mem/ * used_mem/cpu_time from the per-host host_runusage[] entries * update_host_runusage() populates -- replaces update_job_rusages()'s * (job.c) standing "update other fields of job rusage later" FIXME. * used_mem is the SUM across hosts (a multi-host job's total memory * footprint); max_mem/avg_mem are the max/average across hosts; cpu_time is * the SUM across hosts. peak_cpu_time/peak_cpu_duration/mem_efficiency/ * avg_cpu_efficiency/peak_cpu_efficiency are intentionally left untouched -- * bug-1735 was explicitly scope-narrowed to memory + CPU time only; those * fields are bug-1736 (a design task, no formula exists yet). */ void aggregate_job_runusage(xce_job_runusage_t* runusage) { int i, valid = 0; double mem_sum = 0.0; double mem_max = 0.0; int cpu_sum = 0; if ((runusage == NULL) || (runusage->host_runusage == NULL)) { return; } for (i = 0; i < runusage->num_hosts; i++) { xce_host_runusage_t* h = runusage->host_runusage[i]; if (h == NULL) { continue; } mem_sum += h->mem_usage; cpu_sum += h->cpu_time; if ((valid == 0) || (h->mem_usage > mem_max)) { mem_max = h->mem_usage; } valid++; } if (valid == 0) { return; } runusage->used_mem = mem_sum; runusage->max_mem = mem_max; runusage->avg_mem = mem_sum / valid; runusage->cpu_time = cpu_sum; } int update_pids(job_pidinfo_t** pidinfo, int npids, int* pids) { job_pidinfo_t* info; int i; if ((*pidinfo) == NULL) { *pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t)); if ((*pidinfo) == NULL) { return -ENOMEM; } } info = *pidinfo; if (info->pids == NULL) { info->pids = (unsigned int*)calloc(npids, sizeof(unsigned int)); } else if (info->npids < npids) { unsigned int* ptr; ptr = (unsigned int*)realloc(info->pids, npids * sizeof(unsigned int)); if (ptr == NULL) { return -ENOMEM; } info->pids = ptr; } for (i = 0; i < npids; i++) { info->pids[i] = pids[i]; } info->npids = npids; return 0; } int update_pgids(job_pidinfo_t** pidinfo, int npgids, int* pgids) { job_pidinfo_t* info; int i; if ((*pidinfo) == NULL) { *pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t)); if ((*pidinfo) == NULL) { return -ENOMEM; } } info = *pidinfo; if (info->pgids == NULL) { info->pgids = (unsigned int*)calloc(npgids, sizeof(unsigned int)); } else if (info->npgids < npgids) { unsigned int* ptr; ptr = (unsigned int*)realloc(info->pgids, npgids * sizeof(unsigned int)); if (ptr == NULL) { return -ENOMEM; } info->pgids = ptr; } for (i = 0; i < npgids; i++) { info->pgids[i] = pgids[i]; } info->npgids = npgids; return 0; }