go - 如何使用 Go 最大化每秒 UDP 数据包？

Question

我正在尝试在 Go 中编写一个简单的中继服务器，以便在两个客户端之间发送游戏数据。客户端发送 UDP 注册消息，该消息由协议(protocol) ID、匹配 ID 和客户端 ID 组成；这些消息用于设置地址的哈希映射，以便当标准(游戏数据)消息到达时，我们可以将其映射到其预先注册的接收者(无需在每条消息中发送该数据)。 (我使用 github.com/cornelk/hashmap 如果这有任何区别并且服务器是 highcpu-16 GCP Compute 实例)

此设置适用于少量客户端(每秒约 30 条消息)。但是，当我增加负载测试时，输出带宽会随着输入带宽的持续上升而趋于平稳。我已经用 vmstat 完成了日志记录和 ifstat (以及一些 dropwatch 监控，这表明在软件级别丢弃了大量数据包)。在我看来，数据包正在被丢弃，因为我的 go 服务器读取它们的速度不够快。

最初，我为每个 cpu 核心使用了一个 goroutine:

runtime.GOMAXPROCS(runtime.NumCPU())

connection, err := net.ListenUDP("udp", &addr)
if err != nil {
    panic(err)
}

for i := 0; i < runtime.NumCPU(); i++ {
    go listen(connection, c)
}

我也尝试过重用端口并分别监听每个 goroutine(使用 github.com/libp2p/go-reuseport 包)。

最后，我尝试为传入和传出消息设置缓冲 channel ，以尽量减少监听 goroutine 不提取消息的时间。

我尝试一次处理 100K 玩家时缺少什么？感觉无论我的方法如何，我都无法在不丢失大量数据包的情况下超过 3000 名玩家。

我的 listen函数已经徒劳地经历了很多次迭代，试图提高每秒数据包，但通常是:

// Emits SIGABRT to the interrupts channel if an error occurs outside of individual message handling.
func listen(connection *net.UDPConn, interrupts chan os.Signal/*, inbox chan IncomingMessage*/) {
    buffer := make([]byte, 1024)
    n, remoteAddr, err := 0, new(net.UDPAddr), error(nil)
    for err == nil {
        n, remoteAddr, err = connection.ReadFromUDP(buffer)
        if err != nil {
            continue
        }

        //log.Println("Received", n, "bytes", hex.EncodeToString(buffer[:n]))
        if n < 2 {
            log.Println("Minimum packet length is 2 bytes. Received a packet of length", n, "bytes")
            continue
        }

        //inbox<-IncomingMessage{
        //    sender: remoteAddr,
        //    data:   append([]byte(nil), buffer[:n]...),
        //}

        //go handlePacket(inbox, remoteAddr, append([]byte(nil), buffer[:n]...), n)

        protocolId := binary.LittleEndian.Uint16(buffer)
        if protocolId == registrationProtocolId {
            // Start a goroutine to handle the packet (copy the buffer minus the protocol id))
            //go handleRegistrationPacket(connection, remoteAddr, append([]byte(nil), buffer[:n]...))
            handleRegistrationPacket(outbox, remoteAddr, buffer[:n])
        } else if protocolId == matchProtocolId {
            //go handleStandardPacket(connection, remoteAddr.String(), append([]byte(nil), buffer[:n]...))
            handleStandardPacket(outbox, remoteAddr.String(), buffer[:n])
        } else {
            log.Println("Unrecognised protocol id: ", protocolId)
        }
    }
    log.Println("Listener failed:", err)
    interrupts<-syscall.SIGABRT
}

此图显示了 1600 个并发游戏(3200 个客户端)。超过某个点，传出的 KB/s 停止攀升。 CPU甚至没有出汗。

Answer 1

我重读了How to receive a million packets per second ，这表明使用 SO_REUSEPORT套接字上的选项是正确的方法。在下面的代码中，我使用机器每个核心的连接(在我的例子中是 16 个)和每个连接 4 个发件箱处理程序。 CPU 使用率略高，每个连接 4 个，但似乎比每个连接 1 个更可靠(尽管需要测试和明确的答案)。

下面未显示的是 listen函数启动 goroutine 来处理传入的消息并立即返回监听。 outbox传递给处理函数作为发送消息的一种方式。

type OutgoingMessage struct {
    recipient *net.UDPAddr
    data      []byte
}

// ...

func beginListen(c chan os.Signal) {
    addr := net.UDPAddr{
        Port: 1234,
        IP:   net.IP{0, 0, 0, 0},
    }

    connection, err := reuseport.ListenPacket("udp", addr.String())

    if err != nil {
        panic(err)
    }

    outbox := make(chan OutgoingMessage, maxQueueSize)

    sendFromOutbox := func() {
        n, err := 0, error(nil)
        for msg := range outbox {
            n, err = connection.(*net.UDPConn).WriteToUDP(msg.data, msg.recipient)
            if err != nil {
                panic(err)
            }
            if n != len(msg.data) {
                log.Println("Tried to send", len(msg.data), "bytes but only sent ", n)
            }
        }
    }

    for i := 1;  i <= 4; i++ {
        go sendFromOutbox()
    }

    listen(connection.(*net.UDPConn), c, outbox)

    close(outbox)
}

func main() {
    log.Println("Starting...")

    runtime.GOMAXPROCS(runtime.NumCPU())

    c := make(chan os.Signal, 1)

    for i := 0; i < runtime.NumCPU(); i++ {
        go beginListen(c)
    }

    // ...
}

下图显示了与之前的差异:


(到目前为止，这是一个高峰；变量名不会保留，我知道我应该继续写入未发送的字节，直到发送所有数据。)