golang网络框架netpoll(Multi-Reactor模型)核心源码分析

func main(){
    // 1.创建listener
    var listener, _ = CreateListener(network, address)
    // 2.初始化EventLoop
    var eventLoop, _ = NewEventLoop(func(ctx context.Context, connection Connection) error {
        time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
        if l := connection.Reader().Len(); l > 0 {
            var data, err = connection.Reader().Next(l)
            if err != nil {
                return err
            }
            fmt.Printf("data:%+v\n", string(data))
        }
        return nil
    })
    // 3.阻塞等待客户端请求
    eventLoop.Serve(listener)

}

// A EventLoop is a network server.
type EventLoop interface {
    // Serve registers a listener and runs blockingly to provide services, including listening to ports,
    // accepting connections and processing trans data. When an exception occurs or Shutdown is invoked,
    // Serve will return an error which describes the specific reason.
    Serve(ln net.Listener) error

    // Shutdown is used to graceful exit.
    // It will close all idle connections on the server, but will not change the underlying pollers.
    //
    // Argument: ctx set the waiting deadline, after which an error will be returned,
    // but will not force the closing of connections in progress.
    Shutdown(ctx context.Context) error
}


// OnRequest defines the function for handling connection. When data is sent from the connection peer,
// netpoll actively reads the data in LT mode and places it in the connection's input buffer.
// Generally, OnRequest starts handling the data in the following way:
//
//  func OnRequest(ctx context, connection Connection) error {
//      input := connection.Reader().Next(n)
//      handling input data...
//      send, _ := connection.Writer().Malloc(l)
//      copy(send, output)
//      connection.Flush()
//      return nil
//  }
//
// OnRequest will run in a separate goroutine and
// it is guaranteed that there is one and only one OnRequest running at the same time.
// The underlying logic is similar to:
//
//  go func() {
//      for !connection.Reader().IsEmpty() {
//          OnRequest(ctx, connection)
//      }
//  }()
//
// PLEASE NOTE:
// OnRequest must either eventually read all the input data or actively Close the connection,
// otherwise the goroutine will fall into a dead loop.
//
// Return: error is unused which will be ignored directly.
type OnRequest func(ctx context.Context, connection Connection) error

// NewEventLoop .
func NewEventLoop(onRequest OnRequest, ops ...Option) (EventLoop, error) {
    opts := &options{
        onRequest: onRequest,
    }
    for _, do := range ops {
        do.f(opts)
    }
    return &eventLoop{
        opts: opts,
        stop: make(chan error, 1),
    }, nil
}

type eventLoop struct {
    sync.Mutex
    svr  *server

    opts *options
    stop chan error
}

// Serve implements EventLoop.
func (evl *eventLoop) Serve(ln net.Listener) error {
    npln, err := ConvertListener(ln)
    if err != nil {
        return err
    }
    evl.Lock()
    evl.svr = newServer(npln, evl.opts, evl.quit)
    // 开启所有的epoll，然后异步协程阻塞等待
    evl.svr.Run()
    evl.Unlock()

    // 阻塞住
    err = evl.waitQuit()
    // ensure evl will not be finalized until Serve returns
    runtime.SetFinalizer(evl, nil)
    return err
}

// newServer wrap listener into server, quit will be invoked when server exit.
func newServer(ln Listener, opts *options, onQuit func(err error)) *server {
    return &server{
        ln:     ln,
        opts:   opts,
        onQuit: onQuit,
    }
}

type server struct {
    operator    FDOperator
    ln          Listener
    opts        *options
    onQuit      func(err error)
    connections sync.Map // key=fd, value=connection
}

// Run this server.
func (s *server) Run() (err error) {
    s.operator = FDOperator{
        FD:     s.ln.Fd(),
        OnRead: s.OnRead,
        OnHup:  s.OnHup,
    }
    // 从pollmanager中选择出来一个epoll，来管理server fd，也就是设置mainReactor
    s.operator.poll = pollmanager.Pick()
    // 服务端设置可读
    err = s.operator.Control(PollReadable)

    if err != nil {
        s.onQuit(err)
    }
    return err
}


// OnRead implements FDOperator.
// 服务端读就绪时，处理接收客户端连接数据
func (s *server) OnRead(p Poll) error {
    // accept socket
    // 接收客户端连接
    conn, err := s.ln.Accept()
    if err != nil {
        // shut down
        if strings.Contains(err.Error(), "closed") {
            s.operator.Control(PollDetach)
            s.onQuit(err)
            return err
        }
        log.Println("accept conn failed:", err.Error())
        return err
    }
    if conn == nil {
        return nil
    }
    // store & register connection
    // 构建一个新的连接
    var connection = &connection{}
    // 加入到epoll中，并传入opts，
    // 其中包含处理业务逻辑的OnRequest
    connection.init(conn.(Conn), s.opts)

    if !connection.IsActive() {
        return nil
    }
    var fd = conn.(Conn).Fd()

    // 存储新的连接
    connection.AddCloseCallback(func(connection Connection) error {
        s.connections.Delete(fd)
        return nil
    })
    s.connections.Store(fd, connection)

    // trigger onConnect asynchronously
    connection.onConnect()
    return nil
}

    s.operator.poll = pollmanager.Pick()
    // 服务端设置可读
    err = s.operator.Control(PollReadable)

    connection.init(conn.(Conn), s.opts)

