C/C++编程：协程库NtyCo入门

素颜马尾好姑娘i 2022-09-13 14:28 146阅读 0赞

# 协程的起源 #

> 问题：协程存在的原因？协程能够解决哪些问题

在我们线程CS/BS开发模式下，服务器的吞吐量是一个很重要的参数。其实吞吐量是IO处理时间加上业务处理。

为了简单起见，比如，客户端与服务器之间是长连接的，客户端定期给服务器发送心跳包数据。客户端发送一次心跳包到服务器，服务器更新该新客户端状态的。心跳包发送的过程，业务处理时长等于IO读取（RECV系统调用）加上业务处理（更新客户状态）。吞吐量等于1s业务处理次数。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_20_color_FFFFFF_t_70_g_se_x_16]  
业务处理（更新客户端状态）时间，业务不一样的，处理时间不一样，我们就不做讨论。

那如何提升recv的性能。如果只有一个客户端，recv的性能也没有必要提升，也不能提升。如果有百万记的客户端长连接的情况，我们该如何提升。以Linux为例，在这里需要介绍一个“网红”就是epoll。服务器使用epoll管理百万计的客户端长连接，代码框架如下：

while (1) {
        
        int nready = epoll_wait(epfd, events, EVENT_SIZE, -1);
    
        for (i = 0;i < nready;i ++) {
        
    
            int sockfd = events[i].data.fd;
            if (sockfd == listenfd) {
        
                int connfd = accept(listenfd, xxx, xxxx);
                
                setnonblock(connfd);
    
                ev.events = EPOLLIN | EPOLLET;
                ev.data.fd = connfd;
                epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev);
    
            } else {
        
                handle(sockfd);
            }
        }
    }

对于响应式服务器，所有的客户端的操作驱动都是来源于这个大循环。来源于epoll\_wait的反馈结果。  
对于服务器处理百万计的IO。Handle(sockfd)实现方式有两种。

*  handle(sockefd)函数内部对sockfd进行读写操作

int handle(int sockfd) {
        
    
        recv(sockfd, rbuffer, length, 0);
        parser_proto(rbuffer, length);
        send(sockfd, sbuffer, length, 0);
        
    }

*  handler的io操作(send, recv)与epoll\_wait是在同一个处理流程里面的。这就是IO同步操作

优点：

*  sockfd管理方便
 *  操作逻辑清晰

缺点：

*  服务器程序依赖epoll\_wait的循环响应速度慢
 *  程序性能差

当然，Handle(sockfd)有另外的处理方法

*  handle(sockfd)函数内部将sockfd的操作，push到线程池中

int thread_cb(int sockfd){
        
    	// 此函数是在线程池创建的线程中运行
    	// 与handler不在一个线程上下文中运行
        recv(sockfd, rbuffer, length, 0);
        parser_proto(rbuffer, length);
        send(sockfd, sbuffer, length, 0);
    }
    
    int handle(int sockfd){
        
    	// 此函数在主线程main_thread中运行
    	// 在此处之前，确保线程池已经启动
    	push_thread(sockfd, thread_cv); //将sockfd放到其他线程中运行
    }

*  handle函数是将sockfd处理方式放到另一个线程中运行，如此做法，将io操作（recv、send）与epoll\_wait不在一个处理流程里面，使得io操作(recv、send)与epoll\_wait实现解耦。这就叫做io异步操作

优点：

*  子模块好规划
 *  程序性能高

缺点

*  正因为子模块好规划，使得模块之间的socket的管理异常麻烦。每一个子线程都需要管理好socket，必须在IO操作的时候，socket出现关闭或者其他异常。

有没有一种方式，有异步性能，同步的代码逻辑。来方便编程人员对IO操作的组件呢？有，采用一种轻量级的协程来实现。在每次send或者recv之前进行切换，再有调度器来处理epoll\_wait的流程。这就是 [NtyCo：一个C实现的协程库][NtyCo_C]

# 协程的案例 #

> 问题：协程如何使用？与线程使用有何区别

在做网络IO编程的时候，有一个非常理想的情况，就是每次accept返回的时候，就为新来的客户端分配一个线程，这样一个客户端对应一个线程，就不会有多个线程共有一个sockfd。每请求每线程的方式，并且代码逻辑非常易读。但是这只是理想，线程创建代价、调度代价就很大了。

先来看一下每请求每线程的代码如下：

while(1){
        
    	int nfds = (epoll_wait(epoll_fd, events, curfds, -1));
    	if(nfds == -1){
        
    		perror("epoll_wait");
    		break;
    	}
    
    	for(i = 0; i < nfds; i++){
        
    		int sockfd = listenfd(events[i].data.fd, sockfds);
    		if(sockfd)[
    			socklen_t len = sizeof(struct sockaddr_in);
    			int clientfd = accept(sockfd, (struct sockaddr*)&remote, &len);
    
    			pthread_t thread_id;
    			pthread_create(&thread_id, NULL, client_cb, &clientfd);
    		]else{
        
    			...
    		}
    	}
    }

这样的做法，写完放到生产环境下面，如果你的老板不打死你，你来找我。我来帮你老板，为民除害。

如果我们有协程，我们就可以这样实现。参考代码如下：

while (1) {
        
        int nfds = epoll_wait(epoll_fd, events, curfds, -1);
        if (nfds == -1) {
        
            perror("epoll_wait");
            break;
        }
    
        for (i = 0;i < nfds;i ++) {
        
            int sockfd = listenfd(events[i].data.fd, sockfds);
            if (sockfd) {
        
                socklen_t len = sizeof(struct sockaddr_in);
                int clientfd = accept(sockfd, (struct sockaddr*)&remote, &len);
                
                nty_coroutine *read_co;
                nty_coroutine_create(&read_co, server_reader, &clientfd);
            }
            else
            {
        
                ...
            }
    }

这样的代码是完全可以放在生成环境下面的。如果你的老板要打死你，你来找我，我帮你把你老板打死，为民除害。

线程的API思维来使用协程，函数调用的性能来测试协程。

NtyCo封装出来了若干接口，一类是协程本身的，二类是posix的异步封装  
    
协程创建

int nty_coroutine_create(nty_coroutine **new_co, proc_coroutine func, void *arg)

协程调度器的运行

void nty_schedule_run(void)

POSIX异步封装API：

int nty_socket(int domain, int type, int protocol)
    int nty_accept(int fd, struct sockaddr *addr, socklen_t *len)
    int nty_recv(int fd, void *buf, int length)
    int nty_send(int fd, const void *buf, int length)
    int nty_close(int fd)

# 参考 #

*  [NtyCo：一个C实现的协程库][NtyCo_C]
 *  [Linux高性能网络：协程系列][Linux]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/b7b55d7d366b4075ac1bca5d5438d4ee.png
[NtyCo_C]: https://github.com/wangbojing/NtyCo
[Linux]: https://www.jianshu.com/p/8bcec0d2b594