Linux C实现纯用户态抢占式多线程！

温州皮鞋厂老板，只有皮鞋！

2010年的陈年旧事：
纯用户空间抢占式多线程的设计： https://blog.csdn.net/dog250/article/details/5735512

纯用户空间的抢占式多线程库其实是很麻烦的一件事！

嗯，九年前的事情了。当时一直想做一个纯用户态的多线程，然而最终没有找到优雅的方法。

五一放假前的周六单休，在家带娃，想到一个办法。今夜作本文记下。

如果要找 纯用户态多线程 的实现，那太多了，但是几乎都是 非抢占式 的！都是某种协程的形式，需要指令流自己放弃CPU，将执行权主动交给其它的执行流。为什么很少有人研究 纯用户态抢占式多线程 呢？原因我想大概如下吧：

这种多线程严重依赖操作系统内核本身提供的机制，比如Solaris的call up机制这种，每一个系统不一定提供相同的类似机制，甚至不一定提供该机制。
当前线程库非常多，且Linux也早就实现了纯内核多线程，再去实现这个用户态的多线程没有意义。
即便实现了纯用户态线程，每一个线程也会受制于操作系统，比如一个线程在操作系统内核阻塞了，其它的线程也要跟着等…
没有固定的规范化的性能评估体系，很难度量性能开销。这个和基于OS内核的线程不同，因为后者是有严格的软硬件接口profile的。
做这个事情于工作无益，只适合自己玩玩，但是谁会没事了玩这个呢？

我也知道，做这个事情没有意义，但是什么是意义？

直接说吧，做完这个事情，至少对Linux内核对信号处理的理解更深了一步吧，或者退一万步，至少，通过这事的练习，你对sigcontext和sigframe的结构体了解了是吧。

还觉得没有意义？那就尽管喷吧。

所谓 抢占式多线程调度 ，就是不依靠线程自己来放弃CPU从而将执行权交给别的线程，而是靠一种外部主动干扰模式的调度机制，在需要调度的时刻，强行剥夺当前线程的执行权，依靠策略选择另一个线程来运行。

当时之所以没有找到优雅的方案，是因为我没有找到什么地方可以同时做到两件事：

中断当前的线程执行，进入一个handler来根据调度策略实施调度和切换。
在这个handler中修改该进程的寄存器上下文，剥夺当前线程的执行权交给另一个线程。

首先，上述第1点是可以用信号完成的，比如用alarm函数，可以实现分时中断。然而在中断处理函数中，我没有找到修改寄存器的方法，曾经想过用setjmp/longjmp，然而失败，最终使用PTRACE机制实现了一个无比丑陋和粗糙的。

在九年前的文章的中，开篇我就说 纯用户空间的抢占式多线程库其实是很麻烦的一件事！ 确实麻烦，之所以这么认为就是因为上面的难题没有解决。

当时确实是术业不精啊。后面的几年，自己也没怎么看过Linux内核信号处理相关的东西。

周六恰逢正则喝完奶睡着了之后，我一个人又不能出去浪，突然就又想到了这个问题。我发誓要找一个优雅的方案出来，毕竟九年过去了，我想自己的内功应该比那时强太多了。

确实，这个方案也真的是信手拈来。

我知道，Linux进程在执行流返回用户态前处理信号的时候，要调用信号处理函数，而这个信号处理函数是定义在用户态的，所以Linux进程为了可以执行这个handler函数，便需要自己setup一下用户态堆栈。

而这个机制，恰恰给了我们修改寄存器的机会。

九年前，我一直以为用户态的寄存器上下文在完全返回用户态之前，始终是保存在内核栈上，无法修改。但事实上，当执行信号处理函数的时候，内核会把该进程内核栈上的寄存器上下文sigcontext拷贝到用户态的堆栈中，再压入一个sigreturn系统调用作为返回地址，然后等信号处理函数完成后，sigreturn将会自动陷入内核，再将用户态的sigcontext拷贝回内核栈，以彻底完成信号处理，恢复进程的寄存器上下文。

