cherry pick picks cherries

internal/poller/event/defs_linux_386.go

@@ -0,0 +1,15 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events uint32
+	Data   [8]byte // to match amd64
+}

internal/poller/event/defs_linux_amd64.go

@@ -0,0 +1,15 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events uint32
+	Data   [8]byte // unaligned uintptr
+}

internal/poller/event/defs_linux_arm.go

@@ -0,0 +1,16 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events uint32
+	_pad   uint32
+	Data   [8]byte // to match amd64
+}

internal/poller/event/defs_linux_arm64.go

@@ -0,0 +1,16 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events uint32
+	_pad   uint32
+	Data   [8]byte // to match amd64
+}

internal/poller/event/defs_linux_loong64.go

@@ -0,0 +1,16 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events    uint32
+	pad_cgo_0 [4]byte
+	Data      [8]byte // unaligned uintptr
+}

internal/poller/event/defs_linux_mips64x.go

@@ -0,0 +1,18 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+//go:build linux && (mips64 || mips64le)
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events    uint32
+	pad_cgo_0 [4]byte
+	Data      [8]byte // unaligned uintptr
+}

internal/poller/event/defs_linux_mipsx.go

@@ -0,0 +1,18 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+//go:build linux && (mips || mipsle)
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events    uint32
+	pad_cgo_0 [4]byte
+	Data      uint64
+}

internal/poller/event/defs_linux_ppc64x.go

@@ -0,0 +1,18 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+//go:build linux && (ppc64 || ppc64le)
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events    uint32
+	pad_cgo_0 [4]byte
+	Data      [8]byte // unaligned uintptr
+}

internal/poller/event/defs_linux_riscv64.go

@@ -0,0 +1,16 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events    uint32
+	pad_cgo_0 [4]byte
+	Data      [8]byte // unaligned uintptr
+}

internal/poller/event/defs_linux_s390x.go

@@ -0,0 +1,16 @@
+// Copyright 2022 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// This file may have been modified by THL A29 Limited ("Tencent Modifications").
+// All Tencent Modifications are Copyright (C) 2023 THL A29 Limited.
+
+// Package event provides definitions of event data.
+package event
+
+// EpollEvent defines epoll event data.
+type EpollEvent struct {
+	Events    uint32
+	pad_cgo_0 [4]byte
+	Data      [8]byte // unaligned uintptr
+}

internal/poller/poller.go

@@ -9,6 +9,11 @@ package poller
 
 import "fmt"
 
+// GoschedAfterEvent decides whether to call runtime.Gosched() after processing of each event
+// during epoll waiting handling.
+// This global variable can only be changed inside func init().
+var GoschedAfterEvent bool
+
 // Event defines the operation of poll.Control.
 type Event int
 

internal/poller/poller_epoll.go

@@ -19,6 +19,7 @@ import (
 	"github.com/pkg/errors"
 	"golang.org/x/sys/unix"
 	"trpc.group/trpc-go/tnet/internal/iovec"
+	"trpc.group/trpc-go/tnet/internal/poller/event"
 	"trpc.group/trpc-go/tnet/log"
 	"trpc.group/trpc-go/tnet/metrics"
 )
