字节开源的netPoll多路复用器源码解析

typedef union epoll_data {
   int fd;
   //...
} epoll_data_t;

struct epoll_event {
   uint32_t events; /* Epoll events */
   epoll_data_t data; /* User data variable */
};

int epoll_create(int size);
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);

type Conn interface {
// Read reads data from the connection.
// Read can be made to time out and return an error after a fixed
// time limit; see SetDeadline and SetReadDeadline.
Read(b []byte) (n int, err error)

// Write writes data to the connection.
// Write can be made to time out and return an error after a fixed
// time limit; see SetDeadline and SetWriteDeadline.
Write(b []byte) (n int, err error)
}

type server struct {
	operator    FDOperator // ⽤来根据不同事件进⾏不同操作
	ln          Listener. 
	opts        *options // 配置相关回调接口
	onQuit      func(err error) // 退出回调
	connections sync.Map // 记录当前server上accept得到的活跃客户端连接: key=fd, value=connection
}

// FDOperator is a collection of operations on file descriptors.
type FDOperator struct {
	// epoll 监听的 fd
	FD int
    // 监听到读写事件后的回调函数
	OnRead  func(p Poll) error // accept 事件回调 
	OnWrite func(p Poll) error // 客户端 socket 写回调
	OnHup   func(p Poll) error
    
    // linkbuffer 与 socket 缓冲区之间的读写API
	Inputs   func(vs [][]byte) (rs [][]byte)
	InputAck func(n int) (err error)

	Outputs   func(vs [][]byte) (rs [][]byte, supportZeroCopy bool)
	OutputAck func(n int) (err error)

	// epoll 对象
	poll Poll
	...
}

type Poll interface {
	Wait() error
	Close() error
	Trigger() error
	Control(operator *FDOperator, event PollEvent) error
	Alloc() (operator *FDOperator)
	Free(operator *FDOperator)
}

type defaultPoll struct {
	pollArgs
	fd      int            // 监听的fd
	wop     *FDOperator    // eventfd(轻量级进程通信机制), wake epoll_wait -- 用于唤醒wait上阻塞的线程
	...
	Handler func(events []epollevent) (closed bool) // 发生感兴趣事件时，回调该接口处理这些事件
}

type pollArgs struct {
    ...
	events   []epollevent // 发送/接收感兴趣事件
	barriers []barrier。 // 用于实现分散读/集中写的向量缓冲区
}

type epollevent struct {
	events uint32  // 事件位图
	_      int32
	data   [8]byte // 注册感兴趣事件时，可以携带用户数据的指针
}

// poll_manager.go
var pollmanager *manager // 多路复用器池子管理器

func init() {
	var loops = runtime.GOMAXPROCS(0)/20 + 1
	pollmanager = &manager{}
	// 设置负载均衡器，默认采用轮询策略从epoll池中挑选空闲epoll
	pollmanager.SetLoadBalance(RoundRobin)
	// 设置epoll池的大小，同时会初始化池中的epoll对象
	pollmanager.SetNumLoops(loops)
	...
}

// poll_manager.go
func (m *manager) SetNumLoops(numLoops int) error {
	..
	// netpoll支持运行时动态调整epoll池大小,所以此处存在该分支
	// 如果我们打算缩小epoll池大小,则进入下面这个分支
	if numLoops < m.NumLoops {
		// 创建一个新的epoll池
		var polls = make([]Poll, numLoops)
		for idx := 0; idx < m.NumLoops; idx++ {
		    // 对于无需缩减的部分，直接重新指向即可
			if idx < numLoops {
				polls[idx] = m.polls[idx]
			} else {
			// 对于需要缩减的部分，直接Close关闭该多路复用器
				if err := m.polls[idx].Close(); err != nil {
					logger.Printf("NETPOLL: poller close failed: %v\n", err)
				}
			}
		}
		// 更新多路复用池管理器的相关状态
		m.NumLoops = numLoops
		m.polls = polls
		m.balance.Rebalance(m.polls)
		// 如果是动态缩容，缩容完毕后，直接返回
		return nil
	}
    
    // 进入初始化或者扩容逻辑
	m.NumLoops = numLoops
	return m.Run()
}

// poll_manager.go
// 扩容或者初始化epoll池
func (m *manager) Run() (err error) {
	defer func() {
		if err != nil {
			_ = m.Close()
		}
	}()

	// 如果是初始化epoll池，此处的polls大小应该为0
	// 如果时扩容逻辑，此处的polls大小为当前池中已有的多路复用器个数
	for idx := len(m.polls); idx < m.NumLoops; idx++ {
		var poll Poll
		// 创建一个新的多路复用器
		poll, err = openPoll()
		if err != nil {
			return
		}
		// 新创建的多路复用器追加到polls集合 
		m.polls = append(m.polls, poll)
		// 每个多路复用器绑定一个协程，不断轮询注册到该epoll上的fd事件
		go poll.Wait()
	}

	// 更新多路复用池管理器的相关状态
	m.balance.Rebalance(m.polls)
	return nil
}

