Merge release-20210726.0-17-g979d6e7d7 (automated)

4 files changed, 88 insertions, 27 deletions

diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go
index e8e716db0..48356c343 100644
--- a/pkg/tcpip/link/fdbased/endpoint.go
+++ b/pkg/tcpip/link/fdbased/endpoint.go
@@ -56,6 +56,7 @@ import (
 // linkDispatcher reads packets from the link FD and dispatches them to the
 // NetworkDispatcher.
 type linkDispatcher interface {
+	stop()
 	dispatch() (bool, tcpip.Error)
 }
 
@@ -381,16 +382,27 @@ func isSocketFD(fd int) (bool, error) {
 // Attach launches the goroutine that reads packets from the file descriptor and
 // dispatches them via the provided dispatcher.
 func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) {
-	e.dispatcher = dispatcher
-	// Link endpoints are not savable. When transportation endpoints are
-	// saved, they stop sending outgoing packets and all incoming packets
-	// are rejected.
-	for i := range e.inboundDispatchers {
-		e.wg.Add(1)
-		go func(i int) { // S/R-SAFE: See above.
-			e.dispatchLoop(e.inboundDispatchers[i])
-			e.wg.Done()
-		}(i)
+	// nil means the NIC is being removed.
+	if dispatcher == nil && e.dispatcher != nil {
+		for _, dispatcher := range e.inboundDispatchers {
+			dispatcher.stop()
+		}
+		e.Wait()
+		e.dispatcher = nil
+		return
+	}
+	if dispatcher != nil && e.dispatcher == nil {
+		e.dispatcher = dispatcher
+		// Link endpoints are not savable. When transportation endpoints are
+		// saved, they stop sending outgoing packets and all incoming packets
+		// are rejected.
+		for i := range e.inboundDispatchers {
+			e.wg.Add(1)
+			go func(i int) { // S/R-SAFE: See above.
+				e.dispatchLoop(e.inboundDispatchers[i])
+				e.wg.Done()
+			}(i)
+		}
 	}
 }
 
diff --git a/pkg/tcpip/link/fdbased/mmap.go b/pkg/tcpip/link/fdbased/mmap.go
index bfae34ab9..3f516cab5 100644
--- a/pkg/tcpip/link/fdbased/mmap.go
+++ b/pkg/tcpip/link/fdbased/mmap.go
@@ -114,6 +114,7 @@ func (t tPacketHdr) Payload() []byte {
 // packetMMapDispatcher uses PACKET_RX_RING's to read/dispatch inbound packets.
 // See: mmap_amd64_unsafe.go for implementation details.
 type packetMMapDispatcher struct {
+	stopFd
 	// fd is the file descriptor used to send and receive packets.
 	fd int
 
@@ -129,18 +130,18 @@ type packetMMapDispatcher struct {
 	ringOffset int
 }
 
-func (d *packetMMapDispatcher) readMMappedPacket() ([]byte, tcpip.Error) {
+func (d *packetMMapDispatcher) readMMappedPacket() ([]byte, bool, tcpip.Error) {
 	hdr := tPacketHdr(d.ringBuffer[d.ringOffset*tpFrameSize:])
 	for hdr.tpStatus()&tpStatusUser == 0 {
-		event := rawfile.PollEvent{
-			FD:     int32(d.fd),
-			Events: unix.POLLIN | unix.POLLERR,
-		}
-		if _, errno := rawfile.BlockingPoll(&event, 1, nil); errno != 0 {
+		stopped, errno := rawfile.BlockingPollUntilStopped(d.efd, d.fd, unix.POLLIN|unix.POLLERR)
+		if errno != 0 {
 			if errno == unix.EINTR {
 				continue
 			}
-			return nil, rawfile.TranslateErrno(errno)
+			return nil, stopped, rawfile.TranslateErrno(errno)
+		}
+		if stopped {
+			return nil, true, nil
 		}
 		if hdr.tpStatus()&tpStatusCopy != 0 {
 			// This frame is truncated so skip it after flipping the
@@ -158,14 +159,14 @@ func (d *packetMMapDispatcher) readMMappedPacket() ([]byte, tcpip.Error) {
 	// Release packet to kernel.
 	hdr.setTPStatus(tpStatusKernel)
 	d.ringOffset = (d.ringOffset + 1) % tpFrameNR
-	return pkt, nil
+	return pkt, false, nil
 }
 
 // dispatch reads packets from an mmaped ring buffer and dispatches them to the
 // network stack.
 func (d *packetMMapDispatcher) dispatch() (bool, tcpip.Error) {
-	pkt, err := d.readMMappedPacket()
-	if err != nil {
+	pkt, stopped, err := d.readMMappedPacket()
+	if err != nil || stopped {
 		return false, err
 	}
 	var (
diff --git a/pkg/tcpip/link/fdbased/mmap_unsafe.go b/pkg/tcpip/link/fdbased/mmap_unsafe.go
index 58d5dfeef..5b786169a 100644
--- a/pkg/tcpip/link/fdbased/mmap_unsafe.go
+++ b/pkg/tcpip/link/fdbased/mmap_unsafe.go
@@ -47,9 +47,14 @@ func (t tPacketHdr) setTPStatus(status uint32) {
 }
 
 func newPacketMMapDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
+	stopFd, err := newStopFd()
+	if err != nil {
+		return nil, err
+	}
 	d := &packetMMapDispatcher{
-		fd: fd,
-		e:  e,
+		stopFd: stopFd,
+		fd:     fd,
+		e:      e,
 	}
 	pageSize := unix.Getpagesize()
 	if tpBlockSize%pageSize != 0 {
diff --git a/pkg/tcpip/link/fdbased/packet_dispatchers.go b/pkg/tcpip/link/fdbased/packet_dispatchers.go
index ab2855a63..fab34c5fa 100644
--- a/pkg/tcpip/link/fdbased/packet_dispatchers.go
+++ b/pkg/tcpip/link/fdbased/packet_dispatchers.go
@@ -18,6 +18,8 @@
 package fdbased
 
 import (
+	"fmt"
+
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/buffer"
@@ -114,9 +116,36 @@ func (b *iovecBuffer) pullViews(n int) buffer.VectorisedView {
 	return buffer.NewVectorisedView(n, views)
 }
 
+// stopFd is an eventfd used to signal the stop of a dispatcher.
+type stopFd struct {
+	efd int
+}
+
+func newStopFd() (stopFd, error) {
+	efd, err := unix.Eventfd(0, unix.EFD_NONBLOCK)
+	if err != nil {
+		return stopFd{efd: -1}, fmt.Errorf("failed to create eventfd: %w", err)
+	}
+	return stopFd{efd: efd}, nil
+}
+
+// stop writes to the eventfd and notifies the dispatcher to stop. It does not
+// block.
+func (s *stopFd) stop() {
+	increment := []byte{1, 0, 0, 0, 0, 0, 0, 0}
+	if n, err := unix.Write(s.efd, increment); n != len(increment) || err != nil {
+		// There are two possible errors documented in eventfd(2) for writing:
+		// 1. We are writing 8 bytes and not 0xffffffffffffff, thus no EINVAL.
+		// 2. stop is only supposed to be called once, it can't reach the limit,
+		// thus no EAGAIN.
+		panic(fmt.Sprintf("write(efd) = (%d, %s), want (%d, nil)", n, err, len(increment)))
+	}
+}
+
 // readVDispatcher uses readv() system call to read inbound packets and
 // dispatches them.
 type readVDispatcher struct {
+	stopFd
 	// fd is the file descriptor used to send and receive packets.
 	fd int
 
@@ -128,7 +157,15 @@ type readVDispatcher struct {
 }
 
 func newReadVDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
-	d := &readVDispatcher{fd: fd, e: e}
+	stopFd, err := newStopFd()
+	if err != nil {
+		return nil, err
+	}
+	d := &readVDispatcher{
+		stopFd: stopFd,
+		fd:     fd,
+		e:      e,
+	}
 	skipsVnetHdr := d.e.gsoKind == stack.HWGSOSupported
 	d.buf = newIovecBuffer(BufConfig, skipsVnetHdr)
 	return d, nil
@@ -136,8 +173,8 @@ func newReadVDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
 
 // dispatch reads one packet from the file descriptor and dispatches it.
 func (d *readVDispatcher) dispatch() (bool, tcpip.Error) {
-	n, err := rawfile.BlockingReadv(d.fd, d.buf.nextIovecs())
-	if n == 0 || err != nil {
+	n, err := rawfile.BlockingReadvUntilStopped(d.efd, d.fd, d.buf.nextIovecs())
+	if n <= 0 || err != nil {
 		return false, err
 	}
 
@@ -184,6 +221,7 @@ func (d *readVDispatcher) dispatch() (bool, tcpip.Error) {
 // recvMMsgDispatcher uses the recvmmsg system call to read inbound packets and
 // dispatches them.
 type recvMMsgDispatcher struct {
+	stopFd
 	// fd is the file descriptor used to send and receive packets.
 	fd int
 
@@ -207,7 +245,12 @@ const (
 )
 
 func newRecvMMsgDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
+	stopFd, err := newStopFd()
+	if err != nil {
+		return nil, err
+	}
 	d := &recvMMsgDispatcher{
+		stopFd:  stopFd,
 		fd:      fd,
 		e:       e,
 		bufs:    make([]*iovecBuffer, MaxMsgsPerRecv),
@@ -235,8 +278,8 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) {
 		d.msgHdrs[k].Msg.SetIovlen(iovLen)
 	}
 
-	nMsgs, err := rawfile.BlockingRecvMMsg(d.fd, d.msgHdrs)
-	if err != nil {
+	nMsgs, err := rawfile.BlockingRecvMMsgUntilStopped(d.efd, d.fd, d.msgHdrs)
+	if nMsgs == -1 || err != nil {
 		return false, err
 	}
 	// Process each of received packets.