Describe the bug 长连接时可能是惊群问题,当SetNumLoops大于1时，cpu占满，设置为1时一切正常。

To Reproduce Steps to reproduce the behavior:

1. 编写一个最简单的长连接服务端，用map缓存所有conn客户端连接对象池。并设置SetNumLoops(4).
2. 编写一个最简单的tcp客户端，连接建立2000个连接。不断开。保持连接，60s一个心跳包。
3. 现在观察服务端cpu，会发现cpu占满。
4. 然后把服务器端的SetNumLoops(1)，再重新测试观察cpu占用已降低到正常。

Screenshots 服务端参数： 客户端建立2千个连接，并保活。 此时的服务器端cpu情况：

Desktop (please complete the following information):

• OS: mac book pro
• Version

server

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/cloudwego/netpoll"
)

func main() {
	network, address := "tcp", "127.0.0.1:8888"

	// 创建 listener
	listener, err := netpoll.CreateListener(network, address)
	if err != nil {
		panic("create netpoll listener fail")
	}

	// handle: 连接读数据和处理逻辑
	var onRequest netpoll.OnRequest = handler

	// options: EventLoop 初始化自定义配置项
	var opts = []netpoll.Option{
		netpoll.WithReadTimeout(5 * time.Second),
		netpoll.WithIdleTimeout(10 * time.Minute),
		netpoll.WithOnPrepare(nil),
	}

	// 创建 EventLoop
	eventLoop, err := netpoll.NewEventLoop(onRequest, opts...)
	if err != nil {
		panic("create netpoll event-loop fail")
	}

	// 运行 Server
	err = eventLoop.Serve(listener)
	if err != nil {
		panic("netpoll server exit")
	}
}

// 读事件处理
func handler(ctx context.Context, connection netpoll.Connection) error {
	total := 2
	t := time.Now()
	reader := connection.Reader()
	data, err := reader.Peek(reader.Len())
	log.Printf("before Next, len: %v, data: %v", reader.Len(), string(data))
	data, err = reader.Next(total)
	if err != nil {
		log.Printf("Next failed, total: %v, reader.Len: %v, block time: %v, error: %v", total, reader.Len(), int(time.Since(t)), err)
		return err
	}

	log.Printf("after  Next, len: %v, data: %v, timeused: %v", len(data), string(data), int(time.Since(t).Seconds()))

	n, err := connection.Write(data)
	if err != nil {
		return err
	}
	if n != len(data) {
		return fmt.Errorf("write failed: %v < %v", n, len(data))
	}

	return nil
}

client

package main

import (
	"log"
	"net"
	"time"
)

func main() {
	conn, err := net.Dial("tcp", "127.0.0.1:8888")
	if err != nil {
		log.Fatal("dial failed:", err)
	}

	// 用于测试，设定单个完整协议包 2 字节，server端使用 connection.Reader.Next(2) 进行读取
	// server 端超时时间设定为 5s

	// 第一组，在超时时间内发送，server端能读到完整包，但 connection.Reader.Next(2) 阻塞 2s
	conn.Write([]byte("a"))
	time.Sleep(time.Second * 2)
	conn.Write([]byte("a"))

	time.Sleep(time.Second * 1)

	// 第二组，超过超时时间、分开多次发送完整包，server端 connection.Reader.Next(2) 阻塞 5s（超时时间为5s）后报错，但是连接没有断开
	// 30s 内、client 发送完剩余数据前，server 端多次触发 OnRequest，并且每次 OnRequest 中的 connection.Reader.Next(2) 阻塞 5s（超时时间为5s）后报错，但是连接没有断开
	// 30s 后 client 发送完整包剩余数据，server 端 connection.Reader.Next(2) 读到完整包
	conn.Write([]byte("b"))
	time.Sleep(time.Second * 30)
	conn.Write([]byte("b"))

	time.Sleep(time.Second * 1)

	// 第三组，只发送半包，client 端不再有行动
	// server 端多次触发 OnRequest，并且每次 OnRequest 中的 connection.Reader.Next(2) 阻塞 5s（超时时间为5s）后报错，但是连接没有断开
	// 实际场景中，server 端可能收不到 tcp FIN1，比如 client 设备断电，server 端无法及时迅速地释放该连接，如果大量连接进行攻击，存在服务不可用风险
	conn.Write([]byte("c"))

	<-make(chan int)
}

日志

