第十章 超越通用核心的多线程
基于通用核心的附加子句
- 并行构造的附加子句:
- num_threads(integer-expression)
用于设置线程总数. - if(scalar-expression)
用于为并行构造提供条件分支. - copyin(list)
- proc_bind(master|close|spread)
- num_threads(integer-expression)
为了测试num_threads子句与if子句的用法,构造下面所示原型:
#include <iostream>
#include <omp.h>
int main()
{
int NTHREAD, x;
std::cin >> x;
#pragma omp parallel if(x>1) \
num_threads(x/2) //通过\来实现跨行
{
if(omp_get_thread_num()==0)
NTHREAD = omp_get_num_threads();
}
std::cout << "the num of thread is: " << NTHREAD;
return 0;
}
通过程序实验,证明了其相关用法.
-
共享工作循环构造的附加子句:
- lastprivate(list)
如同private与firstprivate子句一样,其为列表中每个变量创建一个私有副本,在区域结束时,列表中每个变量的原始变量将被赋值为最后一次迭代值. - schedule子句中的附加调度类型:
- 启发式调度(guided):动态调度的另一种形式,其中chunk_size一开始是一个大值,每次执行新的分块迭代后chunk_size都会减少,直至chunk_size的最小值.
- 自动调度(auto):编译器和运行时根据自己的选择来安排循环迭代.
- 运行时调度(runtime):调度和可能的chunk_size来自内部控制变量.
据文档所言,其提供了omp_set_schedule与omp_get_schedule两个函数与omp_sched_t枚举类来处理.但是,实际编码时并未成功使用相关函数及枚举类.
- collapse(n)
规定共享工作循环构造之后的n个循环将被合并为一个隐式循环.任何额外的子句,包括数据环境子句或归约都会应用到这个隐式循环中.
- lastprivate(list)
-
任务构造的附加值局:
- untied
该子句用于限制任务队列增长,在任务队列增长时避免任务队列增长速度过大. - priority(priority-value)
该子句可以显式提示任务的执行优先级,优先级值的范围为[0,max-task-prioriity-var].
最大值可以通过环境变量OMP_MAX_TASK_PRIORITY设置.
也可以通过omp_get_max_task_priority(void)函数查询. - depend(dependence-type:list)
用于处理依赖情况下的任务,分析模式类似于DAG(有向无环图).
其中dependence-type包括out,in和inout三种,带有in依赖类型的变量会导致任务等待另一个任务完成,该任务在带有out依赖类型的子句中具有相同的变量. - if(scalar-expression)
如果if子句中表达式为false,那么任务将不会被延迟执行. - final(scalar expression)
当final子句中的表达式为true,那么任务将会被立即执行. - mergeable
用于指示编译器是否可以将两个或多个连续的任务合并为一个任务.
- untied
-
创建一个显式任务调度点:
#pragma omp taskyield
- 创建一个任务循环构造:
#pragma omp taskloop [clause[, clause] ...]
//for-loop
- 创建一个同步任务组:
#pragma omp taskgroup [clause[, clause] ...]
{
//body of taskgroup
}
为了理解depend子句的使用,下面通过一个实例来帮助理解:
#include <omp.h>
int main()
{
int A,B,C,G,F;
#pragma omp parallel shared(A,B,C,G,F)
{
#pragma omp task depend(out:A)
TaskA(&A);
#pragma omp task depend(in:A,G)
TaskB(&B);
#pragma omp task depend(in:A) depend(out:C)
TaskC(&C);
#pragma omp task depend(in:A) depend(out:G)
TaskG(&G);
#pragma omp task depend(in:C,G)
TaskF(&F);
}
return 0;
}
通用核心中缺失的多线程功能
- threadprivate
OpenMP的基本内存模型将内存视为一组给内存中的地址命名的变量.除了shared和private两类,OpenMP还定义了第三种内存类型: threadprivate.
threadprivate内存是一个线程的私有内存,它不能被其他线程访问.然而,其内存中的变量在各个例程中具有可见性. 在非正式情况下,可以认为threadprivate内存是线程的私有内存.它不能被其他线程访问.
threadprivate是一个声明性指令,这意味着它出现在程序中声明变量的地方,并影响其声明的语义.
- 声明threadprivate内存:
#pragma omp threadprivate(list)
为了理解threadprivate的使用,我们回到第七章所述的链表程序:
#include <iostream>
#include <cstdlib>
#include <omp.h>
import <format>;
#define NODE_NUM 20
#define CHUNK 2
#define NTHREADS 3
typedef struct node {
int data;
int procResult;
struct node* next;
node() :data(0), procResult(0), next(nullptr) {}
}Node, * List;
int count = 0;
#pragma omp threadprivate(count)
void incCount()
{
count++;
return;
}
void initList(List p)
{
Node* root{ p };
Node* temp_node;
p->data = 0;
for (int i = 1; i < NODE_NUM; i++) {
temp_node = new Node;
temp_node->data = i;
root->next = temp_node;
root = temp_node;
}
return;
}
void processWork(Node* n)
{
n->procResult = (n->data * n->data);
return;
}
void deleteList(List p)
{
Node* temp_node = p->next;
for (; p != temp_node;) {
temp_node = p;
while (temp_node->next != nullptr && temp_node->next->next != nullptr)
temp_node = temp_node->next;
delete temp_node->next;
temp_node->next = nullptr;
}
delete p;
return;
}
int main()
{
List list = new Node;
Node** parr = new Node * [NODE_NUM];
initList(list);
#ifdef NTHREADS
omp_set_num_threads(NTHREADS);
#endif // NTHREADS
Node* p;
#pragma omp parallel
{
#pragma omp single
{
p = list;
while (p != nullptr)
{
#pragma omp task firstprivate(p)
{
incCount();
processWork(p);
std::cout << std::format("in the {} thrd, the count is {}",
omp_get_thread_num(),
count
) << std::endl;
}//end of task creation
p = p->next;
}
}//end of single region
}//end of parallel region
deleteList(list);
return 0;
}
我们在链表程序的基础上添加了一个threadprivate内存的count,用于统计在线程中执行的task数量.threadprivate数据与特定线程相绑定,因此会在程序中引入错误源.
- master
master 构造定义了一个由线程组的主线程执行的工作块.与single构造不同,它的构造末尾没有隐式的栅栏.
- 声明一个master构造:
#pragma omp master
{
//body of master
}
- atomic
atomic 构造确保了一个变量作为一个独立的,不间断的动作被读取,写入或更新.其保护了一个变量,避免了并发线程对一个存储位置进行多次同步更新的可能性.
atomic构造与critical构造有很大共同点,如果多个线程试图同时执行一个atomic构造,"第一个线程"将执行原子操作,而其他线程将等待轮到自己
atomic构造中通过子句定义原子操作的类型,其中最常见的有三种:读,写和更新(不包括捕获).默认情况(不包含子句)是更新.
| clause | 原子操作示例 |
|---|---|
| read | v=x; |
| write | x=expr; |
| update (default) |
x++;x--;++x;--x; x = expr;v = expr(x); |
现在让我们回到第四章中关于Pi数值积分的部分.
#include <iostream>
#include <omp.h>
#include <fstream>
import <format>;
#define TURNS 100
#define PI 3.141592653589793
long double num_steps = 1e8;
double step;
int main()
{
std::ofstream out;
out.open("example.csv", std::ios::ate);
out << "NTHREADS,pi,err,run_time,num_steps" << std::endl;
double sum = 0.0;
for (int NTHREADS = 1; NTHREADS < TURNS; NTHREADS++) {
double start_time, run_time;
double pi, err;
pi = sum = 0.0;
int actual_nthreads;
step = 1.0 / (double)num_steps;
omp_set_num_threads(NTHREADS);
start_time = omp_get_wtime();
#pragma omp parallel
{
int id = omp_get_thread_num();
int numthreads = omp_get_num_threads();
double x;
double partial = 0.0;
if (id == 0)
actual_nthreads = numthreads;
int istart = id * num_steps / numthreads;
int iend = (id + 1) * num_steps / numthreads;
if (id == (numthreads - 1))
iend = num_steps;
for (int i = istart; i < iend; i++) {
x = (i + 0.5) * step;
partial += 4.0 / (1.0 + x * x);
}
#pragma omp atomic
sum += partial;
}//end of parallel
pi = step * sum;
err = pi - PI;
run_time = omp_get_wtime() - start_time;
std::cout << std::format("pi is {} in {} seconds {} thrds.step is {},err is {}",
pi,
run_time,
actual_nthreads,
step,
err
) << std::endl;
out << std::format("{},{:.15f},{:.15f},{:.15f},{}",
NTHREADS,
pi,
err,
run_time,
num_steps
) << std::endl;
}
out.close();
return 0;
}
我们在这里将critical构造更改为atomic构造实现了相同的功能.
然而,虽然类似于critical构造,但是atomic构造只适用于直接涉及内存中存储位置的操作,也就是说:
#pragma omp atomic
full_sum+=foo();
其中函数foo()的执行不受atomic构造的保护,其等价于:
tmp = foo();
#pragma omp atomic
full_sum+=tmp;
这意味着foo()执行过程中很可能发生数据竞争.
- OMP_STACKSIZE
OpenMP被设计为支持多种系统,操作系统代表正在执行的程序对进程进行管理.
进程分叉出与其关联的线程.
当操作系统创建线程时,它为每个线程预留了一些本地内存,这个内存以栈的形式进行管理.
栈的大小是有限的,如果在线程内部运行的代码创建了大的对象,栈内存可能会溢出,导致潜在的灾难性失败.
为了解决这个问题,OpenMP定义了一个叫做stacksize-var的内部控制变量,它控制线程组中每个线程相关联的内存栈的大小.
设置stacksize-var的命令如下:
export OMP_STACKSIZE=size
OpenMP定义了一系列单位用于处理size:
- size设置以1024字节为单位的大小
- sizeB设置以1字节为单位的大小
- sizeK设置以1024字节为单位的大小
- sizeM设置以1024 * 1024字节为单位的大小
- sizeG设置以1024 * 1024 * 1024字节为单位的大小
举例:
export OMP_STACKSIZE="200K"//200*1024 bytes
- omp_get_max_threads
omp_get_num_threads用于询问OpenMP运行时线程组有多少个线程,但是只能在同一个并行区域内调用.
但是有时候,需要一个可以从并行区域外调用的函数,以找到后续parallel构造所创建的线程组中可能获得的最大线程数..
此时就应当使用 omp_get_max_threads.
int omp_get_max_threads(void)
为了理解其使用,我们提供下面一个例子:
#include <iostream>
#include <omp.h>
int main()
{
int nthread_1, nthread_2;
omp_set_num_threads(2);
nthread_1 = omp_get_max_threads();
#pragma omp parallel
{
if (omp_get_thread_num() == 0)
std::cout << nthread_1 << std::endl;
}
omp_set_num_threads(4);
nthread_2 = omp_get_max_threads();
#pragma omp parallel
{
if (omp_get_thread_num() == 0)
std::cout << nthread_2 << std::endl;
}
return 0;
}
其最终得到的结果为:
2
4
证明了我们调用omp_get_max_threads()所得到的结果的正确.
- omp_set_dynamic
一个OpenMP程序通常由多个被并行区域分隔的顺序部分组成.OpenMP运行时会尝试对一个并行区域到下一个并行区域时,优化线程组的大小,这成为动态模式(dynamic mode).
这意味着OpenMP运行时必须假定与线程相关联的资源可能在并行区域之间发生变化.如果希望在并行区域之间重用线程资源,则需要告诉运行时系统关闭动态线程调度的功能.
通过omp_set_dynamic(),我们可以启用或禁用动态模式.
- 启用或禁用动态模式:
void omp_set_dynamic(int dyn_threads)
其中dyn_threads为一个bool值,其为true时将允许线程组大小再并行区域之间变化.
- omp_in_parallel
让活动线程的数量超过物理核心的数量会影响性能,因为操作系统会因为郭队线程交换而消耗资源,这就是所谓的认购超额.
因此,有些时候想知道自己是否在一个活跃的并行区域内,这样就可以调整后续并行区域中创建的线程数量.
omp_in_parallel()函数用于查询代码是否在并行区域内,如果在活动的并行区域内,那么返回true.
- 查询代码是否在并行区域内:
void omp_in_parallel();