异步 I/O 探秘 —— 为什么说 Go 为我们提供了同步的网络编程接口 ？

package main
import (  "fmt"  "net")
func main() {  listen, err := net.Listen("tcp", "localhost:8080")  ...  for {    conn, err := listen.Accept()    ...    go handleConnection(conn)  }}
func handleConnection(conn net.Conn) {  defer conn.Close()
  buffer := make([]byte, 1024)  for {    n, err := conn.Read(buffer)    if err != nil {      fmt.Println("Error reading:", err.Error())      return    }    data := buffer[:n]    fmt.Println("Received data:", string(data))  }}

type pollDesc struct {  runtimeCtx uintptr}
var serverInit sync.Once
func (pd *pollDesc) init(fd *FD) error {  serverInit.Do(runtime_pollServerInit)  ctx, errno := runtime_pollOpen(uintptr(fd.Sysfd))  if errno != 0 {    return errnoErr(syscall.Errno(errno))  }  pd.runtimeCtx = ctx  return nil}

//go:linkname poll_runtime_pollServerInit internal/poll.runtime_pollServerInitfunc poll_runtime_pollServerInit() {  netpollGenericInit()}

func netpollinit() {  epfd = epollcreate1(_EPOLL_CLOEXEC)  if epfd < 0 {    epfd = epollcreate(1024)    if epfd < 0 {      println("runtime: epollcreate failed with", -epfd)      throw("runtime: netpollinit failed")    }    closeonexec(epfd)  }  r, w, errno := nonblockingPipe()  if errno != 0 {    println("runtime: pipe failed with", -errno)    throw("runtime: pipe failed")  }  ev := epollevent{    events: _EPOLLIN,  }  *(**uintptr)(unsafe.Pointer(&ev.data)) = &netpollBreakRd  errno = epollctl(epfd, _EPOLL_CTL_ADD, r, &ev)  if errno != 0 {    println("runtime: epollctl failed with", -errno)    throw("runtime: epollctl failed")  }  netpollBreakRd = uintptr(r)  netpollBreakWr = uintptr(w)}

net.TCPListener    --->net.TCPListener.Accept        --->net.TCPListener.accept            --->net.netFD.accept                ---->net.netFD.init                    ----->poll.FD.Init                        ------>poll.pollDesc.init                            ------->poll.runtime_pollOpen                                -------->runtime.netpollopen                                    --------->epollctl

// Wrapper around the accept system call that marks the returned file// descriptor as nonblocking and close-on-exec.func accept(s int) (int, syscall.Sockaddr, string, error) {  ns, sa, err := Accept4Func(s, syscall.SOCK_NONBLOCK|syscall.SOCK_CLOEXEC)  if err != nil {    return -1, sa, "accept4", err  }  return ns, sa, "", nil}

net.ListenTCP    --->net.sysListener.listenTCP        --->net.internetSocket            --->net.socket                ---->net.netFD.dial                    ----->net.netFD.init                        ------>poll.FD.Init                            ------->poll.pollDesc.init                                -------->poll.runtime_pollOpen                                    --------->runtime.netpollopen                                        --------->epollctl

net.Dial    --->net.Dialer.Dial        --->net.Dialer.DialContext            ---->net.sysDialer.dialSerial                ----->net.sysDialer.dialSingle                    ------>net.sysDialer.dialTCP                        ------->net.sysDialer.doDialTCP                            -------->net.internetSocket                                --------->net.socket                                    ---------->net.netFD.dial                                        ----------->net.netFD.connect                                            ------------>poll.FD.Init                                                ------------->poll.pollDesc.init                                                    -------------->poll.runtime_pollOpen                                                        --------------->runtime.netpollopen                                                            ---------------->epollctl

func netpollopen(fd uintptr, pd *pollDesc) int32 {  var ev epollevent  ev.events = _EPOLLIN | _EPOLLOUT | _EPOLLRDHUP | _EPOLLET  *(**pollDesc)(unsafe.Pointer(&ev.data)) = pd  return -epollctl(epfd, _EPOLL_CTL_ADD, int32(fd), &ev)}

