深入理解 Go 高性能網(wǎng)絡(luò)框架 nbio
前言
nbio 項(xiàng)目還包括建立在 nbio 基礎(chǔ)上的nbhttp，但這不在我們的討論范圍之內(nèi)。
與 evio 一樣，nbio 也采用經(jīng)典的 Reactor 模式。事實(shí)上，Go 中的許多異步網(wǎng)絡(luò)框架都是基于這種模式設(shè)計(jì)的。
我們先看看如何執(zhí)行 nbio 代碼。
(1) 服務(wù)器：
package main
import (
"fmt"
"github.com/lesismal/nbio"
)
func main() {
g := nbio.NewGopher(nbio.Config{
Network: "tcp",
Addrs: []string{":8888"},
MaxWriteBufferSize: 6 * 1024 * 1024,
})
g.OnData(func(c *nbio.Conn, data []byte) {
c.Write(append([]byte{}, data...))
})
err := g.Start()
if err != nil {
fmt.Printf("nbio.Start failed: %v\n", err)
return
}
defer g.Stop()
g.Wait()
}package main
import (
"fmt"
"github.com/lesismal/nbio"
)
func main() {
g := nbio.NewGopher(nbio.Config{
Network: "tcp",
Addrs: []string{":8888"},
MaxWriteBufferSize: 6 * 1024 * 1024,
})
g.OnData(func(c *nbio.Conn, data []byte) {
c.Write(append([]byte{}, data...))
})
err := g.Start()
if err != nil {
fmt.Printf("nbio.Start failed: %v\n", err)
return
}
defer g.Stop()
g.Wait()
}我們用nbio.NewGopher() 函數(shù)創(chuàng)建新的引擎實(shí)例，通過nbio.Config 結(jié)構(gòu)來(lái)配置引擎實(shí)例，包括:
- Network（網(wǎng)絡(luò)）：使用的網(wǎng)絡(luò)類型，本例中為 "TCP"。
- Addrs（地址）：服務(wù)器應(yīng)該監(jiān)聽的地址和端口，這里是":8888"（監(jiān)聽本地計(jì)算機(jī)的 8888 端口）。
- MaxWriteBufferSize（最大寫緩沖區(qū)大?。簩懢彌_區(qū)的最大大小，此處設(shè)置為 6MB。
我們還可以進(jìn)一步探索其他配置。然后，我們通過引擎實(shí)例g.OnData() 注冊(cè)數(shù)據(jù)接收回調(diào)函數(shù)，該回調(diào)函數(shù)會(huì)在收到數(shù)據(jù)時(shí)調(diào)用?；卣{(diào)函數(shù)需要兩個(gè)參數(shù)：連接對(duì)象c 和接收到的數(shù)據(jù)data。在回調(diào)函數(shù)中，通過c.Write() 方法將接收到的數(shù)據(jù)寫回客戶端。
(2) 客戶端：
package main
import (
"bytes"
"context"
"fmt"
"math/rand"
"time"
"github.com/lesismal/nbio"
"github.com/lesismal/nbio/logging"
)
func main() {
var (
ret []byte
buf = make([]byte, 1024*1024*4)
addr = "localhost:8888"
ctx, _ = context.WithTimeout(context.Background(), 60*time.Second)
)
logging.SetLevel(logging.LevelInfo)
rand.Read(buf)
g := nbio.NewGopher(nbio.Config{})
done := make(chan int)
g.OnData(func(c *nbio.Conn, data []byte) {
ret = append(ret, data...)
if len(ret) == len(buf) {
if bytes.Equal(buf, ret) {
close(done)
}
}
})
err := g.Start()
if err != nil {
fmt.Printf("Start failed: %v\n", err)
}
defer g.Stop()
c, err := nbio.Dial("tcp", addr)
if err != nil {
fmt.Printf("Dial failed: %v\n", err)
}
g.AddConn(c)
c.Write(buf)
select {
case <-ctx.Done():
logging.Error("timeout")
case <-done:
logging.Info("success")
}
}乍一看似乎有點(diǎn)繁瑣，實(shí)際上服務(wù)器和客戶端共享同一套結(jié)構(gòu)。
客戶端通過nbio.Dial 與服務(wù)器連接，連接成功后封裝到nbio.Conn 中。這里nbio.Conn 實(shí)現(xiàn)了標(biāo)準(zhǔn)庫(kù)中的net.Conn 接口，最后通過g.AddConn(c) 添加此連接，并向服務(wù)器寫入數(shù)據(jù)。服務(wù)器收到數(shù)據(jù)后，其處理邏輯是將數(shù)據(jù)原封不動(dòng)發(fā)送回客戶端，客戶端收到數(shù)據(jù)后，會(huì)觸發(fā)OnData 回調(diào)，該回調(diào)會(huì)檢查收到的數(shù)據(jù)長(zhǎng)度是否與發(fā)送的數(shù)據(jù)長(zhǎng)度一致，如果一致，則關(guān)閉連接。
下面深入探討幾個(gè)關(guān)鍵結(jié)構(gòu)。
type Engine struct {
//...
sync.WaitGroup
//...
mux sync.Mutex
wgConn sync.WaitGroup
network string
addrs []string
//...
connsStd map[*Conn]struct{}
connsUnix []*Conn
listeners []*poller
pollers []*poller
onOpen func(c *Conn)
onClose func(c *Conn, err error)
onRead func(c *Conn)
onData func(c *Conn, data []byte)
onReadBufferAlloc func(c *Conn) []byte
onReadBufferFree func(c *Conn, buffer []byte)
//...
}Engine 本質(zhì)上是核心管理器，負(fù)責(zé)管理所有監(jiān)聽器、輪詢器和工作輪詢器。
這兩種輪詢器有什么區(qū)別？
區(qū)別在于責(zé)任不同。
監(jiān)聽輪詢器只負(fù)責(zé)接受新連接。當(dāng)一個(gè)新的客戶端conn 到達(dá)時(shí)，它會(huì)從pollers 中選擇一個(gè)工作輪詢器，并將conn 添加到相應(yīng)的工作輪詢器中。隨后，工作輪詢器負(fù)責(zé)處理該連接的讀/寫事件。
因此當(dāng)我們啟動(dòng)程序時(shí)，如果只監(jiān)聽一個(gè)地址，程序中的輪詢次數(shù)等于 1（監(jiān)聽器輪詢器）+pollerNum。
通過上述字段，可以自定義配置和回調(diào)。例如，可以在新連接到達(dá)時(shí)設(shè)置onOpen 回調(diào)函數(shù)，或在數(shù)據(jù)到達(dá)時(shí)設(shè)置onData 回調(diào)函數(shù)等。
type Conn struct {
mux sync.Mutex
p *poller
fd int
//...
writeBuffer []byte
//...
DataHandler func(c *Conn, data []byte)
}Conn 結(jié)構(gòu)代表網(wǎng)絡(luò)連接，每個(gè)Conn 只屬于一個(gè)輪詢器。當(dāng)數(shù)據(jù)一次寫不完時(shí)，剩余數(shù)據(jù)會(huì)先存儲(chǔ)在writeBuffer 中，等待下一個(gè)可寫事件繼續(xù)寫入。
type poller struct {
g *Engine
epfd int
evtfd int
index int
shutdown bool
listener net.Listener
isListener bool
unixSockAddr string
ReadBuffer []byte
pollType string
}至于poller 結(jié)構(gòu)，這是一個(gè)抽象概念，用于管理底層多路復(fù)用 I/O 操作（如 Linux 的 epoll、Darwin 的 kqueue 等）。
注意pollType，nbio 默認(rèn)使用電平觸發(fā)（LT）模式的 epoll，但用戶也可以將其設(shè)置為邊緣觸發(fā)（ET）模式。
介紹完基本結(jié)構(gòu)后，我們來(lái)看看代碼流程。
當(dāng)啟動(dòng)服務(wù)器代碼時(shí)，調(diào)用Start：
func (g *Engine) Start() error {
//...
switch g.network {
// 第一部分: 初始化 listener
case "unix", "tcp", "tcp4", "tcp6":
for i := range g.addrs {
ln, err := newPoller(g, true, i)
if err != nil {
for j := 0; j < i; j++ {
g.listeners[j].stop()
}
return err
}
g.addrs[i] = ln.listener.Addr().String()
g.listeners = append(g.listeners, ln)
}
//...
// 第二部分: 初始化一定數(shù)量的輪詢器
for i := 0; i < g.pollerNum; i++ {
p, err := newPoller(g, false, i)
if err != nil {
for j := 0; j < len(g.listeners); j++ {
g.listeners[j].stop()
}
for j := 0; j < i; j++ {
g.pollers[j].stop()
}
return err
}
g.pollers[i] = p
}
//...
// 第三部分: 啟動(dòng)所有工作輪詢器
for i := 0; i < g.pollerNum; i++ {
g.pollers[i].ReadBuffer = make([]byte, g.readBufferSize)
g.Add(1)
go g.pollers[i].start()
}
// 第四部分: 啟動(dòng)所有監(jiān)聽器
for _, l := range g.listeners {
g.Add(1)
go l.start()
}
//... (忽略 UDP)
//...
}代碼比較容易理解，分為四個(gè)部分：
第一部分：初始化監(jiān)聽器
根據(jù)g.network 值（如 "unix"、"tcp"、"tcp4"、"tcp6"），為每個(gè)要監(jiān)聽的地址創(chuàng)建一個(gè)新的輪詢器。該輪詢器主要管理監(jiān)聽套接字上的事件。如果在創(chuàng)建過程中發(fā)生錯(cuò)誤，則停止所有先前創(chuàng)建的監(jiān)聽器并返回錯(cuò)誤信息。
第二部分：初始化一定數(shù)量的輪詢器
創(chuàng)建指定數(shù)量（pollerNum）的輪詢器，用于處理已連接套接字上的讀/寫事件。如果在創(chuàng)建過程中發(fā)生錯(cuò)誤，將停止所有監(jiān)聽器和之前創(chuàng)建的工作輪詢器，然后返回錯(cuò)誤信息。
第三部分：?jiǎn)?dòng)所有工作輪詢器投票站
為每個(gè)輪詢器分配讀緩沖區(qū)并啟動(dòng)。
第四部分：?jiǎn)?dòng)所有監(jiān)聽器
啟動(dòng)之前創(chuàng)建的所有監(jiān)聽器，并開始監(jiān)聽各自地址上的連接請(qǐng)求。
關(guān)于輪詢器的啟動(dòng)：
func (p *poller) start() {
defer p.g.Done()
//...
if p.isListener {
p.acceptorLoop()
} else {
defer func() {
syscall.Close(p.epfd)
syscall.Close(p.evtfd)
}()
p.readWriteLoop()
}
}分為兩種情況。如果是監(jiān)聽輪詢器：
func (p *poller) acceptorLoop() {
// 如果不希望將當(dāng)前 goroutine 調(diào)度到其他操作線程。
if p.g.lockListener {
runtime.LockOSThread()
defer runtime.UnlockOSThread()
}
p.shutdown = false
for !p.shutdown {
conn, err := p.listener.Accept()
if err == nil {
var c *Conn
c, err = NBConn(conn)
if err != nil {
conn.Close()
continue
}
// p.g.pollers[c.Hash()%len(p.g.pollers)].addConn(c)
} else {
var ne net.Error
if ok := errors.As(err, &ne); ok && ne.Timeout() {
logging.Error("NBIO[%v][%v_%v] Accept failed: temporary error, retrying...", p.g.Name, p.pollType, p.index)
time.Sleep(time.Second / 20)
} else {
if !p.shutdown {
logging.Error("NBIO[%v][%v_%v] Accept failed: %v, exit...", p.g.Name, p.pollType, p.index, err)
}
break
}
}
}
}監(jiān)聽輪詢器等待新連接的到來(lái)，并在接受后將其封裝到nbio.Conn 中，并將Conn 添加到相應(yīng)的工作輪詢器中。
func (p *poller) addConn(c *Conn) {
c.p = p
if c.typ != ConnTypeUDPServer {
p.g.onOpen(c)
}
fd := c.fd
p.g.connsUnix[fd] = c
err := p.addRead(fd)
if err != nil {
p.g.connsUnix[fd] = nil
c.closeWithError(err)
logging.Error("[%v] add read event failed: %v", c.fd, err)
}
}這里一個(gè)有趣的設(shè)計(jì)是對(duì)conn 的管理。該結(jié)構(gòu)是個(gè)切片，直接使用conn 的fd 作為索引。這樣做的好處是：
- 在連接數(shù)較多的情況下，垃圾回收時(shí)的負(fù)擔(dān)要比使用 map 小。
- 可以防止序列號(hào)問題。
最后，通過調(diào)用addRead 將相應(yīng)的conn fd 添加到 epoll 中。
func (p *poller) addRead(fd int) error {
switch p.g.epollMod {
case EPOLLET:
return syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_ADD, fd, &syscall.EpollEvent{Fd: int32(fd), Events: syscall.EPOLLERR | syscall.EPOLLHUP | syscall.EPOLLRDHUP | syscall.EPOLLPRI | syscall.EPOLLIN | syscall.EPOLLET})
default:
return syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_ADD, fd, &syscall.E
pollEvent{Fd: int32(fd), Events: syscall.EPOLLERR | syscall.EPOLLHUP | syscall.EPOLLRDHUP | syscall.EPOLLPRI | syscall.EPOLLIN})
}
}這里不注冊(cè)寫事件是合理的，因?yàn)樾逻B接上沒有數(shù)據(jù)要發(fā)送。這種方法避免了一些不必要的系統(tǒng)調(diào)用，從而提高了程序性能。
如果啟動(dòng)的是工作輪詢器，它的工作就是等待新增conn 事件，并進(jìn)行相應(yīng)處理。
func (p *poller) readWriteLoop() {
//...
msec := -1
events := make([]syscall.EpollEvent, 1024)
//...
for !p.shutdown {
n, err := syscall.EpollWait(p.epfd, events, msec)
if err != nil && !errors.Is(err, syscall.EINTR) {
return
}
if n <= 0 {
msec = -1
continue
}
msec = 20
// 遍歷事件
for _, ev := range events[:n] {
fd := int(ev.Fd)
switch fd {
case p.evtfd:
default:
c := p.getConn(fd)
if c != nil {
if ev.Events&epollEventsError != 0 {
c.closeWithError(io.EOF)
continue
}
// 如果可寫，則刷新數(shù)據(jù)
if ev.Events&epollEventsWrite != 0 {
c.flush()
}
// 讀取事件
if ev.Events&epollEventsRead != 0 {
if p.g.onRead == nil {
for i := 0; i < p.g.maxConnReadTimesPerEventLoop; i++ {
buffer := p.g.borrow(c)
rc, n, err := c.ReadAndGetConn(buffer)
if n > 0 {
p.g.onData(rc, buffer[:n])
}
p.g.payback(c, buffer)
//...
if n < len(buffer) {
break
}
}
} else {
p.g.onRead(c)
}
}
} else {
syscall.Close(fd)
}
}
}
}
}這段代碼也很簡(jiǎn)單，等待事件到來(lái)，遍歷事件列表，并相應(yīng)處理每個(gè)事件。
func EpollWait(epfd int, events []EpollEvent, msec int) (n int, err error)在EpollWait 中，只有msec 是用戶可修改的。通常，我們?cè)O(shè)置msec = -1 使函數(shù)阻塞，直到至少有一個(gè)事件發(fā)生；否則，函數(shù)將無(wú)限期阻塞。當(dāng)事件較少時(shí)，這種方法非常有用，能最大限度減少 CPU 占用。
如果想盡快響應(yīng)事件，可以設(shè)置msec = 0，這樣EpollWait 就能立即返回，無(wú)需等待任何事件。在這種情況下，程序可能會(huì)更頻繁調(diào)用EpollWait，可以在事件發(fā)生后立即處理事件，從而提高 CPU 使用率。
如果程序可以容忍一定延遲，并且希望降低 CPU 占用率，可以將msec 設(shè)置為正數(shù)。這樣，EpollWait 就會(huì)在指定時(shí)間內(nèi)等待事件發(fā)生。如果在這段時(shí)間內(nèi)沒有事件發(fā)生，函數(shù)將返回，可以選擇稍后再次調(diào)用EpollWait。這種方法可以降低 CPU 占用率，但可能導(dǎo)致響應(yīng)時(shí)間延長(zhǎng)。
nbio 會(huì)根據(jù)事件計(jì)數(shù)調(diào)整msec 值。如果計(jì)數(shù)大于 0，則msec 設(shè)置為 20。
字節(jié)跳動(dòng)的 netpoll 代碼與此類似；如果事件計(jì)數(shù)大于 0 ，則將msec 設(shè)置為 0；如果事件計(jì)數(shù)小于或等于 0，則將msec 設(shè)置為-1，然后調(diào)用Gosched() 以主動(dòng)退出當(dāng)前 goroutine。
var msec = -1
for {
n, err = syscall.EpollWait(epfd, events, msec)
if n <= 0 {
msec = -1
runtime.Gosched()
continue
}
msec = 0
...
}不過，nbio 中的自愿切換代碼已被注釋掉。根據(jù)作者的解釋，最初他參考了字節(jié)跳動(dòng)的方法，并添加了自愿切換功能。
不過，在對(duì) nbio 進(jìn)行性能測(cè)試時(shí)發(fā)現(xiàn)，添加或不添加自愿切換功能對(duì)性能并無(wú)明顯影響，因此最終決定將其刪除。
事件處理部分
如果是可讀事件，則可以通過內(nèi)置或自定義內(nèi)存分配器獲取相應(yīng)的緩沖區(qū)，然后調(diào)用ReadAndGetConn 讀取數(shù)據(jù)，無(wú)需每次都分配緩沖區(qū)。
如果是可寫事件，則會(huì)調(diào)用flush 發(fā)送緩沖區(qū)中未發(fā)送的數(shù)據(jù)。
func (c *Conn) flush() error {
//.....
old := c.writeBuffer
n, err := c.doWrite(old)
if err != nil && !errors.Is(err, syscall.EINTR) && !errors.Is(err, syscall.EAGAIN) {
//.....
}
if n < 0 {
n = 0
}
left := len(old) - n
// 描述尚未完成，因此將其余部分存儲(chǔ)在writeBuffer中以備下次寫入。
if left > 0 {
if n > 0 {
c.writeBuffer = mempool.Malloc(left)
copy(c.writeBuffer, old[n:])
mempool.Free(old)
}
// c.modWrite()
} else {
mempool.Free(old)
c.writeBuffer = nil
if c.wTimer != nil {
c.wTimer.Stop()
c.wTimer = nil
}
// 解釋完成后，首先將conn重置為僅讀取事件。
c.resetRead()
//...
}
c.mux.Unlock()
return nil
}邏輯也很簡(jiǎn)單，有多少就寫多少，如果寫不完，就把剩余數(shù)據(jù)放回writeBuffer，然后在epollWait 觸發(fā)時(shí)再次寫入。
如果寫入完成，則不再有數(shù)據(jù)要寫入，將此連接的事件重置為讀取事件。
主邏輯基本上就是這樣。
等等，最初提到有新連接進(jìn)入時(shí)，只注冊(cè)了連接的讀事件，并沒有注冊(cè)寫事件。寫事件是什么時(shí)候注冊(cè)的？
當(dāng)然是在調(diào)用conn.Write 時(shí)注冊(cè)的。
g := nbio.NewGopher(nbio.Config{
Network: "tcp",
Addrs: []string{":8888"},
MaxWriteBufferSize: 6 * 1024 * 1024,
})
g.OnData(func(c *nbio.Conn, data []byte) {
c.Write(append([]byte{}, data...))
})當(dāng) Conn 數(shù)據(jù)到達(dá)時(shí)，底層會(huì)在讀取數(shù)據(jù)后回調(diào)OnData 函數(shù)，此時(shí)可以調(diào)用Write 向另一端發(fā)送數(shù)據(jù)。
g := nbio.NewGopher(nbio.Config{
Network: "tcp",
Addrs: []string{":8888"},
MaxWriteBufferSize: 6 * 1024 * 1024,
})
g.OnData(func(c *nbio.Conn, data []byte) {
c.Write(append([]byte{}, data...))
})
// 當(dāng)數(shù)據(jù)到達(dá)conn時(shí)，底層將讀取數(shù)據(jù)并回調(diào)OnData函數(shù)。此時(shí)，您可以調(diào)用Write來(lái)向另一端發(fā)送數(shù)據(jù)。
func (c *Conn) Write(b []byte) (int, error) {
//....
n, err := c.write(b)
if err != nil && !errors.Is(err, syscall.EINTR) && !errors.Is(err, syscall.EAGAIN) {
//.....
return n, err
}
if len(c.writeBuffer) == 0 {
if c.wTimer != nil {
c.wTimer.Stop()
c.wTimer = nil
}
} else {
//仍然有數(shù)據(jù)未寫入，添加寫事件。
c.modWrite()
}
//.....
return n, err
}
func (c *Conn) write(b []byte) (int, error) {
//...
if len(c.writeBuffer) == 0 {
n, err := c.doWrite(b)
if err != nil && !errors.Is(err, syscall.EINTR) && !errors.Is(err, syscall.EAGAIN) {
return n, err
}
//.....
left := len(b) - n
// 未完成，將剩余數(shù)據(jù)寫入writeBuffer。
if left > 0 && c.typ == ConnTypeTCP {
c.writeBuffer = mempool.Malloc(left)
copy(c.writeBuffer, b[n:])
c.modWrite()
}
return len(b), nil
}
// 如果writeBuffer中仍有未寫入的數(shù)據(jù)，則還將追加新數(shù)據(jù)。
c.writeBuffer = mempool.Append(c.writeBuffer, b...)
return len(b), nil
}當(dāng)數(shù)據(jù)未完全寫入時(shí)，剩余數(shù)據(jù)將被放入writeBuffer，觸發(fā)執(zhí)行modWrite，并將conn 的寫入事件注冊(cè)到 epoll。
總結(jié)
與 evio 相比，nbio 沒有蜂群效應(yīng)。
Evio 通過不斷喚醒無(wú)效的 epoll 來(lái)實(shí)現(xiàn)邏輯正確性。Nbio 盡量減少系統(tǒng)調(diào)用，減少不必要的開銷。
在可用性方面，nbio 實(shí)現(xiàn)了標(biāo)準(zhǔn)庫(kù)net.Conn，許多設(shè)置都是可配置的，允許用戶進(jìn)行高度靈活的定制。
預(yù)分配緩沖區(qū)用于讀寫操作，以提高應(yīng)用程序性能。
總之，nbio 是個(gè)不錯(cuò)的高性能無(wú)阻塞網(wǎng)絡(luò)框架。
參考資料：
[1]Analyzing High-Performance Network Framework nbio in Go:https://levelup.gitconnected.com/analyzing-high-performance-network-framework-nbio-in-go-9c35f295b5ad
