【云驻共创】数据高速公路—数仓集群通信技术详解

数据仓库服务GaussDB（DWS）是一种基于华为云基础架构和平台的在线数据分析处理数据库，提供即开即用、可扩展且完全托管的分析型数据库服务。GaussDB（DWS）是基于华为融合数据仓库GaussDB产品的云原生服务 ，兼容ANSI/ISO标准的SQL92、SQL99和SQL 2003语法，同时兼容PostgreSQL/Oracle数据库生态，为各行业PB级海量大数据分析提供有竞争力的解决方案。

本文讲解GaussDB（DWS）集群通信技术如何在大规模集群中承载高并发业务，如何实现高性能分布式通信系统。

一、GaussDB（DWS）集群进程架构和通信总览

1. DWS集群进程架构

在GaussDB（DWS）集群中，会有1个或多个协调节点（CN）、若干个数据节点（DN）、全局事物控制器（GTM）、运维管理模块（OM）、集群管理模块（CM）、数据导入导出模块（GDS）。下图为GaussDB(DWS)集群进程架构。

• CoordinatorNode（CN）协调节点：负责请求分解、调度、结果返回；SQL解析和优化；仅保存元数据，不保存数据。

• Data Node（DN）数据节点：负责存储实际表数据（指定分布方式：哈希表、复制表、RroundRobin表）；执行SQL任务并向CN返回执行结果。

• Global Transaction Manager（GTM）全局事物控制器：负责生成和维护全局事务ID、事务快照、时间戳等需要全局唯一的信息。

• Operation Manager（OM）运维管理模块：提供日常运维、配置管理。

• Cluster Manager（CM）集群管理模块：管理和监控分布式系统中各个功能单元和物理资源的运行情况，确保整个系统的稳定运行。

• GDS Loader：批量数据加载，并行加速；支持文本文件格式，错误数据自动识别。

以上模块通过集群网络相互通信，集群通信不同于执行器、优化器、存储等数据库传统模块，集群通信是分布式数据库特有的。对于集群问题定位，集群性能优化有极大的影响。

2. DWS集群通信总览

GaussDB（DWS）是MPP型分布式数据库，使用Share Nothing架构。数据分散存储在各个DN节点。CN不存储数据，作为接收查询的入口，生成的计划会尽量下推到DN并行执行以提升性能。DN执行多表Join时，因为本地DN只有部分数据，需要进行DN间的数据交换对表数据或中间结果集重分布。

本文分享主要讲述客户端、CN、DN三类进程间的通信原理和流程，分为CN通信框架和DN间通信框架。

GaussDB（DWS）一般查询的数据通信流程（如下图绿色箭头所示）：

① 客户端连接CN，下发query。

② CN连接所有DN，生成并下发执行计划。

③ DN间通过网络做表数据或中间结果交换。

④ DN本地做数据加工，将结果集返回给CN。

⑤ CN将结果集聚合加工后返回客户端。

GaussDB（DWS）为全并行分布式执行，也可以用下图来说明：

① 业务应用下发SQL 给 Coordinator，SQL 可以包含对数据的增（insert ）、删 （delete/drop）、改（update ）、查 （select ）。

② Coordinator利用数据库的优化器生成执行计划，每个 DN 会按照执行计划的要求去处理数据。

③ 因为数据是通过一致性Hash 技术均匀分布在每个节点，因此 DN 在处理数据的过程中，可能需要从其他 DN 获取数据， GaussDB 提供了三种 stream 流（广播流、聚合流和重分布流）来降低数据在DN 节点间的流动。

④ DN将结果集返回给 Coordinate 进行汇总。。

⑤ Coordinator将汇总后的结果返回给业务应用。

多线程并行执行
充分利用当前多核特点，通过多线程并发执行，提高系统吞吐量。

二、CN通信框架

1. IP和端口信息

CN中的pgxc_node系统表保存了集群所有节点的IP和端口信息（下图为例）。

• node_port和node_host为主机信息；
• node_port1和node_host1为备机信息。
• hostis_primary为主备关系，为t时，CN会先连接主机再连接备机，反之亦反。
• hostis_primary值由CM集群管理组件在主备切换时自动刷新。

GaussDB进程和监听IP与端口保存在postgresql.conf文件中，与pgxc_node系统表一致。

2. 客户端与CN通信

客户端执行查询流程（如下图所示）：

① 客户端向CN的监听端口发起连接。

