golang将多路复异步io转成阻塞io的方法详解

2019-11-10 11:09:16王振洲

到这里所有的疑问都集中到了pollDesc上,它到底是什么呢?

const (
 pdReady uintptr = 1
 pdWait uintptr = 2
)

// Network poller descriptor.
type pollDesc struct {
 link *pollDesc // in pollcache, protected by pollcache.lock
 lock mutex // protects the following fields
 fd  uintptr
 closing bool
 seq  uintptr // protects from stale timers and ready notifications
 rg  uintptr // pdReady, pdWait, G waiting for read or nil
 rt  timer // read deadline timer (set if rt.f != nil)
 rd  int64 // read deadline
 wg  uintptr // pdReady, pdWait, G waiting for write or nil
 wt  timer // write deadline timer
 wd  int64 // write deadline
 user uint32 // user settable cookie
}

type pollCache struct {
 lock mutex
 first *pollDesc
}

pollDesc网络轮询器是Golang中针对每个socket文件描述符建立的轮询机制。 此处的轮询并不是一般意义上的轮询,而是Golang的runtime在调度goroutine或者GC完成之后或者指定时间之内,调用epoll_wait获取所有产生IO事件的socket文件描述符。当然在runtime轮询之前,需要将socket文件描述符和当前goroutine的相关信息加入epoll维护的数据结构中,并挂起当前goroutine,当IO就绪后,通过epoll返回的文件描述符和其中附带的goroutine的信息,重新恢复当前goroutine的执行。这里我们可以看到pollDesc中有两个变量wg和rg,其实我们可以把它们看作信号量,这两个变量有几种不同的状态:

pdReady:io就绪  pdWait:当前的goroutine正在准备挂起在信号量上,但是还没有挂起。 G pointer:当我们把它改为指向当前goroutine的指针时,当前goroutine挂起  

继续接着上面的WaitRead调用说起,go在这里到底做了什么让当前的goroutine挂起了呢。

func net_runtime_pollWait(pd *pollDesc, mode int) int {
 err := netpollcheckerr(pd, int32(mode))
 if err != 0 {
 return err
 }
 // As for now only Solaris uses level-triggered IO.
 if GOOS == "solaris" {
 netpollarm(pd, mode)
 }
 for !netpollblock(pd, int32(mode), false) {
 err = netpollcheckerr(pd, int32(mode))
 if err != 0 {
 return err
 }
 // Can happen if timeout has fired and unblocked us,
 // but before we had a chance to run, timeout has been reset.
 // Pretend it has not happened and retry.
 }
 return 0
}


// returns true if IO is ready, or false if timedout or closed
// waitio - wait only for completed IO, ignore errors
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
 //根据读写模式获取相应的pollDesc中的读写信号量
 gpp := &pd.rg
 if mode == 'w' {
 gpp = &pd.wg
 }

 for {
 old := *gpp
 //已经准备好直接返回true
 if old == pdReady {
 *gpp = 0
 return true
 }
 if old != 0 {
 throw("netpollblock: double wait")
 }
  //设置gpp pdWait
 if atomic.Casuintptr(gpp, 0, pdWait) {
 break
 }
 }

 if waitio || netpollcheckerr(pd, mode) == 0 {
 gopark(netpollblockcommit, unsafe.Pointer(gpp), "IO wait", traceEvGoBlockNet, 5)
 }

 old := atomic.Xchguintptr(gpp, 0)
 if old > pdWait {
 throw("netpollblock: corrupted state")
 }
 return old == pdReady
}