wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity
myxce/src/controller/testsuite/test.c
Rongsong Shen 6d4d8eb81d feat: rename delaymgr precision→precision_ms, replace sleep() with nanosleep() (P0-2b) 
- Rename `precision` field to `precision_ms` in delaymgr_t struct (delay.h)
- Replace `sleep(mgr->precision)` with `nanosleep()` for 10ms granularity (delay.c)
- Add `#include <time.h>` to delay.c (already in delay.h)
- Update delaymgr_init() to store precision_ms parameter
- Update all call sites: cluster.c (2→200ms), controller/testsuite/test.c (2→2000ms),
  lib/testsuite/test_delay.c (1→1000ms)
- Add TDD test test_delaymgr_precision_ms_field verifying field exists and stores correctly
- delay_task() delay parameter remains in seconds; only sleep granularity changes
2026-05-07 09:08:03 +00:00

2498 lines
64 KiB
C
Raw Blame History

#include <argp.h>
#include <errno.h>
#include <getopt.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
#include <readline/history.h>
#include <readline/readline.h>
#include <cjson/cJSON.h>
#include <cjson/s2j.h>
#include "client.h"
#include "controller.capnp.h"
#include "delay.h"
#include "idgen.h"
#include "preason.h"
#include "xce/types.h"
typedef int (*cmd_func_t)(int, char*[]);
typedef struct {
char* name;
cmd_func_t func;
char* doc;
} command_t;
typedef struct {
char* key;
char* value;
char* defval;
} setting_t;
typedef struct _command_ctx {
struct _command_ctx* parent;
char* prompt;
setting_t* settings;
command_t* commands;
xcecli_ctx_t ctx;
} command_ctx_t;
typedef struct {
uint64_t jobid;
int pid;
time_t ts;
} run_job_t;
typedef struct {
uint64_t jobid;
uint32_t taskid;
int clientidx;
int pid;
time_t ts;
} run_task_t;
typedef struct {
char* host;
int ntasks;
} host_task_t;
typedef struct {
char* PATH;
} sub_envs_t;
typedef struct {
sub_envs_t sub_env_vars;
} exdata_t;
#define MAX_RUN_JOBS (20)
#define MAX_TASK_HOSTS (5)
#define MAX_RUN_TASKS (50)
#define RUSAGE_UPDATE_INTERVAL (60)
#define MAX_LINE_LEN (40960)
#define S2J_JSON_SET_var_string_ELEMENT(to_json, from_struct, _element) \
if (((from_struct) != NULL) && (((from_struct)->_element) != NULL)) { \
cJSON_AddStringToObject((to_json), #_element, \
(from_struct)->_element); \
}
static run_job_t run_jobs[MAX_RUN_JOBS];
static run_task_t run_tasks[MAX_RUN_TASKS];
static int myreq_handler(xce_request_t*, void*);
extern int cmd_help(int, char*[]);
extern int cmd_up(int, char*[]);
extern int cmd_worker(int, char*[]);
extern int cmd_scheduler(int, char*[]);
extern int cmd_client(int, char*[]);
extern int cmd_quit(int, char*[]);
extern int cmd_hello(int, char*[]);
extern int cmd_shutdown(int, char*[]);
extern int cmd_set(int, char*[]);
extern int cmd_reslist(int, char*[]);
extern int cmd_join_worker(int, char*[]);
extern int cmd_join_sched(int, char*[]);
extern int cmd_workerlist(int, char*[]);
extern int cmd_joblist(int, char*[]);
extern int cmd_jobrun(int, char*[]);
extern int cmd_jobsub(int, char*[]);
extern int cmd_getconf(int, char*[]);
extern int cmd_jobfinish(int, char*[]);
extern int cmd_resupdate(int, char*[]);
extern int cmd_resval(int, char*[]);
extern int cmd_jobkill(int, char*[]);
extern int cmd_jobckpt(int, char*[]);
extern int cmd_qctrl(int, char*[]);
extern int cmd_runtasks(int, char*[]);
extern int cmd_setpend(int, char*[]);
extern int cmd_jobinfo(int, char*[]);
static int worker_house_keeping(void*);
static idgen_ctx_t idctx;
delaymgr_t delaymgr;
static setting_t settings[] = {
{ (char*)"host", (char*)NULL, (char*)"localhost" },
{ (char*)"port", (char*)NULL, (char*)"9990" },
{ (char*)"secret", (char*)NULL, (char*)"keepSecret" },
{ (char*)"myname", (char*)NULL, (char*)"localhost" },
{ (char*)NULL, (char*)NULL, (char*)NULL }
};
static command_t top_commands[] = {
{ "help", cmd_help, "Display this text" },
{ "worker", cmd_worker, "In worker context" },
{ "scheduler", cmd_scheduler, "In scheduler context" },
{ "client", cmd_client, "In client context" },
{ "set", cmd_set, "Show/Set settings" },
{ "quit", cmd_quit, "Quit xce test" },
{ (char*)NULL, (cmd_func_t)NULL, (char*)NULL }
};
static command_ctx_t topctx = { .parent = NULL,
.prompt = "xce",
.settings = settings,
.commands = top_commands };
static command_t worker_commands[] = {
{ "..", cmd_up, "Goto upper context" },
{ "help", cmd_help, "Display this text" },
{ "quit", cmd_quit, "Quit xce test" },
{ "join", cmd_join_worker, "Join the cluster" },
{ "jfinish", cmd_jobfinish, "tell controller jobfinish" },
{ "resupdate", cmd_resupdate, "update res vals" },
{ (char*)NULL, (cmd_func_t)NULL, (char*)NULL }
};
static command_ctx_t workerctx = { .parent = &topctx,
.prompt = "worker",
.settings = settings,
.commands = worker_commands };
static command_t client_commands[] = {
{ "..", cmd_up, "Goto upper context" },
{ "hello", cmd_hello, "Say hello to xce controller" },
{ "reslist", cmd_reslist, "Display cluster resources" },
{ "workerlist", cmd_workerlist, "list workers" },
{ "qctrl", cmd_qctrl, "queue control" },
{ "joblist", cmd_joblist, "list jobs" },
{ "shutdown", cmd_shutdown, "shutdown xce controller" },
{ "jobrun", cmd_jobrun, "run a job" },
{ "jobsub", cmd_jobsub, "submit a job" },
{ "jobkill", cmd_jobkill, "signal a job" },
{ "jobckpt", cmd_jobckpt, "checkpoint a job" },
{ "jobinfo", cmd_jobinfo, "job information" },
{ "resval", cmd_resval, "get value of res" },
{ "taskrun", cmd_runtasks, "run tasks" },
{ "help", cmd_help, "Display this text" },
{ "quit", cmd_quit, "Quit xce test" },
{ (char*)NULL, (cmd_func_t)NULL, (char*)NULL }
};
static command_ctx_t clientctx = { .parent = &topctx,
.prompt = "client",
.settings = settings,
.commands = client_commands };
static command_t scheduler_commands[] = {
{ "..", cmd_up, "Goto upper context" },
{ "help", cmd_help, "Display this text" },
{ "quit", cmd_quit, "Quit xce test" },
{ "join", cmd_join_sched, "Join the cluster" },
{ "getconf", cmd_getconf, "Get conf of cluster" },
{ "setpend", cmd_setpend, "set job pend reason" },
{ (char*)NULL, (cmd_func_t)NULL, (char*)NULL }
};
static command_ctx_t schedulerctx = { .parent = &topctx,
.prompt = "scheduler",
.settings = settings,
.commands = scheduler_commands };
static command_ctx_t* active_ctx = &topctx;
static char prompt[256];
static int break_out = 0;
static char doc[] = "xce_test";
static char args_doc[] = "";
const char* argp_program_version
= MK_VERSION("xce v0.0.1-", STR(VERSION_COMMIT_ID));
const char* argp_program_bug_address = BUG_EMAIL;
struct arguments {
int verbose;
};
static struct argp_option options[] = {
{ "verbose", 'v', 0, 0, "Produce verbose logs" },
{ 0 }
};
static error_t parse_opt(int key, char* arg, struct argp_state* state)
{
struct arguments* arguments = state->input;
switch (key) {
case 'v':
arguments->verbose++;
break;
default:
return ARGP_ERR_UNKNOWN;
}
return 0;
}
static struct argp argp = { options, parse_opt, args_doc, doc };
void sigbreak(int signo) { break_out = 1; }
static char* get_setting(char* key)
{
int i;
for (i = 0; active_ctx->settings[i].key; i++) {
if (strcmp(key, active_ctx->settings[i].key) != 0) {
continue;
}
if (active_ctx->settings[i].value != NULL) {
return active_ctx->settings[i].value;
}
return active_ctx->settings[i].defval;
}
printf("Key `%s' is not in settings\n", key);
return NULL;
}
static xce_client_config_t config;
int enter_ctx(command_ctx_t* ctx)
{
config.nmasters = 1;
config.masters[0] = get_setting("host");
config.master_ports[0] = atoi(get_setting("port"));
config.from = get_setting("myname");
config.key = get_setting("secret");
xcecli_ctx_init(&(ctx->ctx), &config);
xcecli_set_handler(ctx->ctx, &myreq_handler, NULL);
xcecli_start(ctx->ctx);
return 0;
}
int leave_ctx(command_ctx_t* ctx)
{
if (ctx == &topctx) {
return 0;
}
if (ctx == &workerctx) {
delaymgr_fini(&delaymgr);
}
xcecli_wakeup(active_ctx->ctx, (int)-EINTR);
xcecli_stop(ctx->ctx);
xcecli_ctx_fini(ctx->ctx);
return 0;
}
static int fill_prompt(command_ctx_t* ctx, char* buf)
{
if (ctx->parent != NULL) {
fill_prompt(ctx->parent, buf);
strcat(buf, "/");
}
strcat(buf, ctx->prompt);
return strlen(buf);
}
command_t* find_command(char* name)
{
int i;
for (i = 0; active_ctx->commands[i].name; i++) {
if (strcmp(active_ctx->commands[i].name, name) == 0) {
return &(active_ctx->commands[i]);
}
}
return NULL;
}
/* Execute a command line. */
int execute_line(char* line)
{
int rc;
command_t* command;
int argc;
char** argv;
argc = parse_argv(line, &argv);
if (argc < 0) {
SET_ERROR(-argc);
perror("execute_line");
return -1;
}
if ((argv == NULL) || (argv[0] == NULL)) {
return 0;
}
command = find_command(argv[0]);
if (!command) {
fprintf(stderr, "%s: No such command for xce test.\n", argv[0]);
free_argv(argv, argc);
return (-1);
}
rc = (command->func)(argc, argv);
free_argv(argv, argc);
return rc;
}
int cmd_help(int argc, char* argv[])
{
int i;
int rc = 0;
if (argc > 1) {
char* cmd = argv[1];
for (i = 0; active_ctx->commands[i].name != NULL; i++) {
if (strcmp(argv[1], active_ctx->commands[i].name) == 0) {
printf("%s\t\t%s.\n", active_ctx->commands[i].name,
active_ctx->commands[i].doc);
return 0;
}
}
printf("No commands matchs `%s'. Possibilities are:\n", cmd);
rc = -1;
}
for (i = 0; active_ctx->commands[i].name != NULL; i++) {
printf("\t%-12s\t%s\n", active_ctx->commands[i].name,
active_ctx->commands[i].doc);
}
return rc;
}
int cmd_up(int argc, char* argv[])
{
if (active_ctx->parent != NULL) {
int rc;
if ((rc = leave_ctx(active_ctx)) < 0) {
return rc;
}
active_ctx = active_ctx->parent;
}
return 0;
}
int cmd_worker(int argc, char* argv[])
{
int rc;
if ((rc = enter_ctx(&workerctx)) < 0) {
return rc;
}
active_ctx = &workerctx;
delaymgr_init(&delaymgr, 2000 /* precision_ms */, NULL);
delaymgr_start(&delaymgr);
delay_task(&delaymgr, 5, -1, (delay_task_entry_fn_t)worker_house_keeping,
&workerctx);
return 0;
}
int cmd_scheduler(int argc, char* argv[])
{
int rc;
if ((rc = enter_ctx(&schedulerctx)) < 0) {
return rc;
}
active_ctx = &schedulerctx;
return 0;
}
int cmd_client(int argc, char* argv[])
{
int rc;
if ((rc = enter_ctx(&clientctx)) < 0) {
return rc;
}
active_ctx = &clientctx;
return 0;
}
static int done = 0;
int cmd_quit(int argc, char* argv[])
{
leave_ctx(active_ctx);
done = 1;
return 0;
}
int cmd_hello(int argc, char* argv[])
{
int rc = 0;
int opt;
char* msg = "Greeting from xce client";
int count = 1;
optind = 1;
while ((opt = getopt(argc, argv, "c:m:")) != -1) {
switch (opt) {
case 'c':
count = atoi(optarg);
break;
case 'm':
msg = optarg;
break;
default:
fprintf(stderr, " hello [-c count] [-w wait_usec] [-m message]\n");
return -1;
break;
}
}
break_out = 0;
while ((!break_out) && (count > 0)) {
rc = xcecli_hello(active_ctx->ctx, msg);
count--;
}
return rc;
}
int cmd_shutdown(int argc, char* argv[])
{
int rc;
rc = xcecli_shutdown(active_ctx->ctx);
/* leave client context */
if (rc >= 0) {
execute_line("..");
}
return rc;
}
int cmd_reslist(int argc, char* argv[])
{
int rc, i;
xce_resource_t** res;
rc = xcecli_reslist(active_ctx->ctx, &res);
if (rc < 0) {
SET_ERROR(-rc);
perror("xce_reslist");
return rc;
}
printf("%10s%10s%10s%64s\n", "Name", "Flags", "Interval", "Description");
for (i = 0; i < rc; i++) {
printf("%10s%10x%10d%64s\n", res[i]->name, res[i]->flags,
res[i]->interval, res[i]->desc);
}
free_Resource_list(rc, res);
return 0;
}
static char* worker_status_str(uint32_t status)
{
if ((status & XCE_WORKER_UNAVAIL) != 0) {
return "UNAVAIL";
} else if ((status & XCE_WORKER_LOCKEDU) != 0) {
return "LOCKU";
} else if ((status & XCE_WORKER_LOCKEDW) != 0) {
return "LOCKW";
} else if ((status & XCE_WORKER_BUSY) != 0) {
return "BUSY";
} else if (status == 0) {
return "OK";
}
return "UNKOWN";
}
int cmd_workerlist(int argc, char* argv[])
{
int rc, i;
xce_worker_t** workers;
rc = xcecli_workerlist(active_ctx->ctx, &workers);
if (rc < 0) {
SET_ERROR(-rc);
perror("xce_workerlist");
return rc;
}
printf("%-20s%-8s%-25s%-25s%12s%12s%12s%12s%12s%12s\n", "HostName",
"Status", "HostType", "HostModel", "CPUfactor", "MaxCPU", "MaxMem",
"MaxSwap", "MaxTmp", "NDisks");
for (i = 0; i < rc; i++) {
printf("%-20s%-8s%-25s%-25s%12f%12d%12f%12f%12f%12d\n",
workers[i]->name, worker_status_str(workers[i]->status),
workers[i]->attrs->hosttype, workers[i]->attrs->hostmodel,
workers[i]->attrs->cpufactor, workers[i]->attrs->maxcpus,
workers[i]->attrs->maxmem, workers[i]->attrs->maxswap,
workers[i]->attrs->maxtmp, workers[i]->attrs->ndisks);
}
free_Worker_list(rc, workers);
return 0;
}
static char* job_state_str(int state)
{
if ((state & JOB_STAT_PEND) != 0) {
if ((state & JOB_STAT_PSUSP) != 0) {
return "PSUSP";
}
return "PEND";
} else if ((state & JOB_STAT_RUN) != 0) {
if ((state & JOB_STAT_USUSP) != 0) {
return "USUSP";
} else if ((state & JOB_STAT_SSUSP) != 0) {
return "SSUSP";
}
return "RUN";
} else if ((state & JOB_STAT_EXIT) != 0) {
return "EXIT";
} else if ((state & JOB_STAT_DONE) != 0) {
return "DONE";
} else if ((state & JOB_STAT_ERR) != 0) {
return "ERROR";
}
return "UNKNOWN";
}
int cmd_joblist(int argc, char* argv[])
{
int rc, i;
xce_job_t** jobs;
rc = xcecli_joblist(active_ctx->ctx, &jobs);
if (rc < 0) {
SET_ERROR(-rc);
perror("xce_joblist");
return rc;
}
printf("%-26s%-10s%-10s%-8s%-16s%-16s%-16s%10s%30s\n", "JobId", "User",
"Queue", "State", "SubmitTime", "StartTime", "FinishTime", "NTasks",
"Command");
for (i = 0; i < rc; i++) {
char buf[128];
fmt_jobid(buf, sizeof(buf) - 1, jobs[i]->jobid);
printf("%-26s%-10s%-10s%-8s%-16ld%-16ld%-16ld%4d,%-5d%30s\n", buf,
jobs[i]->submit->user, jobs[i]->submit->queue,
job_state_str(jobs[i]->state), jobs[i]->queue_time,
jobs[i]->start_time, jobs[i]->finish_time,
jobs[i]->submit->mintasks, jobs[i]->submit->maxtasks,
jobs[i]->submit->start_info->cmdline);
}
for (i = 0; i < rc; i++) {
free_Job_ptr(&(jobs[i]));
}
free(jobs);
return 0;
}
static void print_jobinfo(xce_job_t* job)
{
char idstring[128];
if (job == NULL) {
return;
}
fmt_jobid(idstring, sizeof(idstring) - 1, job->jobid);
printf("JobID: %s\n", idstring);
printf("state: %s\n", job_state_str(job->state));
printf("Queue: %s\n", job->submit->queue);
printf("User: %s\n", job->submit->user);
printf("Slots: %d,%d\n", job->submit->mintasks, job->submit->maxtasks);
printf("Command: %s\n", job->submit->start_info->cmdline);
if (job->pend_reasons != NULL) {
printf("Pend Reason:\n");
if (job->pend_reasons->main_reason != 0) {
printf("\tmain Reason: %d\n", job->pend_reasons->main_reason);
} else {
if (job->pend_reasons->nhost_reasons > 0) {
int i;
printf("\tHost Reasons:\n");
for (i = 0; i < job->pend_reasons->nhost_reasons; i++) {
printf("\t\t(%s, %d)\n",
job->pend_reasons->host_based_reasons[i]->host,
job->pend_reasons->host_based_reasons[i]->reason);
}
}
if (job->pend_reasons->nres_reasons > 0) {
int i;
printf("\tResource Reasons:\n");
for (i = 0; i < job->pend_reasons->nres_reasons; i++) {
printf("\t\t(%s, %s, %d)\n",
job->pend_reasons->res_based_reasons[i]->resource,
job->pend_reasons->res_based_reasons[i]->host,
job->pend_reasons->res_based_reasons[i]->reason);
}
}
}
}
}
int cmd_jobinfo(int argc, char* argv[])
{
int opt;
optind = 1;
while ((opt = getopt(argc, argv, "j:")) != -1) {
uint64_t jobid;
xce_job_t* job;
int rc = 0;
switch (opt) {
case 'j':
parse_jobid(&jobid, optarg);
if (jobid > 0) {
rc = xcecli_jobinfo(active_ctx->ctx, jobid, &job);
if (rc < 0) {
perror("xcecli_jobinfo");
continue;
}
print_jobinfo(job);
free_Job_ptr(&job);
}
break;
}
}
return 0;
}
static int fill_job_alloc(xce_job_alloc_t* jalloc, int nhosts,
host_task_t* hosts)
{
xce_host_alloc_t** halloc = NULL;
int i;
halloc = (xce_host_alloc_t**)calloc(nhosts, sizeof(xce_host_alloc_t*));
if (halloc == NULL) {
return -ENOMEM;
}
for (i = 0; i < nhosts; i++) {
xce_alloc_t* alloc;
xce_alloc_t** allocs;
alloc = (xce_alloc_t*)calloc(1, sizeof(*alloc));
allocs = (xce_alloc_t**)calloc(1, sizeof(*allocs));
halloc[i] = (xce_host_alloc_t*)calloc(1, sizeof(xce_host_alloc_t));
if ((halloc[i] == NULL) || (alloc == NULL) || (allocs == NULL)) {
/* FIXME: handle error */
return -ENOMEM;
}
alloc->resname = "slots";
alloc->reserve = "";
alloc->instid = -1;
alloc->value = (float)hosts[i].ntasks;
alloc->from = time(NULL);
alloc->duration = -1;
allocs[0] = alloc;
halloc[i]->nallocs = 1;
halloc[i]->allocs = allocs;
halloc[i]->host = hosts[i].host;
}
jalloc->nhosts = nhosts;
jalloc->allocs = halloc;
return 0;
}
int cmd_jobrun(int argc, char* argv[])
{
int rc;
int opt;
uint64_t jobid = 0;
host_task_t hosts[MAX_TASK_HOSTS];
xce_job_alloc_t jalloc;
xce_job_dispatch_t jdispatch
= { .njobs = 1, .allocs = (xce_job_alloc_t*[]) { &jalloc } };
int nhosts = 0;
char* host;
int count;
optind = 1;
while ((opt = getopt(argc, argv, "m:j:")) != -1) {
switch (opt) {
case 'm':
if (nhosts >= MAX_TASK_HOSTS) {
fprintf(stderr,
"too many run hosts, we only support max %d hosts\n",
MAX_TASK_HOSTS);
return -1;
}
if (strchr(optarg, (int)':') != NULL) {
char* p;
char buf[256];
p = strchr(optarg, (int)':');
count = atoi(p + 1);
memset(buf, 0x0, sizeof(buf));
strncpy(buf, optarg, (int)(p - optarg));
host = strdup(buf);
} else {
host = strdup(optarg);
count = 1;
}
hosts[nhosts].host = host;
hosts[nhosts].ntasks = count;
nhosts++;
break;
case 'j':
parse_jobid(&jobid, optarg);
if (jobid == 0) {
printf("you need to specify a positive jobid with -j");
return -1;
}
break;
}
}
if (jobid == 0) {
printf("no jobid specified");
return -1;
}
fill_job_alloc(&jalloc, nhosts, &hosts[0]);
jalloc.jobid = jobid;
rc = xcecli_jobdispatch(active_ctx->ctx, &jdispatch);
if (rc < 0) {
SET_ERROR(-rc);
perror("xce_jobdispatch");
return rc;
}
#if 0
printf("job %s will be started on host: %s\n",
argv[1], halloc.host);
#endif
return 0;
}
static int build_job_envs(xce_job_submit_t* submit)
{
exdata_t exdata = { .sub_env_vars = { .PATH = getenv("PATH") } };
s2j_create_json_obj(json_exdata);
s2j_create_json_obj(json_envs);
s2j_json_set_basic_element(json_envs, &(exdata.sub_env_vars), var_string,
PATH);
cJSON_AddItemToObject(json_exdata, "sub_env_vars", json_envs);
submit->submit->start_info->exdata = cJSON_Print(json_exdata);
cJSON_Delete(json_exdata);
return 0;
}
static int build_job_exdata(xce_job_submit_t* submit)
{
build_job_envs(submit);
return 0;
}
int cmd_jobsub(int argc, char* argv[])
{
int rc = 0;
job_submit_data_t submit;
xce_job_submit_t jobsubmit = { .submit = &submit };
int opt;
uint64_t jobid;
uint32_t min = 1, max = 1;
uint32_t arsize = 1;
char jobidStr[128];
optind = 1;
memset(&submit, 0x0, sizeof(submit));
submit.start_info
= (job_startinfo_t*)calloc(1, sizeof(*(submit.start_info)));
if (submit.start_info == NULL) {
perror("calloc");
return -ENOMEM;
}
while ((opt = getopt(argc, argv, "u:q:R:c:x:n:a:")) != -1) {
switch (opt) {
case 'u':
submit.user = strdup(optarg);
break;
case 'q':
submit.queue = strdup(optarg);
break;
case 'R':
submit.resreq = strdup(optarg);
break;
case 'c':
submit.start_info->cmdline = strdup(optarg);
break;
case 'x':
submit.start_info->exdata = strdup(optarg);
break;
case 'n':
if (strchr(optarg, (int)',') != NULL) {
char buf[64];
char* p = strchr(optarg, (int)',');
max = atoi(p + 1);
memset(buf, 0x0, sizeof(buf));
strncpy(buf, optarg, (int)(p - optarg));
min = atoi(buf);
if (max == 0) {
printf("Incorrect setting of max tasks count, set to same "
"as min task value\n");
max = min;
}
} else {
min = atoi(optarg);
max = min;
}
break;
case 'a':
arsize = atoi(optarg);
if (arsize <= 0) {
arsize = 1;
}
break;
default:
printf(
"jobsub -u <user> -q <queue> -R <resreq> -c <command line>\n");
return -EINVAL;
}
}
if (min == 0) {
printf("Incorrect setting of min task count, set to 1 as default\n");
min = 1;
}
if (NULLSTR(submit.user)) {
printf("you need to specify job user with -u\n");
return -EINVAL;
}
if (NULLSTR(submit.queue)) {
printf("you need to specify job queue with -q\n");
return -EINVAL;
}
if (NULLSTR(submit.start_info->cmdline)) {
printf("you need to specify job command with -c\n");
return -EINVAL;
}
if (NULLSTR(submit.resreq)) {
submit.resreq = strdup("type == any");
}
submit.mintasks = min;
submit.maxtasks = max;
idgen(&idctx, 0, 0, arsize, &(jobsubmit.jobid));
if (NULLSTR(submit.start_info->execuser)) {
submit.start_info->execuser = strdup(submit.user);
}
if (NULLSTR(submit.start_info->exechome)) {
submit.start_info->exechome = strdup("/tmp");
}
if (NULLSTR(submit.start_info->execcwd)) {
submit.start_info->execcwd = strdup("/tmp");
}
build_job_exdata(&jobsubmit);
jobsubmit.array_size = arsize;
rc = xcecli_jobsubmit(active_ctx->ctx, &jobsubmit, &jobid);
if (rc < 0) {
SET_ERROR(-rc);
perror("xcecli_jobsubmit");
return rc;
}
FREEUP(submit.user);
FREEUP(submit.queue);
FREEUP(submit.resreq);
FREEUP(submit.start_info->cmdline);
FREEUP(submit.start_info->exdata);
FREEUP(submit.start_info);
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, array_jobid(jobid));
printf("jobsubmit ok: JobId: %s\n", jobidStr);
return 0;
}
int cmd_resval(int argc, char* argv[])
{
char* host = NULL;
char* res = NULL;
xce_resval_t** vals = NULL;
int opt;
int rc;
optind = 1;
while ((opt = getopt(argc, argv, "n:m:")) != -1) {
switch (opt) {
case 'n':
res = strdup(optarg);
break;
case 'm':
host = strdup(optarg);
break;
default:
fprintf(stderr, "resval -n resname [-m host]\n");
FREEUP(host);
FREEUP(res);
return -1;
}
}
rc = xcecli_resval(active_ctx->ctx, host, res, &vals);
if (rc > 0) {
int i;
for (i = 0; i < rc; i++) {
xce_resval_t* val = vals[i];
printf("resource %s@%s:\n", val->name,
(val->host == NULL) ? "cluster" : val->host);
printf("value: ");
switch (val->val->which) {
case Value_u_svalue:
printf("%s\n", val->val->u.svalue);
break;
case Value_u_fvalue:
printf("%f\n", val->val->u.fvalue);
break;
case Value_u_bvalue:
printf("%s\n", (val->val->u.bvalue == 0) ? "false" : "true");
break;
default:
printf(" <unknown type>\n");
break;
}
free_ResourceVal_ptr(&val);
}
FREEUP(vals);
} else {
printf("no resource value found in given host\n");
}
FREEUP(host);
FREEUP(res);
return 0;
}
int cmd_jobkill(int argc, char* argv[])
{
int signo;
uint64_t jobid = 0;
int opt;
int rc;
xce_jobctrl_t jobctrl;
xce_signal_t siginfo;
optind = 1;
signo = SIGTERM;
while ((opt = getopt(argc, argv, "s:j:")) != -1) {
switch (opt) {
case 's':
signo = atoi(optarg);
break;
case 'j':
parse_jobid(&jobid, optarg);
break;
default:
break;
}
}
if (jobid == 0) {
fprintf(stderr, "you need to specify jobid with -j\n");
return -1;
}
jobctrl.action = JobControl_u_siginfo;
jobctrl.jobid = jobid;
jobctrl.u.siginfo = &siginfo;
siginfo.signo = signo;
siginfo.force = 1;
rc = xcecli_jobctrl(active_ctx->ctx, &jobctrl);
if (rc < 0) {
perror("xcecli_jobctrl");
return rc;
}
printf("Job %lu signaled\n", jobid);
return 0;
}
int cmd_jobckpt(int argc, char* argv[])
{
uint64_t jobid = 0;
int opt;
int rc;
xce_jobctrl_t jobctrl;
xce_ckpt_options_t ckptoption;
optind = 1;
ckptoption.kill = 0;
while ((opt = getopt(argc, argv, "kj:m:")) != -1) {
switch (opt) {
case 'k':
ckptoption.kill = 1;
break;
case 'j':
parse_jobid(&jobid, optarg);
break;
case 'm':
ckptoption.method = optarg;
break;
default:
break;
}
}
if (jobid == 0) {
fprintf(stderr, "you need to specify jobid with -j\n");
return -1;
}
if (ckptoption.method == NULL) {
ckptoption.method = "default";
}
jobctrl.action = JobControl_u_ckptoption;
jobctrl.jobid = jobid;
jobctrl.u.ckptoption = &ckptoption;
rc = xcecli_jobctrl(active_ctx->ctx, &jobctrl);
if (rc < 0) {
perror("xcecli_jobctrl");
return rc;
}
printf("Job %lu checkpointed\n", jobid);
return 0;
}
int cmd_qctrl(int argc, char* argv[])
{
int opt;
int rc;
xce_queuectrl_t queuectrl;
int action;
char* queue = NULL;
optind = 1;
while ((opt = getopt(argc, argv, "aiociq:")) != -1) {
switch (opt) {
case 'a':
action = QUEUE_ACTIVE;
break;
case 'i':
action = QUEUE_INACTIVE;
break;
case 'o':
action = QUEUE_OPEN;
break;
case 'c':
action = QUEUE_CLOSE;
break;
case 'q':
queue = strdup(optarg);
break;
default:
break;
}
}
if (queue == NULL) {
fprintf(stderr, "you need to specify queue name with -q\n");
return -1;
}
queuectrl.action = action;
queuectrl.queue = queue;
rc = xcecli_queuectrl(active_ctx->ctx, &queuectrl);
if (rc < 0) {
perror("xcecli_queuectrl");
return rc;
}
printf("Operation on queue %s done\n", queue);
free(queue);
return 0;
}
int cmd_runtasks(int argc, char* argv[])
{
xce_taskhost_t hosts[MAX_TASK_HOSTS];
size_t nhosts = 0;
xce_taskrun_t taskrun;
int opt;
int rc;
int count;
int totaltasks = 0;
char* user = NULL;
char* cmdline = NULL;
char* dir = NULL;
char* exdata = NULL;
char* host = NULL;
char* launch_type = NULL;
int i;
uint64_t jobid = 0;
optind = 1;
while ((opt = getopt(argc, argv, "m:u:c:x:d:j:t:")) != -1) {
switch (opt) {
case 'm':
if (nhosts >= MAX_TASK_HOSTS) {
fprintf(stderr,
"too many run hosts, just support %d run hosts\n",
MAX_TASK_HOSTS);
return -1;
}
if (strchr(optarg, (int)':') != NULL) {
char* p;
char buf[256];
p = strchr(optarg, (int)':');
count = atoi(p + 1);
memset(buf, 0x0, sizeof(buf));
strncpy(buf, optarg, (int)(p - optarg));
host = strdup(buf);
} else {
host = strdup(optarg);
count = 1;
}
hosts[nhosts].host = host;
hosts[nhosts].ntasks = count;
nhosts++;
totaltasks += count;
break;
case 'u':
if (user != NULL) {
free(user);
}
user = strdup(optarg);
break;
case 'c':
if (cmdline != NULL) {
free(cmdline);
}
cmdline = strdup(optarg);
break;
case 'x':
if (exdata != NULL) {
free(exdata);
}
exdata = strdup(optarg);
break;
case 'd':
if (dir != NULL) {
free(dir);
}
dir = strdup(optarg);
break;
case 'j':
parse_jobid(&jobid, optarg);
break;
case 't':
launch_type = strdup(optarg);
break;
default:
break;
}
}
for (i = 0; i < nhosts; i++) {
hosts[i].exechome = dir;
hosts[i].execcwd = dir;
hosts[i].execuser = user;
hosts[i].exdata = exdata;
hosts[i].cmdline = cmdline;
hosts[i].launch_type = launch_type;
if (hosts[i].launch_type == NULL) {
if (totaltasks > 1) {
hosts[i].launch_type = strdup("mpi");
} else {
hosts[i].launch_type = strdup("sequentail");
}
}
}
taskrun.taskhosts
= (xce_taskhost_t**)calloc(nhosts, sizeof(xce_taskhost_t*));
taskrun.clientidx = -1;
if (taskrun.taskhosts != NULL) {
for (i = 0; i < nhosts; i++) {
taskrun.taskhosts[i] = &hosts[i];
}
taskrun.nhosts = nhosts;
taskrun.jobid = jobid;
rc = xcecli_runtask(active_ctx->ctx, &taskrun);
} else {
rc = -ENOMEM;
}
if (rc < 0) {
errno = -rc;
perror("xcecli_runtask");
} else {
printf("run task ok\n");
}
return rc;
}
static int add_pendreason(xce_job_preason_t* preason, char* s)
{
char s1[MAX_LINE_LEN];
char* pcode = NULL;
char* phost = NULL;
char* pres = NULL;
char* p;
int code;
if ((preason == NULL) || (s == NULL)) {
fprintf(stderr, "bad parameter\n");
return -1;
}
if ((p = strrchr(s, (int)':')) == NULL) {
pcode = s;
} else {
pcode = p + 1;
memset(s1, 0x0, sizeof(s1));
strncpy(s1, s, p - s);
if ((p = strrchr(s1, (int)':')) == NULL) {
phost = s1;
} else {
*p = 0x0;
phost = p + 1;
pres = s1;
}
}
code = atoi(pcode);
return add_job_preason(preason, code, phost, pres);
}
int cmd_setpend(int argc, char* argv[])
{
int opt;
xce_job_preason_t jreasons;
xce_pendreason_t preason = { .reasons = &jreasons };
xce_preason_set_t reasons
= { .nreasons = 1, .reasons = (xce_pendreason_t*[]) { &preason } };
int rc = 0;
optind = 1;
while ((opt = getopt(argc, argv, "j:p:")) != -1) {
int rc = 0;
switch (opt) {
case 'j':
parse_jobid(&(preason.jobid), optarg);
break;
case 'p':
rc = add_pendreason(&jreasons, optarg);
if (rc < 0) {
perror("add_pendreason");
return rc;
}
break;
}
}
if (preason.jobid == 0) {
fprintf(stderr, "no or incorrect jobid specified\n");
return -1;
}
if ((jreasons.main_reason == 0) && (jreasons.nhost_reasons == 0)
&& (jreasons.nres_reasons == 0)) {
fprintf(stderr, "no or incorrect pend reason set\n");
return -1;
}
rc = xcecli_set_pend_reason(active_ctx->ctx, &reasons);
if (rc < 0) {
perror("xcecli_set_pend_reason");
erase_job_preason(&jreasons);
return rc;
}
printf("ok\n");
return 0;
}
int cmd_join_worker(int argc, char* argv[])
{
int rc;
int role;
int opt;
hostattrs_t attrs = { .cpufactor = 1.0,
.maxcpus = 4,
.maxmem = 64,
.maxswap = 1024,
.maxtmp = 2048,
.ndisks = 1 };
role = HostRole_worker;
optind = 1;
attrs.hostmodel = "Intel64";
attrs.hosttype = "LINUX64";
while ((opt = getopt(argc, argv, "f:c:m:s:t:d:")) != -1) {
switch (opt) {
case 'f':
attrs.cpufactor = atof(optarg);
break;
case 'c':
attrs.maxcpus = atoi(optarg);
break;
case 'm':
attrs.maxmem = atoi(optarg);
break;
case 's':
attrs.maxswap = atoi(optarg);
break;
case 't':
attrs.maxtmp = atoi(optarg);
break;
case 'd':
attrs.ndisks = atoi(optarg);
break;
default:
printf(" join [-f cpufactor] [-c maxcpus] [-m maxmem] [-s maxswap] "
"[-t maxtmp] [-d ndisks]\n");
return -EINVAL;
}
}
rc = xcecli_join(active_ctx->ctx, role, &attrs);
if (rc < 0) {
SET_ERROR(-rc);
perror("xce_join");
return rc;
}
printf("host join ok\n");
return 0;
}
int cmd_jobfinish(int argc, char* argv[])
{
int rc;
int opt;
uint64_t jobid = 0;
int code = 0;
xce_jobfinish_t finish;
while ((opt = getopt(argc, argv, "j:c:")) != -1) {
switch (opt) {
case 'j':
parse_jobid(&jobid, optarg);
break;
case 'c':
code = atoi(optarg);
break;
default:
printf(" jfinish -j jobid [-c exitcode]\n");
return -EINVAL;
}
}
if (jobid == 0) {
printf(" you need specify a positive jobid\n");
return -EINVAL;
}
finish.jobid = jobid;
finish.exit_code = code;
rc = xcecli_notify_jobfinish(active_ctx->ctx, &finish);
if (rc < 0) {
SET_ERROR(-rc);
perror("xcecli_jobfinish");
return rc;
}
printf("ok\n");
return 0;
}
static int set_resval(xce_resval_t* vals, int idx, char* name, char* val)
{
static struct {
char* name;
int valtype;
} resinfo[] = {
{ "r15s", Value_u_fvalue },
{ "r1m", Value_u_fvalue },
{ "slots", Value_u_fvalue },
{ "ut", Value_u_fvalue },
{ "mem", Value_u_fvalue },
{ "maxcpus", Value_u_fvalue },
{ "ndisks", Value_u_fvalue },
{ "cpuf", Value_u_fvalue },
{ "maxswap", Value_u_fvalue },
{ "maxtmp", Value_u_fvalue },
{ "type", Value_u_svalue },
{ "model", Value_u_svalue },
{ "verilog", Value_u_fvalue },
{ "dyna", Value_u_fvalue },
{ NULL, 0 }
};
int i;
int rc = -1;
if (name == NULL) {
return 0;
}
for (i = 0; resinfo[i].name != NULL; i++) {
if (strcmp(resinfo[i].name, name) == 0) {
rc = 1;
vals[idx].val = (xce_val_t*)calloc(1, sizeof(xce_val_t));
switch (resinfo[i].valtype) {
case Value_u_fvalue:
vals[idx].name = name;
vals[idx].val->which = resinfo[i].valtype;
if (val == NULL) {
vals[idx].val->u.fvalue = 0.0;
} else {
vals[idx].val->u.fvalue = atof(val);
}
break;
case Value_u_ivalue32:
vals[idx].name = name;
vals[idx].val->which = resinfo[i].valtype;
if (val == NULL) {
vals[idx].val->u.ivalue32 = 0;
} else {
vals[idx].val->u.ivalue32 = atoi(val);
}
break;
case Value_u_ivalue64:
vals[idx].name = name;
vals[idx].val->which = resinfo[i].valtype;
if (val == NULL) {
vals[idx].val->u.ivalue64 = 0;
} else {
vals[idx].val->u.ivalue64 = atol(val);
}
break;
case Value_u_bvalue:
vals[idx].name = name;
vals[idx].val->which = resinfo[i].valtype;
if (val == NULL) {
vals[idx].val->u.bvalue = 0;
} else {
if (strcasecmp(val, "t") == 0) {
vals[idx].val->u.bvalue = 1;
} else {
vals[idx].val->u.bvalue = 0;
}
}
break;
}
vals[idx].host = get_setting("myname");
return rc;
}
}
return rc;
}
int cmd_resupdate(int argc, char* argv[])
{
#define MAX_N_RES (16)
int nres;
xce_resval_t vals[MAX_N_RES];
xce_resval_t* valptrs[MAX_N_RES];
char* resname = NULL;
int opt;
int rc;
optind = 1;
nres = 0;
while ((opt = getopt(argc, argv, "n:v:")) != -1) {
switch (opt) {
case 'n':
if (resname != NULL) {
printf(
"res `%s' has no value specified, default will be used\n",
resname);
rc = set_resval(vals, nres, resname, NULL);
if (rc >= 0) {
valptrs[nres] = &vals[nres];
nres++;
}
resname = NULL;
}
resname = strdup(optarg);
break;
case 'v':
if (resname != NULL) {
rc = set_resval(vals, nres, resname, optarg);
if (rc >= 0) {
valptrs[nres] = &vals[nres];
nres++;
}
resname = NULL;
} else {
printf("ignore res value `%s' as no res name found\n", optarg);
}
break;
}
}
rc = xcecli_update_resvals(active_ctx->ctx, nres, &valptrs[0]);
if (rc >= 0) {
printf("resource update ok!\n");
} else {
SET_ERROR(-rc);
perror("xcecli_update_resval");
}
return 0;
}
int cmd_join_sched(int argc, char* argv[])
{
int role = HostRole_scheduler;
int rc = -1;
rc = xcecli_join(active_ctx->ctx, role, NULL);
if (rc < 0) {
SET_ERROR(-rc);
perror("xce_join");
return rc;
}
printf("scheduler join ok\n");
return 0;
}
static int print_reslist(xce_reslist_t reslist)
{
int i;
for (i = 0; i < reslist.count; i++) {
printf("name: %s\nflags: %d\ninterval: %d\ndesc: %s\n\n",
reslist.res[i]->name, reslist.res[i]->flags,
reslist.res[i]->interval, reslist.res[i]->desc);
}
return 0;
}
static int print_queuelist(xce_queuelist_t qlist)
{
int i;
for (i = 0; i < qlist.count; i++) {
printf("name: %s\nstatus: %d\npriority: %d\n\n", qlist.queues[i]->queue,
qlist.queues[i]->status, qlist.queues[i]->priority);
}
return 0;
}
static int print_workerlist(xce_workerlist_t wlist)
{
int i;
for (i = 0; i < wlist.count; i++) {
printf("name: %s\nstatus: %d\n\n", wlist.workers[i]->name,
wlist.workers[i]->status);
}
return 0;
}
static int print_grouplist(xce_group_configlist_t glist) { return 0; }
static int print_configlist(xce_configlist_t clist) { return 0; }
int cmd_getconf(int argc, char* argv[])
{
xce_clusterconf_t* conf;
int rc;
rc = xcecli_clusterconf(active_ctx->ctx, &conf);
if (rc < 0) {
SET_ERROR(-rc);
perror("xce_clusterconf");
return rc;
}
printf("Cluster Name: %s\n", conf->name);
printf("\nResources\n===========\n");
print_reslist(conf->reslist);
printf("\nQueues\n========\n");
print_queuelist(conf->qlist);
printf("\nWorkers\n========\n");
print_workerlist(conf->wlist);
printf("\nUser Groups\n===========\n");
print_grouplist(conf->uglist);
printf("\nHost Groups\n============\n");
print_grouplist(conf->hglist);
printf("\nConfigs\n=======\n");
print_configlist(conf->cfglist);
free_ClusterConf(conf);
return 0;
}
#define KVAL(v) (((v) == NULL) ? "<not set>" : (v))
int cmd_set(int argc, char* argv[])
{
int i;
if (argc == 1) {
for (i = 0; active_ctx->settings[i].key; i++) {
printf("%-12s:%s default: %s\n", active_ctx->settings[i].key,
KVAL(active_ctx->settings[i].value),
KVAL(active_ctx->settings[i].defval));
}
return 0;
}
if (argc != 3) {
printf("set <key> <value>\n");
return -EINVAL;
}
for (i = 0; active_ctx->settings[i].key; i++) {
if (strcmp(active_ctx->settings[i].key, argv[1]) == 0) {
if (active_ctx->settings[i].value != NULL) {
free(active_ctx->settings[i].value);
}
active_ctx->settings[i].value = strdup(argv[2]);
return 0;
}
}
printf("Can not find setting key: %s\n", argv[1]);
return -EINVAL;
}
static int run_command(char* cmdline, pid_t* pid)
{
pid_t child;
child = fork();
*pid = child;
if (child < 0) {
*pid = -errno;
return -errno;
}
if (child == 0) {
execlp("/bin/sh", "/bin/sh", "-c", cmdline, NULL);
perror("execlp");
exit(-1);
}
return 0;
}
static int run_task(char* cmdline, pid_t* pid)
{
return run_command(cmdline, pid);
}
static int get_avail_run_job_slots()
{
int i;
for (i = 0; i < MAX_RUN_JOBS; i++) {
if (run_jobs[i].jobid == 0) {
return i;
}
}
return -1;
}
static int get_avail_run_task_slots()
{
int i;
for (i = 0; i < MAX_RUN_TASKS; i++) {
if (run_tasks[i].pid <= 0) {
return i;
}
}
return -1;
}
static int sched_on_jobs(int njobs, xce_job_t** jobs)
{
printf("%s %d jobs gotten\n", __func__, njobs);
return 0;
}
static int sched_on_workers(int nworkers, xce_worker_t** workers)
{
printf("%s %d workers gotten\n", __func__, nworkers);
return 0;
}
static void notify_job_fail(uint64_t jobid, int code)
{
/* clang-format off */
job_failinfo_t failinfo = {
.jobid = jobid,
.stage = JOB_EXEC_STAGE_MAIN,
.code = code,
.detail = "no detail information"
};
/*clang-format on*/
xcecli_notify_jobfail(active_ctx->ctx, &failinfo);
return;
}
static void notify_task_fail(uint64_t jobid,
int clientidx,
int taskidx,
int code) {
xce_task_failinfo_t failinfo = {
.jobid = jobid,
.host = get_setting("myname"),
.clientidx = clientidx,
.taskidx = taskidx,
.code = code,
.detail = NULL
};
xcecli_notify_taskfail(active_ctx->ctx, &failinfo);
}
static void notify_jobstart(uint64_t jobid,
int pid,
int pgid) {
/* clang-format off */
job_pidinfo_t pidinfo = {
.jobid = jobid,
.npids = 1,
.pids = (unsigned int []){(unsigned int)pid},
.npgids = 0,
.pgids = (unsigned int []){(unsigned int)pgid},
.exdata = NULL
};
/* clang-format on */
xcecli_notify_jobstart(active_ctx->ctx, &pidinfo);
return;
}
static void notify_task_start(uint64_t jobid, int clientidx, int taskidx,
int pid, int pgid)
{
xce_task_pidinfo_t pidinfo
= { .jobid = jobid,
.host = get_setting("myname"),
.clientidx = clientidx,
.taskidx = taskidx,
.npids = 1,
.pids = (unsigned int[]) { (unsigned int)pid },
.npgids = 1,
.pgids = (unsigned int[]) { (unsigned int)pgid },
.exdata = NULL };
xcecli_notify_taskstart(active_ctx->ctx, &pidinfo);
return;
}
static int run_task_remote(uint64_t jobid, job_startinfo_t* startinfo,
xce_launch_t* launchinfo)
{
xce_taskhost_t hosts[launchinfo->nhosts - 1];
xce_taskrun_t taskrun;
int i;
int rc;
taskrun.taskhosts = (xce_taskhost_t**)calloc(launchinfo->nhosts - 1,
sizeof(xce_taskhost_t*));
if (taskrun.taskhosts == NULL) {
return -ENOMEM;
}
taskrun.clientidx = -1;
for (i = 1; i < launchinfo->nhosts; i++) {
hosts[i - 1].exechome = startinfo->exechome;
hosts[i - 1].execcwd = startinfo->execcwd;
hosts[i - 1].cmdline = startinfo->cmdline;
hosts[i - 1].exdata = startinfo->exdata;
hosts[i - 1].execuser = startinfo->execuser;
hosts[i - 1].launch_type = startinfo->launch_type;
hosts[i - 1].host = launchinfo->hosts[i]->host;
hosts[i - 1].ntasks = launchinfo->hosts[i]->nranks;
if (hosts[i - 1].launch_type == NULL) {
/* use mpi as default */
hosts[i - 1].launch_type = "mpi";
}
taskrun.taskhosts[i - 1] = &hosts[i - 1];
}
taskrun.nhosts = launchinfo->nhosts - 1;
taskrun.jobid = jobid;
xcecli_unblock_tasks(active_ctx->ctx);
rc = xcecli_runtask(active_ctx->ctx, &taskrun);
if (rc < 0) {
fprintf(stderr, "run remote task failed\n");
}
FREEUP(taskrun.taskhosts);
return rc;
}
static int run_task_local(uint64_t jobid, job_startinfo_t* startinfo,
int ntasks)
{
pid_t pid;
pid = fork();
if (pid < 0) {
return pid;
}
if (pid == 0) {
int i;
pid_t pids[ntasks];
int rc = 0;
int nlive = 0;
/* disable detection of address sanitizer as
* the request structure will not be freed in
* child and we will exit directly
*/
setenv("ASAN_OPTIONS", "detect_leaks=false", 1);
for (i = 0; i < ntasks; i++) {
rc = run_command(startinfo->cmdline, &pids[i]);
if (rc >= 0) {
nlive++;
}
}
if (rc >= 0) {
rc = 0;
}
while (nlive > 0) {
sleep(1);
for (i = 0; i < ntasks; i++) {
int wstatus;
if (pids[i] <= 0) {
continue;
}
if (waitpid(pids[i], &wstatus, WNOHANG) <= 0) {
continue;
}
if (WIFEXITED(wstatus) || WIFSIGNALED(wstatus)) {
nlive--;
if (WIFSIGNALED(wstatus)) {
rc = 128 + WTERMSIG(wstatus);
} else {
if (WEXITSTATUS(wstatus) != 0) {
rc = WEXITSTATUS(wstatus);
}
}
}
}
}
exit(rc);
}
printf("Job %lu started, PID: %d\n", jobid, pid);
return pid;
}
int run_job(xce_job_exec_t* startmsg)
{
int rc = 0;
uint64_t jobid = startmsg->jobid;
job_startinfo_t* startinfo = startmsg->start_info;
xce_launch_t* launchinfo = startmsg->launchinfo;
printf("%s: exechome= %s\n", __func__,
(startinfo->exechome == NULL) ? "NULL" : startinfo->exechome);
printf("%s: execcwd= %s\n", __func__,
(startinfo->execcwd == NULL) ? "NULL" : startinfo->execcwd);
printf("%s: execuser= %s\n", __func__,
(startinfo->execuser == NULL) ? "NULL" : startinfo->execuser);
printf("%s: cmdline= %s\n", __func__,
(startinfo->cmdline == NULL) ? "NULL" : startinfo->cmdline);
printf("%s: exdata= %s\n", __func__,
(startinfo->exdata == NULL) ? "NULL" : startinfo->exdata);
printf("%s: launch_type= %s\n", __func__,
(startinfo->launch_type == NULL) ? "NULL" : startinfo->launch_type);
printf("%s: q_pre_exec= %s\n", __func__,
(startinfo->q_pre_exec == NULL) ? "NULL" : startinfo->q_pre_exec);
printf("%s: q_post_exec= %s\n", __func__,
(startinfo->q_post_exec == NULL) ? "NULL" : startinfo->q_post_exec);
printf("%s: job_pre_exec= %s\n", __func__,
(startinfo->job_pre_exec == NULL) ? "NULL"
: startinfo->job_pre_exec);
printf("%s: job_post_exec= %s\n", __func__,
(startinfo->job_post_exec == NULL) ? "NULL"
: startinfo->job_post_exec);
if (launchinfo->nhosts > 1) {
rc = run_task_remote(jobid, startinfo, launchinfo);
if (rc < 0) {
fprintf(stderr, "run_task_remote() fail, rc %d\n", rc);
return rc;
}
}
rc = run_task_local(jobid, startinfo, launchinfo->hosts[0]->nranks);
if (rc < 0) {
notify_job_fail(jobid, rc);
} else {
int idx;
int pid;
pid = rc;
idx = get_avail_run_job_slots();
if (idx < 0) {
printf("too many jobs run in this worker\n");
} else {
run_jobs[idx].jobid = jobid;
run_jobs[idx].pid = pid;
run_jobs[idx].ts = 0;
}
notify_jobstart(jobid, pid, pid);
}
return rc;
}
static int handle_startjob(xce_request_t* request)
{
return run_job(request->body.jobstartmsg);
}
static int handle_notify(xce_request_t* request)
{
xce_notify_data_t* data = request->body.notifymsg;
switch (data->event) {
case NotifyData_data_taskPidInfo:
printf("job %lu host %s taskidx %d pid %d\n",
data->data.taskinfo->jobid, data->data.taskinfo->host,
data->data.taskinfo->taskidx, data->data.taskinfo->pids[0]);
break;
case NotifyData_data_taskFailInfo:
printf("job %lu host %s task %d error code %d\n",
data->data.taskfailinfo->jobid, data->data.taskfailinfo->host,
data->data.taskfailinfo->taskidx, data->data.taskfailinfo->code);
break;
case NotifyData_data_taskFinish:
printf("job %lu host %s task %d exit code %d\n",
data->data.taskfinish->jobid, data->data.taskfinish->host,
data->data.taskfinish->taskidx,
data->data.taskfinish->exit_code);
break;
}
return 0;
}
static int job_kill(uint64_t jobid, int signo)
{
int i;
int killed = 0;
for (i = 0; i < MAX_RUN_JOBS; i++) {
if ((run_jobs[i].jobid == 0) || (run_jobs[i].jobid != jobid)) {
continue;
}
if (run_jobs[i].pid > 1) {
kill(run_jobs[i].pid, signo);
killed++;
}
}
return killed;
}
static int job_chkpnt(uint64_t jobid, char* method, int kill)
{
char buf[128];
char info[256];
printf("checkpoint job %ld by method %s\n", jobid, method);
sprintf(buf, "/tmp/%ld/ckpt_dir", jobid);
/* do some actions to checkpoint jobs */
sprintf(info,
"{\nhost_select=\"criu\"; \nrestart_command=\"cd %s; criu restore "
"-vvv\"}\n",
buf);
xcecli_notify_jobchkpt(active_ctx->ctx, jobid, info);
if (kill) {
job_kill(jobid, SIGTERM);
}
return 0;
}
static int handle_jobcontrol(xce_request_t* request)
{
uint64_t jobid;
int signo;
xce_jobctrl_t* jobctrl = request->body.jobctrlmsg;
int rc = 0;
switch (jobctrl->action) {
case JobControl_u_siginfo:
jobid = jobctrl->jobid;
signo = jobctrl->u.siginfo->signo;
rc = job_kill(jobid, signo);
break;
case JobControl_u_ckptoption:
rc = job_chkpnt(jobctrl->jobid, jobctrl->u.ckptoption->method,
jobctrl->u.ckptoption->kill);
break;
default:
rc = -EINVAL;
ERROR("unknown job control action %d", jobctrl->action);
break;
}
return rc;
}
static int handle_runtasks(xce_request_t* request)
{
xce_taskrun_t* runmsg = request->body.runtaskmsg;
int rc = 0;
int i = 0;
#if 0
job_startinfo_t startinfo;
#endif
if (runmsg->nhosts != 1) {
/* something wrong */
rc = -EINVAL;
notify_task_fail(runmsg->jobid, runmsg->clientidx, -1, rc);
return rc;
}
if (strcmp(runmsg->taskhosts[0]->host, get_setting("myname")) != 0) {
/* something wrong */
rc = -EINVAL;
notify_task_fail(runmsg->jobid, runmsg->clientidx, -1, rc);
return rc;
}
printf("run task: jobid = %lu\n", runmsg->jobid);
printf("run task: user = %s\n", runmsg->taskhosts[0]->execuser);
printf("run task: cmdline = %s\n", runmsg->taskhosts[0]->cmdline);
printf("run task: exdata = %s\n", runmsg->taskhosts[0]->exdata);
printf("run task: ntasks =%d\n", runmsg->taskhosts[0]->ntasks);
#if 0
startinfo.exechome = runmsg->taskhosts[0]->exechome;
startinfo.execcwd = runmsg->taskhosts[0]->execcwd;
startinfo.execuser = runmsg->taskhosts[0]->execuser;
startinfo.cmdline = runmsg->taskhosts[0]->cmdline;
startinfo.exdata = runmsg->taskhosts[0]->exdata;
startinfo.launch_type = runmsg->taskhosts[0]->launch_type;
#endif
for (i = 0; i < runmsg->taskhosts[0]->ntasks; i++) {
int pid = 0;
int pgid = 0;
rc = run_task(runmsg->taskhosts[0]->cmdline, &pid);
if (rc < 0) {
notify_task_fail(runmsg->jobid, runmsg->clientidx, i, rc);
} else {
int idx;
printf("job %lu task %d started, PID: %d\n", runmsg->jobid, i, pid);
idx = get_avail_run_task_slots();
if (idx < 0) {
printf("too many running tasks\n");
} else {
run_tasks[idx].jobid = runmsg->jobid;
run_tasks[idx].taskid = i;
run_tasks[idx].clientidx = runmsg->clientidx;
run_tasks[idx].pid = pid;
run_tasks[idx].ts = 0;
}
notify_task_start(runmsg->jobid, runmsg->clientidx, i, pid, pgid);
}
}
return 0;
}
static int myreq_handler(xce_request_t* request, void* args)
{
int rc = 0;
switch (request->opcode) {
case OPCode_push:
if (request->body_which == Request_body_datamsg) {
dataset_t* datamsg = request->body.datamsg;
switch (datamsg->table) {
case Data_data_schedjobs:
case Data_data_jobs:
sched_on_jobs(datamsg->nrows, (xce_job_t**)datamsg->rows);
break;
case Data_data_schedworkers:
case Data_data_workers:
sched_on_workers(datamsg->nrows, (xce_worker_t**)datamsg->rows);
break;
default:
break;
}
}
break;
case OPCode_startjob:
rc = handle_startjob(request);
break;
case OPCode_notify:
rc = handle_notify(request);
break;
case OPCode_jobcontrol:
rc = handle_jobcontrol(request);
break;
case OPCode_runtasks:
rc = handle_runtasks(request);
break;
default:
break;
}
return rc;
}
static void check_run_jobs()
{
int i;
time_t now = time(NULL);
for (i = 0; i < MAX_RUN_JOBS; i++) {
int wstatus;
if (run_jobs[i].jobid == 0) {
continue;
}
if (run_jobs[i].pid <= 0) {
continue;
}
if (waitpid(run_jobs[i].pid, &wstatus, WNOHANG | WUNTRACED | WCONTINUED)
<= 0) {
/* fake job rusage */
int npids = 1;
int npgids = 0;
int nusage = 2;
xce_rusage_t data[] = {
{ "cputime", 100.0 },
{ "mem", 200.0 }
};
xce_rusage_t* usages[] = { &(data[0]), &(data[1]) };
if (now - run_jobs[i].ts > RUSAGE_UPDATE_INTERVAL) {
xcecli_update_jusage(active_ctx->ctx, run_jobs[i].jobid, npids,
&(run_jobs[i].pid), npgids, NULL, nusage,
&usages[0]);
run_jobs[i].ts = now;
}
continue;
}
if (WIFSTOPPED(wstatus)) {
xcecli_notify_jobstop(active_ctx->ctx, run_jobs[i].jobid);
} else if (WIFCONTINUED(wstatus)) {
xcecli_notify_jobresume(active_ctx->ctx, run_jobs[i].jobid);
} else if (WIFEXITED(wstatus) || WIFSIGNALED(wstatus)) {
xce_jobfinish_t finish;
finish.jobid = run_jobs[i].jobid;
if (WIFEXITED(wstatus)) {
finish.exit_code = WEXITSTATUS(wstatus);
} else {
finish.exit_code = 128 + WTERMSIG(wstatus);
}
printf("job %lu exit code %d\n", finish.jobid, finish.exit_code);
xcecli_notify_jobfinish(active_ctx->ctx, &finish);
run_jobs[i].jobid = 0;
run_jobs[i].pid = 0;
}
}
}
static void check_run_tasks()
{
int i;
time_t now = time(NULL);
for (i = 0; i < MAX_RUN_TASKS; i++) {
int wstatus;
if (run_tasks[i].jobid == 0) {
continue;
}
if (run_tasks[i].pid <= 0) {
continue;
}
if (waitpid(run_tasks[i].pid, &wstatus,
WNOHANG | WUNTRACED | WCONTINUED)
<= 0) {
/* fake job rusage */
int npids = 1;
int npgids = 0;
int nusage = 2;
xce_rusage_t data[] = {
{ "cputime", 100.0 },
{ "mem", 200.0 }
};
xce_rusage_t* usages[] = { &(data[0]), &(data[1]) };
if (now - run_tasks[i].ts > RUSAGE_UPDATE_INTERVAL) {
xcecli_update_jusage(active_ctx->ctx, run_tasks[i].jobid, npids,
&(run_tasks[i].pid), npgids, NULL, nusage,
&usages[0]);
run_tasks[i].ts = now;
}
continue;
}
if (WIFEXITED(wstatus) || WIFSIGNALED(wstatus)) {
xce_task_finish_t finish = { .jobid = run_tasks[i].jobid,
.host = get_setting("myname"),
.taskidx = run_tasks[i].taskid,
.clientidx = run_tasks[i].clientidx,
.exit_code = 0 };
if (WIFEXITED(wstatus)) {
finish.exit_code = WEXITSTATUS(wstatus);
} else {
finish.exit_code = 128 + WTERMSIG(wstatus);
}
printf("job %lu task %d exit code %d\n", finish.jobid,
finish.taskidx, finish.exit_code);
xcecli_notify_taskfinish(active_ctx->ctx, &finish);
run_tasks[i].jobid = 0;
run_tasks[i].pid = 0;
}
}
}
#define HEARTBEAT_INTERVAL (60)
static int worker_house_keeping(void* arg)
{
command_ctx_t* ctx = (command_ctx_t*)arg;
int rc = 0;
xce_use(ctx);
check_run_jobs();
check_run_tasks();
rc = xcecli_heartbeat(ctx->ctx, HEARTBEAT_INTERVAL);
if (rc < 0) {
perror("xcecli_heartbeat");
}
return rc;
}
int main(int argc, char* argv[])
{
struct arguments arguments;
char* buf;
char* logfile = "/tmp/xce_test.log";
char* loglevel = "info";
int logsize = 64 * 1024 * 1024;
int logcount = 10;
memset(&arguments, 0x0, sizeof(arguments));
argp_parse(&argp, argc, argv, 0, 0, &arguments);
if (arguments.verbose != 0) {
loglevel = verbose_to_loglevel(arguments.verbose);
}
mylog_init(logfile, loglevel, logsize, logcount);
signal(SIGINT, sigbreak);
memset(&run_jobs, 0x0, sizeof(run_jobs));
memset(&run_tasks, 0x0, sizeof(run_tasks));
idgen_ctx_init(&idctx);
while (!done) {
char* s;
int rc;
prompt[0] = 0x0;
fill_prompt(active_ctx, prompt);
strcat(prompt, ">");
buf = readline(prompt);
if (buf == NULL) {
execute_line("quit");
continue;
}
if (strlen(buf) == 0) {
free(buf);
continue;
}
s = buf;
rc = execute_line(s);
if (rc < 0) {
SET_ERROR(-rc);
} else {
add_history(s);
}
free(buf);
}
return 0;
}
View git blame Copy permalink