深入Go网络模块：netpoll机制与I/O多路复用源码解析

graph TB
    subgraph "Go 网络模块架构"
        A[用户代码] --> B[net包接口层]
        
        subgraph "net包核心组件"
            C[Conn接口]
            D[Listener接口]
            E[netFD网络文件描述符]
            F[Dial/Listen函数]
        end
        
        B --> C
        B --> D
        B --> E
        B --> F
        
        subgraph "internal/poll包"
            G[poll.FD]
            H[pollDesc]
            I[fdMutex]
            J[读写操作封装]
        end
        
        E --> G
        G --> H
        G --> I
        G --> J
        
        subgraph "runtime网络轮询器"
            K[netpoll核心]
            L[pollDesc管理]
            M[事件通知机制]
        end
        
        H --> K
        K --> L
        K --> M
        
        subgraph "平台特定实现"
            N[Linux: epoll]
            O[macOS/BSD: kqueue]
            P[Windows: IOCP]
            Q[Solaris: event ports]
        end
        
        K --> N
        K --> O
        K --> P
        K --> Q
        
        subgraph "系统调用层"
            R[socket系统调用]
            S[I/O多路复用系统调用]
            T[网络协议栈]
        end
        
        N --> S
        O --> S
        P --> S
        Q --> S
        S --> T
        E --> R
    end
    
    style A fill:#e1f5fe
    style B fill:#f3e5f5
    style G fill:#e8f5e8
    style K fill:#fff3e0
    style N fill:#fce4ec

sequenceDiagram
    participant APP as 应用程序
    participant NET as net包
    participant POLL as internal/poll
    participant NETPOLL as runtime netpoll
    participant EPOLL as epoll/kqueue
    participant SCHED as 调度器

    APP->>NET: net.Dial("tcp", addr)
    NET->>NET: 创建socket
    NET->>POLL: 包装为poll.FD
    POLL->>NETPOLL: 注册到netpoll
    NETPOLL->>EPOLL: epoll_ctl(ADD)
    
    APP->>NET: conn.Write(data)
    NET->>POLL: fd.Write()
    alt socket缓冲区有空间
        POLL->>POLL: 直接写入
        POLL->>APP: 返回写入字节数
    else socket缓冲区满
        POLL->>NETPOLL: 等待可写事件
        NETPOLL->>SCHED: gopark当前goroutine
        SCHED->>SCHED: 调度其他goroutine
        
        Note over EPOLL: 等待socket可写
        EPOLL->>NETPOLL: 可写事件就绪
        NETPOLL->>SCHED: 唤醒等待的goroutine
        SCHED->>POLL: 恢复写操作
        POLL->>APP: 返回写入字节数
    end
    
    par 并发读操作
        APP->>NET: conn.Read(buf)
        NET->>POLL: fd.Read()
        alt socket有数据
            POLL->>APP: 返回读取数据
        else socket无数据
            POLL->>NETPOLL: 等待可读事件
            NETPOLL->>SCHED: gopark当前goroutine
            
            Note over EPOLL: 等待数据到达
            EPOLL->>NETPOLL: 可读事件就绪
            NETPOLL->>SCHED: 唤醒等待的goroutine
            SCHED->>POLL: 恢复读操作
            POLL->>APP: 返回读取数据
        end
    end

sequenceDiagram
    participant CLIENT as 客户端
    participant SERVER as 服务端
    participant LISTENER as Listener
    participant NETPOLL as netpoll
    participant EPOLL as epoll

    SERVER->>LISTENER: net.Listen("tcp", ":8080")
    LISTENER->>LISTENER: 创建监听socket
    LISTENER->>NETPOLL: 注册监听socket
    NETPOLL->>EPOLL: epoll_ctl(ADD, listen_fd)
    
    loop 接受连接循环
        SERVER->>LISTENER: listener.Accept()
        LISTENER->>NETPOLL: 等待连接事件
        NETPOLL->>NETPOLL: gopark等待
        
        CLIENT->>CLIENT: net.Dial("tcp", addr)
        CLIENT->>CLIENT: 发起TCP连接
        
        Note over CLIENT,SERVER: TCP三次握手
        CLIENT->>SERVER: SYN
        SERVER->>CLIENT: SYN-ACK
        CLIENT->>SERVER: ACK
        
        EPOLL->>NETPOLL: 监听socket可读
        NETPOLL->>NETPOLL: 唤醒Accept goroutine
        LISTENER->>LISTENER: accept()系统调用
        LISTENER->>SERVER: 返回新连接
        
        par 处理连接
            SERVER->>SERVER: go handleConn(conn)
            SERVER->>SERVER: 处理请求
        end
    end

sequenceDiagram
    participant SYSMON as sysmon
    participant NETPOLL as netpoll
    participant EPOLL as epoll/kqueue
    participant SCHED as 调度器
    participant G as Goroutine

    SYSMON->>NETPOLL: 定期检查网络事件
    NETPOLL->>EPOLL: epoll_wait(timeout=0)
    
    alt 有就绪事件
        EPOLL->>NETPOLL: 返回就绪fd列表
        loop 处理每个就绪fd
            NETPOLL->>NETPOLL: 查找pollDesc
            NETPOLL->>NETPOLL: 检查事件类型
            
            alt 可读事件
                NETPOLL->>G: 唤醒等待读的goroutine
                G->>SCHED: 加入可运行队列
            end
            
            alt 可写事件
                NETPOLL->>G: 唤醒等待写的goroutine
                G->>SCHED: 加入可运行队列
            end
            
            alt 错误事件
                NETPOLL->>G: 唤醒所有等待的goroutine
                G->>G: 处理错误
            end
        end
        
        NETPOLL->>SCHED: 返回就绪goroutine列表
        SCHED->>SCHED: 调度就绪goroutine
        
    else 无就绪事件
        NETPOLL->>SYSMON: 返回空列表
    end

