From a4a404f207c9968cfe7eeb04fcdd53d90933bddc Mon Sep 17 00:00:00 2001 From: Nemoo <18210033860@163.com> Date: Sat, 30 Sep 2023 14:18:56 +0800 Subject: [PATCH] Update threadpool_controler.cpp --- .../threadpool/threadpool_controler.cpp | 369 ++++++++++-------- 1 file changed, 202 insertions(+), 167 deletions(-) diff --git a/src/gausskernel/process/threadpool/threadpool_controler.cpp b/src/gausskernel/process/threadpool/threadpool_controler.cpp index 63c13d4a9..dedb9f47a 100644 --- a/src/gausskernel/process/threadpool/threadpool_controler.cpp +++ b/src/gausskernel/process/threadpool/threadpool_controler.cpp @@ -7,16 +7,12 @@ * * http://license.coscl.org.cn/MulanPSL2 * - * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, - * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, - * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. - * See the Mulan PSL v2 for more details. * ------------------------------------------------------------------------- * * threadpool_controler.cpp - * Controler for thread pool. Class ThreadPoolControler is defined to - * initilize thread pool's worker and listener threads, and dispatch - * new session from postmaster thread to suitable thread group. + * 线程池的控制器。定义了 ThreadPoolControler 类， + * 用于初始化线程池的工作线程和监听线程，并将新会话从 + * 主线程分派到合适的线程组。 * * * IDENTIFICATION @@ -71,32 +67,33 @@ ThreadPoolControler* g_threadPoolControler = NULL; 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; + 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; + 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, @@ -104,12 +101,18 @@ void ThreadPoolControler::Init(bool enableNumaDistribute) 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; @@ -123,16 +126,14 @@ void ThreadPoolControler::Init(bool enableNumaDistribute) while (m_cpuInfo.cpuArrSize[numaId] == 0) numaId++; - /* - * Invoke numa_set_preferred before starting worker thread to make - * more memory allocation local to worker thread. - */ #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)); @@ -143,30 +144,35 @@ void ThreadPoolControler::Init(bool enableNumaDistribute) 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) { - /* Set to interleave mode for other than worker thread */ + /* 设置为交织模式以供除工作线程之外的其他线程使用 */ numa_set_interleave_mask(numa_all_nodes_ptr); } #endif + // 创建线程池调度器对象并启动 m_scheduler = New(CurrentMemoryContext) ThreadPoolScheduler(m_groupNum, m_groups); m_scheduler->StartUp(); } @@ -194,9 +200,10 @@ void AdjustThreadAffinity(void) { cpu_set_t m_cpuset; CPU_ZERO(&m_cpuset); - /* Check if the instance has been attch to some specific CPUs. */ + // 检查实例是否已绑定到某些特定的 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); @@ -204,7 +211,7 @@ void AdjustThreadAffinity(void) { CPU_SET(j, &m_cpuset); } - // set CPU affinity of a thread + // 设置线程的 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))); @@ -213,17 +220,16 @@ void AdjustThreadAffinity(void) { } } - -void ThreadPoolControler::GetInstanceBind(cpu_set_t *cpuset) -{ - /* this function is used to avoid the libgomp bug on some specified OS */ +void ThreadPoolControler::GetInstanceBind(cpu_set_t *cpuset) { + // 修复 libgomp 在某些指定操作系统上的 bug AdjustThreadAffinity(); - /* Check if the instance has been attch to some specific CPUs. */ + // 检查实例是否已绑定到某些特定的 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"); } @@ -235,7 +241,7 @@ void ThreadPoolControler::ParseAttr() m_attr.groupNum = DEFAULT_THREAD_POOL_GROUPS; m_attr.bindCpu = NULL; - /* Do str copy and remove space. */ + /* 复制字符串并移除空格 */ char* attr = TrimStr(g_instance.attr.attr_common.thread_pool_attr); if (IS_NULL_STR(attr)) return; @@ -244,13 +250,13 @@ void ThreadPoolControler::ParseAttr() char* psave = NULL; const char* pdelimiter = ","; - /* Get thread num */ + /* 获取线程数 */ ptoken = TrimStr(strtok_r(attr, pdelimiter, &psave)); if (!IS_NULL_STR(ptoken)) m_attr.threadNum = pg_strtoint32(ptoken); pfree_ext(ptoken); - /* Ger group num */ + /* 获取线程组数 */ ptoken = TrimStr(strtok_r(NULL, pdelimiter, &psave)); if (!IS_NULL_STR(ptoken)) m_attr.groupNum = pg_strtoint32(ptoken); @@ -258,15 +264,15 @@ void ThreadPoolControler::ParseAttr() if (m_attr.threadNum < 0 || m_attr.threadNum > MAX_THREAD_POOL_SIZE) INVALID_ATTR_ERROR( - errdetail("Current thread num %d is out of range [%d, %d].", m_attr.threadNum, 0, MAX_THREAD_POOL_SIZE)); + 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("Current group num %d is out of range [%d, %d].", m_attr.groupNum, 0, MAX_THREAD_POOL_GROUPS)); + errdetail("当前线程组数 %d 超出范围 [%d, %d]。", m_attr.groupNum, 0, MAX_THREAD_POOL_GROUPS)); - /* Get attach cpu */ + /* 获取绑定 CPU */ m_attr.bindCpu = TrimStr(psave); ParseBindCpu(); - + pfree_ext(attr); } @@ -276,7 +282,8 @@ void ThreadPoolControler::ParseStreamAttr() 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; @@ -285,36 +292,36 @@ void ThreadPoolControler::ParseStreamAttr() char* ptoken = NULL; char* psave = NULL; const char* pdelimiter = ","; - - /* Get stream_thread_pool_stream max thread num */ + + // 解析流式线程池的最大线程数 ptoken = TrimStr(strtok_r(attr, pdelimiter, &psave)); if (IS_NULL_STR(ptoken) || !isdigit((unsigned char)*ptoken)) { INVALID_ATTR_ERROR( - errdetail("Current thread_pool_stream_attr format is error, stream_thread_num must be digital.")); + 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("Current stream_thread_num %d is out of range [%d, %d].", + errdetail("当前 stream_thread_num %d 超出范围 [%d, %d]。", m_stream_attr.threadNum, 0, MAX_THREAD_POOL_SIZE)); } - - /* Get proc ratio of stream threads */ + + // 解析流式线程池的处理比例 ptoken = TrimStr(strtok_r(NULL, pdelimiter, &psave)); if (IS_NULL_STR(ptoken) || !isdigit((unsigned char)*ptoken)) { INVALID_ATTR_ERROR( - errdetail("Current thread_pool_stream_attr format is error, stream_proc_ratio must be digital.")); + 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("Current stream_proc_ratio %f is out of range (%d, %d].", + errdetail("当前 stream_proc_ratio %f 超出范围 (%d, %d]。", m_stream_attr.procRatio, 0, MAX_THREAD_POOL_STREAM_PROC_RATIO)); } pfree_ext(attr); - + return; } @@ -325,6 +332,7 @@ void ThreadPoolControler::ParseBindCpu() return; } + // 复制属性字符串以进行解析 char* pattr = pstrdup(m_attr.bindCpu); char* scpu = pattr; char* ptoken = NULL; @@ -332,15 +340,18 @@ void ThreadPoolControler::ParseBindCpu() const char* pdelimiter = ":"; int bindNum = 0; + // 检查属性字符串的格式是否正确 if (scpu[0] != '(' || scpu[strlen(scpu) - 1] != ')') - INVALID_ATTR_ERROR("Use '(' ')' to indicate cpu bind info."); + 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; @@ -360,16 +371,16 @@ void ThreadPoolControler::ParseBindCpu() 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("Only 'nobind', 'allbind', 'cpubind', 'nodebind' and 'numabind' " - "are valid attribute.")); + INVALID_ATTR_ERROR(errdetail("只有 'nobind', 'allbind', 'cpubind', 'nodebind' 和 'numabind' " + "是有效的属性。")); } + // 检查是否找到有效的 CPU 进行线程绑定 if (bindNum == 0) INVALID_ATTR_ERROR( - errdetail("Can not find valid CPU for thread binding, there are two possible reasons:\n" - "1. These CPUs are not active, use lscpu to check On-line CPU(s) list.\n" - "2. The process has been bind to other CPUs and there is no intersection," - "use taskset -pc to check process CPU bind info.\n")); + errdetail("无法找到有效的 CPU 进行线程绑定，可能的原因有两个：\n" + "1. 这些 CPU 不处于活动状态，请使用 lscpu 命令检查在线 CPU 列表。\n" + "2. 进程已绑定到其他 CPU，且没有交集，请使用 taskset -pc 命令检查进程 CPU 绑定信息。\n")); pfree_ext(ptoken); pfree_ext(pattr); } @@ -381,6 +392,7 @@ int ThreadPoolControler::ParseRangeStr(char* attr, bool* arr, int totalNum, char const char* pdelimiter = ","; int retNum = 0; + // 解析属性字符串 ptoken = TrimStr(strtok_r(attr, pdelimiter, &psave)); while (!IS_NULL_STR(ptoken)) { @@ -390,25 +402,29 @@ int ThreadPoolControler::ParseRangeStr(char* attr, bool* arr, int totalNum, char 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("Can not parse attribute %s", pt)); + INVALID_ATTR_ERROR(errdetail("无法解析属性 %s", pt)); if (startid >= totalNum) INVALID_ATTR_ERROR( - errdetail("The %s attribute %d is out of valid range [%d, %d]", bindtype, startid, 0, totalNum - 1)); + errdetail("属性 %s 中的 %d 超出了有效范围 [%d, %d]", bindtype, startid, 0, totalNum - 1)); if (endid >= totalNum) INVALID_ATTR_ERROR( - errdetail("The %s attribute %d is out of valid range [%d, %d]", bindtype, endid, 0, totalNum - 1)); + 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; @@ -421,7 +437,7 @@ int ThreadPoolControler::ParseRangeStr(char* attr, bool* arr, int totalNum, char } } - /* Don't need to free when error ocurrs, errors here are FATAL level! */ + // 释放临时字符串内存，不需要在错误发生时释放，这里的错误是致命的！ pfree_ext(pt); pfree_ext(ptoken); ptoken = TrimStr(strtok_r(NULL, pdelimiter, &psave)); @@ -438,9 +454,8 @@ bool* ThreadPoolControler::GetMcsCpuInfo(int totalCpuNum) bool* isMcsCpuArr = (bool*)palloc0(sizeof(bool) * totalCpuNum); /* - * When the database is deplyed on MCS, we need to read cpuset.cpus to find - * available CPUs in this MCS. If we can read this file, then we think all - * CPUs are available. + * 当数据库部署在MCS上时，我们需要读取cpuset.cpus文件以查找此MCS中可用的CPU。 + * 如果我们能够读取此文件，那么我们认为所有CPU都是可用的。 */ fp = fopen("/sys/fs/cgroup/cpuset/cpuset.cpus", "r"); if (fp == NULL) { @@ -473,11 +488,12 @@ bool* ThreadPoolControler::GetMcsCpuInfo(int totalCpuNum) void ThreadPoolControler::GetActiveCpu(NumaCpuId *numaCpuIdList, int *num) { - *num = 0; + *num = 0; // 初始化传出参数 num 为 0 char buf[BUFSIZE]; - FILE* fp = popen("lscpu -b -e=cpu,node", "r"); + FILE* fp = popen("lscpu -b -e=cpu,node", "r"); // 打开一个用于读取 CPU 信息的流 + if (fp == NULL) { - ereport(WARNING, (errmsg("Unable to use 'lscpu' to read CPU info."))); + ereport(WARNING, (errmsg("Unable to use 'lscpu' to read CPU info."))); // 如果打开流失败，发出警告消息 return; } @@ -486,45 +502,49 @@ void ThreadPoolControler::GetActiveCpu(NumaCpuId *numaCpuIdList, int *num) const char* pdelimiter = " "; int cpuid = 0; int numaid = 0; - /* try to read the header. */ + + // 尝试读取头部信息 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); + cpuid = pg_strtoint32(ptoken); // 解析 CPU ID } ptoken = strtok_r(NULL, pdelimiter, &psave); if (!IS_NULL_STR(ptoken)) { - numaid = pg_strtoint32(ptoken); + numaid = pg_strtoint32(ptoken); // 解析 NUMA ID } - numaCpuIdList[*num].cpuId = cpuid; - numaCpuIdList[*num].numaId = numaid; - (*num)++; + numaCpuIdList[*num].cpuId = cpuid; // 将 CPU ID 存入传出参数 + numaCpuIdList[*num].numaId = numaid; // 将 NUMA ID 存入传出参数 + (*num)++; // 递增传出参数 num } } - pclose(fp); + 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."))); + ereport(WARNING, (errmsg("Fail to read cpu num or numa num."))); // 如果没有读取到 CPU 数量或 NUMA 数量，发出警告消息 return; } - m_cpuInfo.isMcsCpuArr = GetMcsCpuInfo(m_cpuInfo.totalCpuNum); + 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); + 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); + + GetActiveCpu(sysNumaCpuIdList, &sysNumaCpuIdNum); // 获取激活的 CPU 信息 if (sysNumaCpuIdNum == 0) { return; @@ -533,56 +553,58 @@ void ThreadPoolControler::GetSysCpuInfo() 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; - m_cpuInfo.cpuArrSize[numaid]++; - m_cpuInfo.activeCpuNum++; + m_cpuInfo.cpuArr[numaid][m_cpuInfo.cpuArrSize[numaid]] = cpuid; // 将 CPU ID 存入相应 NUMA 节点的数组中 + m_cpuInfo.cpuArrSize[numaid]++; // 递增相应 NUMA 节点的数组大小 + m_cpuInfo.activeCpuNum++; // 递增激活的 CPU 数量 } - } - pfree_ext(sysNumaCpuIdList); + pfree_ext(sysNumaCpuIdList); // 释放内存 for (int i = 0; i < m_cpuInfo.totalNumaNum; i++) { if (m_cpuInfo.cpuArrSize[i] > 0) - m_cpuInfo.activeNumaNum++; + m_cpuInfo.activeNumaNum++; // 统计激活的 NUMA 节点数量 } } void ThreadPoolControler::InitCpuInfo() { - m_cpuInfo.totalCpuNum = 0; - m_cpuInfo.activeCpuNum = 0; - m_cpuInfo.totalNumaNum = 0; - m_cpuInfo.activeNumaNum = 0; - m_cpuInfo.cpuArrSize = NULL; - m_cpuInfo.cpuArr = NULL; + 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; - m_cpuInfo.isBindCpuArr = NULL; - m_cpuInfo.isBindNumaArr = NULL; - m_cpuInfo.isMcsCpuArr = 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); + pclose(fp); // 关闭文件流 } } @@ -591,67 +613,72 @@ 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; + return false; // 默认情况下，返回 false } bool ThreadPoolControler::CheckCpuBind() const { if (m_cpuInfo.bindType == NO_CPU_BIND) - return false; + return false; // 如果没有进行 CPU 绑定，则返回 false + if (m_groupNum != m_cpuInfo.activeNumaNum) { ereport(WARNING, - (errmsg("Can not bind worker thread to CPU because the " - "thread group num must equal to active NUMA num."))); - return false; - } - if (m_cpuInfo.activeCpuNum == 0 || m_cpuInfo.cpuArr == NULL) { - ereport(WARNING, (errmsg("Can not bind worker thread to CPU because no valid CPUs."))); - return false; + (errmsg("无法将工作线程绑定到 CPU，因为线程组数必须等于激活的 NUMA 节点数。"))); + return false; // 如果线程组数不等于激活的 NUMA 节点数，返回 false } - return true; + 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; + 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("allbind should be used to replace nobind in thread_pool_attr when NUMA is activated."))); - return false; + (errmsg("在激活 NUMA 时，应使用 allbind 来替代 nobind 在 thread_pool_attr 中。"))); + return false; // 如果未进行 CPU 绑定，给出警告并返回 false } if (!CheckCpuBind()) { - return false; + return false; // 如果 CPU 绑定检查失败，返回 false } if (m_cpuInfo.totalNumaNum != numaNodeNum || !m_cpuInfo.cpuArrSize) { ereport(WARNING, - (errmsg("Can not activate NUMA distribute because no multiple NUMA nodes or CPUs are available."))); - return false; + (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("Can not activate NUMA distribute because no available cpu in node %d.", i))); - return false; + (errmsg("无法激活 NUMA 分布，因为节点 %d 中没有可用的 CPU。", i))); + return false; // 如果某个 NUMA 节点中没有可用的 CPU，返回 false } } - return true; + return true; // 其他情况下，返回 true } CPUBindType ThreadPoolControler::GetCpuBindType() const @@ -669,55 +696,54 @@ void ThreadPoolControler::SetGroupAndThreadNum() { if (m_attr.groupNum == 0) { if (m_cpuInfo.totalNumaNum > 0) - m_groupNum = m_cpuInfo.activeNumaNum; + m_groupNum = m_cpuInfo.activeNumaNum; // 如果未指定线程组数且存在 NUMA 节点，则使用激活的 NUMA 节点数 else - m_groupNum = DEFAULT_THREAD_POOL_GROUPS; + m_groupNum = DEFAULT_THREAD_POOL_GROUPS; // 否则使用默认的线程组数 } else { - m_groupNum = m_attr.groupNum; + m_groupNum = m_attr.groupNum; // 如果指定了线程组数，则使用指定的线程组数 } if (m_attr.threadNum == 0) { if (m_cpuInfo.activeCpuNum > 0) - m_threadNum = m_cpuInfo.activeCpuNum * THREAD_CORE_RATIO; + m_threadNum = m_cpuInfo.activeCpuNum * THREAD_CORE_RATIO; // 如果未指定线程数且存在激活的 CPU，则计算线程数 else - m_threadNum = DEFAULT_THREAD_POOL_SIZE; + m_threadNum = DEFAULT_THREAD_POOL_SIZE; // 否则使用默认的线程数 } else { - m_threadNum = m_attr.threadNum; + m_threadNum = m_attr.threadNum; // 如果指定了线程数，则使用指定的线程数 } - ConstrainThreadNum(); + ConstrainThreadNum(); // 调用 ConstrainThreadNum 方法进行线程数约束 } void ThreadPoolControler::ConstrainThreadNum() { - /* Thread pool size should not be larger than max_connections. */ + /* 线程池大小不应超过 max_connections。 */ if (MAX_THREAD_POOL_SIZE > g_instance.attr.attr_network.MaxConnections) { ereport(LOG, (errcode(ERRCODE_OPERATE_INVALID_PARAM), - errmsg("Max thread pool size %d should not be larger than max_connections %d, " - "so reduce max thread pool size to max_connections", + 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); - m_threadNum = Min(m_threadNum, m_maxPoolSize); + 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; + return m_maxPoolSize; // 返回最大线程池大小 } ThreadPoolStat* ThreadPoolControler::GetThreadPoolStat(uint32* num) { - ThreadPoolStat* result = (ThreadPoolStat*)palloc(m_groupNum * sizeof(ThreadPoolStat)); + ThreadPoolStat* result = (ThreadPoolStat*)palloc(m_groupNum * sizeof(ThreadPoolStat)); // 分配存储线程池统计信息的内存 int i; for (i = 0; i < m_groupNum; i++) { - m_groups[i]->GetThreadPoolGroupStat(&result[i]); + m_groups[i]->GetThreadPoolGroupStat(&result[i]); // 获取每个线程组的统计信息 } - *num = m_groupNum; - return result; + *num = m_groupNum; // 返回线程组数量 + return result; // 返回线程池统计信息数组 } void ThreadPoolControler::CloseAllSessions() @@ -725,14 +751,14 @@ 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(); + m_sessCtrl->MarkAllSessionClose(); // 标记所有会话为关闭状态 + (void)SignalCancelAllBackEnd(); // 发送取消信号以取消所有后端任务 for (int i = 0; i < m_groupNum; i++) { - m_groups[i]->GetListener()->SendShutDown(); + m_groups[i]->GetListener()->SendShutDown(); // 发送关闭信号给所有监听器 } - /* Check until all groups have closed their sessions. */ + /* 检查直到所有组都关闭了它们的会话。 */ bool allclose = false; while (!allclose) { if (m_sessCtrl->IsActiveListEmpty()) { @@ -741,9 +767,9 @@ void ThreadPoolControler::CloseAllSessions() allclose = true; for (int i = 0; i < m_groupNum; i++) { - allclose = (m_groups[i]->AllSessionClosed() && allclose); + allclose = (m_groups[i]->AllSessionClosed() && allclose); // 检查每个组的会话是否都已关闭 } - pg_usleep(one_hundred_micro_sec); + pg_usleep(one_hundred_micro_sec); // 短暂休眠以减少 CPU 使用 } ereport(LOG, (errmodule(MOD_THREAD_POOL), @@ -753,21 +779,21 @@ void ThreadPoolControler::CloseAllSessions() void ThreadPoolControler::ShutDownThreads(bool forceWait) { for (int i = 0; i < m_groupNum; i++) { - m_groups[i]->ShutDownThreads(); + m_groups[i]->ShutDownThreads(); // 关闭所有线程组中的线程 } ereport(LOG, (errmodule(MOD_THREAD_POOL), errmsg("pmState:%d, shut down all threadpool threads.", pmState))); if (forceWait) { - /* Check until all groups have shut down their workers. */ + /* 检查直到所有组都关闭了它们的工作线程。 */ bool allshut = false; while (!allshut) { allshut = true; for (int i = 0; i < m_groupNum; i++) { - allshut = (m_groups[i]->AllThreadShutDown() && allshut); + allshut = (m_groups[i]->AllThreadShutDown() && allshut); // 检查每个组的线程是否都已关闭 } - pg_usleep(one_hundred_micro_sec); + pg_usleep(one_hundred_micro_sec); // 短暂休眠以减少 CPU 使用 } ereport(LOG, (errmodule(MOD_THREAD_POOL), @@ -778,16 +804,16 @@ void ThreadPoolControler::ShutDownThreads(bool forceWait) void ThreadPoolControler::ShutDownListeners(bool forceWait) { for (int i = 0; i < m_groupNum; i++) { - m_groups[i]->GetListener()->ShutDown(); + 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; + allshut = (m_groups[i]->GetListener()->GetThreadId() == 0) && allshut; // 检查监听器线程是否都已关闭 } - pg_usleep(one_hundred_micro_sec); + pg_usleep(one_hundred_micro_sec); // 短暂休眠以减少 CPU 使用 } } } @@ -828,20 +854,20 @@ void ThreadPoolControler::AddWorkerIfNecessary() ThreadPoolGroup* ThreadPoolControler::FindThreadGroupWithLeastSession() { - int idx = 0; - float4 least_session = 0.0; - float4 session_per_thread = 0.0; + int idx = 0; // 用于记录具有最少会话的线程组的索引 + float4 least_session = 0.0; // 用于记录最少会话数 + float4 session_per_thread = 0.0; // 用于记录每个线程的平均会话数 - least_session = m_groups[0]->GetSessionPerThread(); + 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; + session_per_thread = m_groups[i]->GetSessionPerThread(); // 获取当前线程组的平均会话数 + if (session_per_thread < least_session) { // 如果当前线程组的平均会话数更小 + least_session = session_per_thread; // 更新最少会话数 + idx = i; // 更新具有最少会话的线程组的索引 } } - return m_groups[idx]; + return m_groups[idx]; // 返回具有最少会话的线程组的指针 } bool ThreadPoolControler::StayInAttachMode() @@ -855,15 +881,14 @@ int ThreadPoolControler::DispatchSession(Port* port) knl_session_context* sc = NULL; /* - * In comm_proxy mode, each accepted fd is combined with a fixed communicator thread in one NUMA group, - * we no longer distribute it with old mothod "group with latest sessions", - * so just return the communicator's NUMA group. + * 在comm_proxy模式下，每个接受的文件描述符（fd）都与一个 + * 固定的通信线程合并在一个NUMA组中，我们不再使用旧的方法“ + * 与最新会话的组合”进行分配，因此只需返回通信线程所在的NUMA组即可。 * - * Note: We assume that each connected user session can be equal-possibily distributed to communicators, - * fortunately,it looks like Euler OS can guarantee this(proved), - * otherwise we need revisit it. + * 注意：我们假设每个连接的用户会话可以等可能地分布给通信器， + * 幸运的是，欧拉操作系统似乎可以保证这一点（已经证明），否则我们还需要重新考虑这个问题。 * - * Performance optimization with comm_proxy when thread_pool m_groupNum same as comm_proxy numa groups + * 性能优化：在使用`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); @@ -876,7 +901,7 @@ int ThreadPoolControler::DispatchSession(Port* port) Assert(false); return STATUS_ERROR; } - /* if this group is hanged, we don't accept new session */ + /* 如果这个组挂起了，我们不会接受新的会话。 */ if (grp->IsGroupHanged()) { ereport(WARNING, (errmodule(MOD_THREAD_POOL), @@ -893,29 +918,39 @@ int ThreadPoolControler::DispatchSession(Port* port) } /* - * Bind the specified thread to all the available CPUs. - * This is invoked by auxiliary thread, such as WALSender. + * 将指定的线程绑定到所有可用的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))); } -