WIP: feat: io_uring for netpoll I/O poller

poll_default_bsd.go

@@ -25,7 +25,10 @@ import (
 	"unsafe"
 )
 
-func openPoll() Poll {
+func openPoll(pollType PollType) Poll {
+	if pollType == PollIOURing {
+		return openIOURingPoll()
+	}
 	return openDefaultPoll()
 }
 

poll_default_linux.go

@@ -26,7 +26,10 @@ import (
 )
 
 // Includes defaultPoll/multiPoll/uringPoll...
-func openPoll() Poll {
+func openPoll(pollType PollType) Poll {
+	if pollType == PollIOURing {
+		return openIOURingPoll()
+	}
 	return openDefaultPoll()
 }
 

poll_io_uring.go

@@ -0,0 +1,45 @@
+// Copyright 2021 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 uring "github.com/cloudwego/netpoll/io_uring"
+
+// TODO: init uringPoll
+func openIOURingPoll() *uringPoll {
+	poll := new(uringPoll)
+	ring, err := uring.IOURing(0)
+	if err != nil {
+		panic(err)
+	}
+	poll.fd = ring.Fd()
+	return poll
+}
+
+// TODO: build uringPoll
+type uringPoll struct {
+	fd int
+}
+
+// TODO: Wait implements Poll.
+func (p *uringPoll) Wait() error
+
+// TODO: Close implements Poll.
+func (p *uringPoll) Close() error
+
+// TODO: Trigger implements Poll.
+func (p *uringPoll) Trigger() error
+
+// TODO: Control implements Poll.
+func (p *uringPoll) Control(operator *FDOperator, event PollEvent) error

poll_manager.go

@@ -98,10 +98,16 @@ func (m *manager) Close() error {
 }
 
 // Run all pollers.
-func (m *manager) Run() error {
+func (m *manager) Run(pollTypes ...PollType) error {
+	// set PollDefault as type of poll
+	pollType := PollDefault
+	// set poll type, only executed if the parameter is unique
+	if len(pollTypes) == 1 {
+		pollType = pollTypes[0]
+	}
 	// new poll to fill delta.
 	for idx := len(m.polls); idx < m.NumLoops; idx++ {
-		var poll = openPoll()
+		var poll = openPoll(pollType)
 		m.polls = append(m.polls, poll)
 		go poll.Wait()
 	}
@@ -123,3 +129,15 @@ func (m *manager) Reset() error {
 func (m *manager) Pick() Poll {
 	return m.balance.Pick()
 }
+
+// PollType defines the type of manager.polls.
+type PollType int
+
+const (
+	// PollDefault is used to set poll as epoll on linux systems by default,
+	// and kevent by default on bsd systems.
+	PollDefault PollType = 0x1
+
+	// PollIOURing is used to set poll as io_uring.
+	PollIOURing PollType = 0x2
+)

uring/sys_barrier.go

@@ -0,0 +1,33 @@
+// Copyright 2021 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 uring
+
+import "sync/atomic"
+
+func WRITE_ONCE_U32(p *uint32, v uint32) {
+	atomic.StoreUint32(p, v)
+}
+
+func READ_ONCE_U32(p *uint32) uint32 {
+	return atomic.LoadUint32(p)
+}
+
+func SMP_STORE_RELEASE_U32(p *uint32, v uint32) {
+	atomic.StoreUint32(p, v)
+}
+
+func SMP_LOAD_ACQUIRE_U32(p *uint32) uint32 {
+	return atomic.LoadUint32(p)
+}

uring/sys_enter.go

@@ -0,0 +1,146 @@
+// Copyright 2021 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 uring
+
+import (
+	"syscall"
+)
+
+// Completion Queue Eveny, IO completion data structure
+type URingCQE struct {
+	UserData uint64 // sqe->data submission passed back
+	Res      int32  // result code for this event
+	Flags    uint32
+
+	// If the ring is initialized with IORING_SETUP_CQE32, then this field
+	// contains 16-bytes of padding, doubling the size of the CQE.
+	BigCQE []uint64
+}
+
+// Error implements CQE
+func (c *URingCQE) Error() error {
+	return syscall.Errno(uintptr(-c.Res))
+}
+
+// getData implements CQE
+func (c *URingCQE) getData() uint64 {
+	return c.UserData
+}
+
+// Submission Queue Entry, IO submission data structure
+type URingSQE struct {
+	OpCode      uint8  // type of operation for this sqe
+	Flags       uint8  // IOSQE_ flags
+	IOPrio      uint16 // ioprio for the request
+	Fd          int32  // file descriptor to do IO on
+	Off         uint64 // offset into file
+	Addr        uint64 // pointer to buffer or iovecs
+	Len         uint32 // buffer size or number of iovecs
+	OpcodeFlags uint32
+	UserData    uint64 // data to be passed back at completion time
+
+	BufIG uint16
+
+	Personality uint16 // personality to use, if used
+	SpliceFdIn  int32
+	_pad2       [2]uint64
+}
+
+// setData sets the user data field of the SQE instance passed in.
+func (s *URingSQE) setData(ud uint64) {
+	s.UserData = ud
+}
+
+// setFlags sets the flags field of the SQE instance passed in.
+func (s *URingSQE) setFlags(flags uint8) {
+	s.Flags = flags
+}
+
+// setAddr sets the flags field of the SQE instance passed in.
+func (s *URingSQE) setAddr(addr uintptr) {
+	s.Addr = uint64(addr)
+}
+
+// PrepRW implements SQE
+func (s *URingSQE) PrepRW(op OpFlag, fd int32, addr uintptr, len uint32, offset uint64) {
+	s.OpCode = uint8(op)
+	s.Flags = 0
+	s.IOPrio = 0
+	s.Fd = fd
+	s.Off = offset
+	s.setAddr(addr)
+	s.Len = len
+	s.OpcodeFlags = 0
+	s.UserData = 0
+	s.BufIG = 0
+	s.Personality = 0
+	s.SpliceFdIn = 0
+	s._pad2[0] = 0
+	s._pad2[1] = 0
+}
+
+// Flags of CQE
+// IORING_CQE_F_BUFFER	If set, the upper 16 bits are the buffer ID
+// IORING_CQE_F_MORE	If set, parent SQE will generate more CQE entries
+// IORING_CQE_F_SOCK_NONEMPTY	If set, more data to read after socket recv
+const (
+	IORING_CQE_F_BUFFER OpFlag = 1 << iota
+	IORING_CQE_F_MORE
+	IORING_CQE_F_SOCK_NONEMPTY
+)
+
+const IORING_CQE_BUFFER_SHIFT = 16
+
+// io_uring_enter(2) flags
+const (
+	IORING_ENTER_GETEVENTS uint32 = 1 << iota
+	IORING_ENTER_SQ_WAKEUP
+	IORING_ENTER_SQ_WAIT
+	IORING_ENTER_EXT_ARG
+	IORING_ENTER_REGISTERED_RING
+)
+
+// If sqe->file_index is set to this for opcodes that instantiate a new
+// direct descriptor (like openat/openat2/accept), then io_uring will allocate
+// an available direct descriptor instead of having the application pass one
+// in. The picked direct descriptor will be returned in cqe->res, or -ENFILE
+// if the space is full.
+const (
+	IOSQE_FIXED_FILE_BIT = iota
+	IOSQE_IO_DRAIN_BIT
+	IOSQE_IO_LINK_BIT
+	IOSQE_IO_HARDLINK_BIT
+	IOSQE_ASYNC_BIT
+	IOSQE_BUFFER_SELECT_BIT
+	IOSQE_CQE_SKIP_SUCCESS_BIT
+)
+
+// Flags of SQE
+const (
+	// IOSQE_FIXED_FILE means use fixed fileset
+	IOSQE_FIXED_FILE uint32 = 1 << IOSQE_FIXED_FILE_BIT
+	// IOSQE_IO_DRAIN means issue after inflight IO
+	IOSQE_IO_DRAIN uint32 = 1 << IOSQE_IO_DRAIN_BIT
+	// IOSQE_IO_LINK means links next sqe
+	IOSQE_IO_LINK uint32 = 1 << IOSQE_IO_LINK_BIT
+	// IOSQE_IO_HARDLINK means like LINK, but stronger
+	IOSQE_IO_HARDLINK uint32 = 1 << IOSQE_IO_HARDLINK_BIT
+	// IOSQE_ASYNC means always go async
+	IOSQE_ASYNC uint32 = 1 << IOSQE_ASYNC_BIT
+	// IOSQE_BUFFER_SELECT means select buffer from sqe->buf_group
+	IOSQE_BUFFER_SELECT uint32 = 1 << IOSQE_BUFFER_SELECT_BIT
+	// IOSQE_CQE_SKIP_SUCCESS means don't post CQE if request succeeded
+	IOSQE_CQE_SKIP_SUCCESS uint32 = 1 << IOSQE_CQE_SKIP_SUCCESS_BIT
+)

uring/sys_mmap.go

@@ -0,0 +1,112 @@
+// Copyright 2021 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 uring
+
+import (
+	"syscall"
+	"unsafe"
+)
+
+// sysMmap is used to free the URingSQE and URingCQE,
+func (u *URing) sysMunmap() (err error) {
+	err = syscall.Munmap(u.sqRing.buff)
+	if u.cqRing.buff != nil && &u.cqRing.buff[0] != &u.sqRing.buff[0] {
+		err = syscall.Munmap(u.cqRing.buff)
+	}
+	return
+}
+
+// sysMmap is used to configure the URingSQE and URingCQE,
+// it should only be called after the sysSetUp function has completed successfully.
+func (u *URing) sysMmap(p *ringParams) (err error) {
+	size := unsafe.Sizeof(URingCQE{})
+	if p.flags&IORING_SETUP_CQE32 != 0 {
+		size += unsafe.Sizeof(URingCQE{})
+	}
+	u.sqRing.ringSize = uint64(p.sqOffset.array) + uint64(p.sqEntries*(uint32)(unsafe.Sizeof(uint32(0))))
+	u.cqRing.ringSize = uint64(p.cqOffset.cqes) + uint64(p.cqEntries*(uint32)(size))
+
+	if p.features&IORING_FEAT_SINGLE_MMAP != 0 {
+		if u.cqRing.ringSize > u.sqRing.ringSize {
+			u.sqRing.ringSize = u.cqRing.ringSize
+		}
+		u.cqRing.ringSize = u.sqRing.ringSize
+	}
+
+	// TODO: syscall.MAP_POPULATE unsupport for macox
+	data, err := syscall.Mmap(u.fd, 0, int(u.sqRing.ringSize), syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED)
+	if err != nil {
+		return err
+	}
+	u.sqRing.buff = data
+
+	if p.features&IORING_FEAT_SINGLE_MMAP != 0 {
+		u.cqRing.buff = u.sqRing.buff
+	} else {
+		// TODO: syscall.MAP_POPULATE unsupport for macox
+		data, err = syscall.Mmap(u.fd, int64(IORING_OFF_CQ_RING), int(u.cqRing.ringSize), syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED)
+		if err != nil {
+			u.sysMunmap()
+			return err
+		}
+		u.cqRing.buff = data
+	}
+
+	ringStart := &u.sqRing.buff[0]
+	u.sqRing.kHead = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.sqOffset.head)))
+	u.sqRing.kTail = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.sqOffset.tail)))
+	u.sqRing.kRingMask = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.sqOffset.ringMask)))
+	u.sqRing.kRingEntries = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.sqOffset.ringEntries)))
+	u.sqRing.kFlags = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.sqOffset.flags)))
+	u.sqRing.kDropped = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.sqOffset.dropped)))
+	u.sqRing.array = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.sqOffset.array)))
+
+	size = unsafe.Sizeof(URingSQE{})
+	if p.flags&IORING_SETUP_SQE128 != 0 {
+		size += 64
+	}
+	// TODO: syscall.MAP_POPULATE unsupport for macox
+	buff, err := syscall.Mmap(u.fd, int64(IORING_OFF_SQES), int(size), syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED)
+	if err != nil {
+		_ = u.sysMunmap()
+		return err
+	}
+	u.sqRing.sqeBuff = buff
+
+	cqRingPtr := uintptr(unsafe.Pointer(&u.cqRing.buff[0]))
+	ringStart = &u.cqRing.buff[0]
+
+	u.cqRing.kHead = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.cqOffset.head)))
+	u.cqRing.kTail = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.cqOffset.tail)))
+	u.cqRing.kRingMask = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.cqOffset.ringMsk)))
+	u.cqRing.kRingEntries = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.cqOffset.ringEntries)))
+	u.cqRing.kOverflow = (*uint32)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.cqOffset.overflow)))
+	u.cqRing.cqes = (*URingCQE)(unsafe.Pointer(uintptr(unsafe.Pointer(ringStart)) + uintptr(p.cqOffset.cqes)))
+	if p.cqOffset.flags != 0 {
+		u.cqRing.kFlags = cqRingPtr + uintptr(p.cqOffset.flags)
+	}
+
+	return nil
+}
+
+// Magic offsets for the application to mmap the data it needs
+const (
+	// IORING_OFF_SQ_RING maps sqring to program memory space
+	IORING_OFF_SQ_RING uint64 = 0
+	// IORING_OFF_CQ_RING maps cqring to program memory space
+	IORING_OFF_CQ_RING uint64 = 0x8000000
+	// IORING_OFF_SQES maps sqes array to program memory space
+	IORING_OFF_SQES uint64 = 0x10000000
+)