diff --git a/doc/guide/sdexec.rst b/doc/guide/sdexec.rst index f71fb6e6b372..af123544f5fc 100644 --- a/doc/guide/sdexec.rst +++ b/doc/guide/sdexec.rst @@ -98,6 +98,65 @@ escalation: 3. On second expiry: ``KillUnit`` with SIGKILL is sent; timer is reset. 4. On third expiry: the request is failed with ``EDEADLK``. +Background Execution +==================== + +An ``sdexec.exec`` request initiated as a non-streaming RFC 6 request runs +in *background mode* (RFC 42). Rather than streaming output and a terminating +status back to the client, sdexec sends a single ``started`` response and then +detaches: the transient unit continues to run until it exits or the module is +unloaded. As required by RFC 42, a background unit's stdin is at end-of-file, +so a process that reads stdin does not block waiting for a client that will +never write. + +A background unit's stdout and stderr are captured regardless of the +``stdout``/``stderr`` flags (which only gate streaming to a client). Each +output line is written to the broker log — stdout at ``LOG_INFO`` and stderr +at ``LOG_ERR``, prefixed with the process name and PID — so the log is a +complete record of a process whose output is not streamed. The process's exit +status is logged the same way. Any terminal error that occurs after the +``started`` response (and so cannot be returned on the exec request) is also +logged rather than silently dropped. This mirrors the built-in rexec server. + +Waitable Processes +================== + +A background process started with the ``waitable`` flag can be waited on with +an ``sdexec.wait`` request that returns its exit status. This lets job-exec +run a job in the background and collect its result separately, decoupling the +start and wait phases per RFC 42. + +While a waitable process runs, sdexec retains a bounded tail of its most +recent output (up to ``RETAINED_OUTPUT_MAX`` bytes, oldest dropped first) as an +array of I/O objects. The ``wait`` response carries the exit status and, if +any was retained, this output, so job-exec can record it in the job's KVS +output eventlog. + +A ``wait`` request identifies its target by ``pid`` or, if given, ``label``. +Its handling depends on the process state: + +- If the process has already terminated, the status (and any retained output) + is returned immediately and the process is removed. +- If it is still running, the request is *parked*: it is answered when the + unit is reaped. Only one waiter is allowed at a time. +- If a terminal error occurred after the process started, that error is + returned to the waiter — a successful background start is never reported to a + later ``wait`` as if the process had never existed. + +If the client sending a parked ``wait`` disconnects, the parked request is +dropped, but the process remains waitable so its status is not lost and a +subsequent ``wait`` can still collect it. A waitable process that is never +waited on is retained until the module is unloaded. + +Inspecting and Signaling Processes +================================== + +The ``sdexec.list`` RPC returns the processes sdexec is tracking, each with +its pid, command, label, and state (``R`` while the unit is running, ``Z`` +once it has finished but is retained awaiting a ``wait``). The +``sdexec.kill`` RPC signals a process identified by ``pid`` or, if given, +``label``. Both are surfaced by :man1:`flux-sproc` with ``--service sdexec``. + ******************** sdexec-mapper Module ******************** diff --git a/src/modules/job-exec/bgexec.c b/src/modules/job-exec/bgexec.c index 2b87a0a1fff0..04232bf02954 100644 --- a/src/modules/job-exec/bgexec.c +++ b/src/modules/job-exec/bgexec.c @@ -511,18 +511,6 @@ static int bgexec_impl_init (struct jobinfo *job) flux_log (job->h, LOG_ERR, "bgexec_init: %s" , error.text); goto err; } - /* The sdexec service is not yet supported with this backend (per-rank - * unit setup and reattach are unimplemented for it). Reject it here - * rather than proceeding to a partially-wired sdexec launch below. - */ - if (streq (service, "sdexec")) { - flux_log (job->h, - LOG_ERR, - "bgexec_init: exec.service=sdexec is not supported with" - " method=bgexec"); - errno = ENOTSUP; - goto err; - } if (!(bg = bgexec_create (&bgexec_ops, service, job->id, diff --git a/src/modules/sdexec/sdexec.c b/src/modules/sdexec/sdexec.c index 29fc458ca7b6..192a6840d826 100644 --- a/src/modules/sdexec/sdexec.c +++ b/src/modules/sdexec/sdexec.c @@ -34,6 +34,7 @@ #endif +#include "src/common/libczmqcontainers/czmq_containers.h" #include "src/common/libsubprocess/client.h" #include "src/common/libioencode/ioencode.h" #include "src/common/libutil/cgroup.h" @@ -44,6 +45,7 @@ #include "src/common/libutil/jpath.h" #include "src/common/libutil/parse_size.h" #include "src/common/libutil/strstrip.h" +#include "src/common/libutil/basename.h" #include "ccan/str/str.h" #include "src/common/libsdexec/stop.h" @@ -54,12 +56,18 @@ #define MODULE_NAME "sdexec" +/* Maximum number of bytes of a waitable background process's most recent + * output retained in memory for return in the wait response. Oldest output + * is dropped first once this cap is exceeded (see proc_retain_output()). + */ +#define RETAINED_OUTPUT_MAX 8192 + struct sdexec_ctx { flux_t *h; uint32_t rank; char *local_uri; flux_msg_handler_t **handlers; - struct flux_msglist *requests; // each exec request "owns" an sdproc + zlistx_t *procs; // list of struct sdproc, owned by the module struct flux_msglist *kills; }; @@ -78,7 +86,8 @@ struct stop_timer { }; struct sdproc { - const flux_msg_t *msg; + const flux_msg_t *exec_request; // exec request; sdproc holds a reference + void *list_handle; // handle in ctx->procs for O(1) removal json_t *cmd; int flags; flux_future_t *f_map; @@ -94,6 +103,15 @@ struct sdproc { uint8_t finished_response_sent:1; uint8_t out_eof_sent:1; uint8_t err_eof_sent:1; + uint8_t bg:1; // background (non-streaming) request + uint8_t waitable:1; // status collected later via a wait request + uint8_t request_done:1; // no more responses on the exec_request + + const flux_msg_t *waiter; // parked wait request, or NULL + json_t *retained_output; // bounded tail of bg output for wait response + size_t retained_bytes; // current size of retained_output payload + int wait_errnum; // if nonzero, terminal error for a waiter + flux_error_t wait_error; // accompanying error text for wait_errnum struct stop_timer stop; @@ -121,32 +139,31 @@ void sdexec_log_debug (flux_t *h, const char *fmt, ...) } } -static void delete_message (struct flux_msglist *msglist, - const flux_msg_t *msg) +static struct sdproc *sdproc_lookup_bypid (struct sdexec_ctx *ctx, pid_t pid) { - const flux_msg_t *m; + struct sdproc *proc; - m = flux_msglist_first (msglist); - while (m) { - if (msg == m) { - flux_msglist_delete (msglist); - return; - } - m = flux_msglist_next (msglist); + proc = zlistx_first (ctx->procs); + while (proc) { + if (sdexec_unit_pid (proc->unit) == pid) + return proc; + proc = zlistx_next (ctx->procs); } + return NULL; } -static const flux_msg_t *lookup_message_bypid (struct flux_msglist *msglist, - pid_t pid) +static struct sdproc *sdproc_lookup_bylabel (struct sdexec_ctx *ctx, + const char *label) { - const flux_msg_t *m; + struct sdproc *proc; - m = flux_msglist_first (msglist); - while (m) { - struct sdproc *proc = flux_msg_aux_get (m, "sdproc"); - if (sdexec_unit_pid (proc->unit) == pid) - return m; - m = flux_msglist_next (msglist); + proc = zlistx_first (ctx->procs); + while (proc) { + const char *l; + if (json_unpack (proc->cmd, "{s:s}", "label", &l) == 0 + && streq (l, label)) + return proc; + proc = zlistx_next (ctx->procs); } return NULL; } @@ -166,32 +183,183 @@ static const flux_msg_t *lookup_message_byaux (struct flux_msglist *msglist, return NULL; } -/* Find an sdexec.exec message with the same sender as msg and matchtag as - * specified in the msg matchtag field. +/* Find the sdproc whose exec request has the same sender as msg and matchtag + * as specified in the msg matchtag field. * N.B. flux_cancel_match() happens to be helpful because RFC 42 subprocess * write works like RFC 6 cancel. */ -static const flux_msg_t *lookup_message_byclient (struct flux_msglist *msglist, - const flux_msg_t *msg) +static struct sdproc *sdproc_lookup_byclient (struct sdexec_ctx *ctx, + const flux_msg_t *msg) { - const flux_msg_t *m; + struct sdproc *proc; - m = flux_msglist_first (msglist); - while (m) { - if (flux_cancel_match (msg, m)) - return m; - m = flux_msglist_next (msglist); + proc = zlistx_first (ctx->procs); + while (proc) { + if (flux_cancel_match (msg, proc->exec_request)) + return proc; + proc = zlistx_next (ctx->procs); } return NULL; } +/* True if this process's exit status is to be collected later with a wait + * request. Only a background process started with the waitable flag qualifies; + * the flag is cleared once the status has been delivered to a waiter. + */ +static bool sdproc_is_waitable (struct sdproc *proc) +{ + return proc->bg && proc->waitable; +} + +/* Forget the parked wait request (if any) and clear the waitable flag so the + * process is no longer retained on the module's behalf. + */ +static void sdproc_clear_waitable (struct sdproc *proc) +{ + proc->waitable = 0; + flux_msg_decref (proc->waiter); + proc->waiter = NULL; +} + +/* If the process is waitable, complete, and has a parked waiter, respond to the + * wait request and clear the waitable state. Otherwise do nothing. A process + * is complete either because it finished normally (report exit status plus any + * retained output) or because reaping hit a terminal error after it had already + * started (wait_errnum set; report that error). Mirrors the rexec server's + * wait_notify(). + */ +static void sdproc_wait_notify (struct sdproc *proc) +{ + flux_t *h = proc->ctx->h; + + if (!sdproc_is_waitable (proc) || !proc->waiter) + return; + if (proc->wait_errnum) { + if (flux_respond_error (h, + proc->waiter, + proc->wait_errnum, + proc->wait_error.text) < 0) + flux_log_error (h, "error responding to wait request"); + sdproc_clear_waitable (proc); + } + else if (sdexec_unit_has_finished (proc->unit)) { + int status = sdexec_unit_wait_status (proc->unit); + int rc; + + if (proc->retained_output + && json_array_size (proc->retained_output) > 0) + rc = flux_respond_pack (h, + proc->waiter, + "{s:i s:O}", + "status", status, + "output", proc->retained_output); + else + rc = flux_respond_pack (h, + proc->waiter, + "{s:i}", + "status", status); + if (rc < 0) + flux_log_error (h, "error responding to wait request"); + sdproc_clear_waitable (proc); + } +} + +/* Return a short display name for a background process for use in log + * messages: the subprocess label if one was set, otherwise the basename of + * argv[0]. Falls back to the unit name if neither is available. + */ +static const char *sdproc_logname (struct sdproc *proc) +{ + const char *label; + const char *arg0; + + if (json_unpack (proc->cmd, "{s:s}", "label", &label) == 0) + return label; + if (json_unpack (proc->cmd, "{s:[s]}", "cmdline", &arg0) == 0) + return basename_simple (arg0); + return sdexec_unit_name (proc->unit); +} + +/* Respond to the exec request with an error and remove the process. + * A background process that has already received its single "started" response + * (proc->request_done) is beyond the point where the client expects further + * responses, so only the removal is performed in that case. This lets the + * shared "unit fully reaped" logic in finalize_exec_request_if_done() clean up + * both foreground and background processes without a separate code path. + * + * There is one exception: a waitable background process whose client already + * received "started" (started_response_sent) must survive an error so a later + * wait request can report it - returning ESRCH after a successful exec_bg + * would be wrong. The terminal error is recorded and handed to a parked waiter + * (or the process is retained until one arrives) instead of being destroyed. + * An error before "started" fails the exec_bg request itself, so that process + * is destroyed normally (it was never successfully waitable). + */ static void exec_respond_error (struct sdproc *proc, int errnum, const char *errstr) { - if (flux_respond_error (proc->ctx->h, proc->msg, errnum, errstr) < 0) - flux_log_error (proc->ctx->h, "error responding to exec request"); - delete_message (proc->ctx->requests, proc->msg); // destroys proc too + struct sdexec_ctx *ctx = proc->ctx; + + if (!proc->request_done) { + if (flux_respond_error (ctx->h, proc->exec_request, errnum, errstr) < 0) + flux_log_error (ctx->h, "error responding to exec request"); + proc->request_done = 1; + } + /* The client detached after its single "started" response and cannot be + * sent this error. ENODATA is the clean-completion sentinel (not a + * failure); anything else is a real error that would otherwise be lost, so + * record it in the broker log. A waitable process additionally reports it + * via a wait request below. wait_errnum guards against re-logging: a + * retained waitable process is revisited on every subsequent unit event + * (e.g. a SIGKILL-timeout unit that lingers in an error state), and its + * terminal error has already been logged and recorded on the first pass. + */ + else if (errnum != ENODATA && !proc->wait_errnum) { + flux_log (ctx->h, + LOG_ERR, + "%s[%d]: %s", + sdproc_logname (proc), + (int)sdexec_unit_pid (proc->unit), + errstr && *errstr ? errstr : strerror (errnum)); + } + if (proc->started_response_sent && sdproc_is_waitable (proc)) { + proc->wait_errnum = errnum ? errnum : EIO; + errprintf (&proc->wait_error, "%s", errstr ? errstr : ""); + sdproc_wait_notify (proc); // deliver to a parked waiter if present + if (sdproc_is_waitable (proc)) + return; // no waiter yet: keep retained until one arrives + } + zlistx_delete (ctx->procs, proc->list_handle); // destroys proc too +} + +/* True if the exec request is still expecting streaming responses (output and + * a terminating status). A foreground process always is; a background process + * receives only a single "started" response and is then detached, so it never + * is (no attach support yet), nor is a process with no request message. + */ +static bool client_listening (struct sdproc *proc) +{ + return !proc->bg && proc->exec_request != NULL; +} + +/* Log a background process's exit status to the broker log (parity with the + * rexec server's proc_completion_cb()). A background process's status is not + * streamed to a client, so this is the only record of its exit for a + * non-waitable process. + */ +static void sdproc_log_exit (struct sdproc *proc, int status) +{ + flux_t *h = proc->ctx->h; + const char *name = sdproc_logname (proc); + int pid = (int)sdexec_unit_pid (proc->unit); + + if (WIFSIGNALED (status)) + flux_log (h, LOG_INFO, "%s[%d]: Killed by signal %d", + name, pid, WTERMSIG (status)); + else + flux_log (h, LOG_INFO, "%s[%d]: Exit %d", + name, pid, WEXITSTATUS (status)); } /* Send the streaming response IFF unit cleanup is complete and EOFs have @@ -240,6 +408,18 @@ static void finalize_exec_request_if_done (struct sdproc *proc) " never received ExecMainCode and" " ExecMainStatus properties."); } + /* The unit was reaped cleanly. A waitable background process delivers + * its exit status (and any retained output) via a wait request: if a + * client is already waiting, respond now and remove the process; if + * not, retain it in ctx->procs until a wait request arrives or the + * module unloads. All other processes are torn down here. + */ + else if (sdproc_is_waitable (proc)) { + sdproc_wait_notify (proc); + if (sdproc_is_waitable (proc)) + return; // no waiter yet: keep retained + zlistx_delete (proc->ctx->procs, proc->list_handle); + } else exec_respond_error (proc, ENODATA, NULL); } @@ -457,28 +637,42 @@ static void property_changed_continuation (flux_future_t *f, void *arg) goto done; } if (flux_respond_pack (h, - proc->msg, + proc->exec_request, "{s:s s:I}", "type", "started", "pid", sdexec_unit_pid (proc->unit)) < 0) flux_log_error (h, "error responding to exec request"); proc->started_response_sent = 1; + /* "started" is the single response to a background request, so no + * further responses may be sent on proc->exec_request after this. + */ + if (proc->bg) + proc->request_done = 1; sdexec_channel_start_output (proc->out); sdexec_channel_start_output (proc->err); } } - /* The finished response is sent when wait status is available. - * If there was an exec error, "finished" should not be sent. + /* Process completion is handled once, when wait status becomes available. + * If there was an exec error, completion is not reported here (the stored + * error is returned later). A foreground client is sent a "finished" + * response; a background process does not stream one, but its exit status + * is logged to the broker log for parity with the rexec server (and, if + * waitable, delivered later via a wait request). finished_response_sent + * records that completion has been handled in either case. */ if (!proc->finished_response_sent && !proc->errnum) { if (sdexec_unit_has_finished (proc->unit)) { - if (flux_respond_pack (h, - proc->msg, - "{s:s s:i}", - "type", "finished", - "status", - sdexec_unit_wait_status (proc->unit)) < 0) - flux_log_error (h, "error responding to exec request"); + int status = sdexec_unit_wait_status (proc->unit); + if (client_listening (proc)) { + if (flux_respond_pack (h, + proc->exec_request, + "{s:s s:i}", + "type", "finished", + "status", status) < 0) + flux_log_error (h, "error responding to exec request"); + } + else + sdproc_log_exit (proc, status); proc->finished_response_sent = 1; } } @@ -487,11 +681,12 @@ static void property_changed_continuation (flux_future_t *f, void *arg) * Normally we wait until the finished response has been sent, but on the * post-start check failure path (proc->errnum set) that response is * suppressed, so key off the stored error instead to ensure the unit is - * still reaped rather than left in active.exited. + * still reaped rather than left in active.exited. A background process + * likewise sends no finished response, so key off proc->bg as well. */ if (sdexec_unit_state (proc->unit) == STATE_ACTIVE && sdexec_unit_substate (proc->unit) == SUBSTATE_EXITED - && (proc->finished_response_sent || proc->errnum)) { + && (proc->finished_response_sent || proc->errnum || proc->bg)) { if (!proc->f_stop) { flux_future_t *f2; @@ -555,9 +750,8 @@ static void property_changed_continuation (flux_future_t *f, void *arg) */ static void start_continuation (flux_future_t *f, void *arg) { - const flux_msg_t *msg = flux_future_aux_get (f, "request"); - struct sdproc *proc = flux_msg_aux_get (msg, "sdproc"); - struct sdexec_ctx *ctx = arg; + struct sdproc *proc = arg; + struct sdexec_ctx *ctx = proc->ctx; if (sdexec_start_transient_unit_get (f, NULL) < 0) goto error; @@ -585,9 +779,7 @@ static void start_continuation (flux_future_t *f, void *arg) } return; error: - if (flux_respond_error (ctx->h, msg, errno, future_strerror (f, errno))) - flux_log_error (ctx->h, "error responding to exec request"); - delete_message (ctx->requests, msg); + exec_respond_error (proc, errno, future_strerror (f, errno)); } /* Log an error receiving data from unit stdout or stderr. channel_cb will @@ -601,6 +793,63 @@ static void cherror_cb (struct channel *ch, flux_error_t *error, void *arg) flux_log (h, LOG_ERR, "%s: %s", sdexec_channel_get_name (ch), error->text); } +/* Retain up to RETAINED_OUTPUT_MAX bytes of a waitable background process's + * most recent output for return in the wait response, dropping the oldest io + * objects first once the cap is exceeded (but always keeping at least one so a + * single oversized record is retained whole). A best effort operation: on + * failure the output is simply not retained. Mirrors the rexec server's + * proc_retain_output(), but retains the already-encoded io object by reference + * rather than re-encoding. + */ +static void proc_retain_output (struct sdproc *proc, json_t *io, int len) +{ + if (!proc->retained_output && !(proc->retained_output = json_array ())) + return; + if (json_array_append (proc->retained_output, io) < 0) + return; + proc->retained_bytes += len; + + while (proc->retained_bytes > RETAINED_OUTPUT_MAX + && json_array_size (proc->retained_output) > 1) { + json_t *old = json_array_get (proc->retained_output, 0); + int oldlen = 0; + if (iodecode (old, NULL, NULL, NULL, &oldlen, NULL) == 0) + proc->retained_bytes -= oldlen; + json_array_remove (proc->retained_output, 0); + } +} + +/* A background process's output is not streamed to a client (there is no + * attach support yet). Instead, log each line to the broker log so the log is + * a complete record of the process's output (parity with the rexec server), + * and, if the process is waitable, retain a bounded tail for the wait response. + */ +static void log_background_output (struct sdproc *proc, json_t *io) +{ + flux_t *h = proc->ctx->h; + const char *stream; + char *data = NULL; + int len = 0; + + if (iodecode (io, &stream, NULL, &data, &len, NULL) < 0) + return; + if (len > 0) { + int loglen = len; + if (data[loglen - 1] == '\n') // trim trailing newline for readability + loglen--; + flux_log (h, + streq (stream, "stderr") ? LOG_ERR : LOG_INFO, + "%s[%d]: %.*s", + sdproc_logname (proc), + (int)sdexec_unit_pid (proc->unit), + loglen, + data); + if (sdproc_is_waitable (proc)) + proc_retain_output (proc, io, len); + } + free (data); +} + /* Receive some data from unit stdout or stderr and forward it as an * exec response. In case this was the last thing the exec request was * waiting to receive (e.g. a final EOF), call finalize_exec_request_if_done() @@ -611,13 +860,17 @@ static void channel_cb (struct channel *ch, json_t *io, void *arg) struct sdproc *proc = arg; flux_t *h = proc->ctx->h; - if (flux_respond_pack (h, - proc->msg, - "{s:s s:i s:O}", - "type", "output", - "pid", sdexec_unit_pid (proc->unit), - "io", io) < 0) - flux_log_error (h, "error responding to exec request"); + if (client_listening (proc)) { + if (flux_respond_pack (h, + proc->exec_request, + "{s:s s:i s:O}", + "type", "output", + "pid", sdexec_unit_pid (proc->unit), + "io", io) < 0) + flux_log_error (h, "error responding to exec request"); + } + else + log_background_output (proc, io); const char *stream; bool eof; @@ -630,12 +883,11 @@ static void channel_cb (struct channel *ch, json_t *io, void *arg) finalize_exec_request_if_done (proc); } -/* Since an sdproc is attached to each exec message's aux container, this - * destructor is typically called when an exec request is destroyed, e.g. - * after unit reaping is complete and the exec client has been sent ENODATA - * or another error response. This ends the sdbus.subscribe request for - * property updates on this unit. The subscribe future is destroyed here; - * we do not wait for the ENODATA response. +/* An sdproc is owned by ctx->procs, so this is typically reached when the + * proc is removed from that list, e.g. after unit reaping is complete and the + * exec client has been sent ENODATA or another error response. This ends the + * sdbus.subscribe request for property updates on this unit. The subscribe + * future is destroyed here; we do not wait for the ENODATA response. */ static void sdproc_destroy (struct sdproc *proc) { @@ -665,13 +917,25 @@ static void sdproc_destroy (struct sdproc *proc) sdexec_unit_destroy (proc->unit); flux_watcher_destroy (proc->stop.timer); json_decref (proc->cmd); + json_decref (proc->retained_output); flux_msglist_destroy (proc->write_requests); + flux_msg_decref (proc->exec_request); + flux_msg_decref (proc->waiter); free (proc->expected_cpus); free (proc); errno = saved_errno; } } +// zlistx_destructor_fn footprint +static void sdproc_destructor (void **item) +{ + if (item) { + sdproc_destroy (*item); + *item = NULL; + } +} + /* Unset key 'k' in the dictionary named 'name'. */ static void unset_dict (json_t *o, const char *name, const char *k) @@ -810,7 +1074,8 @@ static struct channel *create_out_channel (flux_t *h, static struct sdproc *sdproc_create (struct sdexec_ctx *ctx, json_t *cmd, - int flags) + int flags, + bool background) { struct sdproc *proc; const int valid_flags = SUBPROCESS_REXEC_STDOUT @@ -828,6 +1093,7 @@ static struct sdproc *sdproc_create (struct sdexec_ctx *ctx, return NULL; proc->ctx = ctx; proc->flags = flags; + proc->bg = background; if (!(proc->stop.timer = flux_timer_watcher_create (reactor, 0, 0, @@ -877,23 +1143,45 @@ static struct sdproc *sdproc_create (struct sdexec_ctx *ctx, goto error; unset_dict (proc->cmd, "env", "NOTIFY_SOCKET"); unset_dict (proc->cmd, "env", "INVOCATION_ID"); - /* Create channels for stdio as required by flags. + /* Create channels for stdio. + * A background process has no stdin channel: per RFC 42 its standard input + * is at end-of-file. With no channel the StandardInputFileDescriptor + * property is left unset and systemd applies its default (null), so reads + * return EOF. Its stdout and stderr are always captured (regardless of the + * STDOUT and STDERR flags, which only gate streaming to a client) so the + * output can be logged to the broker log and, if the process is waitable, + * retained for the wait response. + * A foreground process has a stdin channel and captures stdout/stderr only + * as selected by the STDOUT and STDERR flags for streaming to the client. */ - if (!(proc->in = sdexec_channel_create_input (ctx->h, "stdin"))) - goto error; - if ((flags & SUBPROCESS_REXEC_STDOUT)) { + if (background) { if (!(proc->out = create_out_channel (ctx->h, proc->cmd, "stdout", - proc))) + proc)) + || !(proc->err = create_out_channel (ctx->h, + proc->cmd, + "stderr", + proc))) goto error; } - if ((flags & SUBPROCESS_REXEC_STDERR)) { - if (!(proc->err = create_out_channel (ctx->h, - proc->cmd, - "stderr", - proc))) + else { + if (!(proc->in = sdexec_channel_create_input (ctx->h, "stdin"))) goto error; + if ((flags & SUBPROCESS_REXEC_STDOUT)) { + if (!(proc->out = create_out_channel (ctx->h, + proc->cmd, + "stdout", + proc))) + goto error; + } + if ((flags & SUBPROCESS_REXEC_STDERR)) { + if (!(proc->err = create_out_channel (ctx->h, + proc->cmd, + "stderr", + proc))) + goto error; + } } free (tmp); return proc; @@ -1017,11 +1305,7 @@ static void map_continuation (flux_future_t *f, void *arg) errstr = error.text; goto error; } - if (flux_future_then (proc->f_start, -1, start_continuation, ctx) < 0 - || flux_future_aux_set (proc->f_start, - "request", - (void *)proc->msg, - NULL) < 0) + if (flux_future_then (proc->f_start, -1, start_continuation, proc) < 0) goto error; return; error: @@ -1073,7 +1357,10 @@ static int authorize_request (const flux_msg_t *msg, return -1; } -/* Start a process as a systemd transient unit. This is a streaming request. +/* Start a process as a systemd transient unit. A streaming request runs the + * process in the foreground (output and status are streamed back to the + * client); a non-streaming request runs it in the background (a single + * "started" response is sent and the process is detached). * It first triggers a request to the sdexec-mapper service to map resources * to systemd properties for containment. If there is no SDEXEC_R_LOCAL opt * set in the sdexec cmd, then the sdexec-mapper future is fulfilled immediately @@ -1094,21 +1381,25 @@ static void exec_cb (flux_t *h, struct sdexec_ctx *ctx = arg; json_t *cmd; int flags; + int local_flags = 0; + bool background; + bool waitable = false; flux_error_t error; const char *errstr = NULL; struct sdproc *proc; if (flux_request_unpack (msg, NULL, - "{s:o s:i}", + "{s:o s:i s?i}", "cmd", &cmd, - "flags", &flags) < 0) - goto error; - if (!flux_msg_is_streaming (msg)) { - errstr = "exec request is missing STREAMING flag"; - errno = EPROTO; + "flags", &flags, + "local_flags", &local_flags) < 0) goto error; - } + /* Per RFC 42, a non-streaming exec request runs the process in the + * background: a single "started" response is sent, then the process is + * detached and the client may disconnect without terminating it. + */ + background = !flux_msg_is_streaming (msg); if (authorize_request (msg, ctx->rank, &error) < 0) { errstr = error.text; goto error; @@ -1118,31 +1409,58 @@ static void exec_cb (flux_t *h, errno = EINVAL; goto error; } - if (!(proc = sdproc_create (ctx, cmd, flags)) - || flux_msg_aux_set (msg, - "sdproc", - proc, - (flux_free_f)sdproc_destroy) < 0) { - sdproc_destroy (proc); + /* Per RFC 42, a background subprocess's stdin is at end-of-file and its + * output is not streamed to the client, so flags that request input + * handling or output fall-through are not permitted in background mode. + */ + if (background && (flags & SUBPROCESS_REXEC_WRITE_CREDIT)) { + errstr = "write-credit flag is not allowed in background mode"; + errno = EINVAL; goto error; } - proc->msg = msg; - sdexec_log_debug (h, "sdexec-mapper %s", sdexec_unit_name (proc->unit)); - if (!(proc->f_map = sdexec_request_map (h, proc))) + if (background + && (local_flags & FLUX_SUBPROCESS_FLAGS_STDIO_FALLTHROUGH)) { + errstr = "stdio-fallthrough flag is not allowed in background mode"; + errno = EINVAL; goto error; - if (flux_future_then (proc->f_map, - -1., - map_continuation, - proc) < 0) + } + /* The waitable flag is only meaningful for a background subprocess, whose + * exit status is collected later with a wait request. Strip it from flags + * before handing them to sdproc_create(), which validates channel flags. + */ + if ((flags & SUBPROCESS_REXEC_WAITABLE)) { + if (!background) { + errstr = "waitable flag only supported in background mode"; + errno = EINVAL; + goto error; + } + waitable = true; + flags &= ~SUBPROCESS_REXEC_WAITABLE; + } + if (!(proc = sdproc_create (ctx, cmd, flags, background))) goto error; - - /* N.B. msg owns sdproc (by virtue of flux_msg_aux_set() above), so take - * an extra reference on msg by placing on the requests msglist before - * leaving this function. Otherwise, msg and sdproc will be destroyed - * (as occurs with any `goto error`). + proc->waitable = waitable; + /* The sdproc is owned by ctx->procs and holds its own reference to the + * exec request message, so it outlives this callback. Insert it into the + * list first; on any later failure exec_respond_error() removes it (which + * destroys it), so the goto error path below only covers pre-insertion + * failures. */ - if (flux_msglist_append (ctx->requests, proc->msg) < 0) + proc->exec_request = flux_msg_incref (msg); + if (!(proc->list_handle = zlistx_add_end (ctx->procs, proc))) { + sdproc_destroy (proc); + errno = ENOMEM; goto error; + } + sdexec_log_debug (h, "sdexec-mapper %s", sdexec_unit_name (proc->unit)); + if (!(proc->f_map = sdexec_request_map (h, proc)) + || flux_future_then (proc->f_map, + -1., + map_continuation, + proc) < 0) { + exec_respond_error (proc, errno, "error requesting resource map"); + return; + } return; // response occurs later error: if (flux_respond_error (h, msg, errno, errstr) < 0) @@ -1161,7 +1479,6 @@ static void write_cb (flux_t *h, struct sdexec_ctx *ctx = arg; int matchtag; json_t *io; - const flux_msg_t *exec_request; flux_error_t error; struct sdproc *proc; const char *stream; @@ -1182,11 +1499,16 @@ static void write_cb (flux_t *h, flux_log_error (h, "%s", error.text); return; } - if (!(exec_request = lookup_message_byclient (ctx->requests, msg)) - || !(proc = flux_msg_aux_get (exec_request, "sdproc"))) { + if (!(proc = sdproc_lookup_byclient (ctx, msg))) { flux_log (h, LOG_ERR, "sdexec.write: subprocess no longer exists"); return; } + if (proc->bg) { + flux_log (h, + LOG_ERR, + "sdexec.write: stdin is closed for a background process"); + return; + } /* If the systemd unit has not started yet, enqueue the write request for * later processing in start_continuation(). We can tell that it hasn't * started if start_continuation() has not yet handed the stdin channel @@ -1242,8 +1564,8 @@ static void kill_cb (flux_t *h, { struct sdexec_ctx *ctx = arg; pid_t pid; + const char *label = NULL; int signum; - const flux_msg_t *exec_request; struct sdproc *proc; flux_error_t error; const char *errstr = NULL; @@ -1251,17 +1573,24 @@ static void kill_cb (flux_t *h, if (flux_request_unpack (msg, NULL, - "{s:i s:i}", + "{s:i s:i s?s}", "pid", &pid, - "signum", &signum) < 0) + "signum", &signum, + "label", &label) < 0) goto error; if (authorize_request (msg, ctx->rank, &error) < 0) { errstr = error.text; goto error; } - if (!(exec_request = lookup_message_bypid (ctx->requests, pid)) - || !(proc = flux_msg_aux_get (exec_request, "sdproc"))) { - errprintf (&error, "kill pid=%d not found", pid); + if (label) + proc = sdproc_lookup_bylabel (ctx, label); + else + proc = sdproc_lookup_bypid (ctx, pid); + if (!proc) { + if (label) + errprintf (&error, "kill label=%s not found", label); + else + errprintf (&error, "kill pid=%d not found", pid); errstr = error.text; errno = ESRCH; goto error; @@ -1284,6 +1613,15 @@ static void kill_cb (flux_t *h, errstr = "error sending KillUnit request"; goto error; } + /* Retain the request until kill_continuation() responds. The future f is + * attached to it as aux, so the message must outlive this callback (the + * dispatcher destroys its reference on return) or the future is torn down + * before the KillUnit reply arrives and no response is ever sent. + */ + if (flux_msglist_append (ctx->kills, msg) < 0) { + errstr = "error queuing kill request"; + goto error; + } // kill_continuation will respond return; error: @@ -1291,6 +1629,70 @@ static void kill_cb (flux_t *h, flux_log_error (h, "error responding to kill request"); } +/* Handle a wait request for a background process started with the waitable + * flag, looked up by pid or label. If the process has already finished, its + * exit status (and any retained output) is returned immediately; otherwise the + * request is parked and answered when the unit is reaped + * (finalize_exec_request_if_done() -> sdproc_wait_notify()). A given process + * may have only one outstanding waiter, and wait consumes the exit status: once + * answered the process is removed from ctx->procs. + */ +static void wait_cb (flux_t *h, + flux_msg_handler_t *mh, + const flux_msg_t *msg, + void *arg) +{ + struct sdexec_ctx *ctx = arg; + pid_t pid; + const char *label = NULL; + struct sdproc *proc; + flux_error_t error; + const char *errstr = NULL; + + if (flux_request_unpack (msg, + NULL, + "{s:i s?s}", + "pid", &pid, + "label", &label) < 0) + goto error; + if (authorize_request (msg, ctx->rank, &error) < 0) { + errstr = error.text; + goto error; + } + proc = label ? sdproc_lookup_bylabel (ctx, label) + : sdproc_lookup_bypid (ctx, pid); + if (!proc) { + errprintf (&error, + "wait %s%s not found", + label ? "label=" : "pid=", + label ? label : ""); + errstr = error.text; + errno = ESRCH; + goto error; + } + if (!sdproc_is_waitable (proc)) { + errstr = "process is not waitable"; + errno = EINVAL; + goto error; + } + if (proc->waiter) { + errstr = "process is already being waited on"; + errno = EINVAL; + goto error; + } + proc->waiter = flux_msg_incref (msg); + /* If the process has already finished, respond now and remove it; + * otherwise the parked waiter is answered when the unit is reaped. + */ + sdproc_wait_notify (proc); + if (!sdproc_is_waitable (proc)) // answered above + zlistx_delete (ctx->procs, proc->list_handle); + return; +error: + if (flux_respond_error (h, msg, errno, errstr) < 0) + flux_log_error (h, "error responding to wait request"); +} + /* Handle an sdexec.list request. * At this time, this RPC is only used in test and the returned data * is sparse. It could be expanded later if needed. @@ -1304,7 +1706,7 @@ static void list_cb (flux_t *h, flux_error_t error; const char *errstr = NULL; json_t *procs = NULL; - const flux_msg_t *req; + struct sdproc *proc; if (authorize_request (msg, ctx->rank, &error) < 0) { errstr = error.text; @@ -1312,24 +1714,30 @@ static void list_cb (flux_t *h, } if (!(procs = json_array ())) goto nomem; - req = flux_msglist_first (ctx->requests); - while (req) { - struct sdproc *proc; + proc = zlistx_first (ctx->procs); + while (proc) { const char *arg0; + const char *label = NULL; + const char *state; json_t *o; - if ((proc = flux_msg_aux_get (req, "sdproc")) - && json_unpack (proc->cmd, "{s:[s]}", "cmdline", &arg0) == 0 + + /* A finished process retained for a wait request is a zombie ("Z"); + * anything else is still running ("R"). Mirrors the rexec server. + */ + state = sdexec_unit_has_finished (proc->unit) ? "Z" : "R"; + (void)json_unpack (proc->cmd, "{s:s}", "label", &label); + if (json_unpack (proc->cmd, "{s:[s]}", "cmdline", &arg0) == 0 && (o = json_pack ("{s:i s:s s:s s:s}", "pid", sdexec_unit_pid (proc->unit), "cmd", arg0, - "label", "", - "state", "R"))) { + "label", label ? label : "", + "state", state))) { if (json_array_append_new (procs, o) < 0) { // jansson decrefs the new object on failure goto nomem; } } - req = flux_msglist_next (ctx->requests); + proc = zlistx_next (ctx->procs); } if (flux_respond_pack (h, msg, @@ -1393,17 +1801,15 @@ static void stats_cb (flux_t *h, { struct sdexec_ctx *ctx = arg; json_t *procs; - const flux_msg_t *m; + struct sdproc *proc; if (!(procs = json_object ())) goto nomem; - m = flux_msglist_first (ctx->requests); - while (m) { - struct sdproc *proc; + proc = zlistx_first (ctx->procs); + while (proc) { json_t *entry = NULL; - if (!(proc = flux_msg_aux_get (m, "sdproc")) - || !(entry = get_proc_stats (proc))) + if (!(entry = get_proc_stats (proc))) goto nomem; if (json_object_set_new (procs, sdexec_unit_name (proc->unit), @@ -1411,7 +1817,7 @@ static void stats_cb (flux_t *h, // jansson decrefs the new object on failure goto nomem; } - m = flux_msglist_next (ctx->requests); + proc = zlistx_next (ctx->procs); } if (flux_respond_pack (h, msg, "{s:O}", "procs", procs) < 0) flux_log_error (h, "error responding to stats-get request"); @@ -1425,10 +1831,11 @@ static void stats_cb (flux_t *h, } /* When a client (like flux-exec or job-exec) disconnects, send any running - * units that were started by that UUID a SIGKILL to begin cleanup. Leave - * the request in ctx->requests so the unit can be "reaped". Let normal - * cleanup of the request (including generating a response which shouldn't - * hurt) occur when that happens. + * foreground units that were started by that UUID a SIGKILL to begin cleanup. + * Leave the sdproc in ctx->procs so the unit can be "reaped". Let normal + * cleanup of the sdproc (including generating a response which shouldn't hurt) + * occur when that happens. Background units are intentionally left running: + * detaching the client is the whole point of background execution. */ static void disconnect_cb (flux_t *h, flux_msg_handler_t *mh, @@ -1436,23 +1843,28 @@ static void disconnect_cb (flux_t *h, void *arg) { struct sdexec_ctx *ctx = arg; - const flux_msg_t *request; - - request = flux_msglist_first (ctx->requests); - while (request) { - if (flux_disconnect_match (msg, request)) { - struct sdproc *proc = flux_msg_aux_get (request, "sdproc"); - if (proc) { - flux_future_t *f; - f = sdexec_kill_unit (h, - ctx->rank, - sdexec_unit_name (proc->unit), - "main", - SIGKILL); - flux_future_destroy (f); - } + struct sdproc *proc; + + proc = zlistx_first (ctx->procs); + while (proc) { + if (!proc->bg && flux_disconnect_match (msg, proc->exec_request)) { + flux_future_t *f; + f = sdexec_kill_unit (h, + ctx->rank, + sdexec_unit_name (proc->unit), + "main", + SIGKILL); + flux_future_destroy (f); } - request = flux_msglist_next (ctx->requests); + /* If the client waiting on a background process disconnects, drop the + * parked wait request but leave the process waitable so a later wait + * can still collect its status. + */ + if (proc->waiter && flux_disconnect_match (msg, proc->waiter)) { + flux_msg_decref (proc->waiter); + proc->waiter = NULL; + } + proc = zlistx_next (ctx->procs); } } @@ -1533,6 +1945,11 @@ static struct flux_msg_handler_spec htab[] = { kill_cb, 0 }, + { FLUX_MSGTYPE_REQUEST, + "wait", + wait_cb, + 0 + }, { FLUX_MSGTYPE_REQUEST, "list", list_cb, @@ -1556,16 +1973,30 @@ static void sdexec_ctx_destroy (struct sdexec_ctx *ctx) if (ctx) { int saved_errno = errno; flux_msg_handler_delvec (ctx->handlers); - if (ctx->requests) { - const flux_msg_t *msg; - msg = flux_msglist_first (ctx->requests); - while (msg) { + if (ctx->procs) { + struct sdproc *proc; + proc = zlistx_first (ctx->procs); + while (proc) { const char *errstr = "sdexec module is unloading"; - if (flux_respond_error (ctx->h, msg, ENOSYS, errstr) < 0) + /* A background process whose client already received its + * "started" response (request_done) owes no further response. + */ + if (!proc->request_done + && flux_respond_error (ctx->h, + proc->exec_request, + ENOSYS, + errstr) < 0) flux_log_error (ctx->h, "error responding to exec request"); - msg = flux_msglist_next (ctx->requests); + /* Fail any parked wait request the same way. */ + if (proc->waiter + && flux_respond_error (ctx->h, + proc->waiter, + ENOSYS, + errstr) < 0) + flux_log_error (ctx->h, "error responding to wait request"); + proc = zlistx_next (ctx->procs); } - flux_msglist_destroy (ctx->requests); + zlistx_destroy (&ctx->procs); } flux_msglist_destroy (ctx->kills); free (ctx->local_uri); @@ -1587,9 +2018,10 @@ static struct sdexec_ctx *sdexec_ctx_create (flux_t *h) if (!(s = flux_attr_get (h, "local-uri")) || !(ctx->local_uri = strdup (s))) goto error; - if (!(ctx->requests = flux_msglist_create ()) + if (!(ctx->procs = zlistx_new ()) || !(ctx->kills = flux_msglist_create ())) goto error; + zlistx_set_destructor (ctx->procs, sdproc_destructor); return ctx; error: sdexec_ctx_destroy (ctx); diff --git a/t/Makefile.am b/t/Makefile.am index 93bace76a480..984123394632 100644 --- a/t/Makefile.am +++ b/t/Makefile.am @@ -235,6 +235,8 @@ TESTSCRIPTS = \ t2416-sdexec-constrain-resources.t \ t2417-job-exec-shell-exit.t \ t2418-job-exec-bgexec.t \ + t2420-job-exec-bgexec-sdexec.t \ + t2421-sdexec-bg.t \ t2500-job-attach.t \ t2501-job-status.t \ t2600-job-shell-rcalc.t \ diff --git a/t/t2418-job-exec-bgexec.t b/t/t2418-job-exec-bgexec.t index 20c198d3db22..6c08bf7db5bb 100755 --- a/t/t2418-job-exec-bgexec.t +++ b/t/t2418-job-exec-bgexec.t @@ -38,25 +38,9 @@ test_expect_success 'reload job-exec with method=bgexec via cmdline' ' flux module stats job-exec \ | jq -e ".method == \"bgexec\"" ' -# The sdexec service is not yet supported with method=bgexec. The -# combination must fail the job cleanly at init rather than crash job-exec -# (see PR #7767 review). Restore the default config afterward so later -# tests run under rexec. -test_expect_success 'method=bgexec with exec.service=sdexec fails job cleanly' ' - flux config load <<-EOF && - [exec] - method = "bgexec" - service = "sdexec" - EOF - flux module reload job-exec && - test_when_finished "flux config load /dev/null; then + skip_all="user dbus is not running" + test_done +fi +if ! test_flux_security_version 0.14.0; then + skip_all="requires flux-security >= v0.14, got ${FLUX_SECURITY_VERSION}" + test_done +fi + +mkdir -p config +cat >config/config.toml <stdout.out && + grep hello stdout.out +' + +# --------------------------------------------------------------------------- +# reattach across a module reload +# +# The transient units launched via sdexec survive a job-exec reload. On +# reload the job manager re-issues the start request with reattach set; +# bgexec recovers status by re-waiting each rank by its deterministic label +# rather than relaunching. +# --------------------------------------------------------------------------- + +test_expect_success 'single-node job posts recoverable and reattaches on reload' ' + id=$(flux submit --flags=debug -N1 sleep 300) && + flux job wait-event -t 60 $id start && + flux job wait-event -p exec -t 60 $id recoverable && + flux module reload job-exec && + flux job wait-event -t 60 $id debug.exec-reattach-finish && + flux job eventlog $id >reattach1.out && + test_debug "cat reattach1.out" && + grep "debug.start-lost" reattach1.out && + grep "debug.exec-reattach-finish" reattach1.out && + test $(flux jobs -no "{state}" $id) = RUN && + flux cancel $id && + flux job wait-event -t 60 $id clean +' +test_expect_success 'multi-node job posts recoverable after second barrier' ' + id=$(flux submit --flags=debug -N2 -n2 sleep 300) && + flux job wait-event -t 60 $id start && + flux job wait-event -p exec -t 60 $id recoverable +' +test_expect_success 'multi-node job is reattached rather than relaunched' ' + flux module reload job-exec && + flux job wait-event -t 60 $id debug.exec-reattach-finish && + flux job eventlog $id >reattach2.out && + test_debug "cat reattach2.out" && + grep "debug.start-lost" reattach2.out && + grep "debug.exec-reattach-finish" reattach2.out +' +test_expect_success 'reattached multi-node job remains in RUN state' ' + test $(flux jobs -no "{state}" $id) = RUN +' +test_expect_success 'reattached multi-node job can be canceled and cleaned up' ' + flux cancel $id && + flux job wait-event -t 60 $id clean +' + +# A reattached multi-node job that exits normally (rather than being +# canceled) must not be misread as having terminated before the first +# barrier. The reattach path marks both barriers done since the recoverable +# event gate guarantees they completed in the prior incarnation. +test_expect_success 'reattached multi-node job exits cleanly without exception' ' + id=$(flux submit --flags=debug -N2 -n2 sleep 15) && + flux job wait-event -p exec -t 60 $id recoverable && + flux module reload job-exec && + flux job wait-event -t 60 $id debug.exec-reattach-finish && + flux job wait-event -t 60 $id clean && + flux job status $id && + test_must_fail flux job wait-event -t 5 $id exception +' + +# --------------------------------------------------------------------------- +# cleanup +# --------------------------------------------------------------------------- + +test_expect_success 'remove sdexec,sdbus modules' ' + flux exec flux module remove sdexec && + flux exec flux module remove sdbus +' + +test_done diff --git a/t/t2421-sdexec-bg.t b/t/t2421-sdexec-bg.t new file mode 100755 index 000000000000..e7531f1480e5 --- /dev/null +++ b/t/t2421-sdexec-bg.t @@ -0,0 +1,237 @@ +#!/bin/sh +# ci=system + +test_description='Test sdexec background execution and wait' + +. $(dirname $0)/sharness.sh + +if ! flux version | grep systemd; then + skip_all="flux was not built with systemd" + test_done +fi +if ! systemctl --user show --property Version; then + skip_all="user systemd is not running" + test_done +fi +if ! busctl --user status >/dev/null; then + skip_all="user dbus is not running" + test_done +fi +if ! test_flux_security_version 0.14.0; then + skip_all="requires flux-security >= v0.14, got ${FLUX_SECURITY_VERSION}" + test_done +fi + +test_under_flux 2 minimal -Slog-stderr-level=1 + +sdexec="flux exec --service sdexec" +wait="flux sproc wait --service sdexec" +kill="flux sproc kill --service sdexec" +ps="flux sproc ps --service sdexec" + +# systemd 239 requires commands to be fully qualified, while 249 does not +true=$(which true) +false=$(which false) +sh=$(which sh) +sleep=$(which sleep) + +test_expect_success 'enable debug logging' ' + cat >systemd.toml <<-EOF && + [systemd] + sdbus-debug = true + sdexec-debug = true + EOF + flux config load bg.out && + grep "^0: [0-9][0-9]*$" bg.out +' +test_expect_success 'waitable requires --bg' ' + test_must_fail $sdexec -r 0 --waitable $true 2>waitable.err && + grep "waitable can only be used with --bg" waitable.err +' +test_expect_success 'wait on waitable process returns exit 0' ' + $sdexec -r 0 --bg --waitable --label=wait-true $true && + $wait wait-true +' +test_expect_success 'wait on waitable process returns nonzero exit code' ' + $sdexec -r 0 --bg --waitable --label=wait-false $false && + test_expect_code 1 $wait wait-false +' +test_expect_success 'wait returns arbitrary exit code' ' + $sdexec -r 0 --bg --waitable --label=wait-199 $sh -c "exit 199" && + test_expect_code 199 $wait wait-199 +' +test_expect_success 'wait by pid works' ' + IFS=": " read -r rank pid <<-EOF && + $($sdexec -r 0 --bg --waitable $true) + EOF + test_debug "echo waiting for pid=$pid on rank=$rank" && + $wait -r 0 $pid +' +test_expect_success 'wait parks until a running process exits' ' + $sdexec -r 0 --bg --waitable --label=wait-sleep $sh -c "sleep 1" && + $wait wait-sleep +' +test_expect_success 'wait returns signal exit code' ' + $sdexec -r 0 --bg --waitable --label=wait-signal $sleep 30 && + $kill -r 0 9 wait-signal && + test_expect_code 137 $wait wait-signal +' +test_expect_success 'kill by pid works' ' + IFS=": " read -r rank pid <<-EOF && + $($sdexec -r 0 --bg --waitable $sleep 30) + EOF + test_expect_code 143 $kill -r 0 -w 15 $pid +' +test_expect_success 'kill on nonexistent pid fails' ' + test_must_fail $kill -r 0 15 999999 2>killnoexist.err && + grep -i "not found" killnoexist.err +' +test_expect_success 'wait --output returns retained stdout' ' + $sdexec -r 0 --bg --waitable --label=wait-out \ + $sh -c "echo hello from bg" && + $wait --output wait-out >waitout.out && + grep "hello from bg" waitout.out +' +test_expect_success 'wait --output returns retained stderr' ' + $sdexec -r 0 --bg --waitable --label=wait-err \ + $sh -c "echo oops >&2" && + $wait --output wait-err 2>waiterr.err && + grep "oops" waiterr.err +' +test_expect_success 'wait on non-waitable background process fails' ' + $sdexec -r 0 --bg --label=not-waitable $sleep 30 && + test_must_fail $wait not-waitable 2>notwaitable.err && + test_debug "cat notwaitable.err" && + grep -i "not waitable" notwaitable.err && + $kill -r 0 15 not-waitable +' +test_expect_success 'wait on nonexistent process fails' ' + test_must_fail $wait 999999 2>noexist.err && + grep -i "not found" noexist.err +' +# The $wait CLI blocks until the process exits, so it cannot express "park a +# wait, then probe" without a race. This helper drives Flux handles directly: +# a synchronous round-trip (barrier) exploits in-order per-client request +# processing to know when a parked wait has been registered by the module. +test_expect_success 'create bg wait helper script' ' + cat >bgwait.py <<-'"'"'EOT'"'"' && + import os + import sys + import flux + import flux.subprocess as sp + SERVICE = "sdexec" + RANK = 0 + def barrier(h): + # A round-trip on handle h. The module handles a single client in FIFO + # order, so any request sent on h beforehand has been acted upon by the + # time this returns. + sp.list(h, service=SERVICE, nodeid=RANK).get() + def park_and_exit(label): + # Park a wait on a running process, then exit. Interpreter shutdown + # closes the handle, which the broker delivers to sdexec as a disconnect. + h = flux.Flux() + sp.wait(h, label=label, service=SERVICE, nodeid=RANK) + barrier(h) + def rewait_kill(label): + # Attach a new waiter after the previous one disconnected, retrying while + # the module still reports the process as being waited on (until it has + # processed that disconnect), then kill and collect the status. + h = flux.Flux() + while True: + rpc = sp.wait(h, label=label, service=SERVICE, nodeid=RANK) + barrier(h) + if not rpc.is_ready(): + break # the wait is parked: this client now owns it + try: + rpc.get_status() + sys.exit("wait completed unexpectedly") + except OSError: + pass # already being waited on: disconnect not processed yet + sp.kill(h, signum=15, label=label, service=SERVICE, nodeid=RANK).get() + status = rpc.get_status() + if not (os.WIFSIGNALED(status) and os.WTERMSIG(status) == 15): + sys.exit(f"unexpected wait status {status}") + def double_wait(label): + # Only one outstanding waiter is allowed, so a second wait must fail. + h = flux.Flux() + rpc1 = sp.wait(h, label=label, service=SERVICE, nodeid=RANK) + barrier(h) + try: + sp.wait(h, label=label, service=SERVICE, nodeid=RANK).get_status() + sys.exit("second wait unexpectedly succeeded") + except OSError as exc: + print(exc) + return rpc1 + {"park_and_exit": park_and_exit, + "rewait_kill": rewait_kill, + "double_wait": double_wait}[sys.argv[1]](sys.argv[2]) + EOT + chmod +x bgwait.py +' +test_expect_success 'wait fails when process is already being waited on' ' + $sdexec -r 0 --bg --waitable --label=wait-busy $sleep 30 && + flux python bgwait.py double_wait wait-busy >busy.out 2>&1 && + test_debug "cat busy.out" && + grep -i "already being waited on" busy.out && + $kill -r 0 15 wait-busy && + test_expect_code 143 $wait wait-busy +' +test_expect_success 'new client can wait after previous waiter disconnects' ' + $sdexec -r 0 --bg --waitable --label=wait-reconnect $sleep 30 && + flux python bgwait.py park_and_exit wait-reconnect && + flux python bgwait.py rewait_kill wait-reconnect +' +# Usage: wait_for_ps_state LABEL STATE +# wait up to 30s for the process with the given label to reach STATE (R or Z) +wait_for_ps_state() { + retries=0 + while ! $ps -r 0 -no "{state} {label}" | grep -q "^$2 $1$"; do + retries=$(($retries+1)) + test $retries -eq 300 && return 1 # max 300 * 0.1s = 30s + sleep 0.1 + done +} +test_expect_success 'ps reports label and R state for a running process' ' + $sdexec -r 0 --bg --waitable --label=ps-running $sleep 30 && + wait_for_ps_state ps-running R && + $ps -r 0 >ps-running.out && + test_debug "cat ps-running.out" && + grep ps-running ps-running.out && + $kill -r 0 15 ps-running && + test_expect_code 143 $wait ps-running +' +test_expect_success 'ps reports Z state for a finished waitable process' ' + $sdexec -r 0 --bg --waitable --label=ps-zombie $true && + wait_for_ps_state ps-zombie Z && + $wait ps-zombie +' +# A non-waitable background process is detached, so there is nothing to wait +# on; poll the broker log until its output appears (or time out). +grep_dmesg_retry() { + retries=0 + while ! flux dmesg | grep -q "$1"; do + retries=$(($retries+1)) + test $retries -eq 300 && return 1 # max 300 * 0.1s = 30s + sleep 0.1 + done +} +test_expect_success 'non-waitable background output is logged to broker log' ' + flux dmesg -C && + $sdexec -r 0 --bg $sh -c "echo detached-output" && + grep_dmesg_retry detached-output +' +test_expect_success 'remove sdexec,sdbus modules' ' + flux exec flux module remove sdexec && + flux exec flux module remove sdbus +' +test_done