sequenceDiagram
    participant APP as 应用程序
    participant FD as poll.FD
    participant PD as pollDesc
    participant RUNTIME as runtime
    participant EPOLL as epoll

    APP->>FD: Read/Write操作
    FD->>FD: 尝试非阻塞I/O
    
    alt I/O立即完成
        FD->>APP: 返回结果
    else I/O会阻塞(EAGAIN)
        FD->>PD: 准备等待I/O事件
        PD->>RUNTIME: poll_runtime_pollWait
        
        RUNTIME->>RUNTIME: 检查pollDesc状态
        alt 事件已就绪
            RUNTIME->>FD: 立即返回
        else 需要等待
            RUNTIME->>RUNTIME: gopark当前goroutine
            RUNTIME->>PD: 设置等待状态
            
            Note over EPOLL: 等待I/O事件
            
            EPOLL->>RUNTIME: I/O事件就绪
            RUNTIME->>RUNTIME: netpollready()
            RUNTIME->>RUNTIME: 唤醒等待的goroutine
            RUNTIME->>FD: 恢复I/O操作
        end
        
        FD->>FD: 重试I/O操作
        FD->>APP: 返回结果
    end

flowchart TD
    A[用户调用 conn.Read/Write] --> B[netFD.Read/Write]
    B --> C[poll.FD.Read/Write]
    C --> D{是否可立即完成?}
    
    D -->|是| E[直接系统调用]
    E --> F[返回结果]
    
    D -->|否| G[准备轮询描述符]
    G --> H[runtime_pollReset]
    H --> I[尝试非阻塞I/O]
    I --> J{操作是否完成?}
    
    J -->|是| F
    J -->|否| K[runtime_pollWait]
    K --> L[goroutine进入等待状态]
    L --> M[netpoll检测到事件]
    M --> N[唤醒等待的goroutine]
    N --> O[重新尝试I/O操作]
    O --> F
    
    style A fill:#e1f5fe
    style K fill:#fff3e0
    style M fill:#fce4ec

graph TB
    subgraph "netpoll 架构"
        A[netpoll核心] --> B[pollDesc池]
        A --> C[平台特定实现]
        A --> D[事件循环]
        
        subgraph "pollDesc结构"
            E[文件描述符fd]
            F[读写等待队列rg/wg]
            G[定时器rt/wt]
            H[状态信息atomicInfo]
        end
        
        B --> E
        B --> F
        B --> G
        B --> H
        
        subgraph "平台实现"
            I[Linux epoll]
            J[macOS kqueue]
            K[Windows IOCP]
            L[Solaris event ports]
        end
        
        C --> I
        C --> J
        C --> K
        C --> L
        
        subgraph "事件处理"
            M[netpoll函数]
            N[事件检测]
            O[goroutine唤醒]
            P[就绪队列管理]
        end
        
        D --> M
        M --> N
        N --> O
        O --> P
        
        subgraph "与调度器集成"
            Q[sysmon监控线程]
            R[工作窃取]
            S[空闲P处理]
        end
        
        A --> Q
        A --> R
        A --> S
    end
    
    style A fill:#e1f5fe
    style B fill:#f3e5f5
    style I fill:#e8f5e8
    style M fill:#fff3e0