// init initialize the connection with options
func (c *connection) init(conn Conn, opts *options) (err error) {
    // init buffer, barrier, finalizer
    c.readTrigger = make(chan struct{}, 1)
    c.writeTrigger = make(chan error, 1)
    c.bookSize, c.maxSize = block1k/2, pagesize
    c.inputBuffer, c.outputBuffer = NewLinkBuffer(pagesize), NewLinkBuffer()
    c.inputBarrier, c.outputBarrier = barrierPool.Get().(*barrier), barrierPool.Get().(*barrier)

    // 初始化fd
    c.initNetFD(conn) // conn must be *netFD{}
    // 初始化FDOperator
    c.initFDOperator()
    c.initFinalizer()

    syscall.SetNonblock(c.fd, true)
    // enable TCP_NODELAY by default
    switch c.network {
    case "tcp", "tcp4", "tcp6":
        setTCPNoDelay(c.fd, true)
    }
    // check zero-copy
    if setZeroCopy(c.fd) == nil && setBlockZeroCopySend(c.fd, defaultZeroCopyTimeoutSec, 0) == nil {
        c.supportZeroCopy = true
    }

    // connection initialized and prepare options
   // 做连接初始化和准备工作
    return c.onPrepare(opts)
}

func (c *connection) initNetFD(conn Conn) {
    if nfd, ok := conn.(*netFD); ok {
        c.netFD = *nfd
        return
    }
    c.netFD = netFD{
        fd:         conn.Fd(),
        localAddr:  conn.LocalAddr(),
        remoteAddr: conn.RemoteAddr(),
    }
}



func (c *connection) initFDOperator() {
    // 分配一个FDOperator
    op := allocop()
    op.FD = c.fd
    op.OnRead, op.OnWrite, op.OnHup = nil, nil, c.onHup
    // 设置输入处理方法和输入的响应回调接口
    op.Inputs, op.InputAck = c.inputs, c.inputAck
    // 当outputs为空时，监听可读事件
    op.Outputs, op.OutputAck = c.outputs, c.outputAck

    // if connection has been registered, must reuse poll here.
    if c.pd != nil && c.pd.operator != nil {
        op.poll = c.pd.operator.poll
    }
    c.operator = op
}


// OnPrepare supports close connection, but not read/write data.
// connection will be registered by this call after preparing.
func (c *connection) onPrepare(opts *options) (err error) {
    if opts != nil {
        c.SetOnConnect(opts.onConnect)
        c.SetOnRequest(opts.onRequest)
        c.SetReadTimeout(opts.readTimeout)
        c.SetIdleTimeout(opts.idleTimeout)

        // calling prepare first and then register.
        if opts.onPrepare != nil {
     // 执行onPrepare回调方法
            c.ctx = opts.onPrepare(c)
        }
    }

    if c.ctx == nil {
        c.ctx = context.Background()
    }
    // prepare may close the connection.
    if c.IsActive() {
        // 注册客户端，注册到Poll(epoll)中
        return c.register()
    }
    return nil
}

// register only use for connection register into poll.
// 注册客户端到epoll中
func (c *connection) register() (err error) {
    // 将当前的客户端加入到epoll中管理，此处对应的是subReactor
    if c.operator.poll != nil {
        err = c.operator.Control(PollModReadable)
    } else {
        c.operator.poll = pollmanager.Pick()
        err = c.operator.Control(PollReadable)
    }
    if err != nil {
        log.Println("connection register failed:", err.Error())
        c.Close()
        return Exception(ErrConnClosed, err.Error())
    }
    return nil
}

// 设置loop数量
func setNumLoops(numLoops int) error {
    return pollmanager.SetNumLoops(numLoops)
}

func setLoadBalance(lb LoadBalance) error {
    return pollmanager.SetLoadBalance(lb)
}

// manage all pollers
var pollmanager *manager

// 启动的时候进行初始化
func init() {
    pollmanager = &manager{}
    // 设置负载均衡
    pollmanager.SetLoadBalance(RoundRobin)
    // 设置了循环个数，也就是epoll个数
    pollmanager.SetNumLoops(defaultNumLoops())
}

func defaultNumLoops() int {
    procs := runtime.GOMAXPROCS(0)
    loops := 1
    // Loops produce events that handlers consume,
    // so the producer should be faster than consumer otherwise it will have a bottleneck.
    // But there is no universal option that could be appropriate for any use cases,
    // plz use `SetNumLoops` if you do know what you want.
    if procs > 4 {
        loops = procs
    }
    return loops
}