// poll_default_linux.go

// 打开多路复用器
func openPoll() (Poll, error) {
	return openDefaultPoll()
}

func openDefaultPoll() (*defaultPoll, error) {
	var poll = new(defaultPoll) 
	poll.buf = make([]byte, 8)
	// 创建Epoll对象
	var p, err = EpollCreate(0)
	...
	// 保存epoll的fd
	poll.fd = p
	// eventfd是一种进程/线程通信的机制，他类似信号，不过eventfd只是一种通知机制
	// 无法承载数据（eventfd承载的数据是8个字节），他的好处是简单并且只消耗一个fd
	// 进程间通信机制: https://zhuanlan.zhihu.com/p/383395277
	var r0, _, e0 = syscall.Syscall(syscall.SYS_EVENTFD2, 0, 0, 0)
	...
	// TODO: 这几个回调接口干啥的 ?
	poll.Reset = poll.reset
	// 处理当前epoll fd上所发生的感兴趣的事件
	poll.Handler = poll.handler
	// eventFd 通信机制
	poll.wop = &FDOperator{FD: int(r0)}

	// 在epoll上注册并监听eventFd的可读事件 -- 监听r0上的可读事件
	if err = poll.Control(poll.wop, PollReadable); err != nil {
		_ = syscall.Close(poll.wop.FD)
		_ = syscall.Close(poll.fd)
		return nil, err
	}
	// 初始化FDOperator缓存
	poll.opcache = newOperatorCache()
	return poll, nil
}

// poll_default_linux.go
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)
		}
		// p.fd 就是 epoll fd
		// events 就是挂载到epoll tree上的epoll item
		// mesc 用于指定阻塞时间，是永久阻塞，还是阻塞一段时间，还是非阻塞IO
		// 等待当前epoll上发生感兴趣的事件
		n, err = EpollWait(p.fd, p.events, msec)
		if err != nil && err != syscall.EINTR {
			return err
		}
		// 如果没有发生感兴趣的事件，则将msec设置为-1，表示下一次采用永久阻塞策略来等待感兴趣的事件发生
		// 然后调用Gosched完成协程调度
		if n <= 0 {
			msec = -1
			runtime.Gosched()
			continue
		}
		msec = 0
		// 处理感兴趣的事件
		if p.Handler(p.events[:n]) {
			return nil
		}
		// we can make sure that there is no op remaining if Handler finished
		p.opcache.free()
	}
}

// poll_default_linux.go
// 当epoll上有感兴趣的事件发生的时候,调用该函数进行处理
func (p *defaultPoll) handler(events []epollevent) (closed bool) {
	var triggerRead, triggerWrite, triggerHup, triggerError bool
	var err error
	// 遍历所有感兴趣的事件
	for i := range events {
		// epollevent.data保存的是与之关联的FDOperator对象
		operator := p.getOperator(0, unsafe.Pointer(&events[i].data))
		if operator == nil || !operator.do() {
			continue
		}

		var totalRead int
		// 判断当前发生了什么事件
		evt := events[i].events
		triggerRead = evt&syscall.EPOLLIN != 0
		triggerWrite = evt&syscall.EPOLLOUT != 0
		triggerHup = evt&(syscall.EPOLLHUP|syscall.EPOLLRDHUP) != 0
		triggerError = evt&syscall.EPOLLERR != 0

		// trigger or exit gracefully
		// 是否是eventFD可读事件发生了
		if operator.FD == p.wop.FD {
			// must clean trigger first
			// 从eventFD中读取数据到buf中
			syscall.Read(p.wop.FD, p.buf)
			atomic.StoreUint32(&p.trigger, 0)
			// if closed & exit
			// 说明接收到了关闭信号，那么就关闭当前epoll
			if p.buf[0] > 0 {
			    // 关闭eventFD
				syscall.Close(p.wop.FD)
				// 关闭epoll fd
				syscall.Close(p.fd)
				operator.done()
				return true
			}
			operator.done()
			continue
		}
		// 发生了可读事件
		if triggerRead {
			// 如果FDOperator上的OnRead回调接口不为空，说明发生的是客户端的accept事件
			if operator.OnRead != nil {
				// 调用OnRead来接收并处理客户端连接
				operator.OnRead(p)
		    // 否则说明发生的是某个客户端连接上的可读事件		
			} else if operator.Inputs != nil {
				// 每个poll对象会关联一个barriers结构，该结构用于实现分散读取与集中写入的系统调用
				// 每个poll对象还会关联一个LinkBuffer对象，作为读写数据缓冲区
				// 此处是从LinkBuffer中分配出一块空闲内存
				var bs = operator.Inputs(p.barriers[i].bs)
				if len(bs) > 0 {
					// 读取数据到bs缓存区中
					var n, err = ioread(operator.FD, bs, p.barriers[i].ivs)
					// 推动读指针，让写入缓冲区的数据对消费者可见，同时调用用户注册的OnRequest回调接口，处理读数据
					operator.InputAck(n)
					...
				}
			} else {
				logger.Printf("NETPOLL: operator has critical problem! event=%d operator=%v", evt, operator)
			}
		}
		// 其他感兴趣事件的触发此处暂时不展开
		...
}

// sys_epoll_linux_arm64.go
type epollevent struct {
	events uint32  // events：表示要监听的事件类型，如可读、可写等。这是一个位掩码，可以设置多个事件类型，例如 EPOLLIN 表示可读事件，EPOLLOUT 表示可写事件。
	_      int32
	data   [8]byte // 可以携带用户数据。这里的用户数据通常是一个指针，指向与文件描述符关联的对象或其他相关数据。
}

func EpollCreate(flag int) (fd int, err error) {
	var r0 uintptr
	// 执行epoll_create系统调用
	r0, _, err = syscall.RawSyscall(syscall.SYS_EPOLL_CREATE1, uintptr(flag), 0, 0)
	if err == syscall.Errno(0) {
		err = nil
	}
	return int(r0), err
}

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
}

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_PWAIT, uintptr(epfd), uintptr(_p0), uintptr(len(events)), 0, 0, 0)
	} else {
		r0, _, err = syscall.Syscall6(syscall.SYS_EPOLL_PWAIT, uintptr(epfd), uintptr(_p0), uintptr(len(events)), uintptr(msec), 0, 0)
	}
	if err == syscall.Errno(0) {
		err = nil
	}
	return int(r0), err
}