// pollDesc 包含网络轮询所需的所有信息
// 这是一个不在堆上分配的结构体，避免GC扫描，提高性能
type pollDesc struct {
    _     sys.NotInHeap  // 标记此结构体不在堆上分配，避免GC扫描开销
    
    // 链表管理字段
    link  *pollDesc      // 在pollcache中的链接指针，由pollcache.lock保护
                        // 用于将空闲的pollDesc组织成链表，便于复用
    
    // 文件描述符相关字段
    fd    uintptr        // 底层文件描述符，在pollDesc使用期间保持不变
                        // 这是与操作系统交互的核心标识符
    fdseq atomic.Uintptr // 文件描述符序列号，防止使用过时的pollDesc
                        // 当文件描述符被重用时，序列号会递增，避免ABA问题
    
    // 状态信息字段 - 使用原子操作保证并发安全
    // atomicInfo保存来自closing、rd和wd的位信息
    // 这些位只在持有锁时写入，汇总供netpollcheckerr使用
    // netpollcheckerr无法获取锁，所以需要原子读取状态
    // 在锁下以可能改变摘要的方式写入这些字段后，
    // 代码必须在释放锁前调用publishInfo更新原子状态
    // 改变字段然后调用netpollunblock的代码（仍持有锁）
    // 必须在调用netpollunblock前调用publishInfo，
    // 因为publishInfo阻止netpollblock重新阻塞
    // （通过改变netpollcheckerr的结果）
    // atomicInfo还保存eventErr位，
    // 记录fd上的轮询事件是否出错；
    // atomicInfo是该位的唯一真实来源
    atomicInfo atomic.Uint32 // 原子操作的pollInfo状态位集合
    
    // goroutine等待队列 - 使用原子操作管理等待的goroutine
    // rg、wg原子访问并保存g指针
    // （在这里使用atomic.Uintptr类似于在其他地方使用guintptr）
    rg atomic.Uintptr // 读等待状态：pdReady(就绪)、pdWait(等待中)、等待读取的G指针或pdNil(无等待)
    wg atomic.Uintptr // 写等待状态：pdReady(就绪)、pdWait(等待中)、等待写入的G指针或pdNil(无等待)
    
    // 互斥锁保护的字段 - 以下字段需要在锁保护下访问
    lock    mutex     // 保护以下字段的互斥锁，确保并发安全
    
    // 连接状态字段
    closing bool      // 标记连接是否正在关闭，防止在关闭过程中进行新的I/O操作
    
    // 定时器运行状态
    rrun    bool      // 读取定时器rt是否正在运行，避免重复启动定时器
    wrun    bool      // 写入定时器wt是否正在运行，避免重复启动定时器
    
    // 用户数据
    user    uint32    // 用户可设置的cookie，用于存储用户自定义数据
    
    // 读取超时管理
    rseq    uintptr   // 读取定时器序列号，防止使用过时的读取定时器
                     // 当定时器被取消重新设置时，序列号会递增
    rt      timer     // 读取截止时间定时器，用于实现读取超时功能
    rd      int64     // 读取截止时间（未来的纳秒时间戳，过期时为-1）
                     // 使用纳秒精度的绝对时间，-1表示无超时限制
    
    // 写入超时管理
    wseq    uintptr   // 写入定时器序列号，防止使用过时的写入定时器
                     // 当定时器被取消重新设置时，序列号会递增
    wt      timer     // 写入截止时间定时器，用于实现写入超时功能
    wd      int64     // 写入截止时间（未来的纳秒时间戳，过期时为-1）
                     // 使用纳秒精度的绝对时间，-1表示无超时限制
    
    // 自引用指针
    self    *pollDesc // 指向自身的指针，用于间接接口的存储
                     // 参见(*pollDesc).makeArg方法，用于类型安全的指针传递
}

// pollInfo是netpollcheckerr需要的位，原子存储，
// 主要复制在pollDesc中锁下操作的状态
// 唯一的例外是pollEventErr位，它只在pollInfo中维护
type pollInfo uint32

const (
    pollClosing = 1 << iota
    pollEventErr
    pollExpiredReadDeadline
    pollExpiredWriteDeadline
    pollFDSeq // 20位字段，fdseq字段的低20位
)

// poll_runtime_pollOpen 创建新的pollDesc并注册到网络轮询器
// 这是internal/poll包调用的运行时函数，用于为新的网络连接创建轮询描述符
//go:linkname poll_runtime_pollOpen internal/poll.runtime_pollOpen
func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) {
    // 从pollDesc缓存池中分配一个pollDesc结构
    // pollcache使用链表管理空闲的pollDesc，避免频繁的内存分配
    pd := pollcache.alloc()
    
    // 获取互斥锁，保护pollDesc的初始化过程
    lock(&pd.lock)
    
    // ==================== 状态检查阶段 ====================
    // 检查是否有阻塞的读写操作，确保pollDesc处于干净状态
    // 从缓存池获取的pollDesc应该是已经清理过的，这里是安全检查
    wg := pd.wg.Load()  // 原子加载写等待状态
    if wg != pdNil && wg != pdReady {
        // 如果写等待状态既不是nil也不是ready，说明有goroutine在等待写入
        // 这表明pollDesc没有被正确清理，是一个严重错误
        throw("runtime: blocked write on free polldesc")
    }
    rg := pd.rg.Load()  // 原子加载读等待状态
    if rg != pdNil && rg != pdReady {
        // 如果读等待状态既不是nil也不是ready，说明有goroutine在等待读取
        // 这表明pollDesc没有被正确清理，是一个严重错误
        throw("runtime: blocked read on free polldesc")
    }
    
    // ==================== 初始化阶段 ====================
    // 设置文件描述符
    pd.fd = fd
    
    // 初始化文件描述符序列号
    if pd.fdseq.Load() == 0 {
        // 值0在setEventErr中是特殊的（表示无效），所以从1开始
        // 序列号用于防止ABA问题，确保不会使用过时的pollDesc
        pd.fdseq.Store(1)
    }
    
    // 重置连接状态
    pd.closing = false  // 标记连接未关闭
    pd.setEventErr(false, 0)  // 清除事件错误状态
    
    // 初始化读取相关字段
    pd.rseq++           // 递增读取序列号，使之前的定时器失效
    pd.rg.Store(pdNil)  // 原子设置读等待状态为nil（无等待者）
    pd.rd = 0           // 清除读取截止时间
    
    // 初始化写入相关字段
    pd.wseq++           // 递增写入序列号，使之前的定时器失效
    pd.wg.Store(pdNil)  // 原子设置写等待状态为nil（无等待者）
    pd.wd = 0           // 清除写入截止时间
    
    // 设置自引用指针
    pd.self = pd        // 用于类型安全的指针传递
    
    // 发布状态信息到原子字段，使其对其他goroutine可见
    pd.publishInfo()
    
    // 释放互斥锁，初始化完成
    unlock(&pd.lock)
    
    // ==================== 注册阶段 ====================
    // 将文件描述符注册到平台特定的轮询器（如epoll、kqueue等）
    errno := netpollopen(fd, pd)
    if errno != 0 {
        // 注册失败，释放已分配的pollDesc并返回错误
        pollcache.free(pd)
        return nil, int(errno)
    }
    
    // 成功创建并注册pollDesc
    return pd, 0
}

