fix(worker): lock the worker list against vtask-thread readers (bug-1862)
The main loop adds and reaps worker-list entries (add_worker, cleanup_worker_list -> remove_worker) while controller request handlers on vtask threads walk the same list with no lock. A rejoin resync overlapping a job reap read a freed node (ASAN heap-use-after-free in find_worker_by_jobid), killing the worker mid chaotic HA soak. - xcecd_worker_list_lock/unlock: add_worker links a fully built entry under it; remove_worker unlinks under it and frees after. - handle_jobstatequery holds it from reading the head through the query page (the report send is a non-blocking xce_request); job control and remote task start copy what they need under it instead of keeping the entry pointer. - The finished-job retention store (appended by the main loop, read by resync) gets its own mutex around its public entry points. Test: test_build_report_safe_against_concurrent_list_mutation -- UAF 3/3 with no-op lock stubs, clean 5/5 with the fix; worker suite 10/10; chaotic HA soak 2/2 PASS, 6/6 resync jobs reconciled, no ASAN reports. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
4d77af7323
commit
c29518b21e
GPG key ID:
80533536B0EB651C
6 changed files with 208 additions and 27 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@
|
|||
#include "xcecd_utils.h"
|
||||
#include "xcelog.h"
|
||||
#include <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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) { }
|
||||
|
|
|
|||
|
|
@ -26,6 +26,8 @@
|
|||
* packet.
|
||||
*/
|
||||
|
||||
#include <pthread.h>
|
||||
#include <stdatomic.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
|
|
@ -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();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue