协程

2024-11-27 9 分钟阅读 #计算机#媒体开发

https://blog.csdn.net/weixin_43925427/article/details/140907523

https://blog.csdn.net/qq_62821433/article/details/139480927

概念

协程（Coroutines） 是一种程序组件，允许不同的执行线程或路径在单个进程中共享同一个执行环境。与线程相比，协程是轻量级的，因为它们共享相同的进程空间并具有较小的上下文切换开销。主要特点包括：

  • 非阻塞执行：协程可以在等待 I/O 操作或其他长时间执行的任务时让出 CPU，允许其他协程运行。
  • 协作式多任务处理：协程的切换是显式的，由程序员或协程库控制，不是由操作系统的调度器控制。
  • 高效的并发：协程可以在单个线程中实现高效的并发执行，适用于处理大量并发连接和请求。

在网络应用中，协程可用于优化 I/O 操作，尤其是在高并发环境下。使用协程可以避免为每个客户端连接创建一个线程的高成本，从而减少资源消耗和提高性能。

与进程，线程的区别

  • 进程

    1.程序的一次执行过程；

    2.系统资源分配的单位，具有独立的虚拟内存，PCB控制块，代码，数据，堆栈，文件资源等；

    3.进程上下文切换开销大（主要问题）

  • 线程

    1.系统调度的单位；

    2.共享进程的资源空间；具有自己的栈，寄存器

    3.并发性好，同一进程内的线程切换不会导致进程的上下文切换，所以线程切换开销小；

    4.共享进程资源，多线程会发生冲突，需要进行同步。

  • 协程

    1.可以理解为线程中的线程；

    2.具有自己的上下文；

    3.协程切换不会引起线程的上下文（更轻量）

    4.一个线程同一时刻只有一个协程执行；

    5.协程执行任意时刻都可以退出，下次可以继续执行（与函数的区别，函数一条路走到黑）

    6.高并发场景下性能很好，并且不需要同步

产生原因

协程效率高并不是根本原因。

在高并发场景下，两个原因：

1.用同步代码逻辑简单，并发性差强人意

同步阻塞：阻塞等待业务处理，期间不能做其他事情

同步非阻塞：不断轮询，忙等待占用cpu，

2.用异步，并发性好，代码逻辑又很复杂

基于以上就产生了协程，协程可以做到用同步的代码逻辑达到异步的效果

优缺点

优点：

1.用同步的思想编写出异步的效果，

2.并发性好，性能高

缺点：

单线程中同时只能执行一个协程，无法利用多核功能，还是得结合多线程

协程库

  • Boost.Coroutine2：这是Boost库中的⼀个模块，提供了⼀组灵活的协程实现，包括对称协程和⾮对称协程。它提供了⼀些强⼤的特性，如⽀持多个栈和⾃定义栈⼤⼩。
  • libco：腾讯微信团队开源的⼀个C/C++协程库，据说微信后台⼤量在使⽤这个库，通过⼏个简单的接⼝就能实现协程的创建/调度，同时基于epoll/kqueue实现了⼀个事件反应堆，再加上sys_call（系统调⽤）hook技术，以此给开发者提供同步/异步的开发⽅法
  • libgo ：是⼀个使⽤ C++ 编写的协作式调度的stackful有栈协程库, 同时也是⼀个强⼤的并⾏编程库。⽀持linux平台，MacOS和windows平台，在c++11以上的环境中都能⽤。

ucontext库

概念

在 C 语言中，ucontext 库提供了一套功能丰富的 API，用于管理程序的执行上下文。这些功能主要用于实现用户空间的轻量级线程（协程），允许程序在多个执行流之间进行显式的上下文切换。

  • 上下文（Context）：表示程序在某一时间点的运行状态，包括程序计数器、寄存器集合和堆栈等。
  • 协程（Coroutine）：是一种比线程更轻量的执行单元，它允许进行非抢占式的任务切换，即在用户空间进行切换而无需内核介入。

核心函数

  1. getcontext(ucontext_t \*ucp):

    获取当前执行流的上下文，并保存在 ucp 指向的 ucontext_t 结构体中。（初始化）

  2. setcontext(const ucontext_t \*ucp):

    切换到 ucp 指定的上下文执行。调用此函数后，程序的控制流会转移到新的上下文继续执行，当前函数不会返回。

  3. makecontext(ucontext_t \*ucp, void (\*func)(), int argc, ...):

    修改 ucp 指定的上下文，设置它的执行函数和函数参数。func 应是一个不返回值的函数，因为从该函数返回将导致程序终止。（第一次swap到ctx时，就开始执行func；后面swap回来时，继续执行func）

  4. swapcontext(ucontext_t \*oucp, const ucontext_t \*ucp):

    保存当前上下文到 oucp 并切换到 ucp 指向的上下文。通常用于实现协程之间的切换。

ucontext_t 结构体

ucontext_t 结构体是管理上下文的核心，它包括以下几个部分：

  • uc_link:

    当前上下文结束后，程序执行将继续的上下文的指针。

  • uc_stack:

    描述当前上下文使用的堆栈的信息，包括堆栈的位置、大小和标志。

  • uc_mcontext:

    保存特定于机器的执行上下文信息（如寄存器），这部分是系统特定的。

  • uc_sigmask:

    上下文执行时的信号屏蔽集，用于控制哪些信号在此上下文中被阻塞。

代码示例：

 1#include <stdio.h>
 2#include <ucontext.h>
 3 
 4ucontext_t ctx[3]; // 创建三个上下文对象，对应三个协程
 5ucontext_t main_ctx; // 主上下文，用于协程之间的调度
 6 
 7int count = 0; // 全局变量，用于控制打印次数和协程切换
 8 
 9// coroutine1
10void func1(void) {
11    // 当 count 小于 30 时持续执行
12	while (count ++ < 30) {
13		printf("1\n"); // 打印标识符1
14		swapcontext(&ctx[0], &main_ctx); // 从协程1切换到主上下文
15		printf("4\n"); // 当协程1再次获得控制时，打印标识符4
16	}
17}
18 
19// coroutine2
20void func2(void) {
21    // 当 count 小于 30 时持续执行
22	while (count ++ < 30) {
23		printf("2\n"); // 打印标识符2
24		swapcontext(&ctx[1], &main_ctx); // 从协程2切换到主上下文
25		printf("5\n"); // 当协程2再次获得控制时，打印标识符5
26	}
27}
28 
29// coroutine3
30void func3(void) {
31    // 当 count 小于 30 时持续执行
32	while (count ++ < 30) {
33		printf("3\n"); // 打印标识符3
34		swapcontext(&ctx[2], &main_ctx); // 从协程3切换到主上下文
35		printf("6\n"); // 当协程3再次获得控制时，打印标识符6
36	}
37}
38 
39// schedule
40int main() {
41    // 分配堆栈空间
42	char stack1[2048] = {0};
43	char stack2[2048] = {0};
44	char stack3[2048] = {0};
45 
46    // 初始化协程1的上下文，并设置堆栈和链接
47	getcontext(&ctx[0]);
48	ctx[0].uc_stack.ss_sp = stack1;
49	ctx[0].uc_stack.ss_size = sizeof(stack1);
50	ctx[0].uc_link = &main_ctx;
51	makecontext(&ctx[0], func1, 0);
52 
53    // 初始化协程2的上下文，并设置堆栈和链接
54	getcontext(&ctx[1]);
55	ctx[1].uc_stack.ss_sp = stack2;
56	ctx[1].uc_stack.ss_size = sizeof(stack2);
57	ctx[1].uc_link = &main_ctx;
58	makecontext(&ctx[1], func2, 0);
59 
60    // 初始化协程3的上下文，并设置堆栈和链接
61	getcontext(&ctx[2]);
62	ctx[2].uc_stack.ss_sp = stack3;
63	ctx[2].uc_stack.ss_size = sizeof(stack3);
64	ctx[2].uc_link = &main_ctx;
65	makecontext(&ctx[2], func3, 0);
66 
67    // 开始从主上下文切换到协程
68	printf("swapcontext\n");
69	while (count <= 30) { // 调度器，循环切换到各个协程
70		swapcontext(&main_ctx, &ctx[count % 3]); // 循环切换到三个协程
71	}
72 
73	printf("\n");
74}

关键点：

  1. 初始化上下文：为每个协程分配堆栈并设置其执行的函数。
  2. 协程调度：主函数中的 while 循环实现简单的轮转调度（Round-Robin），根据 count 的值循环切换到三个协程的上下文。
  3. swapcontext()：保存当前上下文并激活新的上下文。在每个协程中，它将控制权交回给主上下文，而主上下文则根据调度逻辑选择下一个要运行的协程。

运行结果：

 11
 22
 33
 44
 51
 65
 72
 86
 93
104
111
125
132
146
153
164
171
185
192
206
213
224
231
245
252
266
273
284
291
305
312
326
333
344
351
365
372
386
393
404
411
425
432
446
453
464
471
485
492
506
513
524
531
545
552
566
573
584

伪代码框架解析

1struct coroutine{
2    int fd;                 // 文件描述符，用于网络 I/O
3    ucontext_t ctx;        // 上下文信息，用于协程切换
4    void *arg;             // 函数参数
5    queue_node(coroutine, ) ready_queue;  // 准备就绪的协程队列
6    rbtree_node(coroutine, ) wait_rb;     // 等待中的协程红黑树
7    rbtree_node(coroutine, ) sleep_rb;    // 睡眠中的协程红黑树
8};

这里，每个协程包括一个文件描述符、上下文信息、函数参数以及三种状态的管理节点：就绪、等待和睡眠。

1void func(void){
2}
3typedef void *(*coroutine_entry)(void *)
4int create_coroutine(co_id, coroutine_entry entry, void *arg){
5    struct coroutine *co = malloc(sizeof(struct coroutine));
6    co->ss_sp = 
7    makecontext(&co->ctx, func, 0);
8}

在这段代码中，create_coroutine 函数用于初始化一个协程。它分配内存，设置栈空间，并使用 makecontext 准备上下文切换。func 是协程将要执行的函数。

1// 调度器
2struct scheduler{
3    int epfd;                          // epoll 文件描述符
4    struct epoll_event events[];       // 事件数组
5 
6    queue_node(coroutine, ) ready_head;  // 就绪队列头
7    rbtree_root(coroutine, ) wait;        // 等待红黑树
8    rbtree_root(coroutine, ) sleep;       // 睡眠红黑树
9}

调度器负责管理所有协程的生命周期和状态转换。它使用 epoll 监听事件，根据 I/O 事件调度协程。就绪、等待和睡眠队列/树用于管理不同状态的协程。

通过使用协程，你可以高效地管理数以万计的并发连接，每个连接使用非常少的内存，同时避免多线程编程中常见的同步和竞争状态问题。

hook实践

基本概念

Hook 是一种编程技术，用于拦截和修改函数或方法的调用。通过 hook，程序员可以在函数调用前后插入自定义逻辑，从而改变程序的行为。Hook 常用于以下几个场景：

  • 调试：插入调试代码来监控函数调用及其参数和返回值。
  • 监控：监控系统或应用程序的行为，例如记录日志。
  • 扩展功能：在不修改原始代码的情况下添加新的功能。
  • 安全性：拦截和修改系统调用以增强安全性。

Hook 的实现方式有多种，常见的包括：

  • 动态链接库插桩：使用动态链接库 (DLL) 的加载机制来替换或扩展现有函数。
  • 代码注入：将自定义代码注入到目标进程中。
  • 函数指针替换：修改函数指针，使其指向自定义的函数。
  • 虚函数表 (VTable) 替换：在 C++ 中，通过替换对象的虚函数表条目来实现 hook。

