gh-liu
commented
5 months ago usage: 网络I/O、文件I/O
type pollDesc struct {
runtimeCtx uintptr
}
var serverInit sync.Once
func (pd *pollDesc) init(fd *FD) error {
serverInit.Do(runtime_pollServerInit) // 在初始化 pollDesc 时，初始化网络轮询器(通过sync.Once仅执行一次)
ctx, errno := runtime_pollOpen(uintptr(fd.Sysfd)) // 加入轮询事件
if errno != 0 {
return errnoErr(syscall.Errno(errno))
}
pd.runtimeCtx = ctx
return nil
}net/fd_posix.go
// Network file descriptor.
type netFD struct {
pfd poll.FD
// ...
}
func newFD(sysfd, family, sotype int, net string) (*netFD, error) {
ret := &netFD{
pfd: poll.FD{
Sysfd: sysfd,
// ...
},
// ...
}
return ret, nil
}
func (fd *netFD) init() error {
return fd.pfd.Init(fd.net, true) // 初始化 pfd，也就是 poll.FD
}
func (fd *netFD) connect(ctx context.Context, la, ra syscall.Sockaddr) (rsa syscall.Sockaddr, ret error) {
// ...
if err := fd.pfd.Init(fd.net, true); err != nil { // 初始化 pfd，也就是 poll.FD
return nil, err
}
// ...
}
os/types.go
// File represents an open file descriptor.
type File struct {
*file // os specific
}
// file is the real representation of *File.
// The extra level of indirection ensures that no clients of os
// can overwrite this data, which could cause the finalizer
// to close the wrong file descriptor.
type file struct {
pfd poll.FD
// ...
}
func newFile(fd int, name string, kind newFileKind) *File {
f := &File{&file{
pfd: poll.FD{
Sysfd: fd,
// ...
},
// ...
}}
// ...
if pollErr := f.pfd.Init("file", pollable); pollErr != nil && clearNonBlock {} // 初始化 pfd，也就是 poll.FD
// ...
return f
}usage: 计时器
// doaddtimer adds t to the current P's heap.
// The caller must have locked the timers for pp.
func doaddtimer(pp *p, t *timer) {
// Timers rely on the network poller, so make sure the poller
// has started.
if netpollInited.Load() == 0 {
netpollGenericInit() // 初始化网络轮询器
}
// ...
}网络轮询器负责处理文件和网络 I/O 的截止日期
// FD is a file descriptor. The net and os packages use this type as a
// field of a larger type representing a network connection or OS file.
type FD struct {
// ...
// I/O poller.
pd pollDesc
// ...
}
// SetDeadline sets the read and write deadlines associated with fd.
func (fd *FD) SetDeadline(t time.Time) error {
return setDeadlineImpl(fd, t, 'r'+'w')
}
// SetReadDeadline sets the read deadline associated with fd.
func (fd *FD) SetReadDeadline(t time.Time) error {
return setDeadlineImpl(fd, t, 'r')
}
// SetWriteDeadline sets the write deadline associated with fd.
func (fd *FD) SetWriteDeadline(t time.Time) error {
return setDeadlineImpl(fd, t, 'w')
}
func setDeadlineImpl(fd *FD, t time.Time, mode int) error {
var d int64
// ...
runtime_pollSetDeadline(fd.pd.runtimeCtx, d, mode)
return nil
}
// 链接到 poll_runtime_pollSetDeadline
func runtime_pollSetDeadline(ctx uintptr, d int64, mode int){}
// 设置超时时间
// 1. 先使用截止日期计算出过期的时间点
// 2. 根据 pd 的状态做出不同处理
// 2.1 计时器没有设置执行函数，则设置函数并调用 resettimer 重置计时器
// 2.2 截至日期发生改变
// 2.2.1 改变大于0，则调用 modtimer 修改计时器
// 2.2.2 改变小于0，则调用 deltimer 删除计时器
// 3. 最后会重新检查截止日期，决定是否唤醒 G
//
//go:linkname poll_runtime_pollSetDeadline internal/poll.runtime_pollSetDeadline
func poll_runtime_pollSetDeadline(pd *pollDesc, d int64, mode int) {
// ...
rd0, wd0 := pd.rd, pd.wd
combo0 := rd0 > 0 && rd0 == wd0
if d > 0 {
d += nanotime()
if d <= 0 {
// If the user has a deadline in the future, but the delay calculation
// overflows, then set the deadline to the maximum possible value.
d = 1<<63 - 1
}
}
if mode == 'r' || mode == 'r'+'w' {
pd.rd = d
}
if mode == 'w' || mode == 'r'+'w' {
pd.wd = d
}
pd.publishInfo()
combo := pd.rd > 0 && pd.rd == pd.wd
rtf := netpollReadDeadline
if combo {
rtf = netpollDeadline
}
if pd.rt.f == nil {
if pd.rd > 0 {
pd.rt.f = rtf
// Copy current seq into the timer arg.
// Timer func will check the seq against current descriptor seq,
// if they differ the descriptor was reused or timers were reset.
pd.rt.arg = pd.makeArg()
pd.rt.seq = pd.rseq
resettimer(&pd.rt, pd.rd)
}
} else if pd.rd != rd0 || combo != combo0 {
pd.rseq++ // invalidate current timers
if pd.rd > 0 {
modtimer(&pd.rt, pd.rd, 0, rtf, pd.makeArg(), pd.rseq)
} else {
deltimer(&pd.rt)
pd.rt.f = nil
}
}
if pd.wt.f == nil {
if pd.wd > 0 && !combo {
pd.wt.f = netpollWriteDeadline
pd.wt.arg = pd.makeArg()
pd.wt.seq = pd.wseq
resettimer(&pd.wt, pd.wd)
}
} else if pd.wd != wd0 || combo != combo0 {
pd.wseq++ // invalidate current timers
if pd.wd > 0 && !combo {
modtimer(&pd.wt, pd.wd, 0, netpollWriteDeadline, pd.makeArg(), pd.wseq)
} else {
deltimer(&pd.wt)
pd.wt.f = nil
}
}
// If we set the new deadline in the past, unblock currently pending IO if any.
// Note that pd.publishInfo has already been called, above, immediately after modifying rd and wd.
delta := int32(0)
var rg, wg *g
if pd.rd < 0 {
rg = netpollunblock(pd, 'r', false, &delta)
}
if pd.wd < 0 {
wg = netpollunblock(pd, 'w', false, &delta)
}
unlock(&pd.lock)
if rg != nil {
netpollgoready(rg, 3)
}
if wg != nil {
netpollgoready(wg, 3)
}
netpollAdjustWaiters(delta)
}
network poller
实现
网络轮询器，与平台无关，具体实现需要定义如下函数：
netpollinit(): 初始化网络轮询器，仅调用一次netpollopen(fd uintptr, pd *pollDesc) int32: 为文件描述符 fd 启用边缘触发通知，pd 参数用于在 fd 就绪时传递给 netpollready，返回一个 errno 值netpollclose(fd uintptr) int32: 禁用文件描述符 fd 的通知，返回一个 errno 值netpoll(delta int64) (gList, int32): 进行网络轮询，如果 delta < 0，则无限期阻塞，如果 delta == 0，则不阻塞，如果 delta > 0，则阻塞至多 delta 纳秒，返回通过调用 netpollready 构建的 goroutine 列表，以及要添加到 netpollWaiters 的 delta。这永远不会返回一个非空列表和一个非零的 delta。netpollBreak(): 唤醒网络轮询器，假设它在 netpoll 中被阻塞netpollIsPollDescriptor(fd uintptr) bool: 报告 fd 是否是轮询器使用的文件描述符epoll 实现
初始化加入待监控的任务获取触发的事件无关实现