golang將多路復(fù)異步io轉(zhuǎn)成阻塞io的方法詳解

package main

import (
 "net"
)

func handleConnection(c net.Conn) {
 //讀寫數(shù)據(jù)
 buffer := make([]byte, 1024)
 c.Read(buffer)
 c.Write([]byte("Hello from server"))
}

func main() {
 l, err := net.Listen("tcp", "host:port")
 if err != nil {
 return
 }
 defer l.Close()
 for {
 c, err := l.Accept()
 if err!= nil {
 return
 }
 go handleConnection(c)
 }
}

// Multiple goroutines may invoke methods on a Conn simultaneously.
//官方注釋：多個(gè)goroutines可能同時(shí)調(diào)用方法在一個(gè)連接上，我的理解就是所謂的驚群效應(yīng)吧
//換句話說就是你多個(gè)goroutines監(jiān)聽同一個(gè)連接同一個(gè)事件，所有的goroutines都會(huì)觸發(fā)，
//這只是我的猜測(cè)，有待驗(yàn)證。
type Conn interface {
 Read(b []byte) (n int, err error)
 Write(b []byte) (n int, err error)
 Close() error
 LocalAddr() Addr
 RemoteAddr() Addr
 SetDeadline(t time.Time) error
 SetReadDeadline(t time.Time) error
 SetWriteDeadline(t time.Time) error
}

type conn struct {
 fd *netFD
}

// Network file descriptor.
type netFD struct {
 // locking/lifetime of sysfd + serialize access to Read and Write methods
 fdmu fdMutex

 // immutable until Close
 sysfd  int
 family  int
 sotype  int
 isConnected bool
 net   string
 laddr  Addr
 raddr  Addr

 // wait server
 pd pollDesc
}

func (fd *netFD) accept() (netfd *netFD, err error) {
 //................
 for {
 s, rsa, err = accept(fd.sysfd)
 if err != nil {
 nerr, ok := err.(*os.SyscallError)
 if !ok {
 return nil, err
 }
 switch nerr.Err {
 /* 如果錯(cuò)誤是EAGAIN說明Socket的緩沖區(qū)為空，未讀取到任何數(shù)據(jù)
    則調(diào)用fd.pd.WaitRead，*/
 case syscall.EAGAIN:
 if err = fd.pd.waitRead(); err == nil {
  continue
 }
 case syscall.ECONNABORTED:
 continue
 }
 return nil, err
 }
 break
 }
 //.........
 //代碼過長(zhǎng)不再列出，感興趣看go的源碼，runtime 下的fd_unix.go
 return netfd, nil
}

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
}

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 {
 //根據(jù)讀寫模式獲取相應(yīng)的pollDesc中的讀寫信號(hào)量
 gpp := pd.rg
 if mode == 'w' {
 gpp = pd.wg
 }

 for {
 old := *gpp
 //已經(jīng)準(zhǔn)備好直接返回true
 if old == pdReady {
 *gpp = 0
 return true
 }
 if old != 0 {
 throw("netpollblock: double wait")
 }
  //設(shè)置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
}

func netpoll(block bool) *g {
 if epfd == -1 {
 return nil
 }
 waitms := int32(-1)
 if !block {
 waitms = 0
 }
 var events [128]epollevent
retry:
 //每次最多監(jiān)聽128個(gè)事件
 n := epollwait(epfd, events[0], int32(len(events)), waitms)
 if n  0 {
 if n != -_EINTR {
 println("runtime: epollwait on fd", epfd, "failed with", -n)
 throw("epollwait failed")
 }
 goto retry
 }
 var gp guintptr
 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'
 }
 if mode != 0 {
  //把epoll中的data轉(zhuǎn)換成pollDesc
 pd := *(**pollDesc)(unsafe.Pointer(ev.data))
 netpollready(gp, pd, mode)
 }
 }
 if block  gp == 0 {
 goto retry
 }
 return gp.ptr()
}