// Control implements Poll.
func (p *defaultPoll) Control(operator *FDOperator, event PollEvent) error {
	var op int
	// TODO: evt.data = operator
	var evt epollevent
	// 将epollevent对象的data指针指向传入的FDOperator对象
	p.setOperator(unsafe.Pointer(&evt.data), operator)
	// 根据监听的事件类型,更新事件位图
	switch event {
	case PollReadable: // server accept a new connection and wait read
		operator.inuse()
		op, evt.events = syscall.EPOLL_CTL_ADD, syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLERR
	case PollWritable: // client create a new connection and wait connect finished
		operator.inuse()
		op, evt.events = syscall.EPOLL_CTL_ADD, EPOLLET|syscall.EPOLLOUT|syscall.EPOLLRDHUP|syscall.EPOLLERR
	case PollDetach: // deregister
		p.delOperator(operator)
		op, evt.events = syscall.EPOLL_CTL_DEL, syscall.EPOLLIN|syscall.EPOLLOUT|syscall.EPOLLRDHUP|syscall.EPOLLERR
	case PollR2RW: // connection wait read/write
		op, evt.events = syscall.EPOLL_CTL_MOD, syscall.EPOLLIN|syscall.EPOLLOUT|syscall.EPOLLRDHUP|syscall.EPOLLERR
	case PollRW2R: // connection wait read
		op, evt.events = syscall.EPOLL_CTL_MOD, syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLERR
	}
	// 完成监听事件信息注册
	return EpollCtl(p.fd, op, operator.FD, &evt)
}

// 1. OnRequest: 有可读数据时回调该接口
var OnRequest = func(ctx context.Context, connection netpoll.Connection) error { return nil }

// 2. OnPrepare: 客户端连接建立完毕后,回调该接口
var OnPrepare = func(connection netpoll.Connection) context.Context { return nil }

func main() {
	// 1. 建立连接
	listen, err := net.Listen("tcp", ":1234")
	if err != nil {
		return
	}

	// 2. 创建eventLoop
	eventLoop, _ := netpoll.NewEventLoop(OnRequest, netpoll.WithOnPrepare(OnPrepare), netpoll.WithReadTimeout(time.Second))

	// 3. 启动服务
	eventLoop.Serve(listen)
}

// Serve implements EventLoop.
func (evl *eventLoop) Serve(ln net.Listener) error {
	// 将原生的listener对象转换为netpoll包装后的Listener对象
	npln, err := ConvertListener(ln)
	if err != nil {
		return err
	}
	evl.Lock()
	// 创建新的server对象
	evl.svr = newServer(npln,
		// opts对象保存了相关事件回调
		evl.opts,
		// 退出事件回调监听函数
		evl.quit)
	// 启动回调
	evl.svr.Run()
	evl.Unlock()
    
    // 监听到停止信号后，从此处返回
	err = evl.waitQuit()
	// ensure evl will not be finalized until Serve returns
	runtime.SetFinalizer(evl, nil)
	return err
}