② CN postmaster主线程accept连接，创建postgres线程并将连接交给此线程处理。

③ 客户端下发query到CN。

④ CN的postgres线程将查询计划下发给其他CN/DN，查询结果沿原路径返回到客户端。

⑤ 客户端查询结束，关闭连接。

⑥ CN上对应的postgres线程销毁退出。

CN与DN建连立流程，和客户端与CN建连立流程基本相同。为了减少CN与DN建立连接，以及DN进程中postgres线程创建、销毁的开销，CN端实现了pooler连接池。

3. Pooler连接池

Pooler连接池是GaussDB CN进程内保存与其他GaussDB进程数据连接的数据结构，保存了CN与其他CN/DN进程的所有连接，每一个连接都对应其他CN/DN上的一个postgres线程。Pooler连接池通过对连接和线程的复用减少了建立连接以及DN创建、销毁postgres线程产生的开销，其主要作用就是连接复用，节省建连、认证、对端线程启动初始化的开销。

Pooler复用流程（如下图所示）：

① session需要连接时，通过DB+USER为key找到正确的pooler连接池，优先从中取走现有连接。

② query结束后，CN的postgres线程并不会归还连接，连接可以用于当前session的下一个查询。

③ session结束后，CN的postgres线程会将连接还到对应的pooler，连接对应的DN上的postgres线程并不会退出，处于ReadCommand中，等待复用后CN新的postgres线程发起任务。

连接和线程是一一对应关系，并且因为线程带有DB、USER等属性连接也会带有这些属性。

4. Pooler视图

pg_pooler_status视图记录了pooler连接池中的所有连接信息（如下图所示）。每一行表示本CN发起的一个连接，对应对端进程的一个postgres线程。

• in_use为“t”表示这个连接正在某线程使用，为"f"表示空闲连接等待复用。

• tid列为本CN的持有此连接的线程号。

• node_name列为对端进程号，remote_pid列为对端线线程号。

• query_id为0或CN/DN不一致时，通过pooler视图查找CN与DN连接关系。

5. Pooler连接清理

Pooler连接清理机制有2种，分别是：Session持有的连接和Pooler空闲连接池中的连接。

Session持有的连接：

• cache_connection，是否使用pooler连接池缓存连接。
• session_timeout，客户端连接空闲超时后报错退出归还连接。
• enable_force_reuse_connections，事务结束后强制归还连接。
• conn_recycle_timeout（8.2.1），CN空闲session超时后归还连接。

Pooler空闲连接池中的连接：

• pg_clean_free_conn，清理1/4的空闲连接池连接，CM定期调用。
• clean connection，清理对应DB或user的所有空闲连接。（e.g.，clean connection to all for database postgres to USER w00308067;）

三、DN通信框架

1. stream算子

前面章节提到，GaussDB(DWS)是MPP型分布式数据库，使用Share Nothing架构，数据分散存储在各个DN节点，而两表满足join条件的数据必须分布在同一个DN上，不满足条件的表需要进行数据重分布，即产生一个stream算子。每个stream算子需要上下两个线程处理异步网络IO，下层发送数据的称为producer，上层接收数据的称为consumer。

分布式执行引擎总体来说分为节点内和跨节点两个部分：

• 节点内部分的执行遵从执行引擎的迭代模型，由核心算子完成 。
• 跨节点部分主要涉及数据在节点之间的流动，由 Stream 算子来支撑数据在计算节点之间的
  流动。

Stream 算子可以在 CN 、 DN 组件上执行。跨节点的数据传输依赖于查询分析阶段根据数据分布以及代价模型构建的数据流动拓扑结构（执行计划树），分布式执行引擎根据此结构来建立节点之间的网络连接，利用Stream算子驱动数据流动于此拓扑结构之上。

2. stream线程

DN上的stream算子都需要启动一个stream线程异步发送网络数据，如果开启了SMP并行，一个stream算子可能需要启动多个stream线程，也会建立更多的DN间连接。

stream算子（Streaming）分为以下三种：

• GATHER：CN与DN通信，收集DN结果集。
• BROADCAST：DN将本地数据全量广播给其他DN。
• REDISTRIBUTE：DN将本地数据Hash后发给对应DN。

下图示例中，wt1表分布列为c1，无需重分布，wt2表c1列不是分布列，需要重分布。

查询由一个postgres线程 （consumer线程）和一个stream线程（producer线程）共同完成。

