golang socket读写同时_epoll在Golang的应用

//用于创建并返回一个epfd句柄，后续关于fd的添加删除等操作都依据这个句柄。
int epoll_create(int size);
//用于向epfd添加，删除，修改要监听的fd。
int epoll_ctl(int epfd, int op, int fd, struct epoll_event* event);
//传入创建返回的epfd句柄，以及超时时间，返回就绪的fd句柄。
int epoll_wait(int epfd, struct epoll_event* events, int maxevents, int timeout);

func main() {fmt.Println("服务端进程id：",os.Getpid())lister, err := net.Listen("tcp", "0.0.0.0:9009")if err != nil {fmt.Println("连接失败", err)return}for {conn, err := lister.Accept() //等待建立连接if err != nil {fmt.Println("建立连接失败", err)continue}//开启协程处理go func() {defer conn.Close()for {buf := make([]byte, 128)n, err := conn.Read(buf)if err != nil{fmt.Println("读出错",err)return}fmt.Println("读取到的数据：",string(buf[:n]))}}()}
}

func (fd *netFD) listenStream(laddr sockaddr, backlog int, ctrlFn func(string, string, syscall.RawConn) error) error {......//绑定该socket接口if err = syscall.Bind(fd.pfd.Sysfd, lsa); err != nil {return os.NewSyscallError("bind", err)}//监听该socketif err = listenFunc(fd.pfd.Sysfd, backlog); err != nil {return os.NewSyscallError("listen", err)}//初始化fd，也就是把socket放入epoll中，进入if err = fd.init(); err != nil {return err}lsa, _ = syscall.Getsockname(fd.pfd.Sysfd)fd.setAddr(fd.addrFunc()(lsa), nil)return nil
}
func (fd *FD) Init(net string, pollable bool) error {......//将socket fd加到poll，进入err := fd.pd.init(fd)......return err
}
//最终跳转到该处，主要关注两个函数runtime_pollServerInit，runtime_pollOpen，
//这两个函数都是runtime实现的，将epoll交由runtime来管理
func (pd *pollDesc) init(fd *FD) error {//sync.once方法，调用epoll_create创建eventpoll对象serverInit.Do(runtime_pollServerInit)//将当前的fd加到epoll中，底层调用epollctl函数ctx, errno := runtime_pollOpen(uintptr(fd.Sysfd))//如果出错，处理相应的fd，删除epoll中fd以及解除状态等操作if errno != 0 {if ctx != 0 {runtime_pollUnblock(ctx)runtime_pollClose(ctx)}return errnoErr(syscall.Errno(errno))}pd.runtimeCtx = ctxreturn nil
}

func poll_runtime_pollServerInit() {//初始化全局epoll对象netpollinit()/全局标志位设置为1atomic.Store(&netpollInited, 1)
}
func netpollinit() {//系统调用，创建一个eventpoll对象epfd = epollcreate1(_EPOLL_CLOEXEC)if epfd >= 0 {return}......
}

func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) {//返回一个存储在Go程序中的一个fd对应的结构体，算是用于记录//goroutine与fd之间的关系，后面会分析到pd := pollcache.alloc()//加锁，防止并发问题lock(&pd.lock)if pd.wg != 0 && pd.wg != pdReady {throw("runtime: blocked write on free polldesc")}if pd.rg != 0 && pd.rg != pdReady {throw("runtime: blocked read on free polldesc")}pd.fd = fdpd.closing = falsepd.everr = falsepd.rseq++pd.rg = 0pd.rd = 0pd.wseq++pd.wg = 0pd.wd = 0unlock(&pd.lock)var errno int32//epoll_ctl系统调用errno = netpollopen(fd, pd)return pd, int(errno)
}
func netpollopen(fd uintptr, pd *pollDesc) int32 {var ev epollevent//注册event事件，这里使用了epoll的ET模式，相对于ET，ET需要每次产生事件时候就要处理事件，//否则容易丢失事件。ev.events = _EPOLLIN | _EPOLLOUT | _EPOLLRDHUP | _EPOLLET//events记录上pd的指针*(**pollDesc)(unsafe.Pointer(&ev.data)) = pd//系统调用将该fd加到eventpoll对象中，交由内核监听return -epollctl(epfd, _EPOLL_CTL_ADD, int32(fd), &ev)
}

