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

fix(worker): keep a signal that arrives before the job has started (bug-1908)

Browse source 
The controller forwards a kill as soon as it has dispatched a job, but
the worker queues the start and only registers the job (and, for the
real worker, connects its IPC client) once it starts -- seconds later
for the first jobs after a worker starts. A signal arriving in between
was dropped ("Failed to find worker"), although the API reported it
accepted, and the job ran to completion.

New pending_signal table: the start handler records the job, a signal
for a queued or still-starting job is kept there, and it is delivered
when the job reaches RUNNING (virtual: killed_signal; real: IPC). A
failed start or a job that ends drops the entry.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-02 02:50:32 +00:00
parent 7d6a6e6043
commit 8b7173bb09
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
10 changed files with 370 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

37
src/worker/include/pending_signal.h Normal file
View file

@ -0,0 +1,37 @@
#ifndef PENDING_SIGNAL_H
#define PENDING_SIGNAL_H
#include <stdbool.h>
#include <stdint.h>
/*
* bug-1908: signals for jobs whose start this worker has accepted but that
* are not running yet.
*
* The controller forwards a job-control signal as soon as it has dispatched
* the job, but the worker queues the start (thread pool) and only registers
* the job in its worker list -- and, for the real worker, only connects the
* job's IPC client -- once it has started. A signal arriving in between used
* to be dropped. The worker now records the start here, keeps the latest
* signal for it, and applies that signal when the job reaches RUNNING.
*
* All functions are thread-safe.
*/
/* The worker has accepted a start for jobid (it is not running yet). */
void pending_signal_start(uint64_t jobid);
/* Record signo (and force) for jobid if its start is pending; a later
* signal replaces an earlier one. Returns true when recorded, false when
* jobid has no pending start. */
bool pending_signal_record(uint64_t jobid, int signo, bool force);
/* jobid's start has finished (it is running, or the start failed): remove
* its entry. Returns true and sets *signo / *force (when non-NULL) when a
* signal was recorded for it, false otherwise. */
bool pending_signal_take(uint64_t jobid, int* signo, bool* force);
/* Number of pending starts (for tests and diagnostics). */
int pending_signal_count(void);
#endif /* PENDING_SIGNAL_H */

4
src/worker/meson.build
View file