go run ./netpoll.go 
2021/07/18 07:25:22 before Next, len: 1, data: a
2021/07/18 07:25:24 after  Next, len: 2, data: aa, timeused: 2
2021/07/18 07:25:25 before Next, len: 1, data: b
2021/07/18 07:25:30 Next failed, total: 2, reader.Len: 1, block time: 5005315692, error: connection read timeout 5s
2021/07/18 07:25:30 before Next, len: 1, data: b
2021/07/18 07:25:35 Next failed, total: 2, reader.Len: 1, block time: 5017124559, error: connection read timeout 5s
2021/07/18 07:25:35 before Next, len: 1, data: b
2021/07/18 07:25:40 Next failed, total: 2, reader.Len: 1, block time: 5009562038, error: connection read timeout 5s
2021/07/18 07:25:40 before Next, len: 1, data: b
2021/07/18 07:25:45 Next failed, total: 2, reader.Len: 1, block time: 5008370180, error: connection read timeout 5s
2021/07/18 07:25:45 before Next, len: 1, data: b
2021/07/18 07:25:50 Next failed, total: 2, reader.Len: 1, block time: 5011104792, error: connection read timeout 5s
2021/07/18 07:25:50 before Next, len: 1, data: b
2021/07/18 07:25:55 after  Next, len: 2, data: bb, timeused: 4
2021/07/18 07:25:56 before Next, len: 1, data: c
2021/07/18 07:26:01 Next failed, total: 2, reader.Len: 1, block time: 5009599769, error: connection read timeout 5s
2021/07/18 07:26:01 before Next, len: 1, data: c
2021/07/18 07:26:06 Next failed, total: 2, reader.Len: 1, block time: 5017649436, error: connection read timeout 5s
2021/07/18 07:26:06 before Next, len: 1, data: c
2021/07/18 07:26:11 Next failed, total: 2, reader.Len: 1, block time: 5015780369, error: connection read timeout 5s
2021/07/18 07:26:11 before Next, len: 1, data: c
2021/07/18 07:26:16 Next failed, total: 2, reader.Len: 1, block time: 5013565228, error: connection read timeout 5s
2021/07/18 07:26:16 before Next, len: 1, data: c
2021/07/18 07:26:21 Next failed, total: 2, reader.Len: 1, block time: 5004234323, error: connection read timeout 5s
2021/07/18 07:26:21 before Next, len: 1, data: c
2021/07/18 07:26:26 Next failed, total: 2, reader.Len: 1, block time: 5014860948, error: connection read timeout 5s
2021/07/18 07:26:26 before Next, len: 1, data: c
2021/07/18 07:26:31 Next failed, total: 2, reader.Len: 1, block time: 5009890510, error: connection read timeout 5s
2021/07/18 07:26:31 before Next, len: 1, data: c
2021/07/18 07:26:36 Next failed, total: 2, reader.Len: 1, block time: 5009386524, error: connection read timeout 5s
2021/07/18 07:26:36 before Next, len: 1, data: c
2021/07/18 07:26:41 Next failed, total: 2, reader.Len: 1, block time: 5009694923, error: connection read timeout 5s
2021/07/18 07:26:41 before Next, len: 1, data: c
2021/07/18 07:26:46 Next failed, total: 2, reader.Len: 1, block time: 5006999390, error: connection read timeout 5s
2021/07/18 07:26:46 before Next, len: 1, data: c
2021/07/18 07:26:51 Next failed, total: 2, reader.Len: 1, block time: 5016639111, error: connection read timeout 5s
2021/07/18 07:26:51 before Next, len: 1, data: c
2021/07/18 07:26:56 Next failed, total: 2, reader.Len: 1, block time: 5004699154, error: connection read timeout 5s
2021/07/18 07:26:56 before Next, len: 1, data: c
2021/07/18 07:27:01 Next failed, total: 2, reader.Len: 1, block time: 5003720648, error: connection read timeout 5s
2021/07/18 07:27:01 before Next, len: 1, data: c
2021/07/18 07:27:06 Next failed, total: 2, reader.Len: 1, block time: 5013684114, error: connection read timeout 5s
2021/07/18 07:27:06 before Next, len: 1, data: c
2021/07/18 07:27:11 Next failed, total: 2, reader.Len: 1, block time: 5008594864, error: connection read timeout 5s
2021/07/18 07:27:11 before Next, len: 1, data: c
2021/07/18 07:27:16 Next failed, total: 2, reader.Len: 1, block time: 5016949058, error: connection read timeout 5s
2021/07/18 07:27:16 before Next, len: 1, data: c
### 一直这样提示

I try to write a simple test on Mac OSX below, but it can't work. I use telnet 127.0.0.1 8080, type hello world, but can't get the response.

package main

import (
	"context"
	"flag"
	"fmt"
	"time"

	"github.com/cloudwego/netpoll"
)

var (
	addr = flag.String("addr", "127.0.0.1:8080", "server address")
)

