diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index ec264c85..40bcd52b 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -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 diff --git a/src/frontend/apiserver/backend/convert_pure.go b/src/frontend/apiserver/backend/convert_pure.go index 04383aa3..24ef9853 100644 --- a/src/frontend/apiserver/backend/convert_pure.go +++ b/src/frontend/apiserver/backend/convert_pure.go @@ -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. diff --git a/src/frontend/apiserver/backend/jobctrl_params_test.go b/src/frontend/apiserver/backend/jobctrl_params_test.go index c36dfd8b..fac2019e 100644 --- a/src/frontend/apiserver/backend/jobctrl_params_test.go +++ b/src/frontend/apiserver/backend/jobctrl_params_test.go @@ -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) } } diff --git a/src/frontend/apiserver/config/app.yaml b/src/frontend/apiserver/config/app.yaml index 730240b6..9927916d 100644 --- a/src/frontend/apiserver/config/app.yaml +++ b/src/frontend/apiserver/config/app.yaml @@ -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: diff --git a/src/frontend/apiserver/controllers/jobs.go b/src/frontend/apiserver/controllers/jobs.go index 62374139..5f125f6d 100644 --- a/src/frontend/apiserver/controllers/jobs.go +++ b/src/frontend/apiserver/controllers/jobs.go @@ -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) { diff --git a/src/frontend/apiserver/models/jobs.go b/src/frontend/apiserver/models/jobs.go index 5c30b4b5..5095fc19 100644 --- a/src/frontend/apiserver/models/jobs.go +++ b/src/frontend/apiserver/models/jobs.go @@ -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", } diff --git a/src/frontend/apiserver/utils/config.go b/src/frontend/apiserver/utils/config.go index ed5a9674..4d4ecf80 100644 --- a/src/frontend/apiserver/utils/config.go +++ b/src/frontend/apiserver/utils/config.go @@ -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 diff --git a/src/include/xce/types.h b/src/include/xce/types.h index dbf7071e..7628b5a0 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -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 { diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 539d523d..f03a2de1 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -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"){ diff --git a/src/worker/include/job_signal.h b/src/worker/include/job_signal.h index 02643ae5..d70d2a8c 100644 --- a/src/worker/include/job_signal.h +++ b/src/worker/include/job_signal.h @@ -2,9 +2,42 @@ #define XCECD_JOB_SIGNAL_H #include +#include +#include /* 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//stat state T) */ +bool job_pid_is_stopped(pid_t pid); + #endif diff --git a/src/worker/include/pending_signal.h b/src/worker/include/pending_signal.h index a85e4192..91cb7027 100644 --- a/src/worker/include/pending_signal.h +++ b/src/worker/include/pending_signal.h @@ -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); diff --git a/src/worker/include/thread_pool.h b/src/worker/include/thread_pool.h index 79a6102b..2e4713dd 100644 --- a/src/worker/include/thread_pool.h +++ b/src/worker/include/thread_pool.h @@ -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; diff --git a/src/worker/include/uv_ipc.h b/src/worker/include/uv_ipc.h index 35526294..e5bf1816 100644 --- a/src/worker/include/uv_ipc.h +++ b/src/worker/include/uv_ipc.h @@ -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); diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index aec314f7..e3643ba9 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -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); diff --git a/src/worker/src/job_signal.c b/src/worker/src/job_signal.c index 4461fa26..893f8001 100644 --- a/src/worker/src/job_signal.c +++ b/src/worker/src/job_signal.c @@ -1,4 +1,6 @@ #include +#include +#include #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')); +} diff --git a/src/worker/src/pending_signal.c b/src/worker/src/pending_signal.c index ae091a10..f02435a8 100644 --- a/src/worker/src/pending_signal.c +++ b/src/worker/src/pending_signal.c @@ -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; diff --git a/src/worker/src/uv_ipc.c b/src/worker/src/uv_ipc.c index af1e16dd..de348da2 100644 --- a/src/worker/src/uv_ipc.c +++ b/src/worker/src/uv_ipc.c @@ -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) { diff --git a/src/worker/src/xcecd_worker.c b/src/worker/src/xcecd_worker.c index 85961016..d2550401 100644 --- a/src/worker/src/xcecd_worker.c +++ b/src/worker/src/xcecd_worker.c @@ -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); diff --git a/src/worker/src/xcecd_worker_utils.c b/src/worker/src/xcecd_worker_utils.c index 8414edbf..18c9ea49 100644 --- a/src/worker/src/xcecd_worker_utils.c +++ b/src/worker/src/xcecd_worker_utils.c @@ -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", diff --git a/src/worker/testsuite/test_controller_connector.c b/src/worker/testsuite/test_controller_connector.c index 079c07ac..996bf9e4 100644 --- a/src/worker/testsuite/test_controller_connector.c +++ b/src/worker/testsuite/test_controller_connector.c @@ -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; diff --git a/src/worker/testsuite/test_finished_job_retention.c b/src/worker/testsuite/test_finished_job_retention.c index 39c0f131..b9ab613d 100644 --- a/src/worker/testsuite/test_finished_job_retention.c +++ b/src/worker/testsuite/test_finished_job_retention.c @@ -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; diff --git a/src/worker/testsuite/test_job_signal.c b/src/worker/testsuite/test_job_signal.c index 90e5650f..59ace938 100644 --- a/src/worker/testsuite/test_job_signal.c +++ b/src/worker/testsuite/test_job_signal.c @@ -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 #include +#include +#include +#include #include #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(); } diff --git a/src/worker/testsuite/test_worker_launch_type_dispatch.c b/src/worker/testsuite/test_worker_launch_type_dispatch.c index f6da6477..57e7394b 100644 --- a/src/worker/testsuite/test_worker_launch_type_dispatch.c +++ b/src/worker/testsuite/test_worker_launch_type_dispatch.c @@ -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; diff --git a/src/worker/testsuite/test_xcecd_resync.c b/src/worker/testsuite/test_xcecd_resync.c index f7789aa5..17cecef4 100644 --- a/src/worker/testsuite/test_xcecd_resync.c +++ b/src/worker/testsuite/test_xcecd_resync.c @@ -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;