批量讀寫網絡包- 只爲更高性能

雖然網絡協議棧提供了豐富的功能，讓我們可以方便的實現網絡的數據交流，但是有時候我們對協議棧的性能還是不是那麼滿意的，前幾篇文章我也介紹了通過 XDP 等技術高效處理網絡數據的方式，但是畢竟 XDP 還沒有那麼廣泛的應用，使用起來也還不是那麼簡單，如果我們通過編程語言提供的標準庫來實現數據的讀寫，還怎麼能提高性能呢？今天就介紹一種方式，批量的讀寫數據包。

直觀上看，我們也能理解批收發網絡包比單個的收發網絡包更有效，因爲在普通的單個收發包的邏輯中，每次收發一個包都至少需要一次系統調用 (Send/SendTo, Read/RecvFrom)，而批處理方式下，一次系統調用就可以處理多個網絡包，所以從理論分析來看，批處理方式更有效。不僅僅網絡處理，很多消息隊列、數據存儲，也都會通過批處理方式獲得更好的性能。

在這篇文章中，我並沒有對批量處理網絡包和普通處理單個網絡包的性能做一個測試 (或許 5.1 假期的時候我能補上)，有人做了簡單的測試 [1], 並沒有發現批處理帶來的好處，當然我猜測他的測試可能過於單一或者簡單，cloudflare 也做過百萬 pps 的測試 [2]，批處理方式性能還是很好的。我想你在評估這項技術的時候也根據你的場景做一下性能測試，這樣才能確保這項技術是否適合採用。

我所說的批量處理收發包的技術是通過系統調用sendmmsg和recvmmsg來實現的，目前只支持 Linux 系統。正如 man 手冊上介紹的，它們是在 socket 上收發多個數據包的系統調用 [3]:

sendmmsg - send multiple messages on a socket: int sendmmsg(int sockfd, struct mmsghdr *msgvec, unsigned int vlen,int flags);
recvmmsg - receive multiple messages on a socket: int recvmmsg(int sockfd, struct mmsghdr *msgvec, unsigned int vlen,int flags, struct timespec *timeout);

這兩個系統調用開始加入到 Linux 的版本是 3.0, glibc 自 2.14 版本開始加入。OpenBSD 7.2 也加入了這個系統調用。

注意recvmmsg是一個blocking系統調用，直到收到 vlen 個消息或者超時纔會返回。這兩個系統調用是對sendmsg和recvmsg的擴展。如果你以前研究過，你可能直到對於網絡數據的讀寫有多個系統調用send、sendto、sendmsg、 sendmmsg和read、recv、``、

man send[4] 和 man recv[5] 分別對它們做了詳細的介紹。

send：send 函數用於發送一個數據包，可以是 TCP 連接或 UDP 數據報。類似write, 只不過write沒有 flag 設置。
sendto：sendto 函數與 send 函數類似，但它可以在發送數據包時指定接收方的地址。如果是面向連接的協議如 TCP, dest_addr 和 addrlen 可以忽略，否則如 UDP, 需要指定這兩個參數。
sendmsg：sendmsg 函數可以發送多個緩衝區的數據。同時，它還可以指定一個或多個附加數據。需要指定
sendmmsg：sendmmsg 函數可以在一次調用中發送多個消息，每個消息可以有一個或多個緩衝區。這可以減少系統調用的次數，從而提高效率。

同樣的，接收用的系統調用主要有下面的幾個：

recv: recv 是最基本的接收函數，從套接字接收數據並返回接收到的字節數。它接收的數據沒有附加的信息（比如目標地址等）。類似read, 只不過read沒有 flag 設置。
recvfrom: recvfrom 也是從套接字接收數據，但它還會返回發送者的地址信息，適合用於 UDP 等有地址信息的協議。
recvmsg: recvmsg 可以在接收數據的同時接收其他相關的數據信息（比如接收的數據是否截斷、發送方的 IP 地址等）。它支持多個數據緩衝區的接收，也支持控制消息（cmsg）的接收。
recvmmsg: recvmmsg 是 recvmsg 的多消息版本，可以同時接收多個消息，適合高併發、高吞吐量的場景。

對應到標準庫的 conn 的方法上，以 UDPConn 爲例：

conn.Write(_) : 使用write系統調用
conn.WriteTo(_, _) : 使用sendto系統調用
conn.WriteToUDP(_, _) : 使用sendto系統調用
conn.WriteMsgUDP(_, _, _) : 使用sendmsg系統調用
conn.WriteMsgUDPAddrPort(_, _, _) : 使用sendmsg系統調用
conn.Read(nil): 使用read系統調用
conn.ReadFrom(nil): 使用recvfrom系統調用
conn.ReadMsgUDP(nil, nil): 使用recvmsg系統調用
conn.ReadMsgUDPAddrPort(nil, nil): 使用recvmsg系統調用
conn.ReadFromUDP(nil): 使用recvmsg系統調用
conn.ReadFromUDPAddrPort(nil): 使用recvmsg系統調用

遺憾的是， Go 標準庫中並沒有提供對系統調用sendmmsg和recvmmsg的包裝，這個系統調用在syscall甚至沒有定義，所以像net.UDPConn還沒有對應的批處理方法。2021 年，在 go issue#45886[6] 中 bradfitz 提出爲*UDPConn增加批讀寫消息的提案，這個提案在 2022 年 11 月 10 日被 Russ Cox 批准爲接受狀態，但是目前這個提案處理依然還沒有人領取去實現。

但是，go 的擴展庫 golang.org/x/net[7] 提供了 ReadBatch和WriteBatch的方法，可以提供批量讀寫消息的方法，它實際是對系統調用readmmsg和sendmmsg的封裝。

當然你也可以類似的實現對系統調用的封裝：

func recvmmsg(s uintptr, hs []mmsghdr, flags int) (int, error) {
 n, _, errno := syscall.Syscall6(sysRECVMMSG, s, uintptr(unsafe.Pointer(&hs[0])), uintptr(len(hs)), uintptr(flags), 0, 0)
 return int(n), errnoErr(errno)
}

func sendmmsg(s uintptr, hs []mmsghdr, flags int) (int, error) {
 n, _, errno := syscall.Syscall6(sysSENDMMSG, s, uintptr(unsafe.Pointer(&hs[0])), uintptr(len(hs)), uintptr(flags), 0, 0)
 return int(n), errnoErr(errno)
}

不過目前它們都有一個問題，類似ReadBatch是 blocking 的，如果沒有足夠的消息被接收到，當前線程會被阻塞住。線程數很少的情況下沒有問題，但是如果線程數很多，就會有資源不足和性能的問題。bradfitz 在提案中期望能和標準庫的 net poller 集成，這樣就避免線程被阻塞中，所以你可以期待這個功能早日能實現。

那麼，我們就看看目前最好的實現批量讀寫 UDP 消息的方式吧，也就是利用 ipv4 包的方式。

使用 ipv4.PacketConn

我們可以利用ipv4.PacketConn的方式，它提供了ReadBatch和WriteBatch方法：

func (c *PacketConn) ReadBatch(ms []Message, flags int) (int, error): 讀取一批消息，它返回讀取的消息數，最大爲 len(ms)
func (c *PacketConn) WriteBatch(ms []Message, flags int) (int, error): 寫一批消息，它返回寫成功的消息數

接下來，我們以一個 UDP 客戶端和服務端的例子，演示批量讀寫的能力。

下面是客戶端的代碼。它首先使用標準庫創建的了一個*net.UDPConn的實例，並使用ipv4.NewPacketConn轉換成*ipv4.PacketConn。接下來準備了 10 條消息，需要準備成ipv4.Message類型。如果你在產品中使用，最好使用 Pool 池化這些對象，這個例子比較簡單，就沒有考慮性能的問題，值演示批量讀寫的功能。準備好數據後就調用WriteBatch批量寫入，這裏也每考慮如果有的每寫成功怎麼處理，假定都寫入成功了。

