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

fix(controller): allow control of running jobs on connected hosts of any status (bug-1941)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-10 08:34:59 +00:00
parent 4d2e80c4bb
commit f244661b14
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
4 changed files with 344 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

90
src/controller/job.c
View file

@ -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 = {

1
src/controller/meson.build
View file

@ -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',

2
src/controller/testsuite/controller_test.c
View file

@ -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);

286
src/controller/testsuite/test_job_ctrl_hosts.c Normal file
View file

@ -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 <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#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);
}