协程是什么线程是操作系统内核态调度的最小单位协程Coroutine是运行在线程之上的用户态轻量级执行单元一个线程可以承载成百上千个协程协程的调度完全在用户态完成内核感知不到协程的存在为什么要有协程在回答这个问题之前我们需要先了解异步的好处。在之前的学习里不论是服务器创建线程池并行执行业务请求还是客户端并发创建连接测试服务器并发度异步带来的好处就是性能提升因为异步可以非阻塞发送多个请求每个请求利用多线程执行回调函数互相间不影响。一个协程对应一个 fd协程在等待 fd 就绪时会主动让出线程控制权用户态非阻塞线程可以立刻调度其他协程执行fd 就绪后再通过用户态调度器唤醒对应协程。整个过程线程始终处于运行状态无内核阻塞 / 唤醒开销。比如同步7600ms while(1){ int nready epoll_wait(); for(int i0;inready;i){ recv(event[i].fd,buffer); send(event[i].fd,buffer); } } 异步1600ms void *task_callback(void *arg){ recv(event[i].fd,buffer); send(event[i].fd,buffer); } while(1){ int nready epoll_wait(); for(int i0;inready;i){ task.fd event[id].fd; task.callback task_callback; push_task_to_threadpool(task); } }但是异步的缺点是思想不符合正常思维当有多层嵌套的时候比如task_callback中业务处理可能也需要并发优化会降低可读性。所以协程的作用就是“同步的思路异步的性能”。下面nty_server的代码更直观在每一个I/O之前都加入一个poll检测如果数据没准备好跳转到下一个I/O执行。假设已有三个协程func1、func2、func3下面用ucontest简单体现协程的跳转。// coroutine1 void func1(void) { while (count 30) { printf(1\n); swapcontext(ctx[0], main_ctx); printf(4\n); } } // coroutine2 void func2(void) { while (count 30) { printf(2\n); swapcontext(ctx[1], main_ctx); printf(5\n); } } // coroutine3 void func3(void) { while (count 30) { printf(3\n); swapcontext(ctx[2], main_ctx); printf(6\n); } } // schedule int main() { char stack1[2048] {0}; char stack2[2048] {0}; char stack3[2048] {0}; getcontext(ctx[0]); //将当前寄存器状态绑定到ctx中 ctx[0].uc_stack.ss_sp stack1; ctx[0].uc_stack.ss_size sizeof(stack1); ctx[0].uc_link main_ctx; makecontext(ctx[0], func1, 0); //绑定入口函数 0为传入参数 getcontext(ctx[1]); ctx[1].uc_stack.ss_sp stack2; ctx[1].uc_stack.ss_size sizeof(stack2); ctx[1].uc_link main_ctx; makecontext(ctx[1], func2, 0); getcontext(ctx[2]); ctx[2].uc_stack.ss_sp stack3; ctx[2].uc_stack.ss_size sizeof(stack3); ctx[2].uc_link main_ctx; makecontext(ctx[2], func3, 0); printf(swapcontext\n); while (count 30) { // scheduler swapcontext(main_ctx, ctx[count%3]); } printf(\n); }其中协程的原语操作就是代码里的swapcontext而从协程换到main函数的操作称为yeild从main函数换到协程的操作称为resume而main函数中的while(count 30) 循环就是一个调度器尽管此处的调度策略只是简单的%3进行调度。汇编上实现yeild和resume也是通过置换寄存器的值协程异步性能的体现在IO操作之前把当前协程负责的fd以及感兴趣的事件注册到epoll然后yeild让出cpu。调度器就会epoll_wait等待某一个fd触发了调度器就会resume这个协程这个协程就会删除在epoll注册的感兴趣的事件然后再执行IO操作。IO 处理完成后把 fd 重新加入 epoll让 epoll 继续监听它的后续事件比如客户端又发了新数据既可以保证后续的 IO 能被正常处理还能够保证 sockfd 只在一个上下文中能够操作 IO 的不会出现在多个上下文同时对一个 IO 进行操作的协程的 IO 异步操作正式是采用此模式进行的。举例具体产品场景如果使用协程如何实现场景1当打开CSDN的时候左边展示了多个子模块跳转按钮中间展示了咨询头条内容、火热开源项目、精选博客右边是活动日历、精品课程以及广告。实现在收到客户请求后服务器需要向数据库请求数据后IO响应给用户此时协程可以用在数据库部分当数据库IO监测到哪一块的服务器请求就返回哪一块的数据也用在服务器上IO监测哪一块收到了数据后就先发送给用户。调度策略是等待IO就yeild由epoll监测处理数据就resume。hook的作用保证在不改变原先代码逻辑的基础上就能用协程。所以要对系统调用的IO API进行一个改名原来的系统调用connect被改名为connect_f , 所以原来的名字connect就空出来了交由我们用户实现。而像mysql-dev源码里面用的都是connect所以就会走我们用户写的函数了。比如以recv举例原先代码 while(1){ recv(); parser(); send(); } 改写recv: typedef ssize_t (*recv_t)(int, void *buf, size_t, int); //一种名为recv_t的类型 recv_t recv_f; //可以指向(int, void *buf, size_t, int)这类函数 recv_f dlsym(RTLD_NEXT, recv); //通过dlsym找到recv系统调用的地址 recv(fd, buffer, length){ struct pollfd fds[1] {0}; fds[0].fd fd; fds[0].event EPOLLIN; if(poll(fd,1,0) 0){ //不可读 add_fd_to_epoll(fd,EPOLLIN); swapcontext(); } recv_f(fd,buffer,length) //底层的recv 调度回来就可以直接执行 }实现一个协程框架思路给每一个需要IO的posix api重写在执行IO操作之前1将fd加入epoll监听 2将协程与fd和监听类型绑定并在调度器中注册等待状态 3将当前协程yeild让出cpu 4当回到yeild下一条命令的时候说明调度器resume回来了将对应的协程从等待状态中移除并恢复其运行。以上述思路编写recv代码static int nty_poll_inner(struct pollfd *fds, nfds_t nfds, int timeout) ssize_t recv(int fd, void *buf, size_t len, int flags) { if (!recv_f) init_hook(); nty_schedule *sched nty_coroutine_get_sched(); if (sched NULL) { return recv_f(fd, buf, len, flags); } struct pollfd fds; fds.fd fd; fds.events POLLIN | POLLERR | POLLHUP; nty_poll_inner(fds, 1, 1); //recv前需要做的工作 int ret recv_f(fd, buf, len, flags); //真正的recv return ret; } static int nty_poll_inner(struct pollfd *fds, nfds_t nfds, int timeout) { nty_schedule *sched nty_coroutine_get_sched(); nty_coroutine *co sched-curr_thread; int i 0; for (i 0;i nfds;i ) {//加入epoll监听 struct epoll_event ev; ev.events nty_pollevent_2epoll(fds[i].events); ev.data.fd fds[i].fd; epoll_ctl(sched-poller_fd, EPOLL_CTL_ADD, fds[i].fd, ev); co-events fds[i].events; nty_schedule_sched_wait(co, fds[i].fd, fds[i].events, timeout); //将协程与特定的文件描述符fd和事件绑定并在调度器中注册等待状态 } nty_coroutine_yield(co); //让出cpu 当执行到下面的for循环说明该协程co已经被调度器选上了 for (i 0;i nfds;i ) { struct epoll_event ev; ev.events nty_pollevent_2epoll(fds[i].events); ev.data.fd fds[i].fd; epoll_ctl(sched-poller_fd, EPOLL_CTL_DEL, fds[i].fd, ev); nty_schedule_desched_wait(fds[i].fd); //在文件描述符fd就绪时将对应的协程从等待状态中移除并恢复其运行用红黑树查找到fd } return nfds; }后续任务实现协程框架并且利用协程框架编写一个tcp服务器可用测试百万并发。实现协程框架未完成nty_server代码其中nty_recv等在内部会调recvrecv再调的recv_f才是底层的读入。#include nty_coroutine.h #include arpa/inet.h #define MAX_CLIENT_NUM 1000000 #define TIME_SUB_MS(tv1, tv2) ((tv1.tv_sec - tv2.tv_sec) * 1000 (tv1.tv_usec - tv2.tv_usec) / 1000) void server_reader(void *arg) { int fd *(int *)arg; free(arg); int ret 0; struct pollfd fds; fds.fd fd; fds.events POLLIN; while (1) { char buf[1024] {0}; ret nty_recv(fd, buf, 1024, 0); if (ret 0) { if(fd MAX_CLIENT_NUM) printf(read from server: %.*s\n, ret, buf); ret nty_send(fd, buf, strlen(buf), 0); if (ret -1) { nty_close(fd); break; } } else if (ret 0) { nty_close(fd); break; } } } void server(void *arg) { unsigned short port *(unsigned short *)arg; free(arg); int fd nty_socket(AF_INET, SOCK_STREAM, 0); if (fd 0) return ; struct sockaddr_in local, remote; local.sin_family AF_INET; local.sin_port htons(port); local.sin_addr.s_addr INADDR_ANY; bind(fd, (struct sockaddr*)local, sizeof(struct sockaddr_in)); listen(fd, 20); printf(listen port : %d\n, port); struct timeval tv_begin; gettimeofday(tv_begin, NULL); while (1) { socklen_t len sizeof(struct sockaddr_in); int cli_fd nty_accept(fd, (struct sockaddr*)remote, len); if (cli_fd % 1000 999) { struct timeval tv_cur; memcpy(tv_cur, tv_begin, sizeof(struct timeval)); gettimeofday(tv_begin, NULL); int time_used TIME_SUB_MS(tv_begin, tv_cur); printf(client fd : %d, time_used: %d\n, cli_fd, time_used); } printf(new client comming\n); nty_coroutine *read_co; int *arg malloc(sizeof(int)); *arg cli_fd; nty_coroutine_create(read_co, server_reader, arg); } } int main(int argc, char *argv[]) { nty_coroutine *co NULL; int i 0; unsigned short base_port 9096; for (i 0;i 100;i ) { unsigned short *port calloc(1, sizeof(unsigned short)); *port base_port i; nty_coroutine_create(co, server, port); ////////no run } nty_schedule_run(); //run return 0; }测试nty_server性能第一次内存服务器 5G 三台客户机3Gcpu服务器4核 三台客户机2核sysctl.conf中的tcp协议栈配置讲解net.ipv4.tcp_mem 252144 378216 630360 //整个tcp协议栈缓存大小。单位页 4K1G 1.5G 2.5G 1.5到2.5触发内存回收大于2.5G强行关闭 net.ipv4.tcp_wmem 512 512 1024 //给每个tcp连接分配的缓冲区大小默认0.5K 最大1K 整体100w1024*1024 * 1K 1G net.ipv4.tcp_rmem 512 512 1024结果跑到了904000个连接内存跑到4.6G 交换跑到2G的时候服务器宕机第二次尝试修改tcp_mem 成1G 2G 3G 因为服务器可用内存其实大于1.5到2.5区间看看是不是这方面的原因结果跑到914000个连接也是交换跑到2G的时候宕机了第三次尝试调大服务器内存到6G应该会有好转感觉能跑到95w结果服务器915000个连接没有好转服务器交互内存差点跑满client2却先杀死。一旦交互内存使用完操作系统会触发OOM-Killer机制把消耗内存最多的进程kill掉以释放内存。第四次上次结果client2先掉了说明给服务器增加内存确实避免了被kill但是客户端撑不住了。这次我们增加客户机的内存3G--4G。结果跑到909000个连接比上次还低。服务器被杀死。第五次最后一次调整内存把服务器6G--8G结果跑到938000个连接服务器交互内存跑满2G宕机。最后client2的cpu跑满到100% 但是又降下去了服务器先结束了所以没报错。第六次调整tcp协议栈内存回收区间 [2G ~ 3G] --- [2G ~ 4G],看看能否达到不增加物理内存也能增加并发度的效果突破93.8w。结果服务器只跑到86.7w个连接client2在调整回收策略后率先倒下cpu跑满了。第七次把回收区间调整回去把tcp协议栈分配的读写缓存缩小到256KB给client2的cpu调整到4核。结果跑到93.3w第八次一模一样再跑一次结果96.6w途中client1 cpu也跑到100文章参考纯c协程框架NtyCo实现与原理-CSDN博客代码参考https://github.com/0voice