// LoadBalance is used to do load balancing among multiple pollers.
// a single poller may not be optimal if the number of cores is large (40C+).
type manager struct {
    NumLoops int
    balance  loadbalance // load balancing method
    // 维护所有的Poll对象
    polls    []Poll      // all the polls
}

// SetNumLoops will return error when set numLoops < 1
func (m *manager) SetNumLoops(numLoops int) error {
    if numLoops < 1 {
        return fmt.Errorf("set invaild numLoops[%d]", numLoops)
    }
    // if less than, reset all; else new the delta.
    if numLoops < m.NumLoops {
        m.NumLoops = numLoops
        return m.Reset()
    }
    m.NumLoops = numLoops
    return m.Run()
}

// SetLoadBalance set load balance.
func (m *manager) SetLoadBalance(lb LoadBalance) error {
    if m.balance != nil && m.balance.LoadBalance() == lb {
        return nil
    }
    m.balance = newLoadbalance(lb, m.polls)
    return nil
}

// Run all pollers.
func (m *manager) Run() error {
    // new poll to fill delta.
    for idx := len(m.polls); idx < m.NumLoops; idx++ {
        // 创建epoll
        var poll = openPoll()
        m.polls = append(m.polls, poll)
        // 每个epoll阻塞等待
        go poll.Wait()
    }
    // LoadBalance must be set before calling Run, otherwise it will panic.
    m.balance.Rebalance(m.polls)
    return nil
}


// Pick will select the poller for use each time based on the LoadBalance.
func (m *manager) Pick() Poll {
    return m.balance.Pick()
}

// Poll monitors fd(file descriptor), calls the FDOperator to perform specific actions,
// and shields underlying differences. On linux systems, poll uses epoll by default,
// and kevent by default on bsd systems.
type Poll interface {
    // Wait will poll all registered fds, and schedule processing based on the triggered event.
    // The call will block, so the usage can be like:
    //
    //  go wait()
    //
    Wait() error

    // Close the poll and shutdown Wait().
    Close() error

    // Trigger can be used to actively refresh the loop where Wait is located when no event is triggered.
    // On linux systems, eventfd is used by default, and kevent by default on bsd systems.
    Trigger() error

    // Control the event of file descriptor and the operations is defined by PollEvent.
    Control(operator *FDOperator, event PollEvent) error
}

// PollEvent defines the operation of poll.Control.
type PollEvent int

const (
    // PollReadable is used to monitor whether the FDOperator registered by
    // listener and connection is readable or closed.
    PollReadable PollEvent = 0x1

    // PollWritable is used to monitor whether the FDOperator created by the dialer is writable or closed.
    // ET mode must be used (still need to poll hup after being writable)
    PollWritable PollEvent = 0x2

    // PollDetach is used to remove the FDOperator from poll.
    PollDetach PollEvent = 0x3

    // PollModReadable is used to re-register the readable monitor for the FDOperator created by the dialer.
    // It is only used when calling the dialer's conn init.
    PollModReadable PollEvent = 0x4

    // PollR2RW is used to monitor writable for FDOperator,
    // which is only called when the socket write buffer is full.
    PollR2RW PollEvent = 0x5

    // PollRW2R is used to remove the writable monitor of FDOperator, generally used with PollR2RW.
    PollRW2R PollEvent = 0x6
)

const EPOLLET = -syscall.EPOLLET

type epollevent struct {
    events uint32
    data   [8]byte // unaligned uintptr
}

// EpollCtl implements epoll_ctl.
func EpollCtl(epfd int, op int, fd int, event *epollevent) (err error) {
    _, _, err = syscall.RawSyscall6(syscall.SYS_EPOLL_CTL, uintptr(epfd), uintptr(op), uintptr(fd), uintptr(unsafe.Pointer(event)), 0, 0)
    if err == syscall.Errno(0) {
        err = nil
    }
    return err
}

// EpollWait implements epoll_wait.
func EpollWait(epfd int, events []epollevent, msec int) (n int, err error) {
    var r0 uintptr
    var _p0 = unsafe.Pointer(&events[0])
    if msec == 0 {
        r0, _, err = syscall.RawSyscall6(syscall.SYS_EPOLL_WAIT, uintptr(epfd), uintptr(_p0), uintptr(len(events)), 0, 0, 0)
    } else {
        r0, _, err = syscall.Syscall6(syscall.SYS_EPOLL_WAIT, uintptr(epfd), uintptr(_p0), uintptr(len(events)), uintptr(msec), 0, 0)
    }
    if err == syscall.Errno(0) {
        err = nil
    }
    return int(r0), err
}



// Includes defaultPoll/multiPoll/uringPoll...
func openPoll() Poll {
    return openDefaultPoll()
}