也就是说，当信号处理函数被执行时，是可以在当前堆栈上找到寄存器上下文的，我们只需要在堆栈上找sigcontext结构体即可。这时，我们对其进行修改，然后这些被修改过的寄存器上下文将会在信号处理完成返回内核时，更新内核栈上的寄存器上下文，从而达到我们的目的。

那么，我先写一个信号处理函数，看看信号处理函数执行时，堆栈上都有什么：

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>int i, j, k = 0;
unsigned char *stack_buffer;
unsigned long *p;void sig_start(int signo)
{unsigned long a = 0x1234567811223344;p = (unsigned char *)&a;stack_buffer = (unsigned char *)&a;// 以下按照8字节为一组，打印堆栈的内容printf("----begin stack----\n");for (i = 0; i < 32; i++) {for (j = 0; j < 8; j++) {printf(" %.2x", stack_buffer[k]);k++;}printf("\n");}printf("----end stack----\n");if (signo = SIGINT)signal(SIGINT, NULL);if (signo = SIGHUP)signal(SIGHUP, NULL);return;
}int main()
{printf("process id is %d  %p %p\n",getpid(), main, wait_start);signal(SIGINT, sig_start);signal(SIGHUP, sig_start);for (;;);
}

让我们执行之，按下Ctrl-C给它一个SIGINT信号，看看打印的堆栈的内容：

[root@localhost ~]# ./a.out
process id is 3036  0x4007a1 0x40068d
^C----begin stack----44 33 22 11 78 56 34 12 # 这便是我们的局部变量 a98 7b 98 fa f8 7f 00 0010 11 92 b5 fc 7f 00 0080 02 3d fa f8 7f 00 00 # 这个便是信号处理函数的返回地址，调用sigreturn的01 00 00 00 00 00 00 00 # 以下的内容，需要参见内核 rt_sigframe 结构体00 00 00 00 00 00 00 0000 00 00 00 00 00 00 0002 00 00 00 01 00 00 0000 00 00 00 00 00 00 0060 10 92 b5 fc 7f 00 0010 0f 92 b5 fc 7f 00 0008 00 00 00 00 00 00 0006 02 00 00 00 00 00 00a0 05 40 00 00 00 00 00f0 11 92 b5 fc 7f 00 0000 00 00 00 00 00 00 0000 00 00 00 00 00 00 0001 00 00 00 00 00 00 0070 0e 92 b5 fc 7f 00 00 # 这玩意儿一看就是堆栈上的地址，0x7ffc...10 11 92 b5 fc 7f 00 0000 00 00 00 00 00 00 0000 00 00 00 00 00 00 0000 00 00 00 00 00 00 008d 03 3d fa f8 7f 00 0010 11 92 b5 fc 7f 00 00e3 07 40 00 00 00 00 0002 02 00 00 00 00 00 0033 00 00 00 00 00 00 0000 00 00 00 00 00 00 0000 00 00 00 00 00 00 0000 00 00 00 00 00 00 0000 00 00 00 00 00 00 00
----end stack----
^C
[root@localhost ~]#

我是在x86_64平台上做的实验，所以我们要看x86_64的rt_sigframe结构体，它位于：
arch/x86/include/asm/sigframe.h:

#ifdef CONFIG_X86_64struct rt_sigframe {char __user *pretcode;struct ucontext uc;struct siginfo info;/* fp state follows here */
};
...
/* 一路追溯，看看rt_sigframe展开后的样子 */
// include/uapi/asm-generic/ucontext.h
struct ucontext {unsigned long     uc_flags;struct ucontext  *uc_link;stack_t       uc_stack;struct sigcontext uc_mcontext;  // 这个就是我们要找的东西！sigset_t      uc_sigmask;   /* mask last for extensibility */
};

