Merge pull request #399 from cloudwego/develop

jayantxie · web-flow · commit 9715df530281 · 2025-06-16T16:24:39.000+08:00
chore: release v0.7.1
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
@@ -31,7 +31,7 @@ Before you submit your Pull Request (PR) consider the following guidelines:
 3. [Fork](https://docs.github.com/en/github/getting-started-with-github/fork-a-repo) the cloudwego/netpoll repo.
 4. In your forked repository, make your changes in a new git branch:
     ```
-    git checkout -b my-fix-branch develop
+    git checkout -b my-fix-branch main
     ```
 5. Create your patch, including appropriate test cases.
 6. Follow our [Style Guides](#code-style-guides).
@@ -41,7 +41,7 @@ Before you submit your Pull Request (PR) consider the following guidelines:
     ```
     git push origin my-fix-branch
     ```
-9. In GitHub, send a pull request to `netpoll:develop`
+9. In GitHub, send a pull request to `netpoll:main`
 
 ## Contribution Prerequisites
 - Our development environment keeps up with [Go Official](https://golang.org/project/).
@@ -50,9 +50,5 @@ Before you submit your Pull Request (PR) consider the following guidelines:
 - Maybe you need familiar with [Actions](https://github.com/features/actions)(our default workflow tool).
 
 ## Code Style Guides
-Also see [Pingcap General advice](https://pingcap.github.io/style-guide/general.html).
-
-Good resources:
 - [Effective Go](https://golang.org/doc/effective_go)
 - [Go Code Review Comments](https://github.com/golang/go/wiki/CodeReviewComments)
-- [Uber Go Style Guide](https://github.com/uber-go/guide/blob/master/style.md)
diff --git a/connection_impl.go b/connection_impl.go
@@ -27,8 +27,6 @@ import (
 type connState = int32
 
 const (
-	defaultZeroCopyTimeoutSec = 60
-
 	connStateNone         = 0
 	connStateConnected    = 1
 	connStateDisconnected = 2
@@ -39,21 +37,20 @@ type connection struct {
 	netFD
 	onEvent
 	locker
-	operator        *FDOperator
-	readTimeout     time.Duration
-	readTimer       *time.Timer
-	readTrigger     chan error
-	waitReadSize    int64
-	writeTimeout    time.Duration
-	writeTimer      *time.Timer
-	writeTrigger    chan error
-	inputBuffer     *LinkBuffer
-	outputBuffer    *LinkBuffer
-	outputBarrier   *barrier
-	supportZeroCopy bool
-	maxSize         int       // The maximum size of data between two Release().
-	bookSize        int       // The size of data that can be read at once.
-	state           connState // Connection state should be changed sequentially.
+	operator      *FDOperator
+	readTimeout   time.Duration
+	readTimer     *time.Timer
+	readTrigger   chan error
+	waitReadSize  int64
+	writeTimeout  time.Duration
+	writeTimer    *time.Timer
+	writeTrigger  chan error
+	inputBuffer   *LinkBuffer
+	outputBuffer  *LinkBuffer
+	outputBarrier *barrier
+	maxSize       int       // The maximum size of data between two Release().
+	bookSize      int       // The size of data that can be read at once.
+	state         connState // Connection state should be changed sequentially.
 }
 
 var (
@@ -351,10 +348,6 @@ func (c *connection) init(conn Conn, opts *options) (err error) {
 	case "tcp", "tcp4", "tcp6":
 		setTCPNoDelay(c.fd, true)
 	}
-	// check zero-copy
-	if setZeroCopy(c.fd) == nil && setBlockZeroCopySend(c.fd, defaultZeroCopyTimeoutSec, 0) == nil {
-		c.supportZeroCopy = true
-	}
 
 	// connection initialized and prepare options
 	return c.onPrepare(opts)
@@ -483,9 +476,8 @@ func (c *connection) flush() error {
 	if c.outputBuffer.IsEmpty() {
 		return nil
 	}
-	// TODO: Let the upper layer pass in whether to use ZeroCopy.
 	bs := c.outputBuffer.GetBytes(c.outputBarrier.bs)
-	n, err := sendmsg(c.fd, bs, c.outputBarrier.ivs, false && c.supportZeroCopy)
+	n, err := sendmsg(c.fd, bs, c.outputBarrier.ivs, false)
 	if err != nil && err != syscall.EAGAIN {
 		return Exception(err, "when flush")
 	}
diff --git a/connection_lock.go b/connection_lock.go
@@ -59,11 +59,11 @@ type locker struct {
 }
 
 func (l *locker) closeBy(w who) (success bool) {
-	return atomic.CompareAndSwapInt32(&l.keychain[closing], 0, int32(w))
+	return atomic.CompareAndSwapInt32(&l.keychain[closing], 0, w)
 }
 
 func (l *locker) isCloseBy(w who) (yes bool) {
-	return atomic.LoadInt32(&l.keychain[closing]) == int32(w)
+	return atomic.LoadInt32(&l.keychain[closing]) == w
 }
 
 func (l *locker) status(k key) int32 {
diff --git a/connection_reactor.go b/connection_reactor.go
@@ -119,13 +119,13 @@ func (c *connection) inputAck(n int) (err error) {
 }
 
 // outputs implements FDOperator.
-func (c *connection) outputs(vs [][]byte) (rs [][]byte, supportZeroCopy bool) {
+func (c *connection) outputs(vs [][]byte) (rs [][]byte, _ bool) {
 	if c.outputBuffer.IsEmpty() {
 		c.rw2r()
-		return rs, c.supportZeroCopy
+		return rs, false
 	}
 	rs = c.outputBuffer.GetBytes(vs)
-	return rs, c.supportZeroCopy
+	return rs, false
 }
 
 // outputAck implements FDOperator.
diff --git a/connection_test.go b/connection_test.go
@@ -688,6 +688,7 @@ func TestConnectionServerClose(t *testing.T) {
 		WithOnPrepare(func(connection Connection) context.Context {
 			// t.Logf("server.OnPrepare: addr=%s", connection.RemoteAddr())
 			defer wg.Done()
+			//nolint:staticcheck // SA1029 no built-in type string as key
 			return context.WithValue(context.Background(), "prepare", "true")
 		}),
 	)
diff --git a/fd_operator.go b/fd_operator.go
@@ -36,6 +36,7 @@ type FDOperator struct {
 	InputAck func(n int) (err error)
 
 	// Outputs will locked if len(rs) > 0, which need unlocked by OutputAck.
+	// supportZeroCopy is not implemented, and it will be ignored
 	Outputs   func(vs [][]byte) (rs [][]byte, supportZeroCopy bool)
 	OutputAck func(n int) (err error)
 
diff --git a/net_listener_test.go b/net_listener_test.go
@@ -80,7 +80,7 @@ func TestListenerDialer(t *testing.T) {
 		conn.SetOnRequest(onRequest)
 
 		MustNil(t, err)
-		n, err := conn.Write([]byte(msg))
+		n, err := conn.Write(msg)
 		MustNil(t, err)
 		Equal(t, n, len(msg))
 		time.Sleep(10 * time.Millisecond)
diff --git a/net_sock.go b/net_sock.go
@@ -40,7 +40,7 @@ type sockaddr interface {
 }
 
 func internetSocket(ctx context.Context, net string, laddr, raddr sockaddr, sotype, proto int, mode string) (conn *netFD, err error) {
-	if (runtime.GOOS == "aix" || runtime.GOOS == "windows" || runtime.GOOS == "openbsd" || runtime.GOOS == "nacl") && raddr.isWildcard() {
+	if (runtime.GOOS == "aix" || runtime.GOOS == "openbsd" || runtime.GOOS == "nacl") && raddr.isWildcard() {
 		raddr = raddr.toLocal(net)
 	}
 	family, ipv6only := favoriteAddrFamily(net, laddr, raddr)
diff --git a/nocopy_linkbuffer.go b/nocopy_linkbuffer.go
@@ -403,7 +403,7 @@ func (b *UnsafeLinkBuffer) MallocAck(n int) (err error) {
 	}
 	// discard the rest
 	for node := b.write.next; node != nil; node = node.next {
-		node.off, node.malloc, node.refer, node.buf = 0, 0, 1, node.buf[:0]
+		node.malloc, node.refer, node.buf = node.off, 1, node.buf[:node.off]
 	}
 	return nil
 }
diff --git a/nocopy_linkbuffer_test.go b/nocopy_linkbuffer_test.go
@@ -21,6 +21,7 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"reflect"
 	"runtime"
 	"sync/atomic"
 	"testing"
@@ -778,6 +779,34 @@ func TestLinkBufferPeekOutOfMemory(t *testing.T) {
 	}
 }
 
+func TestMallocAck(t *testing.T) {
+	sLen := 1024 * 7
+	buf1 := []byte{1, 2, 3, 4}
+	buf2 := []byte{5, 6, 7, 8}
+	lb := NewLinkBuffer(0)
+
+	buf, err := lb.Malloc(4 + sLen)
+	MustNil(t, err)
+	copy(buf[:4], buf1)
+	s := make([]byte, sLen)
+	err = lb.WriteDirect(s, sLen)
+	MustNil(t, err)
+
+	err = lb.MallocAck(4 + sLen)
+	MustNil(t, err)
+	lb.Flush()
+
+	buf, err = lb.Malloc(4)
+	MustNil(t, err)
+	copy(buf[:4], buf2)
+	lb.Flush()
+
+	buf, err = lb.Next(8 + sLen)
+	MustNil(t, err)
+
+	MustTrue(t, reflect.DeepEqual(buf, append(append(buf1, s...), buf2...)))
+}
+
 func BenchmarkStringToSliceByte(b *testing.B) {
 	b.StopTimer()
 	s := "hello world"
diff --git a/poll_default_bsd.go b/poll_default_bsd.go
@@ -52,7 +52,7 @@ func openDefaultPoll() (*defaultPoll, error) {
 type defaultPoll struct {
 	fd      int
 	trigger uint32
-	m       sync.Map       // only used in go:race
+	m       sync.Map       //nolint:unused // only used in go:race
 	opcache *operatorCache // operator cache
 	hups    []func(p Poll) error
 }
diff --git a/poll_default_linux.go b/poll_default_linux.go
@@ -199,10 +199,9 @@ func (p *defaultPoll) handler(events []epollevent) (closed bool) {
 				operator.OnWrite(p)
 			} else if operator.Outputs != nil {
 				// for connection
-				bs, supportZeroCopy := operator.Outputs(p.barriers[i].bs)
+				bs, _ := operator.Outputs(p.barriers[i].bs)
 				if len(bs) > 0 {
-					// TODO: Let the upper layer pass in whether to use ZeroCopy.
-					n, err := iosend(operator.FD, bs, p.barriers[i].ivs, false && supportZeroCopy)
+					n, err := iosend(operator.FD, bs, p.barriers[i].ivs, false)
 					operator.OutputAck(n)
 					if err != nil {
 						p.appendHup(operator)
diff --git a/poll_manager.go b/poll_manager.go
@@ -144,6 +144,7 @@ START:
 	// m.Run() will finish very quickly, so will not many goroutines block on Pick.
 	_ = m.Run()
 
+	//nolint:staticcheck // SA9003: empty branch
 	if !atomic.CompareAndSwapInt32(&m.status, managerInitializing, managerInitialized) {
 		// SetNumLoops called during m.Run() which cause CAS failed
 		// The polls will be adjusted next Pick
diff --git a/sys_exec.go b/sys_exec.go
@@ -72,7 +72,7 @@ func writev(fd int, bs [][]byte, ivs []syscall.Iovec) (n int, err error) {
 	r, _, e := syscall.RawSyscall(syscall.SYS_WRITEV, uintptr(fd), uintptr(unsafe.Pointer(&ivs[0])), uintptr(iovLen))
 	resetIovecs(bs, ivs[:iovLen])
 	if e != 0 {
-		return int(r), syscall.Errno(e)
+		return int(r), e
 	}
 	return int(r), nil
 }
@@ -88,7 +88,7 @@ func readv(fd int, bs [][]byte, ivs []syscall.Iovec) (n int, err error) {
 	r, _, e := syscall.RawSyscall(syscall.SYS_READV, uintptr(fd), uintptr(unsafe.Pointer(&ivs[0])), uintptr(iovLen))
 	resetIovecs(bs, ivs[:iovLen])
 	if e != 0 {
-		return int(r), syscall.Errno(e)
+		return int(r), e
 	}
 	return int(r), nil
 }
diff --git a/sys_sendmsg_bsd.go b/sys_sendmsg_bsd.go
@@ -22,8 +22,6 @@ import (
 	"unsafe"
 )
 
-var supportZeroCopySend bool
-
 // sendmsg wraps the sendmsg system call.
 // Must len(iovs) >= len(vs)
 func sendmsg(fd int, bs [][]byte, ivs []syscall.Iovec, zerocopy bool) (n int, err error) {
@@ -39,7 +37,7 @@ func sendmsg(fd int, bs [][]byte, ivs []syscall.Iovec, zerocopy bool) (n int, er
 	r, _, e := syscall.RawSyscall(syscall.SYS_SENDMSG, uintptr(fd), uintptr(unsafe.Pointer(&msghdr)), uintptr(0))
 	resetIovecs(bs, ivs[:iovLen])
 	if e != 0 {
-		return int(r), syscall.Errno(e)
+		return int(r), e
 	}
 	return int(r), nil
 }
diff --git a/sys_sendmsg_linux.go b/sys_sendmsg_linux.go
@@ -40,11 +40,7 @@ func sendmsg(fd int, bs [][]byte, ivs []syscall.Iovec, zerocopy bool) (n int, er
 		Iov:    &ivs[0],
 		Iovlen: uint64(iovLen),
 	}
-	var flags uintptr
-	if zerocopy {
-		flags = MSG_ZEROCOPY
-	}
-	r, _, e := syscall.RawSyscall(syscall.SYS_SENDMSG, uintptr(fd), uintptr(unsafe.Pointer(&msghdr)), flags)
+	r, _, e := syscall.RawSyscall(syscall.SYS_SENDMSG, uintptr(fd), uintptr(unsafe.Pointer(&msghdr)), 0)
 	resetIovecs(bs, ivs[:iovLen])
 	if e != 0 {
 		return int(r), syscall.Errno(e)
diff --git a/sys_zerocopy_bsd.go b/sys_zerocopy_bsd.go
diff --git a/sys_zerocopy_linux.go b/sys_zerocopy_linux.go

Original file line number	Diff line number	Diff line change
`@@ -59,11 +59,11 @@ type locker struct {`
`59`	`59`	`}`
`60`	`60`
`61`	`61`	`func (l *locker) closeBy(w who) (success bool) {`
`62`		`- return atomic.CompareAndSwapInt32(&l.keychain[closing], 0, int32(w))`
	`62`	`+ return atomic.CompareAndSwapInt32(&l.keychain[closing], 0, w)`
`63`	`63`	`}`
`64`	`64`
`65`	`65`	`func (l *locker) isCloseBy(w who) (yes bool) {`
`66`		`- return atomic.LoadInt32(&l.keychain[closing]) == int32(w)`
	`66`	`+ return atomic.LoadInt32(&l.keychain[closing]) == w`
`67`	`67`	`}`
`68`	`68`
`69`	`69`	`func (l *locker) status(k key) int32 {`
Original file line number	Diff line number	Diff line change
`@@ -119,13 +119,13 @@ func (c *connection) inputAck(n int) (err error) {`
`119`	`119`	`}`
`120`	`120`
`121`	`121`	`// outputs implements FDOperator.`
`122`		`-func (c *connection) outputs(vs [][]byte) (rs [][]byte, supportZeroCopy bool) {`
	`122`	`+func (c *connection) outputs(vs [][]byte) (rs [][]byte, _ bool) {`
`123`	`123`	`if c.outputBuffer.IsEmpty() {`
`124`	`124`	`c.rw2r()`
`125`		`- return rs, c.supportZeroCopy`
	`125`	`+ return rs, false`
`126`	`126`	`}`
`127`	`127`	`rs = c.outputBuffer.GetBytes(vs)`
`128`		`- return rs, c.supportZeroCopy`
	`128`	`+ return rs, false`
`129`	`129`	`}`
`130`	`130`
`131`	`131`	`// outputAck implements FDOperator.`
Original file line number	Diff line number	Diff line change
`@@ -688,6 +688,7 @@ func TestConnectionServerClose(t *testing.T) {`
`688`	`688`	`WithOnPrepare(func(connection Connection) context.Context {`
`689`	`689`	`// t.Logf("server.OnPrepare: addr=%s", connection.RemoteAddr())`
`690`	`690`	`defer wg.Done()`
	`691`	`+ //nolint:staticcheck // SA1029 no built-in type string as key`
`691`	`692`	`return context.WithValue(context.Background(), "prepare", "true")`
`692`	`693`	`}),`
`693`	`694`	`)`
Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@ type sockaddr interface {`
`40`	`40`	`}`
`41`	`41`
`42`	`42`	`func internetSocket(ctx context.Context, net string, laddr, raddr sockaddr, sotype, proto int, mode string) (conn *netFD, err error) {`
`43`		`- if (runtime.GOOS == "aix" \|\| runtime.GOOS == "windows" \|\| runtime.GOOS == "openbsd" \|\| runtime.GOOS == "nacl") && raddr.isWildcard() {`
	`43`	`+ if (runtime.GOOS == "aix" \|\| runtime.GOOS == "openbsd" \|\| runtime.GOOS == "nacl") && raddr.isWildcard() {`
`44`	`44`	`raddr = raddr.toLocal(net)`
`45`	`45`	`}`
`46`	`46`	`family, ipv6only := favoriteAddrFamily(net, laddr, raddr)`
Original file line number	Diff line number	Diff line change
`@@ -403,7 +403,7 @@ func (b *UnsafeLinkBuffer) MallocAck(n int) (err error) {`
`403`	`403`	`}`
`404`	`404`	`// discard the rest`
`405`	`405`	`for node := b.write.next; node != nil; node = node.next {`
`406`		`- node.off, node.malloc, node.refer, node.buf = 0, 0, 1, node.buf[:0]`
	`406`	`+ node.malloc, node.refer, node.buf = node.off, 1, node.buf[:node.off]`
`407`	`407`	`}`
`408`	`408`	`return nil`
`409`	`409`	`}`