func (fd *FD) Accept() (int, syscall.Sockaddr, string, error) {......//检查fd状态是否变化if err := fd.pd.prepareRead(fd.isFile); err != nil {return -1, nil, "", err}for {//accept系统调用，如果有对监听的socket的连接请求，则直接返回发起连接的socket文件描述符//，否则返回EAGAIN错误，被下面捕获到s, rsa, errcall, err := accept(fd.Sysfd)if err == nil {return s, rsa, "", err}switch err {case syscall.EAGAIN:if fd.pd.pollable() {//进入waitRead方法，内部if err = fd.pd.waitRead(fd.isFile); err == nil {continue}}case syscall.ECONNABORTED:continue}return -1, nil, errcall, err}
}
func (pd *pollDesc) wait(mode int, isFile bool) error {if pd.runtimeCtx == 0 {return errors.New("waiting for unsupported file type")}//进入runtime_pollWait方法内部，该方法会跳转到runtime包下，条件满足会park住goroutineres := runtime_pollWait(pd.runtimeCtx, mode)return convertErr(res, isFile)
}
func poll_runtime_pollWait(pd *pollDesc, mode int) int {......//进入netpollblock函数，该函数内部会阻塞住该goroutinefor !netpollblock(pd, int32(mode), false) {err = netpollcheckerr(pd, int32(mode))if err != 0 {return err}}return 0
}
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {gpp := &pd.rgif mode == 'w' {gpp = &pd.wg}......if waitio || netpollcheckerr(pd, mode) == 0 {//gark住该g，此时传参主要关注前两个，一个netpollblockcommit函数，一个gpp为当前pd的rg或者wg，//用于后面记录fd对应的阻塞的goroutinegopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5)}old := atomic.Xchguintptr(gpp, 0)if old > pdWait {throw("runtime: corrupted polldesc")}return old == pdReady
}
func gopark(unlockf func(*g, unsafe.Pointer) bool, lock unsafe.Pointer, reason waitReason, traceEv byte, traceskip int) {......//主要关注两个传参，lock是gpp指针mp.waitlock = lock//unlockf为netpollblockcommit函数mp.waitunlockf = unlockf......//切换到g0栈去执行park_mmcall(park_m)
}
func park_m(gp *g) {//获取当前goroutine_g_ := getg()//修改状态为Gwaiting，代表当前的goroutine被park住了casgstatus(gp, _Grunning, _Gwaiting)//解除m和g关联dropg()if fn := _g_.m.waitunlockf; fn != nil {//调用刚传入的函数参数，也就是netpollblockcommitok := fn(gp, _g_.m.waitlock)//调用完清除_g_.m.waitunlockf = nil_g_.m.waitlock = nilif !ok {if trace.enabled {traceGoUnpark(gp, 2)}casgstatus(gp, _Gwaiting, _Grunnable)execute(gp, true) // Schedule it back, never returns.}}//调度新的g到m上来schedule()
}
func netpollblockcommit(gp *g, gpp unsafe.Pointer) bool {//把当前g的指针存为gpp指针，gpp为pd的rg或wgr := atomic.Casuintptr((*uintptr)(gpp), pdWait, uintptr(unsafe.Pointer(gp)))if r {//将全局变量改为1，代表系统有netpoll的等待者atomic.Xadd(&netpollWaiters, 1)}return r
}

//上面提到过的accept函数，根据序号顺序分析
func (fd *FD) Accept() (int, syscall.Sockaddr, string, error) {......for {//2.使用accept系统调用能获取到新的连接，linux会为新的连接分配一个新的fd，//这个函数会返回新的连接的socket fd对应的进程描述符s, rsa, errcall, err := accept(fd.Sysfd)if err == nil {//3.返回新的进程描述符return s, rsa, "", err}switch err {case syscall.EAGAIN:if fd.pd.pollable() {//1.刚才阻塞到了这个goroutine，后来新的连接请求，该goroutine被唤醒if err = fd.pd.waitRead(fd.isFile); err == nil {continue}}......}......}
}
//返回上一层的函数
func (fd *netFD) accept() (netfd *netFD, err error) {//此时获取到了新的fdd, rsa, errcall, err := fd.pfd.Accept()......//创建新的fd结构体if netfd, err = newFD(d, fd.family, fd.sotype, fd.net); err != nil {poll.CloseFunc(d)return nil, err}//init函数又会进入func (pd *pollDesc) init(fd *FD) error函数，并将新的socket连接通过epoll_ctl传入//epoll的监听事件if err = netfd.init(); err != nil {fd.Close()return nil, err}//系统调用，可以获得客户端的socket的ip信息等lsa, _ := syscall.Getsockname(netfd.pfd.Sysfd)netfd.setAddr(netfd.addrFunc()(lsa), netfd.addrFunc()(rsa))return netfd, nil
}

func findrunnable() (gp *g, inheritTime bool) {_g_ := getg()//分别从本地队列和全局队列寻找可执行的g......//判断是否满足条件，初始化netpoll对象，是否等待者，以及上次调用时间if netpollinited() && atomic.Load(&netpollWaiters) > 0 && atomic.Load64(&sched.lastpoll) != 0 {//netpoll底层调用epoll_wait，传参代表epoll_wait时候是阻塞等待或者非阻塞直接返回//这里是非阻塞模式，会立即返回内核eventpoll对象的rdlist列表if list := netpoll(false); !list.empty() {gp := list.pop()//将可运行G的列表注入调度程序并清除glistinjectglist(&list)//修改gp状态casgstatus(gp, _Gwaiting, _Grunnable)if trace.enabled {traceGoUnpark(gp, 0)}//返回可运行的greturn gp, false}}.......stopm()goto top
}
//对epoll_wait的进一步封装
func netpoll(block bool) gList {if epfd == -1 {return gList{}}waitms := int32(-1)if !block {waitms = 0}//声明一个epollevent事件，在epoll_wait系统调用时候，会给该数组赋值并返回一个索引位，/之后可以遍历数组取出就绪的fd事件。var events [128]epollevent
retry://陷入系统调用，取出内核eventpoll中的rdlist，返回就绪的事件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("runtime: netpoll failed")}goto retry}var toRun gList//遍历event事件数组for i := int32(0); i < n; i++ {ev := &events[i]if ev.events == 0 {continue}var mode int32//是否有就绪的读写事件，放入mode标志位if ev.events&(_EPOLLIN|_EPOLLRDHUP|_EPOLLHUP|_EPOLLERR) != 0 {mode += 'r'}if ev.events&(_EPOLLOUT|_EPOLLHUP|_EPOLLERR) != 0 {mode += 'w'}if mode != 0 {//取出存入的pollDesc的指针pd := *(**pollDesc)(unsafe.Pointer(&ev.data))pd.everr = falseif ev.events == _EPOLLERR {pd.everr = true}//取出pd中的rg或wg，后面放到运行队列netpollready(&toRun, pd, mode)}}if block && toRun.empty() {goto retry}return toRun
}
func netpollready(toRun *gList, pd *pollDesc, mode int32) {var rg, wg *gif mode == 'r' || mode == 'r'+'w' {rg = netpollunblock(pd, 'r', true)}if mode == 'w' || mode == 'r'+'w' {wg = netpollunblock(pd, 'w', true)}//将阻塞的goroutine加入gList返回if rg != nil {toRun.push(rg)}if wg != nil {toRun.push(wg)}
}

func main() {......//开启处理go func() {defer conn.Close()for {buf := make([]byte, 128)//将缓冲区的数据读出来放到buf中n, err := conn.Read(buf)......}}()}
}
func (fd *FD) Read(p []byte) (int, error) {......for {//系统调用读取缓冲区数据，这里没有可读会直接返回，不会阻塞n, err := syscall.Read(fd.Sysfd, p)if err != nil {n = 0if err == syscall.EAGAIN && fd.pd.pollable() {//不可读，进入waitRead方法，park住该goroutine，//并记录goroutine到pd的rg中,等待唤醒if err = fd.pd.waitRead(fd.isFile); err == nil {continue}}}......}
}

//当发生某些情况，如连接断开，fd销毁等，会调用到此处
func poll_runtime_pollClose(pd *pollDesc) {.......netpollclose(pd.fd)//释放对应的pdpollcache.free(pd)
}
//调用epoll_ctl系统调用，删除该fd在eventpoll上对应的epitem
func netpollclose(fd uintptr) int32 {var ev epolleventreturn -epollctl(epfd, _EPOLL_CTL_DEL, int32(fd), &ev)
}

$ strace -f ./server

#....省略内存管理部分以及线程管理部分
#执行到fmt.Println("服务端进程id：",os.Getpid())
[pid 30307] getpid() = 30307
[pid 30307] write(1, "346234215345212241347253257350277233347250213id357274232 30307n", 27服务端进程id：30307
) = 27
......由于过多，省略关于socket的系统调用
[pid 30308] <... nanosleep resumed> NULL) = 0
#打开系统文件，该文件定义tcp最大连接数，会被设置成pollable，并加入epoll节点中
[pid 30307] openat(AT_FDCWD, "/proc/sys/net/core/somaxconn", O_RDONLY|O_CLOEXEC <unfinished ...>
[pid 30308] nanosleep({tv_sec=0, tv_nsec=20000}, <unfinished ...>
[pid 30307] <... openat resumed> ) = 4
#调用epoll_ctl，创建一个eventpoll
[pid 30307] epoll_create1(EPOLL_CLOEXEC) = 5
#将fd加到epoll事件
[pid 30307] epoll_ctl(5, EPOLL_CTL_ADD, 4, {EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, {u32=2174189320, u64=139635855949576}}) = 0
[pid 30307] fcntl(4, F_GETFL) = 0x8000 (flags O_RDONLY|O_LARGEFILE)
[pid 30307] fcntl(4, F_SETFL, O_RDONLY|O_NONBLOCK|O_LARGEFILE) = 0
[pid 30308] <... nanosleep resumed> NULL) = 0
[pid 30307] read(4, <unfinished ...>
#执行epoll_wait查看就绪事件
[pid 30308] epoll_pwait(5, <unfinished ...>
[pid 30307] <... read resumed> "512n", 65536) = 4
[pid 30308] <... epoll_pwait resumed> [{EPOLLIN|EPOLLOUT, {u32=2174189320, u64=139635855949576}}], 128, 0, NULL, 139635812673280) = 1
[pid 30307] read(4, <unfinished ...>
[pid 30308] nanosleep({tv_sec=0, tv_nsec=20000}, <unfinished ...>
[pid 30307] <... read resumed> "", 65532) = 0
#将/proc/sys/net/core/somaxconn文件的fd从epoll中删除
[pid 30307] epoll_ctl(5, EPOLL_CTL_DEL, 4, 0xc00005e8d4) = 0
#关掉打开的somaxconn描述符
[pid 30307] close(4) = 0
#设置监听的socket描述符
[pid 30307] setsockopt(3, SOL_SOCKET, SO_REUSEADDR, [1], 4) = 0
[pid 30307] bind(3, {sa_family=AF_INET6, sin6_port=htons(9009), inet_pton(AF_INET6, "::", &sin6_addr), sin6_flowinfo=htonl(0), sin6_scope_id=0}, 28) = 0
[pid 30307] listen(3, 512 <unfinished ...>
[pid 30308] <... nanosleep resumed> NULL) = 0
[pid 30307] <... listen resumed> ) = 0
[pid 30308] nanosleep({tv_sec=0, tv_nsec=20000}, <unfinished ...>
#将用于监听的socket fd加入到epoll中
[pid 30307] epoll_ctl(5, EPOLL_CTL_ADD, 3, {EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, {u32=2174189320, u64=139635855949576}}) = 0
[pid 30307] getsockname(3, {sa_family=AF_INET6, sin6_port=htons(9009), inet_pton(AF_INET6, "::", &sin6_addr), sin6_flowinfo=htonl(0), sin6_scope_id=0}, [112->28]) = 0
#执行accept4发现没有连接，返回EAGAIN错误
[pid 30307] accept4(3, 0xc00005eb98, [112], SOCK_CLOEXEC|SOCK_NONBLOCK) = -1 EAGAIN (Resource temporarily unavailable)
#查看是否有就绪的fd，此次调用是非阻塞，立即返回
[pid 30307] epoll_pwait(5, [], 128, 0, NULL, 0) = 0
[pid 30308] <... nanosleep resumed> NULL) = 0
#查看是否有就绪的fd，此次会阻塞等待，直到有连接进来
[pid 30307] epoll_pwait(5, <unfinished ...>
[pid 30308] futex(0x60dc70, FUTEX_WAIT_PRIVATE, 0, {tv_sec=60, tv_nsec=0} <unfinished ...>
[pid 30307] <... epoll_pwait resumed> [{EPOLLIN, {u32=2174189320, u64=139635855949576}}], 128, -1, NULL, 0) = 1
[pid 30307] futex(0x60dc70, FUTEX_WAKE_PRIVATE, 1) = 1
[pid 30308] <... futex resumed> ) = 0
#新的连接，代表收到了一个客户端连接,分配了一个fd是4
[pid 30307] accept4(3, <unfinished ...>, <... accept4 resumed> {sa_family=AF_INET6, sin6_port=htons(52082), inet_pton(AF_INET6, "::ffff:127.0.0.1", &sin6_addr), sin6_flowinfo=htonl(0), sin6_scope_id=0}, [112->28], SOCK_CLOEXEC|SOCK_NONBLOCK) = 4
#把4加入到epoll中管理
[pid 30307] epoll_ctl(5, EPOLL_CTL_ADD, 4, {EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, {u32=2174189112, u64=139635855949368}}) = 0
[pid 30307] getsockname(4, {sa_family=AF_INET6, sin6_port=htons(9009), inet_pton(AF_INET6, "::ffff:127.0.0.1", &sin6_addr), sin6_flowinfo=htonl(0), sin6_scope_id=0}, [112->28]) = 0
......
#后来将client端关掉,此时tcp连接断掉了，将epoll中的fd移除
[pid 30309] epoll_ctl(5, EPOLL_CTL_DEL, 4, 0xc00005fdd4 <unfinished ...>
[pid 30308] nanosleep({tv_sec=0, tv_nsec=20000}, <unfinished ...>
[pid 30309] <... epoll_ctl resumed> ) = 0
[pid 30309] close(4) = 0
[pid 30309] epoll_pwait(5, [], 128, 0, NULL, 824634114048) = 0
#阻塞等待
[pid 30309] epoll_pwait(5, <unfinished ...>
........