lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(worker): job ids print human-readable format via fmt_jobid() (bug-1588)

Browse source 
Extends bug-1588's job.c fix to the remaining raw-%lu jobid logging sites:
8 worker daemon files (xcecd_worker.c, controller_connector.c, uv_ipc.c,
xcecd_worker_utils.c, xcecd_dbus_manager.c, xcecd.c, main.c, xcecd_cgroups.c)
and src/controller/testsuite/test.c (xce_test, heavily used for controller
testing). 99 call sites converted to fmt_jobid()+%s, matching the existing
convention. Non-jobid %lu values (lsn, client id, byte offset) and raw-numeric
identifier/path construction (checkpoint dir, systemd unit name, XCE_JOBID
env var) are deliberately left untouched.

Verified via /xce-build worker and controller (asan, --incremental): both
compile clean with zero warnings tied to the change; worker suite 4/4 OK,
controller suite 4/5 OK (the one failure is the pre-existing bug-1561
overlayfs fdatasync flake, unrelated to this change).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-08-22 02:42:01 +00:00
parent 811b5a11ec
commit 1c906e955c
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
9 changed files with 289 additions and 163 deletions
Download patch file Download diff file Expand all files Collapse all files

44
src/controller/testsuite/test.c
View file

@ -2562,6 +2562,7 @@ int cmd_jobkill(int argc, char* argv[])
int rc;
xce_jobctrl_t jobctrl;
xce_signal_t siginfo;
char jobidStr[64];
optind = 1;
signo = SIGTERM;
@ -2596,7 +2597,8 @@ int cmd_jobkill(int argc, char* argv[])
return rc;
}
printf("Job %lu signaled\n", jobid);
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, jobid);
printf("Job %s signaled\n", jobidStr);
return 0;
}
@ -2608,6 +2610,7 @@ int cmd_jobckpt(int argc, char* argv[])
int rc;
xce_jobctrl_t jobctrl;
xce_ckpt_options_t ckptoption;
char jobidStr[64];
optind = 1;
ckptoption.kill = 0;
@ -2646,7 +2649,8 @@ int cmd_jobckpt(int argc, char* argv[])
return rc;
}
printf("Job %lu checkpointed\n", jobid);
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, jobid);
printf("Job %s checkpointed\n", jobidStr);
return 0;
}
@ -3904,6 +3908,7 @@ static int run_task_local(uint64_t jobid, job_startinfo_t* startinfo,
int ntasks)
{
pid_t pid;
char jobidStr[64];
pid = fork();
if (pid < 0) {
@ -3963,7 +3968,8 @@ static int run_task_local(uint64_t jobid, job_startinfo_t* startinfo,
exit(rc);
}
printf("Job %lu started, PID: %d\n", jobid, pid);
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, jobid);
printf("Job %s started, PID: %d\n", jobidStr, pid);
return pid;
}
@ -4037,21 +4043,27 @@ static int handle_startjob(xce_request_t* request)
static int handle_notify(xce_request_t* request)
{
xce_notify_data_t* data = request->body.notifymsg;
char jobidStr[64];
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,
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, data->data.taskinfo->jobid);
printf("job %s host %s taskidx %d pid %d\n",
jobidStr, 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,
fmt_jobid(jobidStr, sizeof(jobidStr) - 1,
data->data.taskfailinfo->jobid);
printf("job %s host %s task %d error code %d\n",
jobidStr, 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,
fmt_jobid(jobidStr, sizeof(jobidStr) - 1,
data->data.taskfinish->jobid);
printf("job %s host %s task %d exit code %d\n",
jobidStr, data->data.taskfinish->host,
data->data.taskfinish->taskidx,
data->data.taskfinish->exit_code);
break;
@ -4132,6 +4144,7 @@ static int handle_runtasks(xce_request_t* request)
xce_taskrun_t* runmsg = request->body.runtaskmsg;
int rc = 0;
int i = 0;
char jobidStr[64];
#if 0
job_startinfo_t startinfo;
#endif
@ -4152,7 +4165,8 @@ static int handle_runtasks(xce_request_t* request)
return rc;
}
printf("run task: jobid = %lu\n", runmsg->jobid);
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, runmsg->jobid);
printf("run task: jobid = %s\n", jobidStr);
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);
@ -4177,7 +4191,7 @@ static int handle_runtasks(xce_request_t* request)
} else {
int idx;
printf("job %lu task %d started, PID: %d\n", runmsg->jobid, i, pid);
printf("job %s task %d started, PID: %d\n", jobidStr, i, pid);
idx = get_avail_run_task_slots();
if (idx < 0) {
@ -4282,6 +4296,7 @@ static void check_run_jobs()
xcecli_notify_jobresume(active_ctx->ctx, run_jobs[i].jobid);
} else if (WIFEXITED(wstatus) || WIFSIGNALED(wstatus)) {
xce_jobfinish_t finish;
char jobidStr[64];
finish.jobid = run_jobs[i].jobid;
@ -4291,7 +4306,8 @@ static void check_run_jobs()
finish.exit_code = 128 + WTERMSIG(wstatus);
}
printf("job %lu exit code %d\n", finish.jobid, finish.exit_code);
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, finish.jobid);
printf("job %s exit code %d\n", jobidStr, finish.exit_code);
xcecli_notify_jobfinish(active_ctx->ctx, &finish);
run_jobs[i].jobid = 0;
run_jobs[i].pid = 0;
@ -4343,6 +4359,7 @@ static void check_run_tasks()
.taskidx = run_tasks[i].taskid,
.clientidx = run_tasks[i].clientidx,
.exit_code = 0 };
char jobidStr[64];
if (WIFEXITED(wstatus)) {
finish.exit_code = WEXITSTATUS(wstatus);
@ -4350,7 +4367,8 @@ static void check_run_tasks()
finish.exit_code = 128 + WTERMSIG(wstatus);
}
printf("job %lu task %d exit code %d\n", finish.jobid,
fmt_jobid(jobidStr, sizeof(jobidStr) - 1, finish.jobid);
printf("job %s task %d exit code %d\n", jobidStr,
finish.taskidx, finish.exit_code);
xcecli_notify_taskfinish(active_ctx->ctx, &finish);
run_tasks[i].jobid = 0;

97
src/worker/src/controller_connector.c
View file

@ -500,12 +500,15 @@ int checking_connection()
#ifndef XRUN
static void dump_startjob_request(xce_job_exec_t* d)
{
char jobidStr[64];
if (!d) {
return;
}
fmt_jobid(jobidStr, sizeof(jobidStr), d->jobid);
XCECD_LOG(XCE_LOG_DEBUG,
"Job ID: %lu, Command: %s, Exec User: %s, Exdata: %s", d->jobid,
"Job ID: %s, Command: %s, Exec User: %s, Exdata: %s", jobidStr,
d->start_info->cmdline ? d->start_info->cmdline : "NULL",
d->start_info->execuser ? d->start_info->execuser : "NULL",
d->start_info->exdata ? d->start_info->exdata : "NULL");
@ -680,11 +683,14 @@ static char* create_mpi_cmdline(char* cmd, char* type, xce_launch_t* launchinfo)
static int job_start_remote_task(uint64_t jobid, job_startinfo_t* startinfo,
xce_launch_t* launchinfo)
{
char jobidStr[64];
xce_taskhost_t hosts[launchinfo->nhosts - 1];
xce_taskrun_t taskrun;
int i;
int rc;
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
if (!startinfo || !launchinfo) {
XCECD_LOG(XCE_LOG_ERROR, "Invalid parameters for job_run_task_remote");
return -1;
@ -724,13 +730,13 @@ static int job_start_remote_task(uint64_t jobid, job_startinfo_t* startinfo,
taskrun.nhosts = launchinfo->nhosts - 1;
taskrun.jobid = jobid;
XCECD_LOG(XCE_LOG_DEBUG, "Launching %d remote tasks for job %lu",
taskrun.nhosts, jobid);
XCECD_LOG(XCE_LOG_DEBUG, "Launching %d remote tasks for job %s",
taskrun.nhosts, jobidStr);
xcecli_unblock_tasks(xcecli_ctx);
rc = xcecli_runtask(xcecli_ctx, &taskrun);
if (rc < 0) {
XCECD_LOG(XCE_LOG_ERROR, "Job %lu run remote task failed.", jobid);
XCECD_LOG(XCE_LOG_ERROR, "Job %s run remote task failed.", jobidStr);
}
} else {
rc = -ENOMEM;
@ -752,6 +758,7 @@ static int job_start_local_task(xce_request_t* request)
{
#ifndef XRUN
xce_job_exec_t* d = request->body.jobstartmsg;
char jobidStr[64];
dump_startjob_request(d);
@ -959,7 +966,8 @@ static int job_start_local_task(xce_request_t* request)
}
}
XCECD_LOG(XCE_LOG_DEBUG, "Job %lu ntasks is: %d", d->jobid,
fmt_jobid(jobidStr, sizeof(jobidStr), d->jobid);
XCECD_LOG(XCE_LOG_DEBUG, "Job %s ntasks is: %d", jobidStr,
job_ctx->ntasks);
/* Job limits */
@ -1017,7 +1025,8 @@ static int job_start_local_task(xce_request_t* request)
static int handle_startjob(xce_request_t* request)
{
int rc = 0;
int rc = 0;
char jobidStr[64];
XCECD_LOG(XCE_LOG_DEBUG, "Received startjob request from controller");
@ -1057,8 +1066,9 @@ static int handle_startjob(xce_request_t* request)
if (launchinfo->nhosts > 1) {
rc = job_start_remote_task(jobid, startinfo, launchinfo);
if (rc < 0) {
XCECD_LOG(XCE_LOG_ERROR, "Job %lu run remote task failed, rc %d.",
jobid, rc);
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
XCECD_LOG(XCE_LOG_ERROR, "Job %s run remote task failed, rc %d.",
jobidStr, rc);
xcecd_task_fail(jobid, request->body.runtaskmsg->clientidx, -1, rc);
return rc;
}
@ -1082,6 +1092,7 @@ static int handle_runtasks(xce_request_t* request)
XCECD_LOG(XCE_LOG_DEBUG, "Received runtasks request from controller");
xce_taskrun_t* runmsg = request->body.runtaskmsg;
char jobidStr[64];
if (!runmsg) {
XCECD_LOG(XCE_LOG_ERROR,
@ -1117,7 +1128,8 @@ static int handle_runtasks(xce_request_t* request)
return -1;
}
XCECD_LOG(XCE_LOG_DEBUG3, "run task: jobid = %lu", runmsg->jobid);
fmt_jobid(jobidStr, sizeof(jobidStr), runmsg->jobid);
XCECD_LOG(XCE_LOG_DEBUG3, "run task: jobid = %s", jobidStr);
XCECD_LOG(XCE_LOG_DEBUG3, "run task: user = %s",
runmsg->taskhosts[0]->execuser ? runmsg->taskhosts[0]->execuser
: "NULL");
@ -1357,10 +1369,12 @@ static int handle_fetchfile(xce_request_t* request)
int fd;
ssize_t nread;
uint32_t maxbytes;
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), msg->jobid);
XCECD_LOG(XCE_LOG_DEBUG,
"Fetchfile request: jobid=%lu path=%s offset=%lu max=%u",
msg->jobid, msg->filepath, msg->offset, msg->max_bytes);
"Fetchfile request: jobid=%s path=%s offset=%lu max=%u",
jobidStr, msg->filepath, msg->offset, msg->max_bytes);
memset(&freply, 0, sizeof(freply));
freply.jobid = msg->jobid;
@ -1434,6 +1448,7 @@ static int handle_jobcontrol(xce_request_t* request)
#ifndef XRUN
XCECD_LOG(XCE_LOG_DEBUG, "Received job control request from controller");
xce_jobctrl_t* d = request->body.jobctrlmsg;
char jobidStr[64];
if (d->jobid <= 0) {
XCECD_LOG(XCE_LOG_ERROR,
@ -1441,11 +1456,13 @@ static int handle_jobcontrol(xce_request_t* request)
return -1;
}
fmt_jobid(jobidStr, sizeof(jobidStr), d->jobid);
if (d->action == JobControl_u_siginfo) {
XCECD_LOG(
XCE_LOG_DEBUG,
"Received job control request to send SIGINFO for job ID: %lu",
d->jobid);
"Received job control request to send SIGINFO for job ID: %s",
jobidStr);
job_control_task_t* task_ctx = calloc(1, sizeof(job_control_task_t));
if (!task_ctx) {
@ -1466,8 +1483,8 @@ static int handle_jobcontrol(xce_request_t* request)
|| worker->job_status == JOB_STARTING) {
XCECD_LOG(XCE_LOG_ERROR,
"Job is pending or starting, controller should not "
"pass the signal to %lu.",
d->jobid);
"pass the signal to %s.",
jobidStr);
// really should not happen, but if it does, we handle it
// gracefully
@ -1476,8 +1493,8 @@ static int handle_jobcontrol(xce_request_t* request)
return 0;
}
} else {
XCECD_LOG(XCE_LOG_ERROR, "Failed to find worker for job ID: %lu",
d->jobid);
XCECD_LOG(XCE_LOG_ERROR, "Failed to find worker for job ID: %s",
jobidStr);
FREEUP(task_ctx);
return -1;
}
@ -1486,13 +1503,13 @@ static int handle_jobcontrol(xce_request_t* request)
if (!ipc_master_send_job_signal(task_ctx->jobid, task_ctx->signal,
task_ctx->force)) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to submit job control task for job ID: %lu",
d->jobid);
"Failed to submit job control task for job ID: %s",
jobidStr);
FREEUP(task_ctx);
return -1;
} else {
XCECD_LOG(XCE_LOG_DEBUG,
"Job control task submitted for job ID: %lu", d->jobid);
"Job control task submitted for job ID: %s", jobidStr);
}
FREEUP(task_ctx);
#else
@ -1501,8 +1518,8 @@ static int handle_jobcontrol(xce_request_t* request)
"Virtual cluster mode, handling job control task directly");
FREEUP(task_ctx);
XCECD_LOG(XCE_LOG_DEBUG,
"Job control task handled directly for job ID: %lu",
d->jobid);
"Job control task handled directly for job ID: %s",
jobidStr);
worker->killed_signal = d->u.siginfo->signo; // Mark job as killed
#endif
@ -1510,18 +1527,18 @@ static int handle_jobcontrol(xce_request_t* request)
#if !defined(VIRTUAL_CLUSTER)
XCECD_LOG(XCE_LOG_DEBUG,
"Received job control request to set checkpoint options for "
"job ID: %lu",
d->jobid);
"job ID: %s",
jobidStr);
if (!ipc_master_send_job_ckpt(d->jobid, d->u.ckptoption->kill,
d->u.ckptoption->method)) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to send checkpoint options for job ID: %lu",
d->jobid);
"Failed to send checkpoint options for job ID: %s",
jobidStr);
return -1;
} else {
XCECD_LOG(XCE_LOG_DEBUG, "Checkpoint options sent for job ID: %lu",
d->jobid);
XCECD_LOG(XCE_LOG_DEBUG, "Checkpoint options sent for job ID: %s",
jobidStr);
}
#else
XCECD_LOG(XCE_LOG_WARN, "Received job control request in virtual "
@ -1529,8 +1546,8 @@ static int handle_jobcontrol(xce_request_t* request)
#endif
} else {
XCECD_LOG(XCE_LOG_ERROR,
"Received unknown job control action: %d for job ID: %lu",
d->action, d->jobid);
"Received unknown job control action: %d for job ID: %s",
d->action, jobidStr);
return -1;
}
#endif // XRUN
@ -1759,7 +1776,10 @@ int xcecd_job_remote_start(uint64_t jobid,
const char* hostname,
const char* command)
{
int rc = 0;
int rc = 0;
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
if (!cc_ctx || !cc_ctx->worker_list) {
XCECD_LOG(XCE_LOG_ERROR,
@ -1783,7 +1803,7 @@ int xcecd_job_remote_start(uint64_t jobid,
jobid);
if (!worker) {
XCECD_LOG(XCE_LOG_ERROR,
"Worker not found for job ID %lu", jobid);
"Worker not found for job ID %s", jobidStr);
return -1;
}
@ -1811,15 +1831,15 @@ int xcecd_job_remote_start(uint64_t jobid,
rc = xcecli_runtask(xcecli_ctx, &taskrun);
if (rc < 0) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to start job %lu on remote host %s, rc: %d",
jobid, hostname, rc);
"Failed to start job %s on remote host %s, rc: %d",
jobidStr, hostname, rc);
xcecd_task_fail(jobid, -1, -1, rc);
return rc;
}
XCECD_LOG(XCE_LOG_DEBUG,
"Remote task of job %lu will run on %s",
jobid, hostname);
"Remote task of job %s will run on %s",
jobidStr, hostname);
return 0;
}
@ -1847,14 +1867,17 @@ int xcecd_jobpost_done(uint64_t jobid)
int xcecd_jobrusage_update(xce_job_rusages_t* rusages)
{
char jobidStr[64];
if (!rusages) {
XCECD_LOG(XCE_LOG_ERROR,
"Invalid rusage data.");
return -1;
}
fmt_jobid(jobidStr, sizeof(jobidStr), rusages->jobid);
XCECD_LOG(XCE_LOG_DEBUG,
"Updating job rusage to controller for job %lu", rusages->jobid);
"Updating job rusage to controller for job %s", jobidStr);
return xcecli_update_jusage(xcecli_ctx,
rusages->jobid,

27
src/worker/src/main.c
View file

@ -157,15 +157,20 @@ static void handle_message_from_queue(void)
static void master_disconnect_callback(uint64_t job_id)
{
XCECD_LOG(XCE_LOG_DEBUG, "Worker with client ID %lu disconnected", job_id);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_LOG(XCE_LOG_DEBUG, "Worker with client ID %s disconnected", jobidStr);
}
static void master_job_status_callback(job_status_t* job_status)
{
int ret = XCECD_OK;
int ret = XCECD_OK;
char jobidStr[64];
XCECD_LOG(XCE_LOG_INFO, "Job %lu status: %d, exdata: %s, %s",
job_status->job_id, job_status->status_code,
fmt_jobid(jobidStr, sizeof(jobidStr), job_status->job_id);
XCECD_LOG(XCE_LOG_INFO, "Job %s status: %d, exdata: %s, %s",
jobidStr, job_status->status_code,
job_status->start_exdata[0] ? job_status->start_exdata
: "no exdata",
job_status->status_message);
@ -222,10 +227,13 @@ static void master_heartbeat_callback(uv_stream_t* client)
static void master_ckpt_result_callback(uint64_t job_id, int status,
const char* restore_cmd)
{
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_LOG(
XCE_LOG_DEBUG,
"Received checkpoint result for job %lu: status %d, restore_cmd: %s",
job_id, status, restore_cmd ? restore_cmd : "NULL");
"Received checkpoint result for job %s: status %d, restore_cmd: %s",
jobidStr, status, restore_cmd ? restore_cmd : "NULL");
if (job_id <= 0) {
XCECD_LOG(XCE_LOG_ERROR, "Job ID is not correct.");
@ -238,12 +246,15 @@ static void master_ckpt_result_callback(uint64_t job_id, int status,
static void master_rusages_callback(xce_job_rusages_t* rusages)
{
char jobidStr[64];
if (!rusages) {
return;
}
XCECD_LOG(XCE_LOG_DEBUG, "Received rusage for job %lu with %d metrics.",
rusages->jobid, rusages->count);
fmt_jobid(jobidStr, sizeof(jobidStr), rusages->jobid);
XCECD_LOG(XCE_LOG_DEBUG, "Received rusage for job %s with %d metrics.",
jobidStr, rusages->count);
xcecd_jobrusage_update(rusages);

103
src/worker/src/uv_ipc.c
View file

@ -62,6 +62,7 @@ static job_client_map_t* job_client_map = NULL;
__attribute__((unused)) static void add_job_client_mapping(uint64_t job_id,
client_t* client)
{
char jobidStr[64];
job_client_map_t* mapping
= (job_client_map_t*)calloc(1, sizeof(job_client_map_t));
if (!mapping)
@ -74,7 +75,8 @@ __attribute__((unused)) static void add_job_client_mapping(uint64_t job_id,
client->current_job_id = job_id;
XCECD_LOG(XCE_LOG_DEBUG, "Added job %lu to client %lu mapping", job_id,
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_LOG(XCE_LOG_DEBUG, "Added job %s to client %lu mapping", jobidStr,
client->id);
}
@ -87,6 +89,7 @@ __attribute__((unused)) static void remove_job_client_mapping(uint64_t job_id)
{
job_client_map_t* current = job_client_map;
job_client_map_t* prev = NULL;
char jobidStr[64];
while (current) {
if (current->job_id == job_id) {
@ -100,8 +103,9 @@ __attribute__((unused)) static void remove_job_client_mapping(uint64_t job_id)
current->client->current_job_id = 0;
}
XCECD_LOG(XCE_LOG_DEBUG, "Removed job %lu from client mapping",
job_id);
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_LOG(XCE_LOG_DEBUG, "Removed job %s from client mapping",
jobidStr);
free(current);
return;
@ -488,28 +492,36 @@ static void on_master_read(uv_stream_t* client, ssize_t nread,
master_status_callback(&msg->data.status);
}
} else if (msg->type == MSG_JOB_SIGNAL_ACK) {
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), msg->data.signal.job_id);
XCECD_LOG(XCE_LOG_DEBUG,
"Received signal acknowledgment for job %lu, signal %d, "
"Received signal acknowledgment for job %s, signal %d, "
"force %d",
msg->data.signal.job_id, msg->data.signal.signal,
jobidStr, msg->data.signal.signal,
msg->data.signal.force);
} else if (msg->type == MSG_JOB_CKPT_ACK) {
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), msg->data.ckpt.job_id);
XCECD_LOG(XCE_LOG_DEBUG,
"Received checkpoint acknowledgment for job %lu, signal "
"Received checkpoint acknowledgment for job %s, signal "
"%d, method %s",
msg->data.ckpt.job_id, msg->data.ckpt.kill,
jobidStr, msg->data.ckpt.kill,
msg->data.ckpt.method);
} else if (msg->type == MSG_JOB_CKPT_RESULT) {
XCECD_LOG(XCE_LOG_DEBUG, "Received checkpoint result for job %lu",
msg->data.ckpt.job_id);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), msg->data.ckpt.job_id);
XCECD_LOG(XCE_LOG_DEBUG, "Received checkpoint result for job %s",
jobidStr);
if (master_ckpt_result_callback) {
master_ckpt_result_callback(msg->data.ckpt_result.job_id,
msg->data.ckpt_result.status,
msg->data.ckpt_result.restore_cmd);
}
} else if (msg->type == MSG_JOB_RUSAGES) {
XCECD_LOG(XCE_LOG_DEBUG, "Received rusage info for job %lu",
msg->data.rusages.job_id);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), msg->data.rusages.job_id);
XCECD_LOG(XCE_LOG_DEBUG, "Received rusage info for job %s",
jobidStr);
if (master_rusages_callback) {
// deserialize payload
xce_job_rusages_t* rusages
@ -638,8 +650,10 @@ static void on_worker_read(uv_stream_t* server, ssize_t nread,
worker_id = msg->data.assignment.job_id;
worker_job_callback(&msg->data.assignment);
} else if (msg->type == MSG_JOB_SIGNAL && worker_signal_callback) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Received signal %d for job %lu",
msg->data.signal.signal, msg->data.signal.job_id);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), msg->data.signal.job_id);
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Received signal %d for job %s",
msg->data.signal.signal, jobidStr);
worker_signal_callback(msg->data.signal.job_id,
msg->data.signal.signal,
@ -649,9 +663,11 @@ static void on_worker_read(uv_stream_t* server, ssize_t nread,
msg->data.signal.signal,
msg->data.signal.force);
} else if (msg->type == MSG_JOB_CKPT && worker_ckpt_callback) {
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), msg->data.ckpt.job_id);
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
"Received checkpoint request for job %lu",
msg->data.ckpt.job_id);
"Received checkpoint request for job %s",
jobidStr);
worker_ckpt_callback(msg->data.ckpt.job_id, msg->data.ckpt.kill,
msg->data.ckpt.method);
@ -977,6 +993,7 @@ void ipc_worker_send_status(job_runtime_info_t* job_info,
return;
ipc_message_t msg;
char jobidStr[64];
memset(&msg, 0, sizeof(msg));
msg.type = MSG_JOB_STATUS;
msg.data.status.job_id = job_info->job_id;
@ -1020,15 +1037,16 @@ void ipc_worker_send_status(job_runtime_info_t* job_info,
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Sending job status to master: %s",
status_message);
fmt_jobid(jobidStr, sizeof(jobidStr), job_info->job_id);
if (job_info->job_status == JOB_DONE) {
XCECD_WORKER_LOG(XCE_LOG_INFO, "Job %lu completed successfully.",
job_info->job_id);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Job %s completed successfully.",
jobidStr);
} else if (job_info->job_status == JOB_FAILED) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Job %lu failed: %s", job_info->job_id,
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Job %s failed: %s", jobidStr,
status_message);
} else if (job_info->job_status == JOB_EXITED) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Job %lu exited with code %d: %s",
job_info->job_id, job_info->exit_code, status_message);
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Job %s exited with code %d: %s",
jobidStr, job_info->exit_code, status_message);
}
}
@ -1232,9 +1250,12 @@ void ipc_worker_set_reconnection_callback(worker_reconnect_callback_t callback)
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
{
client_t* target_client = find_client_by_job_id(job_id);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
if (!target_client) {
XCECD_LOG(XCE_LOG_ERROR, "No worker found for job %lu", job_id);
XCECD_LOG(XCE_LOG_ERROR, "No worker found for job %s", jobidStr);
return false;
}
@ -1251,12 +1272,12 @@ bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
int r = uv_write(req, (uv_stream_t*)target_client->pipe, &buf, 1,
on_write_complete);
if (r == 0) {
XCECD_LOG(XCE_LOG_DEBUG, "Sent signal %d for job %lu to worker %d",
signal, job_id, target_client->worker_pid);
XCECD_LOG(XCE_LOG_DEBUG, "Sent signal %d for job %s to worker %d",
signal, jobidStr, target_client->worker_pid);
return true;
} else {
XCECD_LOG(XCE_LOG_ERROR, "Failed to send signal %d for job %lu: %s",
signal, job_id, uv_strerror(r));
XCECD_LOG(XCE_LOG_ERROR, "Failed to send signal %d for job %s: %s",
signal, jobidStr, uv_strerror(r));
free(req);
return false;
}
@ -1278,6 +1299,7 @@ void ipc_worker_send_signal_ack(uint64_t job_id, int signal, bool force)
return;
ipc_message_t msg;
char jobidStr[64];
memset(&msg, 0, sizeof(msg));
msg.type = MSG_JOB_SIGNAL_ACK;
msg.data.signal.job_id = job_id;
@ -1298,9 +1320,10 @@ void ipc_worker_send_signal_ack(uint64_t job_id, int signal, bool force)
}
}
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_WORKER_LOG(
XCE_LOG_DEBUG,
"Sent signal acknowledgment for job %lu, signal %d, force %d", job_id,
"Sent signal acknowledgment for job %s, signal %d, force %d", jobidStr,
signal, force);
}
@ -1334,9 +1357,12 @@ void ipc_worker_set_ckpt_callback(job_ckpt_callback_t callback)
bool ipc_master_send_job_ckpt(uint64_t job_id, int kill, const char* method)
{
client_t* target_client = find_client_by_job_id(job_id);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
if (!target_client) {
XCECD_LOG(XCE_LOG_ERROR, "No worker found for job %lu", job_id);
XCECD_LOG(XCE_LOG_ERROR, "No worker found for job %s", jobidStr);
return false;
}
@ -1354,13 +1380,13 @@ bool ipc_master_send_job_ckpt(uint64_t job_id, int kill, const char* method)
on_write_complete);
if (r == 0) {
XCECD_LOG(XCE_LOG_DEBUG,
"Sent checkpoint request for job %lu to worker %d",
job_id, target_client->worker_pid);
"Sent checkpoint request for job %s to worker %d",
jobidStr, target_client->worker_pid);
return true;
} else {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to send checkpoint request for job %lu: %s",
job_id, uv_strerror(r));
"Failed to send checkpoint request for job %s: %s",
jobidStr, uv_strerror(r));
free(req);
return false;
}
@ -1382,6 +1408,7 @@ bool ipc_worker_send_ckpt_ack(uint64_t job_id, int kill, const char* method)
return false;
ipc_message_t msg;
char jobidStr[64];
memset(&msg, 0, sizeof(msg));
msg.type = MSG_JOB_CKPT;
msg.data.ckpt.job_id = job_id;
@ -1401,7 +1428,8 @@ bool ipc_worker_send_ckpt_ack(uint64_t job_id, int kill, const char* method)
}
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Sent job checkpoint for job %lu", job_id);
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Sent job checkpoint for job %s", jobidStr);
return true;
}
@ -1440,6 +1468,7 @@ bool ipc_worker_send_ckpt_result(uint64_t job_id, int status,
return false;
ipc_message_t msg;
char jobidStr[64];
memset(&msg, 0, sizeof(msg));
msg.type = MSG_JOB_CKPT_RESULT;
msg.data.ckpt_result.job_id = job_id;
@ -1461,8 +1490,9 @@ bool ipc_worker_send_ckpt_result(uint64_t job_id, int status,
}
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Sent job checkpoint result for job %lu",
job_id);
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Sent job checkpoint result for job %s",
jobidStr);
return true;
}
@ -1478,6 +1508,7 @@ bool ipc_worker_send_rusages(xce_job_rusages_t* rusages)
return false;
ipc_message_t msg;
char jobidStr[64];
memset(&msg, 0, sizeof(msg));
msg.type = MSG_JOB_RUSAGES;
msg.data.rusages.job_id = rusages->jobid;
@ -1538,7 +1569,7 @@ bool ipc_worker_send_rusages(xce_job_rusages_t* rusages)
}
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Sent rusage info for job %lu",
rusages->jobid);
fmt_jobid(jobidStr, sizeof(jobidStr), rusages->jobid);
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Sent rusage info for job %s", jobidStr);
return true;
}