// poll_runtime_pollReset 准备描述符用于指定模式的轮询操作
// 在每次I/O操作前调用，重置轮询状态为初始状态
// mode参数：'r'表示读操作，'w'表示写操作
//go:linkname poll_runtime_pollReset internal/poll.runtime_pollReset
func poll_runtime_pollReset(pd *pollDesc, mode int) int {
    // 首先检查pollDesc的错误状态
    // netpollcheckerr会检查连接是否关闭、是否有错误等
    errcode := netpollcheckerr(pd, int32(mode))
    if errcode != pollNoError {
        // 如果有错误（如连接已关闭、超时等），直接返回错误码
        return errcode
    }
    
    // 根据模式重置相应的等待状态
    if mode == 'r' {
        // 读模式：重置读等待状态为pdNil
        // 这表示当前没有goroutine在等待读操作
        pd.rg.Store(pdNil)
    } else if mode == 'w' {
        // 写模式：重置写等待状态为pdNil  
        // 这表示当前没有goroutine在等待写操作
        pd.wg.Store(pdNil)
    }
    
    // 返回成功状态，表示pollDesc已准备好进行I/O操作
    return pollNoError
}

// poll_runtime_pollWait 等待描述符变为可读或可写
// 这是I/O操作的核心等待函数，当I/O操作无法立即完成时调用
// mode参数：'r'表示等待可读，'w'表示等待可写
// 如果超时，返回pollErrTimeout；如果连接关闭，返回相应错误码
//go:linkname poll_runtime_pollWait internal/poll.runtime_pollWait
func poll_runtime_pollWait(pd *pollDesc, mode int) int {
    // 首先检查pollDesc的当前状态
    errcode := netpollcheckerr(pd, int32(mode))
    if errcode != pollNoError {
        // 如果已经有错误（连接关闭、超时等），直接返回
        return errcode
    }
    
    // 如果我们执行到这里，意味着某个goroutine正在等待网络I/O就绪
    // 调度器使用这个信息来决定是否应该阻塞等待网络轮询器的事件
    
    // 循环尝试阻塞等待，直到I/O就绪或发生错误
    for !netpollblock(pd, int32(mode), false) {
        // netpollblock返回false表示阻塞失败，需要重新检查状态
        
        // 重新检查错误状态，可能在等待过程中发生了变化
        errcode = netpollcheckerr(pd, int32(mode))
        if errcode != pollNoError {
            // 发现错误，返回错误码
            return errcode
        }
        
        // 可能有其他goroutine在我们阻塞之前就通知了我们I/O就绪
        // 或者发生了虚假唤醒，重新尝试阻塞
        // 这种情况下需要再次尝试netpollblock
    }
    
    // netpollblock返回true，表示I/O已就绪或goroutine被正确唤醒
    return pollNoError
}

// netpollblock 阻塞等待网络I/O就绪
// 这是网络轮询的核心阻塞函数，负责将goroutine挂起直到I/O事件发生
// pd: 轮询描述符
// mode: 'r'表示等待读就绪，'w'表示等待写就绪  
// waitio: 是否强制等待I/O（通常为false，让函数自己判断）
// 返回值：true表示I/O已就绪，false表示不需要等待或等待失败
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
    // 根据模式选择对应的等待状态指针
    gpp := &pd.rg  // 默认为读等待状态
    if mode == 'w' {
        gpp = &pd.wg  // 写模式使用写等待状态
    }
    
    // ==================== 状态检查和设置阶段 ====================
    // 使用CAS循环设置等待状态，确保原子性和并发安全
    for {
        old := gpp.Load()  // 原子加载当前等待状态
        
        if old == pdReady {
            // I/O已经就绪，无需等待
            // 将状态重置为pdNil，表示没有等待者
            gpp.Store(pdNil)
            return true  // 立即返回，表示I/O就绪
        }
        
        if old != pdNil {
            // 如果状态不是pdNil，说明已经有其他goroutine在等待
            // 这是一个错误情况，因为同一时间只能有一个goroutine等待同一种I/O
            throw("runtime: double wait")
        }
        
        // 尝试将状态从pdNil设置为pdWait
        // 使用CAS确保原子性，如果失败说明状态被其他goroutine修改了
        if gpp.CompareAndSwap(pdNil, pdWait) {
            break  // 成功设置为等待状态，跳出循环
        }
        // CAS失败，重新尝试
    }
    
    // ==================== 阻塞等待阶段 ====================
    // 检查是否需要进入阻塞等待
    if waitio || netpollcheckerr(pd, mode) == pollNoError {
        // waitio为true表示强制等待，或者当前没有错误状态
        
        // 调用gopark将当前goroutine挂起
        // netpollblockcommit: 提交函数，在goroutine被挂起前调用
        // unsafe.Pointer(gpp): 传递给提交函数的参数
        // waitReasonIOWait: 等待原因，用于调试和监控
        // traceBlockNet: 跟踪事件类型
        // 5: 跳过的栈帧数，用于调用栈跟踪
        gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceBlockNet, 5)
    }
    
    // ==================== 唤醒后检查阶段 ====================
    // goroutine被唤醒，检查最终状态
    old := gpp.Load()  // 原子加载当前状态
    
    if old > pdWait {
        // 状态值异常，pdWait应该是最大的有效值
        // 如果大于pdWait，说明pollDesc数据结构被破坏
        throw("runtime: corrupted polldesc")
    }
    
    // 返回I/O是否就绪
    // old == pdReady 表示I/O已就绪
    // old != pdReady 表示被其他原因唤醒（如超时、连接关闭等）
    return old == pdReady
}

