957 lines
34 KiB
C++
957 lines
34 KiB
C++
/*
|
||
* Copyright (c) 2020 Huawei Technologies Co.,Ltd.
|
||
*
|
||
* openGauss is licensed under Mulan PSL v2.
|
||
* You can use this software according to the terms and conditions of the Mulan PSL v2.
|
||
* You may obtain a copy of Mulan PSL v2 at:
|
||
*
|
||
* http://license.coscl.org.cn/MulanPSL2
|
||
*
|
||
* -------------------------------------------------------------------------
|
||
*
|
||
* threadpool_controler.cpp
|
||
* 线程池的控制器。定义了 ThreadPoolControler 类，
|
||
* 用于初始化线程池的工作线程和监听线程，并将新会话从
|
||
* 主线程分派到合适的线程组。
|
||
*
|
||
*
|
||
* IDENTIFICATION
|
||
* src/gausskernel/process/threadpool/threadpool_controler.cpp
|
||
*
|
||
* ---------------------------------------------------------------------------------------
|
||
*/
|
||
#ifdef __USE_NUMA
|
||
#include <numa.h>
|
||
#endif
|
||
#include "postgres.h"
|
||
#include "knl/knl_variable.h"
|
||
|
||
#include "threadpool/threadpool.h"
|
||
|
||
#include "access/xact.h"
|
||
#include "catalog/pg_collation.h"
|
||
#include "commands/copy.h"
|
||
#include "gssignal/gs_signal.h"
|
||
#include "lib/dllist.h"
|
||
#include "libpq/ip.h"
|
||
#include "libpq/libpq.h"
|
||
#include "libpq/pqformat.h"
|
||
#include "libpq/pqsignal.h"
|
||
#include "miscadmin.h"
|
||
#include "postmaster/postmaster.h"
|
||
#include "pgxc/pgxc.h"
|
||
#include "storage/pmsignal.h"
|
||
#include "tcop/dest.h"
|
||
#include "utils/atomic.h"
|
||
#include "utils/builtins.h"
|
||
#include "utils/formatting.h"
|
||
#include "utils/guc.h"
|
||
#include "utils/memutils.h"
|
||
#include "utils/palloc.h"
|
||
#include "utils/ps_status.h"
|
||
#include "executor/executor.h"
|
||
|
||
#include "communication/commproxy_interface.h"
|
||
|
||
#ifdef HAVE_POLL_H
|
||
#include <poll.h>
|
||
#endif
|
||
|
||
ThreadPoolControler* g_threadPoolControler = NULL;
|
||
|
||
#define BUFSIZE 128
|
||
|
||
#define IS_NULL_STR(str) ((str) == NULL || (str)[0] == '\0')
|
||
#define INVALID_ATTR_ERROR(detail) \
|
||
ereport(FATAL, (errcode(ERRCODE_OPERATE_INVALID_PARAM), errmsg("Invalid attribute for thread pool."), detail))
|
||
|
||
static const long one_hundred_micro_sec = 100;
|
||
|
||
// 线程池控制器的构造函数
|
||
ThreadPoolControler::ThreadPoolControler()
|
||
{
|
||
m_threadPoolContext = NULL; // 线程池上下文
|
||
m_sessCtrl = NULL; // 会话控制
|
||
m_groups = NULL; // 线程组
|
||
m_scheduler = NULL; // 调度器
|
||
m_groupNum = 1; // 线程组数量
|
||
m_threadNum = 0; // 线程数量
|
||
m_maxPoolSize = 0; // 最大线程池大小
|
||
m_maxStreamPoolSize = 0; // 最大流线程池大小
|
||
m_streamProcRatio = 0; // 流处理比例
|
||
}
|
||
|
||
ThreadPoolControler::~ThreadPoolControler()
|
||
{
|
||
delete m_scheduler; // 删除调度器对象
|
||
delete m_sessCtrl; // 删除会话控制对象
|
||
MemoryContextDelete(m_threadPoolContext); // 删除线程池上下文
|
||
m_threadPoolContext = NULL; // 将线程池上下文置为空
|
||
m_groups = NULL; // 将线程组指针置为空
|
||
m_sessCtrl = NULL; // 将会话控制指针置为空
|
||
}
|
||
|
||
void ThreadPoolControler::Init(bool enableNumaDistribute)
|
||
{
|
||
// 创建线程池上下文
|
||
m_threadPoolContext = AllocSetContextCreate(g_instance.instance_context,
|
||
"ThreadPoolContext",
|
||
ALLOCSET_DEFAULT_MINSIZE,
|
||
ALLOCSET_DEFAULT_INITSIZE,
|
||
ALLOCSET_DEFAULT_MAXSIZE,
|
||
SHARED_CONTEXT);
|
||
|
||
// 切换到线程池上下文
|
||
AutoContextSwitch memSwitch(m_threadPoolContext);
|
||
|
||
// 分配线程组数组
|
||
m_groups = (ThreadPoolGroup**)palloc(sizeof(ThreadPoolGroup*) * m_groupNum);
|
||
|
||
// 创建线程池会话控制对象
|
||
m_sessCtrl = New(CurrentMemoryContext) ThreadPoolSessControl(CurrentMemoryContext);
|
||
|
||
// 检查是否绑定 CPU
|
||
bool bindCpu = CheckCpuBind();
|
||
// 检查是否绑定 CPU NUMA
|
||
bool bindCpuNuma = CheckCpuNumaBind();
|
||
int maxThreadNum = 0;
|
||
int expectThreadNum = 0;
|
||
int maxStreamNum = 0;
|
||
int numaId = 0;
|
||
int cpuNum = 0;
|
||
int *cpuArr = NULL;
|
||
|
||
for (int i = 0; i < m_groupNum; i++) {
|
||
if (bindCpu) {
|
||
while (m_cpuInfo.cpuArrSize[numaId] == 0)
|
||
numaId++;
|
||
|
||
#ifdef __USE_NUMA
|
||
// 如果启用 NUMA 分配，则设置首选 NUMA 节点
|
||
if (enableNumaDistribute) {
|
||
numa_set_preferred(numaId);
|
||
}
|
||
#endif
|
||
|
||
// 获取当前 NUMA 节点的 CPU 数量
|
||
Assert(numaId < m_cpuInfo.totalNumaNum);
|
||
expectThreadNum = (int)round(
|
||
(double)m_threadNum * ((double)m_cpuInfo.cpuArrSize[numaId] / (double)m_cpuInfo.activeCpuNum));
|
||
maxThreadNum = (int)round(
|
||
(double)m_maxPoolSize * ((double)m_cpuInfo.cpuArrSize[numaId] / (double)m_cpuInfo.activeCpuNum));
|
||
maxStreamNum = (int)round(
|
||
(double)m_maxPoolSize * ((double)m_cpuInfo.cpuArrSize[numaId] / (double)m_cpuInfo.activeCpuNum));
|
||
cpuNum = m_cpuInfo.cpuArrSize[numaId];
|
||
cpuArr = m_cpuInfo.cpuArr[numaId];
|
||
|
||
// 增加 NUMA ID，以处理下一个 NUMA 节点
|
||
numaId++;
|
||
} else {
|
||
// 如果未绑定 CPU，则均匀分配线程和流的数量
|
||
expectThreadNum = m_threadNum / m_groupNum;
|
||
maxThreadNum = m_maxPoolSize / m_groupNum;
|
||
maxStreamNum = m_maxPoolSize / m_groupNum;
|
||
numaId = -1;
|
||
}
|
||
|
||
// 创建线程池组对象并初始化
|
||
m_groups[i] = New(CurrentMemoryContext)ThreadPoolGroup(maxThreadNum, expectThreadNum,
|
||
maxStreamNum, i, numaId, cpuNum, cpuArr, bindCpuNuma);
|
||
m_groups[i]->Init(enableNumaDistribute);
|
||
}
|
||
|
||
// 等待所有线程池组准备就绪
|
||
for (int i = 0; i < m_groupNum; i++) {
|
||
m_groups[i]->WaitReady();
|
||
}
|
||
|
||
#ifdef __USE_NUMA
|
||
if (enableNumaDistribute) {
|
||
/* 设置为交织模式以供除工作线程之外的其他线程使用 */
|
||
numa_set_interleave_mask(numa_all_nodes_ptr);
|
||
}
|
||
#endif
|
||
|
||
// 创建线程池调度器对象并启动
|
||
m_scheduler = New(CurrentMemoryContext) ThreadPoolScheduler(m_groupNum, m_groups);
|
||
m_scheduler->StartUp();
|
||
}
|
||
|
||
void ThreadPoolControler::SetThreadPoolInfo()
|
||
{
|
||
InitCpuInfo();
|
||
|
||
GetInstanceBind(&m_cpuset);
|
||
|
||
GetCpuAndNumaNum(&m_cpuInfo.totalCpuNum, &m_cpuInfo.totalNumaNum);
|
||
|
||
ParseAttr();
|
||
|
||
ParseStreamAttr();
|
||
|
||
GetSysCpuInfo();
|
||
|
||
SetGroupAndThreadNum();
|
||
|
||
SetStreamInfo();
|
||
}
|
||
|
||
void AdjustThreadAffinity(void) {
|
||
cpu_set_t m_cpuset;
|
||
CPU_ZERO(&m_cpuset);
|
||
|
||
// 检查实例是否已绑定到某些特定的 CPU
|
||
int ret = pthread_getaffinity_np(PostmasterPid, sizeof(cpu_set_t), &m_cpuset);
|
||
if (ret == 0) {
|
||
// 如果绑定到 CPU 0 而未绑定到 CPU 1，则重新设置 CPU 亲和性，以避免绑定到单一 CPU
|
||
if ((CPU_ISSET(0, &m_cpuset)) && (!CPU_ISSET(1, &m_cpuset))) {
|
||
int num_processors = sysconf(_SC_NPROCESSORS_CONF);
|
||
CPU_ZERO(&m_cpuset);
|
||
for (int j = 0; j < num_processors; j++) {
|
||
CPU_SET(j, &m_cpuset);
|
||
}
|
||
|
||
// 设置线程的 CPU 亲和性
|
||
int s = pthread_setaffinity_np(PostmasterPid, sizeof(cpu_set_t), &m_cpuset);
|
||
if (s != 0) {
|
||
ereport(WARNING, (errmsg("AdjustThreadAffinity fail to bind thread %lu, errno: %d", PostmasterPid, ret)));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
void ThreadPoolControler::GetInstanceBind(cpu_set_t *cpuset) {
|
||
// 修复 libgomp 在某些指定操作系统上的 bug
|
||
AdjustThreadAffinity();
|
||
|
||
// 检查实例是否已绑定到某些特定的 CPU
|
||
int ret = pthread_getaffinity_np(PostmasterPid, sizeof(cpu_set_t), cpuset);
|
||
if (ret == 0) {
|
||
return;
|
||
} else {
|
||
// 初始化 cpuset，以避免未绑定的情况
|
||
errno_t rc = memset_s(cpuset, sizeof(cpu_set_t), 0, sizeof(cpu_set_t));
|
||
securec_check(rc, "\0", "\0");
|
||
}
|
||
}
|
||
|
||
void ThreadPoolControler::ParseAttr()
|
||
{
|
||
m_attr.threadNum = DEFAULT_THREAD_POOL_SIZE;
|
||
m_attr.groupNum = DEFAULT_THREAD_POOL_GROUPS;
|
||
m_attr.bindCpu = NULL;
|
||
|
||
/* 复制字符串并移除空格 */
|
||
char* attr = TrimStr(g_instance.attr.attr_common.thread_pool_attr);
|
||
if (IS_NULL_STR(attr))
|
||
return;
|
||
|
||
char* ptoken = NULL;
|
||
char* psave = NULL;
|
||
const char* pdelimiter = ",";
|
||
|
||
/* 获取线程数 */
|
||
ptoken = TrimStr(strtok_r(attr, pdelimiter, &psave));
|
||
if (!IS_NULL_STR(ptoken))
|
||
m_attr.threadNum = pg_strtoint32(ptoken);
|
||
pfree_ext(ptoken);
|
||
|
||
/* 获取线程组数 */
|
||
ptoken = TrimStr(strtok_r(NULL, pdelimiter, &psave));
|
||
if (!IS_NULL_STR(ptoken))
|
||
m_attr.groupNum = pg_strtoint32(ptoken);
|
||
pfree_ext(ptoken);
|
||
|
||
if (m_attr.threadNum < 0 || m_attr.threadNum > MAX_THREAD_POOL_SIZE)
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("当前线程数 %d 超出范围 [%d, %d]。", m_attr.threadNum, 0, MAX_THREAD_POOL_SIZE));
|
||
if (m_attr.groupNum < 0 || m_attr.groupNum > MAX_THREAD_POOL_GROUPS)
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("当前线程组数 %d 超出范围 [%d, %d]。", m_attr.groupNum, 0, MAX_THREAD_POOL_GROUPS));
|
||
|
||
/* 获取绑定 CPU */
|
||
m_attr.bindCpu = TrimStr(psave);
|
||
ParseBindCpu();
|
||
|
||
pfree_ext(attr);
|
||
}
|
||
|
||
void ThreadPoolControler::ParseStreamAttr()
|
||
{
|
||
m_stream_attr.threadNum = DEFAULT_THREAD_POOL_SIZE;
|
||
m_stream_attr.procRatio = DEFAULT_THREAD_POOL_STREAM_PROC_RATIO;
|
||
m_stream_attr.groupNum = DEFAULT_THREAD_POOL_GROUPS;
|
||
m_stream_attr.bindCpu = NULL;
|
||
|
||
// 获取流式线程池属性配置字符串
|
||
char* attr = TrimStr(g_instance.attr.attr_common.thread_pool_stream_attr);
|
||
if (IS_NULL_STR(attr)) {
|
||
return;
|
||
}
|
||
|
||
char* ptoken = NULL;
|
||
char* psave = NULL;
|
||
const char* pdelimiter = ",";
|
||
|
||
// 解析流式线程池的最大线程数
|
||
ptoken = TrimStr(strtok_r(attr, pdelimiter, &psave));
|
||
if (IS_NULL_STR(ptoken) || !isdigit((unsigned char)*ptoken)) {
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("当前 thread_pool_stream_attr 格式错误，stream_thread_num 必须为数字。"));
|
||
}
|
||
m_stream_attr.threadNum = pg_strtoint32(ptoken);
|
||
pfree_ext(ptoken);
|
||
if (m_stream_attr.threadNum < 0 || m_stream_attr.threadNum > MAX_THREAD_POOL_SIZE) {
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("当前 stream_thread_num %d 超出范围 [%d, %d]。",
|
||
m_stream_attr.threadNum, 0, MAX_THREAD_POOL_SIZE));
|
||
}
|
||
|
||
// 解析流式线程池的处理比例
|
||
ptoken = TrimStr(strtok_r(NULL, pdelimiter, &psave));
|
||
if (IS_NULL_STR(ptoken) || !isdigit((unsigned char)*ptoken)) {
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("当前 thread_pool_stream_attr 格式错误，stream_proc_ratio 必须为数字。"));
|
||
}
|
||
m_stream_attr.procRatio = atof(ptoken);
|
||
pfree_ext(ptoken);
|
||
if (m_stream_attr.procRatio <= 0 || m_stream_attr.procRatio > MAX_THREAD_POOL_STREAM_PROC_RATIO) {
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("当前 stream_proc_ratio %f 超出范围 (%d, %d]。",
|
||
m_stream_attr.procRatio, 0, MAX_THREAD_POOL_STREAM_PROC_RATIO));
|
||
}
|
||
pfree_ext(attr);
|
||
|
||
return;
|
||
}
|
||
|
||
void ThreadPoolControler::ParseBindCpu()
|
||
{
|
||
if (IS_NULL_STR(m_attr.bindCpu)) {
|
||
m_cpuInfo.bindType = NO_CPU_BIND;
|
||
return;
|
||
}
|
||
|
||
// 复制属性字符串以进行解析
|
||
char* pattr = pstrdup(m_attr.bindCpu);
|
||
char* scpu = pattr;
|
||
char* ptoken = NULL;
|
||
char* psave = NULL;
|
||
const char* pdelimiter = ":";
|
||
int bindNum = 0;
|
||
|
||
// 检查属性字符串的格式是否正确
|
||
if (scpu[0] != '(' || scpu[strlen(scpu) - 1] != ')')
|
||
INVALID_ATTR_ERROR("使用 '(' ')' 来表示 CPU 绑定信息。");
|
||
|
||
scpu++;
|
||
scpu[strlen(scpu) - 1] = '\0';
|
||
|
||
// 解析属性字符串并转换为小写
|
||
ptoken = TrimStr(strtok_r(scpu, pdelimiter, &psave));
|
||
ptoken = pg_strtolower(ptoken);
|
||
|
||
// 根据不同的属性类型进行解析
|
||
if (strncmp("nobind", ptoken, strlen("nobind")) == 0) {
|
||
m_cpuInfo.bindType = NO_CPU_BIND;
|
||
return;
|
||
} else if (strncmp("allbind", ptoken, strlen("allbind")) == 0) {
|
||
m_cpuInfo.bindType = ALL_CPU_BIND;
|
||
return;
|
||
} else if (strncmp("cpubind", ptoken, strlen("cpubind")) == 0) {
|
||
m_cpuInfo.bindType = CPU_BIND;
|
||
m_cpuInfo.isBindCpuArr = (bool*)palloc0(sizeof(bool) * m_cpuInfo.totalCpuNum);
|
||
bindNum = ParseRangeStr(psave, m_cpuInfo.isBindCpuArr, m_cpuInfo.totalCpuNum, "cpubind");
|
||
} else if (strncmp("nodebind", ptoken, strlen("nodebind")) == 0) {
|
||
m_cpuInfo.bindType = NODE_BIND;
|
||
m_cpuInfo.isBindNumaArr = (bool*)palloc0(sizeof(bool) * m_cpuInfo.totalNumaNum);
|
||
bindNum = ParseRangeStr(psave, m_cpuInfo.isBindNumaArr, m_cpuInfo.totalNumaNum, "nodebind");
|
||
} else if (strncmp("numabind", ptoken, strlen("numabind")) == 0) {
|
||
m_cpuInfo.bindType = NUMA_BIND;
|
||
m_cpuInfo.isBindCpuNumaArr = (bool*)palloc0(sizeof(bool) * m_cpuInfo.totalCpuNum);
|
||
bindNum = ParseRangeStr(psave, m_cpuInfo.isBindCpuNumaArr, m_cpuInfo.totalCpuNum, "numabind");
|
||
} else {
|
||
INVALID_ATTR_ERROR(errdetail("只有 'nobind', 'allbind', 'cpubind', 'nodebind' 和 'numabind' "
|
||
"是有效的属性。"));
|
||
}
|
||
|
||
// 检查是否找到有效的 CPU 进行线程绑定
|
||
if (bindNum == 0)
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("无法找到有效的 CPU 进行线程绑定，可能的原因有两个：\n"
|
||
"1. 这些 CPU 不处于活动状态，请使用 lscpu 命令检查在线 CPU 列表。\n"
|
||
"2. 进程已绑定到其他 CPU，且没有交集，请使用 taskset -pc 命令检查进程 CPU 绑定信息。\n"));
|
||
pfree_ext(ptoken);
|
||
pfree_ext(pattr);
|
||
}
|
||
|
||
int ThreadPoolControler::ParseRangeStr(char* attr, bool* arr, int totalNum, char* bindtype)
|
||
{
|
||
char* ptoken = NULL;
|
||
char* psave = NULL;
|
||
const char* pdelimiter = ",";
|
||
int retNum = 0;
|
||
|
||
// 解析属性字符串
|
||
ptoken = TrimStr(strtok_r(attr, pdelimiter, &psave));
|
||
|
||
while (!IS_NULL_STR(ptoken)) {
|
||
char* pt = NULL;
|
||
char* ps = NULL;
|
||
const char* pd = "-";
|
||
int startid = -1;
|
||
int endid = -1;
|
||
|
||
// 解析范围字符串，可能包含起始和结束值
|
||
pt = TrimStr(strtok_r(ptoken, pd, &ps));
|
||
if (!IS_NULL_STR(pt))
|
||
startid = pg_strtoint32(pt);
|
||
if (!IS_NULL_STR(ps))
|
||
endid = pg_strtoint32(ps);
|
||
|
||
// 检查解析的值是否有效
|
||
if (startid < 0 && endid < 0)
|
||
INVALID_ATTR_ERROR(errdetail("无法解析属性 %s", pt));
|
||
if (startid >= totalNum)
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("属性 %s 中的 %d 超出了有效范围 [%d, %d]", bindtype, startid, 0, totalNum - 1));
|
||
if (endid >= totalNum)
|
||
INVALID_ATTR_ERROR(
|
||
errdetail("属性 %s 中的 %d 超出了有效范围 [%d, %d]", bindtype, endid, 0, totalNum - 1));
|
||
|
||
if (endid == -1) {
|
||
// 单个 CPU 绑定
|
||
retNum += arr[startid] ? 0 : 1;
|
||
arr[startid] = true;
|
||
} else {
|
||
// 范围内的多个 CPU 绑定
|
||
if (startid > endid) {
|
||
int tmpid = startid;
|
||
startid = endid;
|
||
endid = tmpid;
|
||
}
|
||
|
||
for (int i = startid; i <= endid; i++) {
|
||
retNum += arr[i] ? 0 : 1;
|
||
arr[i] = true;
|
||
}
|
||
}
|
||
|
||
// 释放临时字符串内存，不需要在错误发生时释放，这里的错误是致命的！
|
||
pfree_ext(pt);
|
||
pfree_ext(ptoken);
|
||
ptoken = TrimStr(strtok_r(NULL, pdelimiter, &psave));
|
||
}
|
||
|
||
return retNum;
|
||
}
|
||
|
||
bool* ThreadPoolControler::GetMcsCpuInfo(int totalCpuNum)
|
||
{
|
||
FILE* fp = NULL;
|
||
char buf[BUFSIZE];
|
||
|
||
bool* isMcsCpuArr = (bool*)palloc0(sizeof(bool) * totalCpuNum);
|
||
|
||
/*
|
||
* 当数据库部署在MCS上时，我们需要读取cpuset.cpus文件以查找此MCS中可用的CPU。
|
||
* 如果我们能够读取此文件，那么我们认为所有CPU都是可用的。
|
||
*/
|
||
fp = fopen("/sys/fs/cgroup/cpuset/cpuset.cpus", "r");
|
||
if (fp == NULL) {
|
||
ereport(WARNING, (errcode(ERRCODE_OPERATE_INVALID_PARAM),
|
||
errmsg("Failed to open file /sys/fs/cgroup/cpuset/cpuset.cpus")));
|
||
errno_t rc = memset_s(isMcsCpuArr, totalCpuNum, 1, totalCpuNum);
|
||
securec_check(rc, "", "");
|
||
return isMcsCpuArr;
|
||
}
|
||
|
||
if (fgets(buf, BUFSIZE, fp) == NULL) {
|
||
ereport(WARNING, (errcode(ERRCODE_OPERATE_INVALID_PARAM),
|
||
errmsg("Failed to read file /sys/fs/cgroup/cpuset/cpuset.cpus")));
|
||
errno_t rc = memset_s(isMcsCpuArr, totalCpuNum, 1, totalCpuNum);
|
||
securec_check(rc, "", "");
|
||
fclose(fp);
|
||
return isMcsCpuArr;
|
||
}
|
||
|
||
int mcsNum = ParseRangeStr(buf, isMcsCpuArr, totalCpuNum, "Mcs Cpu set");
|
||
if (mcsNum == 0) {
|
||
ereport(WARNING, (errcode(ERRCODE_OPERATE_INVALID_PARAM),
|
||
errmsg("No available CPUs in /sys/fs/cgroup/cpuset/cpuset.cpus")));
|
||
errno_t rc = memset_s(isMcsCpuArr, totalCpuNum, 1, totalCpuNum);
|
||
securec_check(rc, "", "");
|
||
}
|
||
fclose(fp);
|
||
return isMcsCpuArr;
|
||
}
|
||
|
||
void ThreadPoolControler::GetActiveCpu(NumaCpuId *numaCpuIdList, int *num)
|
||
{
|
||
*num = 0; // 初始化传出参数 num 为 0
|
||
char buf[BUFSIZE];
|
||
FILE* fp = popen("lscpu -b -e=cpu,node", "r"); // 打开一个用于读取 CPU 信息的流
|
||
|
||
if (fp == NULL) {
|
||
ereport(WARNING, (errmsg("Unable to use 'lscpu' to read CPU info."))); // 如果打开流失败，发出警告消息
|
||
return;
|
||
}
|
||
|
||
char* ptoken = NULL;
|
||
char* psave = NULL;
|
||
const char* pdelimiter = " ";
|
||
int cpuid = 0;
|
||
int numaid = 0;
|
||
|
||
// 尝试读取头部信息
|
||
if (fgets(buf, sizeof(buf), fp) != NULL) {
|
||
while (fgets(buf, sizeof(buf), fp) != NULL) {
|
||
ptoken = strtok_r(buf, pdelimiter, &psave);
|
||
if (!IS_NULL_STR(ptoken)) {
|
||
cpuid = pg_strtoint32(ptoken); // 解析 CPU ID
|
||
}
|
||
ptoken = strtok_r(NULL, pdelimiter, &psave);
|
||
if (!IS_NULL_STR(ptoken)) {
|
||
numaid = pg_strtoint32(ptoken); // 解析 NUMA ID
|
||
}
|
||
numaCpuIdList[*num].cpuId = cpuid; // 将 CPU ID 存入传出参数
|
||
numaCpuIdList[*num].numaId = numaid; // 将 NUMA ID 存入传出参数
|
||
(*num)++; // 递增传出参数 num
|
||
}
|
||
}
|
||
|
||
pclose(fp); // 关闭流
|
||
}
|
||
|
||
void ThreadPoolControler::GetSysCpuInfo()
|
||
{
|
||
if (m_cpuInfo.totalNumaNum == 0 || m_cpuInfo.totalCpuNum == 0) {
|
||
ereport(WARNING, (errmsg("Fail to read cpu num or numa num."))); // 如果没有读取到 CPU 数量或 NUMA 数量，发出警告消息
|
||
return;
|
||
}
|
||
|
||
m_cpuInfo.isMcsCpuArr = GetMcsCpuInfo(m_cpuInfo.totalCpuNum); // 获取 MCS CPU 信息
|
||
|
||
m_cpuInfo.cpuArr = (int**)palloc0(sizeof(int*) * m_cpuInfo.totalNumaNum);
|
||
m_cpuInfo.cpuArrSize = (int*)palloc0(sizeof(int) * m_cpuInfo.totalNumaNum);
|
||
int cpu_per_numa = m_cpuInfo.totalCpuNum / m_cpuInfo.totalNumaNum;
|
||
|
||
for (int i = 0; i < m_cpuInfo.totalNumaNum; i++) {
|
||
m_cpuInfo.cpuArr[i] = (int*)palloc0(sizeof(int) * cpu_per_numa); // 为每个 NUMA 节点分配内存
|
||
}
|
||
|
||
m_cpuInfo.activeCpuNum = 0;
|
||
NumaCpuId *sysNumaCpuIdList = (NumaCpuId*)palloc0(sizeof(NumaCpuId) * m_cpuInfo.totalCpuNum);
|
||
int sysNumaCpuIdNum = 0;
|
||
|
||
GetActiveCpu(sysNumaCpuIdList, &sysNumaCpuIdNum); // 获取激活的 CPU 信息
|
||
|
||
if (sysNumaCpuIdNum == 0) {
|
||
return;
|
||
}
|
||
|
||
for (int i = 0; i < sysNumaCpuIdNum; ++i) {
|
||
int cpuid = sysNumaCpuIdList[i].cpuId;
|
||
int numaid = sysNumaCpuIdList[i].numaId;
|
||
|
||
if (IsActiveCpu(cpuid, numaid)) {
|
||
m_cpuInfo.cpuArr[numaid][m_cpuInfo.cpuArrSize[numaid]] = cpuid; // 将 CPU ID 存入相应 NUMA 节点的数组中
|
||
m_cpuInfo.cpuArrSize[numaid]++; // 递增相应 NUMA 节点的数组大小
|
||
m_cpuInfo.activeCpuNum++; // 递增激活的 CPU 数量
|
||
}
|
||
}
|
||
|
||
pfree_ext(sysNumaCpuIdList); // 释放内存
|
||
|
||
for (int i = 0; i < m_cpuInfo.totalNumaNum; i++) {
|
||
if (m_cpuInfo.cpuArrSize[i] > 0)
|
||
m_cpuInfo.activeNumaNum++; // 统计激活的 NUMA 节点数量
|
||
}
|
||
}
|
||
|
||
void ThreadPoolControler::InitCpuInfo()
|
||
{
|
||
m_cpuInfo.totalCpuNum = 0; // 初始化总 CPU 数量为 0
|
||
m_cpuInfo.activeCpuNum = 0; // 初始化激活的 CPU 数量为 0
|
||
m_cpuInfo.totalNumaNum = 0; // 初始化总 NUMA 节点数量为 0
|
||
m_cpuInfo.activeNumaNum = 0; // 初始化激活的 NUMA 节点数量为 0
|
||
m_cpuInfo.cpuArrSize = NULL; // 初始化 CPU 数组大小为 NULL
|
||
m_cpuInfo.cpuArr = NULL; // 初始化 CPU 数组为 NULL
|
||
|
||
m_cpuInfo.bindType = NO_CPU_BIND; // 初始化 CPU 绑定类型为 NO_CPU_BIND
|
||
m_cpuInfo.isBindCpuArr = NULL; // 初始化 CPU 绑定数组为 NULL
|
||
m_cpuInfo.isBindNumaArr = NULL; // 初始化 NUMA 节点绑定数组为 NULL
|
||
m_cpuInfo.isMcsCpuArr = NULL; // 初始化 MCS CPU 数组为 NULL
|
||
}
|
||
|
||
void ThreadPoolControler::GetCpuAndNumaNum(int32 *totalCpuNum, int32 *totalNumaNum)
|
||
{
|
||
char buf[BUFSIZE];
|
||
FILE* fp = NULL;
|
||
|
||
// 打开 "lscpu" 命令的输出以获取 CPU 和 NUMA 节点数量信息
|
||
if ((fp = popen("LANG=en_US.UTF-8;lscpu", "r")) != NULL) {
|
||
while (fgets(buf, sizeof(buf), fp) != NULL) {
|
||
if (strncmp("CPU(s)", buf, strlen("CPU(s)")) == 0 &&
|
||
strncmp("On-line CPU(s) list", buf, strlen("On-line CPU(s) list")) != 0 &&
|
||
strncmp("NUMA node", buf, strlen("NUMA node")) != 0) {
|
||
// 当遇到包含 "CPU(s)" 的行时，解析并获取总 CPU 数量
|
||
char* loc = strchr(buf, ':');
|
||
*totalCpuNum = pg_strtoint32(loc + 1);
|
||
} else if (strncmp("NUMA node(s)", buf, strlen("NUMA node(s)")) == 0) {
|
||
// 当遇到包含 "NUMA node(s)" 的行时，解析并获取总 NUMA 节点数量
|
||
char* loc = strchr(buf, ':');
|
||
*totalNumaNum = pg_strtoint32(loc + 1);
|
||
}
|
||
}
|
||
pclose(fp); // 关闭文件流
|
||
}
|
||
}
|
||
|
||
bool ThreadPoolControler::IsActiveCpu(int cpuid, int numaid)
|
||
{
|
||
switch (m_cpuInfo.bindType) {
|
||
case NO_CPU_BIND:
|
||
case ALL_CPU_BIND:
|
||
// 如果未进行 CPU 绑定，或者进行了全局 CPU 绑定，检查 CPU 是否激活
|
||
return (m_cpuInfo.isMcsCpuArr[cpuid] && CPU_ISSET(cpuid, &m_cpuset));
|
||
case NODE_BIND:
|
||
// 如果进行了 NUMA 节点绑定，检查 CPU 和 NUMA 节点是否激活
|
||
return (m_cpuInfo.isBindNumaArr[numaid] && m_cpuInfo.isMcsCpuArr[cpuid] && CPU_ISSET(cpuid, &m_cpuset));
|
||
case CPU_BIND:
|
||
// 如果进行了 CPU 绑定，检查 CPU 是否激活
|
||
return (m_cpuInfo.isBindCpuArr[cpuid] && m_cpuInfo.isMcsCpuArr[cpuid] && CPU_ISSET(cpuid, &m_cpuset));
|
||
case NUMA_BIND:
|
||
// 如果进行了 NUMA 节点和 CPU 绑定，检查 CPU 和 NUMA 节点是否激活
|
||
return (m_cpuInfo.isBindCpuNumaArr[cpuid] && m_cpuInfo.isMcsCpuArr[cpuid] && CPU_ISSET(cpuid, &m_cpuset));
|
||
}
|
||
return false; // 默认情况下，返回 false
|
||
}
|
||
|
||
bool ThreadPoolControler::CheckCpuBind() const
|
||
{
|
||
if (m_cpuInfo.bindType == NO_CPU_BIND)
|
||
return false; // 如果没有进行 CPU 绑定，则返回 false
|
||
|
||
if (m_groupNum != m_cpuInfo.activeNumaNum) {
|
||
ereport(WARNING,
|
||
(errmsg("无法将工作线程绑定到 CPU，因为线程组数必须等于激活的 NUMA 节点数。")));
|
||
return false; // 如果线程组数不等于激活的 NUMA 节点数，返回 false
|
||
}
|
||
|
||
if (m_cpuInfo.activeCpuNum == 0 || m_cpuInfo.cpuArr == NULL) {
|
||
ereport(WARNING, (errmsg("无法将工作线程绑定到 CPU，因为没有有效的 CPU。")));
|
||
return false; // 如果没有有效的 CPU，返回 false
|
||
}
|
||
|
||
return true; // 其他情况下，返回 true
|
||
}
|
||
|
||
bool ThreadPoolControler::CheckCpuNumaBind() const
|
||
{
|
||
return m_cpuInfo.bindType == NUMA_BIND; // 如果进行了 NUMA 绑定，返回 true，否则返回 false
|
||
}
|
||
|
||
bool ThreadPoolControler::CheckNumaDistribute(int numaNodeNum) const
|
||
{
|
||
if (m_cpuInfo.bindType == NO_CPU_BIND) {
|
||
ereport(WARNING,
|
||
(errmsg("在激活 NUMA 时，应使用 allbind 来替代 nobind 在 thread_pool_attr 中。")));
|
||
return false; // 如果未进行 CPU 绑定，给出警告并返回 false
|
||
}
|
||
|
||
if (!CheckCpuBind()) {
|
||
return false; // 如果 CPU 绑定检查失败，返回 false
|
||
}
|
||
|
||
if (m_cpuInfo.totalNumaNum != numaNodeNum || !m_cpuInfo.cpuArrSize) {
|
||
ereport(WARNING,
|
||
(errmsg("无法激活 NUMA 分布，因为没有多个 NUMA 节点或可用的 CPU。")));
|
||
return false; // 如果 NUMA 节点数不等于给定的 numaNodeNum，或者 cpuArrSize 为空，返回 false
|
||
}
|
||
|
||
for (int i = 0; i < m_cpuInfo.totalNumaNum; ++i) {
|
||
if (m_cpuInfo.cpuArrSize[i] <= 0) {
|
||
ereport(WARNING,
|
||
(errmsg("无法激活 NUMA 分布，因为节点 %d 中没有可用的 CPU。", i)));
|
||
return false; // 如果某个 NUMA 节点中没有可用的 CPU，返回 false
|
||
}
|
||
}
|
||
|
||
return true; // 其他情况下，返回 true
|
||
}
|
||
|
||
CPUBindType ThreadPoolControler::GetCpuBindType() const
|
||
{
|
||
return m_cpuInfo.bindType;
|
||
}
|
||
|
||
void ThreadPoolControler::SetStreamInfo()
|
||
{
|
||
m_streamProcRatio = m_stream_attr.procRatio;
|
||
m_maxStreamPoolSize = Min(m_stream_attr.threadNum, m_threadNum);
|
||
}
|
||
|
||
void ThreadPoolControler::SetGroupAndThreadNum()
|
||
{
|
||
if (m_attr.groupNum == 0) {
|
||
if (m_cpuInfo.totalNumaNum > 0)
|
||
m_groupNum = m_cpuInfo.activeNumaNum; // 如果未指定线程组数且存在 NUMA 节点，则使用激活的 NUMA 节点数
|
||
else
|
||
m_groupNum = DEFAULT_THREAD_POOL_GROUPS; // 否则使用默认的线程组数
|
||
} else {
|
||
m_groupNum = m_attr.groupNum; // 如果指定了线程组数，则使用指定的线程组数
|
||
}
|
||
|
||
if (m_attr.threadNum == 0) {
|
||
if (m_cpuInfo.activeCpuNum > 0)
|
||
m_threadNum = m_cpuInfo.activeCpuNum * THREAD_CORE_RATIO; // 如果未指定线程数且存在激活的 CPU，则计算线程数
|
||
else
|
||
m_threadNum = DEFAULT_THREAD_POOL_SIZE; // 否则使用默认的线程数
|
||
} else {
|
||
m_threadNum = m_attr.threadNum; // 如果指定了线程数，则使用指定的线程数
|
||
}
|
||
|
||
ConstrainThreadNum(); // 调用 ConstrainThreadNum 方法进行线程数约束
|
||
}
|
||
|
||
void ThreadPoolControler::ConstrainThreadNum()
|
||
{
|
||
/* 线程池大小不应超过 max_connections。 */
|
||
if (MAX_THREAD_POOL_SIZE > g_instance.attr.attr_network.MaxConnections) {
|
||
ereport(LOG, (errcode(ERRCODE_OPERATE_INVALID_PARAM),
|
||
errmsg("最大线程池大小 %d 不应超过 max_connections %d，因此将最大线程池大小减小到 max_connections",
|
||
MAX_THREAD_POOL_SIZE, g_instance.attr.attr_network.MaxConnections)));
|
||
}
|
||
|
||
m_maxPoolSize = Min(MAX_THREAD_POOL_SIZE, g_instance.attr.attr_network.MaxConnections); // 最大线程池大小受限于 max_connections
|
||
m_threadNum = Min(m_threadNum, m_maxPoolSize); // 线程数不应超过最大线程池大小
|
||
}
|
||
|
||
int ThreadPoolControler::GetThreadNum()
|
||
{
|
||
return m_maxPoolSize; // 返回最大线程池大小
|
||
}
|
||
|
||
ThreadPoolStat* ThreadPoolControler::GetThreadPoolStat(uint32* num)
|
||
{
|
||
ThreadPoolStat* result = (ThreadPoolStat*)palloc(m_groupNum * sizeof(ThreadPoolStat)); // 分配存储线程池统计信息的内存
|
||
int i;
|
||
|
||
for (i = 0; i < m_groupNum; i++) {
|
||
m_groups[i]->GetThreadPoolGroupStat(&result[i]); // 获取每个线程组的统计信息
|
||
}
|
||
|
||
*num = m_groupNum; // 返回线程组数量
|
||
return result; // 返回线程池统计信息数组
|
||
}
|
||
|
||
void ThreadPoolControler::CloseAllSessions()
|
||
{
|
||
ereport(LOG, (errmodule(MOD_THREAD_POOL),
|
||
errmsg("pmState:%d, start to close all sessions in threadpool.", pmState)));
|
||
|
||
m_sessCtrl->MarkAllSessionClose(); // 标记所有会话为关闭状态
|
||
(void)SignalCancelAllBackEnd(); // 发送取消信号以取消所有后端任务
|
||
|
||
for (int i = 0; i < m_groupNum; i++) {
|
||
m_groups[i]->GetListener()->SendShutDown(); // 发送关闭信号给所有监听器
|
||
}
|
||
|
||
/* 检查直到所有组都关闭了它们的会话。 */
|
||
bool allclose = false;
|
||
while (!allclose) {
|
||
if (m_sessCtrl->IsActiveListEmpty()) {
|
||
break;
|
||
}
|
||
|
||
allclose = true;
|
||
for (int i = 0; i < m_groupNum; i++) {
|
||
allclose = (m_groups[i]->AllSessionClosed() && allclose); // 检查每个组的会话是否都已关闭
|
||
}
|
||
pg_usleep(one_hundred_micro_sec); // 短暂休眠以减少 CPU 使用
|
||
}
|
||
|
||
ereport(LOG, (errmodule(MOD_THREAD_POOL),
|
||
errmsg("pmState:%d, all threadpool sessions already closed.", pmState)));
|
||
}
|
||
|
||
void ThreadPoolControler::ShutDownThreads(bool forceWait)
|
||
{
|
||
for (int i = 0; i < m_groupNum; i++) {
|
||
m_groups[i]->ShutDownThreads(); // 关闭所有线程组中的线程
|
||
}
|
||
|
||
ereport(LOG, (errmodule(MOD_THREAD_POOL),
|
||
errmsg("pmState:%d, shut down all threadpool threads.", pmState)));
|
||
|
||
if (forceWait) {
|
||
/* 检查直到所有组都关闭了它们的工作线程。 */
|
||
bool allshut = false;
|
||
while (!allshut) {
|
||
allshut = true;
|
||
for (int i = 0; i < m_groupNum; i++) {
|
||
allshut = (m_groups[i]->AllThreadShutDown() && allshut); // 检查每个组的线程是否都已关闭
|
||
}
|
||
pg_usleep(one_hundred_micro_sec); // 短暂休眠以减少 CPU 使用
|
||
}
|
||
|
||
ereport(LOG, (errmodule(MOD_THREAD_POOL),
|
||
errmsg("pmState:%d, all threadpool threads already shut down.", pmState)));
|
||
}
|
||
}
|
||
|
||
void ThreadPoolControler::ShutDownListeners(bool forceWait)
|
||
{
|
||
for (int i = 0; i < m_groupNum; i++) {
|
||
m_groups[i]->GetListener()->ShutDown(); // 关闭所有线程组的监听器
|
||
}
|
||
if (forceWait) {
|
||
bool allshut = false;
|
||
while (!allshut) {
|
||
allshut = true;
|
||
for (int i = 0; i < m_groupNum; i++) {
|
||
allshut = (m_groups[i]->GetListener()->GetThreadId() == 0) && allshut; // 检查监听器线程是否都已关闭
|
||
}
|
||
pg_usleep(one_hundred_micro_sec); // 短暂休眠以减少 CPU 使用
|
||
}
|
||
}
|
||
}
|
||
|
||
void ThreadPoolControler::ShutDownScheduler(bool forceWait, bool noAdjust)
|
||
{
|
||
if (noAdjust) {
|
||
pg_memory_barrier();
|
||
m_scheduler->m_canAdjustPool = false;
|
||
}
|
||
|
||
m_scheduler->ShutDown();
|
||
if (forceWait) {
|
||
bool allshut = false;
|
||
while (!allshut) {
|
||
allshut = m_scheduler->HasShutDown();
|
||
pg_usleep(one_hundred_micro_sec);
|
||
}
|
||
}
|
||
}
|
||
|
||
void ThreadPoolControler::EnableAdjustPool()
|
||
{
|
||
pg_memory_barrier();
|
||
m_scheduler->m_canAdjustPool = true;
|
||
}
|
||
|
||
void ThreadPoolControler::AddWorkerIfNecessary()
|
||
{
|
||
for (int i = 0; i < m_groupNum; i++)
|
||
m_groups[i]->AddWorkerIfNecessary();
|
||
if (m_scheduler->HasShutDown()) {
|
||
m_scheduler->StartUp();
|
||
m_scheduler->SetShutDown(false);
|
||
}
|
||
EnableAdjustPool();
|
||
}
|
||
|
||
ThreadPoolGroup* ThreadPoolControler::FindThreadGroupWithLeastSession()
|
||
{
|
||
int idx = 0; // 用于记录具有最少会话的线程组的索引
|
||
float4 least_session = 0.0; // 用于记录最少会话数
|
||
float4 session_per_thread = 0.0; // 用于记录每个线程的平均会话数
|
||
|
||
least_session = m_groups[0]->GetSessionPerThread(); // 获取第一个线程组的平均会话数作为初始值
|
||
for (int i = 1; i < m_groupNum; i++) {
|
||
session_per_thread = m_groups[i]->GetSessionPerThread(); // 获取当前线程组的平均会话数
|
||
if (session_per_thread < least_session) { // 如果当前线程组的平均会话数更小
|
||
least_session = session_per_thread; // 更新最少会话数
|
||
idx = i; // 更新具有最少会话的线程组的索引
|
||
}
|
||
}
|
||
|
||
return m_groups[idx]; // 返回具有最少会话的线程组的指针
|
||
}
|
||
|
||
bool ThreadPoolControler::StayInAttachMode()
|
||
{
|
||
return m_sessCtrl->GetActiveSessionCount() < m_threadNum;
|
||
}
|
||
|
||
int ThreadPoolControler::DispatchSession(Port* port)
|
||
{
|
||
ThreadPoolGroup* grp = NULL;
|
||
knl_session_context* sc = NULL;
|
||
|
||
/*
|
||
* 在comm_proxy模式下，每个接受的文件描述符（fd）都与一个
|
||
* 固定的通信线程合并在一个NUMA组中，我们不再使用旧的方法“
|
||
* 与最新会话的组合”进行分配，因此只需返回通信线程所在的NUMA组即可。
|
||
*
|
||
* 注意：我们假设每个连接的用户会话可以等可能地分布给通信器，
|
||
* 幸运的是，欧拉操作系统似乎可以保证这一点（已经证明），否则我们还需要重新考虑这个问题。
|
||
*
|
||
* 性能优化：在使用`comm_proxy`时，当`thread_pool`的`m_groupNum`与`comm_proxy`的NUMA组数相同时，可以实现性能优化。
|
||
*/
|
||
if (AmIProxyModeSockfd(port->sock) && m_groupNum == g_comm_proxy_config.s_numa_num) {
|
||
CommSockDesc* comm_sock = g_comm_controller->FdGetCommSockDesc(port->sock);
|
||
grp = m_groups[comm_sock->m_group_id];
|
||
} else {
|
||
grp = FindThreadGroupWithLeastSession();
|
||
}
|
||
|
||
if (grp == NULL) {
|
||
Assert(false);
|
||
return STATUS_ERROR;
|
||
}
|
||
/* 如果这个组挂起了，我们不会接受新的会话。 */
|
||
if (grp->IsGroupHanged()) {
|
||
ereport(WARNING,
|
||
(errmodule(MOD_THREAD_POOL),
|
||
errmsg("Group[%d] is too busy to add new session for now.", grp->GetGroupId())));
|
||
return STATUS_ERROR;
|
||
}
|
||
|
||
sc = m_sessCtrl->CreateSession(port);
|
||
if (sc == NULL)
|
||
return STATUS_ERROR;
|
||
|
||
grp->GetListener()->AddNewSession(sc);
|
||
return STATUS_OK;
|
||
}
|
||
|
||
/*
|
||
* 将指定的线程绑定到所有可用的CPU。
|
||
* 此操作由辅助线程（例如WAL发送者）调用。
|
||
*/
|
||
void ThreadPoolControler::BindThreadToAllAvailCpu(ThreadId thread) const
|
||
{
|
||
// 如果不需要绑定CPU，直接返回
|
||
if (!CheckCpuBind()) {
|
||
return;
|
||
}
|
||
|
||
// 如果绑定方式是ALL_CPU_BIND，也直接返回
|
||
if (m_cpuInfo.bindType == ALL_CPU_BIND) {
|
||
return;
|
||
}
|
||
|
||
// 创建一个CPU集合，初始化为空
|
||
cpu_set_t availCpuSet;
|
||
CPU_ZERO(&availCpuSet);
|
||
|
||
// 遍历每个NUMA节点
|
||
for (int numaNo = 0; numaNo < m_cpuInfo.totalNumaNum; ++numaNo) {
|
||
int cpuNumber = m_cpuInfo.cpuArrSize[numaNo];
|
||
|
||
// 将每个NUMA节点上的CPU添加到CPU集合中
|
||
for (int i = 0; i < cpuNumber; ++i) {
|
||
CPU_SET(m_cpuInfo.cpuArr[numaNo][i], &availCpuSet);
|
||
}
|
||
}
|
||
|
||
// 使用pthread_setaffinity_np函数将线程绑定到CPU集合中
|
||
int ret = pthread_setaffinity_np(thread, sizeof(cpu_set_t), &availCpuSet);
|
||
|
||
// 如果绑定失败，输出警告信息
|
||
if (ret != 0)
|
||
ereport(WARNING, (errmsg("BindThreadToAllAvailCpu fail to bind thread %lu, errno: %d", thread, ret)));
|
||
}
|