From f244661b14a1305b5066c6056c2e05f745b36888 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 10 Oct 2026 08:34:59 +0000 Subject: [PATCH] fix(controller): allow control of running jobs on connected hosts of any status (bug-1941) Cause: job_ctrl (job.c) sent kill/stop/resume of a running job only when the host's status was exactly 0, so any status bit (CLOSED, CLOSED_BY_ADMIN, BUSY, LOCKED...) refused the request, and the refusal returned -errno, i.e. whatever errno happened to hold (often 0 or -ENOENT). Fix: new job_ctrl_to_host sends the request whenever the worker is connected (endpoint present, no XCE_WORKER_UNAVAIL); an unknown or unconnected worker fails with -E_XCE_BAD_WORKER and an ERROR naming the host. Nothing else changes for jobs on dead hosts. Test: testsuite/test_job_ctrl_hosts.c. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/controller/job.c | 90 +++--- src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_job_ctrl_hosts.c | 286 ++++++++++++++++++ 4 files changed, 344 insertions(+), 35 deletions(-) create mode 100644 src/controller/testsuite/test_job_ctrl_hosts.c diff --git a/src/controller/job.c b/src/controller/job.c index eb1bd6f5..e32bfa5e 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -2078,6 +2078,60 @@ int dispatch_a_job_test(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) } #endif /* XCE_TESTING */ +/* Sends a job control request to the worker on `host`. Returns 0 once it is + * queued (xce_jobctrl_req() only enqueues, it never blocks), or a negative + * error. + * + * bug-1941: what matters is whether the worker is connected -- an endpoint + * and no XCE_WORKER_UNAVAIL. Other status bits (CLOSED, CLOSED_BY_ADMIN, + * BUSY, LOCKED...) only stop new dispatches; the jobs already running there + * must stay controllable. A worker that is unknown or not connected gives + * -E_XCE_BAD_WORKER (it used to be -errno, whatever errno held). */ +static int job_ctrl_to_host(cluster_t* c, const char* host, + xce_jobctrl_t* ctrinfo) +{ + worker_t* w = NULL; + int rc; + + lock(c->workermgr, RTTI_LOCK_READ); + rc = key_find(c->workermgr, (void*)host, (void**)&w); + if (w != NULL) { + reference(w); + } + unlock(c->workermgr); + + if (rc < 0) { + return rc; + } + if (w == NULL) { + ERROR("job ctrl %ld: host %s is not a known worker", ctrinfo->jobid, + host); + SET_ERROR(E_XCE_BAD_WORKER); + return -E_XCE_BAD_WORKER; + } + + lock(w, RTTI_LOCK_READ); + if ((w->ep == NULL) || IS_WORKER_UNAVAIL(w->host.status)) { + ERROR("job ctrl %ld: worker %s is not connected, status %x ep %p", + ctrinfo->jobid, w->host.name, w->host.status, w->ep); + SET_ERROR(E_XCE_BAD_WORKER); + rc = -E_XCE_BAD_WORKER; + } else { + rc = xce_jobctrl_req(w->ep, ctrinfo); + if (rc < 0) { + ERROR("job ctrl %ld on host %s returns %d", ctrinfo->jobid, + w->host.name, rc); + } else { + rc = 0; + } + } + unlock(w); + + dereference(w); + + return rc; +} + int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) { uint64_t jobid = ctrinfo->jobid; @@ -2121,45 +2175,11 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) lock(j, RTTI_LOCK_WRITE); if (j->job.allocs != NULL) { xce_host_alloc_t** allocs = j->job.allocs->allocs; - worker_t* w = NULL; /* job control command will be sent to first node * for parallel jobs */ - - lock(c->workermgr, RTTI_LOCK_READ); - rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w); - if (w != NULL) { - reference(w); - } - unlock(c->workermgr); - - if (rc < 0) { - unlock(j); - dereference(j); - - return rc; - } - if (w != NULL) { - lock(w, RTTI_LOCK_READ); - /* FIXME: better method to check host status */ - if ((w->ep != NULL) && (w->host.status == 0)) { - rc = xce_jobctrl_req(w->ep, ctrinfo); - if (rc < 0) { - ERROR("job ctrl %ld on host %s returns %d", jobid, - w->host.name, rc); - } - } else { - ERROR("worker %s is not in correct state, status %x ep %p", - w->host.name, w->host.status, w->ep); - rc = -errno; - } - unlock(w); - - dereference(w); - } else { - rc = -errno; - } + rc = job_ctrl_to_host(c, allocs[0]->host, ctrinfo); } else { /* job has not been started yet */ job_failinfo_t failinfo = { diff --git a/src/controller/meson.build b/src/controller/meson.build index 6f8ab48d..125ba2a1 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -175,6 +175,7 @@ controller_unit_test_src = [ 'testsuite/test_dispatch_fail_terminal.c', 'testsuite/test_dispatch_fail_retry.c', 'testsuite/test_jobdispatch_no_double_notify.c', + 'testsuite/test_job_ctrl_hosts.c', 'testsuite/test_worker_resync.c', 'testsuite/test_standby_state_peer_idx.c', 'testsuite/test_late_joiner_catchup_kick.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index b18cbee1..43b41db3 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -111,6 +111,7 @@ extern void test_dispatch_commit(void); extern void test_dispatch_fail_terminal(void); extern void test_dispatch_fail_retry(void); extern void test_jobdispatch_no_double_notify(void); +extern void test_job_ctrl_hosts(void); extern void test_worker_resync(void); extern void test_standby_state_peer_idx(void); extern void test_late_joiner_catchup_kick(void); @@ -303,6 +304,7 @@ int main(int argc, char* argv[]) TEST(test_dispatch_fail_terminal); TEST(test_dispatch_fail_retry); TEST(test_jobdispatch_no_double_notify); + TEST(test_job_ctrl_hosts); TEST(test_worker_resync); TEST(test_standby_state_peer_idx); TEST(test_late_joiner_catchup_kick); diff --git a/src/controller/testsuite/test_job_ctrl_hosts.c b/src/controller/testsuite/test_job_ctrl_hosts.c new file mode 100644 index 00000000..251959d5 --- /dev/null +++ b/src/controller/testsuite/test_job_ctrl_hosts.c @@ -0,0 +1,286 @@ +/* test_job_ctrl_hosts.c -- which hosts job_ctrl() (job.c) sends a running + * job's control request to, and when it refuses. + * + * bug-1941: job_ctrl() refused kill/stop/resume of a running job unless its + * host's status was exactly 0, so a job on a connected host that is closed + * (CLOSED, CLOSED_BY_ADMIN), BUSY or locked could not be controlled. Only a + * host without a connection (no endpoint, or UNAVAIL) must fail, with + * -E_XCE_BAD_WORKER. + * + * The request is observed on the worker endpoints' outq: xce_jobctrl_req() + * is a non-blocking enqueue that succeeds against a zeroed endpoint with no + * iomgr (same fixture as test_dispatch_commit.c's g_dc_worker_ep); each + * queued packet is decoded back into a Request. + */ + +#include + +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "cluster.h" +#include "controller.capnp.h" +#include "dataq.h" +#include "hook.h" +#include "jobj.h" +#include "kvmgr.h" +#include "netio.h" +#include "wobj.h" +#include "xce/cluster_def.h" +#include "xce/errnox.h" +#include "xce/types.h" +#include "xcealloc.h" +#include "xstring.h" + +#define JCH_MAX_HOSTS 3 + +static int g_jch_jobobj_ready; +static cluster_t g_jch; +static endpoint_t g_jch_ep[JCH_MAX_HOSTS]; + +static void jch_setup(void) +{ + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 32, .delta = 16 }; + kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING, + .nslots = 32, .delta = 16 }; + int i; + + if (makeString((char*)"jch-probe") == NULL) + xstring_table_init(64); + + if (!g_jch_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_jch_jobobj_ready = 1; + } + + memset(&g_jch, 0, sizeof(g_jch)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_jch.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_jch.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_jch.workermgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_jch.jobmgr); + TEST_ASSERT_NOT_NULL(g_jch.workermgr); + + for (i = 0; i < JCH_MAX_HOSTS; i++) { + memset(&g_jch_ep[i], 0, sizeof(g_jch_ep[i])); + g_jch_ep[i].fd = -1; + } +} + +/* Decoded jobcontrol requests found on one endpoint's outq. */ +typedef struct { + int n; + uint64_t jobid; + uint32_t signo; + int32_t force; +} jch_obs_t; + +/* Drains ep's outq, decoding every OPCode_jobcontrol request on it. */ +static jch_obs_t jch_scan(endpoint_t* ep) +{ + jch_obs_t obs = { 0 }; + data_item_t* it; + + while ((it = endpoint_deq(ep)) != NULL) { + io_packet_t* pkt = ITPTR_TO_PACKET(it); + struct capn c; + struct Request rq; + Request_ptr rp; + + capn_init_mem(&c, (const uint8_t*)pkt->data, pkt->sz, 0); + rp.p = capn_getp(capn_root(&c), 0, 1); + read_Request(&rq, rp); + if (rq.opcode == OPCode_jobcontrol + && rq.body_which == Request_body_jobctrlmsg) { + struct JobControl jc; + + read_JobControl(&jc, rq.body.jobctrlmsg); + obs.n += 1; + obs.jobid = jc.jobid; + if (jc.u_which == JobControl_u_siginfo) { + struct JobSignal js; + + read_JobSignal(&js, jc.u.siginfo); + obs.signo = js.signo; + obs.force = js.force; + } + } + capn_free(&c); + free_packet(pkt); + free(pkt); + } + return obs; +} + +static void jch_teardown(void) +{ + int i; + + for (i = 0; i < JCH_MAX_HOSTS; i++) + (void)jch_scan(&g_jch_ep[i]); + + if (g_jch.workermgr) dereference(g_jch.workermgr); + if (g_jch.jobmgr) dereference(g_jch.jobmgr); + xce_hook_fini(&g_jch.hooksys); +} + +/* A worker named `name` with host.status `status` and endpoint `ep`. */ +static void jch_seed_worker(const char* name, int status, endpoint_t* ep) +{ + xce_worker_t wdef; + hostattrs_t* attrs; + worker_t* w = NULL; + + attrs = calloc(1, sizeof(*attrs)); + TEST_ASSERT_NOT_NULL(attrs); + attrs->hosttype = makeString("LINUX"); + attrs->hostmodel = makeString("x86"); + attrs->cpufactor = 1.0f; + attrs->maxcpus = 4; + memset(&wdef, 0, sizeof(wdef)); + wdef.name = (char*)name; + wdef.attrs = attrs; + TEST_ASSERT_NOT_NULL(makeWorker(g_jch.workermgr, &wdef)); + + lock(g_jch.workermgr, RTTI_LOCK_READ); + key_find(g_jch.workermgr, (void*)name, (void**)&w); + unlock(g_jch.workermgr); + TEST_ASSERT_NOT_NULL(w); + w->ep = ep; + w->host.status = status; +} + +/* A RUN job whose allocation spans hosts[0..nhosts-1] (hosts[0] is the head + * the job was dispatched to). */ +static void jch_seed_running_job(uint64_t jobid, int nhosts, + const char* const* hosts) +{ + job_startinfo_t startinfo; + job_submit_data_t submit; + xce_job_t jdef; + job_t* j = NULL; + xce_host_alloc_t** ha; + int i; + + memset(&startinfo, 0, sizeof(startinfo)); + startinfo.cmdline = (char*)"/bin/sleep 100"; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"jchq"; + submit.user = (char*)"tester"; + submit.start_info = &startinfo; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = JOB_STAT_RUN; + jdef.submit = &submit; + j = makeJob(g_jch.jobmgr, &jdef); + TEST_ASSERT_NOT_NULL(j); + + ha = calloc(nhosts, sizeof(*ha)); + TEST_ASSERT_NOT_NULL(ha); + for (i = 0; i < nhosts; i++) { + ha[i] = make_host_alloc((char*)hosts[i], 1, + make_allocs(1, make_alloc((char*)"slots", NULL, 0, + 1.0f, 0))); + TEST_ASSERT_NOT_NULL(ha[i]); + } + j->job.allocs = make_job_alloc(jobid, nhosts, ha); + TEST_ASSERT_NOT_NULL(j->job.allocs); +} + +static int jch_kill(uint64_t jobid, int signo, int force) +{ + xce_signal_t sig = { .signo = signo, .force = force }; + xce_jobctrl_t ctrl = { .jobid = jobid, + .action = JobControl_u_siginfo, + .u = { .siginfo = &sig } }; + + return job_ctrl(&g_jch, &ctrl); +} + +/* bug-1941: a connected host whose status carries CLOSED, CLOSED_BY_ADMIN, + * BUSY or a lock bit still gets the control request. */ +static void test_jc_connected_flagged_host_is_controllable(void) +{ + static const int statuses[] = { XCE_WORKER_CLOSED, + XCE_WORKER_CLOSED_BY_ADMIN, + XCE_WORKER_BUSY, + XCE_WORKER_LOCKEDU, + XCE_WORKER_CLOSED_BY_ADMIN + | XCE_WORKER_BUSY }; + size_t k; + + for (k = 0; k < sizeof(statuses) / sizeof(statuses[0]); k++) { + const char* hosts[] = { "jch-closed" }; + uint64_t jobid = 93100 + k; + char msg[128]; + jch_obs_t obs; + int rc; + + jch_setup(); + jch_seed_worker("jch-closed", statuses[k], &g_jch_ep[0]); + jch_seed_running_job(jobid, 1, hosts); + + rc = jch_kill(jobid, SIGTERM, 0); + obs = jch_scan(&g_jch_ep[0]); + + snprintf(msg, sizeof(msg), + "bug-1941: running job on a connected host, status %#x", + statuses[k]); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, msg); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n, msg); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(jobid, obs.jobid, msg); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(SIGTERM, obs.signo, msg); + + jch_teardown(); + } +} + +/* bug-1941: a host that is UNAVAIL, or has no endpoint, fails the request + * with -E_XCE_BAD_WORKER and nothing is sent. */ +static void test_jc_unconnected_host_fails_with_bad_worker(void) +{ + const char* hosts[] = { "jch-dead" }; + jch_obs_t obs; + int rc; + + jch_setup(); + jch_seed_worker("jch-dead", XCE_WORKER_UNAVAIL, &g_jch_ep[0]); + jch_seed_running_job(93200, 1, hosts); + + rc = jch_kill(93200, SIGTERM, 0); + obs = jch_scan(&g_jch_ep[0]); + TEST_ASSERT_EQUAL_INT_MESSAGE(-E_XCE_BAD_WORKER, rc, + "bug-1941: control of a job on an UNAVAIL host must fail clearly"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.n, + "nothing may be sent to an UNAVAIL host"); + jch_teardown(); + + jch_setup(); + jch_seed_worker("jch-dead", XCE_WORKER_CLOSED, NULL); + jch_seed_running_job(93201, 1, hosts); + + rc = jch_kill(93201, SIGTERM, 0); + TEST_ASSERT_EQUAL_INT_MESSAGE(-E_XCE_BAD_WORKER, rc, + "bug-1941: control of a job on a host with no endpoint must fail " + "clearly"); + jch_teardown(); +} + +void test_job_ctrl_hosts(void) +{ + UnitySetTestFile(__FILE__); + printf("\n>> job_ctrl() host selection (bug-1941)\n\n"); + + RUN_TEST(test_jc_connected_flagged_host_is_controllable); + RUN_TEST(test_jc_unconnected_host_fails_with_bad_worker); +}