谈谈golang的netpoll原理解析_Golang

今天谈谈golang源码netpoll部分实现的细节和协程阻塞调度原理

epoll原理

epoll是linux环境下i/o多路复用的模型，结合下图简单说明epoll工作原理

上图说明了epoll生成描epoll表的基本流程，生成socket用来绑定和监听新的连接，将该socket放入epoll内核表，然后调用wait等待就绪事件。

当epoll wait返回就绪事件时，判断是否是新的连接，如果是新的连接则将描述符加入epoll表，监听读写事件。如果不是新的连接，说明已建立的连接上有读或写就绪事件，这样我们根据epollout或者epollin进行写或者读操作，上图是echo server的基本原理，实际生产中监听epollin还是epollout根据实际情况而定。以上是单线程下epoll工作原理。

golang 网络层如何封装的epoll

golang 网络层封装epoll核心文件在系统文件src/runtime/netpoll.go, 这个文件中调用了不同平台封装的多路复用api，linux环境下epoll封装的文件在src/runtime/netpoll_epoll.go中，windows环境下多路复用模型实现在src/runtime/netpoll_windows.go。golang的思想意在将epoll操作放在runtime包里，而runtime是负责协程调度的功能模块，程序启动后runtime运行时是在单独的线程里，个人认为是mpg模型中m模型，epoll模型管理放在这个单独m中调度，m其实是运行在内核态的，在这个内核态线程不断轮询检测就绪事件，将读写就绪事件抛出，从而触发用户态协程读写调度。而我们常用的read，write，accept等操作其实是在用户态操作的，也就是mpg模型中的g，举个例子当read阻塞时，将该协程挂起，当epoll读就绪事件触发后查找阻塞的协程列表，将该协程激活，用户态g激活后继续读，这样在用户态操作是阻塞的，在内核态其实一直是轮询的，这就是golang将epoll和协程调度结合的原理。

golang 如何实现协程和描述符绑定

golang 在internal/poll/fd_windows.go和internal/poll/fd_unix.go中实现了基本的描述符结构

type netfd struct {
    pfd poll.fd
    // immutable until close
    family      int
    sotype      int
    isconnected bool // handshake completed or use of association with peer
    net         string
    laddr       addr
    raddr       addr
}

　　netfd中pfd结构如下

type fd struct {
    // lock sysfd and serialize access to read and write methods.
    fdmu fdmutex
    // system file descriptor. immutable until close.
    sysfd syscall.handle
    // read operation.
    rop operation
    // write operation.
    wop operation
    // i/o poller.
    pd polldesc
    // used to implement pread/pwrite.
    l sync.mutex
    // for console i/o.
    lastbits       []byte   // first few bytes of the last incomplete rune in last write
    readuint16     []uint16 // buffer to hold uint16s obtained with readconsole
    readbyte       []byte   // buffer to hold decoding of readuint16 from utf16 to utf8
    readbyteoffset int      // readbyte[readoffset:] is yet to be consumed with file.read
    // semaphore signaled when file is closed.
    csema uint32
    skipsyncnotif bool
    // whether this is a streaming descriptor, as opposed to a
    // packet-based descriptor like a udp socket.
    isstream bool
    // whether a zero byte read indicates eof. this is false for a
    // message based socket connection.
    zeroreadiseof bool
    // whether this is a file rather than a network socket.
    isfile bool
    // the kind of this file.
    kind filekind
}

　　fd是用户态基本的描述符结构，内部几个变量通过注释可以读懂，挑几个难理解的
fdmu 控制读写互斥访问的锁，因为可能几个协程并发读写
sysfd 系统返回的描述符，不会更改除非系统关闭回收
rop 为读操作，这个其实是根据不同系统网络模型封装的统一类型，比如epoll，iocp等都封装为统一的operation，根据不同的系统调用不同的模型
wop 为写操作封装的类型
pd 这个是最重要的结构，内部封装了协程等基本信息，这个变量会和内核epoll线程通信，从而实现epoll通知和控制用户态协程的效果。
下面我们着重看看polldesc结构

type polldesc struct {
    runtimectx uintptr
}

　　polldesc内部存储了一个unintptr的变量，uintptr为四字节大小的变量，可以存储指针。runtimectx顾名思义，为运行时上下文，其初始化代码如下

