#include "config.h" #include #include #include #include "cluster.h" #include "controller.capnp.h" #include "utils.h" #include "xce/cluster_def.h" static int do_reply_reslist(cluster_t* c, endpoint_t* ep, xce_request_t* request) { xce_reply_t reply; xce_resource_data_t resdata; xce_data_t getdata = { .table = request->body.getmsg->table, .data = { .resources = &resdata } }; int rc = 0; xlist_t rlist; init_reply_by_req(&reply, request); xlist_init(&rlist, 512); collect_resources(c, NULL, &rlist); resdata.n_total = rlist.count; resdata.resources = (xce_resource_t**)rlist.data; reply.status = 0; reply.data_which = RequestReply_data_getdata; reply.data.getdata = &getdata; rc = xce_reply_get(ep, c->myhost, &reply, NULL, NULL, -1); xlist_fini(&rlist); return rc; } static int do_reply_resvals(cluster_t* c, endpoint_t* ep, xce_request_t* request) { xce_reply_t reply; xce_resval_data_t resvaldata; xce_get_t* getmsg = request->body.getmsg; xce_data_t getdata = { .data = { .resvals = &resvaldata } }; int rc = 0; xce_val_t val; xce_resval_t resval; xce_resval_t* vals[] = { &resval }; init_reply_by_req(&reply, request); resvaldata.n_total = 0; resvaldata.vals = NULL; if (getmsg != NULL) { xce_match_t* condition = getmsg->condition; if (condition != NULL) { getdata.table = getmsg->table; rc = get_resval(c, condition->condition.reskey->resname, condition->condition.reskey->hostname, &val); if (rc >= 0) { resval.val = &val; resval.name = condition->condition.reskey->resname; resval.host = condition->condition.reskey->hostname; resvaldata.n_total = 1; resvaldata.vals = vals; } } else { rc = -EINVAL; } } else { rc = -EINVAL; } reply.status = rc; reply.data_which = RequestReply_data_getdata; reply.data.getdata = &getdata; rc = xce_reply_get(ep, c->myhost, &reply, NULL, NULL, -1); return rc; } int check_worker_(xce_worker_t* w) { if (w == NULL) { return 0; } return 1; } static int do_reply_workerlist(cluster_t* c, endpoint_t* ep, xce_request_t* request) { xce_reply_t reply; xce_workers_data_t workerdata; /* clang-format off */ xce_data_t getdata = { .table = request->body.getmsg->table, .data = { .workers = &workerdata } }; /* clang-format on */ int rc = 0; int i; xlist_t wlist; init_reply_by_req(&reply, request); xlist_init(&wlist, 1024); collect_workers(c, (collect_check_fn_t)check_worker_, &wlist); workerdata.n_total = wlist.count; workerdata.workers = (xce_worker_t**)wlist.data; reply.status = 0; reply.data_which = RequestReply_data_getdata; reply.data.getdata = &getdata; rc = xce_reply_get(ep, c->myhost, &reply, check_controller, c, 100); for (i = 0; i < workerdata.n_total; i++) { xce_worker_t* w = workerdata.workers[i]; free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched); } unlock_collect_workers(&wlist); xlist_fini(&wlist); return rc; } static int do_reply_joblist(cluster_t* c, endpoint_t* ep, xce_request_t* request) { xce_reply_t reply; xce_jobs_data_t jobdata; xce_data_t getdata = { .table = request->body.getmsg->table, .data = { .jobs = &jobdata } }; int rc = 0; xce_match_t* match = request->body.getmsg->condition; xlist_t jlist; init_reply_by_req(&reply, request); xlist_init(&jlist, 1024 * 1024); if (match->type == MatchCondition_condition_matchjid) { job_t* j = NULL; jobdata.n_total = 0; jobdata.jobs = NULL; lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)match->condition.jobid, (void**)&j); if (j != NULL) { reference(j); } unlock(c->jobmgr); if (rc > 0) { lock(j, RTTI_LOCK_READ); xlist_store(&jlist, &(j->job)); } else { dereference(j); } } else { collect_jobs(c, NULL, &jlist); } jobdata.n_total = jlist.count; jobdata.jobs = (xce_job_t**)jlist.data; reply.status = 0; reply.data_which = RequestReply_data_getdata; reply.data.getdata = &getdata; rc = xce_reply_get(ep, c->myhost, &reply, check_controller, c, 100); unlock_collect_jobs(&jlist); FREEUP(jobdata.jobs); return rc; } int do_get_op(cluster_t* c, endpoint_t* ep, xce_request_t* request) { int rc = 0; xce_reply_t reply; TRACE("%s table %d", __func__, request->body.getmsg->table); init_reply_by_req(&reply, request); switch (request->body.getmsg->table) { case Data_data_resources: rc = do_reply_reslist(c, ep, request); break; case Data_data_workers: case Data_data_schedworkers: rc = do_reply_workerlist(c, ep, request); break; case Data_data_schedjobs: case Data_data_jobs: rc = do_reply_joblist(c, ep, request); break; case Data_data_resvals: rc = do_reply_resvals(c, ep, request); break; default: rc = -ENOENT; reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); break; } return rc; }