Skip to content

Commit 6039df7

Browse files
jowayjoway
committed
feat: epoll managered by runtime netpoller
1 parent bb9c3f7 commit 6039df7

File tree

5 files changed

+140
-32
lines changed

5 files changed

+140
-32
lines changed

‎poll_default_bsd.go‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,19 +19,28 @@ package netpoll
1919

2020
import (
2121
"errors"
22+
"runtime"
2223
"sync"
2324
"sync/atomic"
2425
"syscall"
2526
"unsafe"
2627
)
2728

29+
func defaultPollNum() int {
30+
return runtime.GOMAXPROCS(0)/20 + 1
31+
}
32+
33+
func openPollFile() (int, error) {
34+
return syscall.Kqueue()
35+
}
36+
2837
func openPoll() (Poll, error) {
2938
return openDefaultPoll()
3039
}
3140

3241
func openDefaultPoll() (*defaultPoll, error) {
3342
l := new(defaultPoll)
34-
p, err := syscall.Kqueue()
43+
p, err := openPollFile()
3544
if err != nil {
3645
return nil, err
3746
}

‎poll_default_linux.go‎

Lines changed: 30 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,24 @@ package netpoll
1616

1717
import (
1818
"errors"
19+
"fmt"
1920
"runtime"
2021
"sync"
2122
"sync/atomic"
2223
"syscall"
2324
"unsafe"
2425
)
2526

27+
func defaultPollNum() int {
28+
// more pollers could help poller schedule user work to different P to increase parallelism,
29+
// but also will decrease poller's epoll efficiency, so it's a trade-off
30+
return runtime.GOMAXPROCS(0) / 2
31+
}
32+
33+
func openPollFile() (int, error) {
34+
return EpollCreate(0)
35+
}
36+
2637
func openPoll() (Poll, error) {
2738
return openDefaultPoll()
2839
}
@@ -31,11 +42,17 @@ func openDefaultPoll() (*defaultPoll, error) {
3142
var poll = new(defaultPoll)
3243

3344
poll.buf = make([]byte, 8)
34-
var p, err = EpollCreate(0)
45+
var p, err = openPollFile()
3546
if err != nil {
3647
return nil, err
3748
}
3849
poll.fd = p
50+
// register epollfd into runtime's netpoller
51+
pd, errno := runtime_pollOpen(uintptr(poll.fd))
52+
if errno != 0 {
53+
return nil, Exception(ErrUnsupported, fmt.Sprintf("when poll open: errno=%d", errno))
54+
}
55+
poll.pd = pd
3956

4057
var r0, _, e0 = syscall.Syscall(syscall.SYS_EVENTFD2, 0, 0, 0)
4158
if e0 != 0 {
@@ -60,6 +77,7 @@ func openDefaultPoll() (*defaultPoll, error) {
6077
type defaultPoll struct {
6178
pollArgs
6279
fd int // epoll fd
80+
pd uintptr // the pollDesc of epoll fd in runtime's netpoller
6381
wop *FDOperator // eventfd, wake epoll_wait
6482
buf []byte // read wfd trigger msg
6583
trigger uint32 // trigger flag
@@ -90,23 +108,28 @@ func (a *pollArgs) reset(size, caps int) {
90108
// Wait implements Poll.
91109
func (p *defaultPoll) Wait() (err error) {
92110
// init
93-
var caps, msec, n = barriercap, -1, 0
111+
var caps, n = barriercap, 0
94112
p.Reset(128, caps)
95113
// wait
96114
for {
97115
if n == p.size && p.size < 128*1024 {
98116
p.Reset(p.size<<1, caps)
99117
}
100-
n, err = EpollWait(p.fd, p.events, msec)
118+
n, err = EpollWait(p.fd, p.events, 0)
101119
if err != nil && err != syscall.EINTR {
102120
return err
103121
}
104-
if n <= 0 {
105-
msec = -1
106-
runtime.Gosched()
122+
if n == 0 {
123+
errno := runtime_pollReset(p.pd, 'r')
124+
if errno != 0 {
125+
return Exception(ErrUnsupported, fmt.Sprintf("when poll reset: errno=%d", errno))
126+
}
127+
errno = runtime_pollWait(p.pd, 'r')
128+
if errno != 0 {
129+
return Exception(ErrUnsupported, fmt.Sprintf("when poll wait: errno=%d", errno))
130+
}
107131
continue
108132
}
109-
msec = 0
110133
if p.Handler(p.events[:n]) {
111134
return nil
112135
}

‎poll_default_linux_test.go‎

Lines changed: 71 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,7 @@ import (
2121
"errors"
2222
"syscall"
2323
"testing"
24-
25-
"golang.org/x/sys/unix"
24+
"time"
2625
)
2726

2827
func TestEpollEvent(t *testing.T) {
@@ -54,11 +53,11 @@ func TestEpollEvent(t *testing.T) {
5453
}
5554

5655
// EPOLL: add ,del and add
57-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, rfd, event1)
56+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, rfd, event1)
5857
MustNil(t, err)
59-
err = EpollCtl(epollfd, unix.EPOLL_CTL_DEL, rfd, event1)
58+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_DEL, rfd, event1)
6059
MustNil(t, err)
61-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, rfd, event2)
60+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, rfd, event2)
6261
MustNil(t, err)
6362
_, err = syscall.Write(wfd, send)
6463
MustNil(t, err)
@@ -68,15 +67,15 @@ func TestEpollEvent(t *testing.T) {
6867
Equal(t, events[0].data, eventdata2)
6968
_, err = syscall.Read(rfd, recv)
7069
MustTrue(t, err == nil && string(recv) == string(send))
71-
err = EpollCtl(epollfd, unix.EPOLL_CTL_DEL, rfd, event2)
70+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_DEL, rfd, event2)
7271
MustNil(t, err)
7372

7473
// EPOLL: add ,mod and mod
75-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, rfd, event1)
74+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, rfd, event1)
7675
MustNil(t, err)
77-
err = EpollCtl(epollfd, unix.EPOLL_CTL_MOD, rfd, event2)
76+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_MOD, rfd, event2)
7877
MustNil(t, err)
79-
err = EpollCtl(epollfd, unix.EPOLL_CTL_MOD, rfd, event3)
78+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_MOD, rfd, event3)
8079
MustNil(t, err)
8180
_, err = syscall.Write(wfd, send)
8281
MustNil(t, err)
@@ -88,7 +87,7 @@ func TestEpollEvent(t *testing.T) {
8887
Assert(t, events[0].events&syscall.EPOLLIN != 0)
8988
Assert(t, events[0].events&syscall.EPOLLOUT != 0)
9089

91-
err = EpollCtl(epollfd, unix.EPOLL_CTL_DEL, rfd, event2)
90+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_DEL, rfd, event2)
9291
MustNil(t, err)
9392
}
9493

@@ -110,7 +109,7 @@ func TestEpollWait(t *testing.T) {
110109
events: syscall.EPOLLIN | syscall.EPOLLOUT | syscall.EPOLLRDHUP | syscall.EPOLLERR,
111110
data: eventdata,
112111
}
113-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, rfd, event)
112+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, rfd, event)
114113
MustNil(t, err)
115114
_, err = epollWaitUntil(epollfd, events, -1)
116115
MustNil(t, err)
@@ -146,7 +145,7 @@ func TestEpollWait(t *testing.T) {
146145
// EPOLL: close current fd
147146
rfd2, wfd2 := GetSysFdPairs()
148147
defer syscall.Close(wfd2)
149-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, rfd2, event)
148+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, rfd2, event)
150149
err = syscall.Close(rfd2)
151150
MustNil(t, err)
152151
_, err = epollWaitUntil(epollfd, events, -1)
@@ -156,7 +155,7 @@ func TestEpollWait(t *testing.T) {
156155
Assert(t, events[0].events&syscall.EPOLLRDHUP != 0)
157156
Assert(t, events[0].events&syscall.EPOLLERR == 0)
158157

159-
err = EpollCtl(epollfd, unix.EPOLL_CTL_DEL, rfd, event)
158+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_DEL, rfd, event)
160159
MustNil(t, err)
161160
}
162161

@@ -173,7 +172,7 @@ func TestEpollETClose(t *testing.T) {
173172
}
174173

175174
// EPOLL: init state
176-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, rfd, event)
175+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, rfd, event)
177176
_, err = epollWaitUntil(epollfd, events, -1)
178177
MustNil(t, err)
179178
Assert(t, events[0].events&syscall.EPOLLIN == 0)
@@ -194,7 +193,7 @@ func TestEpollETClose(t *testing.T) {
194193
// EPOLL: close peer fd
195194
// EPOLLIN and EPOLLOUT
196195
rfd, wfd = GetSysFdPairs()
197-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, rfd, event)
196+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, rfd, event)
198197
err = syscall.Close(wfd)
199198
MustNil(t, err)
200199
n, err = epollWaitUntil(epollfd, events, 100)
@@ -224,10 +223,10 @@ func TestEpollETDel(t *testing.T) {
224223
}
225224

226225
// EPOLL: del partly
227-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, rfd, event)
226+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, rfd, event)
228227
MustNil(t, err)
229228
event.events = syscall.EPOLLIN | syscall.EPOLLOUT | syscall.EPOLLRDHUP | syscall.EPOLLERR
230-
err = EpollCtl(epollfd, unix.EPOLL_CTL_DEL, rfd, event)
229+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_DEL, rfd, event)
231230
MustNil(t, err)
232231
_, err = syscall.Write(wfd, send)
233232
MustNil(t, err)
@@ -268,7 +267,7 @@ func TestEpollConnectSameFD(t *testing.T) {
268267
t.Logf("create fd: %d", fd1)
269268
err = syscall.SetNonblock(fd1, true)
270269
MustNil(t, err)
271-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, fd1, event1)
270+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, fd1, event1)
272271
MustNil(t, err)
273272
err = syscall.Connect(fd1, &addr)
274273
t.Log(err)
@@ -278,7 +277,7 @@ func TestEpollConnectSameFD(t *testing.T) {
278277
//Assert(t, events[0].events&syscall.EPOLLRDHUP == 0)
279278
//Assert(t, events[0].events&syscall.EPOLLERR == 0)
280279
// forget to del fd
281-
//err = EpollCtl(epollfd, unix.EPOLL_CTL_DEL, fd1, event1)
280+
//err = EpollCtl(epollfd, syscall.EPOLL_CTL_DEL, fd1, event1)
282281
//MustNil(t, err)
283282
err = syscall.Close(fd1) // close fd1
284283
MustNil(t, err)
@@ -289,7 +288,7 @@ func TestEpollConnectSameFD(t *testing.T) {
289288
t.Logf("create fd: %d", fd2)
290289
err = syscall.SetNonblock(fd2, true)
291290
MustNil(t, err)
292-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, fd2, event2)
291+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, fd2, event2)
293292
MustNil(t, err)
294293
err = syscall.Connect(fd2, &addr)
295294
t.Log(err)
@@ -298,7 +297,7 @@ func TestEpollConnectSameFD(t *testing.T) {
298297
Assert(t, events[0].events&syscall.EPOLLOUT != 0)
299298
Assert(t, events[0].events&syscall.EPOLLRDHUP == 0)
300299
Assert(t, events[0].events&syscall.EPOLLERR == 0)
301-
err = EpollCtl(epollfd, unix.EPOLL_CTL_DEL, fd2, event2)
300+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_DEL, fd2, event2)
302301
MustNil(t, err)
303302
err = syscall.Close(fd2) // close fd2
304303
MustNil(t, err)
@@ -310,7 +309,7 @@ func TestEpollConnectSameFD(t *testing.T) {
310309
t.Logf("create fd: %d", fd3)
311310
err = syscall.SetNonblock(fd3, true)
312311
MustNil(t, err)
313-
err = EpollCtl(epollfd, unix.EPOLL_CTL_ADD, fd3, event1)
312+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_ADD, fd3, event1)
314313
MustNil(t, err)
315314
err = syscall.Connect(fd3, &addr)
316315
t.Log(err)
@@ -320,7 +319,7 @@ func TestEpollConnectSameFD(t *testing.T) {
320319
Assert(t, events[0].events&syscall.EPOLLRDHUP == 0)
321320
Assert(t, events[0].events&syscall.EPOLLERR == 0)
322321
MustNil(t, err)
323-
err = EpollCtl(epollfd, unix.EPOLL_CTL_MOD, fd3, eventin)
322+
err = EpollCtl(epollfd, syscall.EPOLL_CTL_MOD, fd3, eventin)
324323
MustNil(t, err)
325324
err = syscall.Close(fd3) // close fd3
326325
MustNil(t, err)
@@ -329,6 +328,55 @@ func TestEpollConnectSameFD(t *testing.T) {
329328
Assert(t, n == 0)
330329
}
331330

331+
func TestRuntimeNetpoller(t *testing.T) {
332+
pfd, err := openPollFile()
333+
MustNil(t, err)
334+
335+
pd, errno := runtime_pollOpen(uintptr(pfd))
336+
Assert(t, errno == 0, errno)
337+
t.Logf("poll open success: pd=%d", pd)
338+
339+
var rfd, wfd = GetSysFdPairs()
340+
341+
eventin := &epollevent{
342+
events: syscall.EPOLLIN | syscall.EPOLLRDHUP | syscall.EPOLLERR,
343+
data: [8]byte{0, 0, 0, 0, 0, 0, 0, 1},
344+
}
345+
err = EpollCtl(pfd, syscall.EPOLL_CTL_ADD, rfd, eventin)
346+
MustNil(t, err)
347+
348+
go func() {
349+
time.Sleep(time.Millisecond * 100)
350+
351+
iovec := [1]syscall.Iovec{}
352+
buf := []byte("hello")
353+
n, err := writev(wfd, [][]byte{buf}, iovec[:])
354+
MustNil(t, err)
355+
Equal(t, n, 5)
356+
t.Logf("poll read success: %s", string(buf[:n]))
357+
}()
358+
359+
begin := time.Now()
360+
errno = runtime_pollWait(pd, 'r'+'w')
361+
Assert(t, errno == 0, errno)
362+
cost := time.Since(begin)
363+
Assert(t, cost.Milliseconds() >= 100)
364+
365+
events := make([]epollevent, 1)
366+
n, err := EpollWait(pfd, events, 0)
367+
MustNil(t, err)
368+
Equal(t, n, 1)
369+
t.Logf("poll wait success")
370+
371+
iovec := [1]syscall.Iovec{}
372+
buf := make([]byte, 1024)
373+
bs := [1][]byte{buf}
374+
n, err = readv(rfd, bs[:], iovec[:])
375+
MustNil(t, err)
376+
Equal(t, n, 5)
377+
t.Logf("poll read success: %s", string(buf[:n]))
378+
}
379+
332380
func epollWaitUntil(epfd int, events []epollevent, msec int) (n int, err error) {
333381
WAIT:
334382
n, err = EpollWait(epfd, events, msec)

‎poll_manager.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ var pollmanager *manager
4848
var logger *log.Logger
4949

5050
func init() {
51-
pollmanager = newManager(runtime.GOMAXPROCS(0)/20 + 1)
51+
pollmanager = newManager(defaultPollNum())
5252
setLoggerOutput(os.Stderr)
5353
}
5454

‎runtime.go‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
// Copyright 2024 CloudWeGo Authors
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package netpoll
16+
17+
import (
18+
_ "unsafe"
19+
)
20+
21+
//go:linkname runtime_pollOpen internal/poll.runtime_pollOpen
22+
func runtime_pollOpen(fd uintptr) (pd uintptr, errno int)
23+
24+
//go:linkname runtime_pollWait internal/poll.runtime_pollWait
25+
func runtime_pollWait(pd uintptr, mode int) (errno int)
26+
27+
//go:linkname runtime_pollReset internal/poll.runtime_pollReset
28+
func runtime_pollReset(pd uintptr, mode int) (errno int)

0 commit comments

Comments
 (0)