minor: v2.10.0

.github/workflows/test.yml

@@ -49,7 +49,7 @@ jobs:
       - name: Setup and run golangci-lint
         uses: golangci/golangci-lint-action@v7
         with:
-          version: v2.1.6
+          version: v2.12.2
           args: -v -E gocritic -E misspell -E revive -E godot --timeout 5m
   test:
     needs: lint

.github/workflows/test_gc_opt.yml

@@ -49,7 +49,7 @@ jobs:
       - name: Setup and run golangci-lint
         uses: golangci/golangci-lint-action@v7
         with:
-          version: v2.1.6
+          version: v2.12.2
           args: -v -E gocritic -E misspell -E revive -E godot --timeout 5m
   test:
     needs: lint

.github/workflows/test_poll_opt.yml

@@ -48,7 +48,7 @@ jobs:
       - name: Setup and run golangci-lint
         uses: golangci/golangci-lint-action@v7
         with:
-          version: v2.1.6
+          version: v2.12.2
           args: -v -E gocritic -E misspell -E revive -E godot
   test:
     needs: lint

.github/workflows/test_poll_opt_gc_opt.yml

@@ -48,7 +48,7 @@ jobs:
       - name: Setup and run golangci-lint
         uses: golangci/golangci-lint-action@v7
         with:
-          version: v2.1.6
+          version: v2.12.2
           args: -v -E gocritic -E misspell -E revive -E godot
   test:
     needs: lint

client_test.go

@@ -796,3 +796,156 @@ func (cli clientReadOnEOF) OnTraffic(c Conn) (action Action) {
 	}{data: data, err: err}
 	return None
 }
+
+// TestClientSafeContext verifies that a client-side Conn's SafeContext()/
+// SetSafeContext() behave correctly on connections created via
+// Client.Dial/DialContext/EnrollContext: DialContext's ctx argument is
+// reflected by both Context() and SafeContext(), switching the stored
+// concrete type (including nil) across calls never panics, and
+// SafeContext() is safe to call concurrently from a goroutine other than
+// the event-loop goroutine.
+func TestClientSafeContext(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:9978")
+	assert.NoError(t, err)
+	defer ln.Close() //nolint:errcheck
+
+	err = goPool.DefaultWorkerPool.Submit(func() {
+		for i := 0; i < 5; i++ {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			err = goPool.DefaultWorkerPool.Submit(func() {
+				buf := make([]byte, 4)
+				for {
+					_, err := io.ReadFull(conn, buf)
+					if err != nil {
+						conn.Close() //nolint:errcheck
+						return
+					}
+					_, err = conn.Write([]byte("pong"))
+					if err != nil {
+						conn.Close() //nolint:errcheck
+						return
+					}
+				}
+			})
+			assert.NoError(t, err)
+		}
+	})
+	assert.NoError(t, err)
+
+	ev := &clientSafeContextEvents{tester: t, done: make(chan struct{})}
+	cli, err := NewClient(ev)
+	assert.NoError(t, err)
+	defer cli.Stop() //nolint:errcheck
+
+	err = cli.Start()
+	assert.NoError(t, err)
+
+	initialCtx := &safeCtxPayloadA{n: 42}
+	ev.initialCtx = initialCtx
+	c, err := cli.DialContext("tcp", "127.0.0.1:9978", initialCtx)
+	assert.NoError(t, err)
+
+	select {
+	case <-ev.done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("timeout waiting for test completion")
+	}
+
+	// release() (which nils out safeCtx) runs on the event-loop goroutine
+	// immediately after OnClose returns, racing with this goroutine waking
+	// up from ev.done, so poll briefly instead of asserting once.
+	assert.Eventually(t, func() bool {
+		return c.SafeContext() == nil
+	}, time.Second, time.Millisecond, "SafeContext() must be nil once the connection is released")
+}
+
+type clientSafeContextEvents struct {
+	BuiltinEventEngine
+	tester *testing.T
+	done   chan struct{}
+
+	initialCtx any
+
+	trafficCount   int32
+	backgroundHits int32
+	stopBackground chan struct{}
+}
+
+func (ev *clientSafeContextEvents) OnOpen(c Conn) (out []byte, action Action) {
+	// OnOpen always runs before any OnTraffic event for this connection, so
+	// asserting here is guaranteed to observe the dial-time ctx before it can
+	// be overwritten by OnTraffic's SetSafeContext calls.
+	assert.Equal(ev.tester, ev.initialCtx, c.Context())
+	assert.Equal(ev.tester, ev.initialCtx, c.SafeContext())
+
+	ev.stopBackground = make(chan struct{})
+	// Exercise SafeContext() concurrently from a goroutine other than the
+	// event-loop goroutine.
+	err := goPool.DefaultWorkerPool.Submit(func() {
+		for {
+			select {
+			case <-ev.stopBackground:
+				return
+			default:
+			}
+			switch v := c.SafeContext().(type) {
+			case nil:
+			case *safeCtxPayloadA:
+				_ = v.n
+			case *safeCtxPayloadB:
+				_ = v.s
+			case int:
+			default:
+				assert.Failf(ev.tester, "unexpected SafeContext type", "%T", v)
+			}
+			atomic.AddInt32(&ev.backgroundHits, 1)
+		}
+	})
+	assert.NoError(ev.tester, err)
+
+	out = []byte("ping")
+	return
+}
+
+func (ev *clientSafeContextEvents) OnTraffic(c Conn) (action Action) {
+	buf, err := c.Next(-1)
+	assert.NoError(ev.tester, err)
+	assert.Equal(ev.tester, "pong", string(buf))
+
+	n := atomic.AddInt32(&ev.trafficCount, 1)
+	switch n % 4 {
+	case 0:
+		c.SetSafeContext(&safeCtxPayloadA{n: n})
+	case 1:
+		c.SetSafeContext(&safeCtxPayloadB{s: "hi"})
+	case 2:
+		c.SetSafeContext(int(n))
+	case 3:
+		c.SetSafeContext(nil)
+	}
+
+	if n >= 5 {
+		action = Close
+		return
+	}
+
+	_, err = c.Write([]byte("ping"))
+	assert.NoError(ev.tester, err)
+	return
+}
+
+func (ev *clientSafeContextEvents) OnClose(c Conn, _ error) (action Action) {
+	// SafeContext() must still be safely callable (no panic) right up
+	// until release(), which happens after OnClose returns.
+	_ = c.SafeContext()
+
+	close(ev.stopBackground)
+	assert.Greater(ev.tester, atomic.LoadInt32(&ev.backgroundHits), int32(0),
+		"expected background goroutine to observe SafeContext() at least once")
+
+	close(ev.done)
+	return
+}