func openDefaultPoll() *defaultPoll {
    var poll = defaultPoll{}
    poll.buf = make([]byte, 8)
    // 创建epoll
    var p, err = syscall.EpollCreate1(0)
    if err != nil {
        panic(err)
    }
    poll.fd = p
    var r0, _, e0 = syscall.Syscall(syscall.SYS_EVENTFD2, 0, 0, 0)
    if e0 != 0 {
        syscall.Close(p)
        panic(err)
    }

    poll.Reset = poll.reset
    poll.Handler = poll.handler

    poll.wop = &FDOperator{FD: int(r0)}
    poll.Control(poll.wop, PollReadable)
    return &poll
}

type defaultPoll struct {
    pollArgs
    fd      int         // epoll fd
    wop     *FDOperator // eventfd, wake epoll_wait
    buf     []byte      // read wfd trigger msg
    trigger uint32      // trigger flag
    // fns for handle events
    Reset   func(size, caps int)
    Handler func(events []epollevent) (closed bool)
}

type pollArgs struct {
    size     int
    caps     int
    events   []epollevent
    barriers []barrier
}

func (a *pollArgs) reset(size, caps int) {
    a.size, a.caps = size, caps
    a.events, a.barriers = make([]epollevent, size), make([]barrier, size)
    for i := range a.barriers {
        a.barriers[i].bs = make([][]byte, a.caps)
        a.barriers[i].ivs = make([]syscall.Iovec, a.caps)
    }
}

// Wait implements Poll.
func (p *defaultPoll) Wait() (err error) {
    // init
    var caps, msec, n = barriercap, -1, 0
    p.Reset(128, caps)
    // wait
    for {
        if n == p.size && p.size < 128*1024 {
            p.Reset(p.size<<1, caps)
        }
        // 调用epoll_wait()
        n, err = EpollWait(p.fd, p.events, msec)
        if err != nil && err != syscall.EINTR {
            return err
        }
        if n <= 0 {
            msec = -1
            runtime.Gosched()
            continue
        }
        msec = 0
        if p.Handler(p.events[:n]) {
            return nil
        }
    }
}

func (p *defaultPoll) handler(events []epollevent) (closed bool) {
    var hups []*FDOperator // TODO: maybe can use sync.Pool
    for i := range events {
        var operator = *(**FDOperator)(unsafe.Pointer(&events[i].data))
        // trigger or exit gracefully
        if operator.FD == p.wop.FD {
            // must clean trigger first
            syscall.Read(p.wop.FD, p.buf)
            atomic.StoreUint32(&p.trigger, 0)
            // if closed & exit
            if p.buf[0] > 0 {
                syscall.Close(p.wop.FD)
                syscall.Close(p.fd)
                return true
            }
            continue
        }
        if !operator.do() {
            continue
        }

        evt := events[i].events
        switch {
        // check hup first
        case evt&(syscall.EPOLLHUP|syscall.EPOLLRDHUP) != 0:
            hups = append(hups, operator)
        case evt&syscall.EPOLLERR != 0:
            // Under block-zerocopy, the kernel may give an error callback, which is not a real error, just an EAGAIN.
            // So here we need to check this error, if it is EAGAIN then do nothing, otherwise still mark as hup.
            if _, _, _, _, err := syscall.Recvmsg(operator.FD, nil, nil, syscall.MSG_ERRQUEUE); err != syscall.EAGAIN {
                hups = append(hups, operator)
            }
        default:
            // 读事件
            if evt&syscall.EPOLLIN != 0 {
                if operator.OnRead != nil {
                    // for non-connection
                    operator.OnRead(p)
                } else {
                    // for connection
                    var bs = operator.Inputs(p.barriers[i].bs)
                    if len(bs) > 0 {
                        var n, err = readv(operator.FD, bs, p.barriers[i].ivs)
                        // 内部会有处理业务逻辑的调用
                        operator.InputAck(n)
                        if err != nil && err != syscall.EAGAIN && err != syscall.EINTR {
                            log.Printf("readv(fd=%d) failed: %s", operator.FD, err.Error())
                            hups = append(hups, operator)
                            break
                        }
                    }
                }
            }
            // 写事件
            if evt&syscall.EPOLLOUT != 0 {
                if operator.OnWrite != nil {
                    // for non-connection
                    operator.OnWrite(p)
                } else {
                    // for connection
                    var bs, supportZeroCopy = operator.Outputs(p.barriers[i].bs)
                    if len(bs) > 0 {
                        // TODO: Let the upper layer pass in whether to use ZeroCopy.
                        var n, err = sendmsg(operator.FD, bs, p.barriers[i].ivs, false && supportZeroCopy)
                        operator.OutputAck(n)
                        if err != nil && err != syscall.EAGAIN {
                            log.Printf("sendmsg(fd=%d) failed: %s", operator.FD, err.Error())
                            hups = append(hups, operator)
                            break
                        }
                    }
                }
            }
        }
        operator.done()
    }
    // hup conns together to avoid blocking the poll.
    if len(hups) > 0 {
        p.detaches(hups)
    }
    return false
}

