Refactor portforward packages to eliminate storing context.Context

Refactor to avoid storing context.Context in portforward structures. Also,
reduce the number of places where copying between two connections is used.

PiperOrigin-RevId: 508695162

runsc/boot/BUILD

@@ -109,6 +109,7 @@ go_library(
         "//pkg/urpc",
         "//runsc/boot/filter",
         "//runsc/boot/platforms",
+        "//runsc/boot/portforward",
         "//runsc/boot/pprof",
         "//runsc/boot/procfs",
         "//runsc/config",

runsc/boot/controller.go

@@ -382,6 +382,29 @@ func (cm *containerManager) Checkpoint(o *control.SaveOpts, _ *struct{}) error {
 	return state.Save(o, nil)
 }
 
+// PortForwardOpts contains options for port forwarding to a port in a
+// container.
+type PortForwardOpts struct {
+	// FilePayload contains one fd for a UDS (or local port) used for port
+	// forwarding.
+	urpc.FilePayload
+
+	// ContainerID is the container for the process being executed.
+	ContainerID string
+	// Port is the port to to forward.
+	Port uint16
+}
+
+// PortForward initiates a port forward to the container.
+func (cm *containerManager) PortForward(opts *PortForwardOpts, _ *struct{}) error {
+	log.Debugf("containerManager.PortForward, cid: %s, port: %d", opts.ContainerID, opts.Port)
+	if err := cm.l.portForward(opts); err != nil {
+		log.Debugf("containerManager.PortForward failed, opts: %+v, err: %v", opts, err)
+		return err
+	}
+	return nil
+}
+
 // RestoreOpts contains options related to restoring a container's file system.
 type RestoreOpts struct {
 	// FilePayload contains the state file to be restored, followed by the

runsc/boot/loader.go

@@ -27,6 +27,7 @@ import (
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/abi/linux"
 	"gvisor.dev/gvisor/pkg/bpf"
+	"gvisor.dev/gvisor/pkg/cleanup"
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/coverage"
 	"gvisor.dev/gvisor/pkg/cpuid"
@@ -69,6 +70,7 @@ import (
 	"gvisor.dev/gvisor/pkg/tcpip/transport/udp"
 	"gvisor.dev/gvisor/runsc/boot/filter"
 	_ "gvisor.dev/gvisor/runsc/boot/platforms" // register all platforms.
+	pf "gvisor.dev/gvisor/runsc/boot/portforward"
 	"gvisor.dev/gvisor/runsc/boot/pprof"
 	"gvisor.dev/gvisor/runsc/config"
 	"gvisor.dev/gvisor/runsc/profile"
@@ -136,7 +138,15 @@ type Loader struct {
 	// sandboxID is the ID for the whole sandbox.
 	sandboxID string
 
-	// mu guards processes.
+	// mountHints provides extra information about mounts for containers that
+	// apply to the entire pod.
+	mountHints *podMountHints
+
+	// productName is the value to show in
+	// /sys/devices/virtual/dmi/id/product_name.
+	productName string
+
+	// mu guards processes and porForwardProxies.
 	mu sync.Mutex
 
 	// processes maps containers init process and invocation of exec. Root
@@ -146,13 +156,10 @@ type Loader struct {
 	// processes is guarded by mu.
 	processes map[execID]*execProcess
 
-	// mountHints provides extra information about mounts for containers that
-	// apply to the entire pod.
-	mountHints *podMountHints
-
-	// productName is the value to show in
-	// /sys/devices/virtual/dmi/id/product_name.
-	productName string
+	// portForwardProxies is a list of active port forwarding connections.
+	//
+	// portForwardProxies is guarded by mu.
+	portForwardProxies []*pf.Proxy
 }
 
 // execID uniquely identifies a sentry process that is executed in a container.
@@ -1387,3 +1394,100 @@ func createFDTable(ctx context.Context, console bool, stdioFDs []*fd.FD, user sp
 	}
 	return fdTable, ttyFile, nil
 }
+
+// portForward implements initiating a portForward connection in the sandbox. portForwardProxies
+// represent a two connections each copying to each other (read ends to write ends) in goroutines.
+// The proxies are stored and can be cleaned up, or clean up after themselves if the connection
+// is broken.
+func (l *Loader) portForward(opts *PortForwardOpts) error {
+	// Validate that we have a stream FD to write to. If this happens then
+	// it means there is a misbehaved urpc client or a bug has occurred.
+	if len(opts.Files) != 1 {
+		return fmt.Errorf("stream FD is required for port forward")
+	}
+
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	cid := opts.ContainerID
+	tg, err := l.tryThreadGroupFromIDLocked(execID{cid: cid})
+	if err != nil {
+		return fmt.Errorf("failed to get threadgroup from %q: %w", cid, err)
+	}
+	if tg == nil {
+		return fmt.Errorf("container %q not started", cid)
+	}
+
+	// Import the fd for the UDS.
+	ctx := l.k.SupervisorContext()
+	fd, err := l.importFD(ctx, opts.Files[0])
+	if err != nil {
+		return fmt.Errorf("importing stream fd: %w", err)
+	}
+	cu := cleanup.Make(func() { fd.DecRef(ctx) })
+	defer cu.Clean()
+
+	fdConn := pf.NewFileDescriptionConn(fd)
+
+	// Create a proxy to forward data between the fdConn and the sandboxed application.
+	pair := pf.ProxyPair{To: fdConn}
+
+	switch l.root.conf.Network {
+	case config.NetworkSandbox:
+		stack := l.k.RootNetworkNamespace().Stack().(*netstack.Stack).Stack
+		nsConn, err := pf.NewNetstackConn(stack, opts.Port)
+		if err != nil {
+			return fmt.Errorf("creating netstack port forward connection: %w", err)
+		}
+		pair.From = nsConn
+	case config.NetworkHost:
+		hConn, err := pf.NewHostInetConn(opts.Port)
+		if err != nil {
+			return fmt.Errorf("creating hostinet port forward connection: %w", err)
+		}
+		pair.From = hConn
+	default:
+		return fmt.Errorf("unsupported network type %q for container %q", l.root.conf.Network, cid)
+	}
+	cu.Release()
+	proxy := pf.NewProxy(pair, opts.ContainerID)
+
+	// Add to the list of port forward connections and remove when the
+	// connection closes.
+	l.portForwardProxies = append(l.portForwardProxies, proxy)
+	proxy.AddCleanup(func() {
+		l.mu.Lock()
+		defer l.mu.Unlock()
+		for i := range l.portForwardProxies {
+			if l.portForwardProxies[i] == proxy {
+				l.portForwardProxies = append(l.portForwardProxies[:i], l.portForwardProxies[i+1:]...)
+				break
+			}
+		}
+	})
+
+	// Start forwarding on the connection.
+	proxy.Start(ctx)
+	return nil
+}
+
+// importFD generically imports a host file descriptor without adding it to any
+// fd table.
+func (l *Loader) importFD(ctx context.Context, f *os.File) (*vfs.FileDescription, error) {
+	hostFD, err := fd.NewFromFile(f)
+	if err != nil {
+		return nil, err
+	}
+	defer hostFD.Close()
+	fd, err := host.NewFD(ctx, l.k.HostMount(), hostFD.FD(), &host.NewFDOptions{
+		Savable:      false, // We disconnect and close on save.
+		IsTTY:        false,
+		VirtualOwner: false, // FD not visible to the sandboxed app so user can't be changed.
+	})
+
+	if err != nil {
+		return nil, err
+	}
+	hostFD.Release()
+	return fd, nil
+}

runsc/boot/portforward/BUILD

@@ -11,15 +11,16 @@ go_library(
         "portforward_netstack.go",
         "portforward_test_util.go",
     ],
+    visibility = [
+        "//runsc:__subpackages__",
+    ],
     deps = [
         "//pkg/cleanup",
         "//pkg/context",
         "//pkg/errors/linuxerr",
         "//pkg/fd",
         "//pkg/fdnotifier",
-        "//pkg/log",
         "//pkg/sentry/vfs",
-        "//pkg/sync",
         "//pkg/tcpip",
         "//pkg/tcpip/network/ipv4",
         "//pkg/tcpip/stack",
@@ -39,13 +40,11 @@ go_test(
     ],
     library = ":portforward",
     tags = [
-        "manual",
         "requires-net:ipv4",
         "requires-net:loopback",
     ],
     deps = [
         "//pkg/abi/linux",
-        "//pkg/cleanup",
         "//pkg/context",
         "//pkg/errors/linuxerr",
         "//pkg/sentry/contexttest",

runsc/boot/portforward/portforward.go

@@ -16,16 +16,137 @@
 package portforward
 
 import (
+	"fmt"
+	"sync"
+
+	"gvisor.dev/gvisor/pkg/cleanup"
 	"gvisor.dev/gvisor/pkg/context"
 )
 
-// portForwardConn is a port forwarding connection. It is used to manage the
+// proxyConn is a port forwarding connection. It is used to manage the
 // lifecycle of the connection and clean it up if necessary.
-type portForwardConn interface {
-	// start starts the connection goroutines and returns.
-	start(ctx context.Context) error
-	// close closes and cleans up the connection.
-	close(ctx context.Context) error
-	// cleanup registers a callback for when the connection closes.
-	cleanup(func())
+type proxyConn interface {
+	// Name returns a name for this proxyConn.
+	Name() string
+	// Write performs a write on this connection.  Write should block on ErrWouldBlock, but it must
+	// listen to 'cancel' to interrupt blocked calls.
+	Write(ctx context.Context, buf []byte, cancel <-chan struct{}) (int, error)
+	// Read performs a read on this connection. Read should block on ErrWouldBlock by the underlying
+	// connection, but it must listen to `cancel` to interrupt blocked calls.
+	Read(ctx context.Context, buf []byte, cancel <-chan struct{}) (int, error)
+	// Close cleans up all resources owned by this proxyConn.
+	Close(ctx context.Context)
+}
+
+// Proxy controls copying data between two proxyConnections. Proxy takes ownership over the two
+// connections and is responsible for cleaning up their resources (i.e. calling their Close method).
+// Proxy(s) all run internal to the sandbox on the supervisor context.
+type Proxy struct {
+	// containerID for this proxy.
+	cid string
+	// "to" and "from" are the two connections on which this Proxy copies.
+	to         proxyConn
+	from       proxyConn
+	once       sync.Once
+	cancelFrom chan struct{}
+	cancelTo   chan struct{}
+	wg         sync.WaitGroup
+	cu         cleanup.Cleanup
+}
+
+// ProxyPair wraps the to/from arguments for NewProxy so that the user explicitly labels to/from.
+type ProxyPair struct {
+	To   proxyConn
+	From proxyConn
+}
+
+// NewProxy returns a new Proxy.
+func NewProxy(pair ProxyPair, cid string) *Proxy {
+	return &Proxy{
+		to:         pair.To,
+		from:       pair.From,
+		cid:        cid,
+		cancelTo:   make(chan struct{}, 1),
+		cancelFrom: make(chan struct{}, 1),
+	}
+}
+
+// readFrom reads from the application's vfs.FileDescription and writes to the shim.
+func (pf *Proxy) readFrom(ctx context.Context) error {
+	buf := make([]byte, 16384 /* 16kb buffer size */)
+	for ctx.Err() == nil {
+		if err := doCopy(ctx, pf.to, pf.from, buf, pf.cancelFrom); err != nil {
+			return fmt.Errorf("readFrom failed on container %q: %v", pf.cid, err)
+		}
+	}
+	return ctx.Err()
+}
+
+// writeTo writes to the application's vfs.FileDescription and reads from the shim.
+func (pf *Proxy) readTo(ctx context.Context) error {
+	buf := make([]byte, 16384 /* 16kb buffer size */)
+	for ctx.Err() == nil {
+		if err := doCopy(ctx, pf.from, pf.to, buf, pf.cancelTo); err != nil {
+			return fmt.Errorf("readTo failed on container %q: %v", pf.cid, err)
+		}
+	}
+	return ctx.Err()
+}
+
+// doCopy is the shared copy code for each of 'readFrom' and 'readTo'.
+func doCopy(ctx context.Context, dst, src proxyConn, buf []byte, cancel chan struct{}) error {
+	n, err := src.Read(ctx, buf, cancel)
+	if err != nil {
+		return fmt.Errorf("failed to read from %q: err %v", src.Name(), err)
+	}
+
+	_, err = dst.Write(ctx, buf[0:n], cancel)
+	if err != nil {
+		return fmt.Errorf("failed to write to %q: err %v", src.Name(), err)
+	}
+	return nil
+}
+
+// Start starts the proxy. On error on either end, the proxy cleans itself up by stopping both
+// connections.
+func (pf *Proxy) Start(ctx context.Context) {
+	pf.cu.Add(func() {
+		pf.to.Close(ctx)
+		pf.from.Close(ctx)
+	})
+
+	pf.wg.Add(1)
+	go func() {
+		if err := pf.readFrom(ctx); err != nil {
+			ctx.Warningf("Shutting down copy from %q to %q on container %s: %v", pf.from.Name(), pf.to.Name(), pf.cid, err)
+		}
+		pf.wg.Done()
+		pf.Close()
+	}()
+	pf.wg.Add(1)
+	go func() {
+		if err := pf.readTo(ctx); err != nil {
+			ctx.Warningf("Shutting down copy from %q to %q on container %s: %v", pf.to.Name(), pf.from.Name(), pf.cid, err)
+		}
+		pf.wg.Done()
+		pf.Close()
+	}()
+}
+
+// AddCleanup adds a cleanup to this Proxy's cleanup.
+func (pf *Proxy) AddCleanup(cu func()) {
+	pf.cu.Add(cu)
+}
+
+// Close cleans up the resources in this Proxy and blocks until all resources are cleaned up
+// and their goroutines exit.
+func (pf *Proxy) Close() {
+	pf.once.Do(func() {
+		pf.cu.Clean()
+		pf.cancelFrom <- struct{}{}
+		defer close(pf.cancelFrom)
+		pf.cancelTo <- struct{}{}
+		defer close(pf.cancelTo)
+	})
+	pf.wg.Wait()
 }

runsc/boot/portforward/portforward_fd_rw.go

@@ -16,6 +16,7 @@ package portforward
 
 import (
 	"io"
+	"sync"
 
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"
@@ -24,23 +25,31 @@ import (
 	"gvisor.dev/gvisor/pkg/waiter"
 )
 
-// fileDescriptionReadWriter implements io.ReadWriter and allows reading and
-// writing to a vfs.FileDescription.
-type fileDescriptionReadWriter struct {
-	// ctx is the context for the socket reader.
-	ctx context.Context
-
+// fileDescriptionConn
+type fileDescriptionConn struct {
 	// file is the file to read and write from.
 	file *vfs.FileDescription
+	// once makes sure we release the owned FileDescription once.
+	once sync.Once
 }
 
-// Read implements io.Reader.Read. It performs a blocking read on the fd.
-func (r *fileDescriptionReadWriter) Read(buf []byte) (int, error) {
+// NewFileDescriptionConn initializes a fileDescriptionConn.
+func NewFileDescriptionConn(file *vfs.FileDescription) proxyConn {
+	return &fileDescriptionConn{file: file}
+}
+
+// Name implements proxyConn.Name.
+func (r *fileDescriptionConn) Name() string {
+	return "fileDescriptionConn"
+}
+
+// Read implements proxyConn.Read.
+func (r *fileDescriptionConn) Read(ctx context.Context, buf []byte, cancel <-chan struct{}) (int, error) {
 	var (
 		notifyCh  chan struct{}
 		waitEntry waiter.Entry
 	)
-	n, err := r.file.Read(r.ctx, usermem.BytesIOSequence(buf), vfs.ReadOptions{})
+	n, err := r.file.Read(ctx, usermem.BytesIOSequence(buf), vfs.ReadOptions{})
 	for linuxerr.Equals(linuxerr.ErrWouldBlock, err) {
 		if notifyCh == nil {
 			waitEntry, notifyCh = waiter.NewChannelEntry(waiter.ReadableEvents | waiter.WritableEvents | waiter.EventHUp | waiter.EventErr)
@@ -48,8 +57,12 @@ func (r *fileDescriptionReadWriter) Read(buf []byte) (int, error) {
 			r.file.EventRegister(&waitEntry)
 			defer r.file.EventUnregister(&waitEntry)
 		}
-		<-notifyCh
-		n, err = r.file.Read(r.ctx, usermem.BytesIOSequence(buf), vfs.ReadOptions{})
+		select {
+		case <-notifyCh:
+		case <-cancel:
+			return 0, io.EOF
+		}
+		n, err = r.file.Read(ctx, usermem.BytesIOSequence(buf), vfs.ReadOptions{})
 	}
 
 	// host fd FileDescriptions use recvmsg which returns zero when the
@@ -60,11 +73,11 @@ func (r *fileDescriptionReadWriter) Read(buf []byte) (int, error) {
 	return int(n), err
 }
 
-// Write implements io.Writer.Write. It performs a blocking write on the fd.
-func (r *fileDescriptionReadWriter) Write(buf []byte) (int, error) {
+// Write implements proxyConn.Write.
+func (r *fileDescriptionConn) Write(ctx context.Context, buf []byte, cancel <-chan struct{}) (int, error) {
 	var notifyCh chan struct{}
 	var waitEntry waiter.Entry
-	n, err := r.file.Write(r.ctx, usermem.BytesIOSequence(buf), vfs.WriteOptions{})
+	n, err := r.file.Write(ctx, usermem.BytesIOSequence(buf), vfs.WriteOptions{})
 	for linuxerr.Equals(linuxerr.ErrWouldBlock, err) {
 		if notifyCh == nil {
 			waitEntry, notifyCh = waiter.NewChannelEntry(waiter.WritableEvents | waiter.EventHUp | waiter.EventErr)
@@ -72,8 +85,19 @@ func (r *fileDescriptionReadWriter) Write(buf []byte) (int, error) {
 			r.file.EventRegister(&waitEntry)
 			defer r.file.EventUnregister(&waitEntry)
 		}
-		<-notifyCh
-		n, err = r.file.Write(r.ctx, usermem.BytesIOSequence(buf), vfs.WriteOptions{})
+		select {
+		case <-notifyCh:
+		case <-cancel:
+			return 0, io.EOF
+		}
+		n, err = r.file.Write(ctx, usermem.BytesIOSequence(buf), vfs.WriteOptions{})
 	}
 	return int(n), err
 }
+
+// Close implements proxyConn.Close.
+func (r *fileDescriptionConn) Close(ctx context.Context) {
+	r.once.Do(func() {
+		r.file.DecRef(ctx)
+	})
+}

runsc/boot/portforward/portforward_fd_rw_test.go

@@ -116,6 +116,9 @@ func (rw *readerWriter) Read(ctx context.Context, dst usermem.IOSequence, opts v
 
 // Write implements vfs.FileDescriptionImpl.Write details for the parent mockFileDescription.
 func (rw *readerWriter) Write(ctx context.Context, src usermem.IOSequence, opts vfs.WriteOptions) (int64, error) {
+	if rw.released {
+		return 0, io.EOF
+	}
 	buf := make([]byte, src.NumBytes())
 	n, err := src.CopyIn(ctx, buf)
 	if err != nil {
@@ -273,8 +276,7 @@ func TestReaderWriter(t *testing.T) {
 				tc.mockFDImpl.Release(ctx)
 				t.Fatal(err)
 			}
-			readerWriter := fileDescriptionReadWriter{
-				ctx:  ctx,
+			readerWriter := fileDescriptionConn{
 				file: fd,
 			}
 			sendBytes := []([]byte){
@@ -284,7 +286,7 @@ func TestReaderWriter(t *testing.T) {
 				[]byte{'y', 'o', 'u', 'a', 'n', 'd', 'm', 'e'},
 			}
 			for _, buf := range sendBytes {
-				n, err := readerWriter.Write(buf)
+				n, err := readerWriter.Write(ctx, buf, nil)
 				if err != nil {
 					tc.mockFDImpl.Release(ctx)
 					t.Fatalf("write failed: %v", err)
@@ -294,10 +296,11 @@ func TestReaderWriter(t *testing.T) {
 					t.Fatalf("failed to write buf: %s", string(buf))
 				}
 			}
+
 			got := []byte{}
 			buf := make([]byte, 4)
 			for {
-				n, err := readerWriter.Read(buf)
+				n, err := readerWriter.Read(ctx, buf, nil)
 				if err == io.EOF {
 					break
 				}
@@ -310,7 +313,6 @@ func TestReaderWriter(t *testing.T) {
 				got = append(got, buf...)
 				buf = buf[0:]
 			}
-
 			tc.mockFDImpl.Release(ctx)
 
 			want := []byte{}
@@ -322,7 +324,7 @@ func TestReaderWriter(t *testing.T) {
 				t.Fatalf("mismatch types: got: %q want: %q", string(got), string(want))
 			}
 
-			_, err = readerWriter.Read(buf[0:])
+			_, err = readerWriter.Read(ctx, buf[0:], nil)
 			if err != io.EOF {
 				t.Fatalf("expected end of file: got: %v", err)
 			}

runsc/boot/portforward/portforward_hostinet.go

@@ -22,10 +22,9 @@ import (
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/cleanup"
 	"gvisor.dev/gvisor/pkg/context"
+	"gvisor.dev/gvisor/pkg/errors/linuxerr"
 	fileDescriptor "gvisor.dev/gvisor/pkg/fd"
 	"gvisor.dev/gvisor/pkg/fdnotifier"
-	"gvisor.dev/gvisor/pkg/log"
-	"gvisor.dev/gvisor/pkg/sentry/vfs"
 	"gvisor.dev/gvisor/pkg/waiter"
 )
 
@@ -33,41 +32,47 @@ var (
 	localHost = [4]byte{127, 0, 0, 1}
 )
 
-// localHostSocket allows reading and writing to a local host socket for hostinet.
-type localHostSocket struct {
+// hostInetConn allows reading and writing to a local host socket for hostinet.
+// hostInetConn implments proxyConn.
+type hostInetConn struct {
 	// wq is the WaitQueue registered with fdnotifier for this fd.
 	wq waiter.Queue
 	// fd is the file descriptor for the socket.
 	fd *fileDescriptor.FD
+	// port is the port on which to connect.
+	port uint16
+	// once makes sure we close only once.
+	once sync.Once
 }
 
-// newLocalHostSocket creates a hostSocket for an FD and registers the fd for
-// notifications.
-func newLocalHostSocket() (*localHostSocket, error) {
+// NewHostInetConn creates a hostInetConn backed by a host socket on the localhost address.
+func NewHostInetConn(port uint16) (proxyConn, error) {
 	// NOTE: Options must match sandbox seccomp filters. See filter/config.go
 	fd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0)
 	if err != nil {
 		return nil, err
 	}
-	s := localHostSocket{
-		fd: fileDescriptor.New(fd),
+	s := hostInetConn{
+		fd:   fileDescriptor.New(fd),
+		port: port,
 	}
+
+	cu := cleanup.Make(func() {
+		s.fd.Close()
+	})
+	defer cu.Clean()
 	if err := fdnotifier.AddFD(int32(s.fd.FD()), &s.wq); err != nil {
 		return nil, err
 	}
-	return &s, nil
-}
-
-// Connect performs a blocking connect on the socket to an ipv4 address.
-func (s *localHostSocket) Connect(port uint16) error {
+	cu.Add(func() { fdnotifier.RemoveFD(int32(s.fd.FD())) })
 	sockAddr := &unix.SockaddrInet4{
 		Addr: localHost,
-		Port: int(port),
+		Port: int(s.port),
 	}
 
 	if err := unix.Connect(s.fd.FD(), sockAddr); err != nil {
 		if err != unix.EINPROGRESS {
-			return err
+			return nil, fmt.Errorf("unix.Connect: %w", err)
 		}
 
 		// Connect is in progress. Wait for the socket to be writable.
@@ -85,200 +90,83 @@ func (s *localHostSocket) Connect(port uint16) error {
 		// Call getsockopt to get the connection result.
 		val, err := unix.GetsockoptInt(s.fd.FD(), unix.SOL_SOCKET, unix.SO_ERROR)
 		if err != nil {
-			return nil
+			return nil, fmt.Errorf("unix.GetSockoptInt: %w", err)
 		}
 		if val != 0 {
-			return unix.Errno(val)
+			return nil, fmt.Errorf("unix.GetSockoptInt: %w", unix.Errno(val))
 		}
 	}
+	cu.Release()
+	return &s, nil
+}
 
-	return nil
+func (s *hostInetConn) Name() string {
+	return fmt.Sprintf("localhost:port:%d", s.port)
 }
 
 // Read implements io.Reader.Read. It performs a blocking read on the fd.
-func (s *localHostSocket) Read(buf []byte) (int, error) {
+func (s *hostInetConn) Read(ctx context.Context, buf []byte, cancel <-chan struct{}) (int, error) {
 	var ch chan struct{}
 	var e waiter.Entry
 	n, err := s.fd.Read(buf)
-	for err == unix.EWOULDBLOCK {
+	for ctx.Err() == nil && linuxerr.Equals(linuxerr.ErrWouldBlock, err) {
 		if ch == nil {
-			e, ch = waiter.NewChannelEntry(waiter.ReadableEvents | waiter.WritableEvents | waiter.EventHUp | waiter.EventErr)
+			e, ch = waiter.NewChannelEntry(waiter.ReadableEvents | waiter.EventHUp | waiter.EventErr)
 			// Register for when the endpoint is writable or disconnected.
 			s.eventRegister(&e)
 			defer s.eventUnregister(&e)
 		}
-		<-ch
+		select {
+		case <-ch:
+		case <-cancel:
+			return 0, io.EOF
+		case <-ctx.Done():
+			return 0, ctx.Err()
+		}
 		n, err = s.fd.Read(buf)
 	}
 	return n, err
 }
 
 // Write implements io.Writer.Write. It performs a blocking write on the fd.
-func (s *localHostSocket) Write(buf []byte) (int, error) {
+func (s *hostInetConn) Write(ctx context.Context, buf []byte, cancel <-chan struct{}) (int, error) {
 	var ch chan struct{}
 	var e waiter.Entry
 	n, err := s.fd.Write(buf)
-	for err == unix.EWOULDBLOCK {
+	for ctx.Err() == nil && linuxerr.Equals(linuxerr.ErrWouldBlock, err) {
 		if ch == nil {
 			e, ch = waiter.NewChannelEntry(waiter.WritableEvents | waiter.EventHUp | waiter.EventErr)
 			// Register for when the endpoint is writable or disconnected.
 			s.eventRegister(&e)
 			defer s.eventUnregister(&e)
+
+		}
+		select {
+		case <-ch:
+		case <-cancel:
+			return 0, io.EOF
+		case <-ctx.Done():
+			return 0, ctx.Err()
 		}
-		<-ch
 		n, err = s.fd.Write(buf)
 	}
 	return n, err
 }
 
-func (s *localHostSocket) eventRegister(e *waiter.Entry) {
+func (s *hostInetConn) eventRegister(e *waiter.Entry) {
 	s.wq.EventRegister(e)
 	fdnotifier.UpdateFD(int32(s.fd.FD()))
 }
 
-func (s *localHostSocket) eventUnregister(e *waiter.Entry) {
+func (s *hostInetConn) eventUnregister(e *waiter.Entry) {
 	s.wq.EventUnregister(e)
 	fdnotifier.UpdateFD(int32(s.fd.FD()))
 }
 
 // Close closes the host socket and removes it from notifications.
-func (s *localHostSocket) Close() {
-	fdnotifier.RemoveFD(int32(s.fd.FD()))
-	s.fd.Close()
-}
-
-// hostinetportForwardConn is a hostinet port forwarding connection.
-type hostinetPortForwardConn struct {
-	// cid is the container id that this connection is connecting to.
-	cid string
-
-	// Socket is the host socket connected to the application.
-	socket *localHostSocket
-	// fd is the FileDescription for the imported host UDS fd.
-	fd *vfs.FileDescription
-
-	// status holds the status of the connection.
-	status struct {
-		sync.Mutex
-		// started indicates if the connection is started or not.
-		started bool
-		// closed indicates if the connection is closed or not.
-		closed bool
-	}
-
-	// toDone is closed when the copy to the application port is finished.
-	toDone chan struct{}
-
-	// fromDone is closed when the copy from the application socket is finished.
-	fromDone chan struct{}
-
-	// cu is called when the connection finishes.
-	cu cleanup.Cleanup
-}
-
-// newHostinetPortForward starts port forwarding to the given port in hostinet
-// mode.
-func newHostinetPortForward(ctx context.Context, cid string, fd *vfs.FileDescription, port uint16) (portForwardConn, error) {
-	log.Debugf("Handling hostinet port forwarding request for %s on port %d", cid, port)
-	appSocket, err := newLocalHostSocket()
-	if err != nil {
-		return nil, fmt.Errorf("hostinet socket: %w", err)
-	}
-
-	cu := cleanup.Make(func() { appSocket.Close() })
-	defer cu.Clean()
-
-	if err := appSocket.Connect(port); err != nil {
-		return nil, fmt.Errorf("hostinet connect: %w", err)
-	}
-
-	pfConn := hostinetPortForwardConn{
-		cid:      cid,
-		socket:   appSocket,
-		fd:       fd,
-		toDone:   make(chan struct{}),
-		fromDone: make(chan struct{}),
-		cu:       cleanup.Cleanup{},
-	}
-
-	cu.Release()
-	return &pfConn, nil
-}
-
-// Start implements portForwardConn.start.
-func (c *hostinetPortForwardConn) start(ctx context.Context) error {
-	c.status.Lock()
-	defer c.status.Unlock()
-
-	if c.status.closed {
-		return fmt.Errorf("already closed")
-	}
-	if c.status.started {
-		return fmt.Errorf("already started")
-	}
-
-	log.Debugf("Start forwarding to/from container %q and localhost", c.cid)
-
-	importedRW := &fileDescriptionReadWriter{
-		file: c.fd,
-	}
-
-	go func() {
-		_, _ = io.Copy(c.socket, importedRW)
-		// Indicate that this goroutine has completed.
-		close(c.toDone)
-		// Make sure to clean up when one half of the copy has finished.
-		c.close(ctx)
-	}()
-	go func() {
-		_, _ = io.Copy(importedRW, c.socket)
-		// Indicate that this goroutine has completed.
-		close(c.fromDone)
-		// Make sure to clean up when one half of the copy has finished.
-		c.close(ctx)
-	}()
-
-	c.status.started = true
-
-	return nil
-}
-
-// close implements portForwardConn.close.
-func (c *hostinetPortForwardConn) close(ctx context.Context) error {
-	c.status.Lock()
-
-	// This should be a no op if the connection is already closed.
-	if c.status.closed {
-		c.status.Unlock()
-		return nil
-	}
-
-	log.Debugf("Stopping forwarding to/from container %q and localhost...", c.cid)
-
-	// Closing the FileDescription and endpoint should make all
-	// goroutines exit.
-	c.fd.DecRef(ctx)
-	c.socket.Close()
-
-	// Wait for one goroutine to finish or for a save event.
-	<-c.toDone
-	log.Debugf("Stopped forwarding one-half of copy for %q", c.cid)
-
-	// Wait on the other goroutine.
-	<-c.fromDone
-	log.Debugf("Stopped forwarding to/from container %q and localhost", c.cid)
-
-	c.status.closed = true
-
-	c.status.Unlock()
-
-	// Call the cleanup object.
-	c.cu.Clean()
-
-	return nil
-}
-
-// cleanup implements portForwardConn.cleanup.
-func (c *hostinetPortForwardConn) cleanup(f func()) {
-	c.cu.Add(f)
+func (s *hostInetConn) Close(_ context.Context) {
+	s.once.Do(func() {
+		fdnotifier.RemoveFD(int32(s.fd.FD()))
+		s.fd.Close()
+	})
 }

runsc/boot/portforward/portforward_hostinet_test.go

@@ -18,14 +18,19 @@ import (
 	"fmt"
 	"net"
 	"reflect"
+	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"golang.org/x/sync/errgroup"
+	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"
 	"gvisor.dev/gvisor/pkg/sentry/contexttest"
 )
 
 func TestLocalHostSocket(t *testing.T) {
+	ctx := contexttest.Context(t)
 	clientData := append(
 		[]byte("do what must be done\n"),
 		[]byte("do not hesitate\n")...,
@@ -48,7 +53,7 @@ func TestLocalHostSocket(t *testing.T) {
 	g.Go(func() error {
 		conn, err := l.Accept()
 		if err != nil {
-			return fmt.Errorf("could not accept connection: %v", err)
+			t.Fatalf("could not accept connection: %v", err)
 		}
 		defer conn.Close()
 
@@ -75,16 +80,12 @@ func TestLocalHostSocket(t *testing.T) {
 	})
 
 	g.Go(func() error {
-		sock, err := newLocalHostSocket()
+		sock, err := NewHostInetConn(uint16(port))
 		if err != nil {
-			return fmt.Errorf("could not create local host socket: %v", err)
-		}
-		defer sock.Close()
-		if err := sock.Connect(uint16(port)); err != nil {
-			return fmt.Errorf("could not connect to local host socket: %v", err)
+			t.Fatalf("could not create local host socket: %v", err)
 		}
 		for i := 0; i < len(clientData); {
-			n, err := sock.Write(clientData[i:])
+			n, err := sock.Write(ctx, clientData[i:], nil)
 			if err != nil {
 				return fmt.Errorf("could not write to local host socket: %v", err)
 			}
@@ -94,7 +95,7 @@ func TestLocalHostSocket(t *testing.T) {
 		data := make([]byte, 1024)
 		dataLen := 0
 		for dataLen < len(serverData) {
-			n, err := sock.Read(data[dataLen:])
+			n, err := sock.Read(ctx, data[dataLen:], nil)
 			if err != nil {
 				t.Fatalf("could not read from local host socket: %v", err)
 			}
@@ -114,17 +115,27 @@ func TestLocalHostSocket(t *testing.T) {
 
 type netConnMockEndpoint struct {
 	conn net.Conn
+	mu   sync.Mutex
 }
 
 // read implements portforwarderTestHarness.read.
 func (nc *netConnMockEndpoint) read(n int) ([]byte, error) {
+	nc.mu.Lock()
+	defer nc.mu.Unlock()
+
 	buf := make([]byte, n)
-	n, err := nc.conn.Read(buf)
-	return buf[:n], err
+	nc.conn.SetReadDeadline(time.Now().Add(time.Millisecond * 500))
+	res, err := nc.conn.Read(buf)
+	if err != nil && strings.Contains(err.Error(), "timeout") {
+		return nil, linuxerr.ErrWouldBlock
+	}
+	return buf[:res], err
 }
 
 // write implements portforwarderTestHarness write.
 func (nc *netConnMockEndpoint) write(buf []byte) (int, error) {
+	nc.mu.Lock()
+	defer nc.mu.Unlock()
 	written := 0
 	for {
 		n, err := nc.conn.Write(buf[written:])
@@ -138,7 +149,7 @@ func (nc *netConnMockEndpoint) write(buf []byte) (int, error) {
 	}
 }
 
-func TestHostinetPortForwardConn(t *testing.T) {
+func TestHostInetProxy(t *testing.T) {
 	for _, tc := range []struct {
 		name     string
 		requests map[string]string
@@ -167,41 +178,42 @@ func TestHostinetPortForwardConn(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			doHostinetTest(t, tc.requests)
+			doHostinetTest(t, tc.name, tc.requests)
 		})
 	}
 }
 
-func doHostinetTest(t *testing.T, requests map[string]string) {
-	ctx := contexttest.Context(t)
-	appEndpoint := &mockApplicationFDImpl{}
-	defer appEndpoint.Release(ctx)
+func doHostinetTest(t *testing.T, name string, requests map[string]string) {
+	ctx := context.Background()
+	appEndpoint := newMockApplicationFDImpl()
 	client, err := newMockFileDescription(ctx, appEndpoint)
 	if err != nil {
-		t.Fatalf("newMockFileDescription failed: %v", err)
+		t.Fatalf("newMockFileDescription: %v", err)
 	}
+
 	l, err := net.Listen("tcp", ":0")
 	if err != nil {
 		t.Fatalf("net.Listen failed: %v", err)
 	}
 	defer l.Close()
-	port := l.Addr().(*net.TCPAddr).Port
-	portForwardConn, err := newHostinetPortForward(ctx, "", client, uint16(port))
+	port := uint16(l.Addr().(*net.TCPAddr).Port)
+	sock, err := NewHostInetConn(port)
 	if err != nil {
-		t.Fatalf("newHostinetPortForward failed: %v", err)
+		t.Fatalf("could not create local host socket: %v", err)
 	}
-	if err := portForwardConn.start(ctx); err != nil {
-		t.Fatalf("portForwardConn.start failed: %v", err)
-	}
-	conn, err := l.Accept()
-	if err != nil {
-		t.Fatalf("l.Accept failed: %v", err)
-	}
-	defer conn.Close()
 
+	proxy := NewProxy(ProxyPair{To: sock, From: &fileDescriptionConn{file: client}}, name)
+
+	proxy.Start(ctx)
+
+	shim, err := l.Accept()
+	if err != nil {
+		t.Fatalf("could not accept shim connection: %v", err)
+	}
+	defer shim.Close()
 	harness := portforwarderTestHarness{
 		app:  appEndpoint,
-		shim: &netConnMockEndpoint{conn},
+		shim: &netConnMockEndpoint{conn: shim},
 	}
 
 	for req, resp := range requests {
@@ -226,7 +238,6 @@ func doHostinetTest(t *testing.T, requests map[string]string) {
 		if err != nil {
 			t.Fatalf("failed to read from shim: %v", err)
 		}
-
 		if string(got) != resp {
 			t.Fatalf("shim mismatch: got: %s want: %s", string(got), resp)
 		}

runsc/boot/portforward/portforward_netstack.go

@@ -17,12 +17,10 @@ package portforward
 import (
 	"bytes"
 	"fmt"
+	"io"
+	"sync"
 
-	"gvisor.dev/gvisor/pkg/cleanup"
 	"gvisor.dev/gvisor/pkg/context"
-	"gvisor.dev/gvisor/pkg/log"
-	"gvisor.dev/gvisor/pkg/sentry/vfs"
-	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/network/ipv4"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
@@ -30,203 +28,126 @@ import (
 	"gvisor.dev/gvisor/pkg/waiter"
 )
 
-// netstackPortForwardConn is a portForwardConn implementation for netstack.
-type netstackPortForwardConn struct {
-	// cid is the container id that this connection is connecting to.
-	cid string
-
-	// ep is the tcpip.Endpoint to the application port.
+// netstackConn allows reading and writing to a netstack endpoint.
+// netstackConn implements proxyConn.
+type netstackConn struct {
+	// ep is the tcpip.Endpoint on which to read and write.
 	ep tcpip.Endpoint
-	// wq is the endpoint waiter.Queue.
+	// port is the port on which to connect.
+	port uint16
+	// wq is the WaitQueue for this connection to wait on notifications.
 	wq *waiter.Queue
-	// fd is the FileDescription for the imported host UDS fd.
-	fd *vfs.FileDescription
-
-	// status holds the status of the connection.
-	status struct {
-		sync.Mutex
-		// started indicates if the connection is started or not.
-		started bool
-		// closed indicates if the connection is closed or not.
-		closed bool
-	}
-
-	// toDone is closed when the copy to the application port is finished.
-	toDone chan struct{}
-
-	// fromDone is closed when the copy from the application socket is finished.
-	fromDone chan struct{}
-
-	// cu is called when the connection finishes.
-	cu cleanup.Cleanup
+	// once makes sure Close is called once.
+	once sync.Once
 }
 
-// newNetstackPortForward creates a new port forwarding connection to the given
+// NewNetstackConn creates a new port forwarding connection to the given
 // port in netstack mode.
-func newNetstackPortForward(ctx context.Context, stack *stack.Stack, cid string, fd *vfs.FileDescription, port uint16) (portForwardConn, error) {
+func NewNetstackConn(stack *stack.Stack, port uint16) (proxyConn, error) {
 	var wq waiter.Queue
 	ep, tcpErr := stack.NewEndpoint(tcp.ProtocolNumber, ipv4.ProtocolNumber, &wq)
 	if tcpErr != nil {
 		return nil, fmt.Errorf("creating endpoint: %v", tcpErr)
 	}
-	cu := cleanup.Make(func() { ep.Close() })
-	defer cu.Clean()
-
+	n := &netstackConn{
+		ep:   ep,
+		port: port,
+		wq:   &wq,
+	}
 	waitEntry, notifyCh := waiter.NewChannelEntry(waiter.WritableEvents)
-	wq.EventRegister(&waitEntry)
-	defer wq.EventUnregister(&waitEntry)
+	n.wq.EventRegister(&waitEntry)
+	defer n.wq.EventUnregister(&waitEntry)
 
-	tcpErr = ep.Connect(tcpip.FullAddress{
+	tcpErr = n.ep.Connect(tcpip.FullAddress{
 		Addr: "\x7f\x00\x00\x01", // 127.0.0.1
-		Port: port,
+		Port: n.port,
 	})
 	if _, ok := tcpErr.(*tcpip.ErrConnectStarted); ok {
 		<-notifyCh
-		tcpErr = ep.LastError()
+		tcpErr = n.ep.LastError()
 	}
 	if tcpErr != nil {
 		return nil, fmt.Errorf("connecting endpoint: %v", tcpErr)
 	}
-
-	pfConn := netstackPortForwardConn{
-		cid:      cid,
-		ep:       ep,
-		wq:       &wq,
-		fd:       fd,
-		toDone:   make(chan struct{}),
-		fromDone: make(chan struct{}),
-		cu:       cleanup.Cleanup{},
-	}
-
-	cu.Release()
-	return &pfConn, nil
+	return n, nil
 }
 
-// start implements portForwardConn.start.
-func (c *netstackPortForwardConn) start(ctx context.Context) error {
-	c.status.Lock()
-	defer c.status.Unlock()
-
-	if c.status.closed {
-		return fmt.Errorf("already closed")
-	}
-	if c.status.started {
-		return fmt.Errorf("already started")
-	}
-
-	log.Debugf("Start forwarding to/from container %q and localhost", c.cid)
-
-	go c.writeToEP(ctx)
-	go c.readFromEP(ctx)
-
-	c.status.started = true
-
-	return nil
+// Name implements proxyConn.Name.
+func (n *netstackConn) Name() string {
+	return fmt.Sprintf("netstack:port:%d", n.port)
 }
 
-// close implements portForwardConn.close.
-func (c *netstackPortForwardConn) close(ctx context.Context) error {
-	c.status.Lock()
-
-	// This should be a no op if the connection is already closed.
-	if c.status.closed {
-		c.status.Unlock()
-		return nil
-	}
-
-	log.Debugf("Stopping forwarding to/from container %q and localhost...", c.cid)
-
-	// Closing the endpoint will make the other goroutine exit.
-	c.ep.Close()
-	c.fd.DecRef(ctx)
-
-	<-c.toDone
-	log.Debugf("Stopped forwarding one-half of copy for %q", c.cid)
-
-	// Wait on the other goroutine.
-	<-c.fromDone
-	log.Debugf("Stopped forwarding to/from container %q and localhost", c.cid)
-
-	c.status.closed = true
-
-	c.status.Unlock()
-
-	// Call the cleanup object.
-	c.cu.Clean()
-
-	return nil
+// bufWriter is used as an io.Writer to read from tcpip.Endpoint.
+type bufWriter struct {
+	buf    []byte
+	offset int64
 }
 
-// cleanup implements portForwardConn.cleanup.
-func (c *netstackPortForwardConn) cleanup(f func()) {
-	c.cu.Add(f)
+// Write implements io.Writer.
+func (b *bufWriter) Write(buf []byte) (int, error) {
+	n := copy(b.buf[b.offset:], buf)
+	b.offset += int64(n)
+	return n, nil
 }
 
-// readFromEP reads from the tcpip.Endpoint and writes to the given Writer.
-func (c *netstackPortForwardConn) readFromEP(ctx context.Context) {
-	w := &fileDescriptionReadWriter{
-		file: c.fd,
+// Read implements proxyConn.Read.
+func (n *netstackConn) Read(ctx context.Context, buf []byte, cancel <-chan struct{}) (int, error) {
+	var ch chan struct{}
+	var e waiter.Entry
+	b := &bufWriter{
+		buf: buf,
 	}
-
-	// Register for read notifications.
-	waitEntry, notifyCh := waiter.NewChannelEntry(waiter.EventIn | waiter.EventHUp | waiter.EventErr)
-	// Register for when the endpoint is readable or disconnected.
-	c.wq.EventRegister(&waitEntry)
-
-	for {
-		_, err := c.ep.Read(w, tcpip.ReadOptions{})
-		if err != nil {
-			if _, ok := err.(*tcpip.ErrWouldBlock); ok {
-				<-notifyCh
-				continue
-			}
-			log.Infof("Port forward read error; cid: %q: %v", c.cid, err)
-			break
+	res, tcpErr := n.ep.Read(b, tcpip.ReadOptions{})
+	for _, ok := tcpErr.(*tcpip.ErrWouldBlock); ok && ctx.Err() == nil; _, ok = tcpErr.(*tcpip.ErrWouldBlock) {
+		if ch == nil {
+			e, ch = waiter.NewChannelEntry(waiter.ReadableEvents | waiter.EventIn | waiter.EventHUp | waiter.EventErr)
+			n.wq.EventRegister(&e)
+			defer n.wq.EventUnregister(&e)
 		}
+		select {
+		case <-ch:
+		case <-cancel:
+			return 0, io.EOF
+		case <-ctx.Done():
+			return 0, ctx.Err()
+		}
+		res, tcpErr = n.ep.Read(b, tcpip.ReadOptions{})
 	}
-
-	// Clean up when one half of the copy is finished.
-	c.wq.EventUnregister(&waitEntry)
-	c.ep.Shutdown(tcpip.ShutdownRead)
-	close(c.fromDone)
-	c.close(ctx)
+	if tcpErr != nil {
+		return 0, io.EOF
+	}
+	return res.Total, nil
 }
 
-func (c *netstackPortForwardConn) writeToEP(ctx context.Context) {
-	r := &fileDescriptionReadWriter{
-		file: c.fd,
-	}
-
-	// Register for write notifications.
-	waitEntry, notifyCh := waiter.NewChannelEntry(waiter.WritableEvents | waiter.EventHUp | waiter.EventErr)
-	// Register for when the endpoint is writable or disconnected.
-	c.wq.EventRegister(&waitEntry)
-
-	v := make([]byte, 16384 /* 16kb read buffer size */)
-	for {
-		n, err := r.Read(v)
-		if err != nil {
-			break
+// Write implements proxyConn.Write.
+func (n *netstackConn) Write(ctx context.Context, buf []byte, cancel <-chan struct{}) (int, error) {
+	var ch chan struct{}
+	var e waiter.Entry
+	var b bytes.Reader
+	b.Reset(buf)
+	res, tcpErr := n.ep.Write(&b, tcpip.WriteOptions{Atomic: true})
+	for _, ok := tcpErr.(*tcpip.ErrWouldBlock); ok && ctx.Err() == nil; _, ok = tcpErr.(*tcpip.ErrWouldBlock) {
+		if ch == nil {
+			e, ch = waiter.NewChannelEntry(waiter.WritableEvents | waiter.EventIn | waiter.EventHUp | waiter.EventErr)
+			n.wq.EventRegister(&e)
+			defer n.wq.EventUnregister(&e)
 		}
-		var b bytes.Reader
-		b.Reset(v[:n])
-		for b.Len() != 0 {
-			_, err := c.ep.Write(&b, tcpip.WriteOptions{Atomic: true})
-			if err != nil {
-				// If the channel is not ready for writing then wait until it is.
-				if _, ok := err.(*tcpip.ErrWouldBlock); ok {
-					<-notifyCh
-					continue
-				}
-				log.Infof("Port forward read error; cid: %q: %v", c.cid, err)
-				break
-			}
+		select {
+		case <-ch:
+		case <-cancel:
+			return 0, io.EOF
+		case <-ctx.Done():
+			return 0, ctx.Err()
 		}
+		res, tcpErr = n.ep.Write(&b, tcpip.WriteOptions{Atomic: true})
 	}
+	if tcpErr != nil {
+		return 0, io.EOF
+	}
+	return int(res), nil
+}
 
-	// Clean up when one half of the copy is finished.
-	c.wq.EventUnregister(&waitEntry)
-	c.ep.Shutdown(tcpip.ShutdownWrite)
-	close(c.toDone)
-	c.close(ctx)
+// Close implements proxyConn.Close.
+func (n *netstackConn) Close(_ context.Context) {
+	n.once.Do(func() { n.ep.Close() })
 }

runsc/boot/portforward/portforward_netstack_test.go

@@ -20,9 +20,7 @@ import (
 	"sync"
 	"testing"
 
-	"gvisor.dev/gvisor/pkg/cleanup"
 	"gvisor.dev/gvisor/pkg/sentry/contexttest"
-	"gvisor.dev/gvisor/pkg/sentry/vfs"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/waiter"
 )
@@ -32,7 +30,6 @@ type baseTCPEndpointImpl struct {
 	readBuf  bytes.Buffer
 	writeBuf bytes.Buffer
 	mu       sync.Mutex
-	wq       *waiter.Queue
 }
 
 // read reads data from the buffer that "Write" writes to.
@@ -43,7 +40,6 @@ func (b *baseTCPEndpointImpl) read(n int) ([]byte, error) {
 		return nil, io.EOF
 	}
 	ret := b.writeBuf.Next(n)
-	b.wq.Notify(waiter.WritableEvents)
 	return ret, nil
 }
 
@@ -55,7 +51,6 @@ func (b *baseTCPEndpointImpl) write(buf []byte) (int, error) {
 		return 0, io.EOF
 	}
 	n, err := b.readBuf.Write(buf)
-	b.wq.Notify(waiter.ReadableEvents)
 	return n, err
 }
 
@@ -101,22 +96,13 @@ func (b *baseTCPEndpointImpl) Write(payload tcpip.Payloader, _ tcpip.WriteOption
 }
 
 func (b *baseTCPEndpointImpl) Shutdown(shutdown tcpip.ShutdownFlags) tcpip.Error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.closed = true
 	return nil
 }
 
-func newNetstackPortForwardConnWithMock(impl mockTCPEndpointImpl, fd *vfs.FileDescription, wq *waiter.Queue) *netstackPortForwardConn {
-	ep := &mockTCPEndpoint{impl}
-	return &netstackPortForwardConn{
-		ep:       ep,
-		wq:       wq,
-		fd:       fd,
-		toDone:   make(chan struct{}),
-		fromDone: make(chan struct{}),
-		cu:       cleanup.Cleanup{},
-	}
-}
-
-func TestNetstackPortforward(t *testing.T) {
+func TestNetstackProxy(t *testing.T) {
 	for _, tc := range []struct {
 		name     string
 		requests map[string]string
@@ -145,27 +131,30 @@ func TestNetstackPortforward(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			doNetstackTest(t, tc.requests)
+			doNetstackTest(t, tc.name, tc.requests)
 		})
 	}
 }
 
-func doNetstackTest(t *testing.T, responses map[string]string) {
+func doNetstackTest(t *testing.T, name string, responses map[string]string) {
 	ctx := contexttest.Context(t)
-	appEndpoint := &mockApplicationFDImpl{}
-	defer appEndpoint.Release(ctx)
+	appEndpoint := newMockApplicationFDImpl()
 	fd, err := newMockFileDescription(ctx, appEndpoint)
 	if err != nil {
 		t.Fatalf("newMockFileDescription: %v", err)
 	}
 
-	wq := waiter.Queue{}
-	impl := &baseTCPEndpointImpl{wq: &wq}
-	conn := newNetstackPortForwardConnWithMock(impl, fd, &wq)
-	if err := conn.start(ctx); err != nil {
-		t.Fatalf("conn.start: %v", err)
+	wq := &waiter.Queue{}
+	impl := &baseTCPEndpointImpl{}
+	ep := newMockTCPEndpoint(impl, wq)
+	sock := &netstackConn{
+		ep: ep,
+		wq: wq,
 	}
-	defer conn.close(ctx)
+
+	proxy := NewProxy(ProxyPair{To: sock, From: &fileDescriptionConn{file: fd}}, name)
+	proxy.Start(ctx)
+	defer proxy.Close()
 
 	harness := portforwarderTestHarness{
 		app:  appEndpoint,
@@ -200,3 +189,62 @@ func doNetstackTest(t *testing.T, responses map[string]string) {
 		}
 	}
 }
+
+// tcpErrImpl blocks on the first Read/Write and then throws an error afterwards.
+type tcpErrImpl struct {
+	mu     sync.Mutex
+	reads  bool
+	writes bool
+}
+
+// Read implements mockTCPEndpointImpl.Read.
+func (e *tcpErrImpl) Read(w io.Writer, _ tcpip.ReadOptions) (tcpip.ReadResult, tcpip.Error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.reads {
+		return tcpip.ReadResult{}, &tcpip.ErrBadLocalAddress{}
+	}
+	e.reads = true
+	return tcpip.ReadResult{}, &tcpip.ErrWouldBlock{}
+}
+
+// Write implements mockTCPEndpointImpl.Write.
+func (e *tcpErrImpl) Write(payload tcpip.Payloader, _ tcpip.WriteOptions) (int64, tcpip.Error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.writes {
+		return 0, &tcpip.ErrBadLocalAddress{}
+	}
+	e.writes = true
+	return 0, &tcpip.ErrWouldBlock{}
+}
+
+// Shutdown implements mockTCPEndpointImpl.Shutdown.
+func (e *tcpErrImpl) Shutdown(shutdown tcpip.ShutdownFlags) tcpip.Error {
+	return nil
+}
+
+// Close implements mockTCPEndpointImpl.Shutdown.
+func (e *tcpErrImpl) Close() {}
+
+// TestNTestNestackReadsWrites checks that reads/writes check errors from the underlying endpoint
+// multiple times.
+func TestNestackReadsWrites(t *testing.T) {
+	ctx := contexttest.Context(t)
+	wq := &waiter.Queue{}
+	ep := newMockTCPEndpoint(&tcpErrImpl{}, wq)
+	cancel := make(chan struct{})
+	conn := netstackConn{ep: ep, wq: wq}
+	defer close(cancel)
+	defer conn.Close(ctx)
+
+	_, err := conn.Read(ctx, []byte("something"), cancel)
+	if err != io.EOF {
+		t.Fatalf("mismatch read err: want: %v got: %v", io.EOF, err)
+	}
+
+	_, err = conn.Write(ctx, []byte("something"), cancel)
+	if err != io.EOF {
+		t.Fatalf("mismatch write err: want: %v got: %v", io.EOF, err)
+	}
+}

runsc/boot/portforward/portforward_test_util.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"io"
 	"sync"
+	"time"
 
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"
@@ -76,15 +77,22 @@ type mockApplicationFDImpl struct {
 	vfs.FileDescriptionDefaultImpl
 	vfs.NoLockFD
 	vfs.DentryMetadataFileDescriptionImpl
-	mu       sync.Mutex
-	readBuf  bytes.Buffer
-	writeBuf bytes.Buffer
-	released bool
-	queue    waiter.Queue
+	mu         sync.Mutex
+	readBuf    bytes.Buffer
+	writeBuf   bytes.Buffer
+	released   bool
+	queue      waiter.Queue
+	notifyStop chan struct{}
 }
 
 var _ vfs.FileDescriptionImpl = (*mockApplicationFDImpl)(nil)
 
+func newMockApplicationFDImpl() *mockApplicationFDImpl {
+	app := &mockApplicationFDImpl{notifyStop: make(chan struct{})}
+	go app.doNotify()
+	return app
+}
+
 // Read implements vfs.FileDescriptionImpl.Read details for the parent mockFileDescription.
 func (s *mockApplicationFDImpl) Read(ctx context.Context, dst usermem.IOSequence, opts vfs.ReadOptions) (int64, error) {
 	s.mu.Lock()
@@ -109,12 +117,9 @@ func (s *mockApplicationFDImpl) Write(ctx context.Context, src usermem.IOSequenc
 	}
 
 	buf := make([]byte, src.NumBytes())
-	n, err := src.CopyIn(ctx, buf)
-	if err != nil {
-		return int64(n), err
-	}
-	res, err := s.writeBuf.Write(buf)
-	return int64(res), err
+	n, _ := src.CopyIn(ctx, buf)
+	res, _ := s.writeBuf.Write(buf[:n])
+	return int64(res), nil
 }
 
 // write implements mockEndpoint.write.
@@ -125,7 +130,6 @@ func (s *mockApplicationFDImpl) write(buf []byte) (int, error) {
 		return 0, io.EOF
 	}
 	ret, err := s.readBuf.Write(buf)
-	s.queue.Notify(waiter.ReadableEvents)
 	return ret, err
 }
 
@@ -140,10 +144,21 @@ func (s *mockApplicationFDImpl) read(n int) ([]byte, error) {
 		return nil, linuxerr.ErrWouldBlock
 	}
 	ret := s.writeBuf.Next(n)
-	s.queue.Notify(waiter.WritableEvents)
 	return ret, nil
 }
 
+func (s *mockApplicationFDImpl) doNotify() {
+	for {
+		s.queue.Notify(waiter.ReadableEvents | waiter.WritableEvents | waiter.EventHUp)
+		select {
+		case <-s.notifyStop:
+			return
+		default:
+			time.Sleep(time.Millisecond * 50)
+		}
+	}
+}
+
 func (s *mockApplicationFDImpl) IsReadable() bool {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -178,8 +193,8 @@ func (s *mockApplicationFDImpl) EventUnregister(we *waiter.Entry) {
 func (s *mockApplicationFDImpl) Release(context.Context) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s.queue.Notify(waiter.ReadableEvents)
 	s.released = true
+	s.notifyStop <- struct{}{}
 }
 
 // mockTCPEndpointImpl is the subset of methods used by tests for the mockTCPEndpoint struct. This
@@ -193,7 +208,33 @@ type mockTCPEndpointImpl interface {
 
 // mockTCPEndpoint mocks tcpip.Endpoint for tests.
 type mockTCPEndpoint struct {
-	impl mockTCPEndpointImpl // impl implements the subset of methods needed for mockTCPEndpoints.
+	impl       mockTCPEndpointImpl // impl implements the subset of methods needed for mockTCPEndpoints.
+	wq         *waiter.Queue
+	notifyDone chan struct{}
+}
+
+func newMockTCPEndpoint(impl mockTCPEndpointImpl, wq *waiter.Queue) *mockTCPEndpoint {
+	ret := &mockTCPEndpoint{
+		impl:       impl,
+		wq:         wq,
+		notifyDone: make(chan struct{}),
+	}
+
+	go ret.doNotify()
+	return ret
+}
+
+func (m *mockTCPEndpoint) doNotify() {
+	for {
+		m.wq.Notify(waiter.ReadableEvents | waiter.WritableEvents | waiter.EventHUp)
+		select {
+		case <-m.notifyDone:
+			return
+		default:
+			time.Sleep(time.Millisecond * 50)
+		}
+
+	}
 }
 
 // The below are trivial stub methods to get mockTCPEndpoint to implement tcpip.Endpoint. They
@@ -202,6 +243,7 @@ type mockTCPEndpoint struct {
 // Close implements tcpip.Endpoint.Close.
 func (m *mockTCPEndpoint) Close() {
 	m.impl.Close()
+	m.notifyDone <- struct{}{}
 }
 
 // Abort implements tcpip.Endpoint.Abort.