client_unix.go

@@ -274,7 +274,8 @@ func (cli *Client) EnrollContext(c net.Conn, ctx any) (Conn, error) {
 	default:
 		return nil, errorx.ErrUnsupportedProtocol
 	}
-	gc.ctx = ctx
+	gc.SetContext(ctx)
+	gc.SetSafeContext(ctx)
 
 	connOpened := make(chan struct{})
 	ccb := &connWithCallback{c: gc, cb: func() {

connection_unix.go

@@ -20,6 +20,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"sync/atomic"
 	"time"
 
 	"golang.org/x/sys/unix"
@@ -39,6 +40,7 @@ type conn struct {
 	fd             int                    // file descriptor
 	gfd            gfd.GFD                // gnet file descriptor
 	ctx            any                    // user-defined context
+	safeCtx        atomic.Pointer[any]    // safe user-defined context
 	remote         unix.Sockaddr          // remote socket address
 	proto          string                 // protocol name: "tcp", "udp", or "unix".
 	localAddr      net.Addr               // local addr
@@ -91,6 +93,7 @@ func (c *conn) release() {
 	c.opened = false
 	c.isEOF = false
 	c.ctx = nil
+	c.safeCtx.Store(nil)
 	c.buffer = nil
 	if addr, ok := c.localAddr.(*net.TCPAddr); ok && len(c.loop.listeners) == 0 && len(addr.Zone) > 0 {
 		bsPool.Put(bs.StringToBytes(addr.Zone))
@@ -559,3 +562,14 @@ func (*conn) SetReadDeadline(_ time.Time) error {
 func (*conn) SetWriteDeadline(_ time.Time) error {
 	return errorx.ErrUnsupportedOp
 }
+
+func (c *conn) SafeContext() (ctx any) {
+	if p := c.safeCtx.Load(); p != nil {
+		return *p
+	}
+	return nil
+}
+
+func (c *conn) SetSafeContext(ctx any) {
+	c.safeCtx.Store(&ctx)
+}

connection_windows.go

@@ -19,6 +19,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -52,14 +53,15 @@ type openConn struct {
 
 type conn struct {
 	pc            net.PacketConn
-	ctx           any                // user-defined context
-	loop          *eventloop         // owner event-loop
-	buffer        *bbPool.ByteBuffer // reuse memory of inbound data as a temporary buffer
-	cache         []byte             // temporary cache for the inbound data
-	rawConn       net.Conn           // original connection
-	localAddr     net.Addr           // local server addr
-	remoteAddr    net.Addr           // remote addr
-	inboundBuffer elastic.RingBuffer // buffer for data from the remote
+	ctx           any                 // user-defined context
+	safeCtx       atomic.Pointer[any] // safe user-defined context
+	loop          *eventloop          // owner event-loop
+	buffer        *bbPool.ByteBuffer  // reuse memory of inbound data as a temporary buffer
+	cache         []byte              // temporary cache for the inbound data
+	rawConn       net.Conn            // original connection
+	localAddr     net.Addr            // local server addr
+	remoteAddr    net.Addr            // remote addr
+	inboundBuffer elastic.RingBuffer  // buffer for data from the remote
 }
 
 func packTCPConn(c *conn, buf []byte) *tcpConn {
@@ -84,18 +86,21 @@ func packUDPConn(c *conn, buf []byte) *udpConn {
 }
 
 func newStreamConn(el *eventloop, nc net.Conn, ctx any) (c *conn) {
-	return &conn{
+	c = &conn{
 		ctx:        ctx,
 		loop:       el,
 		buffer:     bbPool.Get(),
 		rawConn:    nc,
 		localAddr:  nc.LocalAddr(),
 		remoteAddr: nc.RemoteAddr(),
 	}
+	c.SetSafeContext(ctx)
+	return c
 }
 
 func (c *conn) release() {
 	c.ctx = nil
+	c.safeCtx.Store(nil)
 	c.localAddr = nil
 	if c.rawConn != nil {
 		c.rawConn = nil
@@ -107,7 +112,7 @@ func (c *conn) release() {
 }
 
 func newUDPConn(el *eventloop, pc net.PacketConn, rc net.Conn, localAddr, remoteAddr net.Addr, ctx any) *conn {
-	return &conn{
+	c := &conn{
 		ctx:        ctx,
 		pc:         pc,
 		rawConn:    rc,
@@ -116,6 +121,8 @@ func newUDPConn(el *eventloop, pc net.PacketConn, rc net.Conn, localAddr, remote
 		localAddr:  localAddr,
 		remoteAddr: remoteAddr,
 	}
+	c.SetSafeContext(ctx)
+	return c
 }
 
 func (c *conn) resetBuffer() {
@@ -573,3 +580,14 @@ func (*conn) SetReadDeadline(_ time.Time) error {
 func (*conn) SetWriteDeadline(_ time.Time) error {
 	return errorx.ErrUnsupportedOp
 }
+
+func (c *conn) SafeContext() (ctx any) {
+	if p := c.safeCtx.Load(); p != nil {
+		return *p
+	}
+	return nil
+}
+
+func (c *conn) SetSafeContext(ctx any) {
+	c.safeCtx.Store(&ctx)
+}

eventloop_unix.go

@@ -183,7 +183,8 @@ func (el *eventloop) enroll(c net.Conn, addr net.Addr, ctx any) (resCh chan Regi
 			return
 		}
 
-		gc.ctx = ctx
+		gc.SetContext(ctx)
+		gc.SetSafeContext(ctx)
 
 		connOpened := make(chan struct{})
 		ccb := &connWithCallback{c: gc, cb: func() {

gnet.go

@@ -430,6 +430,10 @@ type Conn interface {
 	// you must invoke it within any method in EventHandler.
 	Context() (ctx any)
 
+	// SafeContext is the concurrency-safe version of Context that can
+	// be invoked on any goroutine.
+	SafeContext() (ctx any)
+
 	// EventLoop returns the event-loop that the connection belongs to.
 	// The returned EventLoop is concurrency-safe.
 	EventLoop() EventLoop
@@ -438,12 +442,20 @@ type Conn interface {
 	// you must invoke it within any method in EventHandler.
 	SetContext(ctx any)
 
+	// SetSafeContext is the concurrency-safe version of SetContext that can
+	// be invoked on any goroutine.
+	SetSafeContext(ctx any)
+
 	// LocalAddr is the connection's local socket address, it's not concurrency-safe,
-	// you must invoke it within any method in EventHandler.
+	// you must invoke it and use the returned value within any method in EventHandler.
+	// If you must use the returned value outside of EventHandler, you should make
+	// a copy of it by calling net.Addr.String().
 	LocalAddr() net.Addr
 
 	// RemoteAddr is the connection's remote address, it's not concurrency-safe,
-	// you must invoke it within any method in EventHandler.
+	// you must invoke it and use the returned value within any method in EventHandler.
+	// If you must use the returned value outside of EventHandler, you should make
+	// a copy of it by calling net.Addr.String().
 	RemoteAddr() net.Addr
 
 	// Wake triggers an OnTraffic event for the current connection, it's concurrency-safe.