Rongsong Shen
e5fde7b55f
Crash: encode_NameValue_list called with count=21, s=NULL on worker rejoin, causing SIGSEGV in persist_worker_hook_fun. Root cause: free_name_val_list(xce_name_val_t***, int) accepted count by value. It zeroed *vals (resvals_to_sched = NULL) but could not zero nresvals_to_sched. Any caller that freed the scheduler resource snapshot (opget.c workerlist, reply.c clusterconf, dataex.c, sched.c dispatch) left nresvals_to_sched stale. When the worker later rejoined and the WAL HOOK fired, the WAL encoder read count=21 with a NULL pointer. Fix: change signature to free_name_val_list(xce_name_val_t***, int*) so both fields are zeroed atomically. Update all 7 call sites.
640 lines
18 KiB
C
640 lines
18 KiB
C
#include <errno.h>
|
|
#include <stdlib.h>
|
|
|
|
#include "cluster.h"
|
|
#include "controller.capnp.h"
|
|
#include "request.h"
|
|
#include "xce/types.h"
|
|
#include "xstring.h"
|
|
|
|
static int xce_apiserver_push_configs(cluster_t* c, endpoint_t* ep)
|
|
{
|
|
xce_configdata_t configdata;
|
|
int rc = 0;
|
|
|
|
memset(&configdata, 0x0, sizeof(configdata));
|
|
configdata.data.clusterconfig = c->myconf->cluster_config;
|
|
configdata.data_which = ConfigData_data_clusterconfig;
|
|
|
|
rc = xce_pushconfig_req(ep, c->myhost, &configdata);
|
|
|
|
return rc;
|
|
}
|
|
|
|
static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep)
|
|
{
|
|
xce_data_t data;
|
|
xce_workers_data_t workers
|
|
= { .n_total = 0, .pos = 0, .count = 0, .workers = NULL };
|
|
int i, rc = 0;
|
|
xlist_t wlist;
|
|
|
|
memset(&data, 0x0, sizeof(data));
|
|
xlist_init(&wlist, 8192);
|
|
|
|
data.table = Data_data_workers;
|
|
data.data.workers = &workers;
|
|
|
|
collect_workers(c, (collect_check_fn_t)check_worker_, &wlist);
|
|
workers.n_total = wlist.count;
|
|
workers.workers = (xce_worker_t**)wlist.data;
|
|
workers.count = workers.n_total;
|
|
|
|
rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);
|
|
|
|
/* free up collected resource value which used
|
|
passed to api server
|
|
*/
|
|
for (i = 0; i < workers.n_total; i++) {
|
|
xce_worker_t* w = workers.workers[i];
|
|
|
|
free_name_val_list(&(w->resvals_to_sched), &(w->nresvals_to_sched));
|
|
}
|
|
|
|
unlock_collect_workers(&wlist);
|
|
|
|
FREEUP(workers.workers);
|
|
|
|
return rc;
|
|
}
|
|
|
|
static int check_job_(xce_job_t* j)
|
|
{
|
|
if (j == NULL) {
|
|
return 0;
|
|
}
|
|
|
|
return 1;
|
|
}
|
|
|
|
static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep)
|
|
{
|
|
xce_data_t data;
|
|
xce_jobs_data_t jobs = { .n_total = 0, .pos = 0, .count = 0, .jobs = NULL };
|
|
int rc = 0;
|
|
xlist_t jlist;
|
|
|
|
memset(&data, 0x0, sizeof(data));
|
|
xlist_init(&jlist, 1024 * 1024);
|
|
|
|
data.table = Data_data_jobs;
|
|
data.data.jobs = &jobs;
|
|
|
|
collect_jobs(c, (collect_check_fn_t)check_job_, &jlist);
|
|
jobs.n_total = jlist.count;
|
|
jobs.jobs = (xce_job_t**)jlist.data;
|
|
jobs.count = jobs.n_total;
|
|
|
|
rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);
|
|
|
|
unlock_collect_jobs(&jlist);
|
|
|
|
FREEUP(jobs.jobs);
|
|
|
|
return rc;
|
|
}
|
|
|
|
static int apiserver_push_joblist(cluster_t* c, xlist_t* l)
|
|
{
|
|
int i;
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) {
|
|
int rc;
|
|
xce_data_t data;
|
|
xce_jobs_data_t jobs = { .n_total = l->count,
|
|
.pos = 0,
|
|
.count = l->count,
|
|
.jobs = (xce_job_t**)l->data };
|
|
|
|
memset(&data, 0x0, sizeof(data));
|
|
data.table = Data_data_jobs;
|
|
data.data.jobs = &jobs;
|
|
|
|
rc = xce_push_data_req(c->apiservers[i].ep, c->myhost, &data,
|
|
check_controller, c, 100);
|
|
if (rc < 0) {
|
|
WARN("%s push job list to api server %s has problem, rc %d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_job_new_fun(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
xlist_t* l = va_arg(ap, xlist_t*);
|
|
int rc = 0;
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
rc = -EINVAL;
|
|
if (l != NULL) {
|
|
rc = apiserver_push_joblist(c, l);
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return rc;
|
|
}
|
|
|
|
static int apiserver_job_start_fun(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
job_pidinfo_t* info = va_arg(ap, job_pidinfo_t*);
|
|
int i;
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (c->apiservers[i].ep != NULL)) {
|
|
int rc;
|
|
xce_notify_data_t xdata = { .event = NotifyData_data_jobPidInfo,
|
|
.data = { .pidinfo = info } };
|
|
|
|
/* FIXME: add job pid to API server? */
|
|
rc = notify(c, c->apiservers[i].ep, &xdata);
|
|
if (rc < 0) {
|
|
WARN("%s notify job start to api server %s fail, rc %d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_job_exit_fun(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
xce_jobfinish_t* info = va_arg(ap, xce_jobfinish_t*);
|
|
int i;
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (c->apiservers[i].ep != NULL)) {
|
|
int rc;
|
|
xce_notify_data_t xdata
|
|
= { .event = NotifyData_data_jobFinishInfo,
|
|
.data = { .finishinfo = info } };
|
|
|
|
/* FIXME: we need to send exit code of job to API server */
|
|
rc = notify(c, c->apiservers[i].ep, &xdata);
|
|
if (rc < 0) {
|
|
WARN("%s notify job exit to api server %s fail, rc %d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_job_fail_fun(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
job_failinfo_t* info = va_arg(ap, job_failinfo_t*);
|
|
int i;
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (c->apiservers[i].ep != NULL)) {
|
|
int rc;
|
|
xce_notify_data_t xdata
|
|
= { .event = NotifyData_data_jobFailInfo,
|
|
.data = { .failinfo = info } };
|
|
|
|
/* FIXME: we need to add fail reason of job to api server */
|
|
rc = notify(c, c->apiservers[i].ep, &xdata);
|
|
if (rc < 0) {
|
|
WARN("%s notify job fail to api server %s fail, rc %d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_job_update_fun(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
job_t* job = va_arg(ap, job_t*);
|
|
int i;
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (c->apiservers[i].ep != NULL)) {
|
|
int rc;
|
|
xce_notify_data_t xdata
|
|
= { .event = NotifyData_data_updateJob,
|
|
.data = { .update_job = &(job->job) } };
|
|
|
|
rc = notify(c, c->apiservers[i].ep, &xdata);
|
|
if (rc < 0) {
|
|
WARN("%s notify job update to api server %s fail, rc %d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_job_rusage_update_fun(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
xce_job_rusages_t* rusages = va_arg(ap, xce_job_rusages_t*);
|
|
int i;
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (c->apiservers[i].ep != NULL)) {
|
|
int rc;
|
|
xce_notify_data_t xdata
|
|
= { .event = NotifyData_data_schedJobRusages,
|
|
.data = { .sched_job_rusages = rusages } };
|
|
rc = notify(c, c->apiservers[i].ep, &xdata);
|
|
if (rc < 0) {
|
|
WARN("%s notify rusage update to api server %s fail, rc %d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_job_hook_fun(int type, int situation, int detail,
|
|
void* data, va_list ap)
|
|
{
|
|
switch (detail) {
|
|
case HOOK_DETAIL_JOB_NEW:
|
|
return apiserver_job_new_fun(type, situation, data, ap);
|
|
case HOOK_DETAIL_JOB_START:
|
|
return apiserver_job_start_fun(type, situation, data, ap);
|
|
case HOOK_DETAIL_JOB_EXIT:
|
|
return apiserver_job_exit_fun(type, situation, data, ap);
|
|
case HOOK_DETAIL_JOB_UPDATE:
|
|
return apiserver_job_update_fun(type, situation, data, ap);
|
|
case HOOK_DETAIL_JOB_FAIL:
|
|
return apiserver_job_fail_fun(type, situation, data, ap);
|
|
case HOOK_DETAIL_JOB_RUSAGE_UPDATE:
|
|
return apiserver_job_rusage_update_fun(type, situation, data, ap);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_push_workerlist(cluster_t* c, endpoint_t* ep, xlist_t* l)
|
|
{
|
|
xce_data_t data;
|
|
xce_workers_data_t workers = { .n_total = l->count,
|
|
.pos = 0,
|
|
.count = l->count,
|
|
.workers = (xce_worker_t**)l->data };
|
|
int rc = 0;
|
|
|
|
memset(&data, 0x0, sizeof(data));
|
|
|
|
data.table = Data_data_workers;
|
|
data.data.workers = &workers;
|
|
|
|
rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);
|
|
return rc;
|
|
}
|
|
|
|
static int apiserver_worker_change(int type, int situation, void* data,
|
|
cluster_t* c, worker_t* w)
|
|
{
|
|
int i;
|
|
xlist_t resvalist;
|
|
|
|
TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx,
|
|
situation);
|
|
|
|
xlist_init(&resvalist, 256);
|
|
collect_resval(c->resmgr, &(w->host.resvals), &resvalist);
|
|
w->host.nresvals_to_sched = resvalist.count;
|
|
w->host.resvals_to_sched = (xce_name_val_t**)resvalist.data;
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (c->apiservers[i].ep != NULL)) {
|
|
xlist_t wlist;
|
|
int rc;
|
|
|
|
xlist_init(&wlist, 2);
|
|
xlist_store(&wlist, w);
|
|
rc = apiserver_push_workerlist(c, c->apiservers[i].ep, &wlist);
|
|
if (rc < 0) {
|
|
WARN("%s notify worker change failed to api server %s, rc "
|
|
"%d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
xlist_fini(&wlist);
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
w->host.nresvals_to_sched = 0;
|
|
w->host.resvals_to_sched = NULL;
|
|
xlist_fini(&resvalist);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_worker_join(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
char* name = va_arg(ap, char*);
|
|
worker_t* w = NULL;
|
|
int rc;
|
|
|
|
if (situation == HOOK_BEFORE_EVENT) {
|
|
/* nothing to do */
|
|
return 0;
|
|
}
|
|
|
|
lock(c->workermgr, RTTI_LOCK_WRITE);
|
|
rc = key_find(c->workermgr, (void*)name, (void**)&w);
|
|
if (w != NULL) {
|
|
reference(w);
|
|
}
|
|
unlock(c->workermgr);
|
|
|
|
if (rc < 0) {
|
|
return rc;
|
|
}
|
|
|
|
if (rc == 0) {
|
|
return -ENOENT;
|
|
}
|
|
|
|
lock(w, RTTI_LOCK_READ);
|
|
rc = apiserver_worker_change(type, situation, data, c, w);
|
|
unlock(w);
|
|
|
|
dereference(w);
|
|
|
|
return rc;
|
|
}
|
|
|
|
static int apiserver_worker_dead(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
worker_t* w = va_arg(ap, worker_t*);
|
|
|
|
/* FIXME: need to handle job status in this worker */
|
|
if (situation == HOOK_BEFORE_EVENT) {
|
|
return 0;
|
|
}
|
|
|
|
return apiserver_worker_change(type, situation, data, c, w);
|
|
}
|
|
|
|
static int apiserver_worker_update(int type, int situation, void* data,
|
|
va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
worker_t* w = va_arg(ap, worker_t*);
|
|
|
|
if (situation == HOOK_BEFORE_EVENT) {
|
|
return 0;
|
|
}
|
|
|
|
return apiserver_worker_change(type, situation, data, c, w);
|
|
}
|
|
|
|
static int apiserver_worker_hook_fun(int type, int situation, int detail,
|
|
void* data, va_list ap)
|
|
{
|
|
switch (detail) {
|
|
case HOOK_DETAIL_WORKER_JOIN:
|
|
return apiserver_worker_join(type, situation, data, ap);
|
|
case HOOK_DETAIL_WORKER_DEAD:
|
|
return apiserver_worker_dead(type, situation, data, ap);
|
|
case HOOK_DETAIL_WORKER_UPDATE:
|
|
return apiserver_worker_update(type, situation, data, ap);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_resource_hook_fun(int type, int situation, int detail,
|
|
void* data, va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
xce_resval_list_t* resvals = va_arg(ap, xce_resval_list_t*);
|
|
int i;
|
|
|
|
if (detail != HOOK_DETAIL_RESOURCE_UPDATE) {
|
|
return 0;
|
|
}
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (c->apiservers[i].ep != NULL)) {
|
|
int rc;
|
|
xce_notify_data_t xdata
|
|
= { .event = NotifyData_data_schedUpdateResvals,
|
|
.data = { .sched_update_resvals = resvals } };
|
|
|
|
rc = notify(c, c->apiservers[i].ep, &xdata);
|
|
if (rc < 0) {
|
|
WARN("%s notify api server %s resource update fail, rc %d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int apiserver_queue_hook_fun(int type, int situation, int detail,
|
|
void* data, va_list ap)
|
|
{
|
|
cluster_t* c = va_arg(ap, cluster_t*);
|
|
queue_t* q = va_arg(ap, queue_t*);
|
|
int i;
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_lock(&(c->apiservers[i].lock));
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (c->apiservers[i].ep != NULL)) {
|
|
int rc;
|
|
xce_notify_data_t xdata
|
|
= { .event = NotifyData_data_schedUpdateQueue,
|
|
.data = { .sched_update_queue = &(q->queue) } };
|
|
|
|
rc = notify(c, c->apiservers[i].ep, &xdata);
|
|
if (rc < 0) {
|
|
WARN(
|
|
"%s notify queue change to api server %s failed, rc %d",
|
|
__func__, c->apiservers[i].name, rc);
|
|
}
|
|
}
|
|
mylock_unlock(&(c->apiservers[i].lock));
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
int apiserver_init(cluster_t* c)
|
|
{
|
|
int i;
|
|
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
mylock_init(&(c->apiservers[i].lock));
|
|
c->apiservers[i].name = NULL;
|
|
c->apiservers[i].ep = NULL;
|
|
}
|
|
|
|
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB,
|
|
(hook_fn_t)apiserver_job_hook_fun, NULL);
|
|
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER,
|
|
(hook_fn_t)apiserver_worker_hook_fun, NULL);
|
|
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_RESOURCE,
|
|
(hook_fn_t)apiserver_resource_hook_fun, NULL);
|
|
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE,
|
|
(hook_fn_t)apiserver_queue_hook_fun, NULL);
|
|
|
|
return 0;
|
|
}
|
|
|
|
int on_apiserver_join(cluster_t* c, endpoint_t* ep)
|
|
{
|
|
xce_apiserver_push_configs(c, ep);
|
|
xce_apiserver_push_workers(c, ep);
|
|
xce_apiserver_push_jobs(c, ep);
|
|
|
|
return 0;
|
|
}
|
|
|
|
int handle_apiserver_leave(cluster_t* c, endpoint_t* ep)
|
|
{
|
|
ep_private_t* priv;
|
|
int i;
|
|
|
|
if (ep == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
priv = (ep_private_t*)ep->data;
|
|
if (priv == NULL) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
i = priv->u.idx;
|
|
if (i < 0) {
|
|
SET_ERROR(EINVAL);
|
|
return -errno;
|
|
}
|
|
|
|
if (c->apiservers[i].name != NULL) {
|
|
c->apiservers[i].ep = NULL;
|
|
}
|
|
|
|
/* FIXME: kill ep? */
|
|
return 0;
|
|
}
|
|
|
|
int add_apiserver(cluster_t* c, char* name, endpoint_t* ep)
|
|
{
|
|
int i;
|
|
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
if ((c->apiservers[i].name != NULL)
|
|
&& (strcmp(c->apiservers[i].name, name) == 0)) {
|
|
if (c->apiservers[i].ep != NULL) {
|
|
/* FIXME: we need to delete 1 */
|
|
}
|
|
c->apiservers[i].ep = ep;
|
|
return i;
|
|
}
|
|
}
|
|
|
|
for (i = 0; i < N_APISERVER_MAX; i++) {
|
|
if (c->apiservers[i].name == NULL) {
|
|
c->apiservers[i].name = makeString(name);
|
|
c->apiservers[i].ep = ep;
|
|
return i;
|
|
}
|
|
}
|
|
|
|
/* better error code */
|
|
return -1;
|
|
}
|