Go 中的 Epoll

關(guān)于計算機網(wǎng)絡(luò)分層與 TCP 通信過程過程此處不再贅述。

• 考慮到 TCP 通信過程中各種復(fù)雜操作，包括三次握手，四次揮手等，多數(shù)操作系統(tǒng)都提供了 Socket 作為 TCP 網(wǎng)絡(luò)連接的抽象。
• Linux -> Internet domain socket -> SOCK_STREAM
• Linux 中 Socket 以 “文件描述符” FD 作為標(biāo)識

在進行 Socket 通信時，服務(wù)端同時操作多個 Socket，此時便需要 IO 模型操作方案。：

• 阻塞 IO。傳統(tǒng) C/C++ 方案，同步讀寫 Socket（一個線程一個 Socket），線程陷入內(nèi)核態(tài)，當(dāng)讀寫成功后，切換回用戶態(tài)繼續(xù)執(zhí)行。
• 非阻塞 IO。應(yīng)用會不斷自旋輪詢，直到 Socket 可以讀寫，如果暫時無法收發(fā)數(shù)據(jù)，會返回錯誤。
• Epoll 多路復(fù)用。提供了事件列表，不需要查詢各個 Socket。其注冊多個 Socket 事件，調(diào)用 epoll ，當(dāng)有事件發(fā)生則返回。

Epoll 是 Linux 下的 event poll，Windows 中為 IOCP, Mac 中為 kqueue。

在 Go 中，內(nèi)部采用結(jié)合阻塞模型和多路復(fù)用的方法。在這里就不再是線程操作 Socket，而是 Goroutine 協(xié)程。每個協(xié)程關(guān)心一個 Socket 連接：

• 在底層使用操作系統(tǒng)的多路復(fù)用 IO。
• 在協(xié)程層次使用阻塞模型。
• 阻塞協(xié)程時，休眠協(xié)程。

我們知道，Go 是一個跨平臺的語言，不同平臺/操作系統(tǒng)下提供的 Epoll 實現(xiàn)不同，所以 Go 在 Epoll/IOCP/kqueue 上再獨立了一層 epoll 抽象層，用于屏蔽各個系統(tǒng)的差異性，抽象各系統(tǒng)對多路復(fù)用器的實現(xiàn)。

Go Network Poller 多路復(fù)用器的抽象，以 Linux 為例：

• Go Network Poller 對于多路復(fù)用器的抽象和適配
• epoll_create() -> netpollinit()
• epoll_ctl() -> netpollopen()
• epoll_wait() -> netpoll()

// Integrated network poller (platform-independent part).
// A particular implementation (epoll/kqueue/port/AIX/Windows)
// must define the following functions:
//
// func netpollinit()
//     Initialize the poller. Only called once.
//
// func netpollopen(fd uintptr, pd *pollDesc) int32
//     Arm edge-triggered notifications for fd. The pd argument is to pass
//     back to netpollready when fd is ready. Return an errno value.
//
// func netpollclose(fd uintptr) int32
//     Disable notifications for fd. Return an errno value.
//
// func netpoll(delta int64) gList
//     Poll the network. If delta < 0, block indefinitely. If delta == 0,
//     poll without blocking. If delta > 0, block for up to delta nanoseconds.
//     Return a list of goroutines built by calling netpollready.
//
// func netpollBreak()
//     Wake up the network poller, assumed to be blocked in netpoll.
//
// func netpollIsPollDescriptor(fd uintptr) bool
//     Reports whether fd is a file descriptor used by the poller.

上訴所有方法的實現(xiàn)都在 %GOROOT/src/runtime/netpoll_epoll.go%

netpollinit() 新建多路復(fù)用器

1. 新建 Epoll，不同系統(tǒng)對應(yīng)不同的實現(xiàn)方式。
2. 新建一個 Pipe 管道用于中斷 Epoll。
3. 將 “管道有數(shù)據(jù)到達” 事件注冊到 Epoll 中。