func (pd *polldesc) init(fd *fd) error {
    serverinit.do(runtime_pollserverinit)
    ctx, errno := runtime_pollopen(uintptr(fd.sysfd))
    if errno != 0 {
        if ctx != 0 {
            runtime_pollunblock(ctx)
            runtime_pollclose(ctx)
        }
        return errnoerr(syscall.errno(errno))
    }
    pd.runtimectx = ctx
    return nil
}

　　runtime_pollopen实际link的是runtime包下的poll_runtime_pollopen函数，具体实现在runtime/netpoll.go

//go:linkname poll_runtime_pollopen internal/poll.runtime_pollopen
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
    unlock(&pd.lock)
    var errno int32
    errno = netpollopen(fd, pd)
    return pd, int(errno)
}

　　可以看出通过pollcache.alloc返回*polldesc类型的变量pd，并且用pd初始化了netpollopen，这里我们稍作停留，谈谈pollcache

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
        }
        // must be in non-gc memory because can be referenced
        // only from epoll/kqueue internals.
        mem := persistentalloc(n*pdsize, 0, &memstats.other_sys)
        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
    unlock(&c.lock)
    return pd
}

　　alloc函数做了这样的操作，如果链表头为空则初始化pdsize个polldesc节点，并pop出头部，如果不为空则直接pop出头部节点，每个节点的类型就是*polldesc类型，具体实现在runtime/netpoll.go中

type polldesc struct {
    link *polldesc // in pollcache, protected by pollcache.lock
    // the lock protects pollopen, pollsetdeadline, pollunblock and deadlineimpl operations.
    // this fully covers seq, rt and wt variables. fd is constant throughout the polldesc lifetime.
    // pollreset, pollwait, pollwaitcanceled and runtime·netpollready (io readiness notification)
    // proceed w/o taking the lock. so closing, everr, rg, rd, wg and wd are manipulated
    // in a lock-free way by all operations.
    // note(dvyukov): the following code uses uintptr to store *g (rg/wg),
    // that will blow up when gc starts moving objects.
    lock    mutex // protects the following fields
    fd      uintptr
    closing bool
    everr   bool    // marks event scanning error happened
    user    uint32  // user settable cookie
    rseq    uintptr // protects from stale read timers
    rg      uintptr // pdready, pdwait, g waiting for read or nil
    rt      timer   // read deadline timer (set if rt.f != nil)
    rd      int64   // read deadline
    wseq    uintptr // protects from stale write timers
    wg      uintptr // pdready, pdwait, g waiting for write or nil
    wt      timer   // write deadline timer
    wd      int64   // write deadline
}

　　其中rt和wt分别是读写定时器，用来防止读写超时。
fd为描述符指针，lock负责保护polldesc内部成员变量读写防止多线程操作导致并发问题。
除此之外最重要的是rg和wg两个变量，rg保存了用户态操作polldesc的读协程地址，wg保存了用户态操作polldesc写协程地址。
举个例子，当我们在在用户态协程调用read阻塞时rg就被设置为该读协程，当内核态epoll_wait检测read就绪后就会通过rg找到这个协程让后恢复运行。
rg,wg默认是0，rg为pdready表示读就绪，可以将协程恢复，为pdwait表示读阻塞，协程将要被挂起。wg也是如此。
所以golang其实是通过polldesc实现用户态和内核态信息的共享的。
回到之前poll_runtime_pollopen函数，我们就理解了其内部生成*polldesc，并且传入netpollopen函数，netpollopen对应实现了epoll的init和wait，从而达到了用户态信息和内核态的关联。

netpollopen函数不同模型的实现不相同，epoll的实现在runtime/netpoll_epoll.go中

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)
}

从而实现了epoll将fd添加至内核epoll表里，同样pd作为event的data传入内核表，从而实现内核态和用户态协程的关联。
runtime/netpoll_epoll.go实现了epoll模型的基本操作，详见源码。

golang如何将一个描述符加入epoll表中

传统的流程为:
生成socket–> bind socket–> listen–> accept
在golang中生成socket，bind，以及listen统一封装好了
listen–> lc.listen –> sl.listentcp –> internetsocket
internetsocket –> socket –> newfd && listenstream
在newfd中完成了描述符创建，在listenstream完成了bind和listen。newfd只初始化了基本的结构，未完成polldesc类型变量pd的初始化。
我们跟随源码查看listen的绑定流程

