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

fix(worker): graceful job kill and catchable stop, as LSF (bug-1957)

Browse source 
A plain kill sent one SIGTERM and never escalated, and a plain stop sent
one uncatchable SIGSTOP. Following LSF's TERMINATE and SUSPEND actions, a
plain kill now sends SIGINT, then SIGTERM, then SIGKILL, and a plain stop
sends SIGTSTP (the job can catch it, prepare and stop itself), then SIGSTOP
if it has not stopped; each next signal follows after the grace period and
only while the job is alive (or not yet stopped). A resume cancels a
pending stop; the job's exit ends any sequence. An explicit signal is
delivered once; a force kill is SIGKILL at once.

- JobSignal gains grace (capnp @2, xce_signal_t.grace): > 0 asks for the
  sequence, in seconds between signals.
- apiserver: job.terminate_interval (default 10 s, LSF's
  JOB_TERMINATE_INTERVAL); jobCtrlParams sends kill = SIGKILL and stop =
  SIGSTOP with that grace when no signal is given; the handler no longer
  pre-fills a default signal.
- worker: grace travels through the job-control task, the kept-signal table
  (pending_signal_*_ex) and the IPC message; the job process runs the
  sequence (job_signal_seq_*, uv timer) and checks stop state in /proc.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-10 10:00:36 +00:00
parent e5a0e42b66
commit 367af11af1
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
24 changed files with 580 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/frontend/apiserver/backend/clib.go
View file

