本文分享自华为云社区《GaussDB(DWS) 集群通信系列二：stream线程池设计》，作者：半岛里有个小铁盒。

1.前言

适用版本：【8.1.0(及以上)】

GaussDB(DWS)分布式架构的Stream算子作为SQL join操作时频繁发生的执行算子，共存在三种模式：Gather、Redistribute、Broadcast，分别负责CN节点GATHER数据，DN节点REDISTRIBUTE和BROACAST数据。大集群高并发场景下，Stream算子过多可能会导致通信的性能瓶颈，引起性能劣化（2000个stream同时启动，进程初始化耗时从ms级劣化到s级），因此需要尽可能减少Stream算子。但是在某些现场环境下，存在数据倾斜、join查询不包含必要分布键等客观情况，Stream算子无法有效减少，为多表join场景下的查询时延保障带来挑战。因此GaussDB(DWS)对于线程初始化->线程任务执行->线程退出执行的流程方面做了stream线程池优化，减少了线程初始化与线程退出所带来的开销。

2.实现原理

stream线程是临时线程，随query启动和退出，负责stream算子的执行，stream线程初始化和退出都会争抢锁等进程级资源，在stream线程个数无法进一步优化的场景下，需要设计有效方案以减少stream线程初始化和退出的时间代价，将进程初始化耗时稳定在ms级，保障数据库的确定性时延查询。Stream线程池的核心思想是等stream线程执行完计划任务，保留必要且可复用的线程信息，将线程放入线程池中。

线程池中的线程执行过程如上图所示，其具体步骤为：

• 步骤一：线程信息初始化
• 步骤二：线程待唤醒后轻量级初始化（query级初始化）
• 步骤三：线程任务执行
• 步骤四：线程清理
• 返回步骤二：继续等待下条query执行

在返回步骤二时，当线程等待超时、超出线程池容量（最大stream线程个数）、异常时线程已不可用，需要销毁。

其中步骤一中在线程初始化时，需要执行的操作有：线程创建、创建相关内存上下文、信号处理函数注册、内存追踪信息初始化、初始化GUC选项等操作；

步骤二中在线程轻量级/查询级初始化时，需要执行的操作有恢复GUC参数、初始化BackendParams、重置GUC参数等操作。

stream线程池为了高效管理线程的出/入池操作，采用无锁队列实现。定义结构体ThreadSlot保存线程池中每一个线程的信息，包含：线程状态、线程号、线程对应的database oid、线程执行所需的信息StreamProducer、线程唤醒所需的锁和条件变量。

当线程还未被创建时，初始化一定数量的ThreadSlot数量以预留stream线程，这些ThreadSlot被保存在数组threadSlots中。当stream线程执行完毕，需要将stream线程放置到表征可复用线程的无锁队列，称之为idleRing；当线程因为超时、异常等原因不再复用，需要退出时，将stream线程对应的ThreadSlot放置到表征未创建线程的无锁队列，称之为emptyRing。

idleRing的作用是为了快速获取并复用线程池中的线程，emptyRing的作用是快速获取一个未被使用的ThreadSlot结构，以创建一个新的stream线程。由于stream线程的初始化信息和database是强相关的，如果不保留database相关的信息，那么线程初始化的时间代价仍然较高，所以线程池中的线程复用时，需要满足database信息匹配。对于设计线程池而言，每一个database都应该对应一个idleRing。

综上所述，基于无锁队列的stream线程池设计如下所示：

从上图可以看出，一个线程池包含预留stream线程结构的threadSlots、一个表征未创建线程的无锁队列emptyRing和表征可复用线程的无锁队列idleRing，由于每个database对应一个idleRing，因此多个idleRing被组织为链表结构。

3.具体实现机制

3.1 数据结构设计

定义结构体ThreadSlot保存线程池中每一个线程的信息，包含：线程状态、线程号、线程对应的database oid、线程执行所需的信息StreamProducer，StreamProducer是父线程向子线程传递的唯一结构、线程唤醒所需的锁和条件变量。

typedef struct
{
    int status;
    uint32 idx;
    ThreadId tid;
    Oid dbOid;
    StreamProducer* streamObj;
    pthread_mutex_t m_mutex;
    pthread_cond_t m_cond;
} ThreadSlot;

