Golang网络:核心API实现剖析二)

func (pd *pollDesc) Wait(mode int) error {
 res := runtime_pollWait(pd.runtimeCtx, mode)
 return convertErr(res)
}
func (pd *pollDesc) WaitRead() error {
 return pd.Wait('r')
}
func (pd *pollDesc) WaitWrite() error {
 return pd.Wait('w')
}

TEXT net·runtime_pollWait(SB),NOSPLIT,0ドル-0 
 JMP runtime·netpollWait(SB)

func netpollWait(pd *pollDesc, mode int) int {
 // 先检查该socket是否有error发生(如关闭、超时等) 
 err := netpollcheckerr(pd, int32(mode))
 if err != 0 {
 return err
 }
 // As for now only Solaris uses level-triggered IO. 
 if GOOS == "solaris" {
 onM(func() {
 netpollarm(pd, mode)
 })
 }
 // 循环等待netpollblock返回值为true 
 // 如果返回值为false且该socket未出现任何错误 
 // 那该协程可能被意外唤醒,需要重新被挂起 
 // 还有一种可能:该socket由于超时而被唤醒 
 // 此时netpollcheckerr就是用来检测超时错误的 
 for !netpollblock(pd, int32(mode), false) {
 err = netpollcheckerr(pd, int32(mode))
 if err != 0 {
 return err
 }
 }
 return 0 
}
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
 gpp := &pd.rg
 if mode == 'w' {
 gpp = &pd.wg
 }
 // set the gpp semaphore to WAIT 
 // 首先将轮询状态设置为pdWait 
 // 为什么要使用for呢?因为casuintptr使用了自旋锁 
 // 为什么使用自旋锁就要加for循环呢? 
 for {
 old := *gpp
 if old == pdReady {
 *gpp = 0 
 return true 
 }
 if old != 0 {
 gothrow("netpollblock: double wait")
 }
 // 将socket轮询相关的状态设置为pdWait 
 if casuintptr(gpp, 0, pdWait) {
 break 
 }
 }
 // 如果未出错将该协程挂起,解锁函数是netpollblockcommit 
 if waitio || netpollcheckerr(pd, mode) == 0 {
 f := netpollblockcommit
 gopark(**(**unsafe.Pointer)(unsafe.Pointer(&f)), unsafe.Pointer(gpp), "IO wait")
 }
 // 可能是被挂起的协程被唤醒 
 // 或者由于某些原因该协程压根未被挂起 
 // 获取其当前状态记录在old中 
 old := xchguintptr(gpp, 0)
 if old > pdWait {
 gothrow("netpollblock: corrupted state")
 }
 return old == pdReady
}

func netpoll(block bool) (gp *g) {
 if epfd == -1 {
 return 
 }
 waitms := int32(-1)
 if !block {
 // 如果调用者不希望block 
 // 设置waitsm为0 
 waitms = 0 
 }
 var events [128]epollevent
retry:
 // 调用epoll_wait获取就绪事件 
 n := epollwait(epfd, &events[0], int32(len(events)), waitms)
 if n < 0 {
 ...
 }
 goto retry
 }
 for i := int32(0); i < n; i++ {
 ev := &events[i]
 if ev.events == 0 {
 continue 
 }
 var mode int32 
 if ev.events&(_EPOLLIN|_EPOLLRDHUP|_EPOLLHUP|_EPOLLERR) != 0 {
 mode += 'r' 
 }
 if ev.events&(_EPOLLOUT|_EPOLLHUP|_EPOLLERR) != 0 {
 mode += 'w' 
 }
 // 对每个事件,调用了netpollready 
 // pd主要记录了与该socket关联的等待协程 
 if mode != 0 {
 pd := *(**pollDesc)(unsafe.Pointer(&ev.data))
 netpollready((**g)(noescape(unsafe.Pointer(&gp))), pd, mode)
 }
 }
 // 如果调用者同步等待且本次未获取到就绪socket 
 // 继续重试 
 if block && gp == nil {
 goto retry
 }
 return gp
}

// make pd ready, newly runnable goroutines (if any) are returned in rg/wg 
func netpollready(gpp **g, pd *pollDesc, mode 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)
 }
 // 将就绪协程添加至链表中 
 if rg != nil {
 rg.schedlink = *gpp
 *gpp = rg
 }
 if wg != nil {
 wg.schedlink = *gpp
 *gpp = wg
 }
}
// 将pollDesc的状态置为pdReady并返回就绪协程 
func netpollunblock(pd *pollDesc, mode int32, ioready bool) *g {
 gpp := &pd.rg
 if mode == 'w' {
 gpp = &pd.wg
 }
 for {
 old := *gpp
 if old == pdReady {
 return nil 
 }
 if old == 0 && !ioready {
 return nil 
 }
 var new uintptr 
 if ioready {
 new = pdReady
 }
 if casuintptr(gpp, old, new) {
 if old == pdReady || old == pdWait {
 old = 0 
 }
 return (*g)(unsafe.Pointer(old))
 }
 }
}

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传