目录

• 一、声明
• 二、epoll基础
• 三、net包到底层类图
• 四、连接的建立
• 五、异步的数据读写
• 六、调度有网络消息的G
• 七、带超时时间的读写
• 八、总结

声明

下面的分析均基于Golang1.14版本。
以下只分析Tcp相关的源码，其它的类似。
网络底层只分析Linux下epoll的实现。

epoll基础

epoll的相关接口

int epoll_create(int size); // 初始化epoll句柄 // 套接字的事件监听注册。含新增，修改，删除操作。可以监听读写事件（一般只监听读事件）int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);// 收集注册在epoll句柄中的已触发的事件int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

在Golang中通过汇编代码调用对应的系统调用来调用epoll的接口。以下是对应关系：

epollcreate ==> epoll_createepollctl ==> epoll_ctlepollwait ==> epoll_wait

net包到底层类图

net包到底层的类图

1.在net包的最上层要提供接口建立网络连接和数据读取。提供net.Listen获取监听socket，提供Connect，Accept和获取连接socket。
2.连接的建立最终需要调用Linux的系统的socket, bind, listen, accept, connect, send, recv等函数。这些系统调用在netFD这一层被调用。
3.连接的管理最终需要调用Linux系统的epoll_create，epoll_ctl，epoll_wait等函数。这些系统调用在runtime.pollDesc这一层被调用。

连接的建立

epoll实现TCP Client + Server

1.在Linux下用epoll实现网络连接比较简单，流程图如上。

Go调用Linux系统调用实现TCP Client + Server

2.在上面我们已经知道Go在Linux下实现网络连接最终是调用Linux下的系统调用。分析Go源码时，可以尝试去对应C实现网络连接的流程。
3.其中scoket函数里调用了Linux相关的 socket，bind，listen，connect等系统调用。
4.poll_runtime_pollOpen调用了Linux相关的epoll_ctl系统调用。
5.以上的流程图侧重于类比Go和C实现网络连接的流程，所以有部分分支未深入，如socket中调用系统调用初始化socket的部分。
PS：accept也是连接建立的一部分，但accept本质也是对监听套接字的数据读写，因此accept部分在一部数据读写部分进行分析。

epoll_create的调用

6.epoll句柄在一个进程中最多只需要一个。当调用pollDesc.init时，说明该进程有网络相关的操作，于是通过sync.Once调用runtime_pollServerinit进行初始化，该函数最终会调用epoll_create创建epoll句柄。
7.使用epoll_ctl注册监听事件。

