From c350d8aca69f3245caa20d73669b733c6845b650 Mon Sep 17 00:00:00 2001 From: Paul Menzel Date: Mon, 25 May 2026 10:44:15 +0200 Subject: [PATCH 1/2] mxqsub: add --callback/-c option to run a script when a job finishes mxqsub accepts --callback=SCRIPT (absolute path). The daemon runs the script after every terminal outcome (finished, failed, killed, unknown), as the submitting user in the job workdir, with MXQ_JOB_ID, MXQ_GROUP_ID, MXQ_JOB_STATUS, and MXQ_JOB_WORKDIR in the environment. Adds job_callback column to mxq_job; migration in migrate_019. Also fixes mxq_set_job_status_unknown not updating job->job_status in-struct, inconsistent with the other status-update functions. Co-Authored-By: Claude Sonnet 4.6 --- mxq_job.c | 7 +++- mxq_job.h | 1 + mxqd.c | 50 ++++++++++++++++++++++++++ mxqsub.c | 26 ++++++++++++-- mysql/create_tables.sql | 1 + mysql/migrate_019_add_job_callback.sql | 2 ++ 6 files changed, 84 insertions(+), 3 deletions(-) create mode 100644 mysql/migrate_019_add_job_callback.sql diff --git a/mxq_job.c b/mxq_job.c index daf4061e..3d02665c 100644 --- a/mxq_job.c +++ b/mxq_job.c @@ -18,7 +18,7 @@ #include "mxq_group.h" #include "mxq_job.h" -#define JOB_FIELDS_CNT 37 +#define JOB_FIELDS_CNT 38 #define JOB_FIELDS \ " job_id, " \ " job_status, " \ @@ -29,6 +29,7 @@ " job_argv, " \ " job_stdout, " \ " job_stderr, " \ + " job_callback, " \ " job_umask, " \ " host_submit, " \ " host_id, " \ @@ -73,6 +74,7 @@ static void bind_result_job_fields(struct mx_mysql_bind *result, struct mxq_job mx_mysql_bind_var(result, idx++, string, &(j->job_argv_str)); mx_mysql_bind_var(result, idx++, string, &(j->job_stdout)); mx_mysql_bind_var(result, idx++, string, &(j->job_stderr)); + mx_mysql_bind_var(result, idx++, string, &(j->job_callback)); mx_mysql_bind_var(result, idx++, uint32, &(j->job_umask)); mx_mysql_bind_var(result, idx++, string, &(j->host_submit)); mx_mysql_bind_var(result, idx++, string, &(j->host_id)); @@ -136,6 +138,7 @@ void mxq_job_free_content(struct mxq_job *j) mx_free_null(j->job_argv_str); mx_free_null(j->job_stdout); mx_free_null(j->job_stderr); + mx_free_null(j->job_callback); if (j->tmp_stderr == j->tmp_stdout) { j->tmp_stdout = NULL; @@ -605,6 +608,8 @@ int mxq_set_job_status_unknown(struct mx_mysql *mysql, struct mxq_job *job) return res; } + job->job_status = MXQ_JOB_STATUS_UNKNOWN; + return res; } diff --git a/mxq_job.h b/mxq_job.h index d73561cb..c008a026 100644 --- a/mxq_job.h +++ b/mxq_job.h @@ -29,6 +29,7 @@ struct mxq_job { char * job_stdout; char * job_stderr; + char * job_callback; char * tmp_stdout; char * tmp_stderr; diff --git a/mxqd.c b/mxqd.c index 3f572a8e..6b62a37b 100644 --- a/mxqd.c +++ b/mxqd.c @@ -1928,6 +1928,52 @@ static void release_gpu(struct mxq_server *server, struct mxq_group *group, stru } } +static void run_job_callback(struct mxq_group *group, struct mxq_job *job) +{ + if (!job->job_callback || !*job->job_callback) + return; + + mx_log_info("job=%s(%d):%lu:%lu :: running callback: %s", + group->user_name, group->user_uid, group->group_id, job->job_id, + job->job_callback); + + pid_t pid = fork(); + if (pid < 0) { + mx_log_err("job=%s(%d):%lu:%lu callback fork(): %m", + group->user_name, group->user_uid, group->group_id, job->job_id); + return; + } + + if (pid == 0) { + if (initgroups(group->user_name, group->user_gid) == -1) + _exit(1); + if (setregid(group->user_gid, group->user_gid) == -1) + _exit(1); + if (setreuid(group->user_uid, group->user_uid) == -1) + _exit(1); + if (chdir(job->job_workdir) == -1) + _exit(1); + + mx_setenvf_forever("MXQ_JOB_ID", "%lu", job->job_id); + mx_setenvf_forever("MXQ_GROUP_ID", "%lu", job->group_id); + mx_setenv_forever("MXQ_JOB_STATUS", mxq_job_status_to_name(job->job_status)); + mx_setenv_forever("MXQ_JOB_WORKDIR", job->job_workdir); + + execl(job->job_callback, job->job_callback, NULL); + _exit(1); + } + + int status; + if (waitpid(pid, &status, 0) == -1) { + mx_log_err("job=%s(%d):%lu:%lu callback waitpid(): %m", + group->user_name, group->user_uid, group->group_id, job->job_id); + return; + } + if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) + mx_log_warning("job=%s(%d):%lu:%lu :: callback exited with non-zero status", + group->user_name, group->user_uid, group->group_id, job->job_id); +} + static int job_has_finished(struct mxq_server *server, struct mxq_group *group, struct mxq_job_list *jlist) { int cnt; @@ -1942,6 +1988,8 @@ static int job_has_finished(struct mxq_server *server, struct mxq_group *group, rename_outfiles(server, group, job); + run_job_callback(group, job); + cnt = jlist->group->slots_per_job; cpuset_clear_running(&server->cpu_set_running, &job->host_cpu_set); release_gpu(server, group, job); @@ -1966,6 +2014,8 @@ static int job_is_lost(struct mxq_server *server,struct mxq_group *group, struct rename_outfiles(server, group, job); + run_job_callback(group, job); + cnt = jlist->group->slots_per_job; cpuset_clear_running(&server->cpu_set_running, &job->host_cpu_set); release_gpu(server, group, job); diff --git a/mxqsub.c b/mxqsub.c index 72893be2..821b281e 100644 --- a/mxqsub.c +++ b/mxqsub.c @@ -60,6 +60,10 @@ static void print_usage(void) " -e, --stderr=FILE set file to capture stderr (default: )\n" " -u, --umask=MASK set mode to use as umask (default: current umask)\n" " -p, --priority=PRIORITY set priority (default: 127)\n" + " -c, --callback=SCRIPT execute SCRIPT when the job finishes (any outcome)\n" + " runs as the submitting user in the job workdir;\n" + " MXQ_JOB_ID, MXQ_GROUP_ID, MXQ_JOB_STATUS, and\n" + " MXQ_JOB_WORKDIR are set in the environment\n" "\n" "Job resource information:\n" " Scheduling is done based on the resources a job needs and\n" @@ -517,6 +521,7 @@ static int add_job(struct mx_mysql *mysql, struct mxq_job *j) " job_stdout = ?," " job_stderr = ?," + " job_callback = ?," " job_umask = ?," @@ -535,8 +540,9 @@ static int add_job(struct mx_mysql *mysql, struct mxq_job *j) mx_mysql_statement_param_bind(stmt, 4, string, &(j->job_argv_str)); mx_mysql_statement_param_bind(stmt, 5, string, &(j->job_stdout)); mx_mysql_statement_param_bind(stmt, 6, string, &(j->job_stderr)); - mx_mysql_statement_param_bind(stmt, 7, uint32, &(j->job_umask)); - mx_mysql_statement_param_bind(stmt, 8, string, &(j->host_submit)); + mx_mysql_statement_param_bind(stmt, 7, string, &(j->job_callback)); + mx_mysql_statement_param_bind(stmt, 8, uint32, &(j->job_umask)); + mx_mysql_statement_param_bind(stmt, 9, string, &(j->host_submit)); res = mx_mysql_statement_execute(stmt, &num_rows); if (res < 0) { @@ -705,6 +711,7 @@ int main(int argc, char *argv[]) char arg_debug; u_int32_t arg_tmpdir; u_int16_t arg_gpu; + char *arg_callback; _mx_cleanup_free_ char *current_workdir = NULL; _mx_cleanup_free_ char *arg_stdout_absolute = NULL; @@ -769,6 +776,7 @@ int main(int argc, char *argv[]) MX_OPTION_REQUIRED_ARG("prerequisites", 10), MX_OPTION_REQUIRED_ARG("tags", 11), MX_OPTION_NO_ARG("gpu", 12), + MX_OPTION_REQUIRED_ARG("callback", 'c'), MX_OPTION_END }; @@ -798,6 +806,7 @@ int main(int argc, char *argv[]) arg_prerequisites = ""; arg_tags = NULL; arg_gpu = 0; + arg_callback = ""; arg_mysql_default_group = getenv("MXQ_MYSQL_DEFAULT_GROUP"); if (!arg_mysql_default_group) @@ -1021,6 +1030,18 @@ int main(int argc, char *argv[]) case 12: arg_gpu = 1; break; + + case 'c': + if (!(*optctl.optarg)) { + mx_log_crit("--callback '%s': String is empty.", optctl.optarg); + exit(EX_CONFIG); + } + if (optctl.optarg[0] != '/') { + mx_log_crit("--callback '%s': must be an absolute path.", optctl.optarg); + exit(EX_CONFIG); + } + arg_callback = optctl.optarg; + break; } } @@ -1133,6 +1154,7 @@ int main(int argc, char *argv[]) job.job_argc = argc; job.job_argv = argv; job.job_argv_str = arg_args; + job.job_callback = arg_callback; /******************************************************************/ diff --git a/mysql/create_tables.sql b/mysql/create_tables.sql index 41585f76..d73d34a0 100644 --- a/mysql/create_tables.sql +++ b/mysql/create_tables.sql @@ -83,6 +83,7 @@ CREATE TABLE IF NOT EXISTS mxq_job ( job_stdout VARCHAR(4096) NOT NULL DEFAULT '/dev/null', job_stderr VARCHAR(4096) NOT NULL DEFAULT '/dev/null', + job_callback VARCHAR(4096) NOT NULL DEFAULT '', job_umask INT4 NOT NULL, diff --git a/mysql/migrate_019_add_job_callback.sql b/mysql/migrate_019_add_job_callback.sql new file mode 100644 index 00000000..c411aba7 --- /dev/null +++ b/mysql/migrate_019_add_job_callback.sql @@ -0,0 +1,2 @@ +ALTER TABLE mxq_job + ADD COLUMN job_callback VARCHAR(4096) NOT NULL DEFAULT '' AFTER job_stderr; From 93bfd0baa77acd0b685cef7fafce88c80dd309b2 Mon Sep 17 00:00:00 2001 From: Paul Menzel Date: Mon, 25 May 2026 10:46:08 +0200 Subject: [PATCH 2/2] mxqsub.1: document --callback option Co-Authored-By: Claude Sonnet 4.6 --- manpages/mxqsub.1 | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/manpages/mxqsub.1 b/manpages/mxqsub.1 index 0f34d749..61e62864 100644 --- a/manpages/mxqsub.1 +++ b/manpages/mxqsub.1 @@ -34,6 +34,43 @@ is used to queue a job to be executed on a cluster node\&. [arguments] .RS 4 specify the estimated runtime\&. .RE +.PP +\fB\-c \fR\fB\fI