计算一下偏移位置，正好是处在 pretcode字段 的 58 字节处。也就是说，只要我们找到信号处理函数的 pretcode 偏移，将其再加 58=40 字节就是sigcontext结构体了，这个结构体里全部都是寄存器：

struct sigcontext {__u64 r8;__u64 r9;__u64 r10;__u64 r11;__u64 r12;__u64 r13;__u64 r14;__u64 r15;__u64 rdi;__u64 rsi;__u64 rbp;__u64 rbx;__u64 rdx;__u64 rax;__u64 rcx;__u64 rsp;__u64 rip;__u64 eflags;       /* RFLAGS */__u16 cs;__u16 gs;__u16 fs;__u16 __pad0;__u64 err;__u64 trapno;__u64 oldmask;__u64 cr2;struct _fpstate *fpstate;   /* zero when no FPU context */
#ifdef __ILP32____u32 __fpstate_pad;
#endif__u64 reserved1[8];
};

我们所谓的纯用户态线程调度，就是在信号处理函数里save/restore上述的结构体就好了，而上述的结构体的位置，我们已经知道它在哪里了。

#include <unistd.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>// 仅仅是测试demo，分配4096字节的stack足够了。
#define STACK_SIZE      4096
/* * 为什么是72？* 因为我们在信号处理中增加了一个局部变量，这样pretcode的偏移就是32字节了。* 于是32+40=72！*/
#define CONTEXT_OFFSET  72
// rip寄存器相对于局部变量a的偏移。注意rip在sigcontext中的偏移是16
#define PC_OFFSET       200int wait_start()
{for (;;) {sleep(1000);}
}// 线程1的处理函数
void thread1()
{int a = 1, ret = 0;char buf[64];int fd = open("./file", O_RDWR);for (;;) {// 线程1持续往一个文件里写内容。snprintf(buf, 32, "user thread 1 stack: %p  value:%d\n", &a, a++);ret = write(fd, buf, 32);printf("write buffer to file:%s  size=%d\n", buf, ret);sleep(1);}
}// 线程2的处理函数
void thread2()
{int a = 2;for (;;) {// 线程2随便打印些自己栈上的什么东西。printf("tcp user cong 2 stack: %p  value:%d\n", &a, a++);sleep(1);}
}unsigned char *buf;
int start = 0;
struct sigcontext context[2];
struct sigcontext *curr_con;
unsigned long pc[2];
int idx = 0;
unsigned char *stack1, *stack2;// SIGINT用来启动所有线程，每次信号启动一个。
void sig_start(int dunno)
{unsigned long a = 0, *p;if (start == 0) {  // 启动第一个线程// 首先定位到sigcontext的rip，启动线程仅仅修改rip即可，目标是跳入到thread1线程处理函数p = (unsigned long*)((unsigned char *)&a + PC_OFFSET);*p = pc[0];// 定位到sigcontextp = (unsigned long *)((unsigned char *)&a + CONTEXT_OFFSET);curr_con = (struct sigcontext *)p;// 初始化其堆栈寄存器为为该线程分配的独立堆栈空间。curr_con->rsp = curr_con->rbp = (unsigned long)((unsigned char *)stack1 + STACK_SIZE);start++;} else if (start == 1) { // 启动第二个线程// 定位线程1的sigcontext，保存其上下文，因为马上就要schedule线程2了。p = (unsigned long *)((unsigned char *)&a + CONTEXT_OFFSET);curr_con = (struct sigcontext *)p;memcpy((void *)&context[0], (const void *)curr_con, sizeof(struct sigcontext));// 保存第一个线程的上下文后再定位到sigcontext的rip并修改之，同线程1p = (unsigned long *)((char*)&a + PC_OFFSET);idx = 1;*p = pc[1];p = (unsigned long *)((unsigned char *)&a + CONTEXT_OFFSET);curr_con = (struct sigcontext *)p;// 初始化其堆栈寄存器为为该线程分配的独立堆栈空间。curr_con->rsp = curr_con->rbp = (unsigned long)((unsigned char *)stack2 + STACK_SIZE);start++;// 两个线程均启动完毕，开启时间片轮转调度吧。alarm(2);signal(SIGINT, NULL);}return;
}void sig_schedule(int unused)
{unsigned long a = 0;unsigned char *p;// 保存当前线程的上下文p = (unsigned char *)((unsigned char *)&a + CONTEXT_OFFSET);curr_con = (struct sigcontext *)p;memcpy((void *)&context[idx%2], curr_con, sizeof(struct sigcontext));// 轮转调度下一个线程，恢复其上下文。idx++;memcpy(curr_con, (void *)&context[idx%2], sizeof(struct sigcontext));// 2秒后再调度alarm(2);return;
}int main()
{printf("process id is %d  %p %p\n",getpid(), thread1, thread2);// 为两个线程分配stack空间。// 注意，线程的stack空间一定要独立，不然函数调用会冲突的。stack1 = (unsigned char *)calloc(1, 4096);stack2 = (unsigned char *)calloc(1, 4096);signal(SIGINT, sig_start);signal(SIGALRM, sig_schedule);pc[0] = (unsigned long)thread1;pc[1] = (unsigned long)thread2;wait_start();
}

效果如下：

[root@localhost ~]# ./a.out
process id is 2994  0x4007cd 0x400869
0x1191010 0x1192020
^Cwrite buffer to file:user thread 1 stack: 0x1191ffc  value:1size=32
^Ctcp user cong 2 stack: 0x1193014  value:2
tcp user cong 2 stack: 0x1193014  value:3
write buffer to file:user thread 1 stack: 0x1191ffc  value:2size=32
write buffer to file:user thread 1 stack: 0x1191ffc  value:3size=32
tcp user cong 2 stack: 0x1193014  value:4
tcp user cong 2 stack: 0x1193014  value:5
write buffer to file:user thread 1 stack: 0x1191ffc  value:4size=32
write buffer to file:user thread 1 stack: 0x1191ffc  value:5size=32
tcp user cong 2 stack: 0x1193014  value:6
^C

可以看出，两个线程完美交替执行！

第一个例子有点复杂了，我们换个简单的：

void thread1()
{int i = 1;while (1) {printf("I am thread:%d\n", i);sleep(1);}
}
void thread2()
{int i = 2;while (1) {printf("I am thread:%d\n", i);sleep(1);}
}

效果如下：

[root@localhost ~]# ./a.out
process id is 3085  0x4006fd 0x40072c
0x11f3010 0x11f4020
^CI am thread:1 # Ctrl-C 启动线程1
^CI am thread:2 # Ctrl-C 启动线程2
I am thread:2
I am thread:1
I am thread:1
I am thread:2
I am thread:2
I am thread:1
I am thread:1
I am thread:2
I am thread:2
I am thread:1
I am thread:1
I am thread:2
I am thread:2
I am thread:1
I am thread:1
^C

这个例子清晰多了。

以上的 纯用户态 多线程设计中，没有使用任何操作系统进程级别以外的数据结构存储线程上下文，这就是 纯用户态 的含义。我们看到所有的线程上下文以及线程调度相关的数据结构都存储在单独的一个进程地址空间。

换句话说， 在单独的该进程之外，没人意识得到这个多线程的存在！ 这个容纳用户态多线程的进程容器，就是一个 虚拟机 实例，它完成了线程硬件上下文的save/restore，调度，切换，就像Linux内核之于Linux进程所做的那般。

我说，我的这个纯用户态抢占式多线程，使用了信号处理机制。

有人会问，不是说 “纯” 用户态吗？干嘛用信号？信号不是内核机制吗？

是的，信号是内核机制，但 这里的关注点不在信号是不是内核机制，而是“需要一个第三方来实施抢占式调度”， 为什么需要 “第三方”？

因为抢占式多线程不是协作式多线程，既然线程自己不参与调度决策，那就必然需要第三方来决策。使用信号只是一种方式，由于我在Linux系统做这个用户态多线程，信号恰恰是可以满足需求的。当然，也可以不用信号，如果你能找到等价的机制也是可以的。

【 注意⚠️： 采用信号机制来抢占的开销确实有点大，但是这只是一种可实现的方式，并不是唯一方式，此外，这种抢占式调度完全是可以和协作式调度比如协程协同工作的，只有在发生 某种不得不抢占 的事件后，才实施信号抢占。】

操作系统内核调度不也依赖第三方的时钟中断吗？时钟晶振可不是内核的一部分，它是硬件。

此时4月29日凌晨2:13，合上电脑，或许还能再睡几个钟头。

浙江温州皮鞋?湿，下雨进水不会胖！

Linux C实现纯用户态抢占式多线程！相关推荐

Linux C实现用户态协作式多线程！
皮鞋?湿,不会胖,下雨也不怕!但皮鞋老板不让老湿说协程,那老湿就不说了,毕竟也真的不懂. 前天半夜写下一篇文章作为对九年前一个疑问的回应: Linux C实现纯用户态抢占式多线程!: https:// ...
Linux mem 1.1 用户态进程空间的创建 --- execve() 详解
文章目录 1. 原理介绍 1.1 固定地址映射 1.2 随机地址映射(ASLR) 1.3 文件映射 1.4 stack 2. 代码详解 2.1 execve() 2.1.1 bprm_mm_init( ...
Linux网络设计之用户态协议栈与dpdk
用户态协议栈设计与dpdk dpdk环境开启 Windowe下配置静态IP表 DPDK API介绍 struct rte_memzone结构体 struct rte_mempool结构体 struct ...
linux 内核信号量与用户态信号量(system v,信号量在Linux多线程机制中的应用
[摘要]本文以信号量原理为基础,重点阐述信号量在Linux多线程同步机制中的实现特色. [关键词]信号量:Linux:多线程:同步 1 信号量 1965年E. W. Dijkstra首次提出信号量的 ...
Linux C语言在用户态实现一个低时延通知（eventfd）+轮询（无锁队列ring）机制的消息队列
目录 fastq.c fastq.h test-0.c test-1.c https://github.com/Rtoax/test/tree/master/ipc/github/fastq fast ...
Linux用户态进程如何监控内存被写事件
上周了解到一个很好玩的问题,即如何捕获到"一块特定的内存的内容变成某一个特定的值"这么一个事件. 嗯,还是那位暴雨天穿着意尔康皮鞋给我们送伞皮鞋湿了的同事,感谢他能提供一些好玩的 ...
Linux内核态抢占机制分析
http://blog.sina.com.cn/s/blog_502c8cc401012pxj.html [摘要]本文首先介绍非抢占式内核(Non-Preemptive Kernel)和可抢占式内核( ...
linux禁止内核抢占,Linux内核态抢占机制分析
[51CTO晃荡]8.26 带你深度懂得清华大年夜学.搜狗基于算法的IT运维实践与摸索本文起首介绍非抢占式内核(Non-Preemptive Kernel)和可抢占式内核(Preemptive Ke ...
linux用户态驱动--VFIO（一）
序言设备驱动可以运行在内核态,也可以运行在用户态,不管用户态驱动还是内核态驱动,他们都有各自的缺点.内核态驱动的问题是:系统调用开销大:学习曲线陡峭:接口稳定性差:调试困难:bug致命:编程语言选择 ...
用户态Linux内核
User Mode Linux 是可以在用户态启动的 Linux版本,最新版linux内核已提供了支持.这使我们能在类似 OpenVZ 虚拟化技术的系统上,使用最新的 Linux 内核:并且可以在非 ...

Linux C实现纯用户态抢占式多线程！

Linux C实现纯用户态抢占式多线程！相关推荐

最新文章

热门文章