diff --git a/src/worker/include/pending_signal.h b/src/worker/include/pending_signal.h new file mode 100644 index 00000000..a85e4192 --- /dev/null +++ b/src/worker/include/pending_signal.h @@ -0,0 +1,37 @@ +#ifndef PENDING_SIGNAL_H +#define PENDING_SIGNAL_H + +#include +#include + +/* + * 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 */ diff --git a/src/worker/meson.build b/src/worker/meson.build index 62b17fcd..4a339a44 100644 --- a/src/worker/meson.build +++ b/src/worker/meson.build @@ -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] diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 48fe6131..fada7598 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -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 #include @@ -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); } diff --git a/src/worker/src/pending_signal.c b/src/worker/src/pending_signal.c new file mode 100644 index 00000000..ae091a10 --- /dev/null +++ b/src/worker/src/pending_signal.c @@ -0,0 +1,106 @@ +#include "pending_signal.h" + +#include +#include + +/* 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; +} diff --git a/src/worker/src/xcecd_worker_utils.c b/src/worker/src/xcecd_worker_utils.c index cf00ff37..8414edbf 100644 --- a/src/worker/src/xcecd_worker_utils.c +++ b/src/worker/src/xcecd_worker_utils.c @@ -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, diff --git a/src/worker/testsuite/meson.build b/src/worker/testsuite/meson.build index 9d435798..4a196673 100644 --- a/src/worker/testsuite/meson.build +++ b/src/worker/testsuite/meson.build @@ -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']) diff --git a/src/worker/testsuite/test_controller_connector.c b/src/worker/testsuite/test_controller_connector.c index 8572bcef..6a15436e 100644 --- a/src/worker/testsuite/test_controller_connector.c +++ b/src/worker/testsuite/test_controller_connector.c @@ -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); diff --git a/src/worker/testsuite/test_finished_job_retention.c b/src/worker/testsuite/test_finished_job_retention.c index d5262105..39c0f131 100644 --- a/src/worker/testsuite/test_finished_job_retention.c +++ b/src/worker/testsuite/test_finished_job_retention.c @@ -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 diff --git a/src/worker/testsuite/test_pending_signal.c b/src/worker/testsuite/test_pending_signal.c new file mode 100644 index 00000000..e5371256 --- /dev/null +++ b/src/worker/testsuite/test_pending_signal.c @@ -0,0 +1,88 @@ +/* bug-1908: pending-start signal table (pending_signal.c). */ + +#include + +#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(); +} diff --git a/src/worker/testsuite/test_xcecd_resync.c b/src/worker/testsuite/test_xcecd_resync.c index 4a28517a..f7789aa5 100644 --- a/src/worker/testsuite/test_xcecd_resync.c +++ b/src/worker/testsuite/test_xcecd_resync.c @@ -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; }