worker: reap unattached TTY jobs via ttyd watchdog with real-pid resolution
An attach job (xcecli -I without --browse) that no client ever connects to sat in RUN forever: ttyd only starts the job command on the first websocket client, so with no client the command never runs, ttyd never exits, and the job holds its ttyd, port and per-job worker indefinitely. New exdata key "attach":1, set by the CLI for TTY jobs that are not --browse and carried through the apiserver's extendDataMap to the worker's transfer_info->attach_mode. For such jobs the worker starts a 1 s watchdog when the job process starts: - no client attached within 60 s -> reap ttyd (SIGKILL to the monitored tree), job EXITs with status message "No client attached within 60 s" - client came and the command ran and exited -> give ttyd 5 s to stop on its own (-Q), then reap: "Command exited; ttyd did not stop on its own" The reap lands in the regular process_exit_callback path (JOB_EXITED, status reason, post-exec, worker exit), so kill and cleanup behave like any other job termination. Field-critical detail: uv_proc_get_pid returns the "/bin/sh -c ttyd ..." wrapper pid, and sh does not reliably exec -- on the field ttyd runs as a child of the wrapper. Judging the wrapper's children would see ttyd itself as "the command has started" and never reap. The watchdog resolves the real ttyd pid first (/proc/<pid>/comm walk down through the sh wrapper, capped depth) and judges client activity by the real ttyd's children; killing still goes through the monitored wrapper tree so both topologies (exec'd or forked) are covered. Web-viewed sessions (--browse / DISPLAY) are untouched: no attach exdata, no watchdog -- ttyd stays until a browser connects or the job is killed, as before. Attach failure on the CLI side prints that the job will be cleaned up by the worker if no client attaches within 60 s. Tests: xcecli (attach flag matrix), apiserver controllers (attach exdata end-to-end in ExtendData), worker native suites; live t1 cluster: normal attach completes DONE with no residue, unattached job EXITs at ~60 s with the reason in the worker log and port released, Ctrl-C-then-wait completes when the command ends, --browse job stays RUN past 60 s and kills cleanly. Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
lei.chen
parent
6ce7bdbe11
commit
4f188310ba
9 changed files with 421 additions and 12 deletions
|
|
@ -2048,6 +2048,14 @@ func processJobRequest(ctx *gin.Context, req *models.JobSubmitRequest, defaultVa
|
|||
extendDataMap["interactive"] = req.Interactive
|
||||
}
|
||||
|
||||
// attach=1: a TTY job whose session ends with its command (xcecli -I
|
||||
// without --browse). The worker uses it to reap ttyd when no client ever
|
||||
// connects -- without it an unattached job would sit in RUN forever,
|
||||
// holding its ttyd, port and per-job worker.
|
||||
if req.Attach != 0 {
|
||||
extendDataMap["attach"] = req.Attach
|
||||
}
|
||||
|
||||
if len(extendDataMap) > 0 {
|
||||
updatedExtendDataBytes, err := json.Marshal(extendDataMap)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -428,6 +428,56 @@ func TestJobHandler_SmokeSubmitErrorIs500(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// TestJobHandler_SubmitInteractiveAttachExdata checks the submit handler
|
||||
// packs interactive/attach into the job's ExtendData: attach=1 only for the
|
||||
// terminal-attach TTY kind (the worker reaps ttyd when no client ever
|
||||
// connects), never for --browse or other interactive modes.
|
||||
func TestJobHandler_SubmitInteractiveAttachExdata(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
req map[string]interface{}
|
||||
wantAttach bool
|
||||
}{
|
||||
{"tty attach", map[string]interface{}{
|
||||
"job_name": "a", "command": "sleep 1", "interactive": 1, "attach": 1,
|
||||
}, true},
|
||||
{"tty web view (no attach)", map[string]interface{}{
|
||||
"job_name": "a", "command": "sleep 1", "interactive": 1,
|
||||
}, false},
|
||||
{"xjob with attach ignored worker-side but passed", map[string]interface{}{
|
||||
"job_name": "a", "command": "sleep 1", "interactive": 2, "attach": 1,
|
||||
}, true},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := newHandlerFixture(t)
|
||||
f.seedQueue("normal")
|
||||
|
||||
w := f.submit(tc.req)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String())
|
||||
}
|
||||
subs := f.submissions()
|
||||
if len(subs) != 1 {
|
||||
t.Fatalf("submitJob calls = %d, want 1", len(subs))
|
||||
}
|
||||
var ex map[string]interface{}
|
||||
if raw := subs[0].Job.ExtendData.String; raw != "" {
|
||||
if err := json.Unmarshal([]byte(raw), &ex); err != nil {
|
||||
t.Fatalf("ExtendData %q is not JSON: %v", raw, err)
|
||||
}
|
||||
}
|
||||
if got, ok := ex["interactive"]; !ok || int(got.(float64)) != tc.req["interactive"].(int) {
|
||||
t.Errorf("ExtendData interactive = %v (%v), want %d", got, ok, tc.req["interactive"])
|
||||
}
|
||||
_, hasAttach := ex["attach"]
|
||||
if hasAttach != tc.wantAttach {
|
||||
t.Errorf("ExtendData attach present = %v, want %v (exdata: %s)", hasAttach, tc.wantAttach, subs[0].Job.ExtendData.String)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestJobHandler_SmokeControlByID controls one seeded ordinary job by ID
|
||||
// and checks the dispatch is recorded once with its unique_job_id.
|
||||
func TestJobHandler_SmokeControlByID(t *testing.T) {
|
||||
|
|
|
|||
|
|
@ -669,6 +669,11 @@ type JobSubmitRequest struct {
|
|||
Exclusive bool `json:"exclusive,omitempty"`
|
||||
Rerunable bool `json:"rerunable,omitempty"`
|
||||
Interactive int `json:"interactive,omitempty"`
|
||||
// Attach marks an interactive TTY job as the terminal-attach kind
|
||||
// (xcecli -I without --browse): the session ends with the command, so
|
||||
// the worker may reap the ttyd when no client ever connects. 0 keeps
|
||||
// the web-viewed semantics.
|
||||
Attach int `json:"attach,omitempty"`
|
||||
WarningTimePeriod int `json:"warning_time_period,omitempty"`
|
||||
WarningAction string `json:"warning_action,omitempty"`
|
||||
DataSpecFile string `json:"data_spec_file,omitempty"`
|
||||
|
|
|
|||
|
|
@ -253,6 +253,12 @@ func (js *JobSubmitter) BuildJobRequest() (*models.JobSubmitRequest, error) {
|
|||
if js.interactive {
|
||||
if js.useTTY {
|
||||
jobReq.Interactive = 1 // TTY mode
|
||||
// attach jobs end with their command, web-viewed ones keep the
|
||||
// session alive for a later browser; the worker needs the
|
||||
// distinction to reap a ttyd no client ever connected to
|
||||
if !js.useBrowse {
|
||||
jobReq.Attach = 1
|
||||
}
|
||||
} else if js.useXJob {
|
||||
jobReq.Interactive = 2 // XJob mode
|
||||
} else {
|
||||
|
|
@ -520,7 +526,11 @@ func (js *JobSubmitter) RunJobsSubmit() error {
|
|||
// user asked for the web view. --browse opens the ttyd web URL
|
||||
// in the browser instead of attaching.
|
||||
if !js.useBrowse {
|
||||
return runAttachSession(serverInfo.URL, serverInfo.Token, jobID)
|
||||
err := runAttachSession(serverInfo.URL, serverInfo.Token, jobID)
|
||||
if err != nil {
|
||||
fmt.Println("note: the job will be cleaned up by the worker if no client attaches within 60 s")
|
||||
}
|
||||
return err
|
||||
}
|
||||
if err := openBrowser(url); err != nil {
|
||||
errMsg := request.ExtractErrorMessage(err)
|
||||
|
|
|
|||
|
|
@ -133,6 +133,47 @@ func TestSubmitHelpShowsArrayForms(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// Attach marks the terminal-attach kind of TTY job so the worker may reap
|
||||
// ttyd when no client ever connects; --browse keeps the web-viewed
|
||||
// semantics and must not set it.
|
||||
func TestBuildJobRequest_AttachFlagMatrix(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
interactive bool
|
||||
useTTY bool
|
||||
useBrowse bool
|
||||
useXJob bool
|
||||
wantInter int
|
||||
wantAttach int
|
||||
}{
|
||||
{"batch job", false, false, false, false, 0, 0},
|
||||
{"tty attach (bare -I)", true, true, false, false, 1, 1},
|
||||
{"tty browse", true, true, true, false, 1, 0},
|
||||
{"xjob", true, false, false, true, 2, 0},
|
||||
{"display (interactive=3)", true, false, false, false, 3, 0},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
js := NewJobSubmitter(nil)
|
||||
js.command = "sleep 1"
|
||||
js.interactive = tc.interactive
|
||||
js.useTTY = tc.useTTY
|
||||
js.useBrowse = tc.useBrowse
|
||||
js.useXJob = tc.useXJob
|
||||
req, err := js.BuildJobRequest()
|
||||
if err != nil {
|
||||
t.Fatalf("BuildJobRequest: %v", err)
|
||||
}
|
||||
if req.Interactive != tc.wantInter {
|
||||
t.Errorf("Interactive = %d, want %d", req.Interactive, tc.wantInter)
|
||||
}
|
||||
if req.Attach != tc.wantAttach {
|
||||
t.Errorf("Attach = %d, want %d", req.Attach, tc.wantAttach)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// jobStateServer answers the per-job status query with the states fed in
|
||||
// sequence, holding the last one forever; every request's arrival time is
|
||||
// recorded so tests can assert the poll cadence.
|
||||
|
|
@ -167,9 +208,9 @@ func newJobStateServer(t *testing.T, states ...string) *jobStateServer {
|
|||
_ = json.NewEncoder(w).Encode(models.JobsResponse{
|
||||
Code: 0,
|
||||
Data: struct {
|
||||
Count int `json:"count"`
|
||||
Count int `json:"count"`
|
||||
Rows []models.QueryJob `json:"rows"`
|
||||
Page_token string `json:"page_token"`
|
||||
Page_token string `json:"page_token"`
|
||||
}{Rows: []models.QueryJob{{JobStatus: held, InteractiveJobURL: "http://h:9000"}}},
|
||||
})
|
||||
}))
|
||||
|
|
@ -250,9 +291,9 @@ func TestWaitForInteractiveURL_RunRaceUsesFastInterval(t *testing.T) {
|
|||
_ = json.NewEncoder(w).Encode(models.JobsResponse{
|
||||
Code: 0,
|
||||
Data: struct {
|
||||
Count int `json:"count"`
|
||||
Count int `json:"count"`
|
||||
Rows []models.QueryJob `json:"rows"`
|
||||
Page_token string `json:"page_token"`
|
||||
Page_token string `json:"page_token"`
|
||||
}{Rows: []models.QueryJob{row}},
|
||||
})
|
||||
}))
|
||||
|
|
|
|||
|
|
@ -343,6 +343,11 @@ type JobSubmitRequest struct {
|
|||
Exclusive bool `json:"exclusive,omitempty"`
|
||||
Rerunable bool `json:"rerunable,omitempty"`
|
||||
Interactive int `json:"interactive,omitempty"`
|
||||
// Attach marks an interactive TTY job as the terminal-attach kind (xcecli
|
||||
// -I without --browse): its session ends with the command, so the worker
|
||||
// may reap the ttyd when no client ever connects. 0 (default) keeps the
|
||||
// web-viewed semantics.
|
||||
Attach int `json:"attach,omitempty"`
|
||||
WarningTimePeriod int `json:"warning_time_period,omitempty"`
|
||||
WarningAction string `json:"warning_action,omitempty"`
|
||||
DataSpecFile string `json:"data_spec_file,omitempty"`
|
||||
|
|
|
|||
|
|
@ -24,6 +24,10 @@ typedef struct {
|
|||
typedef struct {
|
||||
transfer_file_t* ops;
|
||||
int interactive_mode;
|
||||
/* attach=1 (exdata key "attach"): TTY job whose session ends with its
|
||||
* command (xcecli attach path). The worker may reap the ttyd when no
|
||||
* client ever connects; web-viewed jobs (0) keep it for a later browser. */
|
||||
int attach_mode;
|
||||
int count;
|
||||
} transfer_info_t;
|
||||
|
||||
|
|
|
|||
|
|
@ -236,6 +236,22 @@ bool parse_exdata(const char* exdata, env_var_t** env_vars,
|
|||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed interactive mode: %d",
|
||||
transfer_infos->interactive_mode);
|
||||
}
|
||||
// --- Handle 'attach' ---
|
||||
else if (strcmp(key, "attach") == 0) {
|
||||
if (cJSON_IsString(item) && item->valuestring != NULL) {
|
||||
transfer_infos->attach_mode = atoi(item->valuestring);
|
||||
} else if (cJSON_IsNumber(item)) {
|
||||
transfer_infos->attach_mode = item->valueint;
|
||||
} else {
|
||||
XCECD_WORKER_LOG(
|
||||
XCE_LOG_ERROR,
|
||||
"Error: 'attach' is neither a string nor a number.");
|
||||
success = false;
|
||||
continue;
|
||||
}
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed attach mode: %d",
|
||||
transfer_infos->attach_mode);
|
||||
}
|
||||
// --- Skip 'limits' ---
|
||||
else if (strcmp(key, "limits") == 0) {
|
||||
// limits will be handled in parse_limit
|
||||
|
|
|
|||
|
|
@ -53,6 +53,22 @@ static job_signal_seq_t graceful_seq;
|
|||
static bool graceful_seq_ready = false;
|
||||
static int job_rusage_collect_interval = 10000; // 10 seconds
|
||||
|
||||
/* Attach TTY watchdog: ttyd only starts the job command on the first client
|
||||
* connection, so with no client ever arriving the command never runs, ttyd
|
||||
* never exits and the job would sit in RUN forever holding its ttyd, port
|
||||
* and per-job worker. For attach jobs (exdata "attach":1) the timer below
|
||||
* reaps ttyd via tty_stop_attach_session, which lands us in the regular
|
||||
* process_exit_callback path (JOB_EXITED, status reason, post-exec, exit). */
|
||||
#define TTY_ATTACH_WATCHDOG_MS 1000 /* poll cadence */
|
||||
#define TTY_ATTACH_NO_CLIENT_MS 60000 /* no client ever connected */
|
||||
#define TTY_ATTACH_CHILD_GONE_MS 5000 /* client came, command ended */
|
||||
static uv_timer_t tty_attach_timer;
|
||||
static bool tty_attach_timer_ready = false;
|
||||
static int64_t tty_attach_start_ms = 0;
|
||||
static bool tty_attach_saw_child = false;
|
||||
static int64_t tty_attach_child_gone_ms = 0;
|
||||
static const char* tty_reap_reason = NULL;
|
||||
|
||||
// process manager will share the same event loop with ipc
|
||||
uv_loop_t* uv_loop = NULL;
|
||||
|
||||
|
|
@ -63,6 +79,9 @@ static void xjob_stop_vnc();
|
|||
static void tty_stop_attach_session();
|
||||
static bool check_process_status(pid_t target_pid, int expected_signal,
|
||||
int timeout_seconds);
|
||||
static pid_t ttyd_real_pid(pid_t monitored_pid);
|
||||
static void tty_attach_watchdog_start();
|
||||
static void tty_attach_watchdog_stop();
|
||||
|
||||
static void free_xce_job_rusages(xce_job_rusages_t* rusages)
|
||||
{
|
||||
|
|
@ -477,6 +496,8 @@ static void process_exit_callback(int pid, int exit_status, int term_signal,
|
|||
job_signal_seq_cancel(&graceful_seq);
|
||||
}
|
||||
|
||||
tty_attach_watchdog_stop();
|
||||
|
||||
check_and_reconnect();
|
||||
|
||||
stop_rusage_collection();
|
||||
|
|
@ -507,7 +528,10 @@ static void process_exit_callback(int pid, int exit_status, int term_signal,
|
|||
XCECD_WORKER_LOG(XCE_LOG_INFO, "Child %d killed by signal %d\n", pid,
|
||||
term_signal);
|
||||
|
||||
ipc_worker_send_status(&job_runtime_info, "Killed");
|
||||
/* the watchdog's SIGKILL reaps ttyd through this callback: report
|
||||
* why instead of a bare "Killed" */
|
||||
ipc_worker_send_status(&job_runtime_info,
|
||||
tty_reap_reason ? tty_reap_reason : "Killed");
|
||||
return_code = exit_status;
|
||||
}
|
||||
|
||||
|
|
@ -730,25 +754,258 @@ static void xjob_stop_vnc()
|
|||
* is reaped; tty_stop_attach_session is only called for the attach path. */
|
||||
static void tty_stop_attach_session()
|
||||
{
|
||||
pid_t ttyd_pid = uv_proc_get_pid(&uv_proc_manager);
|
||||
pid_t monitored_pid = uv_proc_get_pid(&uv_proc_manager);
|
||||
|
||||
if (ttyd_pid <= 0) {
|
||||
if (monitored_pid <= 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
char jobidStr[64];
|
||||
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Terminating ttyd %d of attach job %s",
|
||||
ttyd_pid, jobidStr);
|
||||
ttyd_real_pid(monitored_pid), jobidStr);
|
||||
|
||||
signal_children_recursive(SIGKILL, ttyd_pid);
|
||||
if (kill(ttyd_pid, SIGKILL) != 0 && errno != ESRCH) {
|
||||
/* kill from the monitored "/bin/sh -c" down: covers both topologies
|
||||
* (sh exec'd ttyd -- the monitored pid is ttyd; sh forked it -- the
|
||||
* real ttyd hangs off the wrapper) */
|
||||
signal_children_recursive(SIGKILL, monitored_pid);
|
||||
if (kill(monitored_pid, SIGKILL) != 0 && errno != ESRCH) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_WARN,
|
||||
"Failed to kill ttyd %d of job %s: %s", ttyd_pid,
|
||||
"Failed to kill ttyd %d of job %s: %s", monitored_pid,
|
||||
jobidStr, strerror(errno));
|
||||
}
|
||||
}
|
||||
|
||||
/* First direct child of `parent`, from a /proc scan (same read as
|
||||
* signal_children_recursive); 0 when it has none. */
|
||||
static pid_t first_child_of(pid_t parent)
|
||||
{
|
||||
DIR* dir;
|
||||
struct dirent* entry;
|
||||
char path[256];
|
||||
char buf[MAXLINELEN];
|
||||
FILE* stat_file;
|
||||
pid_t child = 0;
|
||||
|
||||
if (parent <= 0 || (dir = opendir("/proc")) == NULL) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
while ((entry = readdir(dir)) != NULL && child == 0) {
|
||||
pid_t pid;
|
||||
|
||||
if (sscanf(entry->d_name, "%d", &pid) != 1)
|
||||
continue;
|
||||
|
||||
if (pid == parent)
|
||||
continue;
|
||||
|
||||
snprintf(path, sizeof(path), "/proc/%d/stat", pid);
|
||||
stat_file = fopen(path, "r");
|
||||
if (!stat_file)
|
||||
continue;
|
||||
|
||||
if (fgets(buf, sizeof(buf), stat_file)) {
|
||||
int parsed_pid;
|
||||
char comm[256];
|
||||
char state;
|
||||
pid_t ppid;
|
||||
/* "pid (comm) state ppid": tolerant of spaces in comm */
|
||||
if (sscanf(buf, "%d (%255[^)]) %c %d", &parsed_pid, comm, &state,
|
||||
&ppid)
|
||||
== 4
|
||||
&& ppid == parent) {
|
||||
child = pid;
|
||||
}
|
||||
}
|
||||
fclose(stat_file);
|
||||
}
|
||||
|
||||
closedir(dir);
|
||||
return child;
|
||||
}
|
||||
|
||||
/* The worker monitors "/bin/sh -c ttyd ...": sh either execs ttyd (the
|
||||
* monitored pid IS ttyd) or forks it (ttyd hangs off the wrapper -- seen on
|
||||
* the field). Resolve the real ttyd pid so the watchdog judges client
|
||||
* activity by the ttyd process's children; against the wrapper, ttyd itself
|
||||
* would count as "the command has started" and the no-client reap would
|
||||
* never fire. */
|
||||
static pid_t ttyd_real_pid(pid_t monitored_pid)
|
||||
{
|
||||
char path[64];
|
||||
char comm[64];
|
||||
FILE* fp;
|
||||
char* line;
|
||||
|
||||
pid_t pid = monitored_pid;
|
||||
|
||||
for (int depth = 0; pid > 0 && depth < 3; depth++) {
|
||||
snprintf(path, sizeof(path), "/proc/%d/comm", (int)pid);
|
||||
fp = fopen(path, "r");
|
||||
if (!fp) {
|
||||
return 0; /* gone or unreadable: ttyd not (yet) visible */
|
||||
}
|
||||
line = fgets(comm, sizeof(comm), fp);
|
||||
fclose(fp);
|
||||
if (!line) {
|
||||
return 0;
|
||||
}
|
||||
comm[strcspn(comm, "\n")] = '\0';
|
||||
|
||||
if (strcmp(comm, "ttyd") == 0) {
|
||||
return pid;
|
||||
}
|
||||
|
||||
pid = first_child_of(pid); /* descend through the sh wrapper */
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* True when the real ttyd has forked its child command: ttyd only does that
|
||||
* once the first websocket client connects, so a child is both proof of a
|
||||
* client and of the command having started. */
|
||||
static bool ttyd_has_child(pid_t ttyd_pid)
|
||||
{
|
||||
DIR* dir;
|
||||
struct dirent* entry;
|
||||
char path[256];
|
||||
char buf[MAXLINELEN];
|
||||
FILE* stat_file;
|
||||
pid_t ppid;
|
||||
|
||||
if (ttyd_pid <= 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if ((dir = opendir("/proc")) == NULL) {
|
||||
/* unreadable /proc must not reap a live session */
|
||||
return true;
|
||||
}
|
||||
|
||||
while ((entry = readdir(dir)) != NULL) {
|
||||
pid_t pid;
|
||||
|
||||
if (sscanf(entry->d_name, "%d", &pid) != 1)
|
||||
continue;
|
||||
|
||||
if (pid == ttyd_pid)
|
||||
continue;
|
||||
|
||||
snprintf(path, sizeof(path), "/proc/%d/stat", pid);
|
||||
stat_file = fopen(path, "r");
|
||||
if (!stat_file)
|
||||
continue;
|
||||
|
||||
if (fgets(buf, sizeof(buf), stat_file)) {
|
||||
int parsed_pid;
|
||||
char comm[256];
|
||||
char state;
|
||||
/* "pid (comm) state ppid": tolerant of spaces in comm */
|
||||
if (sscanf(buf, "%d (%255[^)]) %c %d", &parsed_pid, comm, &state,
|
||||
&ppid)
|
||||
== 4
|
||||
&& ppid == ttyd_pid) {
|
||||
fclose(stat_file);
|
||||
closedir(dir);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
fclose(stat_file);
|
||||
}
|
||||
|
||||
closedir(dir);
|
||||
return false;
|
||||
}
|
||||
|
||||
static void tty_attach_watchdog_callback(uv_timer_t* timer)
|
||||
{
|
||||
(void)timer;
|
||||
|
||||
if (worker_exit_flag || TTY != mydata.transfer_info->interactive_mode
|
||||
|| 1 != mydata.transfer_info->attach_mode) {
|
||||
return;
|
||||
}
|
||||
|
||||
pid_t monitored_pid = uv_proc_get_pid(&uv_proc_manager);
|
||||
if (monitored_pid <= 0) {
|
||||
return; /* process_exit_callback owns the job now */
|
||||
}
|
||||
|
||||
/* descend "/bin/sh -c" (it may have exec'd or forked ttyd) */
|
||||
pid_t ttyd_pid = ttyd_real_pid(monitored_pid);
|
||||
if (ttyd_pid <= 0) {
|
||||
/* ttyd not visible yet (sh still starting) or already gone (the
|
||||
* process-exit path owns the job): hold off, do not reap */
|
||||
return;
|
||||
}
|
||||
|
||||
int64_t now_ms = uv_now(uv_loop);
|
||||
|
||||
if (ttyd_has_child(ttyd_pid)) {
|
||||
if (!tty_attach_saw_child) {
|
||||
tty_attach_saw_child = true;
|
||||
char jobidStr[64];
|
||||
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
|
||||
"Attach job %s: command started (client "
|
||||
"connected)",
|
||||
jobidStr);
|
||||
}
|
||||
tty_attach_child_gone_ms = 0;
|
||||
return;
|
||||
}
|
||||
|
||||
char jobidStr[64];
|
||||
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||
|
||||
if (tty_attach_saw_child) {
|
||||
/* the command ran and is gone; give ttyd a moment to exit on its
|
||||
* own (-Q), then reap -- its teardown cannot be relied on with a
|
||||
* relay client still connected */
|
||||
if (tty_attach_child_gone_ms == 0) {
|
||||
tty_attach_child_gone_ms = now_ms;
|
||||
} else if (now_ms - tty_attach_child_gone_ms
|
||||
>= TTY_ATTACH_CHILD_GONE_MS) {
|
||||
XCECD_WORKER_LOG(
|
||||
XCE_LOG_NOTICE,
|
||||
"Attach job %s: command exited but ttyd %d lingers, "
|
||||
"reaping it",
|
||||
jobidStr, ttyd_pid);
|
||||
tty_reap_reason = "Command exited; ttyd did not stop on its own";
|
||||
tty_stop_attach_session();
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (now_ms - tty_attach_start_ms >= TTY_ATTACH_NO_CLIENT_MS) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_NOTICE,
|
||||
"Attach job %s: no client attached within %d s, "
|
||||
"reaping ttyd %d",
|
||||
jobidStr, (int)(TTY_ATTACH_NO_CLIENT_MS / 1000),
|
||||
ttyd_pid);
|
||||
tty_reap_reason = "No client attached within 60 s";
|
||||
tty_stop_attach_session();
|
||||
}
|
||||
}
|
||||
|
||||
static void tty_attach_watchdog_start()
|
||||
{
|
||||
tty_attach_start_ms = uv_now(uv_loop);
|
||||
tty_attach_saw_child = false;
|
||||
tty_attach_child_gone_ms = 0;
|
||||
uv_timer_start(&tty_attach_timer, tty_attach_watchdog_callback,
|
||||
TTY_ATTACH_WATCHDOG_MS, TTY_ATTACH_WATCHDOG_MS);
|
||||
}
|
||||
|
||||
static void tty_attach_watchdog_stop()
|
||||
{
|
||||
if (tty_attach_timer_ready) {
|
||||
uv_timer_stop(&tty_attach_timer);
|
||||
}
|
||||
}
|
||||
|
||||
static void worker_job_callback(job_assignment_t* info)
|
||||
{
|
||||
char* job_cmd = NULL;
|
||||
|
|
@ -938,6 +1195,15 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
start_rusage_collection();
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Starting job rusage collection...");
|
||||
|
||||
tty_reap_reason = NULL;
|
||||
|
||||
/* attach TTY jobs: reap ttyd if no client ever connects (the job
|
||||
* command would otherwise never start -- ttyd is on-demand) */
|
||||
if (TTY == mydata.transfer_info->interactive_mode
|
||||
&& 1 == mydata.transfer_info->attach_mode) {
|
||||
tty_attach_watchdog_start();
|
||||
}
|
||||
|
||||
if (mydata.transfer_info->interactive_mode > 0) {
|
||||
url = generate_url(port_info,
|
||||
mydata.transfer_info->interactive_mode);
|
||||
|
|
@ -1611,6 +1877,10 @@ int main(int argc, char* argv[])
|
|||
/* bug-1957: timer for graceful kill/stop sequences */
|
||||
graceful_seq_ready = (job_signal_seq_init(&graceful_seq, uv_loop) == 0);
|
||||
|
||||
/* attach TTY watchdog (tty_attach_watchdog_callback) */
|
||||
uv_timer_init(uv_loop, &tty_attach_timer);
|
||||
tty_attach_timer_ready = true;
|
||||
|
||||
do {
|
||||
|
||||
ipc_worker_run(UV_RUN_NOWAIT);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue