diff --git a/_typos.toml b/_typos.toml index dfd475b..567e835 100644 --- a/_typos.toml +++ b/_typos.toml @@ -2,3 +2,6 @@ [files] extend-exclude = ["go.mod", "go.sum"] + +[default.extend-words] +typ = "typ" # type \ No newline at end of file diff --git a/connstate/conn.go b/connstate/conn.go new file mode 100644 index 0000000..b7f52ff --- /dev/null +++ b/connstate/conn.go @@ -0,0 +1,103 @@ +// Copyright 2025 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 connstate + +import ( + "errors" + "net" + "sync/atomic" + "syscall" + "unsafe" +) + +type ConnState uint32 + +const ( + // StateOK means the connection is normal. + StateOK ConnState = iota + // StateRemoteClosed means the remote side has closed the connection. + StateRemoteClosed + // StateClosed means the connection has been closed by local side. + StateClosed +) + +// ConnStater is the interface to get the ConnState of a connection. +// Must call Close to release it if you're going to close the connection. +type ConnStater interface { + Close() error + State() ConnState +} + +// ListenConnState returns a ConnStater for the given connection. +// It's generally used for availability checks when obtaining connections from a connection pool. +// Conn must be a syscall.Conn. +func ListenConnState(conn net.Conn) (ConnStater, error) { + pollInitOnce.Do(createPoller) + sysConn, ok := conn.(syscall.Conn) + if !ok { + return nil, errors.New("conn is not syscall.Conn") + } + rawConn, err := sysConn.SyscallConn() + if err != nil { + return nil, err + } + var fd *fdOperator + var opAddErr error + err = rawConn.Control(func(fileDescriptor uintptr) { + fd = pollcache.alloc() + fd.fd = int(fileDescriptor) + atomic.StorePointer(&fd.conn, unsafe.Pointer(&connStater{fd: unsafe.Pointer(fd)})) + opAddErr = poll.control(fd, opAdd) + }) + if fd != nil { + if err != nil && opAddErr == nil { + // if rawConn is closed, poller will delete the fd by itself + _ = rawConn.Control(func(_ uintptr) { + _ = poll.control(fd, opDel) + }) + } + if err != nil || opAddErr != nil { + atomic.StorePointer(&fd.conn, nil) + pollcache.freeable(fd) + } + } + if err != nil { + return nil, err + } + if opAddErr != nil { + return nil, opAddErr + } + return (*connStater)(atomic.LoadPointer(&fd.conn)), nil +} + +type connStater struct { + fd unsafe.Pointer // *fdOperator + state uint32 +} + +func (c *connStater) Close() error { + fd := (*fdOperator)(atomic.LoadPointer(&c.fd)) + if fd != nil && atomic.CompareAndSwapPointer(&c.fd, unsafe.Pointer(fd), nil) { + atomic.StoreUint32(&c.state, uint32(StateClosed)) + _ = poll.control(fd, opDel) + atomic.StorePointer(&fd.conn, nil) + pollcache.freeable(fd) + } + return nil +} + +func (c *connStater) State() ConnState { + return ConnState(atomic.LoadUint32(&c.state)) +} diff --git a/connstate/conn_test.go b/connstate/conn_test.go new file mode 100644 index 0000000..15aae54 --- /dev/null +++ b/connstate/conn_test.go @@ -0,0 +1,349 @@ +// Copyright 2025 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 connstate + +import ( + "errors" + "io" + "net" + "runtime" + "sync" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestListenConnState(t *testing.T) { + ln, err := net.Listen("tcp", "localhost:0") + if err != nil { + panic(err) + } + go func() { + for { + conn, err := ln.Accept() + assert.Nil(t, err) + go func(conn net.Conn) { + buf := make([]byte, 11) + _, err := conn.Read(buf) + assert.Nil(t, err) + conn.Close() + }(conn) + } + }() + conn, err := net.Dial("tcp", ln.Addr().String()) + assert.Nil(t, err) + stater, err := ListenConnState(conn) + assert.Nil(t, err) + assert.Equal(t, StateOK, stater.State()) + _, err = conn.Write([]byte("hello world")) + assert.Nil(t, err) + buf := make([]byte, 1) + _, err = conn.Read(buf) + assert.Equal(t, io.EOF, err) + time.Sleep(10 * time.Millisecond) + assert.Equal(t, StateRemoteClosed, stater.State()) + assert.Nil(t, stater.Close()) + assert.Nil(t, conn.Close()) + assert.Equal(t, StateClosed, stater.State()) +} + +type mockPoller struct { + controlFunc func(fd *fdOperator, op op) error +} + +func (m *mockPoller) wait() error { + return nil +} + +func (m *mockPoller) control(fd *fdOperator, op op) error { + return m.controlFunc(fd, op) +} + +type mockConn struct { + net.Conn + controlFunc func(f func(fd uintptr)) error +} + +func (c *mockConn) SyscallConn() (syscall.RawConn, error) { + return &mockRawConn{ + controlFunc: c.controlFunc, + }, nil +} + +type mockRawConn struct { + syscall.RawConn + controlFunc func(f func(fd uintptr)) error +} + +func (r *mockRawConn) Control(f func(fd uintptr)) error { + return r.controlFunc(f) +} + +func TestListenConnState_Err(t *testing.T) { + // replace poll + pollInitOnce.Do(createPoller) + oldPoll := poll + defer func() { + poll = oldPoll + }() + // test detach + var expectDetach bool + defer func() { + assert.True(t, expectDetach) + }() + cases := []struct { + name string + connControlFunc func(f func(fd uintptr)) error + pollControlFunc func(fd *fdOperator, op op) error + expectErr error + }{ + { + name: "err conn control", + connControlFunc: func(f func(fd uintptr)) error { + return errors.New("err conn control") + }, + expectErr: errors.New("err conn control"), + }, + { + name: "err poll control", + connControlFunc: func(f func(fd uintptr)) error { + f(1) + return nil + }, + pollControlFunc: func(fd *fdOperator, op op) error { + assert.Equal(t, fd.fd, 1) + return errors.New("err poll control") + }, + expectErr: errors.New("err poll control"), + }, + { + name: "err conn control after poll add", + connControlFunc: func(f func(fd uintptr)) error { + f(1) + return errors.New("err conn control after poll add") + }, + pollControlFunc: func(fd *fdOperator, op op) error { + if op == opDel { + expectDetach = true + } + assert.Equal(t, fd.fd, 1) + return nil + }, + expectErr: errors.New("err conn control after poll add"), + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + poll = &mockPoller{ + controlFunc: c.pollControlFunc, + } + conn := &mockConn{ + controlFunc: c.connControlFunc, + } + _, err := ListenConnState(conn) + assert.Equal(t, c.expectErr, err) + }) + } +} + +func BenchmarkListenConnState(b *testing.B) { + ln, err := net.Listen("tcp", "localhost:0") + if err != nil { + panic(err) + } + go func() { + for { + conn, err := ln.Accept() + assert.Nil(b, err) + go func(conn net.Conn) { + buf := make([]byte, 11) + _, err := conn.Read(buf) + assert.Nil(b, err) + conn.Close() + }(conn) + } + }() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + conn, err := net.Dial("tcp", ln.Addr().String()) + assert.Nil(b, err) + stater, err := ListenConnState(conn) + assert.Nil(b, err) + assert.Equal(b, StateOK, stater.State()) + _, err = conn.Write([]byte("hello world")) + assert.Nil(b, err) + buf := make([]byte, 1) + _, err = conn.Read(buf) + assert.Equal(b, io.EOF, err) + time.Sleep(10 * time.Millisecond) + assert.Equal(b, StateRemoteClosed, stater.State()) + assert.Nil(b, stater.Close()) + assert.Nil(b, conn.Close()) + assert.Equal(b, StateClosed, stater.State()) + } + }) +} + +type statefulConn struct { + net.Conn + stater ConnStater +} + +func (s *statefulConn) Close() error { + s.stater.Close() + return s.Conn.Close() +} + +type mockStater struct { +} + +func (m *mockStater) State() ConnState { + return StateOK +} + +func (m *mockStater) Close() error { + return nil +} + +type connpool struct { + mu sync.Mutex + conns []*statefulConn +} + +func (p *connpool) get(dialFunc func() *statefulConn) *statefulConn { + p.mu.Lock() + if len(p.conns) == 0 { + p.mu.Unlock() + return dialFunc() + } + for i := len(p.conns) - 1; i >= 0; i-- { + conn := p.conns[i] + if conn.stater.State() == StateOK { + p.conns = p.conns[:i] + p.mu.Unlock() + return conn + } else { + conn.Close() + } + } + p.conns = p.conns[:0] + p.mu.Unlock() + return dialFunc() +} + +func (p *connpool) put(conn *statefulConn) { + p.mu.Lock() + defer p.mu.Unlock() + p.conns = append(p.conns, conn) +} + +func (p *connpool) Close() error { + p.mu.Lock() + defer p.mu.Unlock() + for _, conn := range p.conns { + conn.Close() + } + p.conns = p.conns[:0] + return nil +} + +var withListenConnState bool + +// BenchmarkWithConnState is used to verify the impact of adding ConnState logic on performance. +// To compare with syscall.EpollWait(), you could run `go test -bench=BenchmarkWith -benchtime=10s .` +// to test the first time, and replace isyscall.EpollWait() with syscall.EpollWait() to test the second time. +func BenchmarkWithConnState(b *testing.B) { + withListenConnState = true + benchmarkConnState(b) +} + +func BenchmarkWithoutConnState(b *testing.B) { + withListenConnState = false + benchmarkConnState(b) +} + +func benchmarkConnState(b *testing.B) { + // set GOMAXPROCS to 1 to make P resources scarce + runtime.GOMAXPROCS(1) + ln, err := net.Listen("tcp", "localhost:0") + if err != nil { + panic(err) + } + go func() { + for { + conn, err := ln.Accept() + assert.Nil(b, err) + go func(conn net.Conn) { + var count uint64 + for { + buf := make([]byte, 11) + _, err := conn.Read(buf) + if err != nil { + conn.Close() + return + } + _, err = conn.Write(buf) + if err != nil { + conn.Close() + return + } + count++ + if count == 1000 { + conn.Close() + return + } + } + }(conn) + } + }() + cp := &connpool{} + dialFunc := func() *statefulConn { + conn, err := net.Dial("tcp", ln.Addr().String()) + assert.Nil(b, err) + var stater ConnStater + if withListenConnState { + stater, err = ListenConnState(conn) + assert.Nil(b, err) + } else { + stater = &mockStater{} + } + return &statefulConn{ + Conn: conn, + stater: stater, + } + } + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + conn := cp.get(dialFunc) + buf := make([]byte, 11) + _, err := conn.Write(buf) + if err != nil { + conn.Close() + continue + } + _, err = conn.Read(buf) + if err != nil { + conn.Close() + continue + } + cp.put(conn) + } + }) + _ = cp.Close() +} diff --git a/connstate/poll.go b/connstate/poll.go new file mode 100644 index 0000000..f6c4264 --- /dev/null +++ b/connstate/poll.go @@ -0,0 +1,49 @@ +// Copyright 2025 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 connstate + +import ( + "fmt" + "sync" +) + +type op int + +const ( + opAdd op = iota + opDel +) + +var ( + pollInitOnce sync.Once + poll poller +) + +type poller interface { + wait() error + control(fd *fdOperator, op op) error +} + +func createPoller() { + var err error + poll, err = openpoll() + if err != nil { + panic(fmt.Sprintf("gopkg.connstate openpoll failed, err: %v", err)) + } + go func() { + err := poll.wait() + fmt.Printf("gopkg.connstate epoll wait exit, err: %v\n", err) + }() +} diff --git a/connstate/poll_bsd.go b/connstate/poll_bsd.go new file mode 100644 index 0000000..818e3e8 --- /dev/null +++ b/connstate/poll_bsd.go @@ -0,0 +1,91 @@ +// Copyright 2025 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 darwin || netbsd || freebsd || openbsd || dragonfly +// +build darwin netbsd freebsd openbsd dragonfly + +package connstate + +import ( + "sync/atomic" + "syscall" + "unsafe" +) + +type kqueue struct { + fd int +} + +func (p *kqueue) wait() error { + events := make([]syscall.Kevent_t, 1024) + for { + n, err := syscall.Kevent(p.fd, nil, events, nil) + if err != nil && err != syscall.EINTR { + // exit gracefully + if err == syscall.EBADF { + return nil + } + return err + } + for i := 0; i < n; i++ { + ev := &events[i] + op := *(**fdOperator)(unsafe.Pointer(&ev.Udata)) + if conn := (*connStater)(atomic.LoadPointer(&op.conn)); conn != nil { + if ev.Flags&(syscall.EV_EOF) != 0 { + atomic.CompareAndSwapUint32(&conn.state, uint32(StateOK), uint32(StateRemoteClosed)) + } + } + } + // we can make sure that there is no op remaining if finished handling all events + pollcache.free() + } +} + +func (p *kqueue) control(fd *fdOperator, op op) error { + evs := make([]syscall.Kevent_t, 1) + evs[0].Ident = uint64(fd.fd) + *(**fdOperator)(unsafe.Pointer(&evs[0].Udata)) = fd + if op == opAdd { + evs[0].Filter = syscall.EVFILT_READ + evs[0].Flags = syscall.EV_ADD | syscall.EV_ENABLE | syscall.EV_CLEAR + // prevent ordinary data from triggering + evs[0].Flags |= syscall.EV_OOBAND + evs[0].Fflags = syscall.NOTE_LOWAT + evs[0].Data = 0x7FFFFFFF + _, err := syscall.Kevent(p.fd, evs, nil, nil) + return err + } else { + evs[0].Filter = syscall.EVFILT_READ + evs[0].Flags = syscall.EV_DELETE + _, err := syscall.Kevent(p.fd, evs, nil, nil) + return err + } +} + +func openpoll() (p poller, err error) { + fd, err := syscall.Kqueue() + if err != nil { + return nil, err + } + _, err = syscall.Kevent(fd, []syscall.Kevent_t{{ + Ident: 0, + Filter: syscall.EVFILT_USER, + Flags: syscall.EV_ADD | syscall.EV_CLEAR, + }}, nil, nil) + if err != nil { + syscall.Close(fd) + return nil, err + } + return &kqueue{fd: fd}, nil +} diff --git a/connstate/poll_cache.go b/connstate/poll_cache.go new file mode 100644 index 0000000..707413a --- /dev/null +++ b/connstate/poll_cache.go @@ -0,0 +1,83 @@ +// Copyright 2025 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 connstate + +import ( + "sync" + "unsafe" +) + +const pollBlockSize = 4 * 1024 + +type fdOperator struct { + link *fdOperator // in pollcache, protected by pollcache.lock + index int32 + + fd int + conn unsafe.Pointer // *connStater +} + +var pollcache pollCache + +type pollCache struct { + lock sync.Mutex + first *fdOperator + cache []*fdOperator + // freelist store the freeable operator + // to reduce GC pressure, we only store op index here + freelist []int32 +} + +func (c *pollCache) alloc() *fdOperator { + c.lock.Lock() + if c.first == nil { + const pdSize = unsafe.Sizeof(fdOperator{}) + n := pollBlockSize / pdSize + if n == 0 { + n = 1 + } + index := int32(len(c.cache)) + for i := uintptr(0); i < n; i++ { + pd := &fdOperator{index: index} + c.cache = append(c.cache, pd) + pd.link = c.first + c.first = pd + index++ + } + } + op := c.first + c.first = op.link + c.lock.Unlock() + return op +} + +// freeable mark the operator that could be freed +// only poller could do the real free action +func (c *pollCache) freeable(op *fdOperator) { + c.lock.Lock() + c.freelist = append(c.freelist, op.index) + c.lock.Unlock() +} + +func (c *pollCache) free() { + c.lock.Lock() + for _, idx := range c.freelist { + op := c.cache[idx] + op.link = c.first + c.first = op + } + c.freelist = c.freelist[:0] + c.lock.Unlock() +} diff --git a/connstate/poll_cache_test.go b/connstate/poll_cache_test.go new file mode 100644 index 0000000..86b5707 --- /dev/null +++ b/connstate/poll_cache_test.go @@ -0,0 +1,34 @@ +// Copyright 2025 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 connstate + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestPollCache(t *testing.T) { + pollcache.free() + fd := pollcache.alloc() + pollcache.freeable(fd) + assert.Equal(t, 1, len(pollcache.freelist)) + fd1 := pollcache.alloc() + assert.NotEqual(t, fd, fd1) + pollcache.free() + assert.Equal(t, 0, len(pollcache.freelist)) + fd2 := pollcache.alloc() + assert.Equal(t, fd, fd2) +} diff --git a/connstate/poll_linux.go b/connstate/poll_linux.go new file mode 100644 index 0000000..50ede74 --- /dev/null +++ b/connstate/poll_linux.go @@ -0,0 +1,74 @@ +// Copyright 2025 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 connstate + +import ( + "sync/atomic" + "syscall" + "unsafe" + + isyscall "github.com/cloudwego/gopkg/internal/syscall" +) + +const _EPOLLET uint32 = 0x80000000 + +type epoller struct { + epfd int +} + +//go:nocheckptr +func (p *epoller) wait() error { + events := make([]syscall.EpollEvent, 1024) + for { + // epoll wait is a blocking syscall, so we need to call entersyscallblock to handoff P, + // and let the P run other goroutines. + n, err := isyscall.EpollWait(p.epfd, events, -1) + if err != nil && err != syscall.EINTR { + return err + } + for i := 0; i < n; i++ { + ev := &events[i] + op := *(**fdOperator)(unsafe.Pointer(&ev.Fd)) + if conn := (*connStater)(atomic.LoadPointer(&op.conn)); conn != nil { + if ev.Events&(syscall.EPOLLHUP|syscall.EPOLLRDHUP|syscall.EPOLLERR) != 0 { + atomic.CompareAndSwapUint32(&conn.state, uint32(StateOK), uint32(StateRemoteClosed)) + } + } + } + // we can make sure that there is no op remaining if finished handling all events + pollcache.free() + } +} + +func (p *epoller) control(fd *fdOperator, op op) error { + if op == opAdd { + var ev syscall.EpollEvent + *(**fdOperator)(unsafe.Pointer(&ev.Fd)) = fd + ev.Events = syscall.EPOLLHUP | syscall.EPOLLRDHUP | syscall.EPOLLERR | _EPOLLET + return syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_ADD, fd.fd, &ev) + } else { + var ev syscall.EpollEvent + return syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_DEL, fd.fd, &ev) + } +} + +func openpoll() (p poller, err error) { + var epfd int + epfd, err = syscall.EpollCreate(1) + if err != nil { + return nil, err + } + return &epoller{epfd: epfd}, nil +} diff --git a/connstate/poll_windows.go b/connstate/poll_windows.go new file mode 100644 index 0000000..ad9404d --- /dev/null +++ b/connstate/poll_windows.go @@ -0,0 +1,35 @@ +// Copyright 2025 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 connstate + +import "errors" + +var ( + errNotSupportedForWindows = errors.New("connstate not supported for windows") +) + +type mockWindowsPoller struct{} + +func (m *mockWindowsPoller) wait() error { + return nil +} + +func (m *mockWindowsPoller) control(fd *fdOperator, op op) error { + return errNotSupportedForWindows +} + +func openpoll() (p poller, err error) { + return &mockWindowsPoller{}, nil +} diff --git a/internal/syscall/sys_epoll_linux.go b/internal/syscall/sys_epoll_linux.go new file mode 100644 index 0000000..190e42f --- /dev/null +++ b/internal/syscall/sys_epoll_linux.go @@ -0,0 +1,43 @@ +// Copyright 2025 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 !arm64 && !loong64 +// +build !arm64,!loong64 + +package netpoll + +import ( + "syscall" + "unsafe" +) + +var _zero uintptr + +// EpollWait implements epoll_wait. +func EpollWait(epfd int, events []syscall.EpollEvent, msec int) (n int, err error) { + var _p0 unsafe.Pointer + if len(events) > 0 { + _p0 = unsafe.Pointer(&events[0]) + } else { + _p0 = unsafe.Pointer(&_zero) + } + entersyscallblock() + r0, _, e1 := syscall.RawSyscall6(syscall.SYS_EPOLL_WAIT, uintptr(epfd), uintptr(_p0), uintptr(len(events)), uintptr(msec), 0, 0) + exitsyscall() + n = int(r0) + if e1 != 0 { + err = e1 + } + return +} diff --git a/internal/syscall/sys_epoll_linux_arm64_.go b/internal/syscall/sys_epoll_linux_arm64_.go new file mode 100644 index 0000000..fe97e9a --- /dev/null +++ b/internal/syscall/sys_epoll_linux_arm64_.go @@ -0,0 +1,44 @@ +// Copyright 2025 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 && (arm64 || loong64) +// +build linux +// +build arm64 loong64 + +package netpoll + +import ( + "syscall" + "unsafe" +) + +var _zero uintptr + +// EpollWait implements epoll_wait. +func EpollWait(epfd int, events []syscall.EpollEvent, msec int) (n int, err error) { + var _p0 unsafe.Pointer + if len(events) > 0 { + _p0 = unsafe.Pointer(&events[0]) + } else { + _p0 = unsafe.Pointer(&_zero) + } + entersyscallblock() + r0, _, e1 := syscall.RawSyscall6(syscall.SYS_EPOLL_PWAIT, uintptr(epfd), uintptr(_p0), uintptr(len(events)), uintptr(msec), 0, 0) + exitsyscall() + n = int(r0) + if e1 != 0 { + err = e1 + } + return +} diff --git a/internal/syscall/unsafe.go b/internal/syscall/unsafe.go new file mode 100644 index 0000000..46ba7cc --- /dev/null +++ b/internal/syscall/unsafe.go @@ -0,0 +1,23 @@ +// Copyright 2025 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 entersyscallblock runtime.entersyscallblock +func entersyscallblock() + +//go:linkname exitsyscall runtime.exitsyscall +func exitsyscall()