diff --git a/src/harness/fake_resync.c b/src/harness/fake_resync.c new file mode 100644 index 00000000..dd5e3be6 --- /dev/null +++ b/src/harness/fake_resync.c @@ -0,0 +1,117 @@ +/* + * fake_resync.c -- xce_test fake worker's job-state query answer + * (bug-1922). See include/fake_resync.h. + */ +#include +#include +#include +#include + +#include "controller.capnp.h" /* JobStateStatus_* */ +#include "fake_resync.h" + +void fr_init(fr_accum_t* a) +{ + a->jobids = NULL; + a->ntotal = 0; + a->received = 0; +} + +void fr_reset(fr_accum_t* a) +{ + free(a->jobids); + fr_init(a); +} + +int fr_add_page(fr_accum_t* a, const xce_job_state_query_t* page) +{ + if (page == NULL) + return -EINVAL; + + if (page->pos == 0) { + fr_reset(a); + if (page->n_total < 0) + return -EINVAL; + if (page->n_total > 0) { + a->jobids = calloc((size_t)page->n_total, sizeof(*a->jobids)); + if (a->jobids == NULL) + return -ENOMEM; + } + a->ntotal = page->n_total; + } + + if (page->count > 0) { + if (page->jobids == NULL || a->jobids == NULL || page->pos < 0 + || page->pos + page->count > a->ntotal) { + fr_reset(a); + return -EINVAL; + } + memcpy(&a->jobids[page->pos], page->jobids, + (size_t)page->count * sizeof(*a->jobids)); + a->received += page->count; + } + + return (a->received >= a->ntotal) ? 1 : 0; +} + +static bool in_list(const uint64_t* ids, int n, uint64_t id) +{ + for (int i = 0; i < n; i++) { + if (ids[i] == id) + return true; + } + return false; +} + +int fr_build_report(const uint64_t* queried, int nqueried, + const uint64_t* running, int nrunning, + xce_job_state_entry_t*** out, int* nout) +{ + int extras = 0; + + *out = NULL; + *nout = 0; + for (int i = 0; i < nrunning; i++) { + if (!in_list(queried, nqueried, running[i])) + extras++; + } + + int total = nqueried + extras; + if (total == 0) + return 0; + + /* one block: the pointer array followed by the entries it points at */ + xce_job_state_entry_t** ptrs = calloc(1, (size_t)total * (sizeof(*ptrs) + + sizeof(**ptrs))); + if (ptrs == NULL) + return -ENOMEM; + xce_job_state_entry_t* e = (xce_job_state_entry_t*)(ptrs + total); + + int idx = 0; + for (int i = 0; i < nqueried; i++, idx++) { + e[idx].jobid = queried[i]; + e[idx].status = in_list(running, nrunning, queried[i]) + ? JobStateStatus_running + : JobStateStatus_unrecognized; + e[idx].queried = 1; + ptrs[idx] = &e[idx]; + } + for (int i = 0; i < nrunning; i++) { + if (in_list(queried, nqueried, running[i])) + continue; + e[idx].jobid = running[i]; + e[idx].status = JobStateStatus_running; + e[idx].queried = 0; + ptrs[idx] = &e[idx]; + idx++; + } + + *out = ptrs; + *nout = total; + return 0; +} + +void fr_free_report(xce_job_state_entry_t** entries) +{ + free(entries); +} diff --git a/src/harness/include/fake_resync.h b/src/harness/include/fake_resync.h new file mode 100644 index 00000000..01d98f3f --- /dev/null +++ b/src/harness/include/fake_resync.h @@ -0,0 +1,36 @@ +#ifndef _XCE_HARNESS_FAKE_RESYNC_H +#define _XCE_HARNESS_FAKE_RESYNC_H +/* + * fake_resync -- xce_test fake worker's answer to the controller's + * worker-rejoin-resync job-state query (bug-1922), mirroring xcecd's + * xcecd_resync.c: reassemble a paginated OPCode_jobStateQuery, then build the + * OPCode_jobStateReport entries -- every queried job ID once, in query + * order (running when the fake worker runs it, otherwise unrecognized), + * followed by each running job that was not queried. The fake worker keeps + * no record of finished jobs (it notifies their finish instead), so a + * finished job is reported unrecognized. + */ +#include + +#include "typemap.h" /* xce_job_state_query_t, xce_job_state_entry_t */ + +typedef struct { + uint64_t* jobids; + int ntotal; + int received; +} fr_accum_t; + +void fr_init(fr_accum_t* a); +void fr_reset(fr_accum_t* a); +/* Add one query page. Returns 1 when the query is complete (a->jobids, + * a->ntotal hold it), 0 when more pages are expected, -EINVAL/-ENOMEM on a + * bad page (the accumulator is reset). */ +int fr_add_page(fr_accum_t* a, const xce_job_state_query_t* page); +/* Build the report entries. *out is an array of *nout pointers into one + * allocation; release with fr_free_report. Returns 0 or -ENOMEM. */ +int fr_build_report(const uint64_t* queried, int nqueried, + const uint64_t* running, int nrunning, + xce_job_state_entry_t*** out, int* nout); +void fr_free_report(xce_job_state_entry_t** entries); + +#endif diff --git a/src/harness/meson.build b/src/harness/meson.build index b8fd595e..19f0e514 100644 --- a/src/harness/meson.build +++ b/src/harness/meson.build @@ -21,6 +21,7 @@ xce_test_src = [ 'joblist_display.c', 'join_worker_args.c', 'worker_notify.c', + 'fake_resync.c', 'rcfile.c', 'flagspec.c', 'interruptcall.c', @@ -94,6 +95,8 @@ harness_unit_test_src = [ 'testsuite/test_test_cli.c', 'testsuite/test_worker_notify.c', 'worker_notify.c', + 'testsuite/test_fake_resync.c', + 'fake_resync.c', ] harness_unit_test_deps = [ diff --git a/src/harness/test.c b/src/harness/test.c index 1bfc8702..43d72e9d 100644 --- a/src/harness/test.c +++ b/src/harness/test.c @@ -79,6 +79,8 @@ #include "resval_typeresolve.h" /* resval_resolve_type (resource-value-verification task 2.1) */ #include "xce/cluster.h" /* xce_parse_peers, xce_cluster_config_t */ #include "xce/types.h" +#include "fake_resync.h" /* job-state query answer (bug-1922) */ +#include "request.h" /* xce_job_state_report_req (bug-1922) */ #include "worker_notify.h" /* deferred worker notifications (bug-1921) */ #include "xcetest_interrupt.h" /* xcetest_install_sigint, xcetest_interrupted (xce-test-cli-ux task 1.3) */ @@ -6757,6 +6759,55 @@ int run_job(xce_job_exec_t* startmsg) return rc; } +/* bug-1922: answer the controller's worker-rejoin-resync job-state query + * (sent on every worker join; dispatch to this worker is withheld until the + * report arrives, or until a 15 s timeout). Pages are accumulated; on the + * last one the report -- queried jobs in order, then running jobs that were + * not queried -- is sent with a non-blocking request (no reply to wait for, + * so it is safe in this handler context). */ +static fr_accum_t g_resync_query; + +static int handle_jobstatequery(xce_request_t* request) +{ + uint64_t running[MAX_RUN_JOBS]; + int nrunning = 0; + xce_job_state_entry_t** entries = NULL; + int nentries = 0; + int rc; + + rc = fr_add_page(&g_resync_query, request->body.jobstatequerymsg); + if (rc <= 0) { + return rc; /* more pages expected, or a bad page */ + } + + for (int i = 0; i < MAX_RUN_JOBS; i++) { + if (run_jobs[i].jobid != 0 && run_jobs[i].pid > 0) { + running[nrunning++] = run_jobs[i].jobid; + } + } + rc = fr_build_report(g_resync_query.jobids, g_resync_query.ntotal, running, + nrunning, &entries, &nentries); + fr_reset(&g_resync_query); + if (rc < 0) { + return rc; + } + + xce_job_state_report_t report = { .n_total = nentries, + .pos = 0, + .count = nentries, + .entries = entries }; + rc = xce_job_state_report_req(xcecli_get_my_endpoint(active_ctx->ctx), + xcecli_get_myinfo(active_ctx->ctx), &report, + NULL, NULL, 0); + fr_free_report(entries); + if (rc < 0) { + fprintf(stderr, "job state report send failed, rc %d\n", rc); + return rc; + } + printf("job state query answered: %d job(s)\n", nentries); + return 0; +} + static int handle_startjob(xce_request_t* request) { return run_job(request->body.jobstartmsg); @@ -6962,6 +7013,9 @@ static int myreq_handler(xce_request_t* request, void* args) case OPCode_startjob: rc = handle_startjob(request); break; + case OPCode_jobStateQuery: + rc = handle_jobstatequery(request); + break; case OPCode_notify: rc = handle_notify(request); break; diff --git a/src/harness/testsuite/harness_test.c b/src/harness/testsuite/harness_test.c index 46208d6c..232a4fd5 100644 --- a/src/harness/testsuite/harness_test.c +++ b/src/harness/testsuite/harness_test.c @@ -19,6 +19,7 @@ extern void test_listfmt_json(void); extern void test_xcetest_process_entry(void); extern void test_test_cli(void); extern void test_worker_notify(void); +extern void test_fake_resync(void); /* test_test_cli.c's leak-safety net (see its own comment): reaps and * closes any clitest session left open by a test function that aborted @@ -116,6 +117,7 @@ int main(int argc, char* argv[]) TEST(test_xcetest_process_entry); TEST(test_test_cli); TEST(test_worker_notify); + TEST(test_fake_resync); return UNITY_END(); } diff --git a/src/harness/testsuite/test_fake_resync.c b/src/harness/testsuite/test_fake_resync.c new file mode 100644 index 00000000..0c67f6a6 --- /dev/null +++ b/src/harness/testsuite/test_fake_resync.c @@ -0,0 +1,94 @@ +/* + * test_fake_resync.c -- bug-1922: xce_test's fake worker answers the + * controller's job-state query (worker-rejoin-resync) instead of letting + * every join wait out the 15 s reconciliation timeout. + */ +#include +#include +#include + +#include + +#include "controller.capnp.h" /* JobStateStatus_* */ +#include "fake_resync.h" + +static void test_single_page_query_completes(void) +{ + fr_accum_t a; + uint64_t ids[] = { 11, 12 }; + xce_job_state_query_t page = { .n_total = 2, .pos = 0, .count = 2, + .jobids = ids }; + fr_init(&a); + TEST_ASSERT_EQUAL_INT(1, fr_add_page(&a, &page)); + TEST_ASSERT_EQUAL_INT(2, a.ntotal); + TEST_ASSERT_EQUAL_UINT64(11, a.jobids[0]); + TEST_ASSERT_EQUAL_UINT64(12, a.jobids[1]); + fr_reset(&a); +} + +static void test_multi_page_query_and_empty_query(void) +{ + fr_accum_t a; + uint64_t p1[] = { 1, 2 }, p2[] = { 3 }; + xce_job_state_query_t first = { .n_total = 3, .pos = 0, .count = 2, + .jobids = p1 }; + xce_job_state_query_t second = { .n_total = 3, .pos = 2, .count = 1, + .jobids = p2 }; + fr_init(&a); + TEST_ASSERT_EQUAL_INT(0, fr_add_page(&a, &first)); + TEST_ASSERT_EQUAL_INT(1, fr_add_page(&a, &second)); + TEST_ASSERT_EQUAL_UINT64(3, a.jobids[2]); + fr_reset(&a); + + /* an empty query is one complete page */ + xce_job_state_query_t empty = { .n_total = 0, .pos = 0, .count = 0 }; + TEST_ASSERT_EQUAL_INT(1, fr_add_page(&a, &empty)); + TEST_ASSERT_EQUAL_INT(0, a.ntotal); + fr_reset(&a); +} + +static void test_bad_page_rejected(void) +{ + fr_accum_t a; + uint64_t ids[] = { 1, 2, 3 }; + xce_job_state_query_t first = { .n_total = 2, .pos = 0, .count = 3, + .jobids = ids }; /* overflows n_total */ + fr_init(&a); + TEST_ASSERT_EQUAL_INT(-EINVAL, fr_add_page(&a, &first)); + TEST_ASSERT_NULL(a.jobids); +} + +static void test_report_queried_in_order_then_running_extras(void) +{ + uint64_t queried[] = { 5, 6, 7 }; + uint64_t running[] = { 7, 9, 5 }; + xce_job_state_entry_t** e = NULL; + int n = 0; + + TEST_ASSERT_EQUAL_INT(0, fr_build_report(queried, 3, running, 3, &e, &n)); + TEST_ASSERT_EQUAL_INT(4, n); + TEST_ASSERT_EQUAL_UINT64(5, e[0]->jobid); + TEST_ASSERT_EQUAL_INT(JobStateStatus_running, e[0]->status); + TEST_ASSERT_EQUAL_INT(1, e[0]->queried); + TEST_ASSERT_EQUAL_UINT64(6, e[1]->jobid); + TEST_ASSERT_EQUAL_INT(JobStateStatus_unrecognized, e[1]->status); + TEST_ASSERT_EQUAL_INT(1, e[1]->queried); + TEST_ASSERT_EQUAL_UINT64(7, e[2]->jobid); + TEST_ASSERT_EQUAL_INT(JobStateStatus_running, e[2]->status); + TEST_ASSERT_EQUAL_UINT64(9, e[3]->jobid); /* running, not queried */ + TEST_ASSERT_EQUAL_INT(JobStateStatus_running, e[3]->status); + TEST_ASSERT_EQUAL_INT(0, e[3]->queried); + fr_free_report(e); + + TEST_ASSERT_EQUAL_INT(0, fr_build_report(NULL, 0, NULL, 0, &e, &n)); + TEST_ASSERT_EQUAL_INT(0, n); + fr_free_report(e); +} + +void test_fake_resync(void) +{ + RUN_TEST(test_single_page_query_completes); + RUN_TEST(test_multi_page_query_and_empty_query); + RUN_TEST(test_bad_page_rejected); + RUN_TEST(test_report_queried_in_order_then_running_extras); +}