1. stream线程扫描wt2表，Hash后发给对应的DN的postgres线程。
2. postgres线程先扫描wt1表，再接收stream线程发出的wt2表数据，再进行HashJoin后返回结果集至CN。

3. stream线程池

• stream线程池实现了DN stream线程的复用，避免了stream线程创建、初始化、清理、销毁的开销。

• stream线程池使用无锁队列实现，2000个stream线程并发启动，耗时从2秒级优化到10ms。

• stream算子需要stream线程时，通过DB name匹配对应的stream线程池，优先复用相同DB的已有线程。已创建的stream线程在查询结束后放入线程池等待复用。

• stream线程池中的线程本身具有空闲时超时退出功能，每60s超时回收1/4。

• max_stream_pool参数设置线程池缓存上限，为0时关闭stream线程池功能，也可以临时设置用于清理stream线程。

4. Libcomm通信库

1000个DN集群，每个stream线程需要建立1000个连接。如果1000 stream线程并发，DN总共需要建立100万个连接，会消耗大量的连接、内存、fd资源。基于这种情况，设计了Libcomm通信库，Libcomm通信库在一个物理长连接上模拟n个逻辑连接，使得所有并发的数据跑在一个物理连接上，解决了物理连接数过多和建连耗时的问题。

• comm_max_stream决定了每个物理连接支持的逻辑连接数量，需要大于最大并发数 * 平均stream算子数 * smp的平方。
• comm_quota_size决定了每个逻辑连接的buffer大小，如果内存充裕可以调大。
• comm_usable_memory为通信可用内存大小，如果实际使用接近此值，通信库会减少单个逻辑连接的buffer大小。

5. Libcomm逻辑连接视图

逻辑连接视图展示了DN间所有Libcomm逻辑连接的信息。

• pgxc_comm_send_stream查看所有逻辑连接发送端的信息。
• state为“READY”表示正在建连，为“HOLD”表示发送阻塞，对端buffer满。

• pgxc_comm_recv_stream查看所有逻辑连接接收端的信息。
• buff_usize为单个逻辑连接已接收到的数据缓存大小。

四、通信问题定位

1. 线程等待状态视图

线程等待状态视图 pgxc_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的其他线程一起退出。

2. 通信hang问题

上图示例中定位hang问题步骤为：

① 在pgxc_stat_activity视图中找到问题查询的query_id。

② 根据query_id查询pgxc_thread_wait_status视图，看到CN在等DN4的数据，而DN4在与DN42建连。

③ 查询pgxc_comm_send_stream看到DN4发起的与DN2的逻辑连接中有一个处于READY半连接状态。

④ 查询pgxc_comm_recv_stream看到DN2已经接受了DN4发起的4个逻辑连接，问题出在DN4逻辑连接应答上。

3. 通信报错问题

常见通信报错问题如下图所示：

4. explain performance性能分析

此示例中，可使用explain perfomance分析，按hang问题定位热点阻塞堆栈，使用gsar工具查看环境是否发生网络丢包重传。下图为各时间响应的原因分析：

5. 网络环境问题

对于网络环境问题，可使用gsar工具确认是否发生网络丢包重传；使用netstat命令确认重传发生在哪一个连接上；使用top命令在连接两端机器排查ksoftirq进程CPU占用是否有异常；还可以使用ping、telnet和tcpdump进一步分析丢包问题。

例如下图所示命令可查看各种信息：

netstat -anop|grep 'on ('|grep -v '/0/0'|sort -rnk3|head

五、总结

• GaussDB（DWS）采用全并行的 MPP 架构数据库，业务数据被分散存储在多个节点上，数据分析任务被推送到数据所在位置就近执行，并行地完成大规模的数据处理工作，实现对数据处理的快速响应。
• 全方位 HA 设计： GaussDB（DWS）所有的软件进程均有主备保证，集群的协调节点 （CN） 、数据节点 （DN）等逻辑组件全部有主备保证，能够保证在任意单点物理故障的情况下系统依然能够保证数据可靠、一致，同时还能对外提供服务。
• 通过多线程并发执行，提高系统吞吐量。
• 善用各种查询命令和系统查询命令进行通信问题的分析和定位。

本文参与华为云社区【内容共创】活动第25期。

【内容共创】活动第25期活动详情：https://bbs.huaweicloud.cn/blogs/418766

任务12.数据高速公路—数仓集群通信技术详解