type epollevent struct {  events uint32  data   [8]byte // unaligned uintptr}

*(**pollDesc)(unsafe.Pointer(&ev.data)) = pd

// 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 nanosecondsfunc netpoll(delay int64) gList {  if epfd == -1 {    return gList{}  }  var waitms int32  if delay < 0 {    waitms = -1  } else if delay == 0 {    waitms = 0  } else if delay < 1e6 {    waitms = 1  } else if delay < 1e15 {    waitms = int32(delay / 1e6)  } else {    // An arbitrary cap on how long to wait for a timer.    // 1e9 ms == ~11.5 days.    waitms = 1e9  }  // 准备 epollevent 数组，最多 poll 128个事件  var events [128]epolleventretry:  // 发出 epoll_wait 调用  n := epollwait(epfd, &events[0], int32(len(events)), waitms)  ...  // toRun 是个 goroutine 链表，表示文件描述符已就绪，可以被调度运行的 goroutine  var toRun gList  // 遍历本次 epoll_wait 返回的就绪事件  for i := int32(0); i < n; i++ {    ev := &events[i]    if ev.events == 0 {      continue    }
    if *(**uintptr)(unsafe.Pointer(&ev.data)) == &netpollBreakRd {      ...    }
    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 {      // 从 ev.data 中取出 pollDesc      pd := *(**pollDesc)(unsafe.Pointer(&ev.data))      pd.setEventErr(ev.events == _EPOLLERR)      // 获取相关 goroutine 并插入到 toRun 链表      netpollready(&toRun, pd, mode)    }  }  return toRun}

// 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:nowritebarrierfunc netpollready(toRun *gList, pd *pollDesc, mode int32) {  var rg, wg *g  if mode == 'r' || mode == 'r'+'w' {        // 获取等待读操作的 g    rg = netpollunblock(pd, 'r', true)  }  if mode == 'w' || mode == 'r'+'w' {        // 获取等待写操作的 g    wg = netpollunblock(pd, 'w', true)  }  if rg != nil {        // 将等待读操作的 g 插入链表    toRun.push(rg)  }  if wg != nil {        // 将等待写操作的 g 插入链表    toRun.push(wg)  }}

type pollDesc struct {  ...  // rg, wg are accessed atomically and hold g pointers.  // (Using atomic.Uintptr here is similar to using guintptr elsewhere.)  rg atomic.Uintptr // pdReady, pdWait, G waiting for read or nil  wg atomic.Uintptr // pdReady, pdWait, G waiting for write or nil  ...}

func netpollunblock(pd *pollDesc, mode int32, ioready bool) *g {  gpp := &pd.rg  if mode == 'w' {    gpp = &pd.wg  }
  for {    old := gpp.Load()    if old == pdReady {      return nil    }    if old == 0 && !ioready {      // Only set pdReady for ioready. runtime_pollWait      // will check for timeout/cancel before waiting.      return nil    }    var new uintptr    if ioready {      new = pdReady    }    // 将 gpp 设置为 pdReady    if gpp.CompareAndSwap(old, new) {      if old == pdWait {        // 如果设置为0，则 (*g)(unsafe.Pointer(old)) 为 nil        old = 0      }      // 将 old 的值转换为 *g 返回，old 通常就是发生等待的 goroutine 地址      return (*g)(unsafe.Pointer(old))    }  }}

// A gList is a list of Gs linked through g.schedlink. A G can only be// on one gQueue or gList at a time.type gList struct {  head guintptr}
// push adds gp to the head of l.func (l *gList) push(gp *g) {  gp.schedlink = l.head  l.head.set(gp)}

// The main goroutine.func main() {  ...  if GOARCH != "wasm" { // no threads on wasm yet, so no sysmon    systemstack(func() {      newm(sysmon, nil, -1)    })  }  ...}

// Always runs without a P, so write barriers are not allowed.////go:nowritebarrierrecfunc sysmon() {  ...  for {    ...    // poll network if not polled for more than 10ms    lastpoll := int64(atomic.Load64(&sched.lastpoll))    if netpollinited() && lastpoll != 0 && lastpoll+10*1000*1000 < now {      atomic.Cas64(&sched.lastpoll, uint64(lastpoll), uint64(now))      list := netpoll(0) // non-blocking - returns list of goroutines      if !list.empty() {        // Need to decrement number of idle locked M's        // (pretending that one more is running) before injectglist.        // Otherwise it can lead to the following situation:        // injectglist grabs all P's but before it starts M's to run the P's,        // another M returns from syscall, finishes running its G,        // observes that there is no work to do and no other running M's        // and reports deadlock.        incidlelocked(-1)        injectglist(&list)        incidlelocked(1)      }    }  ...  }}

// Read implements io.Reader.func (fd *FD) Read(p []byte) (int, error) {  if err := fd.readLock(); err != nil {    return 0, err  }  defer fd.readUnlock()  ...  if err := fd.pd.prepareRead(fd.isFile); err != nil {    return 0, err  }  if fd.IsStream && len(p) > maxRW {    p = p[:maxRW]  }  for {    // 发出 Read 系统调用    n, err := ignoringEINTRIO(syscall.Read, fd.Sysfd, p)    if err != nil {      n = 0      // 如果 收到 EAGAIN 错误，且文件描述符是可 poll 的，则进入等待      if err == syscall.EAGAIN && fd.pd.pollable() {        if err = fd.pd.waitRead(fd.isFile); err == nil {          continue        }      }    }    err = fd.eofError(n, err)    return n, err  }}

// returns true if IO is ready, or false if timedout or closed// waitio - wait only for completed IO, ignore errors// Concurrent calls to netpollblock in the same mode are forbidden, as pollDesc// can hold only a single waiting goroutine for each mode.func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {  gpp := &pd.rg  if mode == 'w' {    gpp = &pd.wg  }
  // set the gpp semaphore to pdWait  for {    // Consume notification if already ready.    if gpp.CompareAndSwap(pdReady, 0) {      return true    }    if gpp.CompareAndSwap(0, pdWait) {      break    }
    // Double check that this isn't corrupt; otherwise we'd loop    // forever.    if v := gpp.Load(); v != pdReady && v != 0 {      throw("runtime: double wait")    }  }
  // need to recheck error states after setting gpp to pdWait  // this is necessary because runtime_pollUnblock/runtime_pollSetDeadline/deadlineimpl  // do the opposite: store to closing/rd/wd, publishInfo, load of rg/wg  if waitio || netpollcheckerr(pd, mode) == pollNoError {    // 开启 park 流程，休眠当前 goroutine    gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5)  }  // be careful to not lose concurrent pdReady notification  old := gpp.Swap(0)  if old > pdWait {    throw("runtime: corrupted polldesc")  }  return old == pdReady}

// Puts the current goroutine into a waiting state and calls unlockf on the// system stack.//// If unlockf returns false, the goroutine is resumed.//// unlockf must not access this G's stack, as it may be moved between// the call to gopark and the call to unlockf.//// Note that because unlockf is called after putting the G into a waiting// state, the G may have already been readied by the time unlockf is called// unless there is external synchronization preventing the G from being// readied. If unlockf returns false, it must guarantee that the G cannot be// externally readied.//// Reason explains why the goroutine has been parked. It is displayed in stack// traces and heap dumps. Reasons should be unique and descriptive. Do not// re-use reasons, add new ones.func gopark(unlockf func(*g, unsafe.Pointer) bool, lock unsafe.Pointer, reason waitReason, traceEv byte, traceskip int) {  if reason != waitReasonSleep {    checkTimeouts() // timeouts may expire while two goroutines keep the scheduler busy  }  mp := acquirem()  gp := mp.curg  status := readgstatus(gp)  if status != _Grunning && status != _Gscanrunning {    throw("gopark: bad g status")  }  // 此时 lock 是 gpp(pollDesc 中的 rg)  mp.waitlock = lock  // 此时 unlockf 是 netpollblockcommit  mp.waitunlockf = unlockf  gp.waitreason = reason  mp.waittraceev = traceEv  mp.waittraceskip = traceskip  releasem(mp)  // can't do anything that might move the G between Ms here.  mcall(park_m)}

// func mcall(fn func(*g))// Switch to m->g0's stack, call fn(g).// Fn must never return. It should gogo(&g->sched)// to keep running g.TEXT runtime·mcall<ABIInternal>(SB), NOSPLIT, $0-8  MOVQ  AX, DX  // DX = fn
  // save state in g->sched  // 下面这 5 行指令是保存当前 goroutine 的状态到 g->sched   MOVQ  0(SP), BX  // caller's PC  MOVQ  BX, (g_sched+gobuf_pc)(R14)  LEAQ  fn+0(FP), BX  // caller's SP  MOVQ  BX, (g_sched+gobuf_sp)(R14)  MOVQ  BP, (g_sched+gobuf_bp)(R14)
  // switch to m->g0 & its stack, call fn  // 准备切换到 g0 及其堆栈  MOVQ  g_m(R14), BX    // 通过 R14 拿到当前的 m  MOVQ  m_g0(BX), SI  // SI = g.m.g0 通过 m 拿到 m 的 g0  CMPQ  SI, R14  // if g == m->g0 call badmcall  JNE  goodm  JMP  runtime·badmcall(SB)goodm:  MOVQ  R14, AX    // AX (and arg 0) = g 把当前 goroutine 地址放入 AX  MOVQ  SI, R14    // g = g.m.g0  SI 是 g.m.g0，这一句将 goroutine 切换到 g0  get_tls(CX)    // Set G in TLS  MOVQ  R14, g(CX)  // 设置 g0 到线程本地存储  MOVQ  (g_sched+gobuf_sp)(R14), SP  // sp = g0.sched.sp 切换到 g0 堆栈  PUSHQ  AX  // open up space for fn's arg spill slot  MOVQ  0(DX), R12  CALL  R12    // fn(g)  POPQ  AX  JMP  runtime·badmcall2(SB)  RET

RAX, RBX, RCX, RDI, RSI, R8, R9, R10, R11

// park continuation on g0.func park_m(gp *g) {  _g_ := getg()
  if trace.enabled {    traceGoPark(_g_.m.waittraceev, _g_.m.waittraceskip)  }
  casgstatus(gp, _Grunning, _Gwaiting)  dropg()
  if fn := _g_.m.waitunlockf; fn != nil {    // 调用 netpollblockcommit    ok := fn(gp, _g_.m.waitlock)    _g_.m.waitunlockf = nil    _g_.m.waitlock = nil    if !ok {      if trace.enabled {        traceGoUnpark(gp, 2)      }      casgstatus(gp, _Gwaiting, _Grunnable)      execute(gp, true) // Schedule it back, never returns.    }  }  // 进入调度循环，不再返回  schedule()}

func netpollblockcommit(gp *g, gpp unsafe.Pointer) bool {  r := atomic.Casuintptr((*uintptr)(gpp), pdWait, uintptr(unsafe.Pointer(gp)))  if r {    // Bump the count of goroutines waiting for the poller.    // The scheduler uses this to decide whether to block    // waiting for the poller if there is nothing else to do.    atomic.Xadd(&netpollWaiters, 1)  }  return r}