graph TB
    subgraph "Linux epoll 实现"
        A[epoll_create1] --> B[创建epoll实例]
        B --> C[epfd全局描述符]
        
        D[netpollopen] --> E[epoll_ctl ADD]
        E --> F[注册文件描述符]
        
        G[netpoll] --> H[epoll_wait]
        H --> I[等待事件]
        I --> J[处理就绪事件]
        
        K[netpollBreak] --> L[eventfd通知]
        L --> M[中断epoll_wait]
        
        subgraph "事件类型"
            N[EPOLLIN 可读]
            O[EPOLLOUT 可写]
            P[EPOLLRDHUP 对端关闭]
            Q[EPOLLET 边缘触发]
        end
        
        F --> N
        F --> O
        F --> P
        F --> Q
    end
    
    style A fill:#e1f5fe
    style G fill:#f3e5f5
    style K fill:#e8f5e8

// Linux epoll相关的全局变量
var (
    epfd           int32         = -1 // epoll文件描述符，-1表示未初始化
    netpollEventFd uintptr            // 用于netpollBreak的eventfd，用于中断epoll_wait
    netpollWakeSig atomic.Uint32      // 唤醒信号标志，用于避免重复调用netpollBreak
)

// netpollinit 初始化Linux epoll网络轮询器
// 这个函数在运行时启动时被调用，设置网络I/O多路复用机制
func netpollinit() {
    var errno uintptr
    
    // ==================== 创建epoll实例 ====================
    // 创建epoll实例，设置CLOEXEC标志确保子进程不会继承这个文件描述符
    // EPOLL_CLOEXEC: 在exec时自动关闭，避免文件描述符泄漏
    epfd, errno = linux.EpollCreate1(linux.EPOLL_CLOEXEC)
    if errno != 0 {
        // epoll创建失败，这是致命错误，因为没有epoll就无法进行网络I/O
        println("runtime: epollcreate failed with", errno)
        throw("runtime: netpollinit failed")
    }
    
    // ==================== 创建中断机制 ====================
    // 创建eventfd用于中断epoll_wait调用
    // 当需要唤醒阻塞在epoll_wait上的线程时，向这个eventfd写入数据
    // EFD_CLOEXEC: exec时自动关闭
    // EFD_NONBLOCK: 非阻塞模式，避免写入时阻塞
    efd, errno := linux.Eventfd(0, linux.EFD_CLOEXEC|linux.EFD_NONBLOCK)
    if errno != 0 {
        println("runtime: eventfd failed with", errno)
        throw("runtime: eventfd failed")
    }
    
    // ==================== 注册中断事件 ====================
    // 将eventfd添加到epoll中，监听可读事件
    ev := linux.EpollEvent{
        Events: linux.EPOLLIN,  // 监听可读事件，当有数据写入eventfd时触发
    }
    // 将netpollEventFd的地址存储在事件数据中，用于识别这是中断事件
    *(**uintptr)(unsafe.Pointer(&ev.Data)) = &netpollEventFd
    
    // 使用EPOLL_CTL_ADD将eventfd添加到epoll监听列表
    errno = linux.EpollCtl(epfd, linux.EPOLL_CTL_ADD, efd, &ev)
    if errno != 0 {
        println("runtime: epollctl failed with", errno)
        throw("runtime: epollctl failed")
    }
    
    // 保存eventfd，供后续netpollBreak使用
    netpollEventFd = uintptr(efd)
}

// netpollopen 将文件描述符添加到epoll中
func netpollopen(fd uintptr, pd *pollDesc) uintptr {
    var ev linux.EpollEvent
    // 设置事件类型：可读、可写、对端关闭、边缘触发
    ev.Events = linux.EPOLLIN | linux.EPOLLOUT | linux.EPOLLRDHUP | linux.EPOLLET
    
    // 将pollDesc指针打包到事件数据中
    tp := taggedPointerPack(unsafe.Pointer(pd), pd.fdseq.Load())
    *(*taggedPointer)(unsafe.Pointer(&ev.Data)) = tp
    
    // 添加到epoll中
    return linux.EpollCtl(epfd, linux.EPOLL_CTL_ADD, int32(fd), &ev)
}

// netpoll 检查就绪的网络连接
// 返回变为可运行的goroutine列表和要添加到netpollWaiters的增量
// delay < 0: 无限期阻塞
// delay == 0: 不阻塞，只是轮询
// delay > 0: 阻塞最多delay纳秒
func netpoll(delay int64) (gList, int32) {
    if epfd == -1 {
        return gList{}, 0
    }
    
    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 {
        waitms = 1e9
    }
    
    var events [128]linux.EpollEvent
retry:
    n := linux.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 waitms > 0 {
            return gList{}, 0
        }
        goto retry
    }
    
    var toRun gList
    delta := int32(0)
    for i := int32(0); i < n; i++ {
        ev := &events[i]
        if ev.Events == 0 {
            continue
        }
        
        if *(**uintptr)(unsafe.Pointer(&ev.Data)) == &netpollEventFd {
            // 这是eventfd通知，清空它
            if netpollWakeSig.Load() != 0 {
                netpollWakeSig.Store(0)
                // 消费eventfd中的数据
                var tmp [16]byte
                read(int32(netpollEventFd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))
            }
            continue
        }
        
        // 解包pollDesc指针
        tp := *(*taggedPointer)(unsafe.Pointer(&ev.Data))
        pd := (*pollDesc)(taggedPointerPtr(tp))
        fdseq := taggedPointerTag(tp)
        
        var mode int32
        if ev.Events&(linux.EPOLLIN|linux.EPOLLRDHUP|linux.EPOLLHUP|linux.EPOLLERR) != 0 {
            mode += 'r'
        }
        if ev.Events&(linux.EPOLLOUT|linux.EPOLLHUP|linux.EPOLLERR) != 0 {
            mode += 'w'
        }
        if mode != 0 {
            pd.setEventErr(ev.Events&linux.EPOLLERR != 0, fdseq)
            delta += netpollready(&toRun, pd, mode)
        }
    }
    return toRun, delta
}