func onRequest(_ context.Context, conn netpoll.Connection) error {
	println("hello world")

	var reader, writer = conn.Reader(), conn.Writer()

	defer reader.Release()
	// client send "Hello World", size is 11
	buf, err := reader.Next(11)
	if err != nil {
		fmt.Printf("error %v\n", err)
		return err
	}

	alloc, _ := writer.Malloc(len(buf))
	copy(alloc, buf)
	err = writer.Flush()
	if err != nil {
		fmt.Printf("flush error %v\n", err)
		return err
	}

	return nil
}

func onPrepare(conn netpoll.Connection) context.Context {
	println("hello prepare")
	return context.TODO()
}

func main() {
	flag.Parse()
	l, err := netpoll.CreateListener("tcp", *addr)
	if err != nil {
		panic("create netpoll listener failed")
	}
	defer l.Close()

	println("hello event loop")

	loop, err1 := netpoll.NewEventLoop(
		onRequest,
		netpoll.WithReadTimeout(time.Second),
		netpoll.WithOnPrepare(onPrepare),
	)
	if err1 != nil {
		panic("create event loop failed")
	}

	println("begin to serve")
	loop.Serve(l)
}

Describe the bug 当客户端主动关闭链接时，服务端获取不到完整的数据。

To Reproduce

1. 客户端发送消息并主动关闭链接。
2. 统计服务端收到的消息长度。
3. 消息实际长度为27260。

Expected behavior A clear and concise description of what you expected to happen.

Screenshots 客户端代码 服务端部分代码 统计结果

Desktop (please complete the following information):

• OS: macOS
• Version: 13.0.1

如上图： d当我监听12345端口后，想请求下该端口查看是netpoll处理io事件流程，却发现监听端口始终无法启动，也没有任何报错。 之前曾经短暂启动成功过 下面是具体的code代码：

import ( "context" "fmt" "github.com/cloudwego/netpoll" "time" )

func main() {

l, err := netpoll.CreateListener("tcp", ":12345")
if err != nil {
	panic("监听错误")
}
eventLoop, _ := netpoll.NewEventLoop(
	//这个就是接收到io事件后处理的handler
	func(ctx context.Context, connection netpoll.Connection) error {
		fmt.Printf("收到请求了")
		return nil
	},
	netpoll.WithReadTimeout(time.Second))
fmt.Printf("准备开始监听请求")
eventLoop.Serve(l)

}

下面是我电脑的配置：

每当server端重启时，client端偶尔会出现panic，server端为C++代码，C++的grpc依赖版本为1.43.0

报错详情

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x20 pc=0x902720]

goroutine 89 [running]:
github.com/cloudwego/netpoll.(*linkBufferNode).Malloc(...)
       github.com/cloudwego/netpoll/nocopy_linkbuffer.go:730
github.com/cloudwego/netpoll.(*LinkBuffer).book(0xc86d608e60, 0x40000, 0x7f0b1)
        github.com/cloudwego/netpoll/nocopy_linkbuffer.go:587 +0xa0
github.com/cloudwego/netpoll.(*connection).inputs(0xcdee844e80?, {0xc000440300, 0x20, 0x20})
        github.com/cloudwego/netpoll/connection_reactor.go:77 +0x4a
github.com/cloudwego/netpoll.(*defaultPoll).handler(0xc000532280, {0xc000544000, 0x1, 0xc000544000?})
        github.com/cloudwego/netpoll/poll_default_linux.go:139 +0x10a
github.com/cloudwego/netpoll.(*defaultPoll).Wait(0xc000532280)
        github.com/cloudwego/netpoll/poll_default_linux.go:103 +0x111
created by github.com/cloudwego/netpoll.(*manager).Run
        github.com/cloudwego/netpoll/poll_manager.go:106 +0x31

环境 OS : linux Go version : 1.18.8 netpoll version : github.com/cloudwego/netpoll v0.3.1 kitex version: github.com/cloudwego/kitex v0.4.3

EventLoop 的 OnRequest 内，链接被主动关闭时，会导致链接（processing）死锁，从而无法关闭。

OnRequest

func (s *server) onHandle(_ context.Context, conn netpoll.Connection) error {
	time.Sleep(time.Second * 3)
	conn.Close()
	return nil
}

connection_onevent.go:115

....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection.go:59:18: undefined: OnRequest ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection_impl.go:30:2: undefined: netFD ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection_impl.go:40:19: undefined: barrier ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection_impl.go:41:19: undefined: barrier ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection_impl.go:252:32: undefined: Conn ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection_impl.go:252:46: undefined: OnPrepare ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection_impl.go:272:38: undefined: Conn ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection_onevent.go:48:39: undefined: OnRequest ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\connection_onevent.go:71:40: undefined: OnPrepare ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\net_sock.go:41:116: undefined: netFD ....\go1.17.1\pkg\mod\github.com\cloudwego\[email protected]\net_sock.go:41:116: too many errors

