From 949ff2d8145380a36297ce6109595a979c1c9ac4 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 12 Nov 2025 08:16:16 +0000 Subject: [PATCH] Fix high-priority issues: concurrency safety, naming consistency, and error handling This commit addresses three critical issues identified in the code review: 1. Fix concurrent safety issue with global variable 'increase' - Changed from plain uint64 to atomic.Uint64 - Use atomic.Add(1) for thread-safe ID generation - Removed unnecessary panic recovery code 2. Fix package naming inconsistency in example/echoserver.go - Changed all 'UringNet' references to 'uringnet' (lowercase) - Ensures consistent package naming throughout the codebase 3. Add comprehensive error handling - Added new error types: ErrSocketCreation, ErrInvalidSocketFd, ErrRingSetup, ErrFileRegister - Updated New() function with proper error checking - Updated NewMany() function with proper error handling and validation - Updated SetLoops() to return error and validate inputs - Added error handling to echoserver example with usage instructions - Improved error messages with context using fmt.Errorf with %w These changes improve code reliability, maintainability, and user experience. --- errors/errors.go | 8 ++++ example/echoserver.go | 28 +++++++++--- ringloop.go | 25 +++++++---- uringnet.go | 99 ++++++++++++++++++++++++++----------------- 4 files changed, 107 insertions(+), 53 deletions(-) diff --git a/errors/errors.go b/errors/errors.go index 4343f7d..23d18cd 100644 --- a/errors/errors.go +++ b/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") ) diff --git a/example/echoserver.go b/example/echoserver.go index d252c9c..764d2f5 100644 --- a/example/echoserver.go +++ b/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
") + 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) diff --git a/ringloop.go b/ringloop.go index 90f82a6..580e040 100644 --- a/ringloop.go +++ b/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, ¶Flags) + fmt.Printf("Add Kernel buffer... for ring %d\n", i) + + _, err = theloop.RingNet[i].ring.Submit(1, ¶Flags) + 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 { diff --git a/uringnet.go b/uringnet.go index 863659e..0b9e365 100644 --- a/uringnet.go +++ b/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,25 +620,34 @@ 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 @@ -646,12 +655,24 @@ func NewMany(addr NetAddress, size uint, sqpoll bool, num int, options socket.So 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 }