詳解Go語言I/O多路複用netpoller模型
阿新 • • 發佈：2021-02-09
> 轉載請宣告出處哦~，本篇文章釋出於luozhiyun的部落格：https://www.luozhiyun.com
>
> 本文使用的go的原始碼15.7
可以從 Go 原始碼目錄結構和對應程式碼檔案瞭解 Go 在不同平臺下的網路 I/O 模式的實現。比如，在 Linux 系統下基於 epoll，freeBSD 系統下基於 kqueue，以及 Windows 系統下基於 iocp。
因為我們的程式碼都是部署在Linux上的，所以本文以epoll封裝實現為例子來講解Go語言中I/O多路複用的原始碼實現。
## 介紹
### I/O多路複用
所謂 I/O 多路複用指的就是 select/epoll 這一系列的多路選擇器：支援單一執行緒同時監聽多個檔案描述符（I/O 事件），阻塞等待，並在其中某個檔案描述符可讀寫時收到通知。以防很多同學對select或epoll不那麼熟悉，所以下面先來講講這兩個選擇器。
首先我們先說一下什麼是檔案描述符（File descriptor），根據它的英文首字母也簡稱FD，它是一個用於表述指向檔案的引用的抽象化概念。它是一個索引值，指向核心為每一個程序所維護的該程序開啟檔案的記錄表。當程式開啟一個現有檔案或者建立一個新檔案時，核心向程序返回一個檔案描述符。
#### select
```go
int select(int nfds,
fd_set *restrict readfds,
fd_set *restrict writefds,
fd_set *restrict errorfds,
struct timeval *restrict timeout);
```
writefds、readfds、和exceptfds是三個檔案描述符集合。select會遍歷每個集合的前nfds個描述符，分別找到可以讀取、可以寫入、發生錯誤的描述符，統稱為就緒的描述符。
timeout引數表示呼叫select時的阻塞時長。如果所有檔案描述符都未就緒，就阻塞呼叫程序，直到某個描述符就緒，或者阻塞超過設定的 timeout 後，返回。如果timeout引數設為 NULL，會無限阻塞直到某個描述符就緒；如果timeout引數設為 0，會立即返回，不阻塞。
當select函式返回後，可以通過遍歷fdset，來找到就緒的描述符。
select的缺點也列舉一下：
1. select最大的缺陷就是單個程序所開啟的FD是有一定限制的，它由FD_SETSIZE設定，預設值是1024;
2. 每次呼叫 select，都需要把 fd 集合從使用者態拷貝到核心態，這個開銷在 fd 很多時會很大;
3. 每次 kernel 都需要線性掃描整個 fd_set，所以隨著監控的描述符 fd 數量增長，其 I/O 效能會線性下降;
#### epoll
epoll是selec的增強版本，避免了“效能開銷大”和“檔案描述符數量少”兩個缺點。
為方便理解後續的內容，先看一下epoll的用法：
```
int listenfd = socket(AF_INET, SOCK_STREAM, 0);
bind(listenfd, ...)
listen(listenfd, ...)
int epfd = epoll_create(...);
epoll_ctl(epfd, ...); //將所有需要監聽的fd新增到epfd中
while(1){
int n = epoll_wait(...)
for(接收到資料的socket){
//處理
}
}
```
先用epoll_create建立一個epoll物件例項epfd，同時返回一個引用該例項的檔案描述符，返回的檔案描述符僅僅指向對應的epoll例項，並不表示真實的磁碟檔案節點。
epoll例項內部儲存：
- 監聽列表：所有要監聽的檔案描述符，使用紅黑樹；
- 就緒列表：所有就緒的檔案描述符，使用連結串列；
再通過epoll_ctl將需要監視的fd新增到epfd中，同時為fd設定一個回撥函式，並監聽事件event，並新增到監聽列表中。當有事件發生時，會呼叫回撥函式，並將fd新增到epoll例項的就緒佇列上。
最後呼叫epoll_wait阻塞監聽 epoll 例項上所有的fd的 I/O 事件。當就緒列表中已有資料，那麼epoll_wait直接返回，解決了select每次都需要輪詢一遍的問題。
epoll的優點：
epoll的監聽列表使用紅黑樹儲存，epoll_ctl 函式新增進來的 fd 都會被放在紅黑樹的某個節點內，而紅黑樹本身插入和刪除效能比較穩定，時間複雜度 O(logN)，並且可以儲存大量的的fd，避免了只能儲存1024個fd的限制；
epoll_ctl 中為每個檔案描述符指定了回撥函式，並在就緒時將其加入到就緒列表，因此不需要像select一樣遍歷檢測每個檔案描述符，只需要判斷就緒列表是否為空即可；
## 解析
netpoll本質上是對 I/O 多路複用技術的封裝，所以自然也是和epoll一樣脫離不了下面幾步：
1. netpoll建立及其初始化；
2. 向netpoll中加入待監控的任務；
3. 從netpoll獲取觸發的事件；
在go中對epoll提供的三個函式進行了封裝：
```
func netpollinit()
func netpollopen(fd uintptr, pd *pollDesc) int32
func netpoll(delay int64) gList
```
netpollinit函式負責初始化netpoll；
netpollopen負責監聽檔案描述符上的事件；
netpoll會阻塞等待返回一組已經準備就緒的 Goroutine；
下面是Go語言中編寫的一個TCP server：
```go
func main() {
listen, err := net.Listen("tcp", ":8888")
if err != nil {
fmt.Println("listen error: ", err)
return
}
for {
conn, err := listen.Accept()
if err != nil {
fmt.Println("accept error: ", err)
break
}
// 建立一個goroutine來負責處理讀寫任務
go HandleConn(conn)
}
}
```
下面我們跟著這個TCP server的原始碼一起看看是在哪裡使用了netpoll來完成epoll的呼叫。
### net.Listen
這個TCP server中會呼叫`net.Listen`建立一個socket同時返回與之對應的fd，該fd用來初始化listener的netFD(go層面封裝的網路檔案描述符)，接著呼叫 netFD的listenStream方法完成對 socket 的 bind&listen和netFD的初始化。
呼叫過程如下：
```go
func socket(ctx context.Context, net string, family, sotype, proto int, ipv6only bool, laddr, raddr sockaddr, ctrlFn func(string, string, syscall.RawConn) error) (fd *netFD, err error) {
// 建立一個socket
s, err := sysSocket(family, sotype, proto)
if err != nil {
return nil, err
}
...
// 建立fd
if fd, err = newFD(s, family, sotype, net); err != nil {
poll.CloseFunc(s)
return nil, err
}
if laddr != nil && raddr == nil {
switch sotype {
case syscall.SOCK_STREAM, syscall.SOCK_SEQPACKET:
// 呼叫 netFD的listenStream方法完成對 socket 的 bind&listen和netFD的初始化
if err := fd.listenStream(laddr, listenerBacklog(), ctrlFn); err != nil {
fd.Close()
return nil, err
}
return fd, nil
case syscall.SOCK_DGRAM:
...
}
}
...
return fd, nil
}
func newFD(sysfd syscall.Handle, family, sotype int, net string) (*netFD, error) {
ret := &netFD{
pfd: poll.FD{
Sysfd: sysfd,
IsStream: sotype == syscall.SOCK_STREAM,
ZeroReadIsEOF: sotype != syscall.SOCK_DGRAM && sotype != syscall.SOCK_RAW,
},
family: family,
sotype: sotype,
net: net,
}
return ret, nil
}
```
sysSocket方法會發起一個系統呼叫建立一個socket，newFD會建立一個netFD，然後呼叫netFD的listenStream方法進行bind&listen操作，並對netFD進行init。
netFD是一個檔案描述符的封裝，netFD中包含一個FD資料結構，FD中包含了Sysfd 和pollDesc兩個重要的資料結構，Sysfd是sysSocket返回的socket系統檔案描述符，pollDesc用於監控檔案描述符的可讀或者可寫。
我們繼續看listenStream：
```go
func (fd *netFD) listenStream(laddr sockaddr, backlog int, ctrlFn func(string, string, syscall.RawConn) error) error {
...
// 完成繫結操作
if err = syscall.Bind(fd.pfd.Sysfd, lsa); err != nil {
return os.NewSyscallError("bind", err)
}
// 進行監聽操作
if err = listenFunc(fd.pfd.Sysfd, backlog); err != nil {
return os.NewSyscallError("listen", err)
}
// 初始化fd
if err = fd.init(); err != nil {
return err
}
lsa, _ = syscall.Getsockname(fd.pfd.Sysfd)
fd.setAddr(fd.addrFunc()(lsa), nil)
return nil
}
```
listenStream方法會呼叫Bind方法完成fd的繫結操作，然後呼叫listenFunc進行監聽，接著呼叫fd的init方法，完成FD、pollDesc初始化。
```go
func (pd *pollDesc) init(fd *FD) error {
// 呼叫到runtime.poll_runtime_pollServerInit
serverInit.Do(runtime_pollServerInit)
// 呼叫到runtime.poll_runtime_pollOpen
ctx, errno := runtime_pollOpen(uintptr(fd.Sysfd))
...
return nil
}
```
runtime_pollServerInit用Once封裝保證只能被呼叫一次，這個函式在Linux平臺上會建立一個epoll檔案描述符例項；
poll_runtime_pollOpen呼叫了netpollopen會將fd註冊到 epoll例項中，並返回一個pollDesc；
#### netpollinit初始化
```go
func poll_runtime_pollServerInit() {
netpollGenericInit()
}
func netpollGenericInit() {
if atomic.Load(&netpollInited) == 0 {
lock(&netpollInitLock)
if netpollInited == 0 {
netpollinit()
atomic.Store(&netpollInited, 1)
}
unlock(&netpollInitLock)
}
}
```
netpollGenericInit會呼叫平臺上特定實現的netpollinit，在Linux中會呼叫到netpoll_epoll.go的netpollinit方法：
```go
var (
epfd int32 = -1 // epoll descriptor
)
func netpollinit() {
// 建立一個新的 epoll 檔案描述符
epfd = epollcreate1(_EPOLL_CLOEXEC)
...
// 建立一個用於通訊的管道
r, w, errno := nonblockingPipe()
...
ev := epollevent{
events: _EPOLLIN,
}
*(**uintptr)(unsafe.Pointer(&ev.data)) = &netpollBreakRd
// 將讀取資料的檔案描述符加入監聽
errno = epollctl(epfd, _EPOLL_CTL_ADD, r, &ev)
...
netpollBreakRd = uintptr(r)
netpollBreakWr = uintptr(w)
}
```
呼叫epollcreate1方法會建立一個epoll檔案描述符例項，需要注意的是epfd是一個全域性的屬性。然後建立一個用於通訊的管道，呼叫epollctl將讀取資料的檔案描述符加入監聽。
#### netpollopen加入事件監聽
下面再看看poll_runtime_pollOpen方法：
```go
func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) {
pd := pollcache.alloc()
lock(&pd.lock)
if pd.wg != 0 && pd.wg != pdReady {
throw("runtime: blocked write on free polldesc")
}
if pd.rg != 0 && pd.rg != pdReady {
throw("runtime: blocked read on free polldesc")
}
pd.fd = fd
pd.closing = false
pd.everr = false
pd.rseq++
pd.rg = 0
pd.rd = 0
pd.wseq++
pd.wg = 0
pd.wd = 0
pd.self = pd
unlock(&pd.lock)
var errno int32
errno = netpollopen(fd, pd)
return pd, int(errno)
}
func netpollopen(fd uintptr, pd *pollDesc) int32 {
var ev epollevent
ev.events = _EPOLLIN | _EPOLLOUT | _EPOLLRDHUP | _EPOLLET
*(**pollDesc)(unsafe.Pointer(&ev.data)) = pd
return -epollctl(epfd, _EPOLL_CTL_ADD, int32(fd), &ev)
}
```
poll_runtime_pollOpen方法會通過`pollcache.alloc`初始化總大小約為 4KB的pollDesc結構體。然後重置pd的屬性，呼叫netpollopen向epoll例項epfd加入新的輪詢事件監聽檔案描述符的可讀和可寫狀態。
下面我們再看看pollCache是如何初始化pollDesc的。
```go
type pollCache struct {
lock mutex
first *pollDesc
}
const pollBlockSize = 4 * 1024
func (c *pollCache) alloc() *pollDesc {
lock(&c.lock)
// 初始化首節點
if c.first == nil {
const pdSize = unsafe.Sizeof(pollDesc{})
n := pollBlockSize / pdSize
if n == 0 {
n = 1
}
mem := persistentalloc(n*pdSize, 0, &memstats.other_sys)
// 初始化pollDesc連結串列
for i := uintptr(0); i < n; i++ {
pd := (*pollDesc)(add(mem, i*pdSize))
pd.link = c.first
c.first = pd
}
}
pd := c.first
c.first = pd.link
lockInit(&pd.lock, lockRankPollDesc)
unlock(&c.lock)
return pd
}
```
pollCache的連結串列頭如果為空，那麼初始化首節點，首節點是一個pollDesc的連結串列頭，每次呼叫該結構體都會返回連結串列頭還沒有被使用的pollDesc。
到這裡就完成了net.Listen的分析，下面我們看看listen.Accept。
### Listener.Accept
Listener.Accept方法最終會呼叫到netFD的accept方法中：
```go
func (fd *netFD) accept() (netfd *netFD, err error) {
// 呼叫netfd.FD的Accept接受新的 socket 連線，返回 socket 的 fd
d, rsa, errcall, err := fd.pfd.Accept()
...
// 構造一個新的netfd
if netfd, err = newFD(d, fd.family, fd.sotype, fd.net); err != nil {
poll.CloseFunc(d)
return nil, err
}
// 呼叫 netFD 的 init 方法完成初始化
if err = netfd.init(); err != nil {
netfd.Close()
return nil, err
}
lsa, _ := syscall.Getsockname(netfd.pfd.Sysfd)
netfd.setAddr(netfd.addrFunc()(lsa), netfd.addrFunc()(rsa))
return netfd, nil
}
```
這個方法首先會呼叫到FD的Accept接受新的 socket 連線，並返回新的socket對應的fd，然後呼叫newFD構造一個新的netfd，並通過init 方法完成初始化。
init方法上面我們已經看過了，下面我們來看看Accept方法：
```go
func (fd *FD) Accept() (int, syscall.Sockaddr, string, error) {
...
for {
// 使用 linux 系統呼叫 accept 接收新連線，建立對應的 socket
s, rsa, errcall, err := accept(fd.Sysfd)
if err == nil {
return s, rsa, "", err
}
switch err {
case syscall.EINTR:
continue
case syscall.EAGAIN:
if fd.pd.pollable() {
// 如果當前沒有發生期待的 I/O 事件，那麼 waitRead 會通過 park goroutine 讓邏輯 block 在這裡
if err = fd.pd.waitRead(fd.isFile); err == nil {
continue
}
}
case syscall.ECONNABORTED:
continue
}
return -1, nil, errcall, err
}
}
```
`FD.Accept`方法會使用 linux 系統呼叫 accept 接收新連線，建立對應的 socket，如果沒有可讀的訊息，waitRead會被阻塞。這些被park住的goroutine會在goroutine的排程中呼叫`runtime.netpoll`被喚醒。
#### pollWait事件等待
`pollDesc.waitRead`實際上是呼叫了`runtime.poll_runtime_pollWait`
```go
func poll_runtime_pollWait(pd *pollDesc, mode int) int {
...
// 進入 netpollblock 並且判斷是否有期待的 I/O 事件發生
for !netpollblock(pd, int32(mode), false) {
...
}
return 0
}
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
gpp := &pd.rg
if mode == 'w' {
gpp = &pd.wg
}
// 這個 for 迴圈是為了等待 io ready 或者 io wait
for {
old := *gpp
// gpp == pdReady 表示此時已有期待的 I/O 事件發生，
// 可以直接返回 unblock 當前 goroutine 並執行響應的 I/O 操作
if old == pdReady {
*gpp = 0
return true
}
if old != 0 {
throw("runtime: double wait")
}
// 如果沒有期待的 I/O 事件發生，則通過原子操作把 gpp 的值置為 pdWait 並退出 for 迴圈
if atomic.Casuintptr(gpp, 0, pdWait) {
break
}
}
if waitio || netpollcheckerr(pd, mode) == 0 {
// 讓出當前執行緒，將 Goroutine 轉換到休眠狀態並等待執行時的喚醒
gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5)
}
// be careful to not lose concurrent pdReady notification
old := atomic.Xchguintptr(gpp, 0)
if old >