func netpollinit() {
    var errno uintptr

    // 1. 新建 Epoll，不同系統(tǒng)對應(yīng)不同的實現(xiàn)方式
    epfd, errno = syscall.EpollCreate1(syscall.EPOLL_CLOEXEC)
    if errno != 0 {
        println("runtime: epollcreate failed with", errno)
        throw("runtime: netpollinit failed")
    }

    // 用來中斷 Epoll 的管道
    r, w, errpipe := nonblockingPipe()
    if errpipe != 0 {
        println("runtime: pipe failed with", -errpipe)
        throw("runtime: pipe failed")
    }

    // 3. 注冊事件
    ev := syscall.EpollEvent{
        Events: syscall.EPOLLIN,
    }
    *(**uintptr)(unsafe.Pointer(&ev.Data)) = &netpollBreakRd
    errno = syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, r, &ev)
    if errno != 0 {
        println("runtime: epollctl failed with", errno)
        throw("runtime: epollctl failed")
    }
    netpollBreakRd = uintptr(r)
    netpollBreakWr = uintptr(w)
}

netpollopen() 插入事件

1. 傳入一個 Socket 的 FD 和 pollDesc 指針。
2. pollDesc 指針是 Socket 相關(guān)詳細信息。
3. pollDesc 指針中記錄了哪個協(xié)程在休眠等待此 Socket。
4. 將 Socket 的可讀/可寫/斷開事件注冊到 Epoll 中。

func netpollopen(fd uintptr, pd *pollDesc) uintptr {
    var ev syscall.EpollEvent

    // 事件類型
    ev.Events = syscall.EPOLLIN | syscall.EPOLLOUT | syscall.EPOLLRDHUP | syscall.EPOLLET
    *(**pollDesc)(unsafe.Pointer(&ev.Data)) = pd
    return syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, int32(fd), &ev)
}

netpoll() 查詢事件

1. 調(diào)用 EpollWait() 方法，查詢有哪些事件發(fā)生
2. 根據(jù) Socket 相關(guān)的 pollDesc 信息，返回哪些協(xié)程可以喚醒。

func netpoll(delay int64) gList {
    // 1. 查詢哪些事件發(fā)生
    n, errno := syscall.EpollWait(epfd, events[:], int32(len(events)), waitms)
    // ...
    if errno != 0 {
        if errno != _EINTR {
            println("runtime: epollwait on fd", epfd, "failed with", errno)
            throw("runtime: netpoll failed")
        }
        // If a timed sleep was interrupted, just return to
        // recalculate how long we should sleep now.
        if waitms > 0 {
            return gList{}
        }
        goto retry
    }
    // 2. 根據(jù) Socket 相關(guān)的 pollDesc 信息，返回哪些協(xié)程可以喚醒。
    var toRun gList
    for i := int32(0); i < n; i++ {
        ev := events[i]
        if ev.Events == 0 {
            continue
        }

        if *(**uintptr)(unsafe.Pointer(&ev.Data)) == &netpollBreakRd {
            if ev.Events != syscall.EPOLLIN {
                println("runtime: netpoll: break fd ready for", ev.Events)
                throw("runtime: netpoll: break fd ready for something unexpected")
            }
            if delay != 0 {
                var tmp [16]byte
                read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))
                netpollWakeSig.Store(0)
            }
            continue
        }

        var mode int32
        if ev.Events&(syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLHUP|syscall.EPOLLERR) != 0 {
            mode += 'r'
        }
        if ev.Events&(syscall.EPOLLOUT|syscall.EPOLLHUP|syscall.EPOLLERR) != 0 {
            mode += 'w'
        }
        if mode != 0 {
            pd := *(**pollDesc)(unsafe.Pointer(&ev.Data))
            pd.setEventErr(ev.Events == syscall.EPOLLERR)
            netpollready(&toRun, pd, mode)
        }
    }

    // 協(xié)程列表
    return toRun
}

Go Network Poller

Network Poller 初始化

• 初始化一個 Network Poller。
• 調(diào)用 netpollinit() 新建多路復(fù)用器。

// %GOROOT%src/runtime/netpoll.go
func poll_runtime_pollServerInit() {
    netpollGenericInit()
}

func netpollGenericInit() {
    // 每個 Go 應(yīng)用只初始化一次
    if netpollInited.Load() == 0 {
        lockInit(&netpollInitLock, lockRankNetpollInit)
        lock(&netpollInitLock)
        if netpollInited.Load() == 0 {
            // 新建多路復(fù)用器
            netpollinit()
            netpollInited.Store(1)
        }
        unlock(&netpollInitLock)
    }
}

