MPI + OpenMP 分段错误和不可预测的行为
MPI + OpenMP segmentation fault and unpredictable behavior
经过数小时的尝试类似问题的解决方案后，我决定发布此内容，但没有成功。我正在编写一个C++ MPI + OpenMP 代码，其中一个 MPI 节点（服务器）将双数组发送到其他节点。服务器生成线程以便同时发送到多个客户端。串行版本（仅使用 MPI）运行良好，单线程版本也是如此。多线程版本 （openmp） 在随机迭代次数后不断抛出分段错误错误。该行printf("%d: cur_idx:%d, opt_k.k:%d, idx:%d, N:%d n", tid, cur_idx,opt_k.k,idx,N)在每次迭代时打印出值。不可预测性是迭代次数（在一个事件中，代码成功运行，只是在我尝试立即再次运行它时抛出 seg 错误错误）。但是，它始终以 num_threads=1 完成。getData 返回结构向量，结构定义为 （int，int，double *）。
这是代码
double *tStatistics=new double[8], tmp_time; // wall clock time
double SY, Sto;
int a_tasks=0, file_p=0;
vector<myDataType *> d = getData();
int idx=0; opt_k.k=1; opt_k.proc_files=0; opt_k.p=this->node_sz;
opt_k.proc_files=0; SY=0; Sto=0;
std::fill(header,header+SZ_HEADER,-1);
omp_set_num_threads(5);// for now
// parallel region
#pragma omp parallel default(none) shared(d,idx,SY,Sto) private(a_tasks)
{
double *myHeader=new double[SZ_HEADER];
std::fill(myHeader,myHeader+SZ_HEADER,0);
int tid = omp_get_thread_num(), cur_idx, cur_k; int N;
//#pragma omp atomic
N=d.size();
while (idx<N) {
// Assign tasks and fetch results where available
cur_idx=N;
#pragma omp critical(update__idx)
{
if (idx<N) {
cur_idx=idx; cur_k=opt_k.k; idx+=cur_k;
}
}
if (cur_idx<N) {
printf("%d: cur_idx:%d, opt_k.k:%d, idx:%d, N:%d n", tid, cur_idx,opt_k.k,idx,N);
MPI_Recv(myHeader,SZ_HEADER,MPI_DOUBLE,MPI_ANY_SOURCE,MPI_ANY_TAG,MY_COMM_GRP,this->Stat);
if(this->Stat->MPI_TAG == TAG_HEADER){ // serve tasks
while (cur_k && cur_idx<N) {
myHeader[1]=d[cur_idx]->nRows; myHeader[2]=d[cur_idx]->nCols; myHeader[3]=cur_idx; myHeader[9]=--cur_k;
MPI_Send(myHeader,SZ_HEADER,MPI_DOUBLE,(int)myHeader[4],TAG_DATA,MY_COMM_GRP);
MPI_Send(d[cur_idx]->data,d[cur_idx]->nRows*d[cur_idx]->nCols,MPI_DOUBLE,(int)myHeader[4],TAG_DATA,MY_COMM_GRP);
delete[] d[cur_idx]->data; ++cur_idx;
}
}else if(this->Stat->MPI_TAG == TAG_RESULT){ // collect results
printf("%d - 4n", tid);
}
} //end if(loopmain)
} // end while(loopmain)
} // end parallel section
message("terminate slaves");
for(int i=1;i<node_sz;++i){ // terminate
MPI_Recv(header,SZ_HEADER,MPI_DOUBLE,MPI_ANY_SOURCE,MPI_ANY_TAG,MY_COMM_GRP,this->Stat);
MPI_Send(header,SZ_HEADER,MPI_DOUBLE,(int)header[4],TAG_TERMINATE,MY_COMM_GRP);
}
return 0;
另一个匹配函数是
void CMpifun::slave2()
{
double *Data; vector<myDataType> dataQ; vector<hist_type> resQ;
char out_opt='b'; // irrelevant
myDataType *out_im = new myDataType; hist_type *out_hist; CLdp ldp;
int file_cnt=0; double tmp_t; //local variables
while (true) { // main while loop
header[4]=myRank; MPI_Send(header,SZ_HEADER,MPI_DOUBLE,MASTER,TAG_HEADER,MY_COMM_GRP);
MPI_Recv(header,SZ_HEADER,MPI_DOUBLE,MASTER,MPI_ANY_TAG,MY_COMM_GRP,this->Stat);
if(this->Stat->MPI_TAG == TAG_TERMINATE) {
break;
}
//receive data
while(true) {
Data=new double[(int)(header[1]*header[2])];
MPI_Recv(Data,(int)(header[1]*header[2]),MPI_DOUBLE,MASTER,TAG_DATA,MY_COMM_GRP,this->Stat);
myDataType d; d.data=Data; d.nRows=(int)header[1]; d.nCols=(int)header[2];
//dataQ.push_back(d);
delete[] Data;
file_cnt++;
if ((int)header[9]) {
MPI_Recv(header,SZ_HEADER,MPI_DOUBLE,MASTER,TAG_DATA,MY_COMM_GRP,this->Stat);
} else break;
}
} // end main while loop
message("terminating");
我已经尝试了解决类似问题的所有建议。这是我的环境设置
export OMP_WAIT_POLICY="active"
export OMP_NUM_THREADS=4
export OMP_DYNAMIC=true # "true","false"
export OMP_STACKSIZE=200M #
export KMP_STACKSIZE=$OMP_STACKSIZE
ulimit -s unlimited
非常感谢所有参与进来的人。我越来越相信这与内存分配有关，但也不明白为什么。我现在有以下代码：
double CMpifun::sendData2()
{
double *tStatistics=new double[8], tmp_time; // wall clock time
double SY, Sto; int a_tasks=0, file_p=0;
vector<myDataType *> d = getData();
int idx=0; opt_k.k=1; opt_k.proc_files=0; opt_k.p=this->node_sz;
opt_k.proc_files=0; SY=0; Sto=0;
std::fill(header,header+SZ_HEADER,-1);
omp_set_num_threads(224);// for now
// parallel region
#pragma omp parallel default(none) shared(idx,SY,Sto,d) private(a_tasks)
{
double *myHeader=new double[SZ_HEADER];
std::fill(myHeader,myHeader+SZ_HEADER,0);
int tid = omp_get_thread_num(), cur_idx, cur_k; int N;
//#pragma omp critical(update__idx)
{
N=d.size();
}
while (idx<N) {
// Assign tasks and fetch results where available
cur_idx=N;
#pragma omp critical(update__idx)
{
if (idx<N) {
cur_idx=idx; cur_k=opt_k.k; idx+=cur_k;
}
}
if (cur_idx<N) {
//printf("%d: cur_idx:%d, opt_k.k:%d, idx:%d, N:%d n", tid, cur_idx,opt_k.k,idx,N);
printf("%d: cur_idx:%d, N:%d n", tid, cur_idx,N);
//#pragma omp critical(update__idx)
{
MPI_Recv(myHeader,SZ_HEADER,MPI_DOUBLE,MPI_ANY_SOURCE,MPI_ANY_TAG,MY_COMM_GRP,this->Stat);
}
if(this->Stat->MPI_TAG == TAG_HEADER){ // serve tasks
while (cur_k && cur_idx<N) {
//#pragma omp critical(update__idx)
{
myHeader[1]=d[cur_idx]->nRows; myHeader[2]=d[cur_idx]->nCols; myHeader[3]=cur_idx;
myHeader[9]=--cur_k;
MPI_Send(myHeader,SZ_HEADER,MPI_DOUBLE,(int)myHeader[4],TAG_DATA,MY_COMM_GRP);
MPI_Send(d[cur_idx]->data,d[cur_idx]->nRows*d[cur_idx]->nCols,MPI_DOUBLE,(int)myHeader[4],TAG_DATA,MY_COMM_GRP);
delete[] d[cur_idx]->data;
}
++cur_idx;
}
}else if(this->Stat->MPI_TAG == TAG_RESULT){ // collect results
printf("%d - 4n", tid);
}
} //end if(loopmain)
} // end while(loopmain)
} // end parallel section
message("terminate slaves");
for(int i=1;i<node_sz;++i){ // terminate
MPI_Recv(header,SZ_HEADER,MPI_DOUBLE,MPI_ANY_SOURCE,MPI_ANY_TAG,MY_COMM_GRP,this->Stat);
MPI_Send(header,SZ_HEADER,MPI_DOUBLE,(int)header[4],TAG_TERMINATE,MY_COMM_GRP);
}
return 0;
而且是一对
void CMpifun::slave2()
{
double *Data; vector<myDataType> dataQ; vector<hist_type> resQ;
char out_opt='b'; // irrelevant
myDataType *out_im = new myDataType; hist_type *out_hist; CLdp ldp;
int file_cnt=0; double tmp_t; //local variables
while (true) { // main while loop
header[4]=myRank; MPI_Send(header,SZ_HEADER,MPI_DOUBLE,MASTER,TAG_HEADER,MY_COMM_GRP);
MPI_Recv(header,SZ_HEADER,MPI_DOUBLE,MASTER,MPI_ANY_TAG,MY_COMM_GRP,this->Stat);
if(this->Stat->MPI_TAG == TAG_TERMINATE) {
break;
}
//receive data
while(true) {
Data=new double[(int)(header[1]*header[2])];
MPI_Recv(Data,(int)(header[1]*header[2]),MPI_DOUBLE,MASTER,TAG_DATA,MY_COMM_GRP,this->Stat);
myDataType *d=new myDataType; d->data=Data; d->nRows=(int)header[1]; d->nCols=(int)header[2];
dataQ.push_back(*d);
delete[] Data;
file_cnt++;
if ((int)header[9]) {
MPI_Recv(header,SZ_HEADER,MPI_DOUBLE,MASTER,TAG_DATA,MY_COMM_GRP,this->Stat);
} else break;
}
// Error section: Uncommenting next line causes seg fault
/*while (dataQ.size()) { // process data
out_hist = new hist_type();
myDataType d = dataQ.back(); dataQ.pop_back(); // critical section
ldp.process(d.data, d.nRows,d.nCols,out_opt,out_im, out_hist);
resQ.push_back(*out_hist); out_hist=0;
delete[] d.data; delete[] out_im->data;
}*/
//time_arr[1] /= file_cnt; time_arr[2] /= file_cnt;
//header[6]=time_arr[0]; header[7]=time_arr[1]; header[8]=time_arr[2];
//header[4]=myRank; header[9]=resQ.size();
} // end main while loop
更新是，如果我取消注释 Slave2（） 函数中的 while 循环，则运行不会完成。我不明白的是，这个函数（slave2）没有任何openmp/线程，但它似乎有效果。此外，它不与线程函数共享任何变量。如果我注释掉麻烦的部分，那么无论我设置的线程数（4、18、300）如何，代码都会运行。我的 OpenMP 环境变量仍保持原样。limit -a的输出如下：
core file size (blocks, -c) 0
data seg size (kbytes, -d) unlimited
scheduling priority (-e) 0
file size (blocks, -f) unlimited
pending signals (-i) 30473
max locked memory (kbytes, -l) 64
max memory size (kbytes, -m) unlimited
open files (-n) 1024
pipe size (512 bytes, -p) 8
POSIX message queues (bytes, -q) 819200
real-time priority (-r) 0
stack size (kbytes, -s) 37355
cpu time (seconds, -t) unlimited
max user processes (-u) 30473
virtual memory (kbytes, -v) unlimited
file locks (-x) unlimited
我的构造函数也调用mpi_init_thread。为了解决这个问题@Tim，我使用动态内存（带new）的原因是为了不使堆栈内存膨胀，遵循类似问题的解决方案的建议。感谢您的帮助。
我看到的最大问题是你的代码表现出的许多竞争条件。您所看到的不稳定行为无疑是由此引起的。请记住，任何时候您访问 OpenMP 中的共享变量（通过 shared 关键字或全局范围声明）时，您都会访问可由该组中的任何其他线程读取或写入的内存，而不能保证顺序。例如
N = d.size();
是一种争用条件，因为std::vector不是线程安全的。由于您在类中使用 OpenMP，因此任何成员变量也被视为"全局"，因此默认情况下不是线程安全的。
如@tim18所述，由于您是从 OpenMP 并行区域中调用 MPI 例程的，因此应使用 MPI_Init_thread 函数将 MPI 运行时初始化为线程安全。
顺便说一句，您的C++需要一些工作。切勿在用户级代码中使用new或delete。使用 RAII 管理对象生存期，并将大型数据结构包装在为您管理生存期的瘦对象中。例如，此行
delete[] d[cur_idx]->data;
告诉我有恶魔潜伏在你的代码中，等待释放给毫无戒心的用户（可能是你！顺便说一下，这也是一个竞争条件。许多恶魔！
相关文章：
- 如何确保在使用基于布尔值的两个方法之一调用方法时避免分支预测错误
- 不可预测的C++睡眠/等待行为
- 超过 N 时出现不可预测的位移结果
- DirectShow CSourceStream::FillBuffer 暂停和查找后对第一帧的不可预测的调用数
- 多线程文件 IO 程序在线程数增加时行为不可预测
- C++映射/设置迭代器不可递增错误
- 使用 std：：find 查找从二进制文件中读取的字符并转换为 std：：vector 中的 std：：string<string> 会产生这种不可预测的行为？
- std::sleep_for在Windows 10上的不可预测行为
- 不可预测的文件描述符泄漏
- 循环程序的行为是不可预测的
- C++中不可预测的输出
- 当来自外部库的线程不可预测地崩溃我的应用程序时，我该怎么办
- 不可预测的伪RNG
- C++ 自定义运算符(+=） 的行为方式不可预测
- 如何优化间接基数排序?(又名如何优化不可预测的内存访问模式）
- 修复不可预测的崩溃
- 不可预测的指针行为
- wxWidgets 不可预测的 Seg 故障
- 是什么导致了这个不可预测的OpenGL错误？
- MPI + OpenMP 分段错误和不可预测的行为