接下來讀取回包，假定服務器都要把每一條消息返回，所以這裏如果批量讀沒有讀取完，還會接着讀，直到把所有的消息都讀取回來。

package main

import (
 "fmt"
 "net"

 "golang.org/x/net/ipv4"
)
func main() {
 remote, err := net.ResolveUDPAddr("udp", "localhost:9999")
 if err != nil {
  panic(err)
 }

 conn, err := net.Dial("udp", "localhost:9999")
 if err != nil {
  panic(err)
 }
 defer conn.Close()

 pconn := ipv4.NewPacketConn(conn.(*net.UDPConn))

 // write with a batch of 10 messages
 batch := 10
 msgs := make([]ipv4.Message, batch)
 for i := 0; i < batch; i++ {
  msgs[i] = ipv4.Message{
   Buffers: [][]byte{[]byte(fmt.Sprintf("hello batch %d", i))},
   Addr:    remote,
  }
 }
 n, err := pconn.WriteBatch(msgs, 0)
 if err != nil {
  panic(err)
 }
 fmt.Printf("sent %d messages\n", n)

 // read 10 messages with batch
 count := 0
 for count < batch {
  n, err := pconn.ReadBatch(msgs, 0)
  if err != nil {
   panic(err)
  }
  count += n

  for i := 0; i < n; i++ {
   fmt.Println(string(msgs[i].Buffers[0]))
  }
 }
}

服務端的代碼類似，先批量接收消息，再把消息原樣批量寫回：

package main

import (
 "fmt"
 "net"

 "golang.org/x/net/ipv4"
)

func main() {
 addr, err := net.ResolveUDPAddr("udp", ":9999")
 if err != nil {
  panic(err)
 }
 conn, err := net.ListenUDP("udp", addr)
 if err != nil {
  panic(err)
 }
 defer conn.Close()

 pconn := ipv4.NewPacketConn(conn)
 fmt.Println("server listening on", addr)

 batch := 10
 msgs := make([]ipv4.Message, batch)
 for i := 0; i < batch; i++ {
  msgs[i] = ipv4.Message{
   Buffers: [][]byte{make([]byte, 1024)},
  }
 }

 for {
  n, err := pconn.ReadBatch(msgs, 0)
  if err != nil {
   panic(err)
  }

  tn := 0
  for tn < n {
   nn, err := pconn.WriteBatch(msgs[tn:n], 0)
   if err != nil {
    panic(err)
   }
   tn += nn
  }
 }
}

運行 server 和 client, 可以在 client 看到 10 條消息都收到了:

ubuntu@lab:~/network-programming/ch04/mmsg/sendmmsg/client$ ./client
sent 10 messages
hello batch 0
hello batch 1
hello batch 2
hello batch 3
hello batch 4
hello batch 5
hello batch 6
hello batch 7
hello batch 8
hello batch 9

使用 ipv4.Conn

其實使用 ipv4.Conn 也能實現批量讀寫，底層是一樣的。這個類型底層是golang.org/x/net/internal/socket.Conn, 它包含下面的批量讀寫消息的方法：

SendMsgs(ms []Message, flags int) (int, error): 批量寫消息，包裝系統調用sendmmsg, 返回發送成功的消息數量
RecvMsgs(ms []Message, flags int) (int, error)：批量讀消息，包裝系統調用recvmmsg, 返回接收成功的消息數量

使用方式幾乎和上面的一樣，只不過方法名不同。下面是客戶端的代碼：

package main

import (
 "fmt"
 "net"

 "golang.org/x/net/ipv4"
)