定义结构体StreamThreadPool表征线程池，其中size表示线程池中拟预留的ThreadSlot个数，ThreadSlot被保存在threadSlots数组中；无锁队列emptyRing用来保存未创建线程的ThreadSlot，对应地，idleRing用来保存空闲的已创建stream线程的ThreadSlot。结构如下所示：

class StreamThreadPool: public BaseObject
{
public:
    StreamThreadPool();
    void Init(int num);                            				     // streamThreadPool init   
    int Call(StreamProducer* obj);                   			     // 获取idle线程 或 create 新线程
    bool Wait();                                                      // idle线程等待唤醒或者超时退出
    ThreadSlot* GetLocalSlot();                                       // get streamThreadSlot
    void SetLocalSlot(int slotIdx);                                   // set streamThreadSlot
    StreamPool* GetLocalPool();                                       // 获取streamDBPool 或 新建一个
    ThreadSlot* PopSlot();                                            // 从idleRing/emptyRing获取一slot
    void PushToEmpty(ThreadSlot* slot);                               // 将slot直接放入emptyRing
    void PushToIdle(StreamPool* pool, ThreadSlot* slot);              // 将slot直接放入idleRing
    void LocalPushToIdle();                                           // 根据状态，将slot放入idleRing
    void LocalPushToEmpty();                                          // 根据状态，将slot放入emptyRing
    int CleanStreamPool(const char *dbName, cleanOption cleanMode);   // 根据db信息清线程
    void CleanInAllStreamPool(int desNum);                            // 调整线程池中stream线程个数
    int GetStreamNum();                                               // 获取线程池中stream线程个数
    bool Release();                                                   // 判断超时线程是否需要清理
    bool TimeoutClean();                                              // 清理超时idle线程

private:
    int size;
    ThreadSlot* threadSlots;
    ArrayLockFreeQueue emptyRing;
    StreamPool* PoolListHead;
}

定义结构体StreamPool，由于stream线程的初始化信息和database是强相关的，如果不保留database相关的信息，那么线程初始化的时间代价仍然较高，所以线程池中的线程复用时，需要满足database信息匹配，所以一个emptyRing和一个database相匹配，保存在链表PoolListHead中。

typedef struct StreamPool
{
    Oid dbOid;
    ArrayLockFreeQueue idleRing;
    struct StreamPool* next;
} StreamPool;

综上，我们可以得到各结构间组织的直观图，如下所示：

上图中threadSlots可以放在idleRing(蓝色)、emptyRing(绿色)和运行空间(黄色)中。

3.2 stream线程状态转移DFA设计

每一个记录线程信息的结构ThreadSlot中都保存了线程当前的状态status，记录线程状态的目的是为了保障线程执行过程的有序控制，也可以通过状态的互斥避免threadSlot不会被两个线程同时使用。

stream线程状态转移用确定性有限状态机（DFA，definite automata）表征，共包含4个状态：

STREAM_SLOT_EXIT、STREAM_SLOT_IDLE、STREAM_SLOT_HOLD和STREAM_SLOT_RUN状态。其物理含义如下：

• STREAM_SLOT_EXIT：线程退出状态，表示线程未被创建或线程已退出；
• STREAM_SLOT_IDLE：线程可复用状态，表示线程在idleRing中，可以被复用；
• STREAM_SLOT_HOLD：线程临时独占状态，表示线程在做进入下一个状态的准备工作；
• STREAM_SLOT_RUN：线程运行状态，表示线程正在执行任务。

状态间转移条件如下所示，图中粗箭头表示状态机主循环部分：

与状态对应的，是slot所处的位置，slot所处的位置有三处，分别是idleRing、emptyRing和运行空间，slot从无锁队列中拿出，运行时所处的位置，我们称之为运行空间。各状态所处的位置情况如下所示：

• STREAM_SLOT_EXIT：idleRing(idle线程超时)、emptyRing(初始化或者FATAL)；
• STREAM_SLOT_IDLE：idleRing
• STREAM_SLOT_HOLD：运行空间（从无锁队列中取出）、idleRing（idle线程超时或中断）；
• STREAM_SLOT_RUN：运行空间。

Slot的位置变化和状态转移的关系如下，图中粗箭头表示状态机主循环部分：