代码实践

定义函数指针类型和变量

1typedef ssize_t (*read_t)(int fd, void *buf, size_t count);
2read_t read_f = NULL;
3 
4typedef ssize_t (*write_t)(int fd, const void *buf, size_t count);
5write_t write_f = NULL;

定义了两个函数指针类型 read_t 和 write_t，并声明了两个全局变量 read_f 和 write_f。这些变量将用于保存原始的 read 和 write 函数地址。 Hook 函数实现

 1ssize_t read(int fd, void *buf, size_t count) {
 2    struct pollfd fds[1] = {0};
 3 
 4    fds[0].fd = fd;
 5    fds[0].events = POLLIN;
 6 
 7    int res = poll(fds, 1, 0);
 8    if (res <= 0) {
 9        getcontext(&main_ctx);
10        swapcontext(&main_ctx, &read_ctx); // fd --> ctx
11    }
12 
13    ssize_t ret = read_f(fd, buf, count);
14    printf("read: %s\n", (char *)buf);
15    return ret;
16}

这是自定义的 read 函数，它会拦截所有对 read 函数的调用，并在调用原始 read 函数之前执行一些逻辑。

  • poll 调用：使用 poll 检查文件描述符 fd 是否有数据可读。
  • 上下文切换：如果没有数据可读，保存当前上下文并切换到 read_ctx。
  • 调用原始 read：调用保存的原始 read 函数 read_f。
  • 打印读取的数据：将读取的数据打印出来。

另一个 Hook 函数

1ssize_t write(int fd, const void *buf, size_t count) {
2    printf("write: %s\n", (const char *)buf);
3    return write_f(fd, buf, count);
4}

自定义的 write 函数会拦截所有对 write 函数的调用，打印写入的数据，并调用原始 write 函数。 初始化 Hook

1void init_hook(void) {
2    if (!read_f) {
3        read_f = dlsym(RTLD_NEXT, "read");
4    }
5    if (!write_f) {
6        write_f = dlsym(RTLD_NEXT, "write");
7    }
8}

init_hook 函数用于初始化 read_f 和 write_f，它们分别指向原始的 read 和 write 函数。dlsym 函数从共享库中查找符号的地址，这里查找的是 read 和 write 函数。 主函数

 1int main() {
 2    init_hook();
 3 
 4    // 初始化 read_ctx 上下文
 5    getcontext(&read_ctx);
 6    read_ctx.uc_link = &main_ctx;
 7    read_ctx.uc_stack.ss_sp = malloc(SIGSTKSZ);
 8    read_ctx.uc_stack.ss_size = SIGSTKSZ;
 9    makecontext(&read_ctx, (void (*)(void))read, 0);
10 
11    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
12 
13    struct sockaddr_in serveraddr;
14    memset(&serveraddr, 0, sizeof(struct sockaddr_in));
15 
16    serveraddr.sin_family = AF_INET;
17    serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
18    serveraddr.sin_port = htons(2048);
19 
20    if (-1 == bind(sockfd, (struct sockaddr*)&serveraddr, sizeof(struct sockaddr))) {
21        perror("bind");
22        return -1;
23    }
24 
25    listen(sockfd, 10);
26 
27    struct sockaddr_in clientaddr;
28    socklen_t len = sizeof(clientaddr);
29    int clientfd = accept(sockfd, (struct sockaddr*)&clientaddr, &len);
30    printf("accept\n");
31 
32    while (1) {
33        char buffer[128] = {0};
34        int count = read(clientfd, buffer, 128);
35        if (count == 0) {
36            break;
37        }
38        write(clientfd, buffer, count);
39        printf("sockfd: %d, clientfd: %d, count: %d, buffer: %s\n", sockfd, clientfd, count, buffer);
40    }
41 
42    free(read_ctx.uc_stack.ss_sp);
43 
44    return 0;
45}

