171 lines
4.3 KiB
C
171 lines
4.3 KiB
C
#include <errno.h>
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <unistd.h>
|
|
|
|
#include "controller.capnp.h"
|
|
#include "dataq.h"
|
|
#include "xce/types.h"
|
|
|
|
#include "scheduler.h"
|
|
#include "utils.h"
|
|
|
|
static void enq_pend_update(xce_scheduler_t* sched,
|
|
xce_sched_update_request_t* req)
|
|
{
|
|
mylock_lock(&(sched->lock));
|
|
dataq_enq(sched->pend_updates[0], &(req->link));
|
|
mylock_unlock(&(sched->lock));
|
|
}
|
|
|
|
static int sched_handle_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
|
|
{
|
|
xce_sched_update_request_t* req;
|
|
|
|
TRACE("%s...", __func__);
|
|
TRACE("my cluster name %s", config->clustername);
|
|
|
|
req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
|
|
if (req == NULL) {
|
|
return -ENOMEM;
|
|
}
|
|
|
|
req->op = XCE_SCHED_OP_UPDATE_CONFIG;
|
|
req->u.config = config;
|
|
|
|
enq_pend_update(sched, req);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int sched_handle_jobpush(xce_scheduler_t* sched, xce_jobs_data_t* jobs)
|
|
{
|
|
TRACE("%s...", __func__);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int sched_handle_workerpush(xce_scheduler_t* sched,
|
|
xce_workers_data_t* workers)
|
|
{
|
|
TRACE("%s...", __func__);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int sched_on_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs)
|
|
{
|
|
xce_sched_update_request_t* req;
|
|
|
|
req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
|
|
if (req == NULL) {
|
|
return -ENOMEM;
|
|
}
|
|
|
|
req->op = XCE_SCHED_OP_UPDATE_JOB;
|
|
req->u.jobupdate.njobs = njobs;
|
|
req->u.jobupdate.jobs = jobs;
|
|
|
|
enq_pend_update(sched, req);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int sched_on_workers(xce_scheduler_t* sched, int nworkers,
|
|
xce_worker_t** workers)
|
|
{
|
|
xce_sched_update_request_t* req;
|
|
|
|
req = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
|
|
if (req == NULL) {
|
|
return -ENOMEM;
|
|
}
|
|
|
|
req->op = XCE_SCHED_OP_UPDATE_WORKER;
|
|
req->u.workerupdate.nworkers = nworkers;
|
|
req->u.workerupdate.workers = workers;
|
|
|
|
enq_pend_update(sched, req);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset)
|
|
{
|
|
int rc = 0;
|
|
|
|
TRACE("%s table = %d", __func__, dataset->table);
|
|
|
|
switch (dataset->table) {
|
|
case Data_data_schedjobs:
|
|
rc = sched_on_jobs(sched, dataset->nrows, (xce_job_t**)dataset->rows);
|
|
if (rc >= 0) {
|
|
/* jobs will be put in pend update request queue */
|
|
dataset->nrows = 0;
|
|
dataset->rows = NULL;
|
|
}
|
|
|
|
break;
|
|
case Data_data_schedworkers:
|
|
rc = sched_on_workers(sched, dataset->nrows,
|
|
(xce_worker_t**)dataset->rows);
|
|
if (rc >= 0) {
|
|
/* workers will be put in pend update request queue */
|
|
dataset->nrows = 0;
|
|
dataset->rows = NULL;
|
|
}
|
|
break;
|
|
case Data_data_resources:
|
|
break;
|
|
case Data_data_resvals:
|
|
break;
|
|
}
|
|
|
|
return rc;
|
|
}
|
|
|
|
int scheduler_req_handler(xce_request_t* request, void* args)
|
|
{
|
|
int rc = 0;
|
|
xce_scheduler_t* sched = (xce_scheduler_t*)args;
|
|
|
|
switch (request->opcode) {
|
|
case OPCode_push:
|
|
switch (request->body_which) {
|
|
case Request_body_configpushmsg:
|
|
/* configuration arriving */
|
|
rc = sched_handle_config(
|
|
sched, request->body.configpushmsg->data.conf4sched);
|
|
if (rc >= 0) {
|
|
request->body.configpushmsg->data.conf4sched = NULL;
|
|
}
|
|
break;
|
|
case Request_body_pushmsg:
|
|
/* data arriving */
|
|
switch (request->body.pushmsg->table) {
|
|
case Data_data_schedjobs:
|
|
/* job arriving */
|
|
rc = sched_handle_jobpush(sched,
|
|
request->body.pushmsg->data.jobs);
|
|
break;
|
|
case Data_data_schedworkers:
|
|
/* workers arriving */
|
|
rc = sched_handle_workerpush(
|
|
sched, request->body.pushmsg->data.workers);
|
|
break;
|
|
}
|
|
break;
|
|
case Request_body_datamsg:
|
|
rc = sched_handle_dataset(sched, request->body.datamsg);
|
|
break;
|
|
default:
|
|
rc = -EINVAL;
|
|
break;
|
|
}
|
|
break;
|
|
default:
|
|
rc = -EINVAL;
|
|
break;
|
|
}
|
|
return rc;
|
|
}
|