根据各状态所处的位置情况，从idleRing中取出的slot可能有三种状态：EXIT、IDLE、HOLD。当取出IDLE状态的slot，说明线程可复用；当取出EXIT状态的slot，说明线程已退出，此时需要将slot转存到emptyRing；当取出HOLD状态，说明线程正在被使用，此时需要放回idleRing。

EmptyRing中slot的状态只能是EXIT，运行空间中slot的状态要么是HOLD（刚取出还未运行），要么是RUN（正在运行），不再赘述

3.3 单个stream线程执行流程

Stream线程池中stream线程整体执行流程如下图所示：

stream线程初始化仅初始化一次，执行完query之后，便将连接归还到连接池里，循环执行上图中黄色部分的语句，如果有异常则线程退出，连接销毁，slot归还至emptyRing；如果正常执行结束，将连接中内容清理，避免下个连接误用，并将slot归还至idleRing等待下个连接复用。

那么stream线程复用时如何保持参数的一致性呢，对应上图中的set GUC params阶段。父线程保存自己的guc_variables在syncGucVariables中，syncGucVariables是需要传递给stream的结构用以保证父子线程guc参数的一致。然后父线程在初始化streamProducer时将syncGucVariables保存在该结构中传递。Stream线程根据streamProducer初始化自己的syncGucVariables变量，首先reset所有的guc变量，然后根据syncGucVariables修正自己的variables。

4.外部接口

4.1 GUC参数

max_stream_pool：设置stream线程池能够容纳stream线程的最大个数。该参数8.1.2及以上版本支持。默认值为65535。设置为-1表示不开启stream线程池。该参数支持reload更新，更新规则：设置max_stream_pool小于当前可用线程个数，支持线程个数实时减少；当设置max_stream_pool大于当前idle线程个数，将由业务驱动线程个数的增加

4.2 视图

pg_thread_wait_status：展示了集群所有CN/DN进程内的所有线程的实时 等待状态，是定位集群通信问题最重要的视图

其中对于wait_status列状态说明如下：

• wait stream task：空闲的stream线程；

• wait node：等待其他DN的数据，需要关注对端状态；

• flush data：发送数据给其他DN时因为对端buffer满而阻塞；

• wait cmd：DN上空闲的postgres线程，等待CN的下一个query；

• none：未定义状态，极有可能是阻塞原因；

• synchronize quit：同步退出状态，自身任务已完成，在等待同一个query的其他线程一起退出；

5.通过表象看stream线程池逻辑

【场景一】集群基础行为场景——建立多数据库场景

Create database ***；(建立多库)

分别执行带stream算子的查询；

例：create table test_01(c1 int, c2 int)with(orientation=column) distribute by hash(c1);
insert into test_01 select generate_series(1,100), generate_series(1,100);analyze test_01;
select * from test_01 a, test_01 b, test_01 c, test_01 d, test_01 e, test_01 f where a.c2 =b.c2 and c.c2 = d.c2 and e.c2=f.c2 limit 100;

查询结束，查pgxc_thread_wait_status看DN节点：预期stream线程状态为wait thread cond。且多database之间stream线程不复用。

【场景二】集群基础行为场景——建立多用户场景

Create user ***；(建立多用户)

分别执行带stream算子的查询；（参考场景一示例）

查询结束，查pgxc_thread_wait_status看DN节点：预期stream线程状态为wait thread cond。且多user之间stream线程可以复用。

例：用户一执行完查询，视图中显示共有四个stream线程在线程池，用户二执行同样查询返回正确结果，视图中的stream线程个数不变，且线程号也是一致的，则说明复用。

【场景三】集群基础行为场景——线程清理场景

调整guc参数max_stream_pool的值，观测是否生效；预期：当设置max_stream_pool小于当前idle线程个数，支持线程个数实时减少；当设置max_stream_pool大于当前idle线程个数，将由业务驱动线程个数的增加，但是不会超过max_stream_pool。

执行clean connection(ALL force)，查看stream线程是否被清理；预期：该database的stream线程被完全清理。

执行drop database命令，查看stream线程是否被清理；预期：该database的stream线程被完全清理。

6.总结

本文详细介绍了stream连接池及其原理，让我们更好的理解GaussDB(DWS)集群通信中数据交互的具体逻辑，对于GaussDB通信运维也具备一定的参考意义。

点击关注，第一时间了解华为云新鲜技术~