// Run this server.
func (s *server) Run() (err error) {
	// 当前FDOperator对象封装的是server socket套接字对象
	s.operator = FDOperator{
		FD:     s.ln.Fd(), //  服务端Socket监听器
		OnRead: s.OnRead,  //  可读事件发生
		OnHup:  s.OnHup,   //   挂断事件发生
	}
	// 挑选一个空闲的多路复用器
	s.operator.poll = pollmanager.Pick()
	// 监听服务端套接字上的可读事件
	err = s.operator.Control(PollReadable)
	if err != nil {
		// 错误退出时,回调该方法
		s.onQuit(err)
	}
	return err
}

// poll_default_linux.go
// 当epoll上有感兴趣的事件发生的时候,调用该函数进行处理
func (p *defaultPoll) handler(events []epollevent) (closed bool) {
	var triggerRead, triggerWrite, triggerHup, triggerError bool
	var err error
	// 遍历所有感兴趣的事件
	for i := range events {
		// epollevent.data保存的是与之关联的FDOperator对象
		operator := p.getOperator(0, unsafe.Pointer(&events[i].data))
		...
		var totalRead int
		// 判断当前发生了什么事件
		evt := events[i].events
		triggerRead = evt&syscall.EPOLLIN != 0
		triggerWrite = evt&syscall.EPOLLOUT != 0
		triggerHup = evt&(syscall.EPOLLHUP|syscall.EPOLLRDHUP) != 0
		triggerError = evt&syscall.EPOLLERR != 0
		...
		// 发生了可读事件
		if triggerRead {
			// 如果FDOperator上的OnRead回调接口不为空，说明发生的是客户端的accept事件
			if operator.OnRead != nil {
				// 调用OnRead来接收并处理客户端连接
				operator.OnRead(p)
		    // 否则说明发生的是某个客户端连接上的可读事件		
			} else if operator.Inputs != nil {
				// 每个poll对象会关联一个barriers结构，该结构用于实现分散读取与集中写入的系统调用
				// 每个poll对象还会关联一个LinkBuffer对象，作为读写数据缓冲区
				// 此处是从LinkBuffer中分配出一块空闲内存
				var bs = operator.Inputs(p.barriers[i].bs)
				if len(bs) > 0 {
					// 读取数据到bs缓存区中
					var n, err = ioread(operator.FD, bs, p.barriers[i].ivs)
					// 推动读指针，让写入缓冲区的数据对消费者可见，同时调用用户注册的OnRequest回调接口，处理读数据
					operator.InputAck(n)
					...
				}
			} else {
				logger.Printf("NETPOLL: operator has critical problem! event=%d operator=%v", evt, operator)
			}
		}
		// 其他感兴趣事件的触发此处暂时不展开
		...
}

// OnRead implements FDOperator.
func (s *server) OnRead(p Poll) error {
	// 获取客户端连接
	conn, err := s.ln.Accept()
	...
	// 包装一下原生的conn连接
	var connection = &connection{}
	// 初始化一下连接
	connection.init(conn.(Conn),
		// 初始化完毕后,回调用户注册进来的prepare接口
		s.opts)
	// 连接不活跃,直接返回
	if !connection.IsActive() {
		return nil
	}
	// 返回客户端连接套接字对应的文件描述符
	var fd = conn.(Conn).Fd()
	// 添加关闭回调接口 --- netpoll回调接口这里采用的是回调链的形式,可以添加多个回调接口
	connection.AddCloseCallback(func(connection Connection) error {
		// 当前连接关闭时,将自己从server连接集合中移除
		s.connections.Delete(fd)
		return nil
	})
	// 在server对象中保存 < fd , 已打开连接 >
	s.connections.Store(fd, connection)

	// 调用连接建立接口
	connection.onConnect()
	return nil
}

// init initialize the connection with options
func (c *connection) init(conn Conn, opts *options) (err error) {
	// init buffer, barrier, finalizer
	c.readTrigger = make(chan error, 1)
	c.writeTrigger = make(chan error, 1)
	// 初始化LinkBuffer相关数据结构
	c.bookSize, c.maxSize = pagesize, pagesize
	c.inputBuffer, c.outputBuffer = NewLinkBuffer(pagesize), NewLinkBuffer()
	c.outputBarrier = barrierPool.Get().(*barrier) // 用于聚集读写的缓冲区

	// 初始化
	c.initNetFD(conn)  // 确保conn是被netpoll包装后的netFD类型
	c.initFDOperator() // 初始化FDOperator
	c.initFinalizer()  // 添加close回调函数

	// 将客户端连接套接字设置为非阻塞模式
	syscall.SetNonblock(c.fd, true)
	// enable TCP_NODELAY by default
	switch c.network {
	case "tcp", "tcp4", "tcp6":
		// 禁用 Nagle 算法
		setTCPNoDelay(c.fd, true)
	}
	// 启用零拷贝传输的 TCP Socket 选项 和 阻塞超时时间
	if setZeroCopy(c.fd) == nil && setBlockZeroCopySend(c.fd, defaultZeroCopyTimeoutSec, 0) == nil {
		c.supportZeroCopy = true
	}

	// connection initialized and prepare options
	// 设置相关回调接口,在poll上注册对当前connection可读事件的监听
	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() {
	// 通过负载均衡器挑选一个可用的poll
	poll := pollmanager.Pick()
	// 从opcache中分配一个可用的poll对象
	op := poll.Alloc()
	// 拿到当前客户端连接对应的socket文件描述符
	op.FD = c.fd
	// 回调接口初始化 -- 注意OnRead回调被设置为了nil
	op.OnRead, op.OnWrite, op.OnHup = nil, nil, c.onHup
	op.Inputs, op.InputAck = c.inputs, c.inputAck
	op.Outputs, op.OutputAck = c.outputs, c.outputAck
	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 {
		// 将用户通过options设置的回调接口都赋值给当前accept得到的客户端连接
		c.SetOnConnect(opts.onConnect)
		c.SetOnRequest(opts.onRequest)
		c.SetReadTimeout(opts.readTimeout)
		c.SetWriteTimeout(opts.writeTimeout)
		c.SetIdleTimeout(opts.idleTimeout)

		// calling prepare first and then register.
		// 如果我们指定了onPrepare回调，此处会执行回调
		if opts.onPrepare != nil {
			c.ctx = opts.onPrepare(c)
		}
	}
	// 初始化连接上下文
	if c.ctx == nil {
		c.ctx = context.Background()
	}
	// prepare may close the connection.
	if c.IsActive() {
		// 在当前poll上注册对当前客户端connection的读事件监听
		return c.register()
	}
	return nil
}

// register only use for connection register into poll.
func (c *connection) register() (err error) {
	err = c.operator.Control(PollReadable)
	... 
	return nil
}

// onConnect is responsible for executing onRequest if there is new data coming after onConnect callback finished.
func (c *connection) onConnect() {
	// 获取用户设置的OnConnect回调和OnRequest回调接口 --- 如果没有设置OnConnect回调，此处直接返回
	var onConnect, _ = c.onConnectCallback.Load().(OnConnect)
	if onConnect == nil {
		return
	}
	var onRequest, _ = c.onRequestCallback.Load().(OnRequest)
	var connected int32
	c.onProcess(
		// 第一个回调函数用于判断当前是否连接此刻是否可被处理
		func(c *connection) bool {
			// 在当前客户端连接初始化完毕后,会在onConnect函数中回调一次客户端提供的OnConnect接口
			// 此处通过标记确保只会调用一次OnConnect函数
			if atomic.LoadInt32(&connected) == 0 {
				return true
			}
			// check for onRequest
			return onRequest != nil &&
				// 存在可读数据
				c.Reader().Len() > 0
		},
		// 第二个回调函数会在第一个回调函数返回true的前提下,进行处理
		func(c *connection) {
			// 回调OnConnect函数
			if atomic.CompareAndSwapInt32(&connected, 0, 1) {
				c.ctx = onConnect(c.ctx, c)
				return
			}
			// 处理可读数据,回调用户提供的回调函数
			if onRequest != nil {
				_ = onRequest(c.ctx, c)
			}
		},
	)
}

// 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) {
	...
	// 准备任务
	var task = func() {
		// 如果当前任务可执行,确保至少被执行过一次
		if isProcessable(c) {
			process(c)
		}
		// 死循环处理任务,直到接收到关闭信号或者任务不再可处理
		var closedBy who
		for {
			closedBy = c.status(closing)
			// close by user or no processable
			if closedBy == user || !isProcessable(c) {
				break
			}
			process(c)
		}
		...
		return
	}

	// 异步跑这个任务 --- gopool.CtxGo 字节开源的协程池
	runTask(c.ctx, task)
	return true
}

// poll_default_linux.go
// 当epoll上有感兴趣的事件发生的时候,调用该函数进行处理
func (p *defaultPoll) handler(events []epollevent) (closed bool) {
	var triggerRead, triggerWrite, triggerHup, triggerError bool
	var err error
	// 遍历所有感兴趣的事件
	for i := range events {
		// epollevent.data保存的是与之关联的FDOperator对象
		operator := p.getOperator(0, unsafe.Pointer(&events[i].data))
		...
		var totalRead int
		// 判断当前发生了什么事件
		evt := events[i].events
		triggerRead = evt&syscall.EPOLLIN != 0
		triggerWrite = evt&syscall.EPOLLOUT != 0
		triggerHup = evt&(syscall.EPOLLHUP|syscall.EPOLLRDHUP) != 0
		triggerError = evt&syscall.EPOLLERR != 0
		...
		// 发生了可读事件
		if triggerRead {
			// 如果FDOperator上的OnRead回调接口不为空，说明发生的是客户端的accept事件
			if operator.OnRead != nil {
				// 调用OnRead来接收并处理客户端连接
				operator.OnRead(p)
		    // 否则说明发生的是某个客户端连接上的可读事件		
			} else if operator.Inputs != nil {
				// 每个poll对象会关联一个barriers结构，该结构用于实现分散读取与集中写入的系统调用
				// 每个poll对象还会关联一个LinkBuffer对象，作为读写数据缓冲区
				// 此处是从LinkBuffer中分配出一块空闲内存
				var bs = operator.Inputs(p.barriers[i].bs)
				if len(bs) > 0 {
					// 读取数据到bs缓存区中
					var n, err = ioread(operator.FD, bs, p.barriers[i].ivs)
					// 推动读指针，让写入缓冲区的数据对消费者可见，同时调用用户注册的OnRequest回调接口，处理读数据
					operator.InputAck(n)
					...
				}
			} else {
				logger.Printf("NETPOLL: operator has critical problem! event=%d operator=%v", evt, operator)
			}
		}
		// 其他感兴趣事件的触发此处暂时不展开
		...
}

func (c *connection) initFDOperator() {
    ...
	op.Inputs, op.InputAck = c.inputs, c.inputAck
	op.Outputs, op.OutputAck = c.outputs, c.outputAck
	...
}

// inputs implements FDOperator.
func (c *connection) inputs(vs [][]byte) (rs [][]byte) {
	vs[0] = c.inputBuffer.book(c.bookSize, c.maxSize)
	return vs[:1]
}

// inputAck implements FDOperator.
func (c *connection) inputAck(n int) (err error) {
    ...
    // 提交预留内存,提交后,用户便可以读取这部分内存数据了
	length, _ := c.inputBuffer.bookAck(n)
	...
	var needTrigger = true
	// 从协程池中取出一个空闲协程来处理当前连接上的可读数据
	if length == n { // first start onRequest
		needTrigger = c.onRequest() // 返回值表示是否读取完毕了所有需要的数据, 如果返回false，说明读完了，否则说明没有读完
	}
	// 单开协程处理客户端连接上的可读数据时,可能在回调用户OnRequest接口时,调用读数据接口从而阻塞等待数据准备就绪
	// 此处当数据就绪时，会唤醒对应的协程
	if needTrigger && length >= int(atomic.LoadInt64(&c.waitReadSize)) {
		c.triggerRead(nil)
	}
	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(
		// 第一个回调函数用于判断当前连接是否活跃并且还有未读取数据
		func(c *connection) bool {
			return c.Reader().Len() > 0
		},
		// 第二个回调才是真正将请求交给用户回调来处理
		func(c *connection) {
			_ = onRequest(c.ctx, c)
		},
	)
	// if not processed, should trigger read
	return !processed
}

// ReadString implements Connection.
func (c *connection) ReadString(n int) (s string, err error) {
	if err = c.waitRead(n); err != nil {
		return s, err
	}
	return c.inputBuffer.ReadString(n)
}

// ReadBinary implements Connection.
func (c *connection) ReadBinary(n int) (p []byte, err error) {
	if err = c.waitRead(n); err != nil {
		return p, err
	}
	return c.inputBuffer.ReadBinary(n)
}

// Next implements Connection.
func (c *connection) Next(n int) (p []byte, err error) {
	if err = c.waitRead(n); err != nil {
		return p, err
	}
	return c.inputBuffer.Next(n)
}

// waitRead will wait full n bytes.
func (c *connection) waitRead(n int) (err error) {
	// 如果当前可读数据大于需要的了,直接返回,无需等待
	if n <= c.inputBuffer.Len() {
		return nil
	}
	// 存储自己希望读取的数据量
	atomic.StoreInt64(&c.waitReadSize, int64(n))
	// 返回时,清空变量
	defer atomic.StoreInt64(&c.waitReadSize, 0)
	// 如果设置了读超时属性,就有限期等待,直到数据就绪
	if c.readTimeout > 0 {
		return c.waitReadWithTimeout(n)
	}
	// wait full n
	// 否则无限期等待,直到数据准备就绪
	for c.inputBuffer.Len() < n {
		switch c.status(closing) {
		case poller:
			return Exception(ErrEOF, "wait read")
		case user:
			return Exception(ErrConnClosed, "wait read")
		default:
		    // 等待接收数据就绪的信号
			err = <-c.readTrigger
			if err != nil {
				return err
			}
		}
	}
	return nil
}

// waitReadWithTimeout will wait full n bytes or until timeout.
// 有限期等待
func (c *connection) waitReadWithTimeout(n int) (err error) {
	// set read timeout
	if c.readTimer == nil {
		c.readTimer = time.NewTimer(c.readTimeout)
	} else {
		c.readTimer.Reset(c.readTimeout)
	}

	for c.inputBuffer.Len() < n {
		switch c.status(closing) {
		case poller:
			// cannot return directly, stop timer first!
			err = Exception(ErrEOF, "wait read")
			goto RET
		case user:
			// cannot return directly, stop timer first!
			err = Exception(ErrConnClosed, "wait read")
			goto RET
		default:
			select {
			case <-c.readTimer.C:
				// double check if there is enough data to be read
				if c.inputBuffer.Len() >= n {
					return nil
				}
				return Exception(ErrReadTimeout, c.remoteAddr.String())
			case err = <-c.readTrigger:
				if err != nil {
					return err
				}
				continue
			}
		}
	}
RET:
	// clean timer.C
	if !c.readTimer.Stop() {
		<-c.readTimer.C
	}
	return err
}

func main() {
	// 1. 建立连接
	dialer := netpoll.NewDialer()
	conn, _ := dialer.DialConnection("tcp", ":1234", time.Second)
	var reader, writer = conn.Reader(), conn.Writer()

	// 2. 写数据
	write_data := []byte("hello world")
	alloc, _ := writer.Malloc(len(write_data))
	copy(alloc, write_data) // write data
	writer.Flush()

	// 3. 读数据
	buf, _ := reader.Next(reader.Len())
	fmt.Println("服务端响应的数据:" + string(buf))
	reader.Release()
}

func (d *dialer) DialConnection(network, address string, timeout time.Duration) (connection Connection, err error) {
    ... 
	switch network {
	case "tcp", "tcp4", "tcp6":
	     // 走tcp连接
		return d.dialTCP(ctx, network, address)
    ...
	}
}

func (d *dialer) dialTCP(ctx context.Context, network, address string) (connection *TCPConnection, err error) {
    ...
    connection, err = DialTCP(ctx, "tcp", nil, tcpAddr)
	...
	return nil, firstErr
}

func DialTCP(ctx context.Context, network string, laddr, raddr *TCPAddr) (*TCPConnection, error) {
    ...
	sd := &sysDialer{network: network, address: raddr.String()}
	c, err := sd.dialTCP(ctx, laddr, raddr)
	...
	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")
	...
	return newTCPConnection(conn)
}

func internetSocket(ctx context.Context, net string, laddr, raddr sockaddr, sotype, proto int, mode string) (conn *netFD, err error) {
    ... 
	return socket(ctx, net, family, sotype, proto, ipv6only, laddr, raddr)
}

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
	}
	...
	// 包装客户端socket fd 为netFD
	netfd = newNetFD(fd, family, sotype, net)
	// 建立与server的连接
	err = netfd.dial(ctx, laddr, raddr)
	...
	return netfd, nil
}

