From dcb1b4fe25ab2801aea8c974eb97e2980098dc72 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 17 Apr 2026 07:46:41 +0000 Subject: [PATCH] feat: P0-2a C-side plumbing for host control Add the full call chain for host control operations from the Go API server through the C library to the controller: - schema/controller.capnp: HostControl struct + hostctrlmsg @21 in Request.body union - include/typemap.h: xce_hostctrl_t* hostctrlmsg in body union (placed after @20 clrpreasonmsg, matching capnp ordinal) - include/client.h, xceapi.h, request.h: declarations for xcecli_hostctrl, xce_hostctrl, xce_hostctrl_req - lib/client.c: xcecli_hostctrl (validate + delegate to xce_hostctrl) - lib/xceapi.c: xce_hostctrl (validate ctx state + send + wait reply) - lib/request.c: xce_hostctrl_req (validate hostname/empty/nil, accept only HOST_OPEN/HOST_CLOSE, reject HOST_TAG/UNTAG) - lib/helper.c: Request_body_hostctrlmsg cases in xce_request_body() and xce_request_msg_name() - controller/cluster.c: OPCode_hostcontrol dispatch - controller/include/cluster.h: do_hostcontrol_op declaration - controller/worker.c: do_hostcontrol_op + fix pre-existing goto-clean bug in worker_ctrl that leaked write lock and reference on HOST_OPEN/HOST_CLOSE no-op paths TDD: 8 xce_hostctrl_req validation tests (null ep, null ctrl, null/empty hostname, HOST_DO_NONE no-op, HOST_TAG/UNTAG reject, negative action reject, HOST_OPEN/CLOSE accept via xce_request). 116/116 xcelib tests pass. CGO_ENABLED=0 Go backend tests pass. Co-Authored-By: Claude Sonnet 4.6 --- src/controller/cluster.c | 3 ++ src/controller/include/cluster.h | 1 + src/controller/worker.c | 21 +++++++-- src/include/client.h | 1 + src/include/request.h | 1 + src/include/typemap.h | 1 + src/include/xceapi.h | 1 + src/lib/client.c | 13 ++++++ src/lib/helper.c | 4 ++ src/lib/meson.build | 1 + src/lib/request.c | 31 +++++++++++++ src/lib/testsuite/test.c | 2 + src/lib/testsuite/test_request.c | 79 ++++++++++++++++++++++++++++++++ src/lib/xceapi.c | 26 +++++++++++ src/schema/controller.capnp | 6 +++ 15 files changed, 187 insertions(+), 4 deletions(-) create mode 100644 src/lib/testsuite/test_request.c diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 5ce01ed0..90d7b49e 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -189,6 +189,9 @@ static int myreq_handler(vtask_t* task) case OPCode_queuecontrol: rc = do_queuecontrol_op(c, ep, &request); break; + case OPCode_hostcontrol: + rc = do_hostcontrol_op(c, ep, &request); + break; case OPCode_runtasks: rc = do_runtasks(c, ep, &request); break; diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 5b5ed58f..6bf0cedb 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -108,6 +108,7 @@ extern int update_resval_list(cluster_t*, worker_t*, xce_resval_list_t*); extern int do_jobcontrol_op(cluster_t*, endpoint_t*, xce_request_t*); extern int job_ctrl(cluster_t*, xce_jobctrl_t*); extern int do_queuecontrol_op(cluster_t*, endpoint_t*, xce_request_t*); +extern int do_hostcontrol_op(cluster_t*, endpoint_t*, xce_request_t*); extern int do_runtasks(cluster_t*, endpoint_t*, xce_request_t*); extern int on_task_start(cluster_t*, xce_task_pidinfo_t*); extern int on_task_fail(cluster_t*, xce_task_failinfo_t*); diff --git a/src/controller/worker.c b/src/controller/worker.c index 6e282dfb..f9b08ada 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -626,10 +626,23 @@ int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl) HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w); } - unlock(w); - - dereference(w); - clean: + unlock(w); + dereference(w); + return rc; +} + +int do_hostcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request) +{ + int rc = 0; + xce_reply_t reply; + + init_reply_by_req(&reply, request); + + rc = worker_ctrl(c, request->body.hostctrlmsg); + reply.status = rc; + reply.data_which = RequestReply_data_nodata; + + xce_reply(ep, c->myhost, &reply); return rc; } diff --git a/src/include/client.h b/src/include/client.h index 39e4ab87..7d5e21cd 100644 --- a/src/include/client.h +++ b/src/include/client.h @@ -54,6 +54,7 @@ extern int xcecli_notify(xcecli_ctx_t, xce_notify_data_t*); extern int xcecli_jobctrl(xcecli_ctx_t, xce_jobctrl_t*); extern int xcecli_notify_jobchkpt(xcecli_ctx_t, uint64_t, char*); extern int xcecli_queuectrl(xcecli_ctx_t, xce_queuectrl_t*); +extern int xcecli_hostctrl(xcecli_ctx_t, xce_hostctrl_t*); extern int xcecli_runtask(xcecli_ctx_t, xce_taskrun_t*); extern int xcecli_notify_taskstart(xcecli_ctx_t, xce_task_pidinfo_t*); extern int xcecli_notify_taskfail(xcecli_ctx_t, xce_task_failinfo_t*); diff --git a/src/include/request.h b/src/include/request.h index f2ad3bc3..a48b4b1a 100644 --- a/src/include/request.h +++ b/src/include/request.h @@ -54,6 +54,7 @@ extern int extract_data(dataset_t*, xce_data_t*); extern int xce_update_req(endpoint_t*, xce_update_data_t*); extern int xce_jobctrl_req(endpoint_t*, xce_jobctrl_t*); extern int xce_queuectrl_req(endpoint_t*, xce_queuectrl_t*); +extern int xce_hostctrl_req(endpoint_t*, xce_hostctrl_t*); extern int xce_runtask_req(endpoint_t*, xce_taskrun_t*); extern int xce_set_pend_reason_req(endpoint_t*, xce_preason_set_t*); extern int xce_clear_reason_req(endpoint_t*, xce_clear_reason_t*); diff --git a/src/include/typemap.h b/src/include/typemap.h index 4bbdc8cd..e7d36eed 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -196,6 +196,7 @@ typedef struct { xce_taskrun_t* runtaskmsg; xce_preason_set_t* preasonmsg; xce_clear_reason_t* clrpreasonmsg; + xce_hostctrl_t* hostctrlmsg; } body; } xce_request_t; diff --git a/src/include/xceapi.h b/src/include/xceapi.h index 4667aaa4..0352b7cf 100644 --- a/src/include/xceapi.h +++ b/src/include/xceapi.h @@ -26,6 +26,7 @@ extern int xce_notify(xce_ctx_t*, xce_notify_data_t*); extern int xce_update_resvals(xce_ctx_t*, int, xce_resval_t**); extern int xce_jobctrl(xce_ctx_t*, xce_jobctrl_t*); extern int xce_queuectrl(xce_ctx_t*, xce_queuectrl_t*); +extern int xce_hostctrl(xce_ctx_t*, xce_hostctrl_t*); extern int xce_runtask(xce_ctx_t*, xce_taskrun_t*); extern int xce_update_jusage(xce_ctx_t*, uint64_t, int, int*, int, int*, int, xce_rusage_t**); diff --git a/src/lib/client.c b/src/lib/client.c index 977c89fa..fc94b86d 100644 --- a/src/lib/client.c +++ b/src/lib/client.c @@ -516,6 +516,19 @@ int xcecli_queuectrl(xcecli_ctx_t c, xce_queuectrl_t* ctrlinfo) return rc; } +int xcecli_hostctrl(xcecli_ctx_t c, xce_hostctrl_t* ctrlinfo) +{ + api_ctx_t* ctx = (api_ctx_t*)c; + int rc; + + if ((ctx == NULL) || (ctrlinfo == NULL)) { + return -EINVAL; + } + + rc = xce_hostctrl(&(ctx->ctx), ctrlinfo); + return rc; +} + int xcecli_runtask(xcecli_ctx_t c, xce_taskrun_t* info) { api_ctx_t* ctx = (api_ctx_t*)c; diff --git a/src/lib/helper.c b/src/lib/helper.c index 33e7d2ae..ecd904de 100644 --- a/src/lib/helper.c +++ b/src/lib/helper.c @@ -39,6 +39,8 @@ void* xce_request_body(xce_request_t* req) return req->body.jobctrlmsg; case Request_body_queuectrlmsg: return req->body.queuectrlmsg; + case Request_body_hostctrlmsg: + return req->body.hostctrlmsg; case Request_body_runtaskmsg: return req->body.runtaskmsg; case Request_body_preasonmsg: @@ -85,6 +87,8 @@ char* xce_request_msg_name(xce_request_t* req) return "jobctrlmsg"; case Request_body_queuectrlmsg: return "queuectrlmsg"; + case Request_body_hostctrlmsg: + return "hostctrlmsg"; case Request_body_runtaskmsg: return "runtaskmsg"; case Request_body_preasonmsg: diff --git a/src/lib/meson.build b/src/lib/meson.build index 19e548b0..a15a63fa 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -146,6 +146,7 @@ libxce_test_src = ['testsuite/test.c', 'testsuite/test_jobj.c', 'testsuite/test_wobj.c', 'testsuite/test_qobj.c', + 'testsuite/test_request.c', ] libxce_test_deps = [libxce_deps, librtti_dep, unity_dep, libxce_dep] diff --git a/src/lib/request.c b/src/lib/request.c index a3f11753..0409ff54 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -844,6 +844,37 @@ int xce_queuectrl_req(endpoint_t* ep, xce_queuectrl_t* ctrlinfo) return xce_request(ep, &request); } +int xce_hostctrl_req(endpoint_t* ep, xce_hostctrl_t* ctrlinfo) +{ + /* clang-format off */ + xce_request_t request = { + .opcode = OPCode_hostcontrol, + .body_which = Request_body_hostctrlmsg, + .body = { + .hostctrlmsg = ctrlinfo + } + }; + /* clang-format on */ + + if ((ep == NULL) || (ctrlinfo == NULL)) { + return -EINVAL; + } + + if ((ctrlinfo->hostname == NULL) || (strlen(ctrlinfo->hostname) == 0)) { + return -EINVAL; + } + + if (ctrlinfo->action == HOST_DO_NONE) { + return 0; + } + + if (ctrlinfo->action != HOST_OPEN && ctrlinfo->action != HOST_CLOSE) { + return -EINVAL; + } + + return xce_request(ep, &request); +} + int xce_runtask_req(endpoint_t* ep, xce_taskrun_t* runinfo) { /* clang-format off */ diff --git a/src/lib/testsuite/test.c b/src/lib/testsuite/test.c index ab9e9c26..e9cff86a 100644 --- a/src/lib/testsuite/test.c +++ b/src/lib/testsuite/test.c @@ -27,6 +27,7 @@ extern void test_kv(void); extern void test_jobobj(void); extern void test_qobj(void); extern void test_wobj(void); +extern void test_request(void); void setUp(void) { return; } @@ -123,6 +124,7 @@ int main(int argc, char* argv[]) TEST(test_jobobj); TEST(test_wobj); TEST(test_qobj); + TEST(test_request); return UNITY_END(); } diff --git a/src/lib/testsuite/test_request.c b/src/lib/testsuite/test_request.c new file mode 100644 index 00000000..caf5eef2 --- /dev/null +++ b/src/lib/testsuite/test_request.c @@ -0,0 +1,79 @@ +#include +#include +#include +#include +#include "request.h" +#include "xce/types.h" + +/* Tests for xce_hostctrl_req() validation paths. + * We test only early-return paths (before xce_request() is called) + * since a live endpoint is not available in unit tests. */ + +static void test_hostctrl_req_null_ep(void) +{ + xce_hostctrl_t ctrl = { .action = HOST_OPEN, .hostname = "host1" }; + int rc = xce_hostctrl_req(NULL, &ctrl); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +static void test_hostctrl_req_null_ctrl(void) +{ + /* non-NULL dummy ep is enough; the NULL ctrl check fires first */ + int rc = xce_hostctrl_req((endpoint_t*)0x1, NULL); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +static void test_hostctrl_req_null_hostname(void) +{ + xce_hostctrl_t ctrl = { .action = HOST_OPEN, .hostname = NULL }; + int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +static void test_hostctrl_req_empty_hostname(void) +{ + xce_hostctrl_t ctrl = { .action = HOST_OPEN, .hostname = "" }; + int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +static void test_hostctrl_req_noop(void) +{ + xce_hostctrl_t ctrl = { .action = HOST_DO_NONE, .hostname = "host1" }; + int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl); + TEST_ASSERT_EQUAL_INT(0, rc); +} + +static void test_hostctrl_req_reject_tag(void) +{ + xce_hostctrl_t ctrl = { .action = HOST_TAG, .hostname = "host1" }; + int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +static void test_hostctrl_req_reject_untag(void) +{ + xce_hostctrl_t ctrl = { .action = HOST_UNTAG, .hostname = "host1" }; + int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +static void test_hostctrl_req_reject_negative_action(void) +{ + xce_hostctrl_t ctrl = { .action = -1, .hostname = "host1" }; + int rc = xce_hostctrl_req((endpoint_t*)0x1, &ctrl); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); +} + +void test_request(void) +{ + printf("\n>> host control request validation tests\n\n"); + RUN_TEST(test_hostctrl_req_null_ep); + RUN_TEST(test_hostctrl_req_null_ctrl); + RUN_TEST(test_hostctrl_req_null_hostname); + RUN_TEST(test_hostctrl_req_empty_hostname); + RUN_TEST(test_hostctrl_req_noop); + RUN_TEST(test_hostctrl_req_reject_tag); + RUN_TEST(test_hostctrl_req_reject_untag); + RUN_TEST(test_hostctrl_req_reject_negative_action); +} diff --git a/src/lib/xceapi.c b/src/lib/xceapi.c index b235835a..96c3d384 100644 --- a/src/lib/xceapi.c +++ b/src/lib/xceapi.c @@ -561,6 +561,32 @@ int xce_queuectrl(xce_ctx_t* ctx, xce_queuectrl_t* ctrl) return status; } +int xce_hostctrl(xce_ctx_t* ctx, xce_hostctrl_t* ctrl) +{ + int rc; + int status; + + if (ctx == NULL) { + return -EINVAL; + } + + if (ctx->state != XCE_CTX_RUN) { + return -EINVAL; + } + + rc = xce_hostctrl_req(ctx->ep, ctrl); + if (rc < 0) { + return rc; + } + + rc = cbuf_get(&(ctx->wbuf), &status); + if (rc < 0) { + return rc; + } + + return status; +} + int xce_runtask(xce_ctx_t* ctx, xce_taskrun_t* runinfo) { int rc; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index ed0883e5..22af69f7 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -515,6 +515,11 @@ struct QueueControl $C.mapname("xce_queuectrl_t") { queue @1: Text $C.mapname("queue"); } +struct HostControl $C.mapname("xce_hostctrl_t") { + action @0: UInt32 $C.mapname("action"); + hostname @1: Text $C.mapname("hostname"); +} + struct RequestReply $C.mapname("xce_reply_t"){ seq @0: UInt64; opcode @1: OPCode; @@ -582,6 +587,7 @@ struct Request $C.mapname("xce_request_t"){ runtaskmsg @18: RunTask; preasonmsg @19: PendReasonSet; clrpreasonmsg @20: ClearPReason; + hostctrlmsg @21: HostControl; } }