6
src/worker/src/xcecd.c
View file

@ -34,6 +34,7 @@ void xcecd_process_command(void* arg)
{
pid_t pid = 0;
char unit_name[256];
char jobidStr[64];
fork_task_t* ctx = (fork_task_t*)arg;
if (!ctx || !ctx->startinfo || !ctx->startinfo->cmdline) {
@ -57,8 +58,9 @@ void xcecd_process_command(void* arg)
if (!ctx->startinfo) {
xcecd_job_start_error(ctx->xce_jobid);
XCECD_LOG(XCE_LOG_FATAL, "No startinfo provided, job %lu start failed.",
ctx->xce_jobid);
fmt_jobid(jobidStr, sizeof(jobidStr), ctx->xce_jobid);
XCECD_LOG(XCE_LOG_FATAL, "No startinfo provided, job %s start failed.",
jobidStr);
return;
}

6
src/worker/src/xcecd_cgroups.c
View file

@ -1384,6 +1384,7 @@ xce_job_rusages_t* collect_job_rusages(uint64_t job_id)
int r;
char unit_name[256];
char unit_path[512];
char jobidStr[64];
xce_job_rusages_t* rusages = NULL;
rusages = (xce_job_rusages_t*)calloc(1, sizeof(xce_job_rusages_t));
@ -1699,10 +1700,11 @@ xce_job_rusages_t* collect_job_rusages(uint64_t job_id)
strcpy(pgids_str, "none");
}
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_LOG(XCE_LOG_DEBUG3,
"Job %lu rusage: CPU: %.2fs, Mem: %.2fMB, Swap: %.2fMB, Tasks: "
"Job %s rusage: CPU: %.2fs, Mem: %.2fMB, Swap: %.2fMB, Tasks: "
"%d, PIDs: %d [%s], PGIDs: %d [%s]",
job_id, rusages->usages[0]->value, rusages->usages[1]->value,
jobidStr, rusages->usages[0]->value, rusages->usages[1]->value,
rusages->usages[2]->value, (int)rusages->usages[3]->value,
rusages->npids, pids_str, rusages->npgids, pgids_str);
#endif

4
src/worker/src/xcecd_dbus_manager.c
View file

@ -408,6 +408,7 @@ void dbus_stop_all_workers(xcecd_worker_t** worker_list, sqlite3* db)
xcecd_worker_t* worker = *worker_list;
while (worker) {
char unit_name[256];
char jobidStr[64];
make_job_sd_unit_name(unit_name, sizeof(unit_name) - 1, "xcecd-worker",
worker->job_info.jobid);
@ -420,7 +421,8 @@ void dbus_stop_all_workers(xcecd_worker_t** worker_list, sqlite3* db)
update_worker_status(*worker_list, worker->worker_pid, WORKER_EXITED,
&job, db);
XCECD_LOG(XCE_LOG_WARN, "Job %lu killed.", worker->job_info.jobid);
fmt_jobid(jobidStr, sizeof(jobidStr), worker->job_info.jobid);
XCECD_LOG(XCE_LOG_WARN, "Job %s killed.", jobidStr);
worker = worker->next;
}

116
src/worker/src/xcecd_worker.c
View file

@ -80,7 +80,10 @@ static void free_xce_job_rusages(xce_job_rusages_t* rusages)
static void on_rusage_timer(uv_timer_t* timer)
{
if (jobid > 0) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Collecting rusage for job %lu", jobid);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Collecting rusage for job %s",
jobidStr);
xce_job_rusages_t* rusages = collect_job_rusages(jobid);
@ -589,8 +592,10 @@ static void check_and_reconnect()
static void xjob_start_vnc()
{
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"Starting VNC server for interactive job %lu", jobid);
"Starting VNC server for interactive job %s", jobidStr);
char vnc_start_cmd[MAXLINELEN] = { 0 };
if (access(XJOB_PRE_EXEC, F_OK) != 0) {
@ -652,13 +657,16 @@ static void xjob_start_vnc()
elapsed_time++;
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "VNC server started for job %lu", jobid);
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "VNC server started for job %s",
jobidStr);
}
static void xjob_stop_vnc()
{
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"Stopping VNC server for interactive job %lu", jobid);
"Stopping VNC server for interactive job %s", jobidStr);
char vnc_stop_cmd[MAXLINELEN] = { 0 };
if (access(XJOB_POST_EXEC, F_OK) != 0) {
@ -686,7 +694,8 @@ static void xjob_stop_vnc()
exit(1);
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "VNC server stopped for job %lu", jobid);
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "VNC server stopped for job %s",
jobidStr);
}
static void worker_job_callback(job_assignment_t* info)
@ -695,6 +704,7 @@ static void worker_job_callback(job_assignment_t* info)
char* url = NULL;
char* start_exdata_info = NULL;
int rc = 0;
char jobidStr[64];
if (!info) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Received null job assignment");
@ -715,6 +725,8 @@ static void worker_job_callback(job_assignment_t* info)
uv_proc_manager.job_id = jobid;
job_runtime_info.job_id = jobid;
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
if (info->q_pre_exec[0]) {
strncpy(job_runtime_info.q_pre_exec, info->q_pre_exec,
sizeof(job_runtime_info.q_pre_exec) - 1);
@ -735,8 +747,8 @@ static void worker_job_callback(job_assignment_t* info)
sizeof(job_runtime_info.job_post_exec) - 1);
}
XCECD_WORKER_LOG(XCE_LOG_INFO, "Received job: %s, job_id: %lu, exdata: %s",
info->cmd, jobid,
XCECD_WORKER_LOG(XCE_LOG_INFO, "Received job: %s, job_id: %s, exdata: %s",
info->cmd, jobidStr,
info->exdata[0] ? info->exdata : "no data");
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Job pre-exec: %s",
@ -761,21 +773,21 @@ static void worker_job_callback(job_assignment_t* info)
if (!parse_exdata(info->exdata, &mydata.env_vars, mydata.transfer_info,
jobid)) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to parse exdata of job %lu",
jobid);
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to parse exdata of job %s",
jobidStr);
exit(1);
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"Parsed exdata for job %lu: interactive=%d", jobid,
"Parsed exdata for job %s: interactive=%d", jobidStr,
mydata.transfer_info->interactive_mode);
int job_port
= check_port(port_info, mydata.transfer_info->interactive_mode);
if (job_port < 0) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Failed to check port availability for job %lu",
jobid);
"Failed to check port availability for job %s",
jobidStr);
exit(1);
} else if (job_port > 5900 && job_port < 10000) {
if (mydata.transfer_info->interactive_mode == TTY
@ -784,13 +796,13 @@ static void worker_job_callback(job_assignment_t* info)
} else if (mydata.transfer_info->interactive_mode == XJOB) {
port_info.xjob_port = job_port;
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Assigned port %d for job %lu",
job_port, jobid);
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Assigned port %d for job %s",
job_port, jobidStr);
} else if (job_port == 0) {
/* normal jobs don't need port info */
} else {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Invalid port %d for job %lu", job_port,
jobid);
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Invalid port %d for job %s", job_port,
jobidStr);
exit(1);
}
@ -842,7 +854,8 @@ static void worker_job_callback(job_assignment_t* info)
mydata.transfer_info->interactive_mode);
if (url) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"URL for interactive job %lu: %s", jobid, url);
"URL for interactive job %s: %s", jobidStr,
url);
}
start_exdata = calloc(1, sizeof(start_exdata_t));
@ -858,12 +871,13 @@ static void worker_job_callback(job_assignment_t* info)
if (!start_exdata_info) {
XCECD_WORKER_LOG(
XCE_LOG_ERROR,
"Failed to generate start_exdata_info for job %lu", jobid);
"Failed to generate start_exdata_info for job %s",
jobidStr);
exit(1);
} else {
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"Generated start_exdata_info for job %lu: %s",
jobid, start_exdata_info);
"Generated start_exdata_info for job %s: %s",
jobidStr, start_exdata_info);
}
strncpy(job_runtime_info.exdata, start_exdata_info,
@ -879,7 +893,7 @@ static void worker_job_callback(job_assignment_t* info)
ipc_worker_send_status(&job_runtime_info, "Running");
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
"Reported job %lu run status to master.", jobid);
"Reported job %s run status to master.", jobidStr);
FREEUP(job_cmd);
FREEUP(url);
@ -898,7 +912,8 @@ static void worker_job_callback(job_assignment_t* info)
ipc_worker_send_status(&job_runtime_info, "Failed to start process");
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
"Reported job %lu failed status to master.", jobid);
"Reported job %s failed status to master.",
jobidStr);
worker_exit_flag = 1;
return_code = 1;
}
@ -1074,13 +1089,18 @@ static bool check_process_status(pid_t target_pid, int expected_signal,
*/
static void worker_signal_callback(uint64_t job_id, int signal, bool force)
{
XCECD_WORKER_LOG(XCE_LOG_INFO, "Received signal %d for job %lu, force=%d",
signal, job_id, force);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Received signal %d for job %s, force=%d",
signal, jobidStr, force);
if (job_id != jobid) {
char curJobidStr[64];
fmt_jobid(curJobidStr, sizeof(curJobidStr), jobid);
XCECD_WORKER_LOG(XCE_LOG_WARN,
"Signal for job %lu but current job is %lu", job_id,
jobid);
"Signal for job %s but current job is %s", jobidStr,
curJobidStr);
return;
}
@ -1095,8 +1115,8 @@ static void worker_signal_callback(uint64_t job_id, int signal, bool force)
switch (signal) {
case SIGTERM:
case SIGKILL:
XCECD_WORKER_LOG(XCE_LOG_INFO, "Terminating job %lu with signal %d",
job_id, signal);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Terminating job %s with signal %d",
jobidStr, signal);
if (force) {
success = signal_gid_except_self(SIGKILL);
} else {
@ -1106,36 +1126,36 @@ static void worker_signal_callback(uint64_t job_id, int signal, bool force)
case SIGSTOP:
case SIGTSTP:
XCECD_WORKER_LOG(XCE_LOG_INFO, "Pausing job %lu", job_id);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Pausing job %s", jobidStr);
success = signal_gid_except_self(SIGSTOP);
if (success) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG2, "Job %lu paused successfully",
job_id);
XCECD_WORKER_LOG(XCE_LOG_DEBUG2, "Job %s paused successfully",
jobidStr);
job_runtime_info.job_status = JOB_SUSPENDED;
ipc_worker_send_status(&job_runtime_info, "Suspended");
} else {
XCECD_WORKER_LOG(XCE_LOG_WARN, "Failed to pause job %lu", job_id);
XCECD_WORKER_LOG(XCE_LOG_WARN, "Failed to pause job %s", jobidStr);
}
break;
case SIGCONT:
XCECD_WORKER_LOG(XCE_LOG_INFO, "Resuming job %lu", job_id);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Resuming job %s", jobidStr);
success = signal_gid_except_self(SIGCONT);
if (success) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG2, "Job %lu resumed successfully",
job_id);
XCECD_WORKER_LOG(XCE_LOG_DEBUG2, "Job %s resumed successfully",
jobidStr);
job_runtime_info.job_status = JOB_RUNNING;
ipc_worker_send_status(&job_runtime_info, "Resumed");
} else {
XCECD_WORKER_LOG(XCE_LOG_WARN, "Failed to resume job %lu", job_id);
XCECD_WORKER_LOG(XCE_LOG_WARN, "Failed to resume job %s", jobidStr);
}
break;
default:
XCECD_WORKER_LOG(XCE_LOG_INFO, "Sending signal %d for job %lu", signal,
job_id);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Sending signal %d for job %s", signal,
jobidStr);
success = signal_gid_except_self(signal);
break;
}
@ -1169,12 +1189,14 @@ static char* criu_path()
static void do_criu_ckpt(uint64_t job_id, int kill)
{
pid_t target_pid = 0;
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
char* path = criu_path();
if (!path) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"CRIU is not available for checkpointing job %lu",
job_id);
"CRIU is not available for checkpointing job %s",
jobidStr);
ipc_worker_send_ckpt_result(job_id, 1, NULL);
return;
@ -1275,9 +1297,9 @@ static void do_criu_ckpt(uint64_t job_id, int kill)
}
fclose(log_file);
if (has_error) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"CRIU dump failed for job %lu, check %s for details",
job_id, dump_log);
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"CRIU dump failed for job %s, check %s for details",
jobidStr, dump_log);
ipc_worker_send_ckpt_result(job_id, 1, NULL);
FREEUP(path);
return;
@ -1299,7 +1321,10 @@ static void do_criu_ckpt(uint64_t job_id, int kill)
static void worker_ckpt_callback(uint64_t job_id, int kill, const char* method)
{
XCECD_WORKER_LOG(XCE_LOG_INFO, "Checkpointing job %lu", job_id);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Checkpointing job %s", jobidStr);
job_runtime_info.job_id = job_id;
@ -1311,7 +1336,8 @@ static void worker_ckpt_callback(uint64_t job_id, int kill, const char* method)
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
"Reported job %lu checkpoint status to master.", job_id);
"Reported job %s checkpoint status to master.",
jobidStr);
}
int main(int argc, char* argv[])

