Rongsong Shen
9318f9977e
bug-1401: xce_free_ctrlstatus(s) was defined as free_CtrlStatus(s), which only frees the fields inside xce_ctrlstatus_t. The real client reply path (decode_CtrlStatus_ptr) calloc's the struct itself, so every ctrlstatus call leaked sizeof(xce_ctrlstatus_t) (120 bytes). Existing unit tests never caught it because they built/freed the struct via the raw stack-decode + bare free_CtrlStatus() helpers, never exercising the actual client macro. Fixed by switching the macro to free_CtrlStatus_ptr(&(s)) and added a regression test that exercises the real heap-decode path; RED confirmed the exact 120-byte/1-alloc LeakSanitizer signature before the fix. bug-1400: the originally reported SEGV in cmd_jobsub on malformed positional input is stale — NULLSTR guards added 2025-11-05 already reject it cleanly. The real residual defect was submit.start_info leaking on all 4 early-return validation paths; fixed by routing them through a shared `goto cleanup` (matches the existing goto-clean idiom in queue_ctrl). bug-1396 (job-count anomaly after cold-restart-all, not yet root-caused): instrumented test-commit-durability.sh's assert_job_count() to dump the live joblist and ACKED_FILE ledger on any count mismatch instead of only the count, per the bug's own recorded next step. Verified: controller ASAN build + full controller_unit_test green; lib xcelib_test 285/285 green, 0 ASAN leaks. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SDWVNQLKxqTVHh42cH7rV3
3497 lines
104 KiB
C
3497 lines
104 KiB
C
#include <argp.h>
|
|
#include <errno.h>
|
|
#include <getopt.h>
|
|
#include <inttypes.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 "config.h" /* XCE_MAXNAME_LEN */
|
|
#include "controller.capnp.h"
|
|
#include "ctrlstatus_fmt.h" /* ctrlstatus_format_time_ms (task 6.3) */
|
|
#include "delay.h"
|
|
#include "idgen.h"
|
|
#include "preason.h"
|
|
#include "walcompare.h" /* cross-member committed-prefix divergence (cmd_walcompare) */
|
|
#include "waldump.h" /* offline WAL inspector (cmd_waldump) */
|
|
#include "snapdiff.h" /* byte-level checkpoint-file comparator (cmd_snapdiff) */
|
|
#include "snapverify.h" /* offline checkpoint-file structural validator (cmd_snapverify) */
|
|
#include "xce/cluster.h" /* xce_parse_peers, xce_cluster_config_t */
|
|
#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*[]);
|
|
extern int cmd_echo(int, char*[]);
|
|
extern int cmd_expect_jobs(int, char*[]);
|
|
extern int cmd_expect_job(int, char*[]);
|
|
extern int cmd_wait_primary(int, char*[]);
|
|
extern int cmd_ctrlstatus(int, char*[]);
|
|
extern int cmd_waldump(int, char*[]);
|
|
extern int cmd_walcompare(int, char*[]);
|
|
extern int cmd_snapverify(int, char*[]);
|
|
extern int cmd_snapdiff(int, char*[]);
|
|
static int worker_house_keeping(void*);
|
|
|
|
/* Session verdict / loop state (xce-test-e2e-primitives task 1.1).
|
|
*
|
|
* g_session_failed set to 1 the first time any command in the session
|
|
* fails; drives the process exit code in main().
|
|
* g_continue_on_error when set (via `set continue-on-error`), a failing
|
|
* command does not stop the read loop; the session still
|
|
* exits non-zero at the end.
|
|
* g_last_status status of the most recently executed command
|
|
* (0 == success, non-zero == failure). Future tasks
|
|
* expose this as the `$?` auto-variable.
|
|
*/
|
|
static int g_session_failed = 0;
|
|
static int g_continue_on_error = 0;
|
|
static int g_last_status = 0;
|
|
/* Interactive (TTY) sessions are a REPL: a failed command prints its error and
|
|
* the loop CONTINUES — never abort the read loop on a typo. The abort-on-first-
|
|
* failure + non-zero-exit verdict is a SCRIPTED/piped-mode concept only. Set
|
|
* from isatty(stdin) in main. */
|
|
static int g_interactive = 0;
|
|
|
|
/* Auto-variable state for the fixed `$jobid` / `$jobcount` substitution
|
|
* (xce-test-e2e-primitives task 3.1, Req 3.1/3.3/3.5/3.6).
|
|
*
|
|
* g_last_jobid most recently acknowledged JobId as a string; empty ("")
|
|
* means unset. Populated later by cmd_jobsub (task 4.1).
|
|
* g_last_jobcount most recent job count; -1 means unset. Populated by
|
|
* cmd_joblist on a successful `joblist` (task 5.1, Req 3.4);
|
|
* a failed joblist leaves it untouched so no fake count is
|
|
* advertised.
|
|
*
|
|
* Until a command sets one, it stays unset — which is exactly what makes an
|
|
* unset `$jobid`/`$jobcount` reference fail the command (Req 3.6). */
|
|
static char g_last_jobid[XCE_MAXNAME_LEN] = { 0 };
|
|
static int g_last_jobcount = -1;
|
|
|
|
static idgen_ctx_t idctx;
|
|
delaymgr_t delaymgr;
|
|
|
|
static setting_t settings[] = {
|
|
/* HA controller list: comma-separated "name=host:port" or "host:port"
|
|
* entries (same syntax as the controller --peers flag), enabling the
|
|
* test client to fail over across controller candidates. */
|
|
{ (char*)"masters", (char*)NULL, (char*)"localhost: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" },
|
|
{ "echo", cmd_echo, "Print a message" },
|
|
{ "waldump", cmd_waldump, "Inspect WAL segments" },
|
|
{ "walcompare", cmd_walcompare, "Diff member committed WAL prefixes" },
|
|
{ "snapverify", cmd_snapverify, "Verify checkpoint file structure" },
|
|
{ "snapdiff", cmd_snapdiff, "Byte-diff two checkpoint files" },
|
|
{ "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" },
|
|
{ "echo", cmd_echo, "Print a message" },
|
|
{ (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" },
|
|
{ "expect-jobs", cmd_expect_jobs, "assert current job count" },
|
|
{ "expect-job", cmd_expect_job, "assert a JobId is present" },
|
|
{ "wait-primary", cmd_wait_primary,
|
|
"wait until a primary serves [timeout_ms]" },
|
|
{ "ctrlstatus", cmd_ctrlstatus, "show controller/cluster status" },
|
|
{ "waldump", cmd_waldump, "Inspect WAL segments" },
|
|
{ "walcompare", cmd_walcompare, "Diff member committed WAL prefixes" },
|
|
{ "snapverify", cmd_snapverify, "Verify checkpoint file structure" },
|
|
{ "snapdiff", cmd_snapdiff, "Byte-diff two checkpoint files" },
|
|
{ "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" },
|
|
{ "echo", cmd_echo, "Print a message" },
|
|
{ "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" },
|
|
{ "echo", cmd_echo, "Print a message" },
|
|
{ (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;
|
|
|
|
/* Holds the parsed HA controller list. Static so the hostname[] buffers it
|
|
* owns outlive enter_ctx(): xcecli_ctx_init() stores the config pointer (it
|
|
* does not copy masters[]), so config.masters[i] must reference storage that
|
|
* persists for the lifetime of the client context. */
|
|
static xce_cluster_config_t master_cfg;
|
|
|
|
int enter_ctx(command_ctx_t* ctx)
|
|
{
|
|
char* masters = get_setting("masters");
|
|
int i;
|
|
|
|
/* Parse the controller HA peer list with the shared libxce parser — the
|
|
* same "name=host:port,..."/"host:port,..." syntax used by the controller
|
|
* --peers flag — so the test client fails over across candidates exactly
|
|
* like the worker/scheduler/apiserver do. */
|
|
memset(&master_cfg, 0, sizeof(master_cfg));
|
|
if ((masters == NULL) || (xce_parse_peers(masters, &master_cfg) != 0) ||
|
|
(master_cfg.candidates_count < 1)) {
|
|
fprintf(stderr, "enter_ctx: invalid 'masters' setting `%s'\n",
|
|
masters ? masters : "(null)");
|
|
return -EINVAL;
|
|
}
|
|
|
|
config.nmasters = master_cfg.candidates_count;
|
|
if (config.nmasters > N_MAX_MASTERS) {
|
|
fprintf(stderr,
|
|
"enter_ctx: %d masters configured, using first %d "
|
|
"(N_MAX_MASTERS)\n",
|
|
config.nmasters, N_MAX_MASTERS);
|
|
config.nmasters = N_MAX_MASTERS;
|
|
}
|
|
|
|
for (i = 0; i < config.nmasters; i++) {
|
|
config.masters[i] = master_cfg.candidates[i].hostname;
|
|
config.master_ports[i] = master_cfg.candidates[i].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;
|
|
}
|
|
|
|
/* Substitute the fixed auto-variables `$?`, `$jobid`, `$jobcount` on a raw line
|
|
* before tokenizing (xce-test-e2e-primitives task 3.1, Req 3.1/3.3/3.5/3.6).
|
|
*
|
|
* Only WHOLE-TOKEN occurrences are recognized: a `$name` must be bounded by the
|
|
* start of the line / whitespace on its left and whitespace / end-of-line on its
|
|
* right. A `$name` glued to other non-whitespace (e.g. `x$jobid`, `$jobidx`) is
|
|
* NOT a recognized auto-variable and is copied through verbatim. No other `$...`
|
|
* syntax is recognized: an unknown whole token such as `$other` is also copied
|
|
* verbatim (it is simply not one of the three names).
|
|
*
|
|
* Replacements:
|
|
* $? -> decimal of g_last_status (always available).
|
|
* $jobid -> g_last_jobid if set (non-empty); unset => command failure.
|
|
* $jobcount -> decimal of g_last_jobcount if set (>=0); unset => failure.
|
|
*
|
|
* Writes the expanded line into `out` (capacity `outsz`, always NUL-terminated
|
|
* on success). Returns 0 on success, -EINVAL on an unset auto-var reference,
|
|
* -ENOSPC if the expansion would overflow `out`. On any error `out` content is
|
|
* unspecified and the caller must fail the command rather than run it. */
|
|
static int substitute_autovars(const char* line, char* out, size_t outsz)
|
|
{
|
|
const char* p = line;
|
|
size_t o = 0;
|
|
|
|
if ((line == NULL) || (out == NULL) || (outsz == 0)) {
|
|
return -EINVAL;
|
|
}
|
|
|
|
while (*p != '\0') {
|
|
/* A `$` starts a candidate auto-var only when it begins a token, i.e.
|
|
* the previous char is the line start or whitespace. */
|
|
int at_token_start
|
|
= (p == line) || (*(p - 1) == ' ') || (*(p - 1) == '\t');
|
|
|
|
if ((*p == '$') && at_token_start) {
|
|
const char* name = p + 1;
|
|
const char* end = name;
|
|
const char* repl = NULL;
|
|
char numbuf[32];
|
|
|
|
/* Scan to the token end (whitespace or NUL). */
|
|
while ((*end != '\0') && (*end != ' ') && (*end != '\t')) {
|
|
end++;
|
|
}
|
|
|
|
size_t namelen = (size_t)(end - name);
|
|
|
|
if ((namelen == 1) && (name[0] == '?')) {
|
|
snprintf(numbuf, sizeof(numbuf), "%d", g_last_status);
|
|
repl = numbuf;
|
|
} else if ((namelen == 5) && (strncmp(name, "jobid", 5) == 0)) {
|
|
if (g_last_jobid[0] == '\0') {
|
|
fprintf(stderr,
|
|
"autovar: $jobid referenced before set\n");
|
|
return -EINVAL;
|
|
}
|
|
repl = g_last_jobid;
|
|
} else if ((namelen == 8) && (strncmp(name, "jobcount", 8) == 0)) {
|
|
if (g_last_jobcount < 0) {
|
|
fprintf(stderr,
|
|
"autovar: $jobcount referenced before set\n");
|
|
return -EINVAL;
|
|
}
|
|
snprintf(numbuf, sizeof(numbuf), "%d", g_last_jobcount);
|
|
repl = numbuf;
|
|
}
|
|
|
|
if (repl != NULL) {
|
|
size_t rlen = strlen(repl);
|
|
|
|
if ((o + rlen) >= outsz) {
|
|
return -ENOSPC;
|
|
}
|
|
memcpy(out + o, repl, rlen);
|
|
o += rlen;
|
|
p = end; /* consume the whole `$name` token */
|
|
continue;
|
|
}
|
|
|
|
/* Not a recognized auto-var: copy the whole token verbatim. */
|
|
while (p < end) {
|
|
if ((o + 1) >= outsz) {
|
|
return -ENOSPC;
|
|
}
|
|
out[o++] = *p++;
|
|
}
|
|
continue;
|
|
}
|
|
|
|
if ((o + 1) >= outsz) {
|
|
return -ENOSPC;
|
|
}
|
|
out[o++] = *p++;
|
|
}
|
|
|
|
out[o] = '\0';
|
|
return 0;
|
|
}
|
|
|
|
/* Main-loop wrapper around execute_line that implements the session verdict
|
|
* (xce-test-e2e-primitives task 1.1):
|
|
* - blank / whitespace-only lines and `#`-comment lines are ignored;
|
|
* - normal lines run via execute_line; g_last_status / g_session_failed are
|
|
* updated from the result;
|
|
* - in default (abort) mode a failure stops the loop; under
|
|
* g_continue_on_error the loop keeps running but the failure is recorded.
|
|
*
|
|
* Returns 0 to keep the read loop running, non-zero to request that the loop
|
|
* stop (abort-on-failure). The leading `!` shell-escape and `$`-substitution
|
|
* are intentionally handled in later tasks (2.1 / 3.1); this function is the
|
|
* single chokepoint where they will be added.
|
|
*/
|
|
int preprocess_and_run(char* line)
|
|
{
|
|
char* p = line;
|
|
int rc;
|
|
|
|
/* Skip leading whitespace to classify the line. */
|
|
while ((*p == ' ') || (*p == '\t')) {
|
|
p++;
|
|
}
|
|
|
|
/* Ignore blank / whitespace-only lines and `#` comments. */
|
|
if ((*p == '\0') || (*p == '#')) {
|
|
return 0;
|
|
}
|
|
|
|
/* Shell-escape: a line whose first non-whitespace char is `!` runs the
|
|
* remainder via system(). The shell command's exit status becomes
|
|
* g_last_status (the `$?` auto-variable); a non-zero status is a command
|
|
* failure subject to the same abort-vs-continue rules as execute_line. The
|
|
* remainder is passed raw to the shell (no parse_argv / auto-var
|
|
* substitution — that is task 3.1). (xce-test-e2e-primitives task 2.1) */
|
|
if (*p == '!') {
|
|
const char* rest = p + 1;
|
|
int sysrc = system(rest);
|
|
int status;
|
|
|
|
if (sysrc == -1) {
|
|
/* system() itself failed to run the shell — treat as failure. */
|
|
status = 127;
|
|
} else if (WIFEXITED(sysrc)) {
|
|
status = WEXITSTATUS(sysrc);
|
|
} else {
|
|
/* Signaled / abnormal termination — non-zero failure status. */
|
|
status = 1;
|
|
}
|
|
|
|
g_last_status = status;
|
|
|
|
if (status != 0) {
|
|
g_session_failed = 1;
|
|
if (!g_continue_on_error && !g_interactive) {
|
|
return 1; /* abort: stop the read loop (scripted mode only) */
|
|
}
|
|
} else {
|
|
add_history(line);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
/* Substitute the fixed auto-variables (`$?`, `$jobid`, `$jobcount`) before
|
|
* tokenizing (xce-test-e2e-primitives task 3.1). An unset `$jobid` /
|
|
* `$jobcount` reference (Req 3.6) — or an overflow — fails the command:
|
|
* we do NOT run execute_line, and the result is treated exactly like a
|
|
* failing command (g_last_status non-zero, g_session_failed, abort vs
|
|
* continue). The leading `!` shell-escape above is intentionally NOT
|
|
* substituted: that line goes raw to system() and the shell owns `$?`. */
|
|
{
|
|
char expanded[MAX_LINE_LEN];
|
|
int subrc = substitute_autovars(p, expanded, sizeof(expanded));
|
|
|
|
if (subrc < 0) {
|
|
g_last_status = subrc;
|
|
g_session_failed = 1;
|
|
if (!g_continue_on_error && !g_interactive) {
|
|
return 1; /* abort: stop the read loop (scripted mode only) */
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
rc = execute_line(expanded);
|
|
}
|
|
|
|
g_last_status = (rc < 0) ? (rc) : 0;
|
|
|
|
if (rc < 0) {
|
|
SET_ERROR(-rc);
|
|
g_session_failed = 1;
|
|
if (!g_continue_on_error) {
|
|
return 1; /* errexit (-e): stop the read loop on first failure */
|
|
}
|
|
} else {
|
|
add_history(line);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
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;
|
|
}
|
|
|
|
/* echo <message...> — print the remainder of the line (argv[1..] joined by
|
|
* single spaces) followed by a newline and succeed (xce-test-e2e-primitives
|
|
* task 1.2, Req 6.3). `echo` with no argument prints an empty line and still
|
|
* succeeds. Returns 0 always so it never marks the session failed. */
|
|
int cmd_echo(int argc, char* argv[])
|
|
{
|
|
int i;
|
|
|
|
for (i = 1; i < argc; i++) {
|
|
if (i > 1) {
|
|
putchar(' ');
|
|
}
|
|
fputs(argv[i], stdout);
|
|
}
|
|
putchar('\n');
|
|
|
|
return 0;
|
|
}
|
|
|
|
/* waldump <path> [<path> ...] — offline WAL inspector for failover/durability
|
|
* triage. Each path may be a seg-*.wal file, a wal/ dir, or a node/data dir.
|
|
* Read-only: lists every entry, decodes JOB_SUBMIT jobids, and flags LSN gaps
|
|
* (a missing committed prefix is THE failover job-loss signal) plus the CRC
|
|
* recovered tip. Returns nonzero if any path failed or any gap was found, so
|
|
* it doubles as an assertion in scripted E2E runs. */
|
|
int cmd_waldump(int argc, char* argv[])
|
|
{
|
|
if (argc < 2) {
|
|
fprintf(stderr,
|
|
"usage: waldump <wal_dir|data_dir|seg_file> [more...]\n");
|
|
return -1;
|
|
}
|
|
|
|
int rc = 0;
|
|
for (int i = 1; i < argc; i++) {
|
|
waldump_stats_t st;
|
|
int r = waldump_path(argv[i], stdout, &st);
|
|
if (r != 0) {
|
|
fprintf(stderr, "waldump: %s: %s\n", argv[i], strerror(-r));
|
|
rc = -1;
|
|
continue;
|
|
}
|
|
if (st.n_gaps > 0 || st.crc_broken)
|
|
rc = -1; /* surface the defect as a nonzero command verdict */
|
|
}
|
|
return rc;
|
|
}
|
|
|
|
/* walcompare [--report] [--committed N] <memberA> <memberB> [more...] — diff
|
|
* member WAL committed prefixes by (lsn -> term, pcrc) and FAIL (nonzero) on any
|
|
* committed-prefix divergence (ha-term-safety R6.1/R6.2/R6.3), reusing the
|
|
* waldump decoder. Each member path may be a seg-*.wal file, a wal/ dir, or a
|
|
* node/data dir (same resolution as waldump). With no --committed, the shared
|
|
* committed boundary = the minimum CRC-intact tip across members. The diverging
|
|
* LSN and per-member (term, pcrc) identities are reported. With --report, also
|
|
* emit a machine-checkable per-member `(lsn, term, pcrc, status)` table and a
|
|
* committed-vs-uncommitted classification for each divergence (task 1.1). Exit
|
|
* 0 = identical, nonzero = divergence or load error — so it doubles as a
|
|
* scripted assertion. */
|
|
int cmd_walcompare(int argc, char* argv[])
|
|
{
|
|
uint64_t committed = 0; /* 0 = derive the shared boundary */
|
|
int report = 0; /* --report => full per-member machine table */
|
|
int ai = 1;
|
|
|
|
/* Optional flags (any order): --report, --committed N. */
|
|
for (;;) {
|
|
if (argc > ai && strcmp(argv[ai], "--report") == 0) {
|
|
report = 1;
|
|
ai += 1;
|
|
continue;
|
|
}
|
|
if (argc > ai && strcmp(argv[ai], "--committed") == 0) {
|
|
if (argc <= ai + 1) {
|
|
fprintf(stderr, "walcompare: --committed needs an LSN\n");
|
|
return -1;
|
|
}
|
|
char* end = NULL;
|
|
unsigned long v = strtoul(argv[ai + 1], &end, 10);
|
|
if (end == argv[ai + 1] || *end != '\0') {
|
|
fprintf(stderr, "walcompare: bad --committed LSN '%s'\n",
|
|
argv[ai + 1]);
|
|
return -1;
|
|
}
|
|
committed = (uint64_t)v;
|
|
ai += 2;
|
|
continue;
|
|
}
|
|
break;
|
|
}
|
|
|
|
int n = argc - ai;
|
|
if (n < 2) {
|
|
fprintf(stderr,
|
|
"usage: walcompare [--report] [--committed N] <memberA> "
|
|
"<memberB> [more...]\n");
|
|
return -1;
|
|
}
|
|
if (n > 16) {
|
|
fprintf(stderr, "walcompare: at most 16 members\n");
|
|
return -1;
|
|
}
|
|
|
|
walcompare_member_t members[16];
|
|
for (int i = 0; i < n; i++) {
|
|
members[i].path = argv[ai + i];
|
|
members[i].committed_lsn = committed; /* 0 => derive shared boundary */
|
|
}
|
|
|
|
walcompare_divergence_t div;
|
|
int rc = report ? walcompare_report(members, n, stdout, &div)
|
|
: walcompare(members, n, stdout, &div);
|
|
if (rc == WALCMP_OK)
|
|
return 0;
|
|
if (rc == WALCMP_DIVERGENT) {
|
|
fprintf(stderr,
|
|
"walcompare: FAIL — committed-prefix divergence at LSN %" PRIu64
|
|
"\n",
|
|
div.lsn);
|
|
return -1; /* nonzero session verdict */
|
|
}
|
|
fprintf(stderr, "walcompare: error: %s\n", strerror(-rc));
|
|
return -1;
|
|
}
|
|
|
|
/* snapverify <snap_file> [<snap_file> ...] — offline checkpoint-file
|
|
* structural validator for force-snapshot triage (bug-1389: OPCode_
|
|
* snapshotTransfer chunk-content corruption). Each path is a snap-*.capnp
|
|
* (locally generated) or snap-*.bin (received via ha_handle_snapshot_chunk)
|
|
* file. Read-only: walks the same layout persist_load_snapshot() parses
|
|
* without restoring anything, and on the first structural defect reports the
|
|
* exact byte offset and how many job/queue frames parsed cleanly before it —
|
|
* the diagnostic a bare "persist_load_snapshot failed rc=-5" log line
|
|
* doesn't give you. Returns nonzero if any path failed to verify. */
|
|
int cmd_snapverify(int argc, char* argv[])
|
|
{
|
|
if (argc < 2) {
|
|
fprintf(stderr, "usage: snapverify <snap_file> [more...]\n");
|
|
return -1;
|
|
}
|
|
|
|
int rc = 0;
|
|
for (int i = 1; i < argc; i++) {
|
|
snapverify_report_t rep;
|
|
int r = snapverify_path(argv[i], stdout, &rep);
|
|
if (r != 0)
|
|
rc = -1;
|
|
}
|
|
return rc;
|
|
}
|
|
|
|
/* snapdiff <source_file> <received_file> — byte-level checkpoint-file
|
|
* comparator for force-snapshot triage (bug-1389). `source_file` is the
|
|
* primary's own locally generated checkpoint (snap-*.capnp); `received_file`
|
|
* is the copy a standby assembled from the wire (snap-*.bin). Reports the
|
|
* first differing byte offset and whether it lands exactly on an
|
|
* OPCode_snapshotTransfer chunk boundary — the decisive signal for "the
|
|
* transport corrupted one chunk" vs "these were never the same content".
|
|
* Returns 0 if identical, nonzero otherwise. */
|
|
int cmd_snapdiff(int argc, char* argv[])
|
|
{
|
|
if (argc != 3) {
|
|
fprintf(stderr, "usage: snapdiff <source_file> <received_file>\n");
|
|
return -1;
|
|
}
|
|
|
|
snapdiff_report_t rep;
|
|
int rc = snapdiff_paths(argv[1], argv[2], stdout, &rep);
|
|
return (rc == 0) ? 0 : -1;
|
|
}
|
|
|
|
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";
|
|
}
|
|
|
|
/* Shared, print-free joblist core for `cmd_joblist` and the `expect-jobs` /
|
|
* `expect-job` assertions (xce-test-e2e-primitives task 5.1, design.md:81,
|
|
* 116-117).
|
|
*
|
|
* Performs the single `xcecli_joblist` call (through the failover-aware client
|
|
* context) and returns the current job count (>= 0), or a negative errno-style
|
|
* value on failure. When `find_jobid != NULL`, `*contains` is set to 1 if a job
|
|
* whose `fmt_jobid(...)` string equals `find_jobid` is present, else 0; the
|
|
* `fmt_jobid` representation matches the one `cmd_joblist` prints and the one
|
|
* `cmd_jobsub` captures into `$jobid` (task 4.1), so `expect-job $jobid` lines
|
|
* up. When `print_table` is set, the human-readable table is printed (the
|
|
* `cmd_joblist` path). The job array is always freed internally.
|
|
*
|
|
* On error this returns the negative rc WITHOUT setting any auto-variable, so a
|
|
* failed joblist never advertises a fake `$jobcount` (Req 3.6). */
|
|
static int joblist_query(const char* find_jobid, int* contains, int print_table,
|
|
int quiet)
|
|
{
|
|
int rc, i;
|
|
xce_job_t** jobs;
|
|
|
|
if (contains != NULL) {
|
|
*contains = 0;
|
|
}
|
|
|
|
rc = xcecli_joblist(active_ctx->ctx, &jobs);
|
|
|
|
if (rc < 0) {
|
|
SET_ERROR(-rc);
|
|
/* `quiet` callers (e.g. the wait-primary poll, which probes repeatedly
|
|
* and EXPECTS failure until a primary serves) suppress the per-attempt
|
|
* error print so a normal failover wait does not spam the console. */
|
|
if (!quiet) {
|
|
perror("xce_joblist");
|
|
}
|
|
return rc;
|
|
}
|
|
|
|
if (print_table) {
|
|
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);
|
|
|
|
if ((find_jobid != NULL) && (contains != NULL)
|
|
&& (strcmp(buf, find_jobid) == 0)) {
|
|
*contains = 1;
|
|
}
|
|
|
|
if (print_table) {
|
|
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 rc;
|
|
}
|
|
|
|
int cmd_joblist(int argc, char* argv[])
|
|
{
|
|
int count;
|
|
|
|
count = joblist_query(NULL, NULL, 1, /*quiet=*/0);
|
|
|
|
if (count < 0) {
|
|
return count;
|
|
}
|
|
|
|
/* Capture $jobcount (Req 3.4): a successful joblist publishes the returned
|
|
* job count so a later `echo $jobcount` / `expect-jobs` substitutes it. On
|
|
* failure we returned above without touching g_last_jobcount, so a failed
|
|
* joblist never advertises a fake count (Req 3.6). */
|
|
g_last_jobcount = count;
|
|
|
|
return 0;
|
|
}
|
|
|
|
/* `expect-jobs N` assertion (xce-test-e2e-primitives task 5.2, Req 5.1/5.2,
|
|
* design.md:116-117).
|
|
*
|
|
* Passes (returns 0) when the current job count equals N; otherwise prints the
|
|
* expected vs actual count and returns a negative errno-style value so
|
|
* preprocess_and_run records the session as failed (Req 1.1).
|
|
*
|
|
* A malformed argument (missing or non-numeric N) is itself an assertion
|
|
* failure, not a no-op. A failed joblist query is treated as a failure too: we
|
|
* cannot determine the count, so we never silently equate "query failed" with
|
|
* "zero jobs" (Req 5.2 reports actual; a failed query has no trustworthy
|
|
* actual). */
|
|
int cmd_expect_jobs(int argc, char* argv[])
|
|
{
|
|
long expected;
|
|
char* end;
|
|
int count;
|
|
|
|
if (argc != 2) {
|
|
printf("expect-jobs <N>\n");
|
|
return -EINVAL;
|
|
}
|
|
|
|
errno = 0;
|
|
expected = strtol(argv[1], &end, 10);
|
|
|
|
if ((errno != 0) || (end == argv[1]) || (*end != '\0') || (expected < 0)) {
|
|
printf("expect-jobs: invalid count '%s' (expected a non-negative "
|
|
"integer)\n",
|
|
argv[1]);
|
|
return -EINVAL;
|
|
}
|
|
|
|
/* Print-free count via the shared helper. A negative rc means the query
|
|
* itself failed (e.g. no reachable master) — that is an assertion failure,
|
|
* not "0 jobs". */
|
|
count = joblist_query(NULL, NULL, 0, /*quiet=*/0);
|
|
|
|
if (count < 0) {
|
|
printf("expect-jobs: could not determine job count (joblist query "
|
|
"failed)\n");
|
|
return count;
|
|
}
|
|
|
|
if (count != (int)expected) {
|
|
printf("expect-jobs: expected %ld, actual %d\n", expected, count);
|
|
return -EINVAL;
|
|
}
|
|
|
|
printf("expect-jobs: ok (%d)\n", count);
|
|
|
|
return 0;
|
|
}
|
|
|
|
/* `expect-job <jobid>` assertion (xce-test-e2e-primitives task 5.3, Req 5.3/5.4,
|
|
* design.md cmd_expect_jobs / cmd_expect_job).
|
|
*
|
|
* Passes (returns 0) when a job whose `fmt_jobid` string equals argv[1] is
|
|
* present in the current job list; otherwise prints the missing id and returns a
|
|
* negative errno-style value so preprocess_and_run records the session as failed
|
|
* (Req 1.1).
|
|
*
|
|
* `$jobid` substitution is transparent: autovar substitution (task 3.1) replaces
|
|
* `$jobid` with the captured id string BEFORE this handler runs, so argv[1] is
|
|
* already the literal id. The membership test in joblist_query uses the SAME
|
|
* `fmt_jobid` representation that cmd_jobsub captured into `$jobid` (task 4.1),
|
|
* so `expect-job $jobid` matches the joblist entry.
|
|
*
|
|
* A missing argument is itself an assertion failure, not a no-op. A failed
|
|
* joblist query is also a failure: we cannot determine membership, so we report
|
|
* the query failure distinctly rather than silently equating it with "absent". */
|
|
int cmd_expect_job(int argc, char* argv[])
|
|
{
|
|
int contains;
|
|
int rc;
|
|
|
|
if (argc != 2) {
|
|
printf("expect-job <jobid>\n");
|
|
return -EINVAL;
|
|
}
|
|
|
|
/* Print-free membership query via the shared helper. A negative rc means the
|
|
* query itself failed (e.g. no reachable master) — that is an assertion
|
|
* failure, distinct from "job absent". */
|
|
rc = joblist_query(argv[1], &contains, 0, /*quiet=*/0);
|
|
|
|
if (rc < 0) {
|
|
printf("expect-job: could not determine membership of '%s' (joblist "
|
|
"query failed)\n",
|
|
argv[1]);
|
|
return rc;
|
|
}
|
|
|
|
if (!contains) {
|
|
printf("expect-job: job %s not found\n", argv[1]);
|
|
return -ENOENT;
|
|
}
|
|
|
|
printf("expect-job: ok (%s)\n", argv[1]);
|
|
|
|
return 0;
|
|
}
|
|
|
|
/* `wait-primary [timeout_ms]` (controller-status task 4.2, Req 9.1-9.4;
|
|
* originally xce-test-e2e-primitives task 6.1; design.md "cmd_wait_primary probe
|
|
* swap — State contract").
|
|
*
|
|
* Blocks until the AUTHORITATIVE controller status probe reaches the serving
|
|
* primary, or the timeout elapses. The probe is xcecli_ctrlstatus_primary: it
|
|
* drives active_ctx + libxce master-rotation/failover and returns 0 ONLY when a
|
|
* reply with responderRole == PRIMARY was obtained (it internally follows the
|
|
* redirect to the primary), or a typed negative when no primary answered within
|
|
* its budget (Req 9.1). A standby/hint answer is NOT success — the probe keeps
|
|
* polling until the primary itself answers (this avoids a false success on a
|
|
* standby's stale primary view after a primary death).
|
|
*
|
|
* On the first authoritative probe we print the served primary's identity and
|
|
* return 0 (Req 9.2). Otherwise we sleep a short poll interval and retry,
|
|
* re-checking a MONOTONIC deadline between attempts so the command returns
|
|
* within a bounded time of the timeout and never hangs indefinitely when there
|
|
* is no primary (Req 9.3). The deadline is derived from CLOCK_MONOTONIC, never
|
|
* wall-clock, so a clock step cannot skew it.
|
|
*
|
|
* The timeout is argv[1] in milliseconds when supplied, else
|
|
* WAIT_PRIMARY_DEFAULT_MS (Req 9.4). A malformed timeout argument is a usage
|
|
* error (return < 0) rather than a silent fallback to the default. */
|
|
#define WAIT_PRIMARY_DEFAULT_MS 30000L /* default max wait when arg omitted */
|
|
#define WAIT_PRIMARY_POLL_MS 200L /* sleep between probe attempts */
|
|
#define MS_PER_SEC 1000L
|
|
#define NS_PER_MS 1000000L
|
|
#define US_PER_MS 1000L
|
|
|
|
/* The client's configured log level (set in main from -v), so wait-primary can
|
|
* temporarily raise the threshold to silence libxce's per-attempt failover INFO
|
|
* spam during its poll and then RESTORE the user's chosen level. */
|
|
static char* g_active_loglevel = (char*)"info";
|
|
|
|
static long monotonic_ms(void)
|
|
{
|
|
struct timespec ts;
|
|
|
|
clock_gettime(CLOCK_MONOTONIC, &ts);
|
|
|
|
return ((long)ts.tv_sec * MS_PER_SEC) + (ts.tv_nsec / NS_PER_MS);
|
|
}
|
|
|
|
int cmd_wait_primary(int argc, char* argv[])
|
|
{
|
|
long timeout_ms = WAIT_PRIMARY_DEFAULT_MS;
|
|
long start_ms, elapsed_ms;
|
|
|
|
if (argc > 2) {
|
|
printf("wait-primary [timeout_ms]\n");
|
|
return -EINVAL;
|
|
}
|
|
|
|
if (argc == 2) {
|
|
char* end;
|
|
|
|
errno = 0;
|
|
timeout_ms = strtol(argv[1], &end, 10);
|
|
|
|
if ((errno != 0) || (end == argv[1]) || (*end != '\0')
|
|
|| (timeout_ms < 0)) {
|
|
printf("wait-primary: invalid timeout '%s' (expected a "
|
|
"non-negative integer in ms)\n",
|
|
argv[1]);
|
|
return -EINVAL;
|
|
}
|
|
}
|
|
|
|
start_ms = monotonic_ms();
|
|
|
|
/* The poll probes repeatedly and EXPECTS failure until a primary serves, so
|
|
* it must not spam the console: xcecli_ctrlstatus_primary suppresses its own
|
|
* per-attempt output, and we raise the client log threshold for the duration
|
|
* of the loop to silence libxce's per-attempt "xcecli failover: attempt N"
|
|
* INFO logs — RESTORED to the user's level on every exit path. */
|
|
mylog_setlevel((char*)"warn");
|
|
|
|
/* Probe, then re-check the monotonic deadline between attempts. Each probe
|
|
* may itself take time (libxce failover + redirect-to-primary has its own
|
|
* internal retry budget); we re-read the clock after every probe returns so
|
|
* the overall loop honors the deadline and returns failure in bounded time. */
|
|
for (;;) {
|
|
xce_ctrlstatus_t* status = NULL;
|
|
int rc = xcecli_ctrlstatus_primary(active_ctx->ctx,
|
|
&status);
|
|
|
|
/* Success ONLY on an authoritative reply from the serving primary
|
|
* (responderRole == PRIMARY, guaranteed by xcecli_ctrlstatus_primary).
|
|
* Print the served primary's identity (the responder IS the primary, so
|
|
* primaryName == responderName) and free the reply (Req 9.1, 9.2). */
|
|
if (rc == 0) {
|
|
const char* primary = (status != NULL)
|
|
? (NULLSTR(status->primaryName)
|
|
? status->responderName
|
|
: status->primaryName)
|
|
: NULL;
|
|
|
|
elapsed_ms = monotonic_ms() - start_ms;
|
|
mylog_setlevel(g_active_loglevel);
|
|
printf("wait-primary: primary '%s' serving after %ld ms\n",
|
|
NULLSTR(primary) ? "unknown" : primary, elapsed_ms);
|
|
xcecli_free_ctrlstatus(status);
|
|
return 0;
|
|
}
|
|
|
|
/* Non-authoritative / failed probe: nothing to free (the typed-failure
|
|
* contract leaves *status untouched), keep polling until the deadline. */
|
|
elapsed_ms = monotonic_ms() - start_ms;
|
|
|
|
if (elapsed_ms >= timeout_ms) {
|
|
mylog_setlevel(g_active_loglevel);
|
|
printf("wait-primary: no primary served within %ld ms (waited "
|
|
"%ld ms)\n",
|
|
timeout_ms, elapsed_ms);
|
|
return -ETIMEDOUT;
|
|
}
|
|
|
|
/* Don't oversleep past the deadline: cap the poll to the time left. */
|
|
{
|
|
long remaining_ms = timeout_ms - elapsed_ms;
|
|
long sleep_ms = (remaining_ms < WAIT_PRIMARY_POLL_MS)
|
|
? remaining_ms
|
|
: WAIT_PRIMARY_POLL_MS;
|
|
|
|
if (sleep_ms > 0) {
|
|
usleep((useconds_t)(sleep_ms * US_PER_MS));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/* Map a ctrl_role_t raw value (CtrlStatus.responderRole, see ha.h:60-63 and
|
|
* design.md "Schema — reply payload") to a display string. PRIMARY==0,
|
|
* STANDBY==1, INIT==2, ELECTING==3; anything else is reported verbatim so an
|
|
* unexpected role is never silently rendered as a known one. */
|
|
static const char* ctrlstatus_role_str(uint8_t role)
|
|
{
|
|
switch (role) {
|
|
case 0:
|
|
return "primary";
|
|
case 1:
|
|
return "standby";
|
|
case 2:
|
|
return "initializing";
|
|
case 3:
|
|
return "electing";
|
|
default:
|
|
return "unknown-role";
|
|
}
|
|
}
|
|
|
|
/* `ctrlstatus` (controller-status task 4.1, Req 1.3/3.3/5.1/5.2/8.1/8.2/8.3,
|
|
* design.md "cmd_ctrlstatus (test client) — State contract").
|
|
*
|
|
* Issues the any-node status read (xcecli_ctrlstatus, is_write=false: whatever
|
|
* controller answers — primary or standby hint view — rides the configured
|
|
* failover/rotation) and prints the identity/health fields in human-readable
|
|
* form.
|
|
*
|
|
* On failure (no reachable controller etc.) it prints a typed error and returns
|
|
* the negative rc so the session records the command as failed and exits
|
|
* non-zero (Req 8.3). On success it prints the fields and returns 0 (Req 8.2). */
|
|
int cmd_ctrlstatus(int argc, char* argv[])
|
|
{
|
|
xce_ctrlstatus_t* status = NULL;
|
|
int rc;
|
|
int i;
|
|
|
|
(void)argv;
|
|
|
|
if (argc != 1) {
|
|
printf("ctrlstatus\n");
|
|
return -EINVAL;
|
|
}
|
|
|
|
rc = xcecli_ctrlstatus(active_ctx->ctx, &status);
|
|
|
|
if (rc < 0) {
|
|
char ebuf[128];
|
|
|
|
ebuf[0] = '\0';
|
|
xce_strerror_r(-rc, ebuf, sizeof(ebuf));
|
|
SET_ERROR(-rc);
|
|
|
|
switch (-rc) {
|
|
case E_XCE_NOMASTER:
|
|
fprintf(stderr, "ctrlstatus failed: no-master (%s)\n", ebuf);
|
|
break;
|
|
case E_XCE_NOT_READY:
|
|
fprintf(stderr, "ctrlstatus failed: not-ready (%s)\n", ebuf);
|
|
break;
|
|
case ETIMEDOUT:
|
|
fprintf(stderr, "ctrlstatus failed: timeout (%s)\n", ebuf);
|
|
break;
|
|
default:
|
|
fprintf(stderr, "ctrlstatus failed: %s\n", ebuf);
|
|
break;
|
|
}
|
|
|
|
return rc;
|
|
}
|
|
|
|
/* Current primary (Req 2.x): an empty primaryName means no elected/known
|
|
* primary — render the well-defined "unknown" (Req 1.3/2.2/2.3) rather than
|
|
* an empty or stale value. */
|
|
printf("Primary: %s\n",
|
|
NULLSTR(status->primaryName) ? "unknown" : status->primaryName);
|
|
|
|
/* Answering node + role, framed as a property of the responder and kept
|
|
* DISTINCT from the authoritative Primary line above (Req 3.1/3.2/3.3): a
|
|
* transient responder role (electing/initializing) must not be misread as
|
|
* the cluster's primary state. */
|
|
printf("answered by %s (%s)\n",
|
|
NULLSTR(status->responderName) ? "unknown" : status->responderName,
|
|
ctrlstatus_role_str(status->responderRole));
|
|
|
|
printf("Cluster: %s\n",
|
|
NULLSTR(status->clusterName) ? "unknown" : status->clusterName);
|
|
|
|
/* Configured controller topology + count (Req 4.1/4.2). */
|
|
printf("Controllers (%u):\n", status->peerCount);
|
|
for (i = 0; i < status->peers.count; i++) {
|
|
xce_ctrlpeer_t* peer = status->peers.list[i];
|
|
|
|
if (peer == NULL) {
|
|
continue;
|
|
}
|
|
|
|
printf("\t%s\t%s:%u\n", NULLSTR(peer->name) ? "?" : peer->name,
|
|
NULLSTR(peer->host) ? "?" : peer->host, peer->port);
|
|
}
|
|
|
|
/* Config version + committed config version (Req 5.1) and the content
|
|
* checksum as the hex string the controller already exposes (Req 5.2). */
|
|
printf("configVersion: %llu\n", (unsigned long long)status->configVersion);
|
|
printf("committedConfigVersion: %llu\n",
|
|
(unsigned long long)status->committedConfigVersion);
|
|
printf("configChecksum: %s\n",
|
|
NULLSTR(status->configChecksum) ? "unknown" : status->configChecksum);
|
|
|
|
/* HA health / progress (Req 6.x). */
|
|
printf("term: %llu\n", (unsigned long long)status->term);
|
|
printf("committedLsn: %llu\n", (unsigned long long)status->committedLsn);
|
|
printf("appliedLsn: %llu\n", (unsigned long long)status->appliedLsn);
|
|
printf("fenced: %s\n", status->fenced ? "true" : "false");
|
|
|
|
/* Responder snapshot timestamp (Req 11.2/11.3): render the responder's
|
|
* sample-time wall clock (epoch ms) in the SAME local-time
|
|
* "YYYY-MM-DD HH:MM:SS,mmm" form the controller logger emits, so it can be
|
|
* compared directly against ctrl-N.log. */
|
|
{
|
|
char tbuf[64];
|
|
if (ctrlstatus_format_time_ms(status->responder_time_ms, tbuf,
|
|
sizeof(tbuf)) != NULL) {
|
|
printf("responderTime: %s\n", tbuf);
|
|
} else {
|
|
printf("responderTime: unknown\n");
|
|
}
|
|
}
|
|
|
|
xcecli_free_ctrlstatus(status);
|
|
|
|
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;
|
|
/* Initialize so a "rc >= 0 but *jobid never written" outcome (a non-clean
|
|
* ack such as a redirect/zero-row/not-ready surfaced as a non-negative rc)
|
|
* is detectable as an invalid (zero) JobId rather than printed as garbage
|
|
* (bug-786). */
|
|
uint64_t jobid = 0;
|
|
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");
|
|
rc = -EINVAL;
|
|
goto cleanup;
|
|
}
|
|
}
|
|
|
|
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");
|
|
rc = -EINVAL;
|
|
goto cleanup;
|
|
}
|
|
|
|
if (NULLSTR(submit.queue)) {
|
|
printf("you need to specify job queue with -q\n");
|
|
rc = -EINVAL;
|
|
goto cleanup;
|
|
}
|
|
|
|
if (NULLSTR(submit.start_info->cmdline)) {
|
|
printf("you need to specify job command with -c\n");
|
|
rc = -EINVAL;
|
|
goto cleanup;
|
|
}
|
|
|
|
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);
|
|
|
|
cleanup:
|
|
FREEUP(submit.user);
|
|
FREEUP(submit.queue);
|
|
FREEUP(submit.resreq);
|
|
FREEUP(submit.start_info->cmdline);
|
|
FREEUP(submit.start_info->exdata);
|
|
FREEUP(submit.start_info);
|
|
|
|
/* Report success only on a clean ack that echoes back the EXACT JobId we
|
|
* submitted (bug-786, Req 7.1/7.2/7.3). xce_jobsubmit() writes *jobid only
|
|
* when the reply status is >= 0; XCECLI_FAILOVER_CALL surfaces a non-clean
|
|
* outcome as a negative rc (e.g. -E_XCE_NOT_READY on no-primary budget
|
|
* exhaustion, -E_XCE_REDIRECT / -E_XCE_NOMASTER on a standby, -ETIMEDOUT /
|
|
* -ENOTCONN on transport loss).
|
|
*
|
|
* The `jobid != 0` gate alone is NOT sufficient: on a failover, libxce's
|
|
* handle_sub_reply() returns -1 on a non-jobid reply but LEAVES the stale
|
|
* previous value in ctx->reply.u.jobid, and xceapi.c copies it out whenever
|
|
* status >= 0 — so a rejected submit can surface as rc >= 0 with a NON-ZERO
|
|
* stale-garbage jobid (the live soak captured `63-15--1-1015808[32751]`
|
|
* while the controller logged "handle_jobsubmit fail to lock job" and never
|
|
* committed it). We therefore require the returned jobid to EQUAL the JobId
|
|
* we generated locally and submitted: on a genuine clean ack the controller
|
|
* commits and echoes back exactly submit->jobid (handle_jobsubmit replies
|
|
* reply.data.jobid = job_submit() = base_id = submit->jobid; verified
|
|
* round-trip). `jobid != jobsubmit.jobid` covers BOTH the never-written
|
|
* (zero) case AND the stale non-zero garbage case in one uint64 compare. */
|
|
if (rc < 0) {
|
|
char ebuf[128];
|
|
|
|
ebuf[0] = '\0';
|
|
xce_strerror_r(-rc, ebuf, sizeof(ebuf));
|
|
SET_ERROR(-rc);
|
|
|
|
switch (-rc) {
|
|
case E_XCE_NOMASTER:
|
|
fprintf(stderr, "jobsubmit failed: no-master (%s)\n", ebuf);
|
|
break;
|
|
case E_XCE_REDIRECT:
|
|
fprintf(stderr, "jobsubmit failed: redirect (%s)\n", ebuf);
|
|
break;
|
|
case E_XCE_NOT_READY:
|
|
fprintf(stderr, "jobsubmit failed: not-ready (%s)\n", ebuf);
|
|
break;
|
|
case ETIMEDOUT:
|
|
fprintf(stderr, "jobsubmit failed: timeout (%s)\n", ebuf);
|
|
break;
|
|
default:
|
|
fprintf(stderr, "jobsubmit failed: %s\n", ebuf);
|
|
break;
|
|
}
|
|
|
|
return rc;
|
|
}
|
|
|
|
if (jobid != jobsubmit.jobid) {
|
|
/* Non-negative rc but the returned JobId does not match the one we
|
|
* submitted: not a clean ack. This rejects BOTH the never-written
|
|
* (jobid == 0) case AND the stale-garbage non-zero case left in the
|
|
* libxce reply union on a failover-rejected submit. Never print "ok"
|
|
* with an uninitialized/garbage id (Req 7.3). */
|
|
char wantStr[128];
|
|
char gotStr[128];
|
|
|
|
fmt_jobid(wantStr, sizeof(wantStr) - 1, jobsubmit.jobid);
|
|
fmt_jobid(gotStr, sizeof(gotStr) - 1, jobid);
|
|
fprintf(stderr,
|
|
"jobsubmit failed: no clean ack (controller did not confirm "
|
|
"the submitted JobId: wanted %s, got %s)\n",
|
|
wantStr, gotStr);
|
|
return -E_XCE_INTERNAL;
|
|
}
|
|
|
|
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, array_jobid(jobid));
|
|
|
|
/* Capture $jobid (Req 3.2) using the SAME representation cmd_joblist prints
|
|
* for a job's JobId column — fmt_jobid(jobs[i]->jobid), the full jobid
|
|
* WITHOUT array_jobid() stripping — so a later `$jobid` substitution and
|
|
* `expect-job $jobid` compare equal to the joblist entry for this job. Only
|
|
* set on this clean-ack + valid-id path; never on a failure path, so an
|
|
* unset $jobid still fails per Req 3.6. */
|
|
fmt_jobid(g_last_jobid, sizeof(g_last_jobid) - 1, 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;
|
|
}
|
|
|
|
/* errexit toggle (xce-test-e2e-primitives task 1.1), shell `-e`/`+e` style.
|
|
* Default is mode-dependent (batch `-e` abort-on-failure, interactive `+e`
|
|
* continue); these override it for the rest of the session. The session
|
|
* still exits non-zero if any command failed. Handled here ahead of the
|
|
* per-context key/value path. `continue-on-error` == `+e` (back-compat). */
|
|
if ((argc == 2) &&
|
|
((strcmp(argv[1], "+e") == 0) ||
|
|
(strcmp(argv[1], "continue-on-error") == 0))) {
|
|
g_continue_on_error = 1; /* +e: keep the read loop running on failure */
|
|
return 0;
|
|
}
|
|
if ((argc == 2) && (strcmp(argv[1], "-e") == 0)) {
|
|
g_continue_on_error = 0; /* -e: abort the read loop on first failure */
|
|
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);
|
|
}
|
|
|
|
/* Remember the user's level so wait-primary can restore it after temporarily
|
|
* raising the threshold to silence its poll-loop failover spam. */
|
|
g_active_loglevel = loglevel;
|
|
|
|
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);
|
|
|
|
/* errexit default, modelled on shell `-e`/`+e`, keyed on the stdin mode:
|
|
* batch (piped/redirected stdin) → `-e`: abort the read loop on the first
|
|
* failing command and exit non-zero (a script wants fail-fast + verdict).
|
|
* interactive (TTY) REPL → `+e`: a failed command prints its error and the
|
|
* loop CONTINUES (never exit on a typo).
|
|
* Override at any time with `set -e` / `set +e` (or `set continue-on-error`). */
|
|
g_interactive = isatty(STDIN_FILENO);
|
|
g_continue_on_error = g_interactive ? 1 : 0;
|
|
|
|
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 = preprocess_and_run(s);
|
|
free(buf);
|
|
if (rc) {
|
|
break; /* abort-on-failure requested loop stop */
|
|
}
|
|
}
|
|
|
|
return g_session_failed ? 1 : 0;
|
|
}
|