func (c *netFD) dial(ctx context.Context, laddr, raddr sockaddr) (err error) {
    ...
	// 连接server
	if crsa, err = c.connect(ctx, lsa, rsa); err != nil {
		return err
	}
	...
	return nil
}

func (c *netFD) connect(ctx context.Context, la, ra syscall.Sockaddr) (rsa syscall.Sockaddr, retErr error) {
	// 系统调用connect,连接server
	syscall.Connect(c.fd, ra)
	...
	// 为当前client socket创建一个新的FDOperator
	c.pd = newPollDesc(c.fd)
	for {
		// 等待直到当前client socket可写为止
		if err := c.pd.WaitWrite(ctx); err != nil {
			return nil, err
		}
		switch err := syscall.Errno(nerr); err {
		...
		// 如果没有错误发生,直接返回Sockaddr
		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
			}
		...
		}
	}
}

func newPollDesc(fd int) *pollDesc {
	pd := &pollDesc{}
	poll := pollmanager.Pick()
	pd.operator = &FDOperator{
		poll:    poll,
		FD:      fd,
		OnWrite: pd.onwrite, // 设置OnWrite回调接口
		OnHup:   pd.onhup,
	}
	pd.writeTrigger = make(chan struct{})
	pd.closeTrigger = make(chan struct{})
	return pd
}

