Change behavior of Subscribe to non-blocking. Fix test.

addr_linux.go

@@ -365,6 +365,9 @@ func addrSubscribeAt(newNs, curNs netns.NsHandle, ch chan<- AddrUpdate, done <-c
 		for {
 			msgs, from, err := s.Receive()
 			if err != nil {
+				if err == syscall.EAGAIN {
+					continue
+				}
 				if cberr != nil {
 					cberr(fmt.Errorf("Receive failed: %v",
 						err))

link_linux.go

@@ -996,28 +996,28 @@ func LinkSetXdpFdWithFlags(link Link, fd, flags int) error {
 // LinkSetGSOMaxSegs sets the GSO maximum segment count of the link device.
 // Equivalent to: `ip link set $link gso_max_segs $maxSegs`
 func LinkSetGSOMaxSegs(link Link, maxSegs int) error {
-       return pkgHandle.LinkSetGSOMaxSegs(link, maxSegs)
+	return pkgHandle.LinkSetGSOMaxSegs(link, maxSegs)
 }
 
 // LinkSetGSOMaxSegs sets the GSO maximum segment count of the link device.
 // Equivalent to: `ip link set $link gso_max_segs $maxSegs`
 func (h *Handle) LinkSetGSOMaxSegs(link Link, maxSize int) error {
-       base := link.Attrs()
-       h.ensureIndex(base)
-       req := h.newNetlinkRequest(unix.RTM_SETLINK, unix.NLM_F_ACK)
+	base := link.Attrs()
+	h.ensureIndex(base)
+	req := h.newNetlinkRequest(unix.RTM_SETLINK, unix.NLM_F_ACK)
 
-       msg := nl.NewIfInfomsg(unix.AF_UNSPEC)
-       msg.Index = int32(base.Index)
-       req.AddData(msg)
+	msg := nl.NewIfInfomsg(unix.AF_UNSPEC)
+	msg.Index = int32(base.Index)
+	req.AddData(msg)
 
-       b := make([]byte, 4)
-       native.PutUint32(b, uint32(maxSize))
+	b := make([]byte, 4)
+	native.PutUint32(b, uint32(maxSize))
 
-       data := nl.NewRtAttr(unix.IFLA_GSO_MAX_SEGS, b)
-       req.AddData(data)
+	data := nl.NewRtAttr(unix.IFLA_GSO_MAX_SEGS, b)
+	req.AddData(data)
 
-       _, err := req.Execute(unix.NETLINK_ROUTE, 0)
-       return err
+	_, err := req.Execute(unix.NETLINK_ROUTE, 0)
+	return err
 }
 
 // LinkSetGSOMaxSize sets the IPv6 GSO maximum size of the link device.
@@ -2353,6 +2353,9 @@ func linkSubscribeAt(newNs, curNs netns.NsHandle, ch chan<- LinkUpdate, done <-c
 		for {
 			msgs, from, err := s.Receive()
 			if err != nil {
+				if err == syscall.EAGAIN {
+					continue
+				}
 				if cberr != nil {
 					cberr(fmt.Errorf("Receive failed: %v",
 						err))

link_test.go

@@ -1605,10 +1605,8 @@ func TestLinkAddDelVxlanFlowBased(t *testing.T) {
 }
 
 func TestLinkAddDelBareUDP(t *testing.T) {
-	if os.Getenv("CI") == "true" {
-		t.Skipf("Fails in CI due to operation not supported (missing kernel module?)")
-	}
-	minKernelRequired(t, 5, 8)
+	minKernelRequired(t, 5, 1)
+	setUpNetlinkTestWithKModule(t, "bareudp")
 	tearDown := setUpNetlinkTest(t)
 	defer tearDown()
 
@@ -1635,6 +1633,7 @@ func TestBareUDPCompareToIP(t *testing.T) {
 	}
 	// requires iproute2 >= 5.10
 	minKernelRequired(t, 5, 9)
+	setUpNetlinkTestWithKModule(t, "bareudp")
 	ns, tearDown := setUpNamedNetlinkTest(t)
 	defer tearDown()
 

neigh_linux.go

@@ -416,6 +416,9 @@ func neighSubscribeAt(newNs, curNs netns.NsHandle, ch chan<- NeighUpdate, done <
 		for {
 			msgs, from, err := s.Receive()
 			if err != nil {
+				if err == syscall.EAGAIN {
+					continue
+				}
 				if cberr != nil {
 					cberr(err)
 				}

nl/nl_linux.go

@@ -722,6 +722,16 @@ func Subscribe(protocol int, groups ...uint) (*NetlinkSocket, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	// Sometimes (socket_linux.go:SocketGet), Subscribe is used to create a socket
+	// that subscirbed to no groups. So we don't need to set nonblock there.
+	if len(groups) > 0 {
+		if err := unix.SetNonblock(fd, true); err != nil {
+			unix.Close(fd)
+			return nil, err
+		}
+	}
+
 	s := &NetlinkSocket{
 		fd: int32(fd),
 	}

proc_event_linux.go

@@ -148,6 +148,9 @@ func ProcEventMonitor(ch chan<- ProcEvent, done <-chan struct{}, errorChan chan<
 		for {
 			msgs, from, err := s.Receive()
 			if err != nil {
+				if err == syscall.EAGAIN {
+					continue
+				}
 				errorChan <- err
 				return
 			}

route_linux.go

@@ -1527,6 +1527,9 @@ func routeSubscribeAt(newNs, curNs netns.NsHandle, ch chan<- RouteUpdate, done <
 		for {
 			msgs, from, err := s.Receive()
 			if err != nil {
+				if err == syscall.EAGAIN {
+					continue
+				}
 				if cberr != nil {
 					cberr(fmt.Errorf("Receive failed: %v",
 						err))

socket_linux.go

@@ -237,6 +237,9 @@ loop:
 	for {
 		msgs, from, err := s.Receive()
 		if err != nil {
+			if err == syscall.EAGAIN {
+				continue
+			}
 			return err
 		}
 		if from.Pid != nl.PidKernel {

xfrm_monitor_linux.go

@@ -2,6 +2,7 @@ package netlink
 
 import (
 	"fmt"
+	"syscall"
 
 	"github.com/vishvananda/netlink/nl"
 	"github.com/vishvananda/netns"
@@ -56,6 +57,9 @@ func XfrmMonitor(ch chan<- XfrmMsg, done <-chan struct{}, errorChan chan<- error
 		for {
 			msgs, from, err := s.Receive()
 			if err != nil {
+				if err == syscall.EAGAIN {
+					continue
+				}
 				errorChan <- err
 				return
 			}