// netpollBreak 中断epoll_wait
func netpollBreak() {
    // CAS失败表示有正在进行的唤醒，所以我们完成了
    if !netpollWakeSig.CompareAndSwap(0, 1) {
        return
    }
    
    // 向eventfd写入数据以唤醒epoll_wait
    var b byte = 1
    write(int32(netpollEventFd), noescape(unsafe.Pointer(&b)), 1)
}

graph TB
    subgraph "macOS kqueue 实现"
        A[kqueue] --> B[创建kqueue实例]
        B --> C[kq全局描述符]
        
        D[netpollopen] --> E[kevent注册]
        E --> F[添加读写事件]
        
        G[netpoll] --> H[kevent等待]
        H --> I[获取就绪事件]
        I --> J[处理事件列表]
        
        K[netpollBreak] --> L[wakeNetpoll]
        L --> M[中断kevent]
        
        subgraph "事件类型"
            N[EVFILT_READ 可读]
            O[EVFILT_WRITE 可写]
            P[EV_ADD 添加事件]
            Q[EV_DELETE 删除事件]
        end
        
        F --> N
        F --> O
        F --> P
        F --> Q
    end
    
    style A fill:#e1f5fe
    style G fill:#f3e5f5
    style K fill:#e8f5e8

var (
    kq               int32 = -1
    netpollWakeSig   atomic.Uint32
)

// netpollinit 初始化kqueue
func netpollinit() {
    kq = kqueue()
    if kq < 0 {
        println("runtime: kqueue failed with", -kq)
        throw("runtime: netpollinit failed")
    }
    closeonexec(kq)
}

// netpollopen 将文件描述符添加到kqueue
func netpollopen(fd uintptr, pd *pollDesc) uintptr {
    // 添加读事件过滤器
    var kev keventt
    kev.ident = uint64(fd)
    kev.filter = _EVFILT_READ
    kev.flags = _EV_ADD
    kev.udata = (*byte)(unsafe.Pointer(pd))
    
    n := kevent(kq, &kev, 1, nil, 0, nil)
    if n < 0 {
        return -n
    }
    
    // 添加写事件过滤器
    kev.filter = _EVFILT_WRITE
    kev.flags = _EV_ADD
    n = kevent(kq, &kev, 1, nil, 0, nil)
    if n < 0 {
        return -n
    }
    return 0
}

// netpoll 检查就绪的网络连接
func netpoll(delay int64) (gList, int32) {
    if kq == -1 {
        return gList{}, 0
    }
    
    var tp *timespec
    var ts timespec
    if delay < 0 {
        tp = nil
    } else if delay == 0 {
        tp = &ts
    } else {
        ts.setNsec(delay)
        if ts.tv_sec > 1e6 {
            // Darwin在睡眠时间过长时返回EINVAL
            ts.tv_sec = 1e6
        }
        tp = &ts
    }
    
    var events [64]keventt
retry:
    n := kevent(kq, nil, 0, &events[0], int32(len(events)), tp)
    if n < 0 {
        if n != -_EINTR && n != -_ETIMEDOUT {
            println("runtime: kevent on fd", kq, "failed with", -n)
            throw("runtime: netpoll failed")
        }
        // 如果定时睡眠被中断，只需返回重新计算睡眠时间
        if delay > 0 {
            return gList{}, 0
        }
        goto retry
    }
    
    var toRun gList
    delta := int32(0)
    for i := 0; i < int(n); i++ {
        ev := &events[i]
        
        var mode int32
        switch ev.filter {
        case _EVFILT_READ:
            mode += 'r'
        case _EVFILT_WRITE:
            mode += 'w'
        }
        
        if mode != 0 {
            pd := (*pollDesc)(unsafe.Pointer(ev.udata))
            pd.setEventErr(ev.flags&_EV_ERROR != 0, 0)
            delta += netpollready(&toRun, pd, mode)
        }
    }
    return toRun, delta
}

graph TB
    subgraph "netFD 架构"
        A[netFD] --> B[poll.FD]
        A --> C[网络信息]
        A --> D[地址信息]
        
        subgraph "poll.FD组件"
            E[系统文件描述符Sysfd]
            F[文件描述符互斥锁fdmu]
            G[轮询描述符pd]
            H[I/O状态标志]
        end
        
        B --> E
        B --> F
        B --> G
        B --> H
        
        subgraph "网络信息"
            I[协议族family]
            J[套接字类型sotype]
            K[连接状态isConnected]
            L[网络类型net]
        end
        
        C --> I
        C --> J
        C --> K
        C --> L
        
        subgraph "地址信息"
            M[本地地址laddr]
            N[远程地址raddr]
        end
        
        D --> M
        D --> N
        
        subgraph "I/O操作"
            O[Read读取]
            P[Write写入]
            Q[Accept接受连接]
            R[Connect建立连接]
        end
        
        A --> O
        A --> P
        A --> Q
        A --> R
    end
    
    style A fill:#e1f5fe
    style B fill:#f3e5f5
    style G fill:#e8f5e8
    style O fill:#fff3e0