// WaitWrite .
func (pd *pollDesc) WaitWrite(ctx context.Context) (err error) {
	if pd.operator.isUnused() {
		// add ET|Write|Hup
		// 在当前连接绑定的poll上注册等待可写事件
		if err = pd.operator.Control(PollWritable); err != nil {
			logger.Printf("NETPOLL: pollDesc register operator failed: %v", err)
			return err
		}
	}
    // 等待直到接收到终止信号或者可写事件(当前client socket缓冲区为空)  
	select {
	case <-pd.closeTrigger: // triggered by poller
		// no need to detach, since poller has done it in OnHup.
		return Exception(ErrConnClosed, "by peer")
	case <-pd.writeTrigger: // triggered by poller
		err = nil
	case <-ctx.Done(): // triggered by ctx
		pd.detach()
		pd.operator.unused()
		err = mapErr(ctx.Err())
	}
	// double check close trigger
	// 如果没有接收到停止信号,此处就直接返回
	select {
	case <-pd.closeTrigger:
		return Exception(ErrConnClosed, "by peer")
	default:
		return err
	}
}

func (c *connection) Malloc(n int) (buf []byte, err error) {
	return c.outputBuffer.Malloc(n)
}

func (c *connection) Flush() error {
    ...
    // 刷新写缓冲区中的数据,让其对外可见 
	c.outputBuffer.Flush()
	// 将数据写入内核socket缓冲区
	return c.flush()
}

