feat: support Conn.SendTo for UDP (#694)

Fixes #686

Author: panjf2000

.github/workflows/cross-compile-bsd.yml

@@ -31,7 +31,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        go: ['1.20', '1.23']
+        go: ['1.20', '1.24']
         os:
           - ubuntu-latest
     name: Go ${{ matrix.go }} @ ${{ matrix.os }}

.github/workflows/test.yml

@@ -47,16 +47,16 @@ jobs:
           cache: false
 
       - name: Setup and run golangci-lint
-        uses: golangci/golangci-lint-action@v6
+        uses: golangci/golangci-lint-action@v7
         with:
-          version: v1.64.8
-          args: -v -E gofumpt -E gocritic -E misspell -E revive -E godot --timeout 5m
+          version: v2.1.5
+          args: -v -E gocritic -E misspell -E revive -E godot --timeout 5m
   test:
     needs: lint
     strategy:
       fail-fast: false
       matrix:
-        go: ['1.20', '1.23']
+        go: ['1.20', '1.24']
         os:
           - ubuntu-latest
           - macos-latest

.github/workflows/test_gc_opt.yml

@@ -47,16 +47,16 @@ jobs:
           cache: false
 
       - name: Setup and run golangci-lint
-        uses: golangci/golangci-lint-action@v6
+        uses: golangci/golangci-lint-action@v7
         with:
-          version: v1.64.8
-          args: -v -E gofumpt -E gocritic -E misspell -E revive -E godot --timeout 5m
+          version: v2.1.5
+          args: -v -E gocritic -E misspell -E revive -E godot --timeout 5m
   test:
     needs: lint
     strategy:
       fail-fast: false
       matrix:
-        go: ['1.20', '1.23']
+        go: ['1.20', '1.24']
         os:
           - ubuntu-latest
           - macos-latest

.github/workflows/test_poll_opt.yml

@@ -46,16 +46,16 @@ jobs:
           cache: false
 
       - name: Setup and run golangci-lint
-        uses: golangci/golangci-lint-action@v6
+        uses: golangci/golangci-lint-action@v7
         with:
-          version: v1.64.8
-          args: -v -E gofumpt -E gocritic -E misspell -E revive -E godot
+          version: v2.1.5
+          args: -v -E gocritic -E misspell -E revive -E godot
   test:
     needs: lint
     strategy:
       fail-fast: false
       matrix:
-        go: ['1.20', '1.23']
+        go: ['1.20', '1.24']
         os: [ubuntu-latest, macos-latest]
     name: Go ${{ matrix.go }} @ ${{ matrix.os }}
     runs-on: ${{ matrix.os }}

.github/workflows/test_poll_opt_gc_opt.yml

@@ -46,16 +46,16 @@ jobs:
           cache: false
 
       - name: Setup and run golangci-lint
-        uses: golangci/golangci-lint-action@v6
+        uses: golangci/golangci-lint-action@v7
         with:
-          version: v1.64.8
-          args: -v -E gofumpt -E gocritic -E misspell -E revive -E godot
+          version: v2.1.5
+          args: -v -E gocritic -E misspell -E revive -E godot
   test:
     needs: lint
     strategy:
       fail-fast: false
       matrix:
-        go: ['1.20', '1.23']
+        go: ['1.20', '1.24']
         os: [ubuntu-latest, macos-latest]
     name: Go ${{ matrix.go }} @ ${{ matrix.os }}
     runs-on: ${{ matrix.os }}

client_test.go

@@ -533,7 +533,7 @@ func startGnetClient(t *testing.T, cli *Client, network, addr string, multicore,
 		c, err = cli.DialContext(network, addr, handler)
 	}
 	require.NoError(t, err)
-	defer c.Close()
+	defer c.Close() //nolint:errcheck
 	err = c.Wake(nil)
 	require.NoError(t, err)
 	rspCh := handler.rspCh
@@ -688,7 +688,7 @@ func TestClientReadOnEOF(t *testing.T) {
 
 	ln, err := net.Listen("tcp", "127.0.0.1:9999")
 	assert.NoError(t, err)
-	defer ln.Close()
+	defer ln.Close() //nolint:errcheck
 
 	go func() {
 		for {
@@ -733,7 +733,7 @@ func TestClientReadOnEOF(t *testing.T) {
 }
 
 func process(conn net.Conn) {
-	defer conn.Close() //noliint:errcheck
+	defer conn.Close() //nolint:errcheck
 	buf := make([]byte, 8)
 	n, err := conn.Read(buf)
 	if err != nil {

client_unix.go

@@ -163,7 +163,7 @@ func (cli *Client) Enroll(c net.Conn) (Conn, error) {
 
 // EnrollContext is like Enroll but also accepts an empty interface ctx that can be obtained later via Conn.Context.
 func (cli *Client) EnrollContext(c net.Conn, ctx any) (Conn, error) {
-	defer c.Close()
+	defer c.Close() //nolint:errcheck
 
 	sc, ok := c.(syscall.Conn)
 	if !ok {
@@ -242,7 +242,7 @@ func (cli *Client) EnrollContext(c net.Conn, ctx any) (Conn, error) {
 	}}
 	err = cli.el.poller.Trigger(queue.HighPriority, cli.el.register, ccb)
 	if err != nil {
-		gc.Close()
+		_ = gc.Close()
 		return nil, err
 	}
 

connection_unix.go

@@ -29,7 +29,6 @@ import (
 	"github.com/panjf2000/gnet/v2/pkg/buffer/elastic"
 	errorx "github.com/panjf2000/gnet/v2/pkg/errors"
 	gio "github.com/panjf2000/gnet/v2/pkg/io"
-	"github.com/panjf2000/gnet/v2/pkg/logging"
 	"github.com/panjf2000/gnet/v2/pkg/netpoll"
 	bsPool "github.com/panjf2000/gnet/v2/pkg/pool/byteslice"
 	"github.com/panjf2000/gnet/v2/pkg/queue"
@@ -158,11 +157,7 @@ loop:
 			}
 			return
 		}
-		if err := c.loop.close(c, os.NewSyscallError("write", err)); err != nil {
-			logging.Errorf("failed to close connection(fd=%d,remote=%+v) on conn.write: %v",
-				c.fd, c.remoteAddr, err)
-		}
-		return 0, os.NewSyscallError("write", err)
+		return 0, c.loop.close(c, os.NewSyscallError("write", err))
 	}
 	data = data[sent:]
 	if isET && len(data) > 0 {
@@ -206,11 +201,7 @@ loop:
 			}
 			return
 		}
-		if err := c.loop.close(c, os.NewSyscallError("writev", err)); err != nil {
-			logging.Errorf("failed to close connection(fd=%d,remote=%+v) on conn.writev: %v",
-				c.fd, c.remoteAddr, err)
-		}
-		return 0, os.NewSyscallError("writev", err)
+		return 0, c.loop.close(c, os.NewSyscallError("writev", err))
 	}
 	pos := len(bs)
 	if remaining -= sent; remaining > 0 {
@@ -280,11 +271,20 @@ func (c *conn) asyncWritev(a any) (err error) {
 	return
 }
 
-func (c *conn) sendTo(buf []byte) error {
-	if c.remote == nil {
-		return unix.Send(c.fd, buf, 0)
+func (c *conn) sendTo(buf []byte, addr unix.Sockaddr) (n int, err error) {
+	defer func() {
+		if err != nil {
+			n = 0
+		}
+	}()
+
+	if addr != nil {
+		return len(buf), unix.Sendto(c.fd, buf, 0, addr)
+	}
+	if c.remote == nil { // connected UDP socket of client
+		return len(buf), unix.Send(c.fd, buf, 0)
 	}
-	return unix.Sendto(c.fd, buf, 0, c.remote)
+	return len(buf), unix.Sendto(c.fd, buf, 0, c.remote) // unconnected UDP socket of server
 }
 
 func (c *conn) resetBuffer() {
@@ -389,14 +389,24 @@ func (c *conn) Discard(n int) (int, error) {
 
 func (c *conn) Write(p []byte) (int, error) {
 	if c.isDatagram {
-		if err := c.sendTo(p); err != nil {
-			return 0, err
-		}
-		return len(p), nil
+		return c.sendTo(p, nil)
 	}
 	return c.write(p)
 }
 
+func (c *conn) SendTo(p []byte, addr net.Addr) (int, error) {
+	if !c.isDatagram {
+		return 0, errorx.ErrUnsupportedOp
+	}
+
+	sa := socket.NetAddrToSockaddr(addr)
+	if sa == nil {
+		return 0, errorx.ErrInvalidNetworkAddress
+	}
+
+	return c.sendTo(p, sa)
+}
+
 func (c *conn) Writev(bs [][]byte) (int, error) {
 	if c.isDatagram {
 		return 0, errorx.ErrUnsupportedOp
@@ -462,7 +472,7 @@ func (c *conn) SetKeepAlivePeriod(d time.Duration) error {
 
 func (c *conn) AsyncWrite(buf []byte, callback AsyncCallback) error {
 	if c.isDatagram {
-		err := c.sendTo(buf)
+		_, err := c.sendTo(buf, nil)
 		// TODO: it will not go asynchronously with UDP, so calling a callback is needless,
 		//  we may remove this branch in the future, please don't rely on the callback
 		// 	to do something important under UDP, if you're working with UDP, just call Conn.Write

connection_windows.go

@@ -222,7 +222,23 @@ func (c *conn) Write(p []byte) (int, error) {
 	return c.pc.WriteTo(p, c.remoteAddr)
 }
 
+func (c *conn) SendTo(p []byte, addr net.Addr) (int, error) {
+	if c.pc == nil {
+		return 0, errorx.ErrUnsupportedOp
+	}
+
+	if addr == nil {
+		return 0, errorx.ErrInvalidNetworkAddress
+	}
+
+	return c.pc.WriteTo(p, addr)
+}
+
 func (c *conn) Writev(bs [][]byte) (int, error) {
+	if c.pc != nil { // not available for UDP
+		return 0, errorx.ErrUnsupportedOp
+	}
+
 	if c.rawConn != nil {
 		bb := bbPool.Get()
 		defer bbPool.Put(bb)
@@ -443,6 +459,10 @@ func (c *conn) AsyncWrite(buf []byte, cb AsyncCallback) error {
 }
 
 func (c *conn) AsyncWritev(bs [][]byte, cb AsyncCallback) error {
+	if c.pc != nil {
+		return errorx.ErrUnsupportedOp
+	}
+
 	buf := bbPool.Get()
 	for _, b := range bs {
 		_, _ = buf.Write(b)

eventloop_unix_test.go

@@ -127,10 +127,7 @@ type benchmarkServerGC struct {
 func (s *benchmarkServerGC) OnBoot(eng Engine) (action Action) {
 	s.eng = eng
 	go func() {
-		for {
-			if s.eng.eng.eventLoops.len() == s.elNum && s.eng.CountConnections() == s.elNum*int(s.initConnCount) {
-				break
-			}
+		for s.eng.eng.eventLoops.len() != s.elNum || s.eng.CountConnections() != s.elNum*int(s.initConnCount) {
 			time.Sleep(time.Millisecond)
 		}
 		close(s.initOk)