關(guān)于 pollDesc，是 runtime 包對 Socket 的詳細描述：

type pollDesc struct {
    _    sys.NotInHeap
    link *pollDesc // in pollcache, protected by pollcache.lock
    fd   uintptr   // constant for pollDesc usage lifetime
    atomicInfo atomic.Uint32 // atomic pollInfo


    rg atomic.Uintptr // pdReady, pdWait, G waiting for read or pdNil
    wg atomic.Uintptr // pdReady, pdWait, G waiting for write or pdNil

    lock    mutex // protects the following fields
    closing bool
    user    uint32    // user settable cookie
    rseq    uintptr   // protects from stale read timers
    rt      timer     // read deadline timer (set if rt.f != nil)
    rd      int64     // read deadline (a nanotime in the future, -1 when expired)
    wseq    uintptr   // protects from stale write timers
    wt      timer     // write deadline timer
    wd      int64     // write deadline (a nanotime in the future, -1 when expired)
    self    *pollDesc // storage for indirect interface. See (*pollDesc).makeArg.
}

Network Poller 新增監(jiān)聽 Socket

• 在 pollCache 鏈表中分配一個 pollDesc。
• 初始化 pollDesc，rg，wg 都為 0。
• 調(diào)用 netpollopen() 插入事件

func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) {
    // 分配 pollDesc
    pd := pollcache.alloc()
    lock(&pd.lock)
    wg := pd.wg.Load()
    if wg != pdNil && wg != pdReady {
        throw("runtime: blocked write on free polldesc")
    }
    rg := pd.rg.Load()
    if rg != pdNil && rg != pdReady {
        throw("runtime: blocked read on free polldesc")
    }

    // 初始化 pollDesc
    pd.fd = fd
    pd.closing = false
    pd.setEventErr(false)
    pd.rseq++
    pd.rg.Store(pdNil)
    pd.rd = 0
    pd.wseq++
    pd.wg.Store(pdNil)
    pd.wd = 0
    pd.self = pd
    pd.publishInfo()
    unlock(&pd.lock)

    // 插入事件
    errno := netpollopen(fd, pd)
    if errno != 0 {
        pollcache.free(pd)
        return nil, int(errno)
    }
    return pd, 0
}

Network Poller 收發(fā)數(shù)據(jù)

• Socket 已經(jīng)可讀寫時
  • runtime 的 g0 協(xié)程循環(huán)調(diào)用 netpoll() 方法。
  • 發(fā)現(xiàn) Socket 可讀寫時，給對應(yīng)的 rg，wg 置為 pdReady(1)。
  • 協(xié)程調(diào)用 poll_runtime_pollWait()。
  • 判斷 rg 或 wg 已置為 pdReady(1)，返回 0。

func poll_runtime_pollWait(pd *pollDesc, mode int) int {
    // ...
    // 判斷 `rg` 或 `wg` 已置為 pdReady(1)，返回 0。
    for !netpollblock(pd, int32(mode), false) {
        errcode = netpollcheckerr(pd, int32(mode))
        if errcode != pollNoError {
            return errcode
        }
    }
    return pollNoError
}

• Socket 暫時無法讀寫時
  • runtime 的 g0 協(xié)程循環(huán)調(diào)用 netpoll() 方法。
  • 協(xié)程調(diào)用 poll_runtime_pollWait()。
  • 發(fā)現(xiàn)對應(yīng)的 rg 或 wg 為 0。
  • 給對應(yīng)的 rg 或 wg 置為協(xié)程地址。
  • 休眠等待。
  • 當(dāng)發(fā)現(xiàn) Socket可讀寫時，查看對應(yīng)的 rg 或 wg。
  • 若為協(xié)程地址，則返回該地址。
  • 調(diào)度器開始調(diào)度該協(xié)程。
    

Socket 通信

net 包中的 Socket 會被定義為一個 netFD 結(jié)構(gòu)體：

type netFD struct {
    // 最終指向的 runtime 中的 Socket 結(jié)構(gòu)體
    pfd poll.FD

    family      int
    sotype      int
    isConnected bool // handshake completed or use of association with peer
    net         string
    laddr       Addr
    raddr       Addr
}

Server 端

以 TCP 協(xié)議為例：

net 的 net.Listen() 操作：

