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

When listening for IPv6 address changes, I found that subscribe is not returning when there is message in the socket. After some researching, I found that libnl suggest setting socket to non-blocking when subscribing to groups. (Ref)[https://www.infradead.org/~tgr/libnl/doc/core.html#:~:text=best%20to%20put%20the%20socket%20in%20non-blocking%20mode]

Also fixed test related to BareUDP, which requires "bareudp" kmod. (Ref)[https://www.kernelconfig.io/config_bareudp]

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

@@ -2427,6 +2427,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

@@ -1607,10 +1607,8 @@ func TestLinkAddDelVxlanFlowBased(t *testing.T) {
 }
 
 func TestLinkAddDelBareUDP(t *testing.T) {
-	if os.Getenv("CI") == "true" {
+	minKernelRequired(t, 5, 1)
-		t.Skipf("Fails in CI due to operation not supported (missing kernel module?)")
+	setUpNetlinkTestWithKModule(t, "bareudp")
-	}
-	minKernelRequired(t, 5, 8)
 	tearDown := setUpNetlinkTest(t)
 	defer tearDown()
 
@@ -1637,6 +1635,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

@@ -727,6 +727,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

@@ -1653,6 +1653,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

@@ -433,6 +433,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
 			}