// Network file descriptor.
type netFD struct {
    pfd poll.FD
    
    // immutable until Close
    family      int    // 协议族 (AF_INET, AF_INET6, AF_UNIX)
    sotype      int    // 套接字类型 (SOCK_STREAM, SOCK_DGRAM)
    isConnected bool   // 握手完成或与对端建立关联
    net         string // 网络类型 ("tcp", "udp", "unix")
    laddr       Addr   // 本地地址
    raddr       Addr   // 远程地址
}

// poll.FD 是文件描述符的核心结构
type FD struct {
    // 锁定sysfd并序列化对Read和Write方法的访问
    fdmu fdMutex
    
    // 系统文件描述符。在Close之前不可变
    Sysfd int
    
    // 文件描述符的平台相关状态
    SysFile
    
    // I/O轮询器
    pd pollDesc
    
    // 文件关闭时发出信号的信号量
    csema uint32
    
    // 如果此文件已设置为阻塞模式，则非零
    isBlocking uint32
    
    // 这是否是流描述符，而不是基于数据包的描述符（如UDP套接字）
    // 不可变
    IsStream bool
    
    // 零字节读取是否表示EOF。对于基于消息的套接字连接，这是false
    ZeroReadIsEOF bool
    
    // 这是否是文件而不是网络套接字
    isFile bool
}

// Read 从网络连接读取数据
func (fd *netFD) Read(p []byte) (n int, err error) {
    n, err = fd.pfd.Read(p)
    runtime.KeepAlive(fd)
    return n, wrapSyscallError(readSyscallName, err)
}

// Write 向网络连接写入数据
func (fd *netFD) Write(p []byte) (n int, err error) {
    n, err = fd.pfd.Write(p)
    runtime.KeepAlive(fd)
    return n, wrapSyscallError(writeSyscallName, err)
}