func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) {    pd := pollcache.alloc() // 分配pollDesc    lock(&pd.lock) // 初始化 pollDesc    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    unlock(&pd.lock)    var errno int32    errno = netpollopen(fd, pd) // epoll_ctl 注册监听事件    return pd, int(errno)}func netpollopen(fd uintptr, pd *pollDesc) int32 {    var ev epollevent    ev.events = _EPOLLIN | _EPOLLOUT | _EPOLLRDHUP | _EPOLLET // 4种类型的事件都需要注册    *(**pollDesc)(unsafe.Pointer(&ev.data)) = pd    return -epollctl(epfd, _EPOLL_CTL_ADD, int32(fd), &ev) // 调用epoll_ctl}

异步的数据读写

异步读写数据

1.简单分析accept，read，write的调用场景。accept是监听套接字等待客户端连接，因此同步必然会导致阻塞。read是读取对方socket发过来的数据，因此也必须是异步读取，否则阻塞。write是往缓冲区写数据，如果缓冲区满则返回写入的数量，因此最好是异步写，否则缓冲区满的情况下也会阻塞。
2.调用accept，read，write这些函数最终都会调用runtime的poll_runtime_pollWait函数，源码如下。

func poll_runtime_pollWait(pd *pollDesc, mode int) int {    err := netpollcheckerr(pd, int32(mode))    if err != 0 {        return err    }    // As for now only Solaris, illumos, and AIX use level-triggered IO.    if GOOS == "solaris" || GOOS == "illumos" || GOOS == "aix" {        netpollarm(pd, mode)    }    for !netpollblock(pd, int32(mode), false) { // 最终调用netpollblock阻塞        err = netpollcheckerr(pd, int32(mode))        if err != 0 {            return err        }        // Can happen if timeout has fired and unblocked us,        // but before we had a chance to run, timeout has been reset.        // Pretend it has not happened and retry.    }    return 0}func netpollcheckerr(pd *pollDesc, mode int32) int {    if pd.closing {        return 1 // ErrFileClosing or ErrNetClosing 如果套接字关闭或即将关闭    }    if (mode == 'r' && pd.rd < 0) || (mode == 'w' && pd.wd < 0) {        return 2 // ErrTimeout 如果套接字已经超时（定了事件的超时时间且过了超时时间）    }    // Report an event scanning error only on a read event.    // An error on a write event will be captured in a subsequent    // write call that is able to report a more specific error.    if mode == 'r' && pd.everr {        return 3 // ErrNotPollable epoll_wati发生_EPOLLERR错误    }    return 0}func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {    gpp := &pd.rg    if mode == 'w' {        gpp = &pd.wg    }    // set the gpp semaphore to WAIT    for {        old := *gpp        if old == pdReady {            *gpp = 0            return true        }        if old != 0 {            throw("runtime: double wait")        }        if atomic.Casuintptr(gpp, 0, pdWait) {            break        }    }    // need to recheck error states after setting gpp to WAIT    // this is necessary because runtime_pollUnblock/runtime_pollSetDeadline/deadlineimpl    // do the opposite: store to closing/rd/wd, membarrier, load of rg/wg    if waitio || netpollcheckerr(pd, mode) == 0 { // 如果可以进入休眠 调用gopark G进入waitting状态        gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5)    }    // be careful to not lose concurrent READY notification    old := atomic.Xchguintptr(gpp, 0)    if old > pdWait {        throw("runtime: corrupted polldesc")    }    return old == pdReady}

3.由上面的源码可知，套接字的读写最终会调用gopark使得协程进入waitting状态。因此需要在网络事件完成时，调用ready唤醒协程，让网络协程继续运行读写socket。

4. SetDeadline使用场景。如果不设置超时时间，那么G会一直阻塞在读写中，直到注册的事件发生。如果业务层要处理长时间没有读写请求的G，不设置超时时间是无法实现的。

调度有网络消息的G

1.对比C用epoll写的TCP客户端或服务器，需要轮询调用epoll_wait，处理触发了读写事件的socket。同理Go这边也需要有对应的代码轮询调用epoll_wait，处理socket的读写事件。
2.轮询调用epoll_wait的函数是netpoll函数，在sysmon中和findrunnable中都有调用该函数（忽略了GC中的调用）。

        lastpoll := int64(atomic.Load64(&sched.lastpoll))        if netpollinited() && lastpoll != 0 && lastpoll+10*1000*1000 < now {            atomic.Cas64(&sched.lastpoll, uint64(lastpoll), uint64(now))            list := netpoll(0) // non-blocking - returns list of goroutines            if !list.empty() { // 触发读写事件的G的List                incidlelocked(-1)                injectglist(&list)                incidlelocked(1)            }        }    if netpollinited() && atomic.Load(&netpollWaiters) > 0 && atomic.Load64(&sched.lastpoll) != 0 {        if list := netpoll(0); !list.empty() { // non-blocking            gp := list.pop() // 取触发读写事件的G的List中的首个元素继续运行            injectglist(&list)            casgstatus(gp, _Gwaiting, _Grunnable)            if trace.enabled {                traceGoUnpark(gp, 0)            }            return gp, false        }    }

3.netpoll主要是调用epoll_wait，将已经触发了读写事件的G取出并串联在gList中。

func netpoll(delay int64) gList {    if epfd == -1 {        return gList{}    }    var waitms int32    if delay < 0 {        waitms = -1    } else if delay == 0 {        waitms = 0    } else if delay < 1e6 {        waitms = 1    } else if delay < 1e15 {        waitms = int32(delay / 1e6)    } else {        // An arbitrary cap on how long to wait for a timer.        // 1e9 ms == ~11.5 days.        waitms = 1e9    }    var events [128]epolleventretry:    n := epollwait(epfd, &events[0], int32(len(events)), waitms)    if n < 0 { // 出现错误        if n != -_EINTR {            println("runtime: epollwait on fd", epfd, "failed with", -n)            throw("runtime: netpoll failed")        }        // If a timed sleep was interrupted, just return to        // recalculate how long we should sleep now.        if waitms > 0 { // 如果设定了大于0的等待时间 不阻塞 直接退出            return gList{}        }        goto retry // 如果超时时间小于等于0 则再次尝试    }    var toRun gList    for i := int32(0); i < n; i++ { // 取出触发读写事件socket绑定的G        ev := &events[i]        if ev.events == 0 {            continue        }        if *(**uintptr)(unsafe.Pointer(&ev.data)) == &netpollBreakRd {            if ev.events != _EPOLLIN {                println("runtime: netpoll: break fd ready for", ev.events)                throw("runtime: netpoll: break fd ready for something unexpected")            }            if delay != 0 {                // netpollBreak could be picked up by a                // nonblocking poll. Only read the byte                // if blocking.                var tmp [16]byte                read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))            }            continue        }        var mode int32        if ev.events&(_EPOLLIN|_EPOLLRDHUP|_EPOLLHUP|_EPOLLERR) != 0 {            mode += 'r'        }        if ev.events&(_EPOLLOUT|_EPOLLHUP|_EPOLLERR) != 0 {            mode += 'w'        }        if mode != 0 {            pd := *(**pollDesc)(unsafe.Pointer(&ev.data))            pd.everr = false            if ev.events == _EPOLLERR {                pd.everr = true            }            netpollready(&toRun, pd, mode) // 将触发读写事件的socket绑定的G放入toRun中        }    }    return toRun}func netpollready(toRun *gList, pd *pollDesc, mode int32) {    var rg, wg *g    if mode == 'r' || mode == 'r'+'w' {        rg = netpollunblock(pd, 'r', true)    }    if mode == 'w' || mode == 'r'+'w' {        wg = netpollunblock(pd, 'w', true)    }    if rg != nil {        toRun.push(rg)    }    if wg != nil {        toRun.push(wg)    }}func netpollunblock(pd *pollDesc, mode int32, ioready bool) *g {    gpp := &pd.rg    if mode == 'w' {        gpp = &pd.wg    }    for {        old := *gpp        if old == pdReady {            return nil        }        if old == 0 && !ioready {            // Only set READY for ioready. runtime_pollWait            // will check for timeout/cancel before waiting.            return nil        }        var new uintptr        if ioready {            new = pdReady        }        if atomic.Casuintptr(gpp, old, new) {            if old == pdReady || old == pdWait {                old = 0            }            return (*g)(unsafe.Pointer(old))        }    }}

4. 上面的源码都比较简单，调用epoll_wait找到触发读写事件的socket和绑定的pollDesc，并根据pollDesc找到对应的G。将G放入gList中，再调度gList中的G，调用ready使G从waitting进入runnable状态。

带超时时间的读写

1. 设置超时时间，主要是增加一个timer，设置timer的到期时间和回调函数。

func poll_runtime_pollSetDeadline(pd *pollDesc, d int64, mode int) {    lock(&pd.lock)    if pd.closing {        unlock(&pd.lock)        return    }    rd0, wd0 := pd.rd, pd.wd    combo0 := rd0 > 0 && rd0 == wd0    if d > 0 {        d += nanotime() // d为到期时间        if d <= 0 {            // If the user has a deadline in the future, but the delay calculation            // overflows, then set the deadline to the maximum possible value.            d = 1<<63 - 1        }    }    if mode == 'r' || mode == 'r'+'w' {        pd.rd = d    }    if mode == 'w' || mode == 'r'+'w' {        pd.wd = d    }    combo := pd.rd > 0 && pd.rd == pd.wd    rtf := netpollReadDeadline // 绑定到期后的处理函数     if combo {        rtf = netpollDeadline    }    if pd.rt.f == nil {        if pd.rd > 0 {            pd.rt.f = rtf            // Copy current seq into the timer arg.            // Timer func will check the seq against current descriptor seq,            // if they differ the descriptor was reused or timers were reset.            pd.rt.arg = pd            pd.rt.seq = pd.rseq            resettimer(&pd.rt, pd.rd) // 新增read timer        }    } else if pd.rd != rd0 || combo != combo0 {         pd.rseq++ // invalidate current timers        if pd.rd > 0 { // 修改原有的read timer            modtimer(&pd.rt, pd.rd, 0, rtf, pd, pd.rseq)        } else { // 删除原有的timer            deltimer(&pd.rt)            pd.rt.f = nil        }    }    if pd.wt.f == nil {        if pd.wd > 0 && !combo {            pd.wt.f = netpollWriteDeadline // 绑定到期后的处理函数             pd.wt.arg = pd            pd.wt.seq = pd.wseq            resettimer(&pd.wt, pd.wd) // 新增timer        }    } else if pd.wd != wd0 || combo != combo0 {        pd.wseq++ // invalidate current timers        if pd.wd > 0 && !combo { // 修改原有的timer            modtimer(&pd.wt, pd.wd, 0, netpollWriteDeadline, pd, pd.wseq)        } else { // 删除原有的timer             deltimer(&pd.wt)            pd.wt.f = nil        }    }    // If we set the new deadline in the past, unblock currently pending IO if any.    var rg, wg *g    if pd.rd < 0 || pd.wd < 0 { // 如果超时时间 < 0  则尝试将对应的G取出并设置为runnable        atomic.StorepNoWB(noescape(unsafe.Pointer(&wg)), nil) // full memory barrier between stores to rd/wd and load of rg/wg in netpollunblock        if pd.rd < 0 {            rg = netpollunblock(pd, 'r', false)        }        if pd.wd < 0 {            wg = netpollunblock(pd, 'w', false)        }    }    unlock(&pd.lock)    if rg != nil {        netpollgoready(rg, 3)    }    if wg != nil {        netpollgoready(wg, 3)    }}

2.读写timer到期后的回调处理函数，设置读写的超时时间并且唤醒协程，源码如下。

func netpolldeadlineimpl(pd *pollDesc, seq uintptr, read, write bool) {    lock(&pd.lock)    // Seq arg is seq when the timer was set.    // If it's stale, ignore the timer event.    currentSeq := pd.rseq    if !read {        currentSeq = pd.wseq    }    if seq != currentSeq { // 如果是先触发了读写事件 再触发超时 则序列号不相等. 此时需要唤醒G        // The descriptor was reused or timers were reset.        unlock(&pd.lock)        return    }    var rg *g    if read {        if pd.rd <= 0 || pd.rt.f == nil {            throw("runtime: inconsistent read deadline")        }        pd.rd = -1 // 设置超时时间 和 回调处理函数        atomic.StorepNoWB(unsafe.Pointer(&pd.rt.f), nil) // full memory barrier between store to rd and load of rg in netpollunblock        rg = netpollunblock(pd, 'r', false)    }    var wg *g    if write {        if pd.wd <= 0 || pd.wt.f == nil && !read {            throw("runtime: inconsistent write deadline")        }        pd.wd = -1 // 设置超时时间 和 回调处理函数        atomic.StorepNoWB(unsafe.Pointer(&pd.wt.f), nil) // full memory barrier between store to wd and load of wg in netpollunblock        wg = netpollunblock(pd, 'w', false)    }    unlock(&pd.lock)    if rg != nil {        netpollgoready(rg, 0) // 唤醒read协程    }    if wg != nil {        netpollgoready(wg, 0) // 唤醒write协程    }}

3. rg/wg值的互相转换。rg/wg初始值为0，调用gopark前，rg/wg的值为pdWait，调用gopark时，将rg/wg值设置为G。唤醒前，如果是通过epoll_wait中注册的事件唤醒，在netpollunblock中被设置为pdReady。如果是在timer超时中被唤醒，则在netpollunblock中设置为0。唤醒后，pdWait设置为0。

   

   rg/wg值的状态转换

4. SetDeadline, SetReadDeadline, SetWriteDeadline这3个函数会修改pollDesc中的rd，wd (read deadline, write deadline)，rd/wd函数的值的设置和含义如下。

总结

1.Go中Linux下epoll实现的TCP连接和C实现的epoll的模型是一样的，相当于用Go的语法和特性为epoll模型做了一层封装。
2.学习的本质是将陌生的东西转化为熟悉的东西，把陌生的模型和熟悉的模型进行类比，能更快的学会陌生的知识。
3.Go提供的TCP连接读写接口是同步阻塞的，同步表示调用结束后，要么读写成功，要么超时，阻塞表示该接口返回的时间是不确定的，上层业务需要封装读写。

给这篇文章的作者打赏

微信扫一扫打赏

支付宝扫一扫打赏

×

打赏 点赞(0)