unc (lc *listenconfig) listen(ctx context.context, network, address string) (listener, error) {
    addrs, err := defaultresolver.resolveaddrlist(ctx, "listen", network, address, nil)
    if err != nil {
        return nil, &operror{op: "listen", net: network, source: nil, addr: nil, err: err}
    }
    sl := &syslistener{
        listenconfig: *lc,
        network:      network,
        address:      address,
    }
    var l listener
    la := addrs.first(isipv4)
    switch la := la.(type) {
    case *tcpaddr:
        l, err = sl.listentcp(ctx, la)
    case *unixaddr:
        l, err = sl.listenunix(ctx, la)
    default:
        return nil, &operror{op: "listen", net: sl.network, source: nil, addr: la, err: &addrerror{err: "unexpected address type", addr: address}}
    }
    if err != nil {
        return nil, &operror{op: "listen", net: sl.network, source: nil, addr: la, err: err} // l is non-nil interface containing nil pointer
    }
    return l, nil
}

　可以看出listen函数返回的类型为listener接口类型，其内部根据la类型调用不同的listen函数，这里查看listentcp　

func (sl *syslistener) listentcp(ctx context.context, laddr *tcpaddr) (*tcplistener, error) {
    fd, err := internetsocket(ctx, sl.network, laddr, nil, syscall.sock_stream, 0, "listen", sl.listenconfig.control)
    if err != nil {
        return nil, err
    }
    return &tcplistener{fd: fd, lc: sl.listenconfig}, nil
}

　　internetsocket内部调用socket生成描述符返回

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) {
    s, err := syssocket(family, sotype, proto)
    if err != nil {
        return nil, err
    }
    if err = setdefaultsockopts(s, family, sotype, ipv6only); err != nil {
        poll.closefunc(s)
        return nil, err
    }
    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:
            if err := fd.listenstream(laddr, listenerbacklog(), ctrlfn); err != nil {
                fd.close()
                return nil, err
            }
            return fd, nil
        case syscall.sock_dgram:
            if err := fd.listendatagram(laddr, ctrlfn); err != nil {
                fd.close()
                return nil, err
            }
            return fd, nil
        }
    }
    if err := fd.dial(ctx, laddr, raddr, ctrlfn); err != nil {
        fd.close()
        return nil, err
    }
    return fd, nil
}

socket函数做了这样几件事
1 调用syssocket生成描述符
2 调用newfd封装描述符，构造netfd类型变量
3 调用netfd的listendatagram方法，实现bind和listen

func (fd *netfd) listenstream(laddr sockaddr, backlog int, ctrlfn func(string, string, syscall.rawconn) error) error {
    var err error
    if err = setdefaultlistenersockopts(fd.pfd.sysfd); err != nil {
        return err
    }
    var lsa syscall.sockaddr
    if lsa, err = laddr.sockaddr(fd.family); err != nil {
        return err
    }
    if ctrlfn != nil {
        c, err := newrawconn(fd)
        if err != nil {
            return err
        }
        if err := ctrlfn(fd.ctrlnetwork(), laddr.string(), c); err != nil {
            return err
        }
    }
    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)
    }
    if err = fd.init(); err != nil {
        return err
    }
    lsa, _ = syscall.getsockname(fd.pfd.sysfd)
    fd.setaddr(fd.addrfunc()(lsa), nil)
    return nil
}

　　listenstream除了bind和listen操作之外，还执行了netfd的init操作，这个init操作就是将netfd和epoll关联，将描述符和协程信息写入epoll表

func (fd *netfd) init() error {
    errcall, err := fd.pfd.init(fd.net, true)
    if errcall != "" {
        err = wrapsyscallerror(errcall, err)
    }
    return err
}

前文讲过fd.pfd为fd类型，是和epoll通信的核心结构，fd的init方法内完成了polldesc类型成员变量pd和epoll的关联。
其内部调用了fd.pd.init(fd)，pd就是fd的polldesc类型成员变量，其init函数上面已经解释过了调用了runtime_pollopen，runtime_pollopen是link到
runtime/netpoll.go中poll_runtime_pollopen函数，这个函数将用户态协程的polldesc信息写入到epoll所在的单独线程，从而实现用户态和内核态的关联。
总结下bind和listen后续的消息流程就是：
listenstream –> bind&listen&init –> polldesc.init –> runtime_pollopen
–> poll_runtime_pollopen –> epollctl(epoll_ctl_add)

到此为止golang网络描述符从生成到绑定和监听，以及写入epoll表的流程分析完毕，下一篇分析accept流程以及用户态协程如何挂起，epoll就绪后如何唤醒协程。

到此这篇关于谈谈golang的netpoll原理解析的文章就介绍到这了,更多相关golang的netpoll原理内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！