Fix high-priority issues: concurrency safety, naming consistency, and…

errors/errors.go

@@ -39,4 +39,12 @@ var (
 	ErrUnsupportedOp = errors.New("unsupported operation")
 	// ErrNegativeSize occurs when trying to pass a negative size to a buffer.
 	ErrNegativeSize = errors.New("negative size is invalid")
+	// ErrSocketCreation occurs when socket creation fails.
+	ErrSocketCreation = errors.New("failed to create socket")
+	// ErrInvalidSocketFd occurs when socket file descriptor is invalid.
+	ErrInvalidSocketFd = errors.New("invalid socket file descriptor")
+	// ErrRingSetup occurs when io_uring setup fails.
+	ErrRingSetup = errors.New("failed to setup io_uring")
+	// ErrFileRegister occurs when registering files to io_uring fails.
+	ErrFileRegister = errors.New("failed to register files to io_uring")
 )

example/echoserver.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"github.com/y001j/uringnet"
 	socket "github.com/y001j/uringnet/sockets"
 	"os"
@@ -9,7 +10,7 @@ import (
 )
 
 type testServer struct {
-	UringNet.BuiltinEventEngine
+	uringnet.BuiltinEventEngine
 
 	testloop *uringnet.Ringloop
 	//ring      *uring_net.URingNet
@@ -28,7 +29,7 @@ func (ts *testServer) OnTraffic(data *uringnet.UserData, ringnet *uringnet.URing
 	return uringnet.Echo
 }
 
-func (ts *testServer) OnWritten(data uringnet.UserData) UringNet.Action {
+func (ts *testServer) OnWritten(data uringnet.UserData) uringnet.Action {
 	//buf, _ := c.Next(-1)
 	//thebuffer := ts.testloop.GetBuffer()
 	//fmt.Println("Send Message to Client: \n", string(data))
@@ -50,14 +51,29 @@ func (ts *testServer) OnOpen(data *uringnet.UserData) ([]byte, uringnet.Action)
 }
 
 func main() {
+	if len(os.Args) < 2 {
+		fmt.Println("Usage: echoserver <address>")
+		fmt.Println("Example: echoserver :8080")
+		os.Exit(1)
+	}
+
 	addr := os.Args[1]
 
-	//accptRingNet, _ := uring_net.New(uring_net.NetAddress{uring_net., addr}, 500, true)
-	//TcpAddr
 	options := socket.SocketOptions{TCPNoDelay: socket.TCPNoDelay, ReusePort: true}
-	ringNets, _ := uringnet.NewMany(uringnet.NetAddress{socket.Tcp4, addr}, 3200, true, runtime.NumCPU()*2-2, options, &testServer{}) //runtime.NumCPU()
+	ringNets, err := uringnet.NewMany(uringnet.NetAddress{socket.Tcp4, addr}, 3200, true, runtime.NumCPU()*2-2, options, &testServer{})
+	if err != nil {
+		fmt.Printf("Failed to create uring instances: %v\n", err)
+		os.Exit(1)
+	}
+
+	loop, err := uringnet.SetLoops(ringNets, 3000)
+	if err != nil {
+		fmt.Printf("Failed to set loops: %v\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Printf("Echo server started on %s with %d rings\n", addr, runtime.NumCPU()*2-2)
 
-	loop := uringnet.SetLoops(ringNets, 3000)
 	var waitgroup sync.WaitGroup
 	waitgroup.Add(1)
 

ringloop.go

@@ -36,15 +36,20 @@ const bufLength uint64 = 2048
 //
 //	@Description: set the ringloop for the engine
 //	@param urings
+//	@param bufferSize
 //	@return *Ringloop
-func SetLoops(urings []*URingNet, bufferSize int) *Ringloop {
+//	@return error
+func SetLoops(urings []*URingNet, bufferSize int) (*Ringloop, error) {
 	size := len(urings)
+	if size == 0 {
+		return nil, fmt.Errorf("no uring instances provided")
+	}
+
 	theloop := &Ringloop{}
 	theloop.RingCount = int32(size)
-	//theloop.connections = map[int]*conn{}
 	theloop.RingNet = urings
-	for i := 0; i < size; i++ {
 
+	for i := 0; i < size; i++ {
 		urings[i].ringloop = theloop
 		theloop.RingNet[i] = urings[i]
 		theloop.socketFd = urings[i].SocketFd
@@ -53,20 +58,24 @@ func SetLoops(urings []*URingNet, bufferSize int) *Ringloop {
 		fdstack = append(fdstack, int32(theloop.socketFd))
 		err := theloop.RingNet[i].ring.RegisterFiles(fdstack)
 		if err != nil {
-			return nil
+			return nil, fmt.Errorf("failed to register files for ring %d: %w", i, err)
 		}
 
-		//set buffer
+		// Set buffer
 		sqe2 := theloop.RingNet[i].ring.GetSQEntry()
 		urings[i].Autobuffer = make([][bufLength]byte, bufferSize)
 		uring.ProvideBuf(sqe2, urings[i].Autobuffer, uint32(bufferSize), uint32(bufLength), uint16(i))
 		data := makeUserData(provideBuffer)
 		sqe2.SetUserData(data.id)
 		theloop.RingNet[i].userDataList.Store(data.id, data)
-		fmt.Println("Add Kernel buffer... for ring ", i)
-		_, _ = theloop.RingNet[i].ring.Submit(1, &paraFlags)
+		fmt.Printf("Add Kernel buffer... for ring %d\n", i)
+
+		_, err = theloop.RingNet[i].ring.Submit(1, &paraFlags)
+		if err != nil {
+			return nil, fmt.Errorf("failed to submit buffer for ring %d: %w", i, err)
+		}
 	}
-	return theloop
+	return theloop, nil
 }
 
 func (loop *Ringloop) GetBuffer() [][bufLength]byte {

uringnet.go

@@ -6,6 +6,7 @@ package uringnet
 import (
 	"crypto/tls"
 	"fmt"
+	"github.com/y001j/uringnet/errors"
 	socket "github.com/y001j/uringnet/sockets"
 	"github.com/y001j/uringnet/uring"
 	"golang.org/x/sys/unix"
@@ -118,25 +119,12 @@ type request struct {
 	done    chan struct{}
 }
 
-var increase uint64 = 1
+var increase atomic.Uint64
 
 func makeUserData(state UserdataState) *UserData {
-	defer func() {
-		err := recover() // 内置函数，可以捕获异常
-		if err != nil {
-			fmt.Println("err:", err)
-			fmt.Println("发生异常............")
-		}
-	}()
-
 	userData := new(UserData)
-	//userData := &UserData{
-	//	//ringNet: ringNet,
-	//	state: uint32(state),
-	//}
 	userData.state = uint32(state)
-	userData.id = increase
-	increase++
+	userData.id = increase.Add(1)
 
 	return userData
 }
@@ -581,32 +569,44 @@ func (ringNet *URingNet) read2(Fd int32, sqe *uring.SQEntry) {
 
 // New Creates a new uRingnNet which is used to
 func New(addr NetAddress, size uint, sqpoll bool, options socket.SocketOptions) (*URingNet, error) {
-	//1. set the socket
-	//var ringNet *URingNet
 	ringNet := &URingNet{}
 	ringNet.userDataMap = make(map[uint64]*UserData)
 	ops := socket.SetOptions(string(addr.AddrType), options)
+
+	var err error
 	switch addr.AddrType {
 	case socket.Tcp, socket.Tcp4, socket.Tcp6:
-		ringNet.SocketFd, _, _ = socket.TCPSocket(string(addr.AddrType), addr.Address, true, ops...) //ListenTCPSocket(addr)
+		ringNet.SocketFd, _, err = socket.TCPSocket(string(addr.AddrType), addr.Address, true, ops...)
 	case socket.Udp, socket.Udp4, socket.Udp6:
-		ringNet.SocketFd, _, _ = socket.UDPSocket(string(addr.AddrType), addr.Address, true, ops...)
+		ringNet.SocketFd, _, err = socket.UDPSocket(string(addr.AddrType), addr.Address, true, ops...)
 	case socket.Unix:
-		ringNet.SocketFd, _, _ = socket.UnixSocket(string(addr.AddrType), addr.Address, true, ops...)
-
+		ringNet.SocketFd, _, err = socket.UnixSocket(string(addr.AddrType), addr.Address, true, ops...)
 	default:
-		ringNet.SocketFd = -1
+		return nil, errors.ErrUnsupportedProtocol
+	}
+
+	if err != nil {
+		return nil, fmt.Errorf("%w: %v", errors.ErrSocketCreation, err)
+	}
+
+	if ringNet.SocketFd < 0 {
+		return nil, errors.ErrInvalidSocketFd
 	}
+
 	ringNet.Addr = addr.Address
 	ringNet.Type = addr.AddrType
 
-	//ringNet.userDataList = make(sync.Map, 1024)
-	//Create the io_uring instance
+	// Create the io_uring instance
+	var params *uring.IOUringParams
 	if sqpoll {
-		ringNet.SetUring(size, &uring.IOUringParams{Flags: uring.IORING_SETUP_SQPOLL | uring.IORING_SETUP_SQ_AFF, SQThreadCPU: 1})
-	} else {
-		ringNet.SetUring(size, nil)
+		params = &uring.IOUringParams{Flags: uring.IORING_SETUP_SQPOLL | uring.IORING_SETUP_SQ_AFF, SQThreadCPU: 1}
 	}
+
+	_, err = ringNet.SetUring(size, params)
+	if err != nil {
+		return nil, fmt.Errorf("%w: %v", errors.ErrRingSetup, err)
+	}
+
 	return ringNet, nil
 }
 
@@ -620,38 +620,59 @@ func New(addr NetAddress, size uint, sqpoll bool, options socket.SocketOptions)
 //	@return *[]URingNet
 //	@return error
 func NewMany(addr NetAddress, size uint, sqpoll bool, num int, options socket.SocketOptions, handler EventHandler) ([]*URingNet, error) {
-	//1. set the socket
 	var sockfd int
 	ops := socket.SetOptions(string(addr.AddrType), options)
+
+	var err error
 	switch addr.AddrType {
 	case socket.Tcp, socket.Tcp4, socket.Tcp6:
-		sockfd, _, _ = socket.TCPSocket(string(addr.AddrType), addr.Address, true, ops...) //ListenTCPSocket(addr)
+		sockfd, _, err = socket.TCPSocket(string(addr.AddrType), addr.Address, true, ops...)
 	case socket.Udp, socket.Udp4, socket.Udp6:
-		sockfd, _, _ = socket.UDPSocket(string(addr.AddrType), addr.Address, true, ops...)
+		sockfd, _, err = socket.UDPSocket(string(addr.AddrType), addr.Address, true, ops...)
 	case socket.Unix:
-		sockfd, _, _ = socket.UnixSocket(string(addr.AddrType), addr.Address, true, ops...)
+		sockfd, _, err = socket.UnixSocket(string(addr.AddrType), addr.Address, true, ops...)
 	default:
-		sockfd = -1
+		return nil, errors.ErrUnsupportedProtocol
+	}
+
+	if err != nil {
+		return nil, fmt.Errorf("%w: %v", errors.ErrSocketCreation, err)
 	}
-	uringArray := make([]*URingNet, num) //*URingNet{}
-	//ringNet.userDataList = make(sync.Map, 1024)
-	//Create the io_uring instance
+
+	if sockfd < 0 {
+		return nil, errors.ErrInvalidSocketFd
+	}
+
+	uringArray := make([]*URingNet, num)
+
+	// Create the io_uring instances
 	for i := 0; i < num; i++ {
 		uringArray[i] = &URingNet{}
-		//uringArray[i].userDataMap = make(map[uint64]*UserData)
 		uringArray[i].ReadBuffer = make([]byte, 1024)
 		uringArray[i].WriteBuffer = make([]byte, 1024)
 		uringArray[i].SocketFd = sockfd
 		uringArray[i].Addr = addr.Address
 		uringArray[i].Type = addr.AddrType
 		uringArray[i].Handler = handler
 
+		var params *uring.IOUringParams
 		if sqpoll {
-			uringArray[i].SetUring(size, &uring.IOUringParams{Flags: uring.IORING_SETUP_SQPOLL, Features: uring.IORING_FEAT_NODROP | uring.IORING_FEAT_FAST_POLL | uring.IORING_FEAT_SQPOLL_NONFIXED}) //Features: uring.IORING_FEAT_FAST_POLL|uring.IORING_FEAT_NODROP})
+			params = &uring.IOUringParams{
+				Flags:    uring.IORING_SETUP_SQPOLL,
+				Features: uring.IORING_FEAT_NODROP | uring.IORING_FEAT_FAST_POLL | uring.IORING_FEAT_SQPOLL_NONFIXED,
+			}
 		} else {
-			uringArray[i].SetUring(size, &uring.IOUringParams{Features: uring.IORING_FEAT_FAST_POLL | uring.IORING_FEAT_NODROP})
+			params = &uring.IOUringParams{
+				Features: uring.IORING_FEAT_FAST_POLL | uring.IORING_FEAT_NODROP,
+			}
+		}
+
+		_, err = uringArray[i].SetUring(size, params)
+		if err != nil {
+			return nil, fmt.Errorf("%w: failed to setup uring instance %d: %v", errors.ErrRingSetup, i, err)
 		}
-		fmt.Println("Uring instance initiated!")
+
+		fmt.Printf("Uring instance %d initiated!\n", i)
 	}
 	return uringArray, nil
 }