Golang epoll

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

var (
    // epoll 描述符, epollcreate 函数创建返回的文件描述符
    epfd int32 = -1 // epoll descriptor

    // 保存的是pipe2()或pip()函数创建的读写描述符，pip2()或pip()函数创建的read和write，只要任意一方操作另一方能获取到数据
    // netpollBreakRd会被注册到epoll中，当写描述符向里写数据时会触发wait监听函数返回就绪的事件集
    // 这对读写描述符的作用在于通信，当有其他协程在wait阻塞等待时，可以通过写描述符写入数据让等待的协程返回
    // 主要针对 epoll_wait() 函数阻塞的线程，通过这里的事件使调用 epoll_wait() 函数的线程陷入内核返回。
    // 对于 epoll_wait() 的 timeout 参数为0的情况，这里的事件会被忽略
    netpollBreakRd, netpollBreakWr uintptr // for netpollBreak

    // 用于避免重复调用 netpollBreak()
    // 在向netpollBreakWr中写入数据时，该值会从0变成1，控制只写一次标志符号
    // 该参数是原子性的
    netpollWakeSig uint32 // used to avoid duplicate calls of netpollBreak
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func netpollGenericInit() {
    // var netpollInited uint32 
    // 是否已初始化epoll标志 0.未初始化 1.已初始化
    if atomic.Load(&netpollInited) == 0 {
        // var netpollInitLock mutex 初始化
        lockInit(&netpollInitLock, lockRankNetpollInit)
        lock(&netpollInitLock)
        // 这里需要判断 netpollInited == 0；原因在于可能存在多个协程并发在等待初始化epoll
        // 当这些协程获取的锁权限时，这里的netpollInited已被设置成1了，已经被初始化了
        if netpollInited == 0 {
            netpollinit()   // 初始化netpoll
            atomic.Store(&netpollInited, 1)	// netpollInited
        }
        unlock(&netpollInitLock)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

func netpollinit() {
    // epollcreate 函数内会调用 epollcreate1
    
    // _EPOLL_CLOEXEC：当前进程fork出来的任何子进程在执行前都会关闭epoll描述符，也因此，子进程不能够访问epoll实例
    epfd = epollcreate1(_EPOLL_CLOEXEC)
    // 小于0表示创建出错，大于0表示返回的创建后的文件句柄ID
    if epfd < 0 {	
        epfd = epollcreate(1024)
        if epfd < 0 {
            println("runtime: epollcreate failed with", -epfd)
            throw("runtime: netpollinit failed")
        }
        // 系统调用 fcntl 设置 FD_CLOEXEC，参看 epollcreate1 函数参数
        // fcntl：fd, F_SETFD, FD_CLOEXEC
        closeonexec(epfd)
    }
    // 创建一个用于通信的管道,返回读写,主要用于那些等待在IO轮询中的线程通信
    // 创建一个非阻塞式pipe，用来唤醒阻塞中的 netpoller。
    // pipe2(_O_NONBLOCK | _O_CLOEXEC)
    r, w, errno := nonblockingPipe()    // 主要用于Break相关的函数，主要用于网络轮询唤醒信号
    if errno != 0 {
        println("runtime: pipe failed with", -errno)
        throw("runtime: pipe failed")
    }
    // epollevent 是事件类型
    // 	type epollevent struct {
    //		events uint32   // 事件类型
    //		data [8]byte    // 存储用户数据，刚好是一个指针存储的大小，该数据用户是可以修改的
    // 	}
    ev := epollevent{
        // 注意：这里默认注册的是水平触发，因此会一直触发
        events: _EPOLLIN,   // EPOLLIN 表示对应的文件描述符可以读（包括对端SOCKET正常关闭）
    }
    // var netpollBreakRd uintptr
    *(**uintptr)(unsafe.Pointer(&ev.data)) = &netpollBreakRd // ev.data = &netpollBreakRd
    // 将读取数据的文件描述符加入监听，当使用w写数据时，该r会被触发
    // func epollctl(epfd, op, fd int32, ev *epollevent) int32
    // _EPOLL_CTL_ADD 将目标文件描述符fd添加到epoll描述符epfd中，并将事件event与fd链接的内部文件关联起来。
    errno = epollctl(epfd, _EPOLL_CTL_ADD, r, &ev)      // r 被添加到epoll中
    if errno != 0 {
        println("runtime: epollctl failed with", -errno)
        throw("runtime: epollctl failed")
    }
    // netpollBreakRd、netpollBreakWr 是非阻塞管道两端的文件描述符，分别被用作读取端和写入端。
    // 读取端 netpollBreakRd 被添加到 epoll 中监听 _EPOLLIN 事件，后续从写入端netpollBreakWr
    // 写入数据就能唤醒阻塞中的 poller。
    netpollBreakRd = uintptr(r) // netpollBreakRd保存pip2的read
    netpollBreakWr = uintptr(w) // netpollBreakRd保存pip2的write
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

func netpollopen(fd uintptr, pd *pollDesc) int32 {
    // 	type epollevent struct {
    //      events uint32   // 事件类型
    //      data [8]byte    // 存储用户数据，刚好是一个指针存储的大小，该数据用户是可以修改的
    // 	}
    var ev epollevent
    // EPOLLIN：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）
    // EPOLLOUT：表示对应的文件描述符可以写
    // EPOLLRDHUP：对端描述符产生一个挂断事件，比如来自对端的socket挂断事件等
    // EPOLLET：将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的
    // pollDesc：类型的数据结构pd作为与fd关联的自定数据会被一同添加到epoll中。
    ev.events = _EPOLLIN | _EPOLLOUT | _EPOLLRDHUP | _EPOLLET
    // epollevent 是由events和data组成，data来自用户空间传入的数据
    *(**pollDesc)(unsafe.Pointer(&ev.data)) = pd // ev.data = pd
    // _EPOLL_CTL_ADD 将目标文件描述符fd添加到epoll描述符epfd中，并将事件event与fd链接的内部文件关联起来。
    return -epollctl(epfd, _EPOLL_CTL_ADD, int32(fd), &ev)
}

1
2
3
4
5
6

func netpollclose(fd uintptr) int32 {
    var ev epollevent
    // _EPOLL_CTL_DEL 从epoll文件描述符epfd中删除目标文件描述符fd。
    // 该事件被忽略，可以为NULL。
    return -epollctl(epfd, _EPOLL_CTL_DEL, int32(fd), &ev)	
}

1
2
3

func netpollIsPollDescriptor(fd uintptr) bool {
    return fd == uintptr(epfd) || fd == netpollBreakRd || fd == netpollBreakWr
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// netpollBreak interrupts an epollwait.
func netpollBreak() {
    // netpollWakeSig信号主要用于本轮已经通知过IO轮询，但是还没处理有其他的调用此方法时
    // 使用原子操作将netpollWakeSig由0变成1，表示正在唤醒epoll。
    if atomic.Cas(&netpollWakeSig, 0, 1) {	
        for {
            var b byte
            // 向netpollBreakWr中写入数据，会导致那些阻塞在netpoll函数中的线程直接返回去执行后面代码
            // netpollBreakWr在pipe2函数中注册时已经设置了非阻塞。因此这里不会阻塞。
            n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
            // 写入成功
            if n == 1 {
                break
            }
            // 在写入任何数据之前，调用被信号中断。
            if n == -_EINTR {
                continue // 重试。
            }
            // 已使用 O_NONBLOCK 选择了非阻塞 I/O，并且写入将阻塞。
            // _EAGAIN：表示目前没有可用的数据
            if n == -_EAGAIN {
                return
            }
            println("runtime: netpollBreak write failed with", -n)
            throw("runtime: netpollBreak write failed")
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

// netpoll checks for ready network connections.
// Returns list of goroutines that become runnable.
// 
// delay < 0: blocks indefinitely;	
// delay == 0: does not block, just polls;
// delay > 0: block for up to that many nanoseconds;
func netpoll(delay int64) gList {
    // 1) epoll没有初始化
    if epfd == -1 {		
        return gList{}
    }
    
    // 2) 下面把纳秒级的 delay 转换成毫秒级的 waitms。
    var waitms int32
    if delay < 0 {
        waitms = -1 // 永久阻塞，直到事件就绪返回
    } else if delay == 0 {
        waitms = 0 	// 不阻塞，立即返回
    } else if delay < 1e6 { // 小于1毫秒，修改为阻塞1毫秒
        waitms = 1 	// 阻塞1毫秒
    } else if delay < 1e15 { // 小于 11.5 day
        // 1e6 表示 1毫秒
        waitms = int32(delay / 1e6) // 阻塞指定毫秒
    } else {
        // An arbitrary cap on how long to wait for a timer.
        // 1e9 ms == ~11.5 days.
        waitms = 1e9    // 最长 11.5天
    }
    
    // 3) 通过epollwait函数等待IO事件，缓冲区大小为128个epollevent。
    // 超时时间是 waitms 毫秒。如果 epollwait函数被中断打断，就通过goto来重试。
    // waitms 大于0时不会重试，因为需要返回调用者中去重新计算超时时间。
    
    // 下面传入epollwait的数量是128，表示一次最大就绪128个事件
    // 这样可能存在此次大于128数量时，需要等到下一个IO轮询时间窗口
    //  type epollevent struct {
    //      events uint32   // 事件类型
    //      data [8]byte    // 存储用户数据，刚好是一个指针存储的大小，该数据用户是可以修改的
    //  }
    var events [128]epollevent  // 用于存储已经准备好的描述符的事件数据
retry:
    // 调用epollwait等待文件描述符转换成可写或可读，如果没有epollwait会阻塞
    n := epollwait(epfd, &events[0], int32(len(events)), waitms)    // 返回活跃的数量n
    if n < 0 {
        // EBADF：epfd 不是有效的文件描述符。
        // EFAULT：事件指向的内存区域无法用写权限访问。
        // EINVAL：epfd 不是epoll文件描述符，或者maxevents(第三个参数)小于等于0。
        if n != -_EINTR { // EINTR 被CPU中断
            println("runtime: epollwait on fd", epfd, "failed with", -n)
            throw("runtime: netpoll failed")
        }
        // If a timed sleep was interrupted, just return to
        // recalculate how long we should sleep now.
        //
        // 如果定时睡眠被中断，只需返回重新计算我们现在应该睡多久。
        // _EINTR: 在任何请求的事件发生或超时到期之前，该调用被信号处理程序中断。
        if waitms > 0 {
            return gList{}
        }
        // 被中断 AND waitms <= 0 情况重试
        goto retry
    }
    // type gList struct { head guintptr }
    var toRun gList
    // 意味着被监控的文件描述符出现了待处理的事件
    for i := int32(0); i < n; i++ {
        // 就绪的IO事件集
        ev := &events[i] 
        // 当前就绪描述符没有事件类型，直接跳过
        if ev.events == 0 {
            continue
        }

        // 事件来源是否是netpollBreakRd，该描述符来之pipe2函数创建
        // 来自pipe2函数创建的通信，netpollBreakRd是LT水平触发如果不读取会一直触发
        // 该事件来自 netpollBreak() 方法，该方法只会在创建timer时和findRunnable()函数中被调用。
        if *(**uintptr)(unsafe.Pointer(&ev.data)) == &netpollBreakRd {
            // 对于文件描述符 netpollBreakRd 而言，只有 _EPOLLIN 事件是正常的，其他都会被视为异常。
            if ev.events != _EPOLLIN {
                println("runtime: netpoll: break fd ready for", ev.events)
                throw("runtime: netpoll: break fd ready for something unexpected")
            }
            
            // 这种情况下只处理不是立即返回情况下，如果是立即返回情况时，数据并未被读取，下次还会触发该信号
            // 这种情况下只在runtime.findrunnable()函数中存在，线程在寻找g无果时，最后只能在IO轮询处等待
            // 只有在 delay 不为0，也就是阻塞式netpoll时，才读取netpollBreakRd中的数据。
            if delay != 0 { // netpollBreakRd 的本意也是只唤醒delay != 0的netpoll，因为这些在阻塞需要返回
                // netpollBreak could be picked up by a
                // nonblocking poll. Only read the byte
                // if blocking.
                // netpollBreak 可以通过非阻塞轮询来获取。 仅在阻塞时读取字节。
                var tmp [16]byte
                // 把写入的数据读取让缓存区为空
                read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))	
                // 将netpollWakeSig由1变成0，表示当前事件已被读取
                atomic.Store(&netpollWakeSig, 0)	
            }
            continue
        }
        
        // 根据epoll返回的IO事件标志位为mode赋值
        //  r 表示可读，w 表示可写，r+w 表示既可读又可写。
        // 检测IO事件中的错误标志位，并相应的为pd.everr赋值。
        
        // 判断发生的事件类型,读类型或者写类型
        var mode int32		// 存储当前类型 
        // EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）
        // EPOLLRDHUP：对端描述符产生一个挂断事件
        // EPOLLHUP：本端描述符产生一个挂断事件，默认监测事件
        // EPOLLERR：表示对应的文件描述符发生错误
        if ev.events&(_EPOLLIN|_EPOLLRDHUP|_EPOLLHUP|_EPOLLERR) != 0 {
            mode += 'r'
        }
        // EPOLLOUT：表示对应的文件描述符可以写
        // EPOLLHUP：本端描述符产生一个挂断事件，默认监测事件
        // EPOLLERR：表示对应的文件描述符发生错误
        if ev.events&(_EPOLLOUT|_EPOLLHUP|_EPOLLERR) != 0 {
            mode += 'w'
        }
        // mode不为0，表示有IO事件，需要从ev.data字段得到与IO事件关联的pollDesc。
        if mode != 0 {
            // 取出保存在 epollevent 里的pollDesc，因为要根据这个内容去恢复g如果g已被挂起时
            pd := *(**pollDesc)(unsafe.Pointer(&ev.data)) // *pollDesc
            // 对应的_EPOLLERR文件描述符出现错误时，标记错误
            pd.setEventErr(ev.events == _EPOLLERR)	// 设置pollDesc的EpollErr错误位，如果是这种状态
            netpollready(&toRun, pd, mode)	// 处理就绪的描述符
        }
    }
    return toRun
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// netpollready is called by the platform-specific netpoll function.
// It declares that the fd associated with pd is ready for I/O.
// The toRun argument is used to build a list of goroutines to return
// from netpoll. The mode argument is 'r', 'w', or 'r'+'w' to indicate
// whether the fd is ready for reading or writing or both.
//
// This may run while the world is stopped, so write barriers are not allowed.
//go:nowritebarrier
func netpollready(toRun *gList, pd *pollDesc, mode int32) {
    var rg, wg *g
    if mode == 'r' || mode == 'r'+'w' {
        // netpollunblock 可能返回 goroutine 或 nil
        rg = netpollunblock(pd, 'r', true)
    }
    if mode == 'w' || mode == 'r'+'w' {
        wg = netpollunblock(pd, 'w', true)
    }
    if rg != nil {
        // 并入 toRun 中，这部分 goroutine 等待放入调度池中
        toRun.push(rg)
    }
    if wg != nil {
        toRun.push(wg)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

func netpollunblock(pd *pollDesc, mode int32, ioready bool) *g {
    // 根据 mode 从rg或wg取出 goroutine。
    gpp := &pd.rg
    if mode == 'w' {
        gpp = &pd.wg
    }

    for {
        // 原子读取
        old := gpp.Load()
        // pdReady：表示fd的数据已经就绪，可供读取或写。该值从g修改pdReady。
        // 这种情况可能 goroutine 已经被返回给调用者了。什么都不做直接返回。
        if old == pdReady {
            return nil
        }
        // nil：没有什么可做的
        if old == 0 && !ioready {
            // Only set pdReady for ioready. runtime_pollWait
            // will check for timeout/cancel before waiting.
            //
            // 只在ioready中设置pdReady。runtime_pollWait将在等待之前检查 timeout/cancel。
            return nil
        }
        // old是0、goroutine、pdWait这三种情况。
        var new uintptr
        if ioready {
            // 修改为pdReady，表示数据已就绪或写
            new = pdReady 
        }
        // CAS 交换
        if gpp.CompareAndSwap(old, new) {
            // pdWait：表示某个goroutine即将挂起并等待fd的可读可写事件。
            if old == pdWait {
                old = 0 // nil
            }
            return (*g)(unsafe.Pointer(old)) // nil 或 *g
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// wakeNetPoller wakes up the thread sleeping in the network poller if it isn't
// going to wake up before the when argument; or it wakes an idle P to service
// timers and the network poller if there isn't one already.
func wakeNetPoller(when int64) {
    // sched.lastpoll：记录的是上次执行netpoll的时间，如果等于0，则表示某个线程正在阻塞式地执行netpoll。
    // sched.lastpoll 被设置为0只会在 findrunnable 函数中。
    if atomic.Load64(&sched.lastpoll) == 0 {
        // In findrunnable we ensure that when polling the pollUntil
        // field is either zero or the time to which the current
        // poll is expected to run. This can have a spurious wakeup
        // but should never miss a wakeup.
        // 
        // 在 findrunnable 中，我们要确保轮询时 pollUntil 字段要么是0，要么为当前poll预期运行的时间。
        // 这里可能会是一个虚假的唤醒，但不应该错过唤醒。
        // sched.pollUntil：表示阻塞式地netpoll将在何时被唤醒。该值在 findrunnable 函数中被设置。
        // sched.pollUntil 值大于0时，表示最近的timer触发时间段。
        pollerPollUntil := int64(atomic.Load64(&sched.pollUntil))
        // pollerPollUntil > when：存在最新的计时器被加入when时间段后触发
        if pollerPollUntil == 0 || pollerPollUntil > when {
            netpollBreak()
        }
    } else {
        // There are no threads in the network poller, try to get
        // one there so it can handle new timers.
        if GOOS != "plan9" { // Temporary workaround - see issue #42303.
            wakep() // 尝试唤醒一个空闲的P起来服务timer和network poller。
        }
    }
}

1
2
3

func netpollarm(pd *pollDesc, mode int) {
	throw("runtime: unused") // runtime: 未使用的
}