mirror of
https://github.com/containerd/containerd.git
synced 2026-11-06 05:17:02 +00:00
runc-shim: don't block State on in-flight Delete
Init.Delete takes p.mu and holds it across initState.Delete, whose delete() shuts the terminal console down and waits up to 10s for stdio to drain, execs `runc delete`, closes the process IO and console and unmounts the rootfs. Init.Status, ExitStatus and ExitedAt take the same mutex, so the shim's task.State RPC, which calls all three, queues behind a Delete. containerd gives State a 2s deadline (io.containerd.timeout.task.state), so the call fails. Depending on the containerd version that is logged as "get state for <id>: context deadline exceeded" and the task reported as UNKNOWN, or the error is propagated, making Get fail and List drop the task. execProcess has the identical shape with e.mu. This is easy to hit when the unmount is slow. On a node with tens of GB of dirty page cache, umount of the overlayfs rootfs calls sync_filesystem() on the upper filesystem and was measured at 17.6s with 12.9 GB dirty (0.22s with 1.0 GB). That node logged 4,661 State timeouts across 284 containers over a window of heavy writes. This change lets State answer while Delete runs. Delete now releases mu for the slow cleanup, and every method that is not a plain reader still waits for the cleanup to end: - Init and execProcess get a `deleting` channel, read and written only with mu held. delete() is entered and left with mu held; it sets `deleting`, unlocks, runs the old body unchanged, and a deferred relock clears `deleting` and closes the channel, whether the cleanup succeeded, failed or panicked. - A new lock() takes mu and, while `deleting` is set, unlocks, waits for the channel and locks again. Every method that changes the process (Init and exec alike) now calls lock() instead of mu.Lock(), so the cleanup is still exclusive with all of them, exactly as when Delete held mu throughout. They then run against deletedState if Delete succeeded, or the previous state if it failed, which is what they saw before. - Status, ExitStatus and ExitedAt keep taking mu directly and so no longer wait for the cleanup. They report the state from before the Delete, since nothing is changed until it ends: a created process stays created with a zero exit status and exit time, and a stopped one keeps its real exit status and exit time. There is no created -> stopped -> created flicker if the Delete fails. The state machine (init_state.go, exec_state.go, deleted_state.go) is unchanged, and there is no second lock: readers still take mu, so Status during an in-flight `runc start` still waits and returns "running", which CRI recovery relies on (a task seen as created is killed). Behavior change: State/Status/ExitStatus/ExitedAt no longer wait for the cleanup part of Delete and report the state from before the Delete. If the Delete fails nothing changes, and an exit that arrived meanwhile is applied afterwards, as it was on main. One consequence: while the Delete of a created task with a non-zero pid runs, a second plain task.Delete from the client (no WithProcessKill) sees "created" and fails at once with "task must be stopped before deletion", where it used to wait for the first Delete, see "stopped" and get NotFound. Everything else is unchanged: other calls wait for Delete as before, a second Delete waits and gets NotFound, exit events wait, and the task service still removes the process only after Delete returns. Because mu is now released inside delete(), any new method that changes a process must take mu through lock(); this is documented on mu. This does not make Delete faster. It still takes as long as before, and the dirty page cache that makes umount slow is a separate problem. Timeouts are unchanged. Tests: Delete is blocked inside an io.Closer; State must return and report the pre-Delete state, mutating methods must wait, and readers must take mu. Assisted-by: Claude Code (Claude Sonnet 5.5) Signed-off-by: Md Tahsin Rahman <tahsin8243@gmail.com>
This commit is contained in:
1 parent
ae65337745
commit
a71f15225d
4 files changed
+487
-19
No files matched your search
@@ -43,7 +43,13 @@ type execProcess struct {
|
||||
|
||||
execState execState
|
||||
|
||||
mu sync.Mutex
|
||||
// mu is released for most of delete; see Init.mu. Methods other than
|
||||
// read-only accessors MUST take it through lock().
|
||||
mu sync.Mutex
|
||||
// deleting is non-nil while delete runs with mu released, and is closed
|
||||
// when delete ends. It is only accessed with mu held.
|
||||
deleting chan struct{}
|
||||
|
||||
id string
|
||||
console console.Console
|
||||
io *processIO
|
||||
@@ -90,7 +96,7 @@ func (e *execProcess) ExitedAt() time.Time {
|
||||
}
|
||||
|
||||
func (e *execProcess) SetExited(status int) {
|
||||
e.mu.Lock()
|
||||
e.lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
e.execState.SetExited(status)
|
||||
@@ -104,13 +110,36 @@ func (e *execProcess) setExited(status int) {
|
||||
}
|
||||
|
||||
func (e *execProcess) Delete(ctx context.Context) error {
|
||||
e.mu.Lock()
|
||||
e.lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
return e.execState.Delete(ctx)
|
||||
}
|
||||
|
||||
// lock acquires mu, waiting first for a delete that has released it.
|
||||
// It returns with mu held and deleting nil. Release with e.mu.Unlock.
|
||||
func (e *execProcess) lock() {
|
||||
e.mu.Lock()
|
||||
for e.deleting != nil {
|
||||
d := e.deleting
|
||||
e.mu.Unlock()
|
||||
<-d
|
||||
e.mu.Lock()
|
||||
}
|
||||
}
|
||||
|
||||
// delete is called with mu held, through lock, and returns with mu held and
|
||||
// deleting cleared. It releases mu for the rest of its body, which can take
|
||||
// up to 10s draining stdio.
|
||||
func (e *execProcess) delete(ctx context.Context) error {
|
||||
done := make(chan struct{})
|
||||
e.deleting = done
|
||||
e.mu.Unlock()
|
||||
defer func() {
|
||||
e.mu.Lock()
|
||||
e.deleting = nil
|
||||
close(done)
|
||||
}()
|
||||
shutdownAndDrain(ctx, e.parent.Platform, e.console, &e.wg, "exec process "+e.id)
|
||||
closeIO(e.closers, e.console, e.io)
|
||||
pidfile := filepath.Join(e.path, e.id+".pid")
|
||||
@@ -120,7 +149,7 @@ func (e *execProcess) delete(ctx context.Context) error {
|
||||
}
|
||||
|
||||
func (e *execProcess) Resize(ws console.WinSize) error {
|
||||
e.mu.Lock()
|
||||
e.lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
return e.execState.Resize(ws)
|
||||
@@ -134,7 +163,7 @@ func (e *execProcess) resize(ws console.WinSize) error {
|
||||
}
|
||||
|
||||
func (e *execProcess) Kill(ctx context.Context, sig uint32, _ bool) error {
|
||||
e.mu.Lock()
|
||||
e.lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
return e.execState.Kill(ctx, sig, false)
|
||||
@@ -164,7 +193,7 @@ func (e *execProcess) Stdio() stdio.Stdio {
|
||||
}
|
||||
|
||||
func (e *execProcess) Start(ctx context.Context) error {
|
||||
e.mu.Lock()
|
||||
e.lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
return e.execState.Start(ctx)
|
||||
|
||||
@@ -61,7 +61,10 @@ func TestExecDeleteReleasesTerminalIO(t *testing.T) {
|
||||
closers: []io.Closer{stdin},
|
||||
console: cons,
|
||||
}
|
||||
if err := e.delete(context.Background()); err != nil {
|
||||
e.mu.Lock()
|
||||
err := e.delete(context.Background())
|
||||
e.mu.Unlock()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if stdin.closed != 1 {
|
||||
@@ -103,7 +106,10 @@ func TestExecDeleteShutsConsoleDownBeforeClosing(t *testing.T) {
|
||||
parent: &Init{Platform: recordingPlatform{events: &events}},
|
||||
console: orderedConsole{events: &events},
|
||||
}
|
||||
if err := e.delete(context.Background()); err != nil {
|
||||
e.mu.Lock()
|
||||
err := e.delete(context.Background())
|
||||
e.mu.Unlock()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := strings.Join(events, ","); got != "shutdown,close" {
|
||||
|
||||
@@ -50,8 +50,19 @@ type Init struct {
|
||||
// the right order when invoked in separate goroutines.
|
||||
// This is the case within the shim implementation as it makes use of
|
||||
// the reaper interface.
|
||||
//
|
||||
// mu is intentionally released for most of delete, which is slow, so
|
||||
// that Status, ExitStatus and ExitedAt, which take mu directly, do not
|
||||
// wait for it. Every method that changes the process, or that must not
|
||||
// run concurrently with a Delete, MUST take mu through lock(), which
|
||||
// waits for delete to end. Taking mu directly is only for read-only
|
||||
// accessors.
|
||||
mu sync.Mutex
|
||||
|
||||
// deleting is non-nil while delete runs with mu released, and is closed
|
||||
// when delete ends. It is only accessed with mu held.
|
||||
deleting chan struct{}
|
||||
|
||||
waitBlock chan struct{}
|
||||
|
||||
WorkDir string
|
||||
@@ -263,7 +274,7 @@ func (p *Init) Status(ctx context.Context) (string, error) {
|
||||
|
||||
// Start the init process
|
||||
func (p *Init) Start(ctx context.Context) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
return p.initState.Start(ctx)
|
||||
@@ -276,7 +287,7 @@ func (p *Init) start(ctx context.Context) error {
|
||||
|
||||
// SetExited of the init process with the next status
|
||||
func (p *Init) SetExited(status int) {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
p.initState.SetExited(status)
|
||||
@@ -291,13 +302,37 @@ func (p *Init) setExited(status int) {
|
||||
|
||||
// Delete the init process
|
||||
func (p *Init) Delete(ctx context.Context) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
return p.initState.Delete(ctx)
|
||||
}
|
||||
|
||||
// lock acquires mu, waiting first for a delete that has released it.
|
||||
// It returns with mu held and deleting nil. Release with p.mu.Unlock.
|
||||
func (p *Init) lock() {
|
||||
p.mu.Lock()
|
||||
for p.deleting != nil {
|
||||
d := p.deleting
|
||||
p.mu.Unlock()
|
||||
<-d
|
||||
p.mu.Lock()
|
||||
}
|
||||
}
|
||||
|
||||
// delete is called with mu held, through lock, and returns with mu held and
|
||||
// deleting cleared. It releases mu for the rest of its body, which can take
|
||||
// tens of seconds (stdio drain, runc delete, rootfs umount). Methods that take
|
||||
// mu through lock wait for it, so it is still exclusive with all of them.
|
||||
func (p *Init) delete(ctx context.Context) error {
|
||||
done := make(chan struct{})
|
||||
p.deleting = done
|
||||
p.mu.Unlock()
|
||||
defer func() {
|
||||
p.mu.Lock()
|
||||
p.deleting = nil
|
||||
close(done)
|
||||
}()
|
||||
shutdownAndDrain(ctx, p.Platform, p.console, &p.wg, "init process "+p.id)
|
||||
err := p.runtime.Delete(ctx, p.id, nil)
|
||||
// ignore errors if a runtime has already deleted the process
|
||||
@@ -324,7 +359,7 @@ func (p *Init) delete(ctx context.Context) error {
|
||||
|
||||
// Resize the init processes console
|
||||
func (p *Init) Resize(ws console.WinSize) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
if p.console == nil {
|
||||
@@ -335,7 +370,7 @@ func (p *Init) Resize(ws console.WinSize) error {
|
||||
|
||||
// Pause the init process and all its child processes
|
||||
func (p *Init) Pause(ctx context.Context) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
return p.initState.Pause(ctx)
|
||||
@@ -343,7 +378,7 @@ func (p *Init) Pause(ctx context.Context) error {
|
||||
|
||||
// Resume the init process and all its child processes
|
||||
func (p *Init) Resume(ctx context.Context) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
return p.initState.Resume(ctx)
|
||||
@@ -351,7 +386,7 @@ func (p *Init) Resume(ctx context.Context) error {
|
||||
|
||||
// Kill the init process
|
||||
func (p *Init) Kill(ctx context.Context, signal uint32, all bool) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
return p.initState.Kill(ctx, signal, all)
|
||||
@@ -366,7 +401,7 @@ func (p *Init) kill(ctx context.Context, signal uint32, all bool) error {
|
||||
|
||||
// KillAll processes belonging to the init process
|
||||
func (p *Init) KillAll(ctx context.Context) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
err := p.runtime.Kill(ctx, p.id, int(unix.SIGKILL), &runc.KillOpts{
|
||||
@@ -387,7 +422,7 @@ func (p *Init) Runtime() *runc.Runc {
|
||||
|
||||
// Exec returns a new child process
|
||||
func (p *Init) Exec(ctx context.Context, path string, r *ExecConfig) (Process, error) {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
return p.initState.Exec(ctx, path, r)
|
||||
@@ -421,7 +456,7 @@ func (p *Init) exec(ctx context.Context, path string, r *ExecConfig) (Process, e
|
||||
|
||||
// Checkpoint the init process
|
||||
func (p *Init) Checkpoint(ctx context.Context, r *CheckpointConfig) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
return p.initState.Checkpoint(ctx, r)
|
||||
@@ -459,7 +494,7 @@ func (p *Init) checkpoint(ctx context.Context, r *CheckpointConfig) error {
|
||||
|
||||
// Update the processes resource configuration
|
||||
func (p *Init) Update(ctx context.Context, r *google_protobuf.Any) error {
|
||||
p.mu.Lock()
|
||||
p.lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
return p.initState.Update(ctx, r)
|
||||
|
||||
@@ -0,0 +1,398 @@
|
||||
//go:build !windows
|
||||
|
||||
/*
|
||||
Copyright The containerd Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"os/exec"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/containerd/console"
|
||||
google_protobuf "github.com/containerd/containerd/v2/pkg/protobuf/types"
|
||||
"github.com/containerd/containerd/v2/pkg/stdio"
|
||||
"github.com/containerd/errdefs"
|
||||
runc "github.com/containerd/go-runc"
|
||||
)
|
||||
|
||||
// failTimeout bounds waits for things that must happen, so only a failure
|
||||
// ever waits that long. settle is how long something that must NOT happen yet
|
||||
// is observed; a correct implementation cannot fail because of it.
|
||||
const (
|
||||
failTimeout = 30 * time.Second
|
||||
settle = 50 * time.Millisecond
|
||||
)
|
||||
|
||||
// gate blocks whoever calls Close or block until released, and tells the test
|
||||
// when it has been entered. It is joined with the goroutines started through
|
||||
// it when the test ends, after being released, so none outlives the test.
|
||||
type gate struct {
|
||||
entered chan struct{}
|
||||
unblock chan struct{}
|
||||
enter func()
|
||||
release func()
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
func newGate(t *testing.T) *gate {
|
||||
t.Helper()
|
||||
g := &gate{entered: make(chan struct{}), unblock: make(chan struct{})}
|
||||
g.enter = sync.OnceFunc(func() { close(g.entered) })
|
||||
g.release = sync.OnceFunc(func() { close(g.unblock) })
|
||||
t.Cleanup(func() {
|
||||
g.release()
|
||||
g.wg.Wait()
|
||||
})
|
||||
return g
|
||||
}
|
||||
|
||||
// Close is called by Delete in its cleanup, when g is in a closers list.
|
||||
func (g *gate) Close() error {
|
||||
g.enter()
|
||||
<-g.unblock
|
||||
return nil
|
||||
}
|
||||
|
||||
// run calls fn in a goroutine and returns a channel closed when it returns.
|
||||
func (g *gate) run(fn func()) <-chan struct{} {
|
||||
done := make(chan struct{})
|
||||
g.wg.Go(func() {
|
||||
defer close(done)
|
||||
fn()
|
||||
})
|
||||
return done
|
||||
}
|
||||
|
||||
func finishes(done <-chan struct{}, d time.Duration) bool {
|
||||
select {
|
||||
case <-done:
|
||||
return true
|
||||
case <-time.After(d):
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func mustFinish(t *testing.T, done <-chan struct{}, what string) {
|
||||
t.Helper()
|
||||
if !finishes(done, failTimeout) {
|
||||
t.Fatalf("timed out waiting for %s", what)
|
||||
}
|
||||
}
|
||||
|
||||
// fakeRuntime returns an OCI runtime on which every command succeeds ("true")
|
||||
// or fails ("false").
|
||||
func fakeRuntime(t *testing.T, command string) *runc.Runc {
|
||||
t.Helper()
|
||||
path, err := exec.LookPath(command)
|
||||
if err != nil {
|
||||
t.Skipf("%s not found: %v", command, err)
|
||||
}
|
||||
return &runc.Runc{Command: path}
|
||||
}
|
||||
|
||||
func newInit(t *testing.T, command string) *Init {
|
||||
t.Helper()
|
||||
p := New("test", fakeRuntime(t, command), stdio.Stdio{})
|
||||
p.Platform = recordingPlatform{events: new([]string)}
|
||||
p.io = &processIO{}
|
||||
return p
|
||||
}
|
||||
|
||||
func newExec(t *testing.T) *execProcess {
|
||||
t.Helper()
|
||||
e := &execProcess{id: "exec", path: t.TempDir(), parent: newInit(t, "true"), waitBlock: make(chan struct{}), io: &processIO{}}
|
||||
e.execState = &execCreatedState{p: e}
|
||||
return e
|
||||
}
|
||||
|
||||
// blockedDelete starts p.Delete and returns once it is blocked in g's Close,
|
||||
// i.e. in its cleanup. finish lets it proceed and returns its result.
|
||||
func blockedDelete(t *testing.T, g *gate, p Process) (finish func() error) {
|
||||
t.Helper()
|
||||
var err error
|
||||
done := g.run(func() { err = p.Delete(t.Context()) })
|
||||
select {
|
||||
case <-g.entered:
|
||||
case <-done:
|
||||
t.Fatalf("Delete returned without reaching its cleanup: %v", err)
|
||||
case <-time.After(failTimeout):
|
||||
t.Fatal("timed out waiting for Delete to reach its cleanup")
|
||||
}
|
||||
return func() error {
|
||||
g.release()
|
||||
mustFinish(t, done, "Delete")
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// state is what State reports for a process.
|
||||
type state struct {
|
||||
status string
|
||||
exit int
|
||||
at time.Time
|
||||
}
|
||||
|
||||
func (s state) equal(o state) bool {
|
||||
return s.status == o.status && s.exit == o.exit && s.at.Equal(o.at)
|
||||
}
|
||||
|
||||
func observe(t *testing.T, p Process) state {
|
||||
t.Helper()
|
||||
s, err := p.Status(t.Context())
|
||||
if err != nil {
|
||||
t.Errorf("Status: %v", err)
|
||||
}
|
||||
return state{s, p.ExitStatus(), p.ExitedAt()}
|
||||
}
|
||||
|
||||
// testStateDuringDelete checks that, while the Delete of p is in its cleanup,
|
||||
// State returns and reports what it did before the Delete; that, if the Delete
|
||||
// fails, nothing has changed after it; and that after a successful one the
|
||||
// process is stopped.
|
||||
func testStateDuringDelete(t *testing.T, g *gate, p Process, deleteFails bool) {
|
||||
t.Helper()
|
||||
want := observe(t, p)
|
||||
finish := blockedDelete(t, g, p)
|
||||
var during state
|
||||
mustFinish(t, g.run(func() { during = observe(t, p) }), "State during Delete")
|
||||
if !during.equal(want) {
|
||||
t.Errorf("State during Delete = %+v; want %+v", during, want)
|
||||
}
|
||||
err := finish()
|
||||
if (err != nil) != deleteFails {
|
||||
t.Errorf("Delete: %v; want error: %v", err, deleteFails)
|
||||
}
|
||||
after := observe(t, p)
|
||||
if deleteFails && !after.equal(want) {
|
||||
t.Errorf("State after failed Delete = %+v; want %+v", after, want)
|
||||
}
|
||||
if !deleteFails && !after.equal(state{"stopped", want.exit, want.at}) {
|
||||
t.Errorf("State after Delete = %+v; want stopped with %d, %v", after, want.exit, want.at)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStateDuringDeleteOfCreatedInit(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, runtime string
|
||||
fails bool
|
||||
}{{"DeleteFails", "false", true}, {"DeleteSucceeds", "true", false}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
g, p := newGate(t), newInit(t, tc.runtime)
|
||||
p.closers = []io.Closer{g}
|
||||
if s := observe(t, p); !s.equal(state{"created", 0, time.Time{}}) {
|
||||
t.Fatalf("State = %+v; want created", s)
|
||||
}
|
||||
testStateDuringDelete(t, g, p, tc.fails)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStateDuringDeleteOfCreatedExec(t *testing.T) {
|
||||
g, e := newGate(t), newExec(t)
|
||||
e.closers = []io.Closer{g}
|
||||
testStateDuringDelete(t, g, e, false)
|
||||
}
|
||||
|
||||
// A stopped process keeps its real exit status and time during the cleanup.
|
||||
func TestStateDuringDeleteOfStoppedInit(t *testing.T) {
|
||||
g, p := newGate(t), newInit(t, "true")
|
||||
p.closers = []io.Closer{g}
|
||||
p.SetExited(137)
|
||||
if s := observe(t, p); s.status != "stopped" || s.exit != 137 || s.at.IsZero() {
|
||||
t.Fatalf("State = %+v; want stopped with exit 137 and a time", s)
|
||||
}
|
||||
testStateDuringDelete(t, g, p, false)
|
||||
}
|
||||
|
||||
func TestStateDuringDeleteOfStoppedExec(t *testing.T) {
|
||||
g, e := newGate(t), newExec(t)
|
||||
e.closers = []io.Closer{g}
|
||||
e.SetExited(3)
|
||||
if s := observe(t, e); s.status != "stopped" || s.exit != 3 || s.at.IsZero() {
|
||||
t.Fatalf("State = %+v; want stopped with exit 3 and a time", s)
|
||||
}
|
||||
testStateDuringDelete(t, g, e, false)
|
||||
}
|
||||
|
||||
// An exit that arrives during a Delete that then fails waits for the cleanup
|
||||
// and is applied to the process afterwards.
|
||||
func TestExitDuringFailedDelete(t *testing.T) {
|
||||
g, p := newGate(t), newInit(t, "false")
|
||||
p.closers = []io.Closer{g}
|
||||
finish := blockedDelete(t, g, p)
|
||||
exited := g.run(func() { p.SetExited(9) })
|
||||
if finishes(exited, settle) {
|
||||
t.Error("SetExited did not wait for Delete")
|
||||
}
|
||||
if err := finish(); err == nil {
|
||||
t.Error("Delete succeeded with a failing runtime")
|
||||
}
|
||||
mustFinish(t, exited, "SetExited")
|
||||
mustFinish(t, g.run(func() { p.Wait(t.Context()) }), "Wait")
|
||||
if s := observe(t, p); s.status != "stopped" || s.exit != 9 {
|
||||
t.Errorf("State = %+v; want stopped with exit 9", s)
|
||||
}
|
||||
}
|
||||
|
||||
// startStub is the state of an Init whose Start blocks in g and then succeeds.
|
||||
type startStub struct {
|
||||
initState
|
||||
g *gate
|
||||
}
|
||||
|
||||
func (s startStub) Start(context.Context) error {
|
||||
s.g.enter()
|
||||
<-s.g.unblock
|
||||
return nil
|
||||
}
|
||||
|
||||
func (startStub) Status(context.Context) (string, error) { return "running", nil }
|
||||
|
||||
// Status waits for an in-flight Start and so reports "running", not
|
||||
// "created"; CRI recovery kills tasks it sees as created.
|
||||
func TestStatusWaitsForStart(t *testing.T) {
|
||||
g, p := newGate(t), newInit(t, "true")
|
||||
p.initState = startStub{p.initState, g}
|
||||
var startErr error
|
||||
started := g.run(func() { startErr = p.Start(t.Context()) })
|
||||
select {
|
||||
case <-g.entered:
|
||||
case <-started:
|
||||
t.Fatalf("Start returned before blocking: %v", startErr)
|
||||
case <-time.After(failTimeout):
|
||||
t.Fatal("timed out waiting for Start")
|
||||
}
|
||||
var status string
|
||||
got := g.run(func() { status, _ = p.Status(t.Context()) })
|
||||
if finishes(got, settle) {
|
||||
t.Errorf("Status returned %q during Start", status)
|
||||
}
|
||||
g.release()
|
||||
mustFinish(t, started, "Start")
|
||||
mustFinish(t, got, "Status")
|
||||
if status != "running" || startErr != nil {
|
||||
t.Errorf("Status = %q, Start: %v; want \"running\", nil", status, startErr)
|
||||
}
|
||||
}
|
||||
|
||||
type operation struct {
|
||||
name string
|
||||
call func() error
|
||||
notFound bool // the deleted process reports it gone
|
||||
}
|
||||
|
||||
// While the cleanup runs, every method that changes a process waits for it,
|
||||
// then runs against the deleted process.
|
||||
func testOperationsWait(t *testing.T, g *gate, p Process, ops []operation) {
|
||||
t.Helper()
|
||||
finish := blockedDelete(t, g, p)
|
||||
errs := make([]error, len(ops))
|
||||
done := make([]<-chan struct{}, len(ops))
|
||||
for i, op := range ops {
|
||||
done[i] = g.run(func() { errs[i] = op.call() })
|
||||
}
|
||||
time.Sleep(settle)
|
||||
for i, op := range ops {
|
||||
select {
|
||||
case <-done[i]:
|
||||
t.Errorf("%s did not wait for Delete", op.name)
|
||||
default:
|
||||
}
|
||||
}
|
||||
if err := finish(); err != nil {
|
||||
t.Errorf("Delete: %v", err)
|
||||
}
|
||||
for i, op := range ops {
|
||||
mustFinish(t, done[i], op.name)
|
||||
if op.notFound && !errdefs.IsNotFound(errs[i]) {
|
||||
t.Errorf("%s: %v; want not found", op.name, errs[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestInitOperationsWaitForDelete(t *testing.T) {
|
||||
g, p := newGate(t), newInit(t, "true")
|
||||
p.closers = []io.Closer{g}
|
||||
ctx := t.Context()
|
||||
testOperationsWait(t, g, p, []operation{
|
||||
{"Start", func() error { return p.Start(ctx) }, false},
|
||||
{"SetExited", func() error { p.SetExited(1); return nil }, false},
|
||||
{"Delete", func() error { return p.Delete(ctx) }, true},
|
||||
{"Resize", func() error { return p.Resize(console.WinSize{}) }, false},
|
||||
{"Pause", func() error { return p.Pause(ctx) }, false},
|
||||
{"Resume", func() error { return p.Resume(ctx) }, false},
|
||||
{"Kill", func() error { return p.Kill(ctx, 9, false) }, true},
|
||||
{"KillAll", func() error { return p.KillAll(ctx) }, false},
|
||||
{"Exec", func() error { _, err := p.Exec(ctx, "", &ExecConfig{}); return err }, false},
|
||||
{"Checkpoint", func() error { return p.Checkpoint(ctx, &CheckpointConfig{}) }, false},
|
||||
{"Update", func() error { return p.Update(ctx, &google_protobuf.Any{}) }, false},
|
||||
})
|
||||
}
|
||||
|
||||
func TestExecOperationsWaitForDelete(t *testing.T) {
|
||||
g, e := newGate(t), newExec(t)
|
||||
e.closers = []io.Closer{g}
|
||||
ctx := t.Context()
|
||||
testOperationsWait(t, g, e, []operation{
|
||||
{"Start", func() error { return e.Start(ctx) }, false},
|
||||
{"SetExited", func() error { e.SetExited(1); return nil }, false},
|
||||
{"Delete", func() error { return e.Delete(ctx) }, true},
|
||||
{"Resize", func() error { return e.Resize(console.WinSize{}) }, false},
|
||||
{"Kill", func() error { return e.Kill(ctx, 9, false) }, true},
|
||||
})
|
||||
}
|
||||
|
||||
// Every reader takes mu, so it waits for a method that holds it, such as an
|
||||
// in-flight Start. Only the exec's own mu is held for the exec readers.
|
||||
func TestReadersTakeMu(t *testing.T) {
|
||||
p, e := newInit(t, "true"), newExec(t)
|
||||
p.SetExited(7)
|
||||
e.SetExited(3)
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
mu *sync.Mutex
|
||||
read func() any
|
||||
want any
|
||||
}{
|
||||
{"Init.Status", &p.mu, func() any { s, _ := p.Status(t.Context()); return s }, "stopped"},
|
||||
{"Init.ExitStatus", &p.mu, func() any { return p.ExitStatus() }, 7},
|
||||
{"Init.ExitedAt", &p.mu, func() any { return !p.ExitedAt().IsZero() }, true},
|
||||
{"Exec.Status", &e.mu, func() any { s, _ := e.Status(t.Context()); return s }, "stopped"},
|
||||
{"Exec.ExitStatus", &e.mu, func() any { return e.ExitStatus() }, 3},
|
||||
{"Exec.ExitedAt", &e.mu, func() any { return !e.ExitedAt().IsZero() }, true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
g := newGate(t)
|
||||
var got any
|
||||
tc.mu.Lock()
|
||||
done := g.run(func() { got = tc.read() })
|
||||
returned := finishes(done, settle)
|
||||
tc.mu.Unlock()
|
||||
if returned {
|
||||
t.Error("returned while mu was held")
|
||||
}
|
||||
mustFinish(t, done, tc.name)
|
||||
if got != tc.want {
|
||||
t.Errorf("got %v; want %v", got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user