https://github.com/cloudwego/netpoll/blob/5607dcbce465c6c75e53fa346da356df96bfb38b/connection_onevent.go#L111 是否仅当 c.Reader().Len() <= 0 进行一次unlock即可，可以减少一次增删锁的开销，task中重新上锁的原因是什么呢？

目前benchmark代码的链接已经失效，希望能早点提供测试用例完整代码

目前文档中benchmark的图，只包括了 100连接（没提到 payload）以及 payload 1k（没提到连接数）的指标，测试内容太少、能够说明的情况也局限

希望能提供不同并发情况下，比如1k、10k、100k、1000k场景下不同payload时与其他网络库、标准库的指标对比

另外由于本人参考文档写的echo server压测时发现netpoll内存占用非常高，而内存占用也是海量并发场景时的重要指标。所以希望也能提供包括内存占用以及其他指标的数据对比

// readBinary cannot use mcache, because the memory allocated by readBinary will not be recycled.
func (b *LinkBuffer) readBinary(n int) (p []byte) {
	b.recalLen(-n) // re-cal length

	// single node
	p = make([]byte, n)
	if b.isSingleNode(n) {
		copy(p, b.read.Next(n))
		return p
	}
	// multiple nodes
	var pIdx int
	var l int
	for ack := n; ack > 0; ack = ack - l {
		l = b.read.Len()
		if l >= ack {
			pIdx += copy(p[pIdx:], b.read.Next(ack))
			break
		} else if l > 0 {
			pIdx += copy(p[pIdx:], b.read.Next(l))
		}
		b.read = b.read.next
	}
	_ = pIdx
	return p
}

Reader 的 接口貌似返回的都是对数组的拷贝, No Copy 是指扩容缩容 No Copy 吗

root@JERRY test $ go build

github.com/cloudwego/netpoll

../../pkg/mod/github.com/cloudwego/[email protected]/sys_sendmsg_linux.go:41:11: cannot use uint64(iovLen) (value of type uint64) as type uint32 in struct literal ../../pkg/mod/github.com/cloudwego/[email protected]/sys_sendmsg_linux.go:47:40: undefined: syscall.SYS_SENDMSG ../../pkg/mod/github.com/cloudwego/[email protected]/sys_zerocopy_linux.go:33:9: cannot use sec (variable of type int64) as type int32 in struct literal ../../pkg/mod/github.com/cloudwego/[email protected]/sys_zerocopy_linux.go:34:9: cannot use usec (variable of type int64) as type int32 in struct literal

github.com/henrylee2cn/ameda

../../pkg/mod/github.com/henrylee2cn/[email protected]/int.go:13:9: cannot use math.MaxInt64 (untyped int constant 9223372036854775807) as int value in return statement (overflows) ../../pkg/mod/github.com/henrylee2cn/[email protected]/uint.go:13:9: cannot use math.MaxUint64 (untyped int constant 18446744073709551615) as uint value in return statement (overflows)

any suggestions?

What type of PR is this?

chore

What this PR does / why we need it (en: English/zh: Chinese):

en: fix: poller read all data before connection close zh: fix: poller 在连接关闭之前，尝试读取完所有数据

Which issue(s) this PR fixes:

Describe the bug I set value to context in func OnConnect like: context.WithValue(ctx, "key", val). Within ten threads, I can get correct value from OnRequest's context, but under high concurrency like 40 threads, the context return 0 value and can't get value which I set in OnConnect.

To Reproduce test with sysbench, --threads=40

Expected behavior Get context values correct.

Panic logger.go:190: [Error] GOPOOL: panic in pool: gopool.DefaultPool: interface conversion: interface {} is nil, not *server.ClientConn: goroutine 1858 [running]

What type of PR is this?

perf

What this PR does / why we need it (en: English/zh: Chinese):

en: replace mcache with pcache zh: 使用 pcache 替代 mcache

Which issue(s) this PR fixes:

The purpose of this issue

We are always interested in finding out who is using Netpoll, what attracted you to using it, how we can listen to your needs and if you are interested, help promote your organization.

• We have people reaching out to us asking, who uses Netpoll in production?
• We’d like to listen to what you would like to see in Netpoll and your scenarios?
• We'd like to help promote your organization and work with you

What we would like from you

Submit a comment in this issue to include the following information

• Your organization or company
• Link to your website
• Your country
• Your contact info to reach out to you: blog, email or Twitter (at least one).
• What is your scenario for using Netpoll?
• Are you running you application in Testing or Production?

Organization/Company: ByteDance
Website: https://bytedance.com
Country: China
Contact: [email protected]
Usage scenario: Using Netpoll as a default net lib in Kitex & Hertz to build large scale Cloud Native applications
Status: Production