python3线程池源码解析_源码解读TDengine中线程池的实现

这篇文章中提到了 tsched 的源码可以一读&＃xff0c;所以去阅读了一下&＃xff0c;总共220来行。

1. 阅读前工作

通过上文了解到这段程序实现的是一个任务队列&＃xff0c;同时带有线程池。这段程序是计算机操作系统里经典的consumer-producer (生产者-消费者)问题的实现。凡是学过操作系统这门课的&＃xff0c;都应该知道这个问题&＃xff0c;做过习题。在阅读源码之前可以先尝试用伪代码实现上述生产者-消费者问题。

2. 如何阅读&＃xff1f;

了解清楚使用场景

这是一个线程池&＃xff0c;客户端可以提交任务&＃xff0c;线程池按照顺序调度执行任务。通过阅读 tsched.h 头文件&＃xff0c;知道主要有三个函数&＃xff1a;

初始化命名的调度器、线程池&＃xff1a;taosInitScheduler

生产者提交某个任务&＃xff1a;taosScheduleTask

程序结束时的清理工作&＃xff1a;taosCleanUpScheduler

通过搜索上述三个函数的调用&＃xff0c; 知道初始化了两个调度器&＃xff0c;有三个地方会提交任务。

两个线程池

定时器里的 tmr 线程池 : 队列长度一万&＃xff0c;只有一个线程服务。此线程会执行到期的 timer 的回调函数。

tsc 线程池&＃xff1a;队列长度一万&＃xff0c;线程数量为所在机器 CPU 核心数的一半。这些线程负责&＃xff1a;异步操作如执行语句&＃xff0c;固定大小滑动窗口流式数据处理

两个生产者

上面提到了&＃xff0c;有三个生产者会提交任务给线程池&＃xff1a;

timer

stream

了解了清楚使用方、使用场景后&＃xff0c;就容易读懂逻辑了。这里是一个标准的操作系统中生产者消费者的问题&＃xff0c;用的也是标准解法&＃xff1a;使用一个互斥量&＃xff0c;两个信号量。线程池使用 pthread 来创建。

关键的数据结构

SSchedQueue 里面就是上述问题中的核心数据结构&＃xff0c;除了放置上述提到的互斥量&＃xff0c;信号量&＃xff0c;还需要一个队列来存储要具体执行的任务。

SSchedMsg 结构来表示线程池任务&＃xff0c;包含要执行的具体函数及所需参数。

源码里注释并不多&＃xff0c;只能通过看具体实现来了解上述支持的执行模式。看到支持两种模式&＃xff1a;执行fp&＃xff0c;或者执行 tfp(ahandle, thandle)。

核心调度逻辑

上面提到了生产者&＃xff0c;一直没有提到消费者。接着读 sched.c 里的源码&＃xff0c;可以看到消费者就是线程池里每个线程的主框架逻辑&＃xff1a; taosProcessSchedQueue。平常这些线程处于阻塞状态&＃xff0c;等待任务。一旦生产者提交任务后&＃xff0c;就会通知到消费者。消费者拿到提交的任务及参数&＃xff0c;去执行。执行完之后继续进入上述阻塞的状态&＃xff0c;这样周而复始。

这里有个疑问&＃xff0c;消费者和生产者之间是异步的。消费完之后&＃xff0c;总得有办法通知消费者&＃xff0c;这一步在哪里做呢&＃xff1f;读到这里可以花点时间翻翻源码&＃xff0c;找找答案。

其实秘密也藏在当时提交任务的数据结构里。TDengine 里有样例代码&＃xff0c;翻了翻&＃xff0c;找到了这个 async demo。可以看到 taos_query_a 就是一个异步的query函数&＃xff0c;里面带了 query语句异步执行完成后的回调函数&＃xff1a;taos_insert_call_back)。

3. 一些思考

看的时候内心不断在思考、对比&＃xff0c;比如优势、劣势是什么&＃xff1f;我会怎么实现

优势

为何使用线程池&＃xff1f;

通过固定线程池大小来固定资源开销&＃xff0c;而且是程序初始化时申请资源&＃xff0c;这在嵌入式设备里是非常重要的&＃xff0c;如果资源不够用&＃xff0c;那就快速失败&＃xff0c;在程序一开始启动时就报错。

复用了线程&＃xff0c;因为创建、销毁线程都是有开销的。这样在频繁创建、销毁线程情况下&＃xff0c;可以节省开销&＃xff0c;复用之前的线程。

任务和线程解耦&＃xff1a;需要使用多线程的地方&＃xff0c;只管提交任务就好了。线程的初始化、运行、状态切换由线程池来负责。

劣势

操作异步化&＃xff0c;对程序员的心智要求更高。需要使用回调函数&＃xff0c;需要存储上下文。但是在上述场景里还好&＃xff0c; 都是一些固定的逻辑。

调试较麻烦&＃xff0c;不是直来直去的逻辑。需要通过分析上下文及回调函数里的日志来分析问题。

有没有其他实现方式&＃xff1f;

如果用 Go 语言实现&＃xff0c;会很简单。使用 channel 来做任务分发&＃xff0c;本身就是线程安全的。

使用 C 来写&＃xff0c;个人觉得会限制 TDengine 的开源参与方。因为现在市场上会 C 的人比较少&＃xff0c;而且主要集中在嵌入式领域。而且 C 的生态一般&＃xff0c;语言的轮子比较少&＃xff0c;所以很多工作都需要自己做&＃xff0c;比如 http server&＃xff0c;rpc 等。如果让我来设计实现 TDengine&＃xff0c;我可能会优先考虑 Rust&＃xff0c;既能精准控制内存&＃xff0c;又有比较完善的社区&＃xff0c;而且语言处于上升期&＃xff0c;容易成为其中的明星项目&＃xff0c;会有推广优势&＃xff0c;比如能吸引一些本身对数据库不怎么关注&＃xff0c;但是对 Rust 感兴趣的程序员。

4. 一个思考题

通过搜索 pthread_create 可以发现系统中还有其他创建线程的地方&＃xff0c;并没有用到上述的线程池&＃xff0c;比如 dnodeMWrite, TcpPool&＃xff0c;cache&＃xff0c;sync等。这些地方为什么没有使用线程池呢&＃xff1f;

欢迎关注我的微信公众账号&＃xff0c;会在第一时间更新&＃xff0c;博客园上只有部分文章会发布