49
src/worker/src/xcecd_worker_utils.c
View file

@ -198,7 +198,8 @@ void remove_worker(xcecd_worker_t** worker_list, int worker_pid)
int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
int worker_status, xcecd_job_t* job, sqlite3* db)
{
int ret = XCECD_OK;
int ret = XCECD_OK;
char jobidStr[64];
if (!worker_list) {
XCECD_LOG(XCE_LOG_DEBUG, "Worker list is empty");
@ -226,6 +227,7 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
worker->job_info.pidinfo.pids[0] = (unsigned int)job->job_pid;
}
job->job_id = worker->job_info.jobid;
fmt_jobid(jobidStr, sizeof(jobidStr), job->job_id);
if (job->job_status == JOB_DONE || job->job_status == JOB_EXITED) {
worker->job_status = job->job_status;
@ -237,8 +239,8 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
&& (job->job_status == JOB_RUNNING)) {
XCECD_LOG(
XCE_LOG_DEBUG,
"Job ID %lu running - sending status to controller",
job->job_id);
"Job ID %s running - sending status to controller",
jobidStr);
worker->job_status = JOB_RUNNING;
xcecd_reply_startjob(job);
@ -252,8 +254,8 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
}
} else if (job->job_status == JOB_SUSPENDED) {
// Send controller that job is stopped
XCECD_LOG(XCE_LOG_DEBUG, "Job %lu is suspended.",
job->job_id);
XCECD_LOG(XCE_LOG_DEBUG, "Job %s is suspended.",
jobidStr);
worker->job_status = JOB_SUSPENDED;
xcecd_job_stopped(job->job_id);
@ -268,8 +270,8 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
}
} else if (worker->job_status == JOB_SUSPENDED
&& job->job_status == JOB_RUNNING) {
XCECD_LOG(XCE_LOG_DEBUG, "Job %lu is resumed.",
job->job_id);
XCECD_LOG(XCE_LOG_DEBUG, "Job %s is resumed.",
jobidStr);
worker->job_status = JOB_RUNNING;
xcecd_job_resumed(job->job_id);
@ -292,8 +294,8 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
worker->worker_status = WORKER_DONE;
worker->job_status = JOB_DONE;
XCECD_LOG(XCE_LOG_DEBUG,
"Job %lu done and sending status to controller",
job->job_id);
"Job %s done and sending status to controller",
jobidStr);
xcecd_job_finished(job);
/* FIXME: add error handling when msg send failure */
@ -312,8 +314,8 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
worker->worker_status = WORKER_EXITED;
worker->job_status = JOB_EXITED;
XCECD_LOG(XCE_LOG_DEBUG,
"Job %lu exited and sending status to controller",
job->job_id);
"Job %s exited and sending status to controller",
jobidStr);
xcecd_job_finished(job);
if (db
@ -331,9 +333,9 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
worker->worker_status = WORKER_UNKNOWN;
worker->job_status = JOB_EXITED;
XCECD_LOG(XCE_LOG_DEBUG,
"Job %lu status is unknown and sending status to "
"Job %s status is unknown and sending status to "
"controller",
job->job_id);
jobidStr);
xcecd_job_finished(job);
/* FIXME: we don't have unknown status currently */
@ -424,8 +426,11 @@ xcecd_worker_t* find_worker_by_pid(void* worker_list, int pid)
xcecd_worker_t* find_worker_by_jobid(xcecd_worker_t* worker_list,
uint64_t jobid)
{
XCECD_LOG(XCE_LOG_DEBUG, "Finding worker with job ID %lu in worker list",
jobid);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
XCECD_LOG(XCE_LOG_DEBUG, "Finding worker with job ID %s in worker list",
jobidStr);
xcecd_worker_t* worker = worker_list;
while (worker) {
@ -459,8 +464,11 @@ bool resend_job_status(xcecd_worker_t* worker_list)
xcecd_worker_t* worker = worker_list;
while (worker) {
if (worker->worker_status == WORKER_RUNNING) {
XCECD_LOG(XCE_LOG_DEBUG, "Resending job %lu status to controller.",
worker->job_info.jobid);
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), worker->job_info.jobid);
XCECD_LOG(XCE_LOG_DEBUG, "Resending job %s status to controller.",
jobidStr);
job.job_id = worker->job_info.jobid;
xcecd_reply_startjob(&job);
@ -499,11 +507,14 @@ bool check_worker_sleep_time_elapsed(xcecd_worker_t* worker)
double time_diff = difftime(current_time, worker->start_time);
if (time_diff >= worker->sleep_time) {
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), worker->job_info.jobid);
XCECD_LOG(
XCE_LOG_DEBUG,
"Sleep time %.0f >= %d seconds elapsed for worker %d (job %lu)",
"Sleep time %.0f >= %d seconds elapsed for worker %d (job %s)",
time_diff, worker->sleep_time, worker->worker_pid,
worker->job_info.jobid);
jobidStr);
return true;
} else {
// XCECD_LOG(XCE_LOG_TRACE, "Sleep time remaining for worker %d: %.0f