@ -703,6 +703,7 @@ func ControlJob(jobid uint64, action int, content string) error {
var sig C.xce_signal_t
sig.signo = C.int(args.Signo)
sig.force = C.int(args.Force)
sig.grace = C.int(args.Grace)
C.jobctrl_set_siginfo(&ctrl, &sig)
case models.JOB_CHKPNT:
var ckpt C.xce_ckpt_options_t

27
src/frontend/apiserver/backend/convert_pure.go
View file

@ -2,6 +2,7 @@ package backend
import (
"apiserver/models"
"apiserver/utils"
"encoding/json"
"errors"
"fmt"
@ -222,6 +223,7 @@ func jobCtrlUnionDiscriminant(action int) int {
type jobCtrlArgs struct {
Signo int
Force int
Grace int // seconds; > 0: graceful kill/stop sequence (bug-1957)
CkptKill int
Method string
}
@ -266,6 +268,11 @@ func jobCtrlParams(action int, content string) (jobCtrlArgs, error) {
case models.JOB_STOP, models.JOB_RESUME, models.JOB_KILL:
if signo == 0 {
signo = defaultActionSignal(action)
// bug-1957: a plain kill or stop is graceful, as LSF's
// TERMINATE and SUSPEND actions; a force kill is immediate
if action == models.JOB_STOP || (action == models.JOB_KILL && !forceKill) {
args.Grace = jobTerminateInterval()
}
}
args.Signo = signo
if action == models.JOB_KILL && forceKill {
@ -281,16 +288,30 @@ func jobCtrlParams(action int, content string) (jobCtrlArgs, error) {
// defaultActionSignal is the signal a stop/resume/kill sends when the caller
// gave none.
//
// bug-1957: kill and stop name the sequence's last signal (SIGKILL,
// SIGSTOP); with a grace period the worker sends SIGINT/SIGTERM or SIGTSTP
// first, so the job can clean up or prepare to stop.
func defaultActionSignal(action int) int {
switch action {
case models.JOB_STOP:
// bug-1954: the worker delivers the requested signal, so a plain
// stop asks for the uncatchable SIGSTOP
return models.JobSignalMapTable["STOP"] // 19
case models.JOB_RESUME:
return models.JobSignalMapTable["CONT"] // 18
}
return models.JobSignalMapTable["TERM"] // 15
return models.JobSignalMapTable["KILL"] // 9
}
// defaultJobTerminateInterval is LSF's JOB_TERMINATE_INTERVAL default.
const defaultJobTerminateInterval = 10
// jobTerminateInterval is job.terminate_interval, or the default when it is
// unset or 0.
func jobTerminateInterval() int {
if utils.JobYamlConfig != nil && utils.JobYamlConfig.TerminateInterval > 0 {
return utils.JobYamlConfig.TerminateInterval
}
return defaultJobTerminateInterval
}
// isKnownSignal reports whether n is one of models.JobSignalMapTable's values.

45
src/frontend/apiserver/backend/jobctrl_params_test.go
View file

@ -2,6 +2,7 @@ package backend
import (
"apiserver/models"
"apiserver/utils"
"encoding/json"
"testing"
)
@ -50,13 +51,17 @@ func TestJobCtrlParams_KillWithoutForce(t *testing.T) {
// An event without a signal keeps the per-action defaults.
func TestJobCtrlParams_DefaultSignalPerAction(t *testing.T) {
// bug-1957: a plain kill or stop is graceful: the worker runs LSF's
// TERMINATE (INT, TERM, KILL) or SUSPEND (TSTP, STOP) sequence with the
// configured interval; resume is a single SIGCONT
cases := []struct {
action int
want int
grace int
}{
{models.JOB_KILL, 15},
{models.JOB_STOP, 19}, // bug-1954: SIGSTOP
{models.JOB_RESUME, 18},
{models.JOB_KILL, 9, defaultJobTerminateInterval},
{models.JOB_STOP, 19, defaultJobTerminateInterval},
{models.JOB_RESUME, 18, 0},
}
for _, tc := range cases {
content := jobCtrlEventContent(t, models.JobControlEvent{UniqueJobID: 42})
@ -64,8 +69,9 @@ func TestJobCtrlParams_DefaultSignalPerAction(t *testing.T) {
if err != nil {
t.Fatalf("action %d: unexpected error %v", tc.action, err)
}
if got.Signo != tc.want {
t.Errorf("action %d: got signo %d, want %d", tc.action, got.Signo, tc.want)
if got.Signo != tc.want || got.Grace != tc.grace {
t.Errorf("action %d: got signo %d grace %d, want %d grace %d",
tc.action, got.Signo, got.Grace, tc.want, tc.grace)
}
}
}
@ -74,7 +80,7 @@ func TestJobCtrlParams_DefaultSignalPerAction(t *testing.T) {
// accepted (case-insensitive, optional SIG prefix).
func TestJobCtrlParams_BareSignalNameOrNumber(t *testing.T) {
for content, want := range map[string]int{
"USR1": 10, "usr2": 12, "SIGHUP": 1, "9": 9, "": 15,
"USR1": 10, "usr2": 12, "SIGHUP": 1, "9": 9, "": 9, // "": graceful kill (bug-1957)
} {
got, err := jobCtrlParams(models.JOB_KILL, content)
if err != nil {
@ -127,7 +133,30 @@ func TestJobCtrlParams_PlainStopSendsSigstop(t *testing.T) {
if args.Signo != models.JobSignalMapTable["STOP"] {
t.Errorf("plain stop: signo %d, want SIGSTOP (%d)", args.Signo, models.JobSignalMapTable["STOP"])
}
if models.DefaultActionSignal["stop"] != "STOP" {
t.Errorf("DefaultActionSignal[stop] = %q, want STOP", models.DefaultActionSignal["stop"])
}
// bug-1957: an explicit signal or a force kill is a single signal (no grace)
func TestJobCtrlParams_ExplicitOrForcedKillIsNotGraceful(t *testing.T) {
explicit := jobCtrlEventContent(t, models.JobControlEvent{UniqueJobID: 42, Signal: 15})
got, err := jobCtrlParams(models.JOB_KILL, explicit)
if err != nil || got.Signo != 15 || got.Grace != 0 {
t.Errorf("explicit TERM: got %+v, %v; want signo 15 grace 0", got, err)
}
forced := jobCtrlEventContent(t, models.JobControlEvent{UniqueJobID: 42, ForceKill: true})
got, err = jobCtrlParams(models.JOB_KILL, forced)
if err != nil || got.Signo != 9 || got.Force != 1 || got.Grace != 0 {
t.Errorf("force kill: got %+v, %v; want signo 9 force 1 grace 0", got, err)
}
}
// bug-1957: the interval comes from job.terminate_interval
func TestJobCtrlParams_GraceFromConfig(t *testing.T) {
old := utils.JobYamlConfig
t.Cleanup(func() { utils.JobYamlConfig = old })
utils.JobYamlConfig = &utils.JobConfig{MaxArraySize: 1000, TerminateInterval: 30}
got, err := jobCtrlParams(models.JOB_KILL, jobCtrlEventContent(t, models.JobControlEvent{UniqueJobID: 42}))
if err != nil || got.Grace != 30 {
t.Errorf("got %+v, %v; want grace 30", got, err)
}
}

1
src/frontend/apiserver/config/app.yaml
View file

@ -78,6 +78,7 @@ file_upload:
- "application/javascript" # .js file
job:
terminate_interval: 10 # seconds between the signals of a plain kill (INT, TERM, KILL) or stop (TSTP, STOP); default 10 if unset or 0
max_array_size: 1000 # max elements per job array (default 1000 if unset or 0, at most 32767; larger values are clamped to 32767 with a startup warning; negative refuses startup)
cache:

9
src/frontend/apiserver/controllers/jobs.go
View file

@ -2579,13 +2579,8 @@ func preCheckJobControlRequest(req *models.JobControlRequest) error {
return errors.New("signal must be empty for stop or resume actions")
}
// set default signal value if not provided
if req.Signal == "" {
if defaultSignal, ok := models.DefaultActionSignal[strings.ToLower(req.Action)]; ok {
req.Signal = defaultSignal
}
}
// bug-1957: no signal means the action's default, chosen by the backend
// (jobCtrlParams): a graceful kill or stop, a plain resume
// transform signal to numeric value if it is a string
if req.Signal != "" && !utils.IsNumeric(req.Signal) {

7
src/frontend/apiserver/models/jobs.go
View file

@ -125,9 +125,12 @@ var JobKillSignalControlMap = map[int][]int{
JOB_KILL: {JOBSTAT["RUN"], JOBSTAT["PEND"], JOBSTAT["SSUSP"], JOBSTAT["USUSP"], JOBSTAT["PSUSP"]},
}
// DefaultActionSignal names each action's default signal. bug-1957: kill and
// stop name the last signal of their graceful sequences (backend
// jobCtrlParams sends them with the configured grace period).
var DefaultActionSignal = map[string]string{
"kill": "TERM",
"stop": "STOP", // bug-1954: the worker delivers exactly this signal
"kill": "KILL",
"stop": "STOP",
"resume": "CONT",
}

11
src/frontend/apiserver/utils/config.go
View file

@ -94,6 +94,10 @@ type FileUploadConfig struct {
type JobConfig struct {
MaxArraySize int `yaml:"max_array_size"`
// TerminateInterval is the seconds between the signals of a plain kill
// (SIGINT, SIGTERM, SIGKILL) and a plain stop (SIGTSTP, SIGSTOP), as
// LSF's JOB_TERMINATE_INTERVAL (bug-1957)
TerminateInterval int `yaml:"terminate_interval"`
}
type CORSConfig struct {
@ -239,8 +243,13 @@ func InitConf() error {
zap.Int("configured", configuredArraySize),
zap.Int("effective", maxArraySize))
}
terminateInterval := viper.GetInt("job.terminate_interval")
if terminateInterval < 0 {
return fmt.Errorf("job.terminate_interval must not be negative: %d", terminateInterval)
}
JobYamlConfig = &JobConfig{
MaxArraySize: maxArraySize,
MaxArraySize: maxArraySize,
TerminateInterval: terminateInterval,
}
// parse cache config section

1
src/include/xce/types.h
View file

@ -672,6 +672,7 @@ typedef struct {
typedef struct {
int signo;
int force;
int grace; /* bug-1957: > 0: graceful kill/stop, seconds between signals */
} xce_signal_t;
typedef struct {

1
src/schema/controller.capnp
View file

@ -603,6 +603,7 @@ struct JobSubmit $C.mapname("xce_job_submit_t") {
struct JobSignal $C.mapname("xce_signal_t") {
signo @0: UInt32 $C.mapname("signo");
force @1: Int32 $C.mapname("force");
grace @2: Int32 $C.mapname("grace"); # bug-1957: > 0 graceful kill/stop, seconds between signals
}
struct CheckpointOptions $C.mapname("xce_ckpt_options_t"){

33
src/worker/include/job_signal.h
View file

@ -2,9 +2,42 @@
#define XCECD_JOB_SIGNAL_H
#include <stdbool.h>
#include <sys/types.h>
#include <uv.h>
/* The signal the job process group receives for a job-control request of
* `signal` with the force flag. */
int job_signal_to_deliver(int signal, bool force);
/* bug-1957: graceful kill and stop, as LSF's TERMINATE and SUSPEND actions.
* A kill (SIGKILL) runs SIGINT, SIGTERM, SIGKILL and a stop (SIGSTOP) runs
* SIGTSTP, SIGSTOP: the first signal is sent at once, each next one after
* the grace period, unless the sequence's goal is reached first (the job
* is gone, or stopped). */
typedef bool (*job_signal_send_fn)(int signo, void* arg);
typedef bool (*job_signal_done_fn)(void* arg);
typedef struct {
uv_timer_t timer;
int steps[3];
int nsteps;
int next;
unsigned grace_ms;
job_signal_send_fn send;
job_signal_done_fn done;
void* arg;
bool active;
} job_signal_seq_t;
int job_signal_seq_init(job_signal_seq_t* seq, uv_loop_t* loop);
/* false (nothing sent) when `signo` has no graceful sequence */
bool job_signal_seq_start(job_signal_seq_t* seq, int signo, unsigned grace_ms,
job_signal_send_fn send, job_signal_done_fn done,
void* arg);
void job_signal_seq_cancel(job_signal_seq_t* seq);
bool job_signal_seq_active(const job_signal_seq_t* seq);
/* true when process `pid` is stopped (/proc/<pid>/stat state T) */
bool job_pid_is_stopped(pid_t pid);
#endif

6
src/worker/include/pending_signal.h
View file

@ -25,11 +25,17 @@ void pending_signal_start(uint64_t jobid);
* 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);
/* bug-1957: as pending_signal_record(), with the graceful kill/stop grace */
bool pending_signal_record_ex(uint64_t jobid, int signo, bool force, int grace);
/* 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);
/* bug-1957: as pending_signal_take(), also returning the grace (when
* non-NULL) */
bool pending_signal_take_ex(uint64_t jobid, int* signo, bool* force,
int* grace);
/* Number of pending starts (for tests and diagnostics). */
int pending_signal_count(void);

1
src/worker/include/thread_pool.h
View file

@ -35,6 +35,7 @@ typedef struct {
pid_t target_pid;
int signal;
int force;
int grace; /* bug-1957 */
bool checkpoint;
int ckpt_kill;
char* ckpt_method;

7
src/worker/include/uv_ipc.h
View file

@ -79,6 +79,7 @@ typedef struct {
uint64_t job_id;
int signal;
bool force;
int grace; /* bug-1957: > 0 graceful kill/stop, seconds */
} job_signal_t;
typedef struct {
@ -138,7 +139,8 @@ typedef void (*worker_reconnect_callback_t)(bool connected);
typedef void (*job_status_callback_t)(job_status_t* job_status);
typedef void (*client_disconnect_callback_t)(uint64_t client_id);
typedef void (*heartbeat_callback_t)(uv_stream_t* client);
typedef void (*job_signal_callback_t)(uint64_t job_id, int signal, bool force);
typedef void (*job_signal_callback_t)(uint64_t job_id, int signal, bool force,
int grace);
typedef void (*job_ckpt_callback_t)(uint64_t job_id, int kill,
const char* method);
typedef void (*job_ckpt_result_callback_t)(uint64_t job_id, int status,
@ -152,7 +154,8 @@ bool ipc_master_check_connection();
void ipc_master_set_status_callback(job_status_callback_t callback);
void ipc_master_set_disconnect_callback(client_disconnect_callback_t callback);
void ipc_master_set_heartbeat_callback(heartbeat_callback_t callback);
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force);
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force,
int grace);
bool ipc_master_send_job_ckpt(uint64_t job_id, int kill, const char* method);
void ipc_master_set_ckpt_callback(job_ckpt_callback_t callback);
void ipc_master_set_ckpt_result_callback(job_ckpt_result_callback_t callback);

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

@ -1568,6 +1568,7 @@ static int handle_jobcontrol(xce_request_t* request)
task_ctx->jobid = d->jobid;
task_ctx->signal = d->u.siginfo->signo;
task_ctx->force = d->u.siginfo->force;
task_ctx->grace = d->u.siginfo->grace;
/* bug-1862: vtask thread vs the main loop reaping entries -- read
* what we need under the list lock, never keep the pointer */
@ -1583,8 +1584,9 @@ static int handle_jobcontrol(xce_request_t* request)
* 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)) {
&& pending_signal_record_ex(d->jobid, d->u.siginfo->signo,
d->u.siginfo->force != 0,
d->u.siginfo->grace)) {
XCECD_LOG(XCE_LOG_DEBUG,
"Job %s has not started yet; signal %d will be "
"delivered when it runs",
@ -1616,7 +1618,7 @@ static int handle_jobcontrol(xce_request_t* request)
#if !defined(VIRTUAL_CLUSTER)
if (!ipc_master_send_job_signal(task_ctx->jobid, task_ctx->signal,
task_ctx->force)) {
task_ctx->force, task_ctx->grace)) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to submit job control task for job ID: %s",
jobidStr);

92
src/worker/src/job_signal.c
View file

@ -1,4 +1,6 @@
#include <signal.h>
#include <stdio.h>
#include <string.h>
#include "job_signal.h"
@ -13,3 +15,93 @@ int job_signal_to_deliver(int signal, bool force)
return signal;
}
/* bug-1957: graceful kill and stop (job_signal.h) */
static void seq_on_timer(uv_timer_t* timer)
{
job_signal_seq_t* seq = timer->data;
if (!seq->active) {
return;
}
if (seq->done(seq->arg)) {
job_signal_seq_cancel(seq);
return;
}
(void)seq->send(seq->steps[seq->next++], seq->arg);
if (seq->next >= seq->nsteps) {
job_signal_seq_cancel(seq);
return;
}
uv_timer_start(&seq->timer, seq_on_timer, seq->grace_ms, 0);
}
int job_signal_seq_init(job_signal_seq_t* seq, uv_loop_t* loop)
{
memset(seq, 0, sizeof(*seq));
seq->timer.data = seq;
return uv_timer_init(loop, &seq->timer);
}
bool job_signal_seq_start(job_signal_seq_t* seq, int signo, unsigned grace_ms,
job_signal_send_fn send, job_signal_done_fn done,
void* arg)
{
if (signo == SIGKILL) {
seq->steps[0] = SIGINT;
seq->steps[1] = SIGTERM;
seq->steps[2] = SIGKILL;
seq->nsteps = 3;
} else if (signo == SIGSTOP) {
seq->steps[0] = SIGTSTP;
seq->steps[1] = SIGSTOP;
seq->nsteps = 2;
} else {
return false;
}
job_signal_seq_cancel(seq);
seq->grace_ms = grace_ms;
seq->send = send;
seq->done = done;
seq->arg = arg;
seq->next = 1;
seq->active = true;
(void)send(seq->steps[0], arg);
uv_timer_start(&seq->timer, seq_on_timer, grace_ms, 0);
return true;
}
void job_signal_seq_cancel(job_signal_seq_t* seq)
{
seq->active = false;
uv_timer_stop(&seq->timer);
}
bool job_signal_seq_active(const job_signal_seq_t* seq) { return seq->active; }
bool job_pid_is_stopped(pid_t pid)
{
char path[64];
char buf[512];
FILE* f;
char* rp;
snprintf(path, sizeof(path), "/proc/%d/stat", (int)pid);
f = fopen(path, "r");
if (f == NULL) {
return false;
}
rp = fgets(buf, sizeof(buf), f);
fclose(f);
if (rp == NULL) {
return false;
}
/* "pid (comm) S ...": the state follows the last ')' */
rp = strrchr(buf, ')');
return (rp != NULL) && (rp[1] == ' ') && ((rp[2] == 'T') || (rp[2] == 't'));
}

17
src/worker/src/pending_signal.c
View file

@ -9,6 +9,7 @@ typedef struct pending_signal_entry {
uint64_t jobid;
int signo; /* 0: no signal recorded */
bool force;
int grace; /* bug-1957 */
struct pending_signal_entry* next;
} pending_signal_entry_t;
@ -43,11 +44,17 @@ void pending_signal_start(uint64_t jobid)
/* a new start of the same job does not inherit an old signal */
e->signo = 0;
e->force = false;
e->grace = 0;
}
pthread_mutex_unlock(&g_lock);
}
bool pending_signal_record(uint64_t jobid, int signo, bool force)
{
return pending_signal_record_ex(jobid, signo, force, 0);
}
bool pending_signal_record_ex(uint64_t jobid, int signo, bool force, int grace)
{
pending_signal_entry_t* e;
@ -56,6 +63,7 @@ bool pending_signal_record(uint64_t jobid, int signo, bool force)
if (e != NULL) {
e->signo = signo;
e->force = force;
e->grace = grace;
}
pthread_mutex_unlock(&g_lock);
@ -63,6 +71,12 @@ bool pending_signal_record(uint64_t jobid, int signo, bool force)
}
bool pending_signal_take(uint64_t jobid, int* signo, bool* force)
{
return pending_signal_take_ex(jobid, signo, force, NULL);
}
bool pending_signal_take_ex(uint64_t jobid, int* signo, bool* force,
int* grace)
{
pending_signal_entry_t** pp;
bool found = false;
@ -81,6 +95,9 @@ bool pending_signal_take(uint64_t jobid, int* signo, bool* force)
if (force != NULL) {
*force = e->force;
}
if (grace != NULL) {
*grace = e->grace;
}
}
free(e);
break;

7
src/worker/src/uv_ipc.c
View file

@ -658,7 +658,8 @@ static void on_worker_read(uv_stream_t* server, ssize_t nread,
worker_signal_callback(msg->data.signal.job_id,
msg->data.signal.signal,
msg->data.signal.force);
msg->data.signal.force,
msg->data.signal.grace);
ipc_worker_send_signal_ack(msg->data.signal.job_id,
msg->data.signal.signal,
@ -1265,7 +1266,8 @@ void ipc_worker_set_reconnection_callback(worker_reconnect_callback_t callback)
* @param force - force send signal
* @return bool - true if sent successfully
*/
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force,
int grace)
{
client_t* target_client = find_client_by_job_id(job_id);
char jobidStr[64];
@ -1283,6 +1285,7 @@ bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
msg.data.signal.job_id = job_id;
msg.data.signal.signal = signal;
msg.data.signal.force = force;
msg.data.signal.grace = grace;
uv_write_t* req = (uv_write_t*)calloc(1, sizeof(uv_write_t));
if (req) {

74
src/worker/src/xcecd_worker.c
View file

@ -48,6 +48,9 @@ static job_runtime_info_t job_runtime_info = { 0 };
char* log_file = NULL;
start_exdata_t* start_exdata = NULL;
static uv_timer_t rusage_timer;
/* bug-1957: graceful kill/stop sequence (see worker_signal_callback) */
static job_signal_seq_t graceful_seq;
static bool graceful_seq_ready = false;
static int job_rusage_collect_interval = 10000; // 10 seconds
// process manager will share the same event loop with ipc
@ -468,6 +471,11 @@ static void job_post_exec()
static void process_exit_callback(int pid, int exit_status, int term_signal,
void* user_data)
{
/* bug-1957: the job is gone; no further graceful-sequence signals */
if (graceful_seq_ready) {
job_signal_seq_cancel(&graceful_seq);
}
check_and_reconnect();
stop_rusage_collection();
@ -1088,7 +1096,46 @@ static bool check_process_status(pid_t target_pid, int expected_signal,
* @param force - whether signal is forced
* @return void
*/
static void worker_signal_callback(uint64_t job_id, int signal, bool force)
/* bug-1957: graceful kill (SIGINT, SIGTERM, SIGKILL) and stop (SIGTSTP,
* SIGSTOP), as LSF's TERMINATE and SUSPEND actions (job_signal.h) */
static bool graceful_send(int signo, void* arg)
{
bool ok = signal_gid_except_self(signo);
(void)arg;
/* a stopped job must run to act on SIGINT */
if (signo == SIGINT) {
(void)signal_gid_except_self(SIGCONT);
}
return ok;
}
static bool graceful_job_gone(void* arg)
{
pid_t pid = uv_proc_get_pid(&uv_proc_manager);
(void)arg;
return (pid <= 0) || (kill(pid, 0) != 0);
}
static bool graceful_job_stopped(void* arg)
{
pid_t pid = uv_proc_get_pid(&uv_proc_manager);
(void)arg;
return (pid <= 0) || job_pid_is_stopped(pid);
}
static void graceful_cancel_stop(void)
{
if (graceful_seq_ready && job_signal_seq_active(&graceful_seq)
&& (graceful_seq.steps[0] == SIGTSTP)) {
job_signal_seq_cancel(&graceful_seq);
}
}
static void worker_signal_callback(uint64_t job_id, int signal, bool force,
int grace)
{
char jobidStr[64];
fmt_jobid(jobidStr, sizeof(jobidStr), job_id);
@ -1113,6 +1160,27 @@ static void worker_signal_callback(uint64_t job_id, int signal, bool force)
bool success = false;
/* bug-1957: a plain kill or stop runs its graceful sequence */
if ((grace > 0) && !force && ((signal == SIGKILL) || (signal == SIGSTOP))
&& graceful_seq_ready) {
bool stop = (signal == SIGSTOP);
XCECD_WORKER_LOG(XCE_LOG_INFO,
"%s job %s gracefully, %d s between signals",
stop ? "Stopping" : "Terminating", jobidStr, grace);
if (job_signal_seq_start(&graceful_seq, signal,
(unsigned)grace * 1000U, graceful_send,
stop ? graceful_job_stopped
: graceful_job_gone,
NULL)) {
if (stop) {
job_runtime_info.job_status = JOB_SUSPENDED;
ipc_worker_send_status(&job_runtime_info, "Suspended");
}
return;
}
}
switch (signal) {
case SIGTERM:
case SIGKILL:
@ -1139,6 +1207,7 @@ static void worker_signal_callback(uint64_t job_id, int signal, bool force)
case SIGCONT:
XCECD_WORKER_LOG(XCE_LOG_INFO, "Resuming job %s", jobidStr);
graceful_cancel_stop(); /* bug-1957: a resume ends a pending stop */
success = signal_gid_except_self(SIGCONT);
if (success) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG2, "Job %s resumed successfully",
@ -1458,6 +1527,9 @@ int main(int argc, char* argv[])
_exit(1);
}
/* bug-1957: timer for graceful kill/stop sequences */
graceful_seq_ready = (job_signal_seq_init(&graceful_seq, uv_loop) == 0);
do {
ipc_worker_run(UV_RUN_NOWAIT);

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

@ -362,9 +362,10 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
{
int signo = 0;
bool force = false;
int grace = 0;
if (pending_signal_take(worker->job_info.jobid,
&signo, &force)) {
if (pending_signal_take_ex(worker->job_info.jobid,
&signo, &force, &grace)) {
XCECD_LOG(XCE_LOG_DEBUG,
"Job %s: delivering signal %d kept "
"while it was starting",
@ -373,7 +374,8 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
worker->killed_signal = signo;
#else
if (!ipc_master_send_job_signal(
worker->job_info.jobid, signo, force)) {
worker->job_info.jobid, signo, force,
grace)) {
XCECD_LOG(XCE_LOG_ERROR,
"Job %s: failed to deliver kept "
"signal %d",

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

@ -117,8 +117,10 @@ bool parse_limits(const char* exdata, res_limits_t* limits)
return true;
}
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force,
int grace)
{
(void)grace;
(void)job_id;
(void)signal;
(void)force;

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

@ -84,8 +84,10 @@ 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)
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force,
int grace)
{
(void)grace;
(void)job_id;
(void)signal;
(void)force;

247
src/worker/testsuite/test_job_signal.c
View file

@ -1,13 +1,31 @@
/* bug-1954: the job receives the signal the user asked for. */
/* bug-1954: the job receives the signal the user asked for.
* bug-1957: graceful kill (INT, TERM, KILL) and stop (TSTP, STOP). */
#include <fcntl.h>
#include <signal.h>
#include <stdlib.h>
#include <sys/wait.h>
#include <unistd.h>
#include <unity/unity.h>
#include "job_signal.h"
void setUp(void) {}
void tearDown(void) {}
/* children spawned by the current test: killed in tearDown even when the
* test fails early, so none outlives the run (they hold no output pipe) */
static pid_t spawned[4];
static int nspawned;
void setUp(void) { nspawned = 0; }
void tearDown(void)
{
for (int i = 0; i < nspawned; i++) {
kill(spawned[i], SIGKILL);
waitpid(spawned[i], NULL, 0);
}
nspawned = 0;
}
static void test_requested_signal_is_delivered(void)
{
@ -25,10 +43,233 @@ static void test_force_kill_escalates_terminate(void)
TEST_ASSERT_EQUAL_INT(SIGKILL, job_signal_to_deliver(SIGKILL, true));
}
/* ── graceful sequences against real child processes ── */
typedef struct {
pid_t pid;
int sent[8];
int nsent;
bool reaped;
} child_t;
static bool child_send(int signo, void* arg)
{
child_t* c = arg;
c->sent[c->nsent++] = signo;
return kill(c->pid, signo) == 0;
}
static bool child_gone(void* arg)
{
child_t* c = arg;
int st;
if (!c->reaped && waitpid(c->pid, &st, WNOHANG) == c->pid) {
c->reaped = true;
}
return c->reaped;
}
static bool child_stopped(void* arg)
{
return job_pid_is_stopped(((child_t*)arg)->pid);
}
static void on_tstp_ignore(int sig) { (void)sig; }
/* a job that prepares, then stops itself (the catchable-stop use case) */
static void on_tstp_stop_self(int sig)
{
(void)sig;
raise(SIGSTOP);
}
enum { IGNORE_INT_TERM, DEFAULT, CATCH_TSTP, TSTP_STOPS_SELF };
static pid_t spawn(int mode)
{
pid_t pid = fork();
TEST_ASSERT_TRUE(pid >= 0);
if (pid == 0) {
int devnull = open("/dev/null", O_RDWR);
if (devnull >= 0) {
dup2(devnull, STDOUT_FILENO);
dup2(devnull, STDERR_FILENO);
}
if (mode == IGNORE_INT_TERM) {
signal(SIGINT, SIG_IGN);
signal(SIGTERM, SIG_IGN);
} else if (mode == CATCH_TSTP) {
signal(SIGTSTP, on_tstp_ignore);
} else if (mode == TSTP_STOPS_SELF) {
signal(SIGTSTP, on_tstp_stop_self);
}
for (;;) {
pause();
}
}
spawned[nspawned++] = pid;
usleep(50 * 1000); /* let the child install its handlers */
return pid;
}
static void run_until_idle(uv_loop_t* loop, job_signal_seq_t* seq)
{
for (int i = 0; i < 400 && job_signal_seq_active(seq); i++) {
uv_run(loop, UV_RUN_NOWAIT);
usleep(5 * 1000);
}
}
static void finish(child_t* c)
{
if (!c->reaped) {
kill(c->pid, SIGKILL);
waitpid(c->pid, NULL, 0);
}
}
static void test_kill_escalates_to_sigkill_when_ignored(void)
{
uv_loop_t loop;
job_signal_seq_t seq;
child_t c = { .pid = spawn(IGNORE_INT_TERM) };
uv_loop_init(&loop);
TEST_ASSERT_EQUAL_INT(0, job_signal_seq_init(&seq, &loop));
TEST_ASSERT_TRUE(job_signal_seq_start(&seq, SIGKILL, 100, child_send,
child_gone, &c));
run_until_idle(&loop, &seq);
(void)child_gone(&c);
TEST_ASSERT_EQUAL_INT(3, c.nsent);
TEST_ASSERT_EQUAL_INT(SIGINT, c.sent[0]);
TEST_ASSERT_EQUAL_INT(SIGTERM, c.sent[1]);
TEST_ASSERT_EQUAL_INT(SIGKILL, c.sent[2]);
TEST_ASSERT_TRUE_MESSAGE(c.reaped, "the job must be gone after SIGKILL");
finish(&c);
uv_close((uv_handle_t*)&seq.timer, NULL);
uv_run(&loop, UV_RUN_DEFAULT);
uv_loop_close(&loop);
}
static void test_kill_stops_when_job_exits_on_sigint(void)
{
uv_loop_t loop;
job_signal_seq_t seq;
child_t c = { .pid = spawn(DEFAULT) };
uv_loop_init(&loop);
TEST_ASSERT_EQUAL_INT(0, job_signal_seq_init(&seq, &loop));
TEST_ASSERT_TRUE(job_signal_seq_start(&seq, SIGKILL, 100, child_send,
child_gone, &c));
run_until_idle(&loop, &seq);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.nsent, "no TERM/KILL after the job ended");
TEST_ASSERT_EQUAL_INT(SIGINT, c.sent[0]);
finish(&c);
uv_close((uv_handle_t*)&seq.timer, NULL);
uv_run(&loop, UV_RUN_DEFAULT);
uv_loop_close(&loop);
}
static void test_stop_is_catchable_then_enforced(void)
{
uv_loop_t loop;
job_signal_seq_t seq;
child_t c = { .pid = spawn(CATCH_TSTP) };
uv_loop_init(&loop);
TEST_ASSERT_EQUAL_INT(0, job_signal_seq_init(&seq, &loop));
TEST_ASSERT_TRUE(job_signal_seq_start(&seq, SIGSTOP, 100, child_send,
child_stopped, &c));
run_until_idle(&loop, &seq);
usleep(20 * 1000);
TEST_ASSERT_EQUAL_INT(2, c.nsent);
TEST_ASSERT_EQUAL_INT_MESSAGE(SIGTSTP, c.sent[0], "stop starts catchable");
TEST_ASSERT_EQUAL_INT(SIGSTOP, c.sent[1]);
TEST_ASSERT_TRUE(job_pid_is_stopped(c.pid));
kill(c.pid, SIGCONT);
finish(&c);
uv_close((uv_handle_t*)&seq.timer, NULL);
uv_run(&loop, UV_RUN_DEFAULT);
uv_loop_close(&loop);
}
static void test_stop_not_enforced_when_job_stops_itself(void)
{
uv_loop_t loop;
job_signal_seq_t seq;
child_t c = { .pid = spawn(TSTP_STOPS_SELF) };
uv_loop_init(&loop);
TEST_ASSERT_EQUAL_INT(0, job_signal_seq_init(&seq, &loop));
TEST_ASSERT_TRUE(job_signal_seq_start(&seq, SIGSTOP, 100, child_send,
child_stopped, &c));
run_until_idle(&loop, &seq);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.nsent, "no SIGSTOP when TSTP stopped it");
TEST_ASSERT_EQUAL_INT(SIGTSTP, c.sent[0]);
kill(c.pid, SIGCONT);
finish(&c);
uv_close((uv_handle_t*)&seq.timer, NULL);
uv_run(&loop, UV_RUN_DEFAULT);
uv_loop_close(&loop);
}
static void test_cancel_stops_the_sequence(void)
{
uv_loop_t loop;
job_signal_seq_t seq;
child_t c = { .pid = spawn(CATCH_TSTP) };
uv_loop_init(&loop);
TEST_ASSERT_EQUAL_INT(0, job_signal_seq_init(&seq, &loop));
TEST_ASSERT_TRUE(job_signal_seq_start(&seq, SIGSTOP, 100, child_send,
child_stopped, &c));
job_signal_seq_cancel(&seq); /* e.g. a resume arrived */
for (int i = 0; i < 40; i++) {
uv_run(&loop, UV_RUN_NOWAIT);
usleep(5 * 1000);
}
TEST_ASSERT_EQUAL_INT(1, c.nsent);
TEST_ASSERT_FALSE(job_signal_seq_active(&seq));
TEST_ASSERT_FALSE(job_pid_is_stopped(c.pid));
finish(&c);
uv_close((uv_handle_t*)&seq.timer, NULL);
uv_run(&loop, UV_RUN_DEFAULT);
uv_loop_close(&loop);
}
static void test_other_signals_have_no_sequence(void)
{
uv_loop_t loop;
job_signal_seq_t seq;
uv_loop_init(&loop);
TEST_ASSERT_EQUAL_INT(0, job_signal_seq_init(&seq, &loop));
TEST_ASSERT_FALSE(job_signal_seq_start(&seq, SIGUSR1, 100, child_send,
child_gone, NULL));
uv_close((uv_handle_t*)&seq.timer, NULL);
uv_run(&loop, UV_RUN_DEFAULT);
uv_loop_close(&loop);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_requested_signal_is_delivered);
RUN_TEST(test_force_kill_escalates_terminate);
RUN_TEST(test_kill_escalates_to_sigkill_when_ignored);
RUN_TEST(test_kill_stops_when_job_exits_on_sigint);
RUN_TEST(test_stop_is_catchable_then_enforced);
RUN_TEST(test_stop_not_enforced_when_job_stops_itself);
RUN_TEST(test_cancel_stops_the_sequence);
RUN_TEST(test_other_signals_have_no_sequence);
return UNITY_END();
}

4
src/worker/testsuite/test_worker_launch_type_dispatch.c
View file

@ -229,8 +229,10 @@ bool parse_limits(const char* exdata, res_limits_t* limits)
return true;
}
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force,
int grace)
{
(void)grace;
(void)job_id;
(void)signal;
(void)force;

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

@ -95,8 +95,10 @@ 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)
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force,
int grace)
{
(void)grace;
(void)job_id;
(void)signal;
(void)force;