@ -7,6 +7,7 @@ xcecd_worker_srcs = [
'src/static_array.c',
'src/xcecd_cgroups.c',
'src/xcecd_worker_utils.c',
'src/pending_signal.c',
'src/xcecd_resync.c',
'src/controller_connector.c',
'src/xcecd_dbus_manager.c',
@ -88,7 +89,8 @@ xcecd_srcs = [
'src/xcecd_resync.c',
'src/hardware_monitor.c',
'src/node_information.c',
'src/job_information.c'
'src/job_information.c',
'src/pending_signal.c'
]
xcecd_deps = [xcecd_worker_deps]

23
src/worker/src/controller_connector.c
View file

@ -7,6 +7,7 @@
#include "xcecd_exdata.h"
#include "xcecd_resync.h"
#include "xcecd_worker_utils.h"
#include "pending_signal.h"
#include "xcelog.h"
#include <errno.h>
#include <fcntl.h>
@ -1056,6 +1057,10 @@ static int job_start_local_task(xce_request_t* request)
job_ctx->limits->cpu_affinity = NULL;
job_ctx->limits->io_device_path = NULL;
/* bug-1908: from here until the job runs, a signal for it is kept
* (pending_signal.h) instead of being dropped */
pending_signal_start(d->jobid);
if (submit_fork_task(cc_ctx->thread_pool, job_ctx) != 0) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to add task to thread pool");
@ -1568,6 +1573,21 @@ static int handle_jobcontrol(xce_request_t* request)
int worker_job_status = worker ? worker->job_status : 0;
xcecd_worker_list_unlock();
/* bug-1908: the job's start is accepted but it is not running yet
* (still queued, or registered but starting): keep the signal and
* deliver it when the job reaches RUNNING (update_worker_status) */
if ((!worker_found || worker_job_status == JOB_PENDING
|| worker_job_status == JOB_STARTING)
&& pending_signal_record(d->jobid, d->u.siginfo->signo,
d->u.siginfo->force != 0)) {
XCECD_LOG(XCE_LOG_DEBUG,
"Job %s has not started yet; signal %d will be "
"delivered when it runs",
jobidStr, d->u.siginfo->signo);
FREEUP(task_ctx);
return 0;
}
if (worker_found) {
if (worker_job_status == JOB_PENDING
|| worker_job_status == JOB_STARTING) {
@ -1985,6 +2005,9 @@ int xcecd_job_resumed(uint64_t jobid)
*/
int xcecd_job_start_error(uint64_t jobid)
{
/* bug-1908: the start is over; drop any signal kept for it */
(void)pending_signal_take(jobid, NULL, NULL);
return xcecd_notify_job_fail(jobid, JOB_EXEC_STAGE_MAIN, -1);
}

106
src/worker/src/pending_signal.c Normal file
View file

@ -0,0 +1,106 @@
#include "pending_signal.h"
#include <pthread.h>
#include <stdlib.h>
/* bug-1908: see pending_signal.h. A short singly linked list -- it holds
* only jobs whose start is queued on this worker. */
typedef struct pending_signal_entry {
uint64_t jobid;
int signo; /* 0: no signal recorded */
bool force;
struct pending_signal_entry* next;
} pending_signal_entry_t;
static pending_signal_entry_t* g_pending = NULL;
static pthread_mutex_t g_lock = PTHREAD_MUTEX_INITIALIZER;
static pending_signal_entry_t* find_locked(uint64_t jobid)
{
pending_signal_entry_t* e = g_pending;
while (e != NULL && e->jobid != jobid) {
e = e->next;
}
return e;
}
void pending_signal_start(uint64_t jobid)
{
pending_signal_entry_t* e;
pthread_mutex_lock(&g_lock);
e = find_locked(jobid);
if (e == NULL) {
e = (pending_signal_entry_t*)calloc(1, sizeof(*e));
if (e != NULL) {
e->jobid = jobid;
e->next = g_pending;
g_pending = e;
}
}
if (e != NULL) {
/* a new start of the same job does not inherit an old signal */
e->signo = 0;
e->force = false;
}
pthread_mutex_unlock(&g_lock);
}
bool pending_signal_record(uint64_t jobid, int signo, bool force)
{
pending_signal_entry_t* e;
pthread_mutex_lock(&g_lock);
e = find_locked(jobid);
if (e != NULL) {
e->signo = signo;
e->force = force;
}
pthread_mutex_unlock(&g_lock);
return (e != NULL);
}
bool pending_signal_take(uint64_t jobid, int* signo, bool* force)
{
pending_signal_entry_t** pp;
bool found = false;
pthread_mutex_lock(&g_lock);
for (pp = &g_pending; *pp != NULL; pp = &(*pp)->next) {
if ((*pp)->jobid == jobid) {
pending_signal_entry_t* e = *pp;
*pp = e->next;
if (e->signo != 0) {
found = true;
if (signo != NULL) {
*signo = e->signo;
}
if (force != NULL) {
*force = e->force;
}
}
free(e);
break;
}
}
pthread_mutex_unlock(&g_lock);
return found;
}
int pending_signal_count(void)
{
pending_signal_entry_t* e;
int n = 0;
pthread_mutex_lock(&g_lock);
for (e = g_pending; e != NULL; e = e->next) {
n++;
}
pthread_mutex_unlock(&g_lock);
return n;
}

31
src/worker/src/xcecd_worker_utils.c
View file

@ -1,4 +1,5 @@
#include "xcecd_worker_utils.h"
#include "pending_signal.h"
#include "uv_ipc.h"
#include "xcecd_dbus_manager.h"
#include "xcecd_utils.h"
@ -356,6 +357,32 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
worker->job_status = JOB_RUNNING;
xcecd_reply_startjob(job);
/* bug-1908: deliver a signal that arrived while the job
* was still starting */
{
int signo = 0;
bool force = false;
if (pending_signal_take(worker->job_info.jobid,
&signo, &force)) {
XCECD_LOG(XCE_LOG_DEBUG,
"Job %s: delivering signal %d kept "
"while it was starting",
jobidStr, signo);
#ifdef VIRTUAL_CLUSTER
worker->killed_signal = signo;
#else
if (!ipc_master_send_job_signal(
worker->job_info.jobid, signo, force)) {
XCECD_LOG(XCE_LOG_ERROR,
"Job %s: failed to deliver kept "
"signal %d",
jobidStr, signo);
}
#endif
}
}
if (db
&& !xcecd_db_update_worker(db, worker->worker_pid,
WORKER_RUNNING,
@ -403,6 +430,9 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
break;
case WORKER_DONE:
/* bug-1908: a job that ends without ever reporting RUNNING
* leaves no kept signal behind */
(void)pending_signal_take(worker->job_info.jobid, NULL, NULL);
worker->worker_status = WORKER_DONE;
worker->job_status = JOB_DONE;
XCECD_LOG(XCE_LOG_DEBUG,
@ -434,6 +464,7 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
break;
case WORKER_EXITED:
(void)pending_signal_take(worker->job_info.jobid, NULL, NULL);
worker->worker_status = WORKER_EXITED;
worker->job_status = JOB_EXITED;
XCECD_LOG(XCE_LOG_DEBUG,

22
src/worker/testsuite/meson.build
View file

@ -12,7 +12,8 @@ worker_test_incdir = include_directories('../include', '../../include', '.',
worker_test_deps = [xcecd_worker_deps, unity_dep]
controller_connector_test = executable('controller_connector_test',
'test_controller_connector.c',
['test_controller_connector.c',
'../src/pending_signal.c'],
c_args: ['-DDBUS_MANAGER',
'-DXCE_CONTROLLER_CONNECTOR_C_PATH="'
+ (meson.current_source_dir() / '..' / 'src' / 'controller_connector.c')
@ -112,7 +113,7 @@ test('worker uv_ipc job_exit_severity test', uv_ipc_job_exit_severity_test,
# finished-job retention -- white-box #include of xcecd_worker_utils.c,
# same convention as resource_sync_test above.
finished_job_retention_test = executable('finished_job_retention_test',
'test_finished_job_retention.c',
['test_finished_job_retention.c', '../src/pending_signal.c'],
include_directories: [worker_test_incdir],
dependencies: worker_test_deps,
install: false)
@ -128,7 +129,7 @@ test('worker finished_job_retention test', finished_job_retention_test,
# ../src/xcecd_worker_utils.c and ../src/xcecd_resync.c, same convention
# as finished_job_retention_test above.
xcecd_resync_test = executable('xcecd_resync_test',
'test_xcecd_resync.c',
['test_xcecd_resync.c', '../src/pending_signal.c'],
include_directories: [worker_test_incdir],
dependencies: worker_test_deps,
install: false)
@ -148,7 +149,7 @@ test('worker xcecd_resync test', xcecd_resync_test,
# own header comment for why xcecli_runtask()/xcecli_unblock_tasks() cannot
# be stubbed here the way the other job-handling externs above are.
launch_type_dispatch_test = executable('launch_type_dispatch_test',
'test_worker_launch_type_dispatch.c',
['test_worker_launch_type_dispatch.c', '../src/pending_signal.c'],
c_args: ['-DDBUS_MANAGER',
'-DXCE_CONTROLLER_CONNECTOR_C_PATH="'
+ (meson.current_source_dir() / '..' / 'src' / 'controller_connector.c')
@ -232,3 +233,16 @@ test('worker exdata_jobindex test', exdata_jobindex_test,
suite: 'worker',
timeout: 60,
env: ['ASAN_OPTIONS=detect_leaks=0'])
# bug-1908: pending-start signal table (pending_signal.c).
pending_signal_test = executable('pending_signal_test',
['test_pending_signal.c',
'../src/pending_signal.c'],
include_directories: [worker_test_incdir],
dependencies: worker_test_deps,
install: false)
test('worker pending_signal test', pending_signal_test,
suite: 'worker',
timeout: 60,
env: ['ASAN_OPTIONS=detect_leaks=0'])

50
src/worker/testsuite/test_controller_connector.c
View file

@ -81,6 +81,7 @@
/* Pull the real connect/rejoin logic under test into this TU. */
#include "../src/controller_connector.c"
#include "pending_signal.h"
#ifndef XCE_CONTROLLER_CONNECTOR_C_PATH
#error \
@ -1547,6 +1548,53 @@ static void test_fetchfile_rejects_symlink_and_hardlink_at_output_name(void)
* main
* ════════════════════════════════════════════════════════════════════════ */
/* bug-1908: a kill for a job whose start this worker has accepted but not
* yet registered (still queued in the thread pool) is kept and reported as
* handled, not dropped with "Failed to find worker". */
static void test_jobcontrol_signal_for_queued_start_is_kept(void)
{
controller_connector_ctx_t fake_cc = { 0 };
void* worker_list_head = NULL;
fake_cc.running = true;
fake_cc.worker_list = &worker_list_head;
cc_ctx = &fake_cc;
xce_signal_t sig = { .signo = 15, .force = 0 };
xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 4242 };
xce_request_t request = { 0 };
int signo = 0;
ctrl.u.siginfo = &sig;
request.body.jobctrlmsg = &ctrl;
pending_signal_start(4242);
TEST_ASSERT_EQUAL_INT(0, handle_jobcontrol(&request));
TEST_ASSERT_TRUE_MESSAGE(pending_signal_take(4242, &signo, NULL),
"the signal must be kept for the queued start");
TEST_ASSERT_EQUAL_INT(15, signo);
}
/* A kill for a job this worker knows nothing about still fails. */
static void test_jobcontrol_signal_for_unknown_job_fails(void)
{
controller_connector_ctx_t fake_cc = { 0 };
void* worker_list_head = NULL;
fake_cc.running = true;
fake_cc.worker_list = &worker_list_head;
cc_ctx = &fake_cc;
xce_signal_t sig = { .signo = 15, .force = 0 };
xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 4343 };
xce_request_t request = { 0 };
(void)pending_signal_take(4242, NULL, NULL); /* independent of above */
ctrl.u.siginfo = &sig;
request.body.jobctrlmsg = &ctrl;
TEST_ASSERT_EQUAL_INT(-1, handle_jobcontrol(&request));
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
}
int main(void)
{
xce_init_xcecd_log("/tmp/test_controller_connector.log", 1024 * 1024, 2);
@ -1554,6 +1602,8 @@ int main(void)
UNITY_BEGIN();
RUN_TEST(test_build_cluster_config_no_peers_synthesizes_single_candidate);
RUN_TEST(test_jobcontrol_signal_for_queued_start_is_kept);
RUN_TEST(test_jobcontrol_signal_for_unknown_job_fails);
RUN_TEST(
test_build_cluster_config_single_entry_peers_matches_synthesized_shape);
RUN_TEST(test_build_cluster_config_multi_candidate_passthrough);

7
src/worker/testsuite/test_finished_job_retention.c
View file

@ -84,6 +84,13 @@ bool dbus_worker_clean(xcecd_worker_t* worker, xcecd_job_t* job)
* exit(1) fatal path -- must return true, or every test below would kill
* the test process. */
bool ipc_master_check_connection() { return true; }
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
{
(void)job_id;
(void)signal;
(void)force;
return true;
}
/* block_signal()/unblock_signal() (xcecd_sig.c) -- not #include'd here
* (only xcecd_worker_utils.c is), no-op stand-ins matching the header

88
src/worker/testsuite/test_pending_signal.c Normal file
View file

@ -0,0 +1,88 @@
/* bug-1908: pending-start signal table (pending_signal.c). */
#include <unity/unity.h>
#include "pending_signal.h"
void setUp(void)
{
/* drain anything a previous test left behind */
for (uint64_t id = 1; id <= 10; id++) {
(void)pending_signal_take(id, NULL, NULL);
}
}
void tearDown(void) {}
static void test_record_requires_pending_start(void)
{
TEST_ASSERT_FALSE(pending_signal_record(1, 15, false));
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
}
static void test_recorded_signal_is_taken_once(void)
{
int signo = 0;
bool force = false;
pending_signal_start(2);
TEST_ASSERT_TRUE(pending_signal_record(2, 15, true));
TEST_ASSERT_TRUE(pending_signal_take(2, &signo, &force));
TEST_ASSERT_EQUAL_INT(15, signo);
TEST_ASSERT_TRUE(force);
TEST_ASSERT_FALSE(pending_signal_take(2, &signo, &force));
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
}
static void test_take_without_signal_removes_entry(void)
{
pending_signal_start(3);
TEST_ASSERT_EQUAL_INT(1, pending_signal_count());
TEST_ASSERT_FALSE(pending_signal_take(3, NULL, NULL));
TEST_ASSERT_EQUAL_INT(0, pending_signal_count());
TEST_ASSERT_FALSE(pending_signal_record(3, 9, false));
}
static void test_later_signal_replaces_earlier(void)
{
int signo = 0;
pending_signal_start(4);
TEST_ASSERT_TRUE(pending_signal_record(4, 15, false));
TEST_ASSERT_TRUE(pending_signal_record(4, 9, false));
TEST_ASSERT_TRUE(pending_signal_take(4, &signo, NULL));
TEST_ASSERT_EQUAL_INT(9, signo);
}
static void test_jobs_are_independent(void)
{
int signo = 0;
pending_signal_start(5);
pending_signal_start(6);
TEST_ASSERT_TRUE(pending_signal_record(6, 15, false));
TEST_ASSERT_FALSE(pending_signal_take(5, &signo, NULL));
TEST_ASSERT_TRUE(pending_signal_take(6, &signo, NULL));
TEST_ASSERT_EQUAL_INT(15, signo);
}
static void test_restart_of_same_job_resets_signal(void)
{
pending_signal_start(7);
TEST_ASSERT_TRUE(pending_signal_record(7, 15, false));
pending_signal_start(7); /* the job is dispatched again */
TEST_ASSERT_EQUAL_INT(1, pending_signal_count());
TEST_ASSERT_FALSE(pending_signal_take(7, NULL, NULL));
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_record_requires_pending_start);
RUN_TEST(test_recorded_signal_is_taken_once);
RUN_TEST(test_take_without_signal_removes_entry);
RUN_TEST(test_later_signal_replaces_earlier);
RUN_TEST(test_jobs_are_independent);
RUN_TEST(test_restart_of_same_job_resets_signal);
return UNITY_END();
}

7
src/worker/testsuite/test_xcecd_resync.c
View file

@ -95,6 +95,13 @@ bool dbus_worker_clean(xcecd_worker_t* worker, xcecd_job_t* job)
}
bool ipc_master_check_connection() { return true; }
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
{
(void)job_id;
(void)signal;
(void)force;
return true;
}
void block_signal(int sig) { (void)sig; }
void unblock_signal(int sig) { (void)sig; }