func main() {
 remote, err := net.ResolveUDPAddr("udp", "localhost:9999")
 if err != nil {
  panic(err)
 }

 conn, err := net.Dial("udp", "localhost:9999")
 if err != nil {
  panic(err)
 }
 defer conn.Close()

 pconn := ipv4.NewConn(conn)

 // write with a batch of 10 messages
 batch := 10
 msgs := make([]ipv4.Message, batch)
 for i := 0; i < batch; i++ {
  msgs[i] = ipv4.Message{
   Buffers: [][]byte{[]byte(fmt.Sprintf("hello batch %d", i))},
   Addr:    remote,
  }
 }
 n, err := pconn.SendMsgs(msgs, 0)
 if err != nil {
  panic(err)
 }
 fmt.Printf("sent %d messages\n", n)

 // read 10 messages with batch
 count := 0
 for count < batch {
  n, err := pconn.RecvMsgs(msgs, 0)
  if err != nil {
   panic(err)
  }
  count += n

  for i := 0; i < n; i++ {
   fmt.Println(string(msgs[i].Buffers[0]))
  }
 }
}

下面是服務端的代碼：

package main

import (
 "fmt"
 "net"

 "golang.org/x/net/ipv4"
)

func main() {
 addr, err := net.ResolveUDPAddr("udp", ":9999")
 if err != nil {
  panic(err)
 }
 conn, err := net.ListenUDP("udp", addr)
 if err != nil {
  panic(err)
 }
 defer conn.Close()

 pconn := ipv4.NewConn(conn)
 fmt.Println("server listening on", addr)

 batch := 10
 msgs := make([]ipv4.Message, batch)
 for i := 0; i < batch; i++ {
  msgs[i] = ipv4.Message{
   Buffers: [][]byte{make([]byte, 1024)},
  }
 }

 for {
  n, err := pconn.RecvMsgs(msgs, 0)
  if err != nil {
   panic(err)
  }

  tn := 0
  for tn < n {
   nn, err := pconn.SendMsgs(msgs[tn:n], 0)
   if err != nil {
    panic(err)
   }
   tn += nn
  }
 }
}

參考資料

[1]

測試: https://hpnpl.net/posts/batchconn-mmsg-go/

[2]

測試: https://blog.cloudflare.com/how-to-receive-a-million-packets/

[3]

系統調用: https://man7.org/linux/man-pages/man2/sendmmsg.2.html

[4]

man send: https://manpages.ubuntu.com/manpages/bionic/man2/send.2.html

[5]

man recv: https://manpages.ubuntu.com/manpages/bionic/man2/recv.2.html

[6]

go issue#45886: https://github.com/golang/go/issues/45886

[7]

golang.org/x/net: https://pkg.go.dev/golang.org/x/net/ipv4#PacketConn.ReadBatch

[8]

使用 AF_XDP Socket 更高效的網絡傳輸: https://colobu.com/2023/04/17/use-af-xdp-socket/

[9]

使用底層的 syscall.Socket 實現網絡編程: https://colobu.com/2023/04/09/use-syscall-Socket-in-go/

[10]

每秒 1 百萬的包傳輸，幾乎不耗 CPU 的那種: https://colobu.com/2023/04/02/support-1m-pps-with-zero-cpu-usage/

[11]

像閃電一樣快速掃描全網的 IP 地址: https://colobu.com/2023/03/19/scan-all-IP-addresses-of-mainland-fastly-like-lightning/

[12]

零拷貝技術第二篇：Go 語言中的應用: https://colobu.com/2022/11/21/zero-copy-and-how-to-use-it-in-go-2/

[13]

零拷貝技術第一篇：綜述: https://colobu.com/2022/11/19/zero-copy-and-how-to-use-it-in-go/

[14]

使用 BPF, 將 Go 網絡程序的吞吐提升 8 倍: https://colobu.com/2022/06/05/use-bpf-to-make-the-go-network-program-8x-faster/

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/8h2QSHznd3K1MMclhdJv3w

使用 ipv4.PacketConn

使用 ipv4.Conn

猜你喜歡