func (c *connection) flush() error {
	...
	// netpoll采用聚集写,所以第一步是将写缓冲区中的数据都读取到写缓冲区向量数组中
	var bs = c.outputBuffer.GetBytes(c.outputBarrier.bs)
	// 将数据都写入内核socket缓冲区中
	var n, err = sendmsg(c.fd, bs, c.outputBarrier.ivs, false && c.supportZeroCopy)
	...
	// 如果写入了部分数据,则释放掉这部分内存中间
	if n > 0 {
		err = c.outputBuffer.Skip(n)
		c.outputBuffer.Release()
		if err != nil {
			return Exception(err, "when flush")
		}
	}
	// 如果所有数据都已经成功写入内核socket缓冲区中,则直接返回
	if c.outputBuffer.IsEmpty() {
		return nil
	}
	// 可能是因为socket缓冲区满了，导致还有一部分数据没写完
	// 此处注册对可写事件的监听
	err = c.operator.Control(PollR2RW)
	...
	// 等待直到可写才会返回
	return c.waitFlush()
}

func (c *connection) waitFlush() (err error) {
    // 如果我们没有设置写超时事件，则进行无限期等待
	if c.writeTimeout == 0 {
		select {
		case err = <-c.writeTrigger:
		}
		return err
	}

	// 如果我们设置了写超时事件，则执行有限期等待
	if c.writeTimer == nil {
		c.writeTimer = time.NewTimer(c.writeTimeout)
	} else {
		c.writeTimer.Reset(c.writeTimeout)
	}

	select {
	case err = <-c.writeTrigger:
		if !c.writeTimer.Stop() { // clean timer
			<-c.writeTimer.C
		}
		return err
	case <-c.writeTimer.C:
		select {
		// try fetch writeTrigger if both cases fires
		case err = <-c.writeTrigger:
			return err
		default:
		}
		// if timeout, remove write event from poller
		// we cannot flush it again, since we don't if the poller is still process outputBuffer
		c.operator.Control(PollRW2R)
		return Exception(ErrWriteTimeout, c.remoteAddr.String())
	}
}

