當前位置：
  1. 首頁
  2. 代理加盟
  3. 資訊
  4. 創業項目
  5. 正文

前面一篇文章提到，Go內置的 net/http中使用了Blocking IO，主要體現在兩層 for 循環。但真的是這樣嗎？

本文我們看看 Go net庫中 Server.ListenAndServe 的實現細節。

net.Listen("tcp", addr) 方法通過系統調用 socket、bind、listen 生成 net.Listener 對象，在后面的for 循環中，通過系統調用 accept 等待新的tcp conn，將其包裝成一個 conn 對象，在新的 goroutine 中對這個conn進行處理。這里是典型的 per goroutine per connection 模型。這個環節看起來是阻塞的，但創建 socket 時設置了syscall.SOCK_NONBLOCK，對后來有什么影響？

// net/http/server.go struct Server
func (srv *Server) ListenAndServe() error {
  ln, err := net.Listen("tcp", addr)
  // ... 省略部分代碼
  return srv.Serve(ln)
}

func (srv *Server) Serve(l net.Listener) Error {
  for {
    // ...
    rw, err := l.Accept()
    // ...
    c := srv.newConn(rw)
    c.setState(c.rwc, StateNew, runHooks) // before Serve can return
    go c.serve(connCtx)
}

net.Listener

net.Listen 觸發一系列的系統調用（主要是 socket、bind、listen），生成一個 net.Listener 對象。這個函數創建兩類Listener: TCP 支持跨機器的網絡通信，UNIX支持本機的多進程通信。

func (lc *ListenConfig) Listen(ctx context.Context, network, address string) (Listener, error) {
  // ... 省略部分代碼
  var l Listener
  la := addrs.first(isIPV4)
  switch la := la.(type) {
  case *TCPAddr:
    l, err = sl.listenTCP(ctx, la)
  case *UnixAddr:
    l, err = sl.listenUnix(ctx, la)
  // ... 省略部分代碼
}

由于兩者都是先觸發 syscall.Socket，我們從 socket 系統調用的視角來看兩者的區別。

// https://man7.org/Linux/man-pages/man2/socket.2.html
#include <sys/socket.h>
int socket(int family, int type, int protocol);

socket() 創建一個用于網絡通信的endpoint，并返回對應的套接字，也叫socket File descriptor。它是一個 int 值，Linux C代碼里一般用 sockfd 作為變量名，而 Go net庫里一般用 fd 作為變量名。

第一個參數 family 參數用來指定通信的協議族(protocol family)，常用的Enum值有:

  1. AF_UNIX/AF_LOCAL: Unix域協議, 用于本機的進程間通信
  2. AF_INET: IPV4協議
  3. AF_INET6: IPV6協議
  4. AF_ROUTE: 路由套接字
  5. 全量Enum定義在linux <sys/socket.h> 下

第二個參數 type 參數用來指定通信語義，常用enum值有:

  1. SOCK_STREAM=1: 基于TCP, 提供有序、可靠、雙向、基于連接的字節流，不限制消息長度，支持消息的優先級傳輸;
  2. SOCK_DGRAM=2: 基于UDP, 支持數據報，不是基于連接的、不保證可靠性，且消息的最大長度是固定的;
  3. SOCK_RAW=3: 支持通過原始的網絡協議訪問
  4. SOCK_RDM=4:
  5. SOCK_SEQPACKET=5: 基于TCP, 提供有序、可靠、雙向、基于連接的字節流，但消息的最大長度是固定的，超出的部分會被丟棄;

除了這幾個，還有兩個enum值在 Go net/http 被用到了，分別是:

  1. SOCK_NONBLOCK: 設置 accept 和 read/write操作為 O_NONBLOCK, 對應的場景有：
  • 接收連接 accept: 同步模式下沒有新連接時, 線程會被休眠, 異步模式下會返回EWOULDBLOCK/EAGAIN錯誤
  • read類操作: 同步模式下socket緩沖區沒有數據可讀時, 線程會被休眠, 異步模式下會返回EWOULDBLOCK/EAGAIN錯誤
  • write類操作: 同步模式下socket緩沖區已滿無法寫入時, 線程會被休眠, 異步模式下會返回EWOULDBLOCK/EAGAIN錯誤
  1. SOCK_CLOEXEC: 由于fork時，子進程默認拷貝父進程的數據空間、堆、棧等，當然也包含socket, 通過設置這個flag, 可以保證fork出來的子進程不持有父進程創建的socket.

第三個參數 protocol 指定通信協議，對于domain=AF_INET/AF_INET6來說，常見的enum值有 IPPROTO_TCP IPPROTO_UDP，全量.

socket() 返回一個 socket file descriptor，但并沒有協議和地址與其關聯。對于tcp client端而言，可以由系統隨機指定一個端口；對于一個 tcp server 而言，必須設置一個公開可訪問的ip地址和端口。bind函數實現了這個功能:

#include <sys/socket.h>

int bind(int sockfd, const struct sockaddr *addr, socklen_t addrlen);

// sockaddr 包含
struct sockaddr {
  sa_family_t sa_family;
  char        sa_data[14];
}

其中 sockfd 參數是 socket函數的返回值，后面兩個參數指定協議類型和地址。

當socket被創建以后，它并不能被動地接收創建連接請求，此時它只能作為一個client使用。要被動地接收請求，轉化為server，需要依賴 listen函數。該函數調用以后，sockfd的狀態會從 closed 轉換為 listen (netstat 命令可以進行查看)。listen函數的聲明如下:

#include <sys/socket.h>

int listen(int sockfd, int backlog);

第一個參數 sockfd 是 socket函數的返回值，第二個參數指定了處于ESTABLISHED狀態的sockets的隊列大小(從Linux 2.2起), 而不是處于 SYNC_RCVD狀態的sockets隊列的大小。這里提到的兩個狀態在TCP連接的三次握手中有所定義:

瀏覽器errconnectionrefused免費指紋瀏覽器插圖

tcp三次握手

backlog 默認值是0x80即128，通常可以配置在文件 /proc/sys/net/ipv4/tcp_max_syn_backlog 中，同時受到/proc/sys/net/core/somaxconn 的限制。在 Go 中，這個參數可以通過 func maxListenerBacklog() int 獲取。如果隊列滿了，Client端會收到 ECONNREFUSED 錯誤，即 connection refused。

小結一下，Linux操作系統層面創建一個tcp server，走的邏輯是：

  1. socket函數創建一個套接字 sockfd，默認狀態是 Closed
  2. bind函數綁定 sockfd 與特定的協議地址，比如 tcp 0.0.0.0:8080
  3. listen函數修改sockfd的狀態為 LISTEN，內核開始監聽套接字，三次握手建立連接

回到 Go net 庫的處理流程，我們關注的函數是 sysListener.listenTCP:

// net/tcpsock_posix.go struct sysListener
func (sl *sysListener) listenTCP(ctx context.Context, laddr *TCPAddr) (*TCPListener, error) {
   fd, err := internetSocket(ctx, sl.network, laddr, nil, syscall.SOCK_STREAM, 0, "listen", sl.ListenConfig.Control)
   if err != nil {
      return nil, err
   }
   return &TCPListener{fd: fd, lc: sl.ListenConfig}, nil
}

對于一個 tcp server，實例 sl 可以被這樣賦值，系統調用被封裝在函數 internetSocket 里:

sl := &sysListener{
  ListenConfig: *lc,     // lc 是默認值
  network:      network, // string "tcp"
  address:      address, // string "0.0.0.0:8080"
}

對于一個 tcp server，函數 internetSocket 接收到的參數可以是:

func internetSocket(
  ctx context.Context,  // context.Background()
  net string,           // "tcp"
  laddr sockaddr,       // &TCPAddr{IP:"0.0.0.0",Port:8080,Zone:"不知道是啥"}, DNS服務讀取的地址
  raddr sockaddr,       // nil, os=aix|windows|openbsd && mode="dail" 才需要
  sotype int,           // syscall.SOCK_STREAM
  proto int,            // 0
  mode string,          // "listen"
  ctrlFn func(string, string, syscall.RawConn) error // sl.ListenConfig.Control
) (fd *netFD, err error) {

函數 internetSocket 同樣只是一層封裝，內部調用的是函數 socket。函數socket內部按照順序調用了socket/bind/listen，返回一個套接字，這個套接字使用network poller，支持異步IO。函數 socket的主要邏輯如下:

// 通過sysSocket執行socket系統調用
// 返回一個套接字s
s, err := sysSocket(family, sotype, proto)

// 封裝int類型的套接字為一個結構體
if fd, err = newFD(s, family, sotype, net); err != nil {
// ...省略部分代碼

// 對于 SOCK_STREAM, SOCK_SEQPACKET類型，調用bind和listen
if laddr != nil && raddr == nil {
  switch sotype {
  case syscall.SOCK_STREAM, syscall.SOCK_SEQPACKET:
    if err := fd.listenStream(laddr, listenerBacklog(), ctrlFn); err != nil {
      fd.Close()
      return nil, err
    }
    return fd, nil

對于linux tcp server 而言，函數 sysSocket 的關鍵只有一行代碼:

s, err := socketFunc(family, sotype|syscall.SOCK_NONBLOCK|syscall.SOCK_CLOEXEC, proto)

SOCK_STREAM、SOCK_NONBLOCK 和 SOCK_CLOEXEC 的語義前面已經講過，不再贅述。

socketFunc 定義存放在 net/hook_unix.go 里，與listenFunc在一塊:

// Placeholders for socket system calls.
socketFunc        func(int, int, int) (int, error)  = syscall.Socket
connectFunc       func(int, syscall.Sockaddr) error = syscall.Connect
listenFunc        func(int, int) error              = syscall.Listen
getsockoptIntFunc func(int, int, int) (int, error)  = syscall.GetsockoptInt

通過sysSocket拿到套接字以后，通過函數newFD將其封裝成一個結構體，類型是 *net.netFD:

func newFD(sysfd, family, sotype int, net string) (*netFD, error) {
   ret := &netFD{
      pfd: poll.FD{
         Sysfd:         sysfd,
         IsStream:      sotype == syscall.SOCK_STREAM,
         ZeroReadIsEOF: sotype != syscall.SOCK_DGRAM && sotype != syscall.SOCK_RAW,
      },
      family: family,
      sotype: sotype,
      net:    net,
   }
   return ret, nil
}

其中，結構體內部poll.FD定義了讀寫的邏輯，它封裝了6個系統調用:

readSyscallName     = "read"
readFromSyscallName = "recvfrom"
readMsgSyscallName  = "recvmsg"
writeSyscallName    = "write"
writeToSyscallName  = "sendto"
writeMsgSyscallName = "sendmsg"

在創建套接字時，已經設置了 SOCK_NONBLOCK flag，如果沒有可用的連接，讀寫數據時，會收到 EWOULDBLOCK/EAGAIN 錯誤。Go net庫的處理是等待一段時間，我們看其中一個例子:

// ReadMsgInet4 is ReadMsg, but specialized for syscall.SockaddrInet4.
func (fd *FD) ReadMsgInet4(p []byte, oob []byte, flags int, sa4 *syscall.SockaddrInet4) (int, int, int, error) {
  if err := fd.readLock(); err != nil {
    return 0, 0, 0, err
  }
  defer fd.readUnlock()
  if err := fd.pd.prepareRead(fd.isFile); err != nil {
    return 0, 0, 0, err
  }
  for {
    n, oobn, sysflags, err := unix.RecvmsgInet4(fd.Sysfd, p, oob, flags, sa4)
    if err != nil {
        if err == syscall.EINTR {
          continue
        }
        // TODO(dfc) should n and oobn be set to 0
        if err == syscall.EAGAIN && fd.pd.pollable() {
          if err = fd.pd.waitRead(fd.isFile); err == nil {
              continue
          }
        }
    }
    err = fd.eofError(n, err)
    return n, oobn, sysflags, err
  }
}

回到 func (sl *sysListener) listenTCP 方法，函數 internetSocket 返回一個套接字結構體的實例，用來構建 TCPListener 對象 &TCPListener{fd: fd, lc: sl.ListenConfig}。后面 Accept tcp conn 時，會用到 net.netFD 的 accept 方法，后者只是封裝了 poll.DF 的 Accept 方法。

回到 net/http 下的 struct Server 的 ListenAndServe 方法，它包含兩步:

  1. net.Listen 方法獲取 ln *TCPListener
  2. srv.Serve(ln)

前面詳細說明了第一步的細節，后面我們看第二步如何Serve。

TCPListenr.Accept

對于 linux下的 tcp server，系統調用 accept 發生在 socket、bind、listen 之后，它從內核中的 ESTABLISHED 隊列中獲取一個建立完成的鏈接。通過函數socket生成的套接字sockfd可以是阻塞或非阻塞(NONBLOCK)，它的函數聲明如下:

#include <sys/socket.h>

int accept(int sockfd, struct sockaddr *restrict addr,
          socklen_t *restrict addrlen);

#define _GNU_SOURCE             /* See feature_test_macros(7) */
#include <sys/socket.h>

int accept4(int sockfd, struct sockaddr *restrict addr,
          socklen_t *restrict addrlen, int flags);

對于阻塞/非阻塞的套接字, accept 的表現并不相同:

  1. 阻塞的sockfd: 調用方會一直被阻塞，直到有一個ESTABLISHED的tcp conn
  2. 非阻塞的sockfd: 函數accept會返回 EAGAIN 或 EWOULDBLOCK 的錯誤

Go net庫使用的是非阻塞的套接字，我們看這部分代碼的邏輯:

// net/net.go struct Server
func (srv *Server) Serve(l net.Listener) error {
  for {
    rw, err := l.Accept()
    // ... 省略部分代碼

這里 ln 的類型是 *TCPListener, 其方法Accept的定義如下:

// Accept implements the Accept method in the Listener interface; it
// waits for the next call and returns a generic Conn.
func (l *TCPListener) Accept() (Conn, error) {
  if !l.ok() {
    return nil, syscall.EINVAL
  }
  c, err := l.accept()
  if err != nil {
    return nil, &OpError{Op: "accept", Net: l.fd.net, Source: nil, Addr: l.fd.laddr, Err: err}
  }
  return c, nil
}

func (ln *TCPListener) accept() (*TCPConn, error) {
  fd, err := ln.fd.accept()
  if err != nil {
    return nil, err
  }
  tc := newTCPConn(fd)
  if ln.lc.KeepAlive >= 0 {
    setKeepAlive(fd, true)
    ka := ln.lc.KeepAlive
    if ln.lc.KeepAlive == 0 {
      ka = defaultTCPKeepAlive
    }
    setKeepAlivePeriod(fd, ka)
  }
  return tc, nil
}

struct TCPListener 的結構如下, accept 依賴成員變量 fd *net.netFD，它通過 pdf poll.FD 的 Accept 方法獲取client端的套接字，并封裝成一個 net.netFD 對象:

// TCPListener is a TCP network listener. Clients should typically
// use variables of type Listener instead of assuming TCP.
type TCPListener struct {
  fd *netFD
  lc ListenConfig
}

func (fd *netFD) accept() (netfd *netFD, err error) {
  d, rsa, errcall, err := fd.pfd.Accept()
  // ... 省略部分代碼

  if netfd, err = newFD(d, fd.family, fd.sotype, fd.net); err != nil {
    poll.CloseFunc(d)
    return nil, err
  }
  if err = netfd.init(); err != nil {
    netfd.Close()
    return nil, err
  }
  lsa, _ := syscall.Getsockname(netfd.pfd.Sysfd)
  netfd.setAddr(netfd.addrFunc()(lsa), netfd.addrFunc()(rsa))
  return netfd, nil
}

繼續看 poll.FD 的方法 Accept。它內部是一個 for 循環，先嘗試通過系統調用accpt4 獲取一個套接字，結果會有下面幾種情況:

  1. 獲取成功, err == nil, 函數直接return即可
  2. syscall.EINTR 表示系統調用期間收到操作系統的信號，但并沒有實質的錯誤發生，所以選擇重試
  3. syscall.EAGAIN 表示目前并沒有establish新的tcp conn，處理是通過 waitRead 將當前goroutine掛起，等待被喚醒
  4. syscall.ECONNABORTED 表示遠程連接已經在ESTABLISHED隊列，還沒有被Accept時，client端放棄連接
  5. 其他錯誤: 返回一個錯誤
// Accept wraps the accept network call.
func (fd *FD) Accept() (int, syscall.Sockaddr, string, error) {
  if err := fd.readLock(); err != nil {
    return -1, nil, "", err
  }
  defer fd.readUnlock()

  if err := fd.pd.prepareRead(fd.isFile); err != nil {
    return -1, nil, "", err
  }
  for {
    s, rsa, errcall, err := accept(fd.Sysfd)
    if err == nil {
      return s, rsa, "", err
    }
    switch err {
    case syscall.EINTR:
      continue
    case syscall.EAGAIN:
      if fd.pd.pollable() {
        if err = fd.pd.waitRead(fd.isFile); err == nil {
          continue
        }
      }
    case syscall.ECONNABORTED:
      // This means that a socket on the listen
      // queue was closed before we Accept()ed it;
      // it's a silly error, so try again.
      continue
    }
    return -1, nil, errcall, err
  }
}

// 代碼路徑: internal/poll/sock_cloexec.go

// 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)
  // ... 省略部分代碼

另外可以看到，accept4 系統調用時，傳入了 SOCK_NONBLOCK 和 SOCK_CLOEXEC 兩個 flag，socket 系統調用也使用了這兩個 flag。

通過 ln.Accept 獲取到ESTABLISHED連接的套接字以后，就可以對遠端的client進行服務了。

在本文中，總共有兩類套接字(socket):

  1. server端監聽的套接字, 通過socket系統調用創建。它的生命周期和server同樣長
  2. 已連接的套接字, 通過accept系統調用創建。它的生命周期比較短，尤其是對于應用層是HTTP短鏈接的情況

在第一篇文章"Go BIO/NIO探討(1)：Gin框架中如何處理HTTP請求"中，我們提到了兩層 for 循環，本文只是講了第一層。從阻塞、非阻塞的角度來看，TCPListener.Accept 方法看起來是block的實現，但底層的套接字和系統調用設置了 NONBLOCK flag，可以說是基于 NONBLOCK 的方式實現的。單純從網絡的視角看，這稱得上是 Non-blocking IO 了。