feat: epoll managered by runtime netpoller

Author: joway

go.mod

@@ -1,6 +1,6 @@
 module github.com/cloudwego/netpoll
 
-go 1.15
+go 1.18
 
 require (
 	github.com/bytedance/gopkg v0.0.0-20220413063733-65bf48ffb3a7

poll_default_bsd.go

@@ -25,13 +25,17 @@ import (
 	"unsafe"
 )
 
+func openPollFile() (int, error) {
+	return syscall.Kqueue()
+}
+
 func openPoll() (Poll, error) {
 	return openDefaultPoll()
 }
 
 func openDefaultPoll() (*defaultPoll, error) {
 	l := new(defaultPoll)
-	p, err := syscall.Kqueue()
+	p, err := openPollFile()
 	if err != nil {
 		return nil, err
 	}

poll_default_linux.go

@@ -16,13 +16,17 @@ package netpoll
 
 import (
 	"errors"
-	"runtime"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"syscall"
 	"unsafe"
 )
 
+func openPollFile() (int, error) {
+	return EpollCreate(0)
+}
+
 func openPoll() (Poll, error) {
 	return openDefaultPoll()
 }
@@ -31,11 +35,16 @@ func openDefaultPoll() (*defaultPoll, error) {
 	var poll = new(defaultPoll)
 
 	poll.buf = make([]byte, 8)
-	var p, err = EpollCreate(0)
+	var p, err = openPollFile()
 	if err != nil {
 		return nil, err
 	}
 	poll.fd = p
+	pd, errno := runtime_pollOpen(uintptr(poll.fd))
+	if errno != 0 {
+		return nil, Exception(ErrUnsupported, fmt.Sprintf("when poll open: errno=%d", errno))
+	}
+	poll.pd = pd
 
 	var r0, _, e0 = syscall.Syscall(syscall.SYS_EVENTFD2, 0, 0, 0)
 	if e0 != 0 {
@@ -60,6 +69,7 @@ func openDefaultPoll() (*defaultPoll, error) {
 type defaultPoll struct {
 	pollArgs
 	fd      int            // epoll fd
+	pd      uintptr        // runtime pd for epoll fd
 	wop     *FDOperator    // eventfd, wake epoll_wait
 	buf     []byte         // read wfd trigger msg
 	trigger uint32         // trigger flag
@@ -90,23 +100,28 @@ func (a *pollArgs) reset(size, caps int) {
 // Wait implements Poll.
 func (p *defaultPoll) Wait() (err error) {
 	// init
-	var caps, msec, n = barriercap, -1, 0
+	var caps, n = barriercap, 0
 	p.Reset(128, caps)
 	// wait
 	for {
 		if n == p.size && p.size < 128*1024 {
 			p.Reset(p.size<<1, caps)
 		}
-		n, err = EpollWait(p.fd, p.events, msec)
+		n, err = EpollWait(p.fd, p.events, 0)
 		if err != nil && err != syscall.EINTR {
 			return err
 		}
-		if n <= 0 {
-			msec = -1
-			runtime.Gosched()
+		if n == 0 {
+			errno := runtime_pollReset(p.pd, 'r')
+			if errno != 0 {
+				return Exception(ErrUnsupported, fmt.Sprintf("when poll reset: errno=%d", errno))
+			}
+			errno = runtime_pollWait(p.pd, 'r')
+			if errno != 0 {
+				return Exception(ErrUnsupported, fmt.Sprintf("when poll wait: errno=%d", errno))
+			}
 			continue
 		}
-		msec = 0
 		if p.Handler(p.events[:n]) {
 			return nil
 		}

runtime.go

@@ -0,0 +1,28 @@
+// Copyright 2024 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package netpoll
+
+import (
+	_ "unsafe"
+)
+
+//go:linkname runtime_pollOpen internal/poll.runtime_pollOpen
+func runtime_pollOpen(fd uintptr) (pd uintptr, errno int)
+
+//go:linkname runtime_pollWait internal/poll.runtime_pollWait
+func runtime_pollWait(pd uintptr, mode int) (errno int)
+
+//go:linkname runtime_pollReset internal/poll.runtime_pollReset
+func runtime_pollReset(pd uintptr, mode int) (errno int)

runtime_linux_test.go

@@ -0,0 +1,73 @@
+// Copyright 2024 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build linux || loong64
+// +build linux loong64
+
+package netpoll
+
+import (
+	"syscall"
+	"testing"
+	"time"
+)
+
+func TestRuntimeNetpoller(t *testing.T) {
+	pfd, err := openPollFile()
+	MustNil(t, err)
+
+	pd, errno := runtime_pollOpen(uintptr(pfd))
+	Assert(t, errno == 0, errno)
+	t.Logf("poll open success: pd=%d", pd)
+
+	var rfd, wfd = GetSysFdPairs()
+
+	eventin := &epollevent{
+		events: syscall.EPOLLIN | syscall.EPOLLRDHUP | syscall.EPOLLERR,
+		data:   [8]byte{0, 0, 0, 0, 0, 0, 0, 1},
+	}
+	err = EpollCtl(pfd, syscall.EPOLL_CTL_ADD, rfd, eventin)
+	MustNil(t, err)
+
+	go func() {
+		time.Sleep(time.Millisecond * 100)
+
+		iovec := [1]syscall.Iovec{}
+		buf := []byte("hello")
+		n, err := writev(wfd, [][]byte{buf}, iovec[:])
+		MustNil(t, err)
+		Equal(t, n, 5)
+		t.Logf("poll read success: %s", string(buf[:n]))
+	}()
+
+	begin := time.Now()
+	errno = runtime_pollWait(pd, 'r'+'w')
+	Assert(t, errno == 0, errno)
+	cost := time.Since(begin)
+	Assert(t, cost.Milliseconds() >= 100)
+
+	events := make([]epollevent, 1)
+	n, err := EpollWait(pfd, events, 0)
+	MustNil(t, err)
+	Equal(t, n, 1)
+	t.Logf("poll wait success")
+
+	iovec := [1]syscall.Iovec{}
+	buf := make([]byte, 1024)
+	bs := [1][]byte{buf}
+	n, err = readv(rfd, bs[:], iovec[:])
+	MustNil(t, err)
+	Equal(t, n, 5)
+	t.Logf("poll read success: %s", string(buf[:n]))
+}