9、网络编程：如何进行网络IO编程降消耗，提吞吐

for {
   // 获取本轮待处理的 fd
   fd = accept() // 获取连接
   new Thread(){ // 创建线程处理连接
       // 从 fd 中读数据
       data = read(fd)  
       // 处理数据 
       handle(data)  
   }
}

// 多个待服务的 fd 
fds = [fd1,fd2,fd3,...]
i = 0
for {
   fd = fds[i]        
   // 尝试从连接中获取数据，socket未就绪直接返回，不再阻塞
   data,err = tryRead(fd)  
   // 读取数据成功，处理数据
   if err == nil{
      handle(data) 
   } 
   // 10ms后再推进流程，否则不断轮询会消耗过多CPU资源
   sleep(10ms)
   i++
   if i == len(fds){
     i = 0
   }
}

// 多个待服务的 fd 
fds = [fd1,fd2,fd3,...]
for {
   // 多路复用，阻塞同时监听多个连接
   readyFds=epoll_wait(fds)
   for i=0;i<len(readyFds);i++{
      // 开线程处理已就绪连接
      new Thread(){
          fd = readyFds[i]
          data,err = read(fd)  
           // 读取数据成功，处理数据
           if err == nil{
              handle(data) 
           } 
      }
   }
}

package main

import (
    "fmt"
    "net"
)

func handleConnection(conn net.Conn) {
    defer conn.Close()
    // 用于读取客户端发送数据的缓冲区
    buffer := make([]byte, 1024)
    for {
        n, err := conn.Read(buffer)
        if err!= nil {
            fmt.Println("读取客户端数据出错:", err)
            break
        }
        // 输出客户端发送的数据
        fmt.Printf("从客户端接收到: %s\n", buffer[:n])
        // 向客户端发送响应信息
        _, err = conn.Write([]byte("已收到你的消息\n"))
        if err!= nil {
            fmt.Println("向客户端发送响应出错:", err)
            break
        }
    }
}

func main() {
    // 监听的地址和端口
    listenAddr := ":8888"
    // 创建监听的TCP套接字
    listener, err := net.Listen("tcp", listenAddr)
    if err!= nil {
        fmt.Println("监听出错:", err)
        return
    }
    defer listener.Close()

    fmt.Printf("服务器正在监听 %s\n", listenAddr)
    for {
        // 接受客户端连接
        conn, err := listener.Accept()
        if err!= nil {
            fmt.Println("接受客户端连接出错:", err)
            continue
        }
        // 启动一个协程来处理客户端连接
        go handleConnection(conn)
    }
}

//runtime/netpoll_epoll.go
func netpollinit() {
    // 调通epoll_create，创建epoll池
    epfd = epollcreate1(_EPOLL_CLOEXEC)
}

//runtime/netpoll_epoll.go
func netpollopen(fd uintptr, pd *pollDesc) int32 {
    var ev epollevent
    ev.events = _EPOLLIN | _EPOLLOUT | _EPOLLRDHUP | _EPOLLET
    *(**pollDesc)(unsafe.Pointer(&ev.data)) = pd
    // 调用epoll_ctl，将需要监听的连接加到epoll池里
    return -epollctl(epfd, _EPOLL_CTL_ADD, int32(fd), &ev)
}

//runtime/netpoll_epoll.go
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)
}

//runtime/netpoll.go

// 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 {
    // IO未就绪，协程阻塞
    if waitio || netpollcheckerr(pd, mode) == 0 {
        gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5)
    }
}

//runtime/netpoll.go

// 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 {
    // IO未就绪，协程阻塞
    if waitio || netpollcheckerr(pd, mode) == 0 {
        gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5)
    }
}

// runtime/proc.go
func main() {
    // 启动一个线程
    systemstack(func() {
        newm(sysmon, nil, -1)
    })
    
}

func sysmon() {
    for {
        // 每隔10s周期处理
        // poll network if not polled for more than 10ms
        lastpoll := sched.lastpoll.Load()
        if netpollinited() && lastpoll != 0 && lastpoll+10*1000*1000 < now {
            sched.lastpoll.CompareAndSwap(lastpoll, now)
            // 获取网络IO就绪的协程列表
            list, delta := netpoll(0) // non-blocking - returns list of goroutines
            if !list.empty() {
                // 唤醒就绪协程
                injectglist(&list)
            }
        }
    }
}

//  runtime/netpoll_epoll.go
// netpoll checks for ready network connections.
// Returns list of goroutines that become runnable.
func netpoll(delay int64) (gList, int32) {
    var events [128]syscall.EpollEvent
    // 调用epoll_wait非阻塞获取就绪的网络IO事件
    n, errno := syscall.EpollWait(epfd, events[:], int32(len(events)), waitms)
    //从events中获取事件对应的协程
    
    return toRun, delta
}

// runtime/proc.go
// Finds a runnable goroutine to execute.
func findRunnable() (gp *g, inheritTime, tryWakeP bool) {
    // Poll network.
    
    // P 本地队列和全局队列都没有待执行的协程
    if netpollinited() && netpollAnyWaiters() && sched.lastpoll.Load() != 0 {
        // 调用epoll_wait获取IO事件就绪协程列表
        if list, delta := netpoll(0); !list.empty() { // non-blocking
            gp := list.pop()
            // 唤醒相关协程
            injectglist(&list)
        }
    }
}

//runtime/proc.go

// stattTheWorldWithSema returns now.
func startTheWorldWithSema(now int64, w worldStop) int64 {
    if netpollinited() {
        list, delta := netpoll(0) // non-blocking
        injectglist(&list) // 唤醒IO就绪的协程列表
    }
}