Go语言源码笔记 --- netpoller

func netpollinit() // 初始化轮询器
func netpollopen(fd uintptr, pd *pollDesc) int32 // 为fd和pd启动边缘触发通知

type pollDesc struct { // Poller对象
    link *pollDesc // 链表
    lock mutex // 保护下面字段
    fd uintptr // fd是底层网络io文件描述符，整个生命期内，不能改变值
    closing bool
    seq uintptr // protect from stale(过时) timers and ready notifications
    rg uintptr // reader goroutine addr
    rt timer
    rd int64
    wg uintptr // writer goroutine addr
    wt timer
    wd int64
    user int32 // user-set cookie用户自定义数据
}

type pollCache struct { // 全局Poller链表
    lock mutex // 保护Poller链表
    first *pollDesc
}

func poll_runtime_pollServerInit() // 调用netpollinit()
func poll_runtime_pollOpen() // 调用netpollopen()
func poll_runtime_pollClose() // 调用netpollclose()
func poll_runtime_pollReset(pd, mode) // 先check(netpollcheckerr(pd, mode))是否有err发生，没有的话重置pd对应字段
func poll_runtime_pollWait(pd, mode) // 先chekerr，再调用netpollblock(pd, mode, false)
func poll_runtime_pollWaitCanceled(pd, mode) // windows下专用
func poll_runtime_pollSetDeadline(pd, deadline, mode)
//1. 重置定时器，并seq++
//2. 设置超时函数netpollDeadline(或者netpollReadDeadline、netpollWriteDeadline)
//3. 如果已经过期，调用netpollunblock和netpollgoready
func poll_runtime_pollUnblock(pd) // netpollUnblock、netpollgoready

/*------------------部分实现------------------*/
func netpollcheckerr(pd, mode) // 检查是否超时或正在关闭
func netpollblockcommit(gp *g, gpp unsafe.Pointer)
func netpollready(gpp *guintptr, pd, mode) // 调用netpollunblock，更新g的schedlink
func netpollgoready(gp *g, traceskip) // 更新统计数据，调用goready --- 通知调度器协程g从parked变为ready
func netpollblock(pd, mode, waitio) // Set rg/wg = pdWait，调用gopark挂起pd对应的g。
func netpollunblock(pd, mode, ioready)
func netpoll(Write/Read)Deadline(arg, seq)

func poll_runtime_ServerInit() {
    netpollinit() // 具现化到Linux下，调用epoll_create
    ...
}

/*****************internal/poll/fd_unix.go*******************/
type FD struct {
    // Lock sysfd and serialize access to Read and Write methods.
    fdmu fdMutex
    // System file descriptor. Immutable until Close.
    Sysfd int
    // I/O poller.
    pd pollDesc
    ...
}
func(fd *FD) Init(net string, pollable bool) error {
    ...
    err := fd.pd.init(fd) // 初始化pd
    ...
}
...
/*****************internal/poll/fd_poll_runtime.go*****************/
type pollDesc struct {
    runtimeCtx uintptr
}
func (pd *pollDesc) init(fd *FD) error {
    serverInit.Do(runtime_pollServerInit) // 初始化PollServer(sync.Once)
    ctx, errno := runtime_pollOpen(uintptr(fd.Sysfd))
    ...
    runtimeCtx = ctx
    return nil
}
...
/*****************runtime/netpoll.go*****************/
func poll_runtime_pollOpen(fd uintptr) (*epDesc, int32) {
    ...
    errno := netpollopen(fd, pd) // 具现化到Linux下，调用epoll_ctl
    ...
}

// runtime/proc.go
...
lastpoll := int64(atomic.Load64(&sched.lastpoll))
now := nanotime()
// 如果10ms内没有poll过，则poll。(1ms=1000000ns)
if lastpoll != 0 && lastpoll+10*1000*1000 < now {
    atomic.Cas64(&sched.lastpoll, uint64(lastpoll), uint64(now))
    gp := netpoll(false) // netpoll在Linux具现为epoll_wait
    if gp != nil {
       injectglist(gp) //把g放到sched中去执行，底层仍然是调用的之前在goroutine里面提到的startm函数。
	}
}
...

conn, err := net.Dial("tcp", "localhost:1208")
...

func socket(...) ... {
	/*
	1. 调用sysSocket创建原生socket
	2. 调用同名包下netFd()，初始化网络文件描述符netFd
	3. 调用fd.dial()，其中最终有调用runtime_pollOpen()加入监听列表
	*/
}

func (fd *FD) Read(p []byte) (int, error) {
    if err := fd.pd.prepareRead(fd.isFile); err != nil {
        return 0, err
    }
    if fd.IsStream && len(p) > maxRW {
        p = p[:maxRW]
    }
    for {
        n, err := syscall.Read(fd.Sysfd, p)
        if err != nil {
            n = 0
            if err == syscall.EAGAIN && fd.pd.pollable() {
                if err = fd.pd.waitRead(fd.isFile); err == nil {
                    continue
                }
            }
        }
        err = fd.eofError(n, err)
        return n, err
    }
}

func (pd *pollDesc) waitRead(isFile bool) error {
    return pd.wait('r', isFile)
}

func (pd *pollDesc) wait(mode int, isFile bool) error {
    if pd.runtimeCtx == 0 {
        return errors.New("waiting for unsupported file type")
    }
    res := runtime_pollWait(pd.runtimeCtx, mode)
    return convertErr(res, isFile)
}

func convertErr(res int, isFile bool) error {
    switch res {
    case 0:
        return nil
    case 1:
        return errClosing(isFile)
    case 2:
        return ErrTimeout
    }
    println("unreachable: ", res)
    panic("unreachable")
}

func poll_runtime_pollWait(pd *pollDesc, mode int) int {
    err := netpollcheckerr(pd, int32(mode))
    if err != 0 {
        return err
    }
    for !netpollblock(pd, int32(mode), false) {
        err = netpollcheckerr(pd, int32(mode))
        if err != 0 {
            return err
        }
    }
    return 0
}

func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
	gpp := &pd.rg
	if mode == 'w' {
		gpp = &pd.wg
	}
    for {
        old := *gpp
        if old == pdReady {
            *gpp = 0
            return true
        }
        if old != 0 {
            throw("runtime: double wait")
        }
        if atomic.Casuintptr(gpp, 0, pdWait) {
            break
        }
    }
    if waitio || netpollcheckerr(pd, mode) == 0 {
	    gopark(netpollblockcommit, unsafe.Pointer(gpp), "IO wait", traceEvGoBlockNet, 5)
    }
    old := atomic.Xchguintptr(gpp, 0)
    if old > pdWait {
        throw("runtime: corrupted polldesc")
    }
    return old == pdReady
}