@@ -45,7 +46,7 @@ func newPoller(ignoreTaskError bool) (Poller, error) {
 	poller := &epoll{
 		fd:              fd,
 		desc:            desc,
-		events:          make([]epollevent, defaultEventCount),
+		events:          make([]event.EpollEvent, defaultEventCount),
 		ioData:          iovec.NewIOData(),
 		buf:             make([]byte, 8),
 		ignoreTaskError: ignoreTaskError,
@@ -61,13 +62,13 @@ type epoll struct {
 	desc            *Desc
 	ioData          iovec.IOData
 	buf             []byte
-	events          []epollevent
+	events          []event.EpollEvent
 	fd              int
 	notified        int32
 	ignoreTaskError bool
 }
 
-func epollWait(epfd int, events []epollevent, msec int) (n int, err error) {
+func epollWait(epfd int, events []event.EpollEvent, msec int) (n int, err error) {
 	var r0 uintptr
 	var _p0 = unsafe.Pointer(&events[0])
 	if msec == 0 {
@@ -118,7 +119,7 @@ func (ep *epoll) handle(n int) {
 	var wakeUp bool
 	for i := 0; i < n; i++ {
 		event := ep.events[i]
-		desc := *(**Desc)(unsafe.Pointer(&event.data))
+		desc := *(**Desc)(unsafe.Pointer(&event.Data))
 		if desc.FD == ep.desc.FD {
 			_, _ = unix.Read(ep.desc.FD, ep.buf)
 			wakeUp = true
@@ -128,11 +129,11 @@ func (ep *epoll) handle(n int) {
 		var inHup bool
 		// Read/Write and error events may be triggered at the same time,
 		// so use if/else instead of switch/case to determine them separately.
-		if event.events&(unix.EPOLLHUP|unix.EPOLLRDHUP|unix.EPOLLERR) != 0 {
+		if event.Events&(unix.EPOLLHUP|unix.EPOLLRDHUP|unix.EPOLLERR) != 0 {
 			inHup = true
 		}
-		readable := event.events&(unix.EPOLLIN|unix.EPOLLPRI) != 0
-		writable := event.events&(unix.EPOLLOUT) != 0
+		readable := event.Events&(unix.EPOLLIN|unix.EPOLLPRI) != 0
+		writable := event.Events&(unix.EPOLLOUT) != 0
 		// The handler function may change at runtime, so for consistency,
 		// we store them in a temporary variable.
 		onRead, onWrite, data := desc.OnRead, desc.OnWrite, desc.Data
@@ -157,6 +158,9 @@ func (ep *epoll) handle(n int) {
 		if inHup {
 			hups = append(hups, desc)
 		}
+		if GoschedAfterEvent {
+			runtime.Gosched()
+		}
 	}
 	if wakeUp {
 		ep.runAsyncTasks()
@@ -206,32 +210,32 @@ func (ep *epoll) Trigger(job Job) error {
 }
 
 // Control the event of Desc and the operations is defined by Event.
-func (ep *epoll) Control(desc *Desc, event Event) (err error) {
-	evt := &epollevent{}
-	*(**Desc)(unsafe.Pointer(&evt.data)) = desc
+func (ep *epoll) Control(desc *Desc, e Event) (err error) {
+	evt := &event.EpollEvent{}
+	*(**Desc)(unsafe.Pointer(&evt.Data)) = desc
 	defer func() {
 		if err != nil { // Prevent unconditional execution of fmt.Sprintf.
-			err = errors.Wrap(err, fmt.Sprintf("event: %s, connection may be closed", event))
+			err = errors.Wrap(err, fmt.Sprintf("event: %s, connection may be closed", e))
 		}
 	}()
-	switch event {
+	switch e {
 	case Readable:
-		evt.events = rflags
+		evt.Events = rflags
 		return ep.insert(desc.FD, evt)
 	case Writable:
-		evt.events = wflags
+		evt.Events = wflags
 		return ep.insert(desc.FD, evt)
 	case ReadWriteable:
-		evt.events = rflags | wflags
+		evt.Events = rflags | wflags
 		return ep.insert(desc.FD, evt)
 	case ModReadable:
-		evt.events = rflags
+		evt.Events = rflags
 		return ep.interest(desc.FD, evt)
 	case ModWritable:
-		evt.events = wflags
+		evt.Events = wflags
 		return ep.interest(desc.FD, evt)
 	case ModReadWriteable:
-		evt.events = rflags | wflags
+		evt.Events = rflags | wflags
 		return ep.interest(desc.FD, evt)
 	case Detach:
 		return ep.remove(desc.FD)
@@ -240,14 +244,14 @@ func (ep *epoll) Control(desc *Desc, event Event) (err error) {
 	}
 }
 
-func (ep *epoll) insert(fd int, event *epollevent) error {
+func (ep *epoll) insert(fd int, event *event.EpollEvent) error {
 	if err := epollCtl(ep.fd, unix.EPOLL_CTL_ADD, fd, event); err != nil {
 		return os.NewSyscallError("epoll_ctl add", err)
 	}
 	return nil
 }
 
-func (ep *epoll) interest(fd int, event *epollevent) error {
+func (ep *epoll) interest(fd int, event *event.EpollEvent) error {
 	if err := epollCtl(ep.fd, unix.EPOLL_CTL_MOD, fd, event); err != nil {
 		return os.NewSyscallError("epoll_ctl mod", err)
 	}
@@ -261,12 +265,7 @@ func (ep *epoll) remove(fd int) error {
 	return nil
 }
 
-type epollevent struct {
-	events uint32
-	data   [8]byte
-}
-
-func epollCtl(epfd int, op int, fd int, event *epollevent) error {
+func epollCtl(epfd int, op int, fd int, event *event.EpollEvent) error {
 	var err error
 	_, _, err = unix.RawSyscall6(
 		unix.SYS_EPOLL_CTL,

internal/poller/poller_kqueue.go

@@ -125,6 +125,9 @@ func (k *kqueue) handle(n int) {
 				}
 			}
 		}
+		if GoschedAfterEvent {
+			runtime.Gosched()
+		}
 	}
 
 	if wakeUp {

internal/timer/timer_test.go

@@ -21,7 +21,6 @@ func TestTimerNormal(t *testing.T) {
 	time.Sleep(time.Millisecond * 10)
 	assert.Equal(t, true, t1.Expired())
 
-	t1.Start()
 	t1.Reset(time.Time{})
 	assert.Equal(t, true, t1.IsZero())
 	assert.Equal(t, false, t1.Expired())

options.go

@@ -25,6 +25,13 @@ func NumPollers() int {
 	return poller.NumPollers()
 }
 
+// EnablePollerGoschedAfterEvent enables calling runtime.Gosched() after processing of each event
+// during epoll wait handling.
+// This function can only be called inside func init().
+func EnablePollerGoschedAfterEvent() {
+	poller.GoschedAfterEvent = true
+}
+
 // OnTCPOpened fires when the tcp connection is established.
 type OnTCPOpened func(conn Conn) error
 

tcpconn_test.go

@@ -309,7 +309,7 @@ func TestConnWrite_ServHandleErr(t *testing.T) {
 			assert.Nil(t, err)
 		},
 		ctrlHandle: func(t *testing.T, server tnet.Conn, client net.Conn, ch chan int) {
-			time.Sleep(time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			assert.Equal(t, false, server.IsActive())
 		},
 		isTnetCliConn: true,