diff --git a/runner/internal/runner/executor/executor.go b/runner/internal/runner/executor/executor.go index d78bf629fb..832cd226c9 100644 --- a/runner/internal/runner/executor/executor.go +++ b/runner/internal/runner/executor/executor.go @@ -43,6 +43,11 @@ const ( // Maximum buffer size for ansistrip MaxBufferSize = 32 * 1024 // 32KB + + // intrChar is the terminal's INTR character (Ctrl-C) in the default configuration. + intrChar = 0x03 + // intrWriteTimeout bounds how long writing INTR to the pty master may block. + intrWriteTimeout = 5 * time.Second ) type ConnectionTracker interface { @@ -505,13 +510,6 @@ func (ex *RunExecutor) execJob(ctx context.Context, jobLogFile io.Writer) error } cmd := exec.CommandContext(ctx, ex.jobSpec.Commands[0], ex.jobSpec.Commands[1:]...) - cmd.Cancel = func() error { - // returns error on Windows - if signalErr := cmd.Process.Signal(os.Interrupt); signalErr != nil { - return fmt.Errorf("send interrupt signal: %w", signalErr) - } - return nil - } cmd.WaitDelay = ex.killDelay // kills the process if it doesn't exit in time if err := utils.MkdirAll(ctx, ex.jobWorkingDir, ex.jobUser.Uid, ex.jobUser.Gid, 0o755); err != nil { @@ -802,6 +800,9 @@ func startCommand(cmd *exec.Cmd) (*os.File, error) { } } + // Cancel must be set before Start, which installs the goroutine that calls it. + cmd.Cancel = func() error { return interruptJob(ptm) } + if err := cmd.Start(); err != nil { _ = ptm.Close() return nil, fmt.Errorf("start command: %w", err) @@ -809,6 +810,40 @@ func startCommand(cmd *exec.Cmd) (*os.File, error) { return ptm, nil } +// interruptJob asks the job to stop the way Ctrl-C does. Writing the terminal's INTR character +// to the pty master makes the line discipline raise SIGINT in the terminal's foreground process +// group -- the command the shell is currently running, together with everything sharing its +// process group. +// +// Signalling cmd.Process reaches the wrong process instead. The server runs commands under +// `sh -i -c`, and an interactive shell turns on job control, which puts the job in a process +// group of its own, while the shell ignores SIGINT for as long as it is waiting for that job. +// The signal reached neither, so nothing stopped the job until WaitDelay expired and SIGKILL +// went to the shell alone. +// +// The job may still ignore this: a program that puts the terminal in raw mode clears ISIG, and +// the INTR character then delivers no signal at all. WaitDelay stays the backstop. +func interruptJob(ptm *os.File) error { + // The master is pollable (see openPty), so a deadline is honoured here. Without one, a job + // that never reads its stdin could fill the terminal's input buffer and block this write + // indefinitely -- and Cmd only starts the WaitDelay timer once Cancel has returned. + if err := ptm.SetWriteDeadline(time.Now().Add(intrWriteTimeout)); err != nil { + return fmt.Errorf("set INTR write deadline: %w", err) + } + defer func() { _ = ptm.SetWriteDeadline(time.Time{}) }() + + if _, err := ptm.Write([]byte{intrChar}); err != nil { + if isPtyError(err) || errors.Is(err, os.ErrClosed) { + // The terminal is gone, so the job is gone with it. Reporting the process as + // already done keeps Wait returning the command's own exit status rather than + // replacing it with the context error. + return fmt.Errorf("write INTR: %w", errors.Join(err, os.ErrProcessDone)) + } + return fmt.Errorf("write INTR: %w", err) + } + return nil +} + func prepareUserSshDir(user *linuxuser.User) (string, error) { sshDir := filepath.Join(user.HomeDir, ".ssh") info, err := os.Stat(sshDir) diff --git a/runner/internal/runner/executor/executor_test.go b/runner/internal/runner/executor/executor_test.go index acbf49381a..85851343eb 100644 --- a/runner/internal/runner/executor/executor_test.go +++ b/runner/internal/runner/executor/executor_test.go @@ -16,6 +16,7 @@ import ( "testing" "time" + "github.com/dstackai/dstack/runner/internal/common/types" linuxuser "github.com/dstackai/dstack/runner/internal/runner/linux/user" "github.com/dstackai/dstack/runner/internal/runner/schemas" "github.com/stretchr/testify/assert" @@ -140,7 +141,14 @@ func TestExecutor_MaxDuration(t *testing.T) { makeCodeTar(t, ex) err := ex.Run(t.Context()) - assert.ErrorContains(t, err, "killed") + // The job is interrupted rather than killed: INTR reaches the workload through the + // terminal, so it exits on SIGINT long before the SIGKILL backstop would fire. + assert.ErrorContains(t, err, "interrupt") + + history := ex.GetHistory(0) + lastState := history.JobStates[len(history.JobStates)-1] + assert.Equal(t, schemas.JobStateTerminated, lastState.State) + assert.Equal(t, types.TerminationReasonMaxDurationExceeded, lastState.TerminationReason) } func TestExecutor_LogQuota(t *testing.T) { @@ -207,6 +215,64 @@ func TestExecutor_SurvivingProcessDoesNotHangRun(t *testing.T) { assert.Contains(t, logs.String(), "done") } +// Stopping a job must interrupt the workload, not just the wrapper shell: the workload gets a +// chance to shut down cleanly instead of being killed once the grace period expires. +func TestExecutor_StopInterruptsWorkload(t *testing.T) { + if testing.Short() { + t.Skip() + } + + workload := filepath.Join(t.TempDir(), "workload.sh") + require.NoError(t, os.WriteFile(workload, []byte( + "trap 'echo graceful shutdown; exit 0' INT\n"+ + "echo ready\n"+ + "sleep 300\n"), 0o600)) + + ex := makeTestExecutor(t) + // The SIGKILL backstop must not be what stops the job. + ex.killDelay = 60 * time.Second + // The trailing `&& :` is load-bearing. Given a single simple command, bash and BusyBox ash + // exec it in place, leaving no shell at all -- and the interrupt used to reach a workload + // that was the direct child just fine. A command list keeps the wrapper shell, which is the + // arrangement that used to swallow the interrupt. Verified against bash, dash and BusyBox + // ash; dash does not do the optimization either way. + ex.jobSpec.Commands = []string{"/bin/sh", "-i", "-c", "/bin/sh " + workload + " && :"} + makeCodeTar(t, ex) + + ctx, cancel := context.WithCancel(t.Context()) + runDone := make(chan error, 1) + go func() { runDone <- ex.Run(ctx) }() + + require.Eventually(t, func() bool { + return strings.Contains(jobLogsSoFar(ex), "ready") + }, 30*time.Second, 100*time.Millisecond, "the workload never started") + + cancel() // what /api/stop does + stoppedAt := time.Now() + + select { + case <-runDone: + case <-time.After(30 * time.Second): + t.Fatal("Run did not return after the job was stopped") + } + assert.Less(t, time.Since(stoppedAt), ex.killDelay, + "the job was stopped by the SIGKILL backstop rather than by INTR") + + history := ex.GetHistory(0) + assert.Contains(t, combineLogMessages(history.JobLogs), "graceful shutdown", + "the workload did not receive SIGINT") + lastState := history.JobStates[len(history.JobStates)-1] + assert.Equal(t, schemas.JobStateTerminated, lastState.State) +} + +// jobLogsSoFar reads the log history while the job is still running, under the lock the +// executor's writers share. +func jobLogsSoFar(ex *RunExecutor) string { + ex.mu.RLock() + defer ex.mu.RUnlock() + return combineLogMessages(ex.jobLogs.history) +} + func TestExecutor_RemoteRepo(t *testing.T) { if testing.Short() { t.Skip()