From b54c08de263e81ff7c990add1ddb9fe95d84dcab Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti Date: Thu, 25 Jun 2026 15:07:30 +0000 Subject: [PATCH 1/4] guest/stdio: keep stdio alive across LCOW live migration A live migration pauses the UVM and drops the GCS vsock bridge, which then re-dials on the destination. Today the stdio relay tears the process down when that happens, and io.Copy throws away whatever it had buffered, so in-flight stdout/stderr is lost. Now the relay pauses instead. On a copy failure mid-migration a manager goroutine re-dials, swaps the conn, and resumes. A bridgeDown flag, set by cmd/gcs around the reconnect loop, is what tells a migration pause apart from a real process exit. The output copy drops io.Copy for a read/write-all loop that keeps the unwritten tail and replays it on the new conn, so the relay does not lose bytes. I tried two other designs first: a per-conn wrapper that parks Read/Write until reconnect (leaves the relays alone but wraps every read/write), and rewriting both relays into pump loops with a registry (touches the shared relay path every container and exec runs through). Went with pause-on-error because the conn is only swapped after the copiers stop, so nothing mutates a live conn under a running copy. Caveat: it is reactive. Whatever the kernel or socket already dropped at the blackout is gone; this only removes the relay-level drop. And if the host closes the conns just before bridgeDown flips, that stream still tears down. Tested under -race and on a two-node migration: the container keeps streaming across the move. Signed-off-by: Shreyansh Sancheti --- cmd/gcs/main.go | 11 + internal/guest/stdio/connection.go | 25 +- internal/guest/stdio/pauserelay.go | 200 ++++++ internal/guest/stdio/pauserelay_test.go | 916 ++++++++++++++++++++++++ internal/guest/stdio/stdio.go | 339 ++++++--- 5 files changed, 1393 insertions(+), 98 deletions(-) create mode 100644 internal/guest/stdio/pauserelay.go create mode 100644 internal/guest/stdio/pauserelay_test.go diff --git a/cmd/gcs/main.go b/cmd/gcs/main.go index fd9575ec3b..91ff2d6e1e 100644 --- a/cmd/gcs/main.go +++ b/cmd/gcs/main.go @@ -27,6 +27,7 @@ import ( "github.com/Microsoft/hcsshim/internal/guest/prot" "github.com/Microsoft/hcsshim/internal/guest/runtime/hcsv2" "github.com/Microsoft/hcsshim/internal/guest/runtime/runc" + "github.com/Microsoft/hcsshim/internal/guest/stdio" "github.com/Microsoft/hcsshim/internal/guest/transport" "github.com/Microsoft/hcsshim/internal/log" "github.com/Microsoft/hcsshim/internal/oc" @@ -449,6 +450,10 @@ func main() { bridgeOut = bridgeCon } + // The bridge is up again; let the stdio relays resume any copiers that + // paused when the previous bridge dropped during a live migration. + stdio.SetBridgeDown(false) + logrus.Info("bridge connected, serving") serveErr := b.ListenAndServe(bridgeIn, bridgeOut) @@ -458,6 +463,12 @@ func main() { break } + // The bridge dropped (for example during a live migration). Tell the + // stdio relays to pause rather than tear down so each process's stdio + // survives the reconnect; the relays re-dial and resume once the bridge + // is back and SetBridgeDown(false) is set above. + stdio.SetBridgeDown(true) + logrus.WithError(serveErr).Warn("bridge connection lost, will reconnect") time.Sleep(reconnectInterval) } diff --git a/internal/guest/stdio/connection.go b/internal/guest/stdio/connection.go index d4c7cbdf18..16847d5eda 100644 --- a/internal/guest/stdio/connection.go +++ b/internal/guest/stdio/connection.go @@ -19,7 +19,9 @@ type ConnectionSettings struct { // Connect returns new transport.Connection instances, one for each stdio pipe // to be used. If CreateStd*Pipe for a given pipe is false, the given Connection -// is set to nil. +// is set to nil. The returned set carries a redial closure so the stdio relays +// can re-establish the connections over the same vsock ports and pause and +// resume the process stdio across a live-migration bridge drop. func Connect(tport transport.Transport, settings ConnectionSettings) (_ *ConnectionSet, err error) { connSet := &ConnectionSet{} defer func() { @@ -28,25 +30,34 @@ func Connect(tport transport.Transport, settings ConnectionSettings) (_ *Connect } }() if settings.StdIn != nil { - c, err := tport.Dial(*settings.StdIn) + port := *settings.StdIn + c, err := tport.Dial(port) if err != nil { return nil, errors.Wrap(err, "failed creating stdin Connection") } - connSet.In = transport.NewLogConnection(c, *settings.StdIn) + connSet.In = transport.NewLogConnection(c, port) } if settings.StdOut != nil { - c, err := tport.Dial(*settings.StdOut) + port := *settings.StdOut + c, err := tport.Dial(port) if err != nil { return nil, errors.Wrap(err, "failed creating stdout Connection") } - connSet.Out = transport.NewLogConnection(c, *settings.StdOut) + connSet.Out = transport.NewLogConnection(c, port) } if settings.StdErr != nil { - c, err := tport.Dial(*settings.StdErr) + port := *settings.StdErr + c, err := tport.Dial(port) if err != nil { return nil, errors.Wrap(err, "failed creating stderr Connection") } - connSet.Err = transport.NewLogConnection(c, *settings.StdErr) + connSet.Err = transport.NewLogConnection(c, port) + } + // redial re-establishes a fresh ConnectionSet over the same vsock ports + // after a bridge drop, so the stdio relays can pause and resume across a + // live migration instead of tearing the process stdio down. + connSet.redial = func() (*ConnectionSet, error) { + return Connect(tport, settings) } return connSet, nil } diff --git a/internal/guest/stdio/pauserelay.go b/internal/guest/stdio/pauserelay.go new file mode 100644 index 0000000000..4e2e73b69b --- /dev/null +++ b/internal/guest/stdio/pauserelay.go @@ -0,0 +1,200 @@ +//go:build linux +// +build linux + +package stdio + +import ( + "io" + "sync/atomic" + "time" + + "github.com/Microsoft/hcsshim/internal/guest/transport" + "github.com/pkg/errors" + "github.com/sirupsen/logrus" +) + +// bridgeDown reports whether the GCS bridge to the host is currently down (for +// example because a live migration is tearing down and re-establishing the +// vsock connections). The stdio relays consult it to tell a live-migration +// disconnect (pause and resume) apart from a normal process-driven copy error +// (tear down). It is set by cmd/gcs around the bridge reconnect loop. +var bridgeDown atomic.Bool + +// SetBridgeDown records whether the host bridge is currently down. cmd/gcs sets +// it true when ListenAndServe returns (and it is not a shutdown) and false once +// the bridge has reconnected. +func SetBridgeDown(v bool) { + bridgeDown.Store(v) +} + +const ( + // redialInterval is the delay between attempts to re-establish the stdio + // connections after a bridge drop. + redialInterval = 100 * time.Millisecond + // maxRedialAttempts bounds how long a relay waits for the bridge to come + // back before giving up and tearing the process stdio down. + maxRedialAttempts = 60 +) + +// redialWithRetry re-establishes a ConnectionSet using the provided redial +// closure, retrying up to maxRedialAttempts with redialInterval between +// attempts. It returns the new set on success or the last error after the +// attempts are exhausted. +func redialWithRetry(redial func() (*ConnectionSet, error)) (*ConnectionSet, error) { + var err error + for i := 0; i < maxRedialAttempts; i++ { + var ns *ConnectionSet + ns, err = redial() + if err == nil { + return ns, nil + } + time.Sleep(redialInterval) + } + return nil, err +} + +// relayBufferSize is the size of the per-stream copy buffer the output relays +// read into before writing to the host connection. It matches io.Copy's +// default buffer size; the unwritten tail of one of these buffers is what is +// held and replayed across a live-migration pause. +const relayBufferSize = 32 * 1024 + +// writeAll writes all of p to w, looping until every byte is written or a write +// returns an error. It returns the number of bytes written so a caller +// recovering from a bridge drop can hold and replay the unwritten remainder +// p[n:] on a freshly dialed connection. copyOut uses it to write process output +// to the host conn and copyIn to write host input to the process pipe, so the +// parameter is the wide io.Writer that both a transport.Connection and a pipe +// satisfy. +func writeAll(w io.Writer, p []byte) (int, error) { + written := 0 + for written < len(p) { + n, err := w.Write(p[written:]) + written += n + if err != nil { + return written, err + } + if n == 0 { + // A well-behaved io.Writer never returns (0, nil) with bytes + // still to write; treat it as a short write (matching io.Copy's + // contract) so a misbehaving connection cannot spin this loop + // forever. + return written, io.ErrShortWrite + } + } + return written, nil +} + +// copyIn copies host input from conn r into the process sink w (a stdin pipe or +// a pty master). It decides by which side failed, mirroring copyOut: a read +// error on r is the host conn, so a bridge drop during a live migration +// (pauseOnError and bridgeDown) returns true (a pause) and the manager re-dials +// so the next call resumes reading from the fresh conn; a write error on w is +// the process closing its stdin (EPIPE), which is a normal end and returns false +// even while the bridge is down (otherwise a post-resume stdin close would be +// mistaken for a pause and spin the manager re-dialing). A bridge-down read +// yields no bytes, so there is nothing to hold. On host EOF and any other +// unexpected error it returns false. +func copyIn(w io.Writer, r io.Reader, name string, pauseOnError bool) (paused bool) { + l := logrus.WithField("file", name) + buf := make([]byte, relayBufferSize) + for { + nr, rerr := r.Read(buf) + if nr > 0 { + if _, werr := writeAll(w, buf[:nr]); werr != nil { + // A pipe write failure is the process closing its stdin: a + // normal end, never a pause, even while the bridge is down. + l.WithError(werr).Error("opengcs::stdio::copyIn - error writing to process input") + return false + } + } + if rerr != nil { + // io.EOF is the host closing stdin: a normal end. + if rerr == io.EOF { + return false + } + if pauseOnError && bridgeDown.Load() { + // A conn read error while the bridge is down is the + // live-migration drop: pause so the manager re-dials. + return true + } + l.WithError(rerr).Error("opengcs::stdio::copyIn - error reading input") + return false + } + } +} + +// copyOut copies process output from pipe r into host conn c using a retained +// buffer. pending holds the in-flight remainder from a prior bridge-drop pause +// and is replayed onto c first. On a bridge-down write failure it returns the +// still-unwritten bytes as held and paused=true so the manager re-dials and the +// next call replays them on the fresh conn (no relay-level drop). On pipe EOF +// (the process closed the stream) it performs the existing clean socket shutdown +// and returns paused=false. +func copyOut(c transport.Connection, r io.Reader, name string, pending []byte, pauseOnError bool) (held []byte, paused bool) { + l := logrus.WithField("file", name) + + // copyDone is set when the copy phase is over for a non-pause reason (pipe + // EOF or a real copy error); the relay then cleanly shuts the socket down. + // A pause returns early, before any shutdown, so the manager can re-dial and + // the next call replays the held remainder. + copyDone := false + + // Replay any in-flight remainder retained from a prior pause first, so the + // bytes already read from the pipe before the bridge dropped are not lost. + if len(pending) > 0 { + if n, err := writeAll(c, pending); err != nil { + if pauseOnError && bridgeDown.Load() { + l.WithError(err).Info("opengcs::stdio::copyOut - pausing on bridge down") + h := make([]byte, len(pending)-n) + copy(h, pending[n:]) + return h, true + } + l.WithError(err).Error("opengcs::stdio::copyOut - error replaying retained output") + copyDone = true + } + } + + buf := make([]byte, relayBufferSize) + for !copyDone { + nr, rerr := r.Read(buf) + if nr > 0 { + if nw, werr := writeAll(c, buf[:nr]); werr != nil { + if pauseOnError && bridgeDown.Load() { + l.WithError(werr).Info("opengcs::stdio::copyOut - pausing on bridge down") + h := make([]byte, nr-nw) + copy(h, buf[nw:nr]) + return h, true + } + l.WithError(werr).Error("opengcs::stdio::copyOut - error copying from pipe") + break + } + } + if rerr != nil { + if rerr != io.EOF { + l.WithError(rerr).Error("opengcs::stdio::copyOut - error reading from pipe") + } + break + } + } + + // Shut down the write end of the socket, then read a byte (which should + // yield EOF) to wait for the other endpoint to finish reading and close + // the connection. + if err := c.CloseWrite(); err == nil { + var b [1]byte + _, err = c.Read(b[:]) + if err == nil { + err = errors.New("unexpected data in socket") + } + if err != io.EOF { //nolint:errorlint + l.WithError(err).Error("opengcs::stdio::copyOut - error reading for clean close") + } + } else { + l.WithError(err).Error("opengcs::stdio::copyOut - error shutting down socket") + } + if err := c.Close(); err != nil { + l.WithError(err).Error("opengcs::stdio::copyOut - error closing socket") + } + return nil, false +} diff --git a/internal/guest/stdio/pauserelay_test.go b/internal/guest/stdio/pauserelay_test.go new file mode 100644 index 0000000000..17300ac882 --- /dev/null +++ b/internal/guest/stdio/pauserelay_test.go @@ -0,0 +1,916 @@ +//go:build linux +// +build linux + +package stdio + +import ( + "bytes" + "errors" + "fmt" + "io" + "net" + "os" + "sync" + "testing" + "time" + + "github.com/Microsoft/hcsshim/internal/guest/transport" +) + +// fakeConn adapts a net.Conn (one end of net.Pipe) to transport.Connection +// (io.ReadWriteCloser + CloseRead + CloseWrite + File). net.Pipe has no +// half-close, so CloseRead/CloseWrite close the whole connection, which is +// sufficient for the relay's pause/resume and clean-close paths. File is never +// exercised by the relay copiers, so it returns an error. +type fakeConn struct { + net.Conn +} + +func (f *fakeConn) CloseRead() error { return f.Close() } +func (f *fakeConn) CloseWrite() error { return f.Close() } +func (f *fakeConn) File() (*os.File, error) { + return nil, fmt.Errorf("fakeConn does not support File") +} + +var _ transport.Connection = (*fakeConn)(nil) + +// closeSignalConn is a transport.Connection whose Read blocks until unblock is +// closed (then returns EOF) and whose Close/CloseRead/CloseWrite are no-ops that +// introduce NO cross-goroutine synchronization. The no-op closes are the whole +// point: TestPipeRelayWaitRaceWithTeardown needs Wait's read of ConnectionSet.In +// and the relay manager goroutine's write of it (s.In = nil in Close) to race on +// the struct field itself. A fakeConn would route both Wait's CloseRead and the +// teardown's Close through net.Pipe's internal mutex, whose happens-before edge +// would mask that field race from -race; a real vsock conn's CloseRead/Close are +// independent syscalls and add no such edge, which is the production condition +// this double reproduces. +type closeSignalConn struct { + unblock chan struct{} +} + +func (c *closeSignalConn) Read([]byte) (int, error) { <-c.unblock; return 0, io.EOF } +func (c *closeSignalConn) Write(p []byte) (int, error) { return len(p), nil } +func (c *closeSignalConn) Close() error { return nil } +func (c *closeSignalConn) CloseRead() error { return nil } +func (c *closeSignalConn) CloseWrite() error { return nil } +func (c *closeSignalConn) File() (*os.File, error) { + return nil, fmt.Errorf("closeSignalConn does not support File") +} + +var _ transport.Connection = (*closeSignalConn)(nil) + +// TestPipeRelayPauseResume verifies that a PipeRelay pauses (rather than tears +// down) when the bridge drops mid-copy, re-dials the stdio connection, resumes +// copying onto the new connection, and only completes Wait once the process +// stdio is truly finished. +func TestPipeRelayPauseResume(t *testing.T) { + SetBridgeDown(false) + defer SetBridgeDown(false) + + // Each redial hands the test the new host end of a fresh net.Pipe so it can + // read what the resumed relay writes. + redialedHosts := make(chan net.Conn, 4) + var redial func() (*ConnectionSet, error) + redial = func() (*ConnectionSet, error) { + host, guest := net.Pipe() + redialedHosts <- host + return &ConnectionSet{Out: &fakeConn{Conn: guest}, redial: redial}, nil + } + + host1, guest1 := net.Pipe() + out1 := &fakeConn{Conn: guest1} + set := &ConnectionSet{Out: out1, redial: redial} + + pr, err := NewPipeRelay(nil) + if err != nil { + t.Fatalf("NewPipeRelay: %v", err) + } + pr.ReplaceConnectionSet(set) + pr.CloseUnusedPipes() + // Capture the stdout write pipe before Start so the test never races the + // relay manager's closePipes on the pr.pipes fields. + stdoutW := pr.pipes[3] + pr.Start() + + // Pre-pause: write "hello" through the relay and read it on the host. + if _, err := stdoutW.Write([]byte("hello")); err != nil { + t.Fatalf("write hello: %v", err) + } + _ = host1.SetReadDeadline(time.Now().Add(5 * time.Second)) + got := make([]byte, 5) + if _, err := io.ReadFull(host1, got); err != nil { + t.Fatalf("read hello: %v", err) + } + if !bytes.Equal(got, []byte("hello")) { + t.Fatalf("pre-pause got %q, want hello", got) + } + + // Simulate a bridge drop: mark the bridge down, close the host stdio conn, + // then nudge the copier so its next write hits the closed conn and pauses. + // The "x" is read from the pipe but cannot be written to the dead conn, so + // the relay must hold it and replay it on the re-dialed conn (no drop). + SetBridgeDown(true) + _ = out1.Close() + if _, err := stdoutW.Write([]byte("x")); err != nil { + t.Fatalf("write pause trigger: %v", err) + } + + // The relay re-dials; grab the new host end. + var host2 net.Conn + select { + case host2 = <-redialedHosts: + case <-time.After(5 * time.Second): + t.Fatal("relay did not redial after pause") + } + + // Post-resume: write "world"; assert the held "x" is replayed first and + // then "world" arrives on the new host conn (the in-flight byte is not + // dropped across the migration pause). + if _, err := stdoutW.Write([]byte("world")); err != nil { + t.Fatalf("write world: %v", err) + } + _ = host2.SetReadDeadline(time.Now().Add(5 * time.Second)) + got2 := make([]byte, len("xworld")) + if _, err := io.ReadFull(host2, got2); err != nil { + t.Fatalf("read xworld: %v", err) + } + if !bytes.Equal(got2, []byte("xworld")) { + t.Fatalf("post-resume got %q, want xworld", got2) + } + + // Process exit: close the stdout write pipe and the host conn so the relay + // finishes and Wait returns. + _ = stdoutW.Close() + _ = host2.Close() + SetBridgeDown(false) + + waitDone := make(chan struct{}) + go func() { + pr.Wait() + close(waitDone) + }() + select { + case <-waitDone: + case <-time.After(5 * time.Second): + t.Fatal("Wait did not return after process exit") + } +} + +// errTestWriteFail is the failure a dead host connection's Write returns after a +// live-migration bridge drop. Both the BEFORE baseline (failWriter) and the +// AFTER proofs (recordConn) raise it so the two behaviors are compared against +// the same simulated fault. +var errTestWriteFail = errors.New("stdio test: simulated dead-conn write failure") + +// failWriter is a minimal io.Writer whose Write always fails immediately without +// accepting any bytes. It is the destination in TestIoCopyDropsInFlightBytes: +// io.Copy reads the source into its internal buffer, this Write fails, and the +// buffered bytes are discarded. That discard is the pre-fix in-flight drop the +// new copyOut prevents. +type failWriter struct{} + +func (failWriter) Write(p []byte) (int, error) { return 0, errTestWriteFail } + +// countingReader is a source over a fixed payload that deliberately does NOT +// implement io.WriterTo, forcing io.Copy down its generic +// read-into-internal-buffer path (the path whose buffer is lost on a write +// error; the real relay source, an *os.File pipe end, has no faster path to a +// plain connection either). It records how many bytes io.Copy has read so the +// test can prove the payload was drained into, and then dropped by, io.Copy. +type countingReader struct { + data []byte + off int + read int +} + +func (r *countingReader) Read(p []byte) (int, error) { + if r.off >= len(r.data) { + return 0, io.EOF + } + n := copy(p, r.data[r.off:]) + r.off += n + r.read += n + return n, nil +} + +// recordConn is a transport.Connection test double for driving copyOut directly. +// When writeErr is non-nil every Write fails with it (modeling a dead host conn +// after a bridge drop); if partialN > 0 the Write first accepts (and records) +// that many bytes before failing, modeling a true partial write where the socket +// took a prefix before the bridge dropped. When zeroWrite is true every Write +// returns (0, nil) to exercise writeAll's short-write guard. Otherwise Write +// records the bytes so a test can prove the retained in-flight remainder is +// replayed in order. Read returns EOF so copyOut's clean-close path completes; +// CloseRead/CloseWrite/Close are inert and File is never used by the relay +// copiers. +type recordConn struct { + mu sync.Mutex + written []byte + writeErr error + zeroWrite bool + partialN int +} + +func (c *recordConn) Read(p []byte) (int, error) { return 0, io.EOF } + +func (c *recordConn) Write(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.zeroWrite { + return 0, nil + } + if c.writeErr != nil { + // A partial write accepts (and records) the first partialN bytes as + // socket-level loss, then fails; partialN == 0 is the immediate-failure + // conn that accepts nothing. + if c.partialN > 0 { + n := c.partialN + if n > len(p) { + n = len(p) + } + c.written = append(c.written, p[:n]...) + return n, c.writeErr + } + return 0, c.writeErr + } + c.written = append(c.written, p...) + return len(p), nil +} + +func (c *recordConn) Close() error { return nil } +func (c *recordConn) CloseRead() error { return nil } +func (c *recordConn) CloseWrite() error { return nil } +func (c *recordConn) File() (*os.File, error) { + return nil, fmt.Errorf("recordConn does not support File") +} + +// recorded returns a copy of the bytes written to the connection so far. +func (c *recordConn) recorded() []byte { + c.mu.Lock() + defer c.mu.Unlock() + b := make([]byte, len(c.written)) + copy(b, c.written) + return b +} + +var _ transport.Connection = (*recordConn)(nil) + +// testPattern returns n bytes of a deterministic, position-dependent pattern so +// any gap, duplication, or reordering of relayed bytes is detectable on +// comparison. 251 is prime, so the pattern does not align with 256 or the copy +// buffer sizes. +func testPattern(n int) []byte { + b := make([]byte, n) + for i := range b { + b[i] = byte(i % 251) + } + return b +} + +// firstDiff returns the index of the first byte at which a and b differ, or -1 +// when they are equal. It pinpoints where a relayed stream lost contiguity for +// a compact failure message instead of dumping kilobytes of %q. +func firstDiff(a, b []byte) int { + n := len(a) + if len(b) < n { + n = len(b) + } + for i := 0; i < n; i++ { + if a[i] != b[i] { + return i + } + } + if len(a) != len(b) { + return n + } + return -1 +} + +// window returns up to 16 bytes of p starting at i (clamped to p's bounds) for +// compact %q logging around the point two byte streams diverge. +func window(p []byte, i int) []byte { + if i < 0 { + i = 0 + } + if i > len(p) { + i = len(p) + } + end := i + 16 + if end > len(p) { + end = len(p) + } + return p[i:end] +} + +// TestIoCopyDropsInFlightBytes is the BEFORE baseline. In isolation it +// demonstrates the in-flight drop the old io.Copy-based relay suffered on a +// bridge drop: io.Copy reads the whole payload from the source into its internal +// buffer, the destination Write then fails, and those buffered bytes are gone. +// Zero bytes reach the destination yet the source is fully drained, so the bytes +// are unrecoverable. copyOut (proved by TestCopyOutRetainsInFlightBytes) instead +// holds and replays exactly these bytes. +func TestIoCopyDropsInFlightBytes(t *testing.T) { + payload := testPattern(5000) + src := &countingReader{data: payload} + + n, err := io.Copy(failWriter{}, src) + + if !errors.Is(err, errTestWriteFail) { + t.Fatalf("io.Copy err = %v, want %v", err, errTestWriteFail) + } + if n != 0 { + t.Fatalf("io.Copy transferred %d bytes to the failed destination, want 0", n) + } + // The bytes were read out of the source into io.Copy's internal buffer and + // then lost when the Write failed: the source is fully drained but nothing + // landed at the destination. This is the dropped in-flight data. + if src.read != len(payload) { + t.Fatalf("source read %d bytes, want %d (io.Copy did not drain the in-flight buffer it then dropped)", src.read, len(payload)) + } + if src.off != len(payload) { + t.Fatalf("source has %d bytes still unread, want 0 (expected fully drained)", len(payload)-src.off) + } +} + +// TestCopyOutRetainsInFlightBytes is the AFTER proof at the copyOut unit level. +// It shows copyOut holds the in-flight remainder that io.Copy would have dropped +// (see TestIoCopyDropsInFlightBytes) and replays it intact, in order, on the +// re-dialed connection. +func TestCopyOutRetainsInFlightBytes(t *testing.T) { + SetBridgeDown(true) + defer SetBridgeDown(false) + + // > one read but < relayBufferSize, so the payload is a single in-flight + // chunk held whole on the failed write. + payload := testPattern(5000) + pr, pw, err := os.Pipe() + if err != nil { + t.Fatalf("os.Pipe: %v", err) + } + defer pr.Close() + if _, err := pw.Write(payload); err != nil { + t.Fatalf("preload pipe: %v", err) + } + + // First copyOut: the conn's Write fails while the bridge is down, so copyOut + // must pause and hand back the full payload it read but could not deliver. + deadConn := &recordConn{writeErr: errTestWriteFail} + held, paused := copyOut(deadConn, pr, "stdout", nil, true) + if !paused { + t.Fatal("copyOut paused = false, want true on a bridge-down write failure") + } + if !bytes.Equal(held, payload) { + t.Fatalf("copyOut held %d in-flight bytes, want %d retained intact: got %q want %q", + len(held), len(payload), held, payload) + } + if got := deadConn.recorded(); len(got) != 0 { + t.Fatalf("dead conn received %d bytes, want 0 (nothing should reach a failed write): %q", len(got), got) + } + + // Second copyOut: replay the held remainder onto a fresh recording conn. Run + // it in a goroutine because, after the replay, copyOut blocks reading the + // still-open pipe; closing the write end lets it reach clean shutdown. + rec := &recordConn{} + done := make(chan struct{}) + var held2 []byte + var paused2 bool + go func() { + held2, paused2 = copyOut(rec, pr, "stdout", held, true) + close(done) + }() + _ = pw.Close() // unblock the post-replay read with EOF so copyOut returns + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("second copyOut did not return after pipe close") + } + + if paused2 { + t.Fatal("second copyOut paused = true, want false (write succeeds on the fresh conn)") + } + if held2 != nil { + t.Fatalf("second copyOut held %q, want nil (nothing left to retain)", held2) + } + got := rec.recorded() + if !bytes.HasPrefix(got, payload) { + t.Fatalf("re-dialed conn did not receive the replayed remainder first: got %q want prefix %q", got, payload) + } + if !bytes.Equal(got, payload) { + t.Fatalf("re-dialed conn received %q, want exactly the replayed payload %q", got, payload) + } +} + +// TestCopyOutRetainsAcrossDoublePause proves the retained remainder survives a +// re-dial that ALSO fails. A held buffer that cannot be replayed on the first +// re-dialed conn (bridge still down) must be re-held, not dropped or truncated, +// and is replayed once a working conn appears. +func TestCopyOutRetainsAcrossDoublePause(t *testing.T) { + SetBridgeDown(true) + defer SetBridgeDown(false) + + payload := testPattern(5000) + + // First re-dial still bridge-down: the replay of the held remainder fails, + // so copyOut must hand the same bytes back to be held again. The reader is + // never consulted because the pending replay fails before copyOut's read + // loop. + deadConn1 := &recordConn{writeErr: errTestWriteFail} + held1, paused1 := copyOut(deadConn1, bytes.NewReader(nil), "stdout", payload, true) + if !paused1 { + t.Fatal("first re-dial copyOut paused = false, want true (write still failing)") + } + if !bytes.Equal(held1, payload) { + t.Fatalf("first re-dial dropped/truncated the held remainder: got %q want %q", held1, payload) + } + + // Second re-dial also bridge-down: the re-held remainder must again be + // returned whole. + deadConn2 := &recordConn{writeErr: errTestWriteFail} + held2, paused2 := copyOut(deadConn2, bytes.NewReader(nil), "stdout", held1, true) + if !paused2 { + t.Fatal("second re-dial copyOut paused = false, want true (write still failing)") + } + if !bytes.Equal(held2, payload) { + t.Fatalf("second re-dial dropped/truncated the re-held remainder: got %q want %q", held2, payload) + } + + // Finally a working conn: the twice-held remainder replays intact and in + // order. + pr, pw, err := os.Pipe() + if err != nil { + t.Fatalf("os.Pipe: %v", err) + } + defer pr.Close() + rec := &recordConn{} + done := make(chan struct{}) + var paused3 bool + go func() { + _, paused3 = copyOut(rec, pr, "stdout", held2, true) + close(done) + }() + _ = pw.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("final copyOut did not return after pipe close") + } + if paused3 { + t.Fatal("final copyOut paused = true, want false (write succeeds)") + } + if got := rec.recorded(); !bytes.Equal(got, payload) { + t.Fatalf("final replay delivered %q, want the full twice-held payload %q", got, payload) + } +} + +// TestWriteAllShortWriteGuard verifies the hardening that a misbehaving +// Connection returning (0, nil) makes writeAll fail with io.ErrShortWrite +// instead of spinning forever, matching io.Copy's contract. It runs writeAll in +// a goroutine so a regression that drops the guard fails as a timeout rather +// than hanging the package. +func TestWriteAllShortWriteGuard(t *testing.T) { + conn := &recordConn{zeroWrite: true} + type result struct { + n int + err error + } + res := make(chan result, 1) + go func() { + n, err := writeAll(conn, testPattern(64)) + res <- result{n, err} + }() + select { + case r := <-res: + if !errors.Is(r.err, io.ErrShortWrite) { + t.Fatalf("writeAll on a (0, nil) conn err = %v, want io.ErrShortWrite", r.err) + } + if r.n != 0 { + t.Fatalf("writeAll on a (0, nil) conn wrote %d bytes, want 0", r.n) + } + case <-time.After(5 * time.Second): + t.Fatal("writeAll did not return on a (0, nil) conn: the short-write guard is missing (infinite loop)") + } +} + +// TestPipeRelayRetainsBurstAcrossPause drives a multi-KB burst through a full +// PipeRelay, drops the bridge after the host has read only the first half, and +// asserts the re-dialed host receives the exact remaining bytes so the two host +// reads concatenate to the original burst with no gap, duplication, or +// reordering. This is the end-to-end proof that the in-flight bytes io.Copy +// dropped are now retained and replayed. +func TestPipeRelayRetainsBurstAcrossPause(t *testing.T) { + SetBridgeDown(false) + defer SetBridgeDown(false) + + const ( + burstLen = 16 * 1024 + splitAt = 8 * 1024 + ) + payload := testPattern(burstLen) + + redialedHosts := make(chan net.Conn, 4) + var redial func() (*ConnectionSet, error) + redial = func() (*ConnectionSet, error) { + host, guest := net.Pipe() + redialedHosts <- host + return &ConnectionSet{Out: &fakeConn{Conn: guest}, redial: redial}, nil + } + + host1, guest1 := net.Pipe() + out1 := &fakeConn{Conn: guest1} + set := &ConnectionSet{Out: out1, redial: redial} + + pr, err := NewPipeRelay(nil) + if err != nil { + t.Fatalf("NewPipeRelay: %v", err) + } + pr.ReplaceConnectionSet(set) + pr.CloseUnusedPipes() + // Capture the stdout write pipe before Start so the test never races the + // relay manager's closePipes on the pr.pipes fields. + stdoutW := pr.pipes[3] + pr.Start() + + // The process emits the whole burst at once. The os.Pipe buffers it (the + // burst is smaller than a pipe's capacity), so this Write does not block on + // the relay; a goroutine guards against any partial-buffer edge anyway. + writeErr := make(chan error, 1) + go func() { + _, werr := stdoutW.Write(payload) + writeErr <- werr + }() + + // Read only the first half on the original host. net.Pipe is synchronous, so + // this returns once the relay's in-flight write has delivered exactly splitAt + // bytes; the relay is then blocked mid-write holding the remainder. + _ = host1.SetReadDeadline(time.Now().Add(5 * time.Second)) + first := make([]byte, splitAt) + if _, err := io.ReadFull(host1, first); err != nil { + t.Fatalf("read first half on original host: %v", err) + } + + // Drop the bridge: mark it down, then close the host conn so the relay's + // blocked write fails. copyOut must hold the undelivered remainder and replay + // it on the re-dialed conn rather than drop it (the old io.Copy bug). + SetBridgeDown(true) + _ = out1.Close() + + var host2 net.Conn + select { + case host2 = <-redialedHosts: + case <-time.After(5 * time.Second): + t.Fatal("relay did not redial after pause") + } + + // The relay replays the held remainder onto the new host. Read exactly the + // remaining bytes; io.ReadFull collects them across however many writes the + // relay needs. + _ = host2.SetReadDeadline(time.Now().Add(5 * time.Second)) + second := make([]byte, burstLen-splitAt) + if _, err := io.ReadFull(host2, second); err != nil { + t.Fatalf("read held remainder on re-dialed host: %v", err) + } + if werr := <-writeErr; werr != nil { + t.Fatalf("process burst write: %v", werr) + } + + // The held remainder must be exactly the bytes that follow the split point, + // and the two host reads must reconstruct the original burst contiguously. + if !bytes.Equal(second, payload[splitAt:]) { + d := firstDiff(second, payload[splitAt:]) + t.Fatalf("held remainder differs from the dropped bytes at offset %d: got %q want %q", + d, window(second, d), window(payload[splitAt:], d)) + } + got := append(append([]byte{}, first...), second...) + if !bytes.Equal(got, payload) { + d := firstDiff(got, payload) + t.Fatalf("relayed burst not contiguous across pause: got %d bytes want %d, first diff at %d: got %q want %q", + len(got), len(payload), d, window(got, d), window(payload, d)) + } + + // Process exit: close stdout and the host conn so the relay finishes. + _ = stdoutW.Close() + _ = host2.Close() + SetBridgeDown(false) + + waitDone := make(chan struct{}) + go func() { + pr.Wait() + close(waitDone) + }() + select { + case <-waitDone: + case <-time.After(5 * time.Second): + t.Fatal("Wait did not return after process exit") + } +} + +// TestCopyOutHoldsPartialWriteRemainder pins the partial-loss boundary. The conn +// accepts the first K bytes of an in-flight chunk and then fails while the bridge +// is down (a true partial write, the socket having taken a prefix before the +// drop). copyOut must hold exactly the tail the socket never accepted: the K +// accepted bytes are socket-level loss the relay cannot recover, and only +// payload[K:] is retained for replay on the re-dialed conn. +func TestCopyOutHoldsPartialWriteRemainder(t *testing.T) { + SetBridgeDown(true) + defer SetBridgeDown(false) + + // A single in-flight chunk (< relayBufferSize) so copyOut issues exactly one + // Write; the conn takes k bytes then fails. + const k = 1500 + payload := testPattern(5000) + pr, pw, err := os.Pipe() + if err != nil { + t.Fatalf("os.Pipe: %v", err) + } + defer pr.Close() + if _, err := pw.Write(payload); err != nil { + t.Fatalf("preload pipe: %v", err) + } + + conn := &recordConn{writeErr: errTestWriteFail, partialN: k} + held, paused := copyOut(conn, pr, "stdout", nil, true) + if !paused { + t.Fatal("copyOut paused = false, want true on a bridge-down partial write") + } + // held must be exactly the bytes after the accepted prefix. + want := payload[k:] + if !bytes.Equal(held, want) { + d := firstDiff(held, want) + t.Fatalf("copyOut held the wrong partial-write remainder: got %d bytes want %d, first diff at %d: got %q want %q", + len(held), len(want), d, window(held, d), window(want, d)) + } + // The conn recorded exactly the first K bytes (the socket-level loss). + if got := conn.recorded(); !bytes.Equal(got, payload[:k]) { + d := firstDiff(got, payload[:k]) + t.Fatalf("conn recorded the wrong accepted prefix: got %d bytes want %d, first diff at %d: got %q want %q", + len(got), k, d, window(got, d), window(payload[:k], d)) + } +} + +// errTestReadFail is the read-side failure a dropped bridge conn returns. It is +// the conn READ error (distinct from a pipe WRITE error) that must make copyIn +// pause while the bridge is down, proving copyIn decides by which side failed. +var errTestReadFail = errors.New("stdio test: simulated bridge-drop conn read failure") + +// blockUntilReader yields data once and then blocks on block (closed by the +// test) on any further Read. In TestCopyInFinishesOnStdinCloseWhileBridgeDown the +// write fails on the first chunk so the second Read is never reached; the block +// is a guard that turns a regression (looping instead of returning) into a +// timeout rather than a wrong pass, and ensures the only way copyIn can return is +// via the write side. +type blockUntilReader struct { + data []byte + sent bool + block chan struct{} +} + +func (r *blockUntilReader) Read(p []byte) (int, error) { + if !r.sent { + r.sent = true + return copy(p, r.data), nil + } + <-r.block + return 0, io.EOF +} + +// errReader returns err on every Read with no bytes, modeling a host conn whose +// read fails (a bridge drop) rather than a normal EOF. +type errReader struct{ err error } + +func (r errReader) Read(p []byte) (int, error) { return 0, r.err } + +// TestCopyInFinishesOnStdinCloseWhileBridgeDown is the gap the side-aware copyIn +// closes. With the bridge down, a process that closes its stdin (a pipe WRITE +// failure) must finish (paused=false), not be mistaken for a live-migration +// pause; a real bridge-drop conn READ error must still pause (paused=true). +// Together they prove copyIn decides by which side failed, not by a single +// collapsed error as the old io.Copy did. +func TestCopyInFinishesOnStdinCloseWhileBridgeDown(t *testing.T) { + SetBridgeDown(true) + defer SetBridgeDown(false) + + // Process closed stdin: the reader yields bytes (so a write is attempted) + // then blocks, and the sink's Write fails (EPIPE). copyIn must return via the + // write side as false (done); the block guarantees no read error or EOF can + // supply the return instead. + block := make(chan struct{}) + t.Cleanup(func() { close(block) }) + r := &blockUntilReader{data: []byte("stdin-bytes"), block: block} + + done := make(chan bool, 1) + go func() { + // failWriter models the process having closed its stdin: every Write + // returns the EPIPE-like errTestWriteFail. + done <- copyIn(failWriter{}, r, "stdin", true) + }() + select { + case paused := <-done: + if paused { + t.Fatalf("copyIn paused = %v on a process stdin-close write failure while bridge down, want false: a pipe write error must finish, not pause", paused) + } + case <-time.After(5 * time.Second): + t.Fatal("copyIn did not return on a stdin-close write failure: it must decide by the write side, not loop while the bridge is down") + } + + // Real bridge-drop conn read error while bridge down: copyIn must pause so + // the manager re-dials and the next call resumes on the fresh conn. + if paused := copyIn(io.Discard, errReader{err: errTestReadFail}, "stdin", true); !paused { + t.Fatalf("copyIn paused = %v on a bridge-down conn read error, want true: a conn read failure during migration must pause", paused) + } +} + +// TestPipeRelaySurvivesMultipleMigrations drives a full PipeRelay through several +// back-and-forth live-migration pauses in a row: each cycle reads a chunk on the +// current host, drops the bridge and holds a single in-flight byte, re-dials a +// fresh net.Pipe host, and resumes. It asserts the concatenation of what every +// successive host conn received reconstructs everything written, contiguously and +// in order, proving each migration holds and replays correctly and that no +// held-buffer state leaks from one migration into the next. Synchronization is +// net.Pipe's blocking reads plus the redial handoff channel; there are no sleeps. +func TestPipeRelaySurvivesMultipleMigrations(t *testing.T) { + SetBridgeDown(false) + defer SetBridgeDown(false) + + const cycles = 4 // > 3 back-and-forth migrations + + // Carve every chunk from one position-dependent pattern so the concatenation + // of all host receptions must reproduce it byte-for-byte; any gap, + // duplication, or reorder across the repeated pause/replay cycles surfaces as + // a firstDiff. + base := testPattern(8 * 1024) + off := 0 + take := func(n int) []byte { + s := base[off : off+n] + off += n + return s + } + + redialedHosts := make(chan net.Conn, cycles+2) + var redial func() (*ConnectionSet, error) + redial = func() (*ConnectionSet, error) { + host, guest := net.Pipe() + redialedHosts <- host + return &ConnectionSet{Out: &fakeConn{Conn: guest}, redial: redial}, nil + } + + host, guest := net.Pipe() + set := &ConnectionSet{Out: &fakeConn{Conn: guest}, redial: redial} + + pr, err := NewPipeRelay(nil) + if err != nil { + t.Fatalf("NewPipeRelay: %v", err) + } + pr.ReplaceConnectionSet(set) + pr.CloseUnusedPipes() + // Capture the stdout write pipe before Start so the test never races the + // relay manager's closePipes on the pr.pipes fields. + stdoutW := pr.pipes[3] + pr.Start() + + // written accumulates every byte handed to the process stdout pipe; read + // accumulates every byte the successive host conns deliver. They must match. + var written, read []byte + + for i := 0; i < cycles; i++ { + // Pre-pause chunk for this connection (a distinct length per cycle). On + // every cycle after the first the host also leads with the single held + // byte replayed from the prior pause, so the expected read is that byte + // plus this chunk. + pre := take(96 + i) + if _, err := stdoutW.Write(pre); err != nil { + t.Fatalf("cycle %d: write pre-pause chunk: %v", i, err) + } + written = append(written, pre...) + + expect := len(pre) + if i > 0 { + expect++ // leading replayed held byte from the previous pause + } + buf := make([]byte, expect) + _ = host.SetReadDeadline(time.Now().Add(5 * time.Second)) + if _, err := io.ReadFull(host, buf); err != nil { + t.Fatalf("cycle %d: read %d bytes on host: %v", i, expect, err) + } + read = append(read, buf...) + + // Drop the bridge and close this host so the relay's next write fails and + // pauses. The single nudge byte is read from the pipe but cannot be + // delivered, so the relay must hold it and replay it on the next conn. + SetBridgeDown(true) + _ = host.Close() + nudge := take(1) + if _, err := stdoutW.Write(nudge); err != nil { + t.Fatalf("cycle %d: write held nudge byte: %v", i, err) + } + written = append(written, nudge...) + + select { + case host = <-redialedHosts: + case <-time.After(5 * time.Second): + t.Fatalf("cycle %d: relay did not redial after pause", i) + } + SetBridgeDown(false) + } + + // Final connection: read the last held byte plus a final chunk, then drive a + // clean process exit and assert Wait returns. + final := take(64) + if _, err := stdoutW.Write(final); err != nil { + t.Fatalf("write final chunk: %v", err) + } + written = append(written, final...) + tail := make([]byte, 1+len(final)) // leading replayed held byte + final + _ = host.SetReadDeadline(time.Now().Add(5 * time.Second)) + if _, err := io.ReadFull(host, tail); err != nil { + t.Fatalf("read final held byte + chunk: %v", err) + } + read = append(read, tail...) + + _ = stdoutW.Close() + _ = host.Close() + SetBridgeDown(false) + + waitDone := make(chan struct{}) + go func() { + pr.Wait() + close(waitDone) + }() + select { + case <-waitDone: + case <-time.After(5 * time.Second): + t.Fatal("Wait did not return after process exit") + } + + // The concatenation of everything the successive host conns received must + // reconstruct everything written, contiguously: no relay-level gap, + // duplication, or reorder across the repeated migrations, and no held-buffer + // state leaking from one migration into the next. + if !bytes.Equal(read, written) { + d := firstDiff(read, written) + t.Fatalf("relayed stream not contiguous across %d migrations: read %d bytes want %d, first diff at %d: got %q want %q", + cycles, len(read), len(written), d, window(read, d), window(written, d)) + } +} + +// TestPipeRelayWaitRaceWithTeardown drives Wait concurrently with the relay +// manager goroutine's teardown so Wait's CloseRead on the stdin conn overlaps +// run's Close of the same ConnectionSet. Its whole value is under -race -count: +// before the fix, Wait read ConnectionSet.In outside the relay mutex while the +// manager goroutine nil'd and closed it outside the mutex (a torn read / +// nil-deref that panics the GCS goroutine = UVM DoS); the fix moves both the +// CloseRead and the conn Close under the relay mutex. The test asserts only that +// nothing panics and that Wait returns. +func TestPipeRelayWaitRaceWithTeardown(t *testing.T) { + SetBridgeDown(false) + defer SetBridgeDown(false) + + // In is a closeSignalConn so Wait's read of ConnectionSet.In and the manager + // goroutine's write of it race on the field with no conn-level happens-before + // edge to mask it (see closeSignalConn). The stdin copier blocks on the conn's + // Read until the test closes stdinUnblock, ending it independently of Wait so + // the teardown that follows runs concurrently with Wait's CloseRead. Out is a + // fakeConn over net.Pipe whose copier ends on the stdout pipe EOF; there is no + // redial closure, so a copier ending means process exit (run tears down). + stdinUnblock := make(chan struct{}) + outHost, outGuest := net.Pipe() + set := &ConnectionSet{In: &closeSignalConn{unblock: stdinUnblock}, Out: &fakeConn{Conn: outGuest}} + + pr, err := NewPipeRelay(nil) + if err != nil { + t.Fatalf("NewPipeRelay: %v", err) + } + pr.ReplaceConnectionSet(set) + pr.CloseUnusedPipes() + // Capture the stdout write pipe before Start so closing it (to end the + // stdout copier) never races the manager's closePipes on pr.pipes. + stdoutW := pr.pipes[3] + pr.Start() + + // Release Wait and the teardown trigger together. The trigger ends both + // copiers WITHOUT Wait's help (stdinUnblock gives the stdin copier EOF; the + // stdout pipe close gives the stdout copier EOF), so run's teardown Close of + // the set runs genuinely concurrently with Wait's CloseRead / read of the same + // ConnectionSet.In rather than after it. That overlap is what -race needs to + // observe the unsynchronized field accesses the fix removes. + start := make(chan struct{}) + waitDone := make(chan struct{}) + go func() { + <-start + pr.Wait() + close(waitDone) + }() + + close(start) + close(stdinUnblock) // EOF to the stdin copier (independent of Wait) + _ = stdoutW.Close() // EOF to the stdout copier + + select { + case <-waitDone: + case <-time.After(10 * time.Second): + t.Fatal("Wait did not return: relay teardown overlapping Wait deadlocked") + } + + // The relay closed the guest side of Out during teardown; close the host side + // so the test owns its cleanup (double close is harmless). + _ = outHost.Close() +} diff --git a/internal/guest/stdio/stdio.go b/internal/guest/stdio/stdio.go index 9352bc58fc..a230a68c1f 100644 --- a/internal/guest/stdio/stdio.go +++ b/internal/guest/stdio/stdio.go @@ -4,10 +4,10 @@ package stdio import ( - "io" "os" "strings" "sync" + "sync/atomic" "github.com/Microsoft/hcsshim/internal/guest/transport" "github.com/pkg/errors" @@ -18,6 +18,10 @@ import ( // implementation should forward a process's stdio through. type ConnectionSet struct { In, Out, Err transport.Connection + // redial, when non-nil, re-establishes a fresh ConnectionSet over the same + // vsock ports. The relay manager goroutine uses it to recover the process + // stdio after a live-migration bridge drop. + redial func() (*ConnectionSet, error) } // Close closes each stdio connection. @@ -138,10 +142,19 @@ func NewPipeRelay(s *ConnectionSet) (_ *PipeRelay, err error) { // PipeRelay is a relay built to expose a pipe interface // for stdin, stdout, stderr on top of a ConnectionSet. type PipeRelay struct { - wg sync.WaitGroup + // mu guards s, which the relay manager goroutine swaps when it re-dials the + // stdio connections after a bridge drop. + mu sync.Mutex s *ConnectionSet // pipes format is stdin [0 read, 1 write], stdout [2 read, 3 write], stderr [4 read, 5 write]. pipes [6]*os.File + // pauseOnError is true when s carries a redial closure, meaning a copy error + // caused by a bridge drop should pause and resume rather than tear down the + // process stdio. + pauseOnError bool + // done is closed by the relay manager goroutine once the relay is truly + // finished (process exit), so Wait can block until then across any pauses. + done chan struct{} } // ReplaceConnectionSet allows the caller to add a new destination set after @@ -166,96 +179,156 @@ func (pr *PipeRelay) Files() (*FileSet, error) { return fs, nil } -func copyAndCleanClose(c transport.Connection, r io.Reader, name string) { - if n, err := io.Copy(c, r); err != nil { - logrus.WithFields(logrus.Fields{ - logrus.ErrorKey: err, - "bytes": n, - "file": name, - }).Error("opengcs::PipeRelay::copyAndCleanClose - error copying from pipe") - } - // Shut down the write end of the socket, then read a byte (which should - // yield EOF) to wait for the other endpoint to finish reading and close - // the connection. - if err := c.CloseWrite(); err == nil { - var b [1]byte - _, err = c.Read(b[:]) - if err == nil { - err = errors.New("unexpected data in socket") +// Start starts the relay operation. The caller must call Wait to wait +// for the relay to finish and release the associated resources. +func (pr *PipeRelay) Start() { + pr.pauseOnError = pr.s != nil && pr.s.redial != nil + pr.done = make(chan struct{}) + go pr.run() +} + +// run manages the relay copiers across live-migration pauses. It runs the +// copiers, and if they exit because the bridge dropped, re-dials the stdio +// connections and restarts the copiers so the process stdio survives the +// migration. When the copiers finish for a non-pause reason (process exit) it +// tears down the pipes and connections and signals done. +func (pr *PipeRelay) run() { + // outPending and errPending carry the in-flight bytes a copier read from + // the process pipe but had not yet written to the host conn when the bridge + // dropped, so the next iteration replays them on the re-dialed conn. + var outPending, errPending []byte + for { + var paused bool + paused, outPending, errPending = pr.runCopiers(outPending, errPending) + if !paused { + break } - if err != io.EOF { //nolint:errorlint - logrus.WithFields(logrus.Fields{ - logrus.ErrorKey: err, - "file": name, - }).Error("opengcs::PipeRelay::copyAndCleanClose - error reading for clean close") + pr.mu.Lock() + redial := pr.s.redial + pr.mu.Unlock() + ns, err := redialWithRetry(redial) + if err != nil { + logrus.WithError(err).Error("opengcs::PipeRelay::run - redial failed; ending relay") + break } - } else { - logrus.WithFields(logrus.Fields{ - logrus.ErrorKey: err, - "file": name, - }).Error("opengcs::PipeRelay::copyAndCleanClose - error shutting down socket") + // Swap in the re-dialed set and close the old one all under the lock so + // the conn Close cannot race Wait's CloseRead on the same conn. The dial + // above stays outside the lock because it can block for seconds. + pr.mu.Lock() + old := pr.s + pr.s = ns + old.Close() + pr.mu.Unlock() } - if err := c.Close(); err != nil { - logrus.WithFields(logrus.Fields{ - logrus.ErrorKey: err, - "file": name, - }).Error("opengcs::PipeRelay::copyAndCleanClose - error closing socket") + pr.closePipes() + // Close the live set under the lock so the conn Close cannot race Wait's + // CloseRead; closePipes stays outside since it only touches the pipes. + pr.mu.Lock() + s := pr.s + pr.s = nil + if s != nil { + s.Close() } + pr.mu.Unlock() + close(pr.done) } -// Start starts the relay operation. The caller must call Wait to wait -// for the relay to finish and release the associated resources. -func (pr *PipeRelay) Start() { - if pr.s.In != nil { - pr.wg.Add(1) +// runCopiers launches one copier per present stdio stream and waits for them +// all to finish. outPending/errPending are the in-flight output remainders held +// from a prior bridge-drop pause; each is replayed on the (re-dialed) conn +// before fresh output. It returns true if any copier paused because of a bridge +// drop (a live migration), along with the new held remainders for stdout and +// stderr so the caller can thread them into the next iteration. +func (pr *PipeRelay) runCopiers(outPending, errPending []byte) (paused bool, outPend, errPend []byte) { + pr.mu.Lock() + s := pr.s + pr.mu.Unlock() + + var cwg sync.WaitGroup + var pausedFlag atomic.Bool + + if s.In != nil && pr.pipes[1] != nil { + in := s.In + cwg.Add(1) go func() { - if n, err := io.Copy(pr.pipes[1], pr.s.In); err != nil { - logrus.WithFields(logrus.Fields{ - logrus.ErrorKey: err, - "bytes": n, - }).Error("opengcs::PipeRelay::Start - error copying stdin to pipe") + defer cwg.Done() + if copyIn(pr.pipes[1], in, "stdin", pr.pauseOnError) { + // Live migration pause: leave the stdin write pipe open so the + // process keeps its stdin across the redial. + pausedFlag.Store(true) + return } - if err := pr.pipes[1].Close(); err != nil { - logrus.WithFields(logrus.Fields{ - logrus.ErrorKey: err, - }).Error("opengcs::PipeRelay::Start - error closing stdin write pipe") + // Normal end (host closed stdin, or the process closed its stdin): + // close the stdin write pipe so the process observes EOF, matching + // the original relay behavior. + if cerr := pr.pipes[1].Close(); cerr != nil { + logrus.WithError(cerr).Error("opengcs::PipeRelay::runCopiers - error closing stdin write pipe") } pr.pipes[1] = nil - pr.wg.Done() }() } - if pr.s.Out != nil { - pr.wg.Add(1) + if s.Out != nil { + out := s.Out + cwg.Add(1) go func() { - copyAndCleanClose(pr.s.Out, pr.pipes[2], "stdout") - pr.wg.Done() + defer cwg.Done() + held, p := copyOut(out, pr.pipes[2], "stdout", outPending, pr.pauseOnError) + outPend = held + if p { + pausedFlag.Store(true) + } }() } - if pr.s.Err != nil { - pr.wg.Add(1) + if s.Err != nil { + errc := s.Err + cwg.Add(1) go func() { - copyAndCleanClose(pr.s.Err, pr.pipes[4], "stderr") - pr.wg.Done() + defer cwg.Done() + held, p := copyOut(errc, pr.pipes[4], "stderr", errPending, pr.pauseOnError) + errPend = held + if p { + pausedFlag.Store(true) + } }() } + cwg.Wait() + return pausedFlag.Load(), outPend, errPend } // Wait waits for the relaying to finish and closes the associated // pipes and connections. func (pr *PipeRelay) Wait() { + // Snapshot the set, close stdin's read side, and snapshot done all under the + // lock so the CloseRead cannot race the relay manager goroutine's Close of + // the same ConnectionSet (the redial swap or the final teardown). + pr.mu.Lock() + s := pr.s // Close stdin so that the copying goroutine is safely unblocked; this is necessary // because the host expects stdin to be closed before it will report process // exit back to the client, and the client expects the process notification before - // it will close its side of stdin (which io.Copy is waiting on in the copying goroutine). - if pr.s != nil && pr.s.In != nil { - _ = pr.s.In.CloseRead() + // it will close its side of stdin (which the input copier is blocked reading). + if s != nil && s.In != nil { + _ = s.In.CloseRead() } + done := pr.done + pr.mu.Unlock() - pr.wg.Wait() - pr.closePipes() - if pr.s != nil { - pr.s.Close() + if done == nil { + // Start was never called (no stdio requested); tear down synchronously + // to match the original no-op relay behavior. Close the set under the + // lock for consistency with the relay manager goroutine's teardown. + pr.closePipes() + pr.mu.Lock() + s = pr.s + pr.s = nil + if s != nil { + s.Close() + } + pr.mu.Unlock() + return } + + <-done } // CloseUnusedPipes gives the caller the ability to close any pipes that do not @@ -303,11 +376,18 @@ func NewTtyRelay(s *ConnectionSet, pty *os.File) *TtyRelay { // TtyRelay relays IO between a set of stdio connections and a master PTY file. type TtyRelay struct { + // m guards closed, the pty teardown, and s (which the relay manager goroutine + // swaps when it re-dials the stdio connections after a bridge drop). m sync.Mutex closed bool - wg sync.WaitGroup s *ConnectionSet pty *os.File + // pauseOnError is true when s carries a redial closure, meaning a copy error + // caused by a bridge drop should pause and resume. + pauseOnError bool + // done is closed once the relay is truly finished, so Wait can block across + // any live-migration pauses. + done chan struct{} } // ReplaceConnectionSet allows the caller to add a new destination set after @@ -330,50 +410,127 @@ func (r *TtyRelay) ResizeConsole(height, width uint16) error { // Start starts the relay operation. The caller must call Wait to wait // for the relay to finish and release the associated resources. func (r *TtyRelay) Start() { - if r.s.In != nil { - r.wg.Add(1) + r.pauseOnError = r.s != nil && r.s.redial != nil + r.done = make(chan struct{}) + go r.run() +} + +// run manages the TTY relay copiers across live-migration pauses, re-dialing +// and restarting them when the bridge drops, and closing the pty and +// connections once the copiers finish for a non-pause reason (process exit). +func (r *TtyRelay) run() { + // outPending carries the in-flight pty output bytes read but not yet + // written to the host conn when the bridge dropped, replayed next iteration. + var outPending []byte + for { + var paused bool + paused, outPending = r.runCopiers(outPending) + if !paused { + break + } + r.m.Lock() + redial := r.s.redial + r.m.Unlock() + ns, err := redialWithRetry(redial) + if err != nil { + logrus.WithError(err).Error("opengcs::TtyRelay::run - redial failed; ending relay") + break + } + // Swap in the re-dialed set and close the old one all under the lock so + // the conn Close cannot race Wait's CloseRead on the same conn. The dial + // above stays outside the lock because it can block for seconds. + r.m.Lock() + old := r.s + r.s = ns + old.Close() + r.m.Unlock() + } + // Close the pty and the live set under the lock so the conn Close cannot race + // Wait's CloseRead. + r.m.Lock() + r.pty.Close() + r.closed = true + s := r.s + r.s = nil + if s != nil { + s.Close() + } + r.m.Unlock() + close(r.done) +} + +// runCopiers launches the stdin->pty and pty->stdout copiers and waits for them +// to finish. outPending is the in-flight pty output remainder held from a prior +// bridge-drop pause, replayed on the (re-dialed) conn before fresh output. It +// returns true if a copier paused because the bridge dropped during a live +// migration, along with the new held output remainder for the next iteration. +func (r *TtyRelay) runCopiers(outPending []byte) (paused bool, outPend []byte) { + r.m.Lock() + s := r.s + r.m.Unlock() + + var cwg sync.WaitGroup + var pausedFlag atomic.Bool + + if s.In != nil { + in := s.In + cwg.Add(1) go func() { - if _, err := io.Copy(r.pty, r.s.In); err != nil { - logrus.WithFields(logrus.Fields{ - logrus.ErrorKey: err, - }).Error("opengcs::TtyRelay::Start - error copying stdin to pty") + defer cwg.Done() + if copyIn(r.pty, in, "stdin", r.pauseOnError) { + pausedFlag.Store(true) } - r.wg.Done() }() } - if r.s.Out != nil { - r.wg.Add(1) + if s.Out != nil { + out := s.Out + cwg.Add(1) go func() { - if _, err := io.Copy(r.s.Out, r.pty); err != nil { - logrus.WithFields(logrus.Fields{ - logrus.ErrorKey: err, - }).Error("opengcs::TtyRelay::Start - error copying pty to stdout") + defer cwg.Done() + held, p := copyOut(out, r.pty, "stdout", outPending, r.pauseOnError) + outPend = held + if p { + pausedFlag.Store(true) } - r.wg.Done() }() } + cwg.Wait() + return pausedFlag.Load(), outPend } // Wait waits for the relaying to finish and closes the associated // files and connections. func (r *TtyRelay) Wait() { + // Snapshot the set, close stdin's read side, and snapshot done all under the + // lock so the CloseRead cannot race the relay manager goroutine's Close of + // the same ConnectionSet (the redial swap or the final teardown). + r.m.Lock() + s := r.s // Close stdin so that the copying goroutine is safely unblocked; this is necessary // because the host expects stdin to be closed before it will report process // exit back to the client, and the client expects the process notification before - // it will close its side of stdin (which io.Copy is waiting on in the copying goroutine). - if r.s != nil && r.s.In != nil { - _ = r.s.In.CloseRead() + // it will close its side of stdin (which the input copier is blocked reading). + if s != nil && s.In != nil { + _ = s.In.CloseRead() } + done := r.done + r.m.Unlock() - // Wait for all users of stdioSet and master to finish before closing them. - r.wg.Wait() - - r.m.Lock() - defer r.m.Unlock() - - r.pty.Close() - r.closed = true - if r.s != nil { - r.s.Close() + if done == nil { + // Start was never called; tear down synchronously to match the original + // no-op relay behavior. Close the set under the lock for consistency with + // the relay manager goroutine's teardown. + r.m.Lock() + r.pty.Close() + r.closed = true + s = r.s + r.s = nil + if s != nil { + s.Close() + } + r.m.Unlock() + return } + + <-done } From 69a87e7b5f5ebdf6c2bca4e0bbfa8c29ee58b22f Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti Date: Fri, 26 Jun 2026 06:36:58 +0000 Subject: [PATCH 2/4] guest/stdio: note why writeAll counts n before the error check The io.Writer contract returns n as the bytes written even on a partial-write error, so counting it before the error check lets copyOut replay only the unwritten tail and never re-send those bytes. Addresses a review question. Signed-off-by: Shreyansh Sancheti --- internal/guest/stdio/pauserelay.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/internal/guest/stdio/pauserelay.go b/internal/guest/stdio/pauserelay.go index 4e2e73b69b..af124afcbe 100644 --- a/internal/guest/stdio/pauserelay.go +++ b/internal/guest/stdio/pauserelay.go @@ -70,6 +70,9 @@ func writeAll(w io.Writer, p []byte) (int, error) { written := 0 for written < len(p) { n, err := w.Write(p[written:]) + // Count n before the error check: per the io.Writer contract these n + // bytes were written even on a partial-write error, so the caller + // replays only p[written:] and never re-sends them. written += n if err != nil { return written, err From fe87c61c0080535aa06b8ba23cfe51a8e1ceea06 Mon Sep 17 00:00:00 2001 From: Harsh Rawat Date: Mon, 20 Jul 2026 03:58:39 +0530 Subject: [PATCH 3/4] refactor stdio connection handling to simplify bridge state management Signed-off-by: Harsh Rawat --- cmd/gcs/main.go | 11 ---- internal/guest/stdio/connection.go | 15 ++--- internal/guest/stdio/pauserelay.go | 25 +------ internal/guest/stdio/pauserelay_test.go | 87 ++++++++++++++----------- 4 files changed, 57 insertions(+), 81 deletions(-) diff --git a/cmd/gcs/main.go b/cmd/gcs/main.go index 91ff2d6e1e..fd9575ec3b 100644 --- a/cmd/gcs/main.go +++ b/cmd/gcs/main.go @@ -27,7 +27,6 @@ import ( "github.com/Microsoft/hcsshim/internal/guest/prot" "github.com/Microsoft/hcsshim/internal/guest/runtime/hcsv2" "github.com/Microsoft/hcsshim/internal/guest/runtime/runc" - "github.com/Microsoft/hcsshim/internal/guest/stdio" "github.com/Microsoft/hcsshim/internal/guest/transport" "github.com/Microsoft/hcsshim/internal/log" "github.com/Microsoft/hcsshim/internal/oc" @@ -450,10 +449,6 @@ func main() { bridgeOut = bridgeCon } - // The bridge is up again; let the stdio relays resume any copiers that - // paused when the previous bridge dropped during a live migration. - stdio.SetBridgeDown(false) - logrus.Info("bridge connected, serving") serveErr := b.ListenAndServe(bridgeIn, bridgeOut) @@ -463,12 +458,6 @@ func main() { break } - // The bridge dropped (for example during a live migration). Tell the - // stdio relays to pause rather than tear down so each process's stdio - // survives the reconnect; the relays re-dial and resume once the bridge - // is back and SetBridgeDown(false) is set above. - stdio.SetBridgeDown(true) - logrus.WithError(serveErr).Warn("bridge connection lost, will reconnect") time.Sleep(reconnectInterval) } diff --git a/internal/guest/stdio/connection.go b/internal/guest/stdio/connection.go index 16847d5eda..814ce38ea6 100644 --- a/internal/guest/stdio/connection.go +++ b/internal/guest/stdio/connection.go @@ -30,28 +30,25 @@ func Connect(tport transport.Transport, settings ConnectionSettings) (_ *Connect } }() if settings.StdIn != nil { - port := *settings.StdIn - c, err := tport.Dial(port) + c, err := tport.Dial(*settings.StdIn) if err != nil { return nil, errors.Wrap(err, "failed creating stdin Connection") } - connSet.In = transport.NewLogConnection(c, port) + connSet.In = transport.NewLogConnection(c, *settings.StdIn) } if settings.StdOut != nil { - port := *settings.StdOut - c, err := tport.Dial(port) + c, err := tport.Dial(*settings.StdOut) if err != nil { return nil, errors.Wrap(err, "failed creating stdout Connection") } - connSet.Out = transport.NewLogConnection(c, port) + connSet.Out = transport.NewLogConnection(c, *settings.StdOut) } if settings.StdErr != nil { - port := *settings.StdErr - c, err := tport.Dial(port) + c, err := tport.Dial(*settings.StdErr) if err != nil { return nil, errors.Wrap(err, "failed creating stderr Connection") } - connSet.Err = transport.NewLogConnection(c, port) + connSet.Err = transport.NewLogConnection(c, *settings.StdErr) } // redial re-establishes a fresh ConnectionSet over the same vsock ports // after a bridge drop, so the stdio relays can pause and resume across a diff --git a/internal/guest/stdio/pauserelay.go b/internal/guest/stdio/pauserelay.go index af124afcbe..806959380d 100644 --- a/internal/guest/stdio/pauserelay.go +++ b/internal/guest/stdio/pauserelay.go @@ -5,7 +5,6 @@ package stdio import ( "io" - "sync/atomic" "time" "github.com/Microsoft/hcsshim/internal/guest/transport" @@ -13,20 +12,6 @@ import ( "github.com/sirupsen/logrus" ) -// bridgeDown reports whether the GCS bridge to the host is currently down (for -// example because a live migration is tearing down and re-establishing the -// vsock connections). The stdio relays consult it to tell a live-migration -// disconnect (pause and resume) apart from a normal process-driven copy error -// (tear down). It is set by cmd/gcs around the bridge reconnect loop. -var bridgeDown atomic.Bool - -// SetBridgeDown records whether the host bridge is currently down. cmd/gcs sets -// it true when ListenAndServe returns (and it is not a shutdown) and false once -// the bridge has reconnected. -func SetBridgeDown(v bool) { - bridgeDown.Store(v) -} - const ( // redialInterval is the delay between attempts to re-establish the stdio // connections after a bridge drop. @@ -116,9 +101,7 @@ func copyIn(w io.Writer, r io.Reader, name string, pauseOnError bool) (paused bo if rerr == io.EOF { return false } - if pauseOnError && bridgeDown.Load() { - // A conn read error while the bridge is down is the - // live-migration drop: pause so the manager re-dials. + if pauseOnError { return true } l.WithError(rerr).Error("opengcs::stdio::copyIn - error reading input") @@ -147,8 +130,7 @@ func copyOut(c transport.Connection, r io.Reader, name string, pending []byte, p // bytes already read from the pipe before the bridge dropped are not lost. if len(pending) > 0 { if n, err := writeAll(c, pending); err != nil { - if pauseOnError && bridgeDown.Load() { - l.WithError(err).Info("opengcs::stdio::copyOut - pausing on bridge down") + if pauseOnError { h := make([]byte, len(pending)-n) copy(h, pending[n:]) return h, true @@ -163,8 +145,7 @@ func copyOut(c transport.Connection, r io.Reader, name string, pending []byte, p nr, rerr := r.Read(buf) if nr > 0 { if nw, werr := writeAll(c, buf[:nr]); werr != nil { - if pauseOnError && bridgeDown.Load() { - l.WithError(werr).Info("opengcs::stdio::copyOut - pausing on bridge down") + if pauseOnError { h := make([]byte, nr-nw) copy(h, buf[nw:nr]) return h, true diff --git a/internal/guest/stdio/pauserelay_test.go b/internal/guest/stdio/pauserelay_test.go index 17300ac882..8204fefa7f 100644 --- a/internal/guest/stdio/pauserelay_test.go +++ b/internal/guest/stdio/pauserelay_test.go @@ -64,9 +64,6 @@ var _ transport.Connection = (*closeSignalConn)(nil) // copying onto the new connection, and only completes Wait once the process // stdio is truly finished. func TestPipeRelayPauseResume(t *testing.T) { - SetBridgeDown(false) - defer SetBridgeDown(false) - // Each redial hands the test the new host end of a fresh net.Pipe so it can // read what the resumed relay writes. redialedHosts := make(chan net.Conn, 4) @@ -105,11 +102,10 @@ func TestPipeRelayPauseResume(t *testing.T) { t.Fatalf("pre-pause got %q, want hello", got) } - // Simulate a bridge drop: mark the bridge down, close the host stdio conn, - // then nudge the copier so its next write hits the closed conn and pauses. - // The "x" is read from the pipe but cannot be written to the dead conn, so - // the relay must hold it and replay it on the re-dialed conn (no drop). - SetBridgeDown(true) + // Simulate a bridge drop: close the host stdio conn so the copier's next + // write hits the closed conn and pauses. The "x" is read from the pipe but + // cannot be written to the dead conn, so the relay must hold it and replay + // it on the re-dialed conn (no drop). _ = out1.Close() if _, err := stdoutW.Write([]byte("x")); err != nil { t.Fatalf("write pause trigger: %v", err) @@ -142,7 +138,6 @@ func TestPipeRelayPauseResume(t *testing.T) { // finishes and Wait returns. _ = stdoutW.Close() _ = host2.Close() - SetBridgeDown(false) waitDone := make(chan struct{}) go func() { @@ -337,9 +332,6 @@ func TestIoCopyDropsInFlightBytes(t *testing.T) { // (see TestIoCopyDropsInFlightBytes) and replays it intact, in order, on the // re-dialed connection. func TestCopyOutRetainsInFlightBytes(t *testing.T) { - SetBridgeDown(true) - defer SetBridgeDown(false) - // > one read but < relayBufferSize, so the payload is a single in-flight // chunk held whole on the failed write. payload := testPattern(5000) @@ -405,9 +397,6 @@ func TestCopyOutRetainsInFlightBytes(t *testing.T) { // re-dialed conn (bridge still down) must be re-held, not dropped or truncated, // and is replayed once a working conn appears. func TestCopyOutRetainsAcrossDoublePause(t *testing.T) { - SetBridgeDown(true) - defer SetBridgeDown(false) - payload := testPattern(5000) // First re-dial still bridge-down: the replay of the held remainder fails, @@ -498,9 +487,6 @@ func TestWriteAllShortWriteGuard(t *testing.T) { // reordering. This is the end-to-end proof that the in-flight bytes io.Copy // dropped are now retained and replayed. func TestPipeRelayRetainsBurstAcrossPause(t *testing.T) { - SetBridgeDown(false) - defer SetBridgeDown(false) - const ( burstLen = 16 * 1024 splitAt = 8 * 1024 @@ -548,10 +534,9 @@ func TestPipeRelayRetainsBurstAcrossPause(t *testing.T) { t.Fatalf("read first half on original host: %v", err) } - // Drop the bridge: mark it down, then close the host conn so the relay's - // blocked write fails. copyOut must hold the undelivered remainder and replay - // it on the re-dialed conn rather than drop it (the old io.Copy bug). - SetBridgeDown(true) + // Drop the bridge: close the host conn so the relay's blocked write fails. + // copyOut must hold the undelivered remainder and replay it on the re-dialed + // conn rather than drop it (the old io.Copy bug). _ = out1.Close() var host2 net.Conn @@ -590,7 +575,6 @@ func TestPipeRelayRetainsBurstAcrossPause(t *testing.T) { // Process exit: close stdout and the host conn so the relay finishes. _ = stdoutW.Close() _ = host2.Close() - SetBridgeDown(false) waitDone := make(chan struct{}) go func() { @@ -611,9 +595,6 @@ func TestPipeRelayRetainsBurstAcrossPause(t *testing.T) { // accepted bytes are socket-level loss the relay cannot recover, and only // payload[K:] is retained for replay on the re-dialed conn. func TestCopyOutHoldsPartialWriteRemainder(t *testing.T) { - SetBridgeDown(true) - defer SetBridgeDown(false) - // A single in-flight chunk (< relayBufferSize) so copyOut issues exactly one // Write; the conn takes k bytes then fails. const k = 1500 @@ -686,9 +667,6 @@ func (r errReader) Read(p []byte) (int, error) { return 0, r.err } // Together they prove copyIn decides by which side failed, not by a single // collapsed error as the old io.Copy did. func TestCopyInFinishesOnStdinCloseWhileBridgeDown(t *testing.T) { - SetBridgeDown(true) - defer SetBridgeDown(false) - // Process closed stdin: the reader yields bytes (so a write is attempted) // then blocks, and the sink's Write fails (EPIPE). copyIn must return via the // write side as false (done); the block guarantees no read error or EOF can @@ -728,9 +706,6 @@ func TestCopyInFinishesOnStdinCloseWhileBridgeDown(t *testing.T) { // held-buffer state leaks from one migration into the next. Synchronization is // net.Pipe's blocking reads plus the redial handoff channel; there are no sleeps. func TestPipeRelaySurvivesMultipleMigrations(t *testing.T) { - SetBridgeDown(false) - defer SetBridgeDown(false) - const cycles = 4 // > 3 back-and-forth migrations // Carve every chunk from one position-dependent pattern so the concatenation @@ -793,10 +768,9 @@ func TestPipeRelaySurvivesMultipleMigrations(t *testing.T) { } read = append(read, buf...) - // Drop the bridge and close this host so the relay's next write fails and + // Drop the bridge: close this host so the relay's next write fails and // pauses. The single nudge byte is read from the pipe but cannot be // delivered, so the relay must hold it and replay it on the next conn. - SetBridgeDown(true) _ = host.Close() nudge := take(1) if _, err := stdoutW.Write(nudge); err != nil { @@ -809,7 +783,6 @@ func TestPipeRelaySurvivesMultipleMigrations(t *testing.T) { case <-time.After(5 * time.Second): t.Fatalf("cycle %d: relay did not redial after pause", i) } - SetBridgeDown(false) } // Final connection: read the last held byte plus a final chunk, then drive a @@ -828,7 +801,6 @@ func TestPipeRelaySurvivesMultipleMigrations(t *testing.T) { _ = stdoutW.Close() _ = host.Close() - SetBridgeDown(false) waitDone := make(chan struct{}) go func() { @@ -861,9 +833,6 @@ func TestPipeRelaySurvivesMultipleMigrations(t *testing.T) { // CloseRead and the conn Close under the relay mutex. The test asserts only that // nothing panics and that Wait returns. func TestPipeRelayWaitRaceWithTeardown(t *testing.T) { - SetBridgeDown(false) - defer SetBridgeDown(false) - // In is a closeSignalConn so Wait's read of ConnectionSet.In and the manager // goroutine's write of it race on the field with no conn-level happens-before // edge to mask it (see closeSignalConn). The stdin copier blocks on the conn's @@ -914,3 +883,43 @@ func TestPipeRelayWaitRaceWithTeardown(t *testing.T) { // so the test owns its cleanup (double close is harmless). _ = outHost.Close() } + +// fakeTransport is a transport.Transport whose Dial returns an inert recordConn, +// for exercising Connect. +type fakeTransport struct{} + +func (fakeTransport) Dial(uint32) (transport.Connection, error) { return &recordConn{}, nil } + +// TestConnectPopulatesRedial asserts Connect always wires the redial closure, the +// invariant the relay relies on to recover stdio across a migration. +func TestConnectPopulatesRedial(t *testing.T) { + port := uint32(1) + set, err := Connect(fakeTransport{}, ConnectionSettings{StdOut: &port}) + if err != nil { + t.Fatalf("Connect: %v", err) + } + if set.redial == nil { + t.Fatal("Connect left redial nil; the relay cannot recover stdio across a migration") + } +} + +// TestRedialWithRetrySucceedsAfterRetries verifies redialWithRetry keeps retrying +// a failing redial (the process stdio ports return after the migration) and +// resumes once one succeeds. +func TestRedialWithRetrySucceedsAfterRetries(t *testing.T) { + calls := 0 + redial := func() (*ConnectionSet, error) { + calls++ + if calls < 3 { + return nil, errors.New("redial failed") + } + return &ConnectionSet{}, nil + } + ns, err := redialWithRetry(redial) + if err != nil || ns == nil { + t.Fatalf("redialWithRetry err = %v ns = %v, want success", err, ns) + } + if calls != 3 { + t.Fatalf("redialWithRetry made %d attempts, want 3", calls) + } +} From 93e5130a698dba6edb82c1b37a7e1ffd36e19f0b Mon Sep 17 00:00:00 2001 From: Shreyansh Sancheti Date: Mon, 20 Jul 2026 08:28:04 +0000 Subject: [PATCH 4/4] guest/stdio: simplify relay locking per review Collapse the repeated connection swap and teardown lock blocks into a small setConn helper (plus a teardown helper for the tty relay), and drop the snapshot-under-lock-then-use-outside reads in run and runCopiers since the manager goroutine is the sole writer of the connection set. Same serialization of the conn close against the Wait CloseRead, fewer lock sites. Also drop a stale bridgeDown mention left in a comment. Signed-off-by: Shreyansh Sancheti --- internal/guest/stdio/pauserelay.go | 2 +- internal/guest/stdio/stdio.go | 128 ++++++++++++++--------------- 2 files changed, 61 insertions(+), 69 deletions(-) diff --git a/internal/guest/stdio/pauserelay.go b/internal/guest/stdio/pauserelay.go index 806959380d..12554848e8 100644 --- a/internal/guest/stdio/pauserelay.go +++ b/internal/guest/stdio/pauserelay.go @@ -76,7 +76,7 @@ func writeAll(w io.Writer, p []byte) (int, error) { // copyIn copies host input from conn r into the process sink w (a stdin pipe or // a pty master). It decides by which side failed, mirroring copyOut: a read // error on r is the host conn, so a bridge drop during a live migration -// (pauseOnError and bridgeDown) returns true (a pause) and the manager re-dials +// returns true (a pause) when pauseOnError is set and the manager re-dials // so the next call resumes reading from the fresh conn; a write error on w is // the process closing its stdin (EPIPE), which is a normal end and returns false // even while the bridge is down (otherwise a post-resume stdin close would be diff --git a/internal/guest/stdio/stdio.go b/internal/guest/stdio/stdio.go index a230a68c1f..5f11c2f55f 100644 --- a/internal/guest/stdio/stdio.go +++ b/internal/guest/stdio/stdio.go @@ -187,6 +187,20 @@ func (pr *PipeRelay) Start() { go pr.run() } +// setConn publishes ns as the current connection set (nil when tearing down) +// and closes the previous set. It holds mu across the swap and close so the +// close cannot race Wait's CloseRead on the same connection; the slow redial +// stays off the lock in the caller. +func (pr *PipeRelay) setConn(ns *ConnectionSet) { + pr.mu.Lock() + defer pr.mu.Unlock() + old := pr.s + pr.s = ns + if old != nil { + old.Close() + } +} + // run manages the relay copiers across live-migration pauses. It runs the // copiers, and if they exit because the bridge dropped, re-dials the stdio // connections and restarts the copiers so the process stdio survives the @@ -203,33 +217,17 @@ func (pr *PipeRelay) run() { if !paused { break } - pr.mu.Lock() - redial := pr.s.redial - pr.mu.Unlock() - ns, err := redialWithRetry(redial) + // run is the only writer of pr.s, so it reads the redial closure without + // mu; the dial can block for seconds and must stay off the lock. + ns, err := redialWithRetry(pr.s.redial) if err != nil { logrus.WithError(err).Error("opengcs::PipeRelay::run - redial failed; ending relay") break } - // Swap in the re-dialed set and close the old one all under the lock so - // the conn Close cannot race Wait's CloseRead on the same conn. The dial - // above stays outside the lock because it can block for seconds. - pr.mu.Lock() - old := pr.s - pr.s = ns - old.Close() - pr.mu.Unlock() + pr.setConn(ns) } pr.closePipes() - // Close the live set under the lock so the conn Close cannot race Wait's - // CloseRead; closePipes stays outside since it only touches the pipes. - pr.mu.Lock() - s := pr.s - pr.s = nil - if s != nil { - s.Close() - } - pr.mu.Unlock() + pr.setConn(nil) close(pr.done) } @@ -240,9 +238,9 @@ func (pr *PipeRelay) run() { // drop (a live migration), along with the new held remainders for stdout and // stderr so the caller can thread them into the next iteration. func (pr *PipeRelay) runCopiers(outPending, errPending []byte) (paused bool, outPend, errPend []byte) { - pr.mu.Lock() + // run is the only writer of pr.s and does not swap it until this returns, so + // reading it here without mu is safe. s := pr.s - pr.mu.Unlock() var cwg sync.WaitGroup var pausedFlag atomic.Bool @@ -315,16 +313,9 @@ func (pr *PipeRelay) Wait() { if done == nil { // Start was never called (no stdio requested); tear down synchronously - // to match the original no-op relay behavior. Close the set under the - // lock for consistency with the relay manager goroutine's teardown. + // to match the original no-op relay behavior. pr.closePipes() - pr.mu.Lock() - s = pr.s - pr.s = nil - if s != nil { - s.Close() - } - pr.mu.Unlock() + pr.setConn(nil) return } @@ -415,6 +406,34 @@ func (r *TtyRelay) Start() { go r.run() } +// setConn publishes ns as the current connection set and closes the previous +// set under m so the close cannot race Wait's CloseRead on the same connection. +func (r *TtyRelay) setConn(ns *ConnectionSet) { + r.m.Lock() + defer r.m.Unlock() + old := r.s + r.s = ns + if old != nil { + old.Close() + } +} + +// teardown closes the pty and the current connection set under m, which also +// guards ResizeConsole's use of the pty. +func (r *TtyRelay) teardown() { + r.m.Lock() + defer r.m.Unlock() + if r.closed { + return + } + r.pty.Close() + r.closed = true + if r.s != nil { + r.s.Close() + r.s = nil + } +} + // run manages the TTY relay copiers across live-migration pauses, re-dialing // and restarting them when the bridge drops, and closing the pty and // connections once the copiers finish for a non-pause reason (process exit). @@ -428,34 +447,16 @@ func (r *TtyRelay) run() { if !paused { break } - r.m.Lock() - redial := r.s.redial - r.m.Unlock() - ns, err := redialWithRetry(redial) + // run is the only writer of r.s, so it reads the redial closure without + // m; the dial can block for seconds and must stay off the lock. + ns, err := redialWithRetry(r.s.redial) if err != nil { logrus.WithError(err).Error("opengcs::TtyRelay::run - redial failed; ending relay") break } - // Swap in the re-dialed set and close the old one all under the lock so - // the conn Close cannot race Wait's CloseRead on the same conn. The dial - // above stays outside the lock because it can block for seconds. - r.m.Lock() - old := r.s - r.s = ns - old.Close() - r.m.Unlock() + r.setConn(ns) } - // Close the pty and the live set under the lock so the conn Close cannot race - // Wait's CloseRead. - r.m.Lock() - r.pty.Close() - r.closed = true - s := r.s - r.s = nil - if s != nil { - s.Close() - } - r.m.Unlock() + r.teardown() close(r.done) } @@ -465,9 +466,9 @@ func (r *TtyRelay) run() { // returns true if a copier paused because the bridge dropped during a live // migration, along with the new held output remainder for the next iteration. func (r *TtyRelay) runCopiers(outPending []byte) (paused bool, outPend []byte) { - r.m.Lock() + // run is the only writer of r.s and does not swap it until this returns, so + // reading it here without m is safe. s := r.s - r.m.Unlock() var cwg sync.WaitGroup var pausedFlag atomic.Bool @@ -518,17 +519,8 @@ func (r *TtyRelay) Wait() { if done == nil { // Start was never called; tear down synchronously to match the original - // no-op relay behavior. Close the set under the lock for consistency with - // the relay manager goroutine's teardown. - r.m.Lock() - r.pty.Close() - r.closed = true - s = r.s - r.s = nil - if s != nil { - s.Close() - } - r.m.Unlock() + // no-op relay behavior. + r.teardown() return }