1. 新建 Socket，并執(zhí)行 bind 操作
2. 新建一個 FD（net 包對 Socket 的詳情描述）。
3. 返回一個 TCPListener 對象
4. 將 TCPListener 的 FD 信息加入監(jiān)聽。

func main() {
    ln, err := net.Listen("tcp", ":8888")
    if err != nil {
        panic(err)
    }
}

TCPListener 本質(zhì)是一個 LISTEN 狀態(tài)的 Scoket。

TCPListener.Accept() 操作：

1. 直接調(diào)用 Socket 的 accept()。
2. 如果失敗，休眠等待新的連接。
3. 將新的 Socket 包裝成 TCPConn 變量返回。
4. 將 TCPConn 的 FD 信息加入監(jiān)聽。

func main() {
    ln, err := net.Listen("tcp", ":8888")
    if err != nil {
        panic(err)
    }

    conn, err := ln.Accept()
    if err != nil {
        panic(err)
    }
    defer conn.Close()
}

TCPConn 本質(zhì)是一個 ESTANBLISHED 狀態(tài)的 Scoket。

TCPConn 收發(fā)數(shù)據(jù)

func main() {
    // 1. 監(jiān)聽端口
    ln, err1 := net.Listen("tcp", ":8888")
    if err1 != nil {
        panic(err1)
    }

    // 2. 建立連接
    conn, err2 := ln.Accept()
    if err2 != nil {
        panic(err2)
    }
    defer conn.Close()

    var recv [1024]byte
    // 使用 bufio 標(biāo)準(zhǔn)庫提供的緩沖區(qū)功能
 send := bufio.NewReader(conn)

    for {

        // 3. 獲取數(shù)據(jù)
        _, err3 := conn.Read(recv[:])
        if err3 != nil {
            break
        }

        fmt.Printf("n: %v\n", string(recv[:]))

        // 4. 發(fā)送數(shù)據(jù)
        msg, err := send.ReadString('\n')
        if strings.ToUpper(msg) == "Q" {
            return
        }
        if err != nil {
            return
        }
        _, err4 := conn.Write([]byte(msg))
        if err4 != nil {
            break
        }
    }
}

Client 端

func main() {
    // 與服務(wù)端建立連接
    conn, err := net.Dial("tcp", ":8888")
    if err != nil {
        panic(err)
    }
    
    var recv [1024]byte

    send := bufio.NewReader(os.Stdin)
    for {
        s, _ := send.ReadString('\n')
        if strings.ToUpper(s) == "Q" {
            return  
        }

        // 發(fā)送數(shù)據(jù)
        _, err = conn.Write([]byte(s))
        if err != nil {
            panic(err)
        }

        // 接收數(shù)據(jù) 
        _, err := conn.Read(recv[:])
        if err != nil {
            break
        }
        fmt.Printf(":%v\n", string(recv[:]))
    }
}

goroutine-per-connection style code

一個協(xié)程服務(wù)一個新的連接文章來源地址http://www.zghlxwxcb.cn/news/detail-515679.html

package main

import (
    "bufio"
    "fmt"
    "net"
    "strings"
)

func handleConnection(conn net.Conn) {
    defer conn.Close()

    var recv [1024]byte
    // 使用 bufio 標(biāo)準(zhǔn)庫提供的緩沖區(qū)功能
    send := bufio.NewReader(conn)

    for {

        // 3. 獲取數(shù)據(jù)
        _, err3 := conn.Read(recv[:])
        if err3 != nil {
            break
        }

        fmt.Printf("n: %v\n", string(recv[:]))

        // 4. 發(fā)送數(shù)據(jù)
        msg, err := send.ReadString('\n')
        if strings.ToUpper(msg) == "Q" {
            return
        }
        if err != nil {
            return
        }
        _, err4 := conn.Write([]byte(msg))
        if err4 != nil {
            break
        }
    }
}

func main() {
    // 1. 監(jiān)聽端口
    ln, err1 := net.Listen("tcp", ":8888")
    if err1 != nil {
        panic(err1)
    }

    for {
        // 2. 建立連接
        conn, err2 := ln.Accept()
        if err2 != nil {
            panic(err2)
        }

        go handleConnection(conn)
    }
}

到了這里，關(guān)于深入理解 Golang: 網(wǎng)絡(luò)編程的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！