// 当感兴趣事件发生的时候,调用该函数进行处理
func (p *defaultPoll) handler(events []epollevent) (closed bool) {
	var triggerRead, triggerWrite, triggerHup, triggerError bool
	var err error
	// 遍历所有感兴趣的事件
	for i := range events {
	    ...
		// 触发写事件
		if triggerWrite {
			// 处理client socket可写事件
			if operator.OnWrite != nil {
				operator.OnWrite(p)
			} else if operator.Outputs != nil {
				// 处理服务端socket可写事件
				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 = iosend(operator.FD, bs, p.barriers[i].ivs, false && supportZeroCopy)
					operator.OutputAck(n)
					if err != nil {
						p.appendHup(operator)
						continue
					}
				}
			} else {
				logger.Printf("NETPOLL: operator has critical problem! event=%d operator=%v", evt, operator)
			}
		}
		operator.done()
	}
	...
}

func (pd *pollDesc) onwrite(p Poll) error {
	select {
	case <-pd.writeTrigger:
	default:
		pd.detach()
		close(pd.writeTrigger)
	}
	return nil
}

// outputs implements FDOperator.
func (c *connection) outputs(vs [][]byte) (rs [][]byte, supportZeroCopy bool) {
	// 如果写缓冲区为空
	if c.outputBuffer.IsEmpty() {
		// 移除对当前fd上可写事件的监听
		c.rw2r()
		return rs, c.supportZeroCopy
	}
	// 读取数据到rs中
	rs = c.outputBuffer.GetBytes(vs)
	return rs, c.supportZeroCopy
}

// 不再监听当前FD上的可写事件
func (c *connection) rw2r() {
	c.operator.Control(PollRW2R)
	c.triggerWrite(nil)  // 唤醒等到可写事件的线程
}

	var n, err = iosend(operator.FD, bs, p.barriers[i].ivs, false && supportZeroCopy)

func (c *connection) outputAck(n int) (err error) {
    // 将已经用完的部分内存回收掉 
	if n > 0 {
		c.outputBuffer.Skip(n)
		c.outputBuffer.Release()
	}
	// 如果此时发现所有待写入数据都写入完毕了,那么就移除对当前fd上可写事件的监听
	if c.outputBuffer.IsEmpty() {
		c.rw2r()
	}
	return nil
}