// Close will write 10000000
func (p *defaultPoll) Close() error {
    _, err := syscall.Write(p.wop.FD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
    return err
}

// Trigger implements Poll.
func (p *defaultPoll) Trigger() error {
    if atomic.AddUint32(&p.trigger, 1) > 1 {
        return nil
    }
    // MAX(eventfd) = 0xfffffffffffffffe
    _, err := syscall.Write(p.wop.FD, []byte{0, 0, 0, 0, 0, 0, 0, 1})
    return err
}

// Control implements Poll.
func (p *defaultPoll) Control(operator *FDOperator, event PollEvent) error {
    var op int
    var evt epollevent
    *(**FDOperator)(unsafe.Pointer(&evt.data)) = operator
    switch event {
    case PollReadable:
        operator.inuse()
        op, evt.events = syscall.EPOLL_CTL_ADD, syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLERR
    case PollModReadable:
        operator.inuse()
        op, evt.events = syscall.EPOLL_CTL_MOD, syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLERR
    case PollDetach:
        defer operator.unused()
        op, evt.events = syscall.EPOLL_CTL_DEL, syscall.EPOLLIN|syscall.EPOLLOUT|syscall.EPOLLRDHUP|syscall.EPOLLERR
    case PollWritable:
        operator.inuse()
        op, evt.events = syscall.EPOLL_CTL_ADD, EPOLLET|syscall.EPOLLOUT|syscall.EPOLLRDHUP|syscall.EPOLLERR
    case PollR2RW:
        op, evt.events = syscall.EPOLL_CTL_MOD, syscall.EPOLLIN|syscall.EPOLLOUT|syscall.EPOLLRDHUP|syscall.EPOLLERR
    case PollRW2R:
        op, evt.events = syscall.EPOLL_CTL_MOD, syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLERR
    }
    // epoll_ctl
    return EpollCtl(p.fd, op, operator.FD, &evt)
}

func (p *defaultPoll) detaches(hups []*FDOperator) error {
    var onhups = make([]func(p Poll) error, len(hups))
    for i := range hups {
        onhups[i] = hups[i].OnHup
        p.Control(hups[i], PollDetach)
    }
    go func(onhups []func(p Poll) error) {
        for i := range onhups {
            if onhups[i] != nil {
                onhups[i](p)
            }
        }
    }(onhups)
    return nil
}

// inputAck implements FDOperator.
func (c *connection) inputAck(n int) (err error) {
    if n < 0 {
        n = 0
    }
    // Auto size bookSize.
    if n == c.bookSize && c.bookSize < mallocMax {
        c.bookSize <<= 1
    }

    length, _ := c.inputBuffer.bookAck(n)
    if c.maxSize < length {
        c.maxSize = length
    }
    if c.maxSize > mallocMax {
        c.maxSize = mallocMax
    }

    var needTrigger = true
    if length == n { // first start onRequest
        // 读完数据后调用处理逻辑方法
        needTrigger = c.onRequest()
    }
    if needTrigger && length >= int(atomic.LoadInt32(&c.waitReadSize)) {
        c.triggerRead()
    }
    return nil
}

// onRequest is responsible for executing the closeCallbacks after the connection has been closed.
func (c *connection) onRequest() (needTrigger bool) {
    var onRequest, ok = c.onRequestCallback.Load().(OnRequest)
    if !ok {
        return true
    }
    processed := c.onProcess(
        // only process when conn active and have unread data
        func(c *connection) bool {
            return c.Reader().Len() > 0 && c.IsActive()
        },
        func(c *connection) {
        // 执行业务逻辑方法
            _ = onRequest(c.ctx, c)
        },
    )
    // if not processed, should trigger read
    return !processed
}

// onProcess is responsible for executing the process function serially,
// and make sure the connection has been closed correctly if user call c.Close() in process function.
func (c *connection) onProcess(isProcessable func(c *connection) bool, process func(c *connection)) (processed bool) {
    if process == nil {
        return false
    }
    // task already exists
    if !c.lock(processing) {
        return false
    }
    // add new task
    var task = func() {
    START:
        // Single request processing, blocking allowed.
        for isProcessable(c) {
            process(c)
        }
        // Handling callback if connection has been closed.
        if !c.IsActive() {
            c.closeCallback(false)
            return
        }
        c.unlock(processing)
        // Double check when exiting.
        if isProcessable(c) && c.lock(processing) {
            goto START
        }
        // task exits
        return
    }

    runTask(c.ctx, task)
    return true
}

var runTask = gopool.CtxGo

// Flush will send all malloc data to the peer,
// so must confirm that the allocated bytes have been correctly assigned.
//
// Flush first checks whether the out buffer is empty.
// If empty, it will call syscall.Write to send data directly,
// otherwise the buffer will be sent asynchronously by the epoll trigger.
func (c *connection) Flush() error {
    if !c.lock(flushing) {
        return Exception(ErrConnClosed, "when flush")
    }
    defer c.unlock(flushing)
    c.outputBuffer.Flush()
    return c.flush()
}

// Write will Flush soon.
func (c *connection) Write(p []byte) (n int, err error) {
    if !c.lock(flushing) {
        return 0, Exception(ErrConnClosed, "when write")
    }
    defer c.unlock(flushing)

    dst, _ := c.outputBuffer.Malloc(len(p))
    n = copy(dst, p)
    c.outputBuffer.Flush()
    err = c.flush()
    return n, err
}



// flush write data directly.
func (c *connection) flush() error {
    if c.outputBuffer.IsEmpty() {
        return nil
    }
    // TODO: Let the upper layer pass in whether to use ZeroCopy.
    var bs = c.outputBuffer.GetBytes(c.outputBarrier.bs)
    var n, err = sendmsg(c.fd, bs, c.outputBarrier.ivs, false && c.supportZeroCopy)
    if err != nil && err != syscall.EAGAIN {
        return Exception(err, "when flush")
    }
    if n > 0 {
        err = c.outputBuffer.Skip(n)
        c.outputBuffer.Release()
        if err != nil {
            return Exception(err, "when flush")
        }
    }
    // return if write all buffer.
    if c.outputBuffer.IsEmpty() {
        return nil
    }
    //更新描述符状态为可读写
    err = c.operator.Control(PollR2RW)
    if err != nil {
        return Exception(err, "when flush")
    }

    err = <-c.writeTrigger
    return err
}

// Dialer extends net.Dialer's API, just for interface compatibility.
// DialConnection is recommended, but of course all functions are practically the same.
// The returned net.Conn can be directly asserted as Connection if error is nil.
type Dialer interface {
    DialConnection(network, address string, timeout time.Duration) (connection Connection, err error)

    DialTimeout(network, address string, timeout time.Duration) (conn net.Conn, err error)
}

// DialConnection is a default implementation of Dialer.
func DialConnection(network, address string, timeout time.Duration) (connection Connection, err error) {
    return defaultDialer.DialConnection(network, address, timeout)
}

// NewDialer only support TCP and unix socket now.
func NewDialer() Dialer {
    return &dialer{}
}

var defaultDialer = NewDialer()

type dialer struct{}

// DialTimeout implements Dialer.
func (d *dialer) DialTimeout(network, address string, timeout time.Duration) (net.Conn, error) {
    conn, err := d.DialConnection(network, address, timeout)
    return conn, err
}

// DialConnection implements Dialer.
func (d *dialer) DialConnection(network, address string, timeout time.Duration) (connection Connection, err error) {
    ctx := context.Background()
    if timeout > 0 {
        subCtx, cancel := context.WithTimeout(ctx, timeout)
        defer cancel()
        ctx = subCtx
    }

    switch network {
    case "tcp", "tcp4", "tcp6":
        var raddr *TCPAddr
        raddr, err = ResolveTCPAddr(network, address)
        if err != nil {
            return nil, err
        }
        connection, err = DialTCP(ctx, network, nil, raddr)
    // case "udp", "udp4", "udp6":  // TODO: unsupport now
    case "unix", "unixgram", "unixpacket":
        var raddr *UnixAddr
        raddr, err = ResolveUnixAddr(network, address)
        if err != nil {
            return nil, err
        }
        connection, err = DialUnix(network, nil, raddr)
    default:
        return nil, net.UnknownNetworkError(network)
    }
    return connection, err
}

// sysDialer contains a Dial's parameters and configuration.
type sysDialer struct {
    net.Dialer
    network, address string
}

// TCPConnection implements Connection.
type TCPConnection struct {
    connection
}

// newTCPConnection wraps *TCPConnection.
func newTCPConnection(conn Conn) (connection *TCPConnection, err error) {
    connection = &TCPConnection{}
    err = connection.init(conn, nil)
    if err != nil {
        return nil, err
    }
    return connection, nil
}

// DialTCP acts like Dial for TCP networks.
//
// The network must be a TCP network name; see func Dial for details.
//
// If laddr is nil, a local address is automatically chosen.
// If the IP field of raddr is nil or an unspecified IP address, the
// local system is assumed.
func DialTCP(ctx context.Context, network string, laddr, raddr *TCPAddr) (*TCPConnection, error) {
    switch network {
    case "tcp", "tcp4", "tcp6":
    default:
        return nil, &net.OpError{Op: "dial", Net: network, Source: laddr.opAddr(), Addr: raddr.opAddr(), Err: net.UnknownNetworkError(network)}
    }
    if raddr == nil {
        return nil, &net.OpError{Op: "dial", Net: network, Source: laddr.opAddr(), Addr: nil, Err: errMissingAddress}
    }
    if ctx == nil {
        ctx = context.Background()
    }
    sd := &sysDialer{network: network, address: raddr.String()}
    c, err := sd.dialTCP(ctx, laddr, raddr)
    if err != nil {
        return nil, &net.OpError{Op: "dial", Net: network, Source: laddr.opAddr(), Addr: raddr.opAddr(), Err: err}
    }
    return c, nil
}

func (sd *sysDialer) dialTCP(ctx context.Context, laddr, raddr *TCPAddr) (*TCPConnection, error) {
   // 内容实现建立连接的逻辑
    conn, err := internetSocket(ctx, sd.network, laddr, raddr, syscall.SOCK_STREAM, 0, "dial")

    // TCP has a rarely used mechanism called a 'simultaneous connection' in
    // which Dial("tcp", addr1, addr2) run on the machine at addr1 can
    // connect to a simultaneous Dial("tcp", addr2, addr1) run on the machine
    // at addr2, without either machine executing Listen. If laddr == nil,
    // it means we want the kernel to pick an appropriate originating local
    // address. Some Linux kernels cycle blindly through a fixed range of
    // local ports, regardless of destination port. If a kernel happens to
    // pick local port 50001 as the source for a Dial("tcp", "", "localhost:50001"),
    // then the Dial will succeed, having simultaneously connected to itself.
    // This can only happen when we are letting the kernel pick a port (laddr == nil)
    // and when there is no listener for the destination address.
    // It's hard to argue this is anything other than a kernel bug. If we
    // see this happen, rather than expose the buggy effect to users, we
    // close the conn and try again. If it happens twice more, we relent and
    // use the result. See also:
    //  https://golang.org/issue/2690
    //  https://stackoverflow.com/questions/4949858/
    //
    // The opposite can also happen: if we ask the kernel to pick an appropriate
    // originating local address, sometimes it picks one that is already in use.
    // So if the error is EADDRNOTAVAIL, we have to try again too, just for
    // a different reason.
    //
    // The kernel socket code is no doubt enjoying watching us squirm.
    for i := 0; i < 2 && (laddr == nil || laddr.Port == 0) && (selfConnect(conn, err) || spuriousENOTAVAIL(err)); i++ {
        if err == nil {
            conn.Close()
        }
        conn, err = internetSocket(ctx, sd.network, laddr, raddr, syscall.SOCK_STREAM, 0, "dial")
    }

    if err != nil {
        return nil, err
    }
    return newTCPConnection(conn)
}

func internetSocket(ctx context.Context, net string, laddr, raddr sockaddr, sotype, proto int, mode string) (conn *netFD, err error) {
    if (runtime.GOOS == "aix" || runtime.GOOS == "windows" || runtime.GOOS == "openbsd" || runtime.GOOS == "nacl") && raddr.isWildcard() {
        raddr = raddr.toLocal(net)
    }
    family, ipv6only := favoriteAddrFamily(net, laddr, raddr)
    // 创建socket
    return socket(ctx, net, family, sotype, proto, ipv6only, laddr, raddr)
}

// socket returns a network file descriptor that is ready for
// asynchronous I/O using the network poller.
func socket(ctx context.Context, net string, family, sotype, proto int, ipv6only bool, laddr, raddr sockaddr) (netfd *netFD, err error) {
    // syscall.Socket & set socket options
    var fd int
    // 调用Socket系统调用函数
    fd, err = sysSocket(family, sotype, proto)
    if err != nil {
        return nil, err
    }
    err = setDefaultSockopts(fd, family, sotype, ipv6only)
    if err != nil {
        syscall.Close(fd)
        return nil, err
    }

    netfd = newNetFD(fd, family, sotype, net)
    // 调用握手方法
    // 内部调用connect方法
    err = netfd.dial(ctx, laddr, raddr)
    if err != nil {
        netfd.Close()
        return nil, err
    }
    return netfd, nil
}


// Wrapper around the socket system call that marks the returned file
// descriptor as nonblocking and close-on-exec.
func sysSocket(family, sotype, proto int) (int, error) {
    // See ../syscall/exec_unix.go for description of ForkLock.
    syscall.ForkLock.RLock()
    // 调用Socket系统调用
    s, err := syscall.Socket(family, sotype, proto)
    if err == nil {
        syscall.CloseOnExec(s)
    }
    syscall.ForkLock.RUnlock()
    if err != nil {
        return -1, os.NewSyscallError("socket", err)
    }
    // 设置非阻塞
    if err = syscall.SetNonblock(s, true); err != nil {
        syscall.Close(s)
        return -1, os.NewSyscallError("setnonblock", err)
    }
    return s, nil
}

func newNetFD(fd, family, sotype int, net string) *netFD {
    var ret = &netFD{}
    ret.fd = fd
    ret.network = net
    ret.family = family
    ret.sotype = sotype
    ret.isStream = sotype == syscall.SOCK_STREAM
    ret.zeroReadIsEOF = sotype != syscall.SOCK_DGRAM && sotype != syscall.SOCK_RAW
    return ret
}


// if dial connection error, you need exec netFD.Close actively
func (c *netFD) dial(ctx context.Context, laddr, raddr sockaddr) (err error) {
    var lsa syscall.Sockaddr
    if laddr != nil {
        if lsa, err = laddr.sockaddr(c.family); err != nil {
            return err
        } else if lsa != nil {
            // bind local address
            // 调用bind方法
            if err = syscall.Bind(c.fd, lsa); err != nil {
                return os.NewSyscallError("bind", err)
            }
        }
    }
    var rsa syscall.Sockaddr  // remote address from the user
    var crsa syscall.Sockaddr // remote address we actually connected to
    if raddr != nil {
        if rsa, err = raddr.sockaddr(c.family); err != nil {
            return err
        }
    }
    // 调用connect()连接方法
    // remote address we actually connected to
    if crsa, err = c.connect(ctx, lsa, rsa); err != nil {
        return err
    }
    c.isConnected = true

    // Record the local and remote addresses from the actual socket.
    // Get the local address by calling Getsockname.
    // For the remote address, use
    // 1) the one returned by the connect method, if any; or
    // 2) the one from Getpeername, if it succeeds; or
    // 3) the one passed to us as the raddr parameter.
    lsa, _ = syscall.Getsockname(c.fd)
    c.localAddr = sockaddrToAddr(lsa)
    if crsa != nil {
        c.remoteAddr = sockaddrToAddr(crsa)
    } else if crsa, _ = syscall.Getpeername(c.fd); crsa != nil {
        c.remoteAddr = sockaddrToAddr(crsa)
    } else {
        c.remoteAddr = sockaddrToAddr(rsa)
    }
    return nil
}

func (c *netFD) connect(ctx context.Context, la, ra syscall.Sockaddr) (rsa syscall.Sockaddr, ret error) {
    // Do not need to call c.writing here,
    // because c is not yet accessible to user,
    // so no concurrent operations are possible.
    // 调用Connect系统调用
    switch err := syscall.Connect(c.fd, ra); err {
    case syscall.EINPROGRESS, syscall.EALREADY, syscall.EINTR:
    case nil, syscall.EISCONN:
        select {
        case <-ctx.Done():
            return nil, mapErr(ctx.Err())
        default:
        }
        return nil, nil
    case syscall.EINVAL:
        // On Solaris we can see EINVAL if the socket has
        // already been accepted and closed by the server.
        // Treat this as a successful connection--writes to
        // the socket will see EOF.  For details and a test
        // case in C see https://golang.org/issue/6828.
        if runtime.GOOS == "solaris" {
            return nil, nil
        }
        fallthrough
    default:
        return nil, os.NewSyscallError("connect", err)
    }

    // TODO: can't support interrupter now.
    // Start the "interrupter" goroutine, if this context might be canceled.
    // (The background context cannot)
    //
    // The interrupter goroutine waits for the context to be done and
    // interrupts the dial (by altering the c's write deadline, which
    // wakes up waitWrite).
    if ctx != context.Background() {
        // Wait for the interrupter goroutine to exit before returning
        // from connect.
        done := make(chan struct{})
        interruptRes := make(chan error)
        defer func() {
            close(done)
            if ctxErr := <-interruptRes; ctxErr != nil && ret == nil {
                // The interrupter goroutine called SetWriteDeadline,
                // but the connect code below had returned from
                // waitWrite already and did a successful connect (ret
                // == nil). Because we've now poisoned the connection
                // by making it unwritable, don't return a successful
                // dial. This was issue 16523.
                ret = mapErr(ctxErr)
                c.Close() // prevent a leak
            }
        }()
        go func() {
            select {
            case <-ctx.Done():
                // Force the runtime's poller to immediately give up
                // waiting for writability, unblocking waitWrite
                // below.
                c.SetWriteDeadline(aLongTimeAgo)
                interruptRes <- ctx.Err()
            case <-done:
                interruptRes <- nil
            }
        }()
    }

    c.pd = newPollDesc(c.fd)
    for {
        // Performing multiple connect system calls on a
        // non-blocking socket under Unix variants does not
        // necessarily result in earlier errors being
        // returned. Instead, once runtime-integrated network
        // poller tells us that the socket is ready, get the
        // SO_ERROR socket option to see if the connection
        // succeeded or failed. See issue 7474 for further
        // details.
        if err := c.pd.WaitWrite(ctx); err != nil {
            return nil, err
        }
        nerr, err := syscall.GetsockoptInt(c.fd, syscall.SOL_SOCKET, syscall.SO_ERROR)
        if err != nil {
            return nil, os.NewSyscallError("getsockopt", err)
        }
        switch err := syscall.Errno(nerr); err {
        case syscall.EINPROGRESS, syscall.EALREADY, syscall.EINTR:
        case syscall.EISCONN:
            return nil, nil
        case syscall.Errno(0):
            // The runtime poller can wake us up spuriously;
            // see issues 14548 and 19289. Check that we are
            // really connected; if not, wait again.
            if rsa, err := syscall.Getpeername(c.fd); err == nil {
                return rsa, nil
            }
        default:
            return nil, os.NewSyscallError("connect", err)
        }
    }
}