在主函数中，首先调用 init_hook 初始化 read_f 和 write_f。然后初始化 read_ctx 上下文，并使用 makecontext 函数设置上下文的执行函数为 read。

接下来，创建一个套接字，并绑定到本地地址，开始监听连接。接受一个客户端连接后，进入循环，读取客户端发送的数据，并将数据回写给客户端。

通过这种方式，read 和 write 的行为被拦截，并增加了自定义逻辑。

完整代码

  1 
  2 
  3#define _GNU_SOURCE
  4 
  5#include <dlfcn.h>
  6 
  7#include <stdio.h>
  8#include <ucontext.h>
  9#include <string.h>
 10#include <unistd.h>
 11#include <fcntl.h>
 12 
 13#include <sys/socket.h>
 14#include <errno.h>
 15#include <netinet/in.h>
 16 
 17#include <pthread.h>
 18#include <sys/poll.h>
 19#include <sys/epoll.h>
 20 
 21#if 1
 22// hook
 23typedef ssize_t (*read_t)(int fd, void *buf, size_t count);
 24read_t read_f = NULL;
 25 
 26typedef ssize_t (*write_t)(int fd, const void *buf, size_t count);
 27write_t write_f = NULL;
 28 
 29ssize_t read(int fd, void *buf, size_t count)
 30{
 31 
 32	struct pollfd fds[1] = {0};
 33 
 34	fds[0].fd = fd;
 35	fds[0].events = POLLIN;
 36 
 37	int res = poll(fds, 1, 0);
 38	if (res <= 0)
 39	{ //
 40 
 41		// fd --> epoll_ctl();
 42 
 43		swapcontext(); // fd --> ctx
 44	}
 45	// io
 46 
 47	ssize_t ret = read_f(fd, buf, count);
 48	printf("read: %s\n", (char *)buf);
 49	return ret;
 50}
 51 
 52ssize_t write(int fd, const void *buf, size_t count)
 53{
 54 
 55	printf("write: %s\n", (const char *)buf);
 56 
 57	return write_f(fd, buf, count);
 58}
 59 
 60void init_hook(void)
 61{
 62 
 63	if (!read_f)
 64	{
 65		read_f = dlsym(RTLD_NEXT, "read");
 66	}
 67 
 68	if (!write_f)
 69	{
 70		write_f = dlsym(RTLD_NEXT, "write");
 71	}
 72}
 73 
 74#endif
 75 
 76int main()
 77{
 78 
 79	init_hook();
 80 
 81	int sockfd = socket(AF_INET, SOCK_STREAM, 0);
 82 
 83	struct sockaddr_in serveraddr;
 84	memset(&serveraddr, 0, sizeof(struct sockaddr_in));
 85 
 86	serveraddr.sin_family = AF_INET;
 87	serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
 88	serveraddr.sin_port = htons(2048);
 89 
 90	if (-1 == bind(sockfd, (struct sockaddr *)&serveraddr, sizeof(struct sockaddr)))
 91	{
 92		perror("bind");
 93		return -1;
 94	}
 95 
 96	listen(sockfd, 10);
 97 
 98	struct sockaddr_in clientaddr;
 99	socklen_t len = sizeof(clientaddr);
100	int clientfd = accept(sockfd, (struct sockaddr *)&clientaddr, &len);
101	printf("accept\n");
102 
103	while (1)
104	{
105 
106		char buffer[128] = {0};
107		int count = read(clientfd, buffer, 128);
108		if (count == 0)
109		{
110			break;
111		}
112		write(clientfd, buffer, count);
113		printf("sockfd: %d, clientfd: %d, count: %d, buffer: %s\n", sockfd, clientfd, count, buffer);
114	}
115 
116	return 0;
117}

通过 hook 技术，可以拦截系统函数的调用，并在调用之前或之后执行自定义逻辑。这在调试、监视和修改函数行为时非常有用。在代码中，hook 技术用于拦截 read 和 write 函数的调用，打印读取和写入的数据，并实现上下文切换的逻辑。

相关文章