// poll.FD的Read实现
func (fd *FD) Read(p []byte) (int, error) {
    if err := fd.readLock(); err != nil {
        return 0, err
    }
    defer fd.readUnlock()
    
    if len(p) == 0 {
        // 如果缓冲区为空，仍然需要检查错误
        return 0, nil
    }
    
    if err := fd.pd.prepareRead(fd.isFile); err != nil {
        return 0, err
    }
    
    if fd.IsStream && len(p) > maxRW {
        p = p[:maxRW]
    }
    
    for {
        // 尝试非阻塞读取
        n, err := ignoringEINTRIO(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
    }
}

// poll.FD的Write实现
func (fd *FD) Write(p []byte) (int, error) {
    if err := fd.writeLock(); err != nil {
        return 0, err
    }
    defer fd.writeUnlock()
    
    if err := fd.pd.prepareWrite(fd.isFile); err != nil {
        return 0, err
    }
    
    var nn int
    for {
        max := len(p)
        if fd.IsStream && max-nn > maxRW {
            max = nn + maxRW
        }
        
        // 尝试非阻塞写入
        n, err := ignoringEINTRIO(syscall.Write, fd.Sysfd, p[nn:max])
        if n > 0 {
            nn += n
        }
        if nn == len(p) {
            return nn, err
        }
        if err == syscall.EAGAIN && fd.pd.pollable() {
            // 缓冲区已满，等待轮询器通知
            if err = fd.pd.waitWrite(fd.isFile); err == nil {
                continue
            }
        }
        if err != nil {
            return nn, err
        }
        if n == 0 {
            return nn, io.ErrUnexpectedEOF
        }
    }
}

sequenceDiagram
    participant C as Client
    participant S as Server
    participant N as netpoll
    participant K as Kernel

    Note over C,K: TCP连接建立过程
    
    S->>K: socket() + bind() + listen()
    S->>N: netpollopen(listenfd)
    N->>K: epoll_ctl(ADD, listenfd)
    
    C->>K: socket() + connect()
    K->>K: TCP三次握手
    K->>N: 连接就绪事件
    
    N->>N: netpoll()检测到事件
    N->>S: 唤醒Accept goroutine
    S->>K: accept()系统调用
    K->>S: 返回新连接fd
    
    S->>N: netpollopen(connfd)
    N->>K: epoll_ctl(ADD, connfd)
    
    Note over S: 连接建立完成

sequenceDiagram
    participant G as Goroutine
    participant FD as netFD
    participant PD as poll.FD
    participant NP as netpoll
    participant K as Kernel

    Note over G,K: 网络读取操作时序
    
    G->>FD: conn.Read(buf)
    FD->>PD: fd.pfd.Read(buf)
    PD->>PD: readLock()
    PD->>PD: pd.prepareRead()
    
    PD->>K: syscall.Read(fd, buf) 非阻塞
    alt 数据就绪
        K->>PD: 返回数据
        PD->>FD: 返回结果
        FD->>G: 返回数据
    else 数据未就绪 (EAGAIN)
        K->>PD: EAGAIN错误
        PD->>NP: pd.waitRead() 等待可读
        NP->>NP: netpollblock() 阻塞goroutine
        
        Note over NP: 等待网络事件...
        
        K->>NP: 数据到达，触发可读事件
        NP->>NP: netpoll()检测到事件
        NP->>PD: 唤醒等待的goroutine
        PD->>K: 重新尝试 syscall.Read()
        K->>PD: 返回数据
        PD->>FD: 返回结果
        FD->>G: 返回数据
    end
    
    PD->>PD: readUnlock()

sequenceDiagram
    participant G as Goroutine
    participant FD as netFD
    participant PD as poll.FD
    participant NP as netpoll
    participant K as Kernel

    Note over G,K: 网络写入操作时序
    
    G->>FD: conn.Write(data)
    FD->>PD: fd.pfd.Write(data)
    PD->>PD: writeLock()
    PD->>PD: pd.prepareWrite()
    
    loop 直到所有数据写完
        PD->>K: syscall.Write(fd, data) 非阻塞
        alt 缓冲区有空间
            K->>PD: 返回写入字节数
            PD->>PD: 更新已写入字节数
        else 缓冲区已满 (EAGAIN)
            K->>PD: EAGAIN错误
            PD->>NP: pd.waitWrite() 等待可写
            NP->>NP: netpollblock() 阻塞goroutine
            
            Note over NP: 等待网络事件...
            
            K->>NP: 缓冲区可写，触发可写事件
            NP->>NP: netpoll()检测到事件
            NP->>PD: 唤醒等待的goroutine
            Note over PD: 继续写入循环
        end
    end
    
    PD->>FD: 返回总写入字节数
    FD->>G: 返回结果
    PD->>PD: writeUnlock()

graph TB
    subgraph "netpoll与调度器集成"
        A[sysmon监控线程] --> B[定期调用netpoll]
        B --> C[检查网络事件]
        C --> D[获取就绪goroutine列表]
        
        E[工作窃取] --> F[findRunnable]
        F --> G[netpoll(0) 非阻塞检查]
        G --> H[获取可运行goroutine]
        
        I[空闲P处理] --> J[park等待工作]
        J --> K[netpoll(-1) 阻塞等待]
        K --> L[网络事件唤醒]
        
        subgraph "事件处理流程"
            M[netpollready]
            N[构建就绪列表]
            O[设置goroutine状态]
            P[加入运行队列]
        end
        
        D --> M
        H --> M
        L --> M
        M --> N
        N --> O
        O --> P
        
        subgraph "调度决策"
            Q[是否有网络等待者]
            R[netpollWaiters计数]
            S[调度策略调整]
        end
        
        A --> Q
        E --> Q
        I --> Q
        Q --> R
        R --> S
    end
    
    style A fill:#e1f5fe
    style E fill:#f3e5f5
    style I fill:#e8f5e8
    style M fill:#fff3e0

// sysmon中的网络轮询
func sysmon() {
    // ... 其他监控逻辑
    
    // 定期检查网络I/O
    if netpollinited() && netpollAnyWaiters() && sched.lastpoll.Load() != 0 {
        if GOOS != "plan9" { // plan9 runtime does not implement netpoll yet
            list, delta := netpoll(0) // 非阻塞检查
            if !list.empty() {
                // 有就绪的网络I/O，注入到调度器
                incidlelocked(-1)
                injectglist(&list)
                incidlelocked(1)
                netpollAdjustWaiters(delta)
            }
        }
    }
    
    // ... 其他监控逻辑
}

// findRunnable中的网络轮询
func findRunnable() (gp *g, inheritTime, tryWakeP bool) {
    // ... 其他查找逻辑
    
    // 尝试从网络轮询器获取工作
    if netpollinited() && netpollAnyWaiters() {
        list, delta := netpoll(0) // 非阻塞
        if !list.empty() {
            gp := list.pop()
            injectglist(&list)
            netpollAdjustWaiters(delta)
            casgstatus(gp, _Gwaiting, _Grunnable)
            if traceEnabled() {
                traceGoUnpark(gp, 0)
            }
            return gp, false, false
        }
    }
    
    // ... 其他查找逻辑
}

// stopm中的网络轮询（P进入空闲状态时）
func stopm() {
    // ... 其他逻辑
    
    // 如果有网络等待者，阻塞等待网络I/O
    if netpollinited() && netpollAnyWaiters() {
        list, delta := netpoll(-1) // 阻塞等待
        if !list.empty() {
            // 有网络I/O就绪，重新激活P
            acquirep(pp)
            gp := list.pop()
            injectglist(&list)
            netpollAdjustWaiters(delta)
            casgstatus(gp, _Gwaiting, _Grunnable)
            return gp
        }
    }
    
    // ... 其他逻辑
}

// netpollready 处理就绪的网络I/O
func netpollready(toRun *gList, pd *pollDesc, mode int32) 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)
    }
    
    delta := int32(0)
    if rg != nil {
        toRun.push(rg)
        delta++
    }
    if wg != nil {
        toRun.push(wg)
        delta++
    }
    return delta
}

// netpollunblock 唤醒等待网络I/O的goroutine
func netpollunblock(pd *pollDesc, mode int32, ioready bool) *g {
    gpp := &pd.rg
    if mode == 'w' {
        gpp = &pd.wg
    }
    
    for {
        old := gpp.Load()
        if old == pdReady {
            return nil
        }
        if old == pdNil && !ioready {
            // 只有在I/O就绪时才设置pdReady
            return nil
        }
        var new uintptr
        if ioready {
            new = pdReady
        } else {
            new = pdNil
        }
        if gpp.CompareAndSwap(old, new) {
            if old == pdWait {
                old = 0
            }
            return (*g)(unsafe.Pointer(old))
        }
    }
}