diff --git a/src/worker/include/xcecd_worker_utils.h b/src/worker/include/xcecd_worker_utils.h index 23c397e3..180bc125 100644 --- a/src/worker/include/xcecd_worker_utils.h +++ b/src/worker/include/xcecd_worker_utils.h @@ -133,6 +133,14 @@ uint64_t xcecd_finished_job_retention_get_window_ms(void); void xcecd_finished_job_retention_note_reconnect(uint64_t now_ms); void xcecd_finished_job_retention_set_connection_healthy(bool healthy); +/* bug-1862: the worker list is mutated only by the main loop, but read by + * controller request handlers on vtask threads. add_worker/remove_worker + * take this lock around their link/unlink; a reader holds it for its whole + * lookup and must copy out what it needs before releasing it. Never hold it + * across a blocking call. */ +void xcecd_worker_list_lock(void); +void xcecd_worker_list_unlock(void); + /* worker-rejoin-resync task 4.2 (Requirement 3.6): copies every currently * retained, non-expired (as of now_ms) finished-job entry into a freshly * heap-allocated array of xcecd_finished_job_t VALUES (not linked -- each diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index bbeeeb6b..ee94b6c0 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -1527,12 +1527,18 @@ static int handle_jobcontrol(xce_request_t* request) task_ctx->signal = d->u.siginfo->signo; task_ctx->force = d->u.siginfo->force; + /* bug-1862: vtask thread vs the main loop reaping entries -- read + * what we need under the list lock, never keep the pointer */ + xcecd_worker_list_lock(); xcecd_worker_t* worker = find_worker_by_jobid( (xcecd_worker_t*)*cc_ctx->worker_list, d->jobid); + int worker_found = (worker != NULL); + int worker_job_status = worker ? worker->job_status : 0; + xcecd_worker_list_unlock(); - if (worker) { - if (worker->job_status == JOB_PENDING - || worker->job_status == JOB_STARTING) { + if (worker_found) { + if (worker_job_status == JOB_PENDING + || worker_job_status == JOB_STARTING) { XCECD_LOG(XCE_LOG_ERROR, "Job is pending or starting, controller should not " "pass the signal to %s.", @@ -1573,7 +1579,13 @@ static int handle_jobcontrol(xce_request_t* request) "Job control task handled directly for job ID: %s", jobidStr); - worker->killed_signal = d->u.siginfo->signo; // Mark job as killed + xcecd_worker_list_lock(); + worker = find_worker_by_jobid((xcecd_worker_t*)*cc_ctx->worker_list, + d->jobid); + if (worker) { + worker->killed_signal = d->u.siginfo->signo; // Mark job as killed + } + xcecd_worker_list_unlock(); #endif } else if (d->action == JobControl_u_ckptoption) { #if !defined(VIRTUAL_CLUSTER) @@ -1649,8 +1661,7 @@ static int handle_jobstatequery(xce_request_t* request) { endpoint_t* ep = xcecli_get_my_endpoint(xcecli_ctx); xce_peerinfo_t* from = xcecli_get_myinfo(xcecli_ctx); - xcecd_worker_t* worker_list - = (cc_ctx != NULL) ? (xcecd_worker_t*)*cc_ctx->worker_list : NULL; + xcecd_worker_t* worker_list; int rc; if (ep == NULL || from == NULL) { @@ -1659,9 +1670,17 @@ static int handle_jobstatequery(xce_request_t* request) return -EINVAL; } + /* bug-1862: this runs on a vtask thread while the main loop adds and + * reaps list entries -- hold the list lock from reading the head + * through the whole page (the report build walks the list several + * times; the report send is a non-blocking xce_request). */ + xcecd_worker_list_lock(); + worker_list + = (cc_ctx != NULL) ? (xcecd_worker_t*)*cc_ctx->worker_list : NULL; rc = xcecd_resync_handle_query_page( worker_list, ep, from, request->body.jobstatequerymsg, resync_wallclock_now_ms(), XCECD_RESYNC_REPORT_PKTLIMIT); + xcecd_worker_list_unlock(); if (rc < 0) { XCECD_LOG(XCE_LOG_ERROR, "handle_jobstatequery: failed to process query page " @@ -1972,8 +1991,26 @@ int xcecd_job_remote_start(uint64_t jobid, return -1; } + /* bug-1862: copy the fields under the list lock -- the main loop may + * reap this entry while xcecli_runtask below is still in flight */ + char* exechome = NULL; + char* execcwd = NULL; + char* exdata = NULL; + char* execuser = NULL; + char* launch_type = NULL; + + xcecd_worker_list_lock(); xcecd_worker_t* worker = find_worker_by_jobid((xcecd_worker_t *)*cc_ctx->worker_list, jobid); + if (worker) { + exechome = worker->exechome ? strdup(worker->exechome) : NULL; + execcwd = worker->execcwd ? strdup(worker->execcwd) : NULL; + exdata = worker->job_exdata ? strdup(worker->job_exdata) : NULL; + execuser = worker->execuser ? strdup(worker->execuser) : NULL; + launch_type = worker->launch_type ? strdup(worker->launch_type) : NULL; + } + xcecd_worker_list_unlock(); + if (!worker) { XCECD_LOG(XCE_LOG_ERROR, "Worker not found for job ID %s", jobidStr); @@ -1981,12 +2018,12 @@ int xcecd_job_remote_start(uint64_t jobid, } xce_taskhost_t host[] = { - {.exechome = worker->exechome, - .execcwd = worker->execcwd, + {.exechome = exechome, + .execcwd = execcwd, .cmdline = strdup(command), - .exdata = worker->job_exdata, - .execuser = worker->execuser, - .launch_type = worker->launch_type, + .exdata = exdata, + .execuser = execuser, + .launch_type = launch_type, .host = strdup(hostname), .ntasks = 1}, {0} @@ -2002,6 +2039,11 @@ int xcecd_job_remote_start(uint64_t jobid, taskrun.taskhosts[0] = &host[0]; rc = xcecli_runtask(xcecli_ctx, &taskrun); + FREEUP(exechome); + FREEUP(execcwd); + FREEUP(exdata); + FREEUP(execuser); + FREEUP(launch_type); if (rc < 0) { XCECD_LOG(XCE_LOG_ERROR, "Failed to start job %s on remote host %s, rc: %d", diff --git a/src/worker/src/xcecd_worker_utils.c b/src/worker/src/xcecd_worker_utils.c index 9a5c0b3f..e88065f7 100644 --- a/src/worker/src/xcecd_worker_utils.c +++ b/src/worker/src/xcecd_worker_utils.c @@ -4,6 +4,7 @@ #include "xcecd_utils.h" #include "xcelog.h" #include +#include #include #include #include @@ -24,6 +25,10 @@ #define XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS 60000ULL static xcecd_finished_job_t* g_finished_job_retention = NULL; +/* bug-1862: the store is appended to by the main loop (job exit) and read + * by the resync query handler on a vtask thread; every public entry point + * below takes this lock around a *_locked_ body. */ +static pthread_mutex_t g_finished_job_retention_lock = PTHREAD_MUTEX_INITIALIZER; static uint64_t g_finished_job_retention_window_ms = XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS; @@ -244,6 +249,8 @@ void add_worker(xcecd_worker_t** worker_list, int worker_pid, int worker_status, add_worker_no_startinfo: + /* bug-1862: fully built before it becomes visible to readers */ + xcecd_worker_list_lock(); if (*worker_list == NULL) { *worker_list = worker; } else { @@ -254,6 +261,7 @@ add_worker_no_startinfo: current->next = worker; worker->prev = current; } + xcecd_worker_list_unlock(); } void remove_worker(xcecd_worker_t** worker_list, int worker_pid) @@ -261,6 +269,9 @@ void remove_worker(xcecd_worker_t** worker_list, int worker_pid) XCECD_LOG(XCE_LOG_DEBUG, "Removing worker %d from worker list", worker_pid); + /* bug-1862: unlink under the lock so no reader can still be holding + * this entry; free it after, once it is unreachable */ + xcecd_worker_list_lock(); xcecd_worker_t* worker = *worker_list; while (worker) { @@ -273,18 +284,21 @@ void remove_worker(xcecd_worker_t** worker_list, int worker_pid) if (worker->next) { worker->next->prev = worker->prev; } - - FREEUP(worker->exechome); - FREEUP(worker->execcwd); - FREEUP(worker->execuser); - FREEUP(worker->job_cmd); - FREEUP(worker->job_info.pidinfo.pids); - FREEUP(worker->job_exdata); - FREEUP(worker); break; } worker = worker->next; } + xcecd_worker_list_unlock(); + + if (worker) { + FREEUP(worker->exechome); + FREEUP(worker->execcwd); + FREEUP(worker->execuser); + FREEUP(worker->job_cmd); + FREEUP(worker->job_info.pidinfo.pids); + FREEUP(worker->job_exdata); + FREEUP(worker); + } } int update_worker_status(xcecd_worker_t* worker_list, int worker_pid, @@ -517,8 +531,8 @@ void cleanup_worker_list(xcecd_worker_t** worker_list, sqlite3* db) } } -void xcecd_retain_finished_job(uint64_t jobid, int exit_status, int exit_code, - uint64_t finish_time_ms) +static void retain_finished_job_locked_(uint64_t jobid, int exit_status, + int exit_code, uint64_t finish_time_ms) { char jobidStr[64]; xcecd_finished_job_t** cur = &g_finished_job_retention; @@ -568,8 +582,8 @@ void xcecd_retain_finished_job(uint64_t jobid, int exit_status, int exit_code, (unsigned long long)finish_time_ms); } -bool xcecd_finished_job_lookup(uint64_t jobid, uint64_t now_ms, - xcecd_finished_job_t* out) +static bool finished_job_lookup_locked_(uint64_t jobid, uint64_t now_ms, + xcecd_finished_job_t* out) { xcecd_finished_job_t* e; @@ -613,9 +627,20 @@ void xcecd_finished_job_retention_set_connection_healthy(bool healthy) g_finished_job_retention_conn_healthy = healthy; } -int xcecd_finished_job_retention_snapshot(uint64_t now_ms, - xcecd_finished_job_t** out, - int* out_n) +static pthread_mutex_t g_worker_list_lock = PTHREAD_MUTEX_INITIALIZER; + +void xcecd_worker_list_lock(void) +{ + pthread_mutex_lock(&g_worker_list_lock); +} + +void xcecd_worker_list_unlock(void) +{ + pthread_mutex_unlock(&g_worker_list_lock); +} + +static int retention_snapshot_locked_(uint64_t now_ms, xcecd_finished_job_t** out, + int* out_n) { xcecd_finished_job_t* e; xcecd_finished_job_t* copy; @@ -656,7 +681,7 @@ int xcecd_finished_job_retention_snapshot(uint64_t now_ms, return 0; } -void xcecd_finished_job_retention_reset(void) +static void retention_reset_locked_(void) { xcecd_finished_job_t* e = g_finished_job_retention; @@ -672,6 +697,44 @@ void xcecd_finished_job_retention_reset(void) g_finished_job_retention_conn_healthy = true; } +void xcecd_retain_finished_job(uint64_t jobid, int exit_status, int exit_code, + uint64_t finish_time_ms) +{ + pthread_mutex_lock(&g_finished_job_retention_lock); + retain_finished_job_locked_(jobid, exit_status, exit_code, finish_time_ms); + pthread_mutex_unlock(&g_finished_job_retention_lock); +} + +bool xcecd_finished_job_lookup(uint64_t jobid, uint64_t now_ms, + xcecd_finished_job_t* out) +{ + bool found; + + pthread_mutex_lock(&g_finished_job_retention_lock); + found = finished_job_lookup_locked_(jobid, now_ms, out); + pthread_mutex_unlock(&g_finished_job_retention_lock); + return found; +} + +int xcecd_finished_job_retention_snapshot(uint64_t now_ms, + xcecd_finished_job_t** out, + int* out_n) +{ + int rc; + + pthread_mutex_lock(&g_finished_job_retention_lock); + rc = retention_snapshot_locked_(now_ms, out, out_n); + pthread_mutex_unlock(&g_finished_job_retention_lock); + return rc; +} + +void xcecd_finished_job_retention_reset(void) +{ + pthread_mutex_lock(&g_finished_job_retention_lock); + retention_reset_locked_(); + pthread_mutex_unlock(&g_finished_job_retention_lock); +} + xcecd_worker_t* find_worker_by_pid(void* worker_list, int pid) { XCECD_LOG(XCE_LOG_DEBUG, "Finding worker %d in worker list", pid); diff --git a/src/worker/testsuite/test_controller_connector.c b/src/worker/testsuite/test_controller_connector.c index 5d93b62d..33e87ad3 100644 --- a/src/worker/testsuite/test_controller_connector.c +++ b/src/worker/testsuite/test_controller_connector.c @@ -184,6 +184,10 @@ void xcecd_finished_job_retention_set_connection_healthy(bool healthy) (void)healthy; } +/* bug-1862 worker-list lock (xcecd_worker_utils.c is not linked here) */ +void xcecd_worker_list_lock(void) {} +void xcecd_worker_list_unlock(void) {} + bool xcecd_resync_is_pending(void) { return g_test_resync_pending; } void xcecd_resync_reset(void) diff --git a/src/worker/testsuite/test_worker_launch_type_dispatch.c b/src/worker/testsuite/test_worker_launch_type_dispatch.c index ab5bdc88..f6da6477 100644 --- a/src/worker/testsuite/test_worker_launch_type_dispatch.c +++ b/src/worker/testsuite/test_worker_launch_type_dispatch.c @@ -262,6 +262,10 @@ void xcecd_finished_job_retention_set_connection_healthy(bool healthy) (void)healthy; } +/* bug-1862 worker-list lock (xcecd_worker_utils.c is not linked here) */ +void xcecd_worker_list_lock(void) {} +void xcecd_worker_list_unlock(void) {} + bool xcecd_resync_is_pending(void) { return false; } void xcecd_resync_reset(void) { } diff --git a/src/worker/testsuite/test_xcecd_resync.c b/src/worker/testsuite/test_xcecd_resync.c index 9f62396c..4a28517a 100644 --- a/src/worker/testsuite/test_xcecd_resync.c +++ b/src/worker/testsuite/test_xcecd_resync.c @@ -26,6 +26,8 @@ * packet. */ +#include +#include #include #include #include @@ -347,6 +349,63 @@ static void test_build_report_answers_every_queried_id_exactly_once(void) free_worker_list(list); } +/* bug-1862: the main loop adds/reaps worker-list entries (add_worker, + * remove_worker via cleanup_worker_list) and retains finished outcomes + * while a controller job-state query is answered on a vtask thread that + * walks the same list and retention store. Under ASAN this interleaving + * was a heap-use-after-free in find_worker_by_jobid. The reader holds the + * worker-list lock for the whole report, exactly as handle_jobstatequery + * does; add_worker/remove_worker and the retention store lock internally. */ +static xcecd_worker_t* g_stress_list; +static atomic_int g_stress_stop; + +static void* stress_mutator(void* arg) +{ + int pid = 1000; + + (void)arg; + while (!atomic_load(&g_stress_stop)) { + add_worker(&g_stress_list, pid, WORKER_RUNNING, + (uint64_t)(5000 + pid % 8), 1, NULL); + if (pid >= 1004) { + remove_worker(&g_stress_list, pid - 4); + } + xcecd_retain_finished_job((uint64_t)(7000 + pid % 16), WORKER_DONE, + 0, 1000); + pid++; + } + return NULL; +} + +static void test_build_report_safe_against_concurrent_list_mutation(void) +{ + pthread_t th; + uint64_t jobids[4] = { 5001, 5002, 5003, 9999 }; + int i; + + g_stress_list = NULL; + atomic_store(&g_stress_stop, 0); + xcecd_finished_job_retention_set_window_ms(1000000); + TEST_ASSERT_EQUAL_INT(0, + pthread_create(&th, NULL, stress_mutator, NULL)); + + for (i = 0; i < 20000; i++) { + xce_job_state_entry_t** entries = NULL; + int n = 0; + + xcecd_worker_list_lock(); + TEST_ASSERT_EQUAL_INT(0, xcecd_resync_build_report( + g_stress_list, jobids, 4, 2000, &entries, &n)); + xcecd_worker_list_unlock(); + xcecd_resync_free_entries(entries, n); + } + + atomic_store(&g_stress_stop, 1); + pthread_join(th, NULL); + free_worker_list(g_stress_list); + g_stress_list = NULL; +} + /* Requirement 3.6: a job the worker is running that was NOT part of the * query is reported as an extra (queried=0). */ static void test_build_report_includes_running_extra(void) @@ -1043,6 +1102,7 @@ int main(void) RUN_TEST(test_reset_clears_gate_stuck_by_a_stalled_multipage_query); RUN_TEST(test_reset_is_a_safe_noop_when_nothing_pending); + RUN_TEST(test_build_report_safe_against_concurrent_list_mutation); return UNITY_END(); }