弗兰克说得对，你目前的问题可能是你正在使用一个非原子操作：
average[k] += vector[k]
您可以通过使用以下方式进行修复：
average[k] += vector[k]
但更大的概念问题是，这可能不会加速你的代码。你正在执行的操作非常简单，而且内存（至少行）是连续的。
确实，我已经为您的代码制作了一个最小工作示例（您应该对您的问题做到这一点）：
#include <vector>
#include <cstdlib>
#include <chrono>
#include <iostream>
using namespace std;
vector<float> average(const vector<vector<unsigned char>>& original){
vector<float> average(original[0].size(), 0.0);
#pragma omp parallel for
for (int i=0; i<original.size(); i++) {
const vector<unsigned char>& vector = original[i];
for (int k = 0; k < vector.size(); ++k) {
#pragma omp atomic
average[k] += vector[k];
}
}
for (float& val : average) {
val /= original.size();
}
return average;
}
int main(){
vector<vector<unsigned char>> mat(1000);
for(int y=0;y<mat.size();y++)
for(int x=0;x<mat.size();x++)
mat.at(y).emplace_back(rand()%255);
std::chrono::steady_clock::time_point begin = std::chrono::steady_clock::now();
double dont_optimize = 0;
for(int i=0;i<100;i++){
auto ret = average(mat);
dont_optimize += ret[0];
}
std::chrono::steady_clock::time_point end = std::chrono::steady_clock::now();
std::cout<<"Time = "<<(std::chrono::duration_cast<std::chrono::microseconds>(end - begin).count()/100)<<std::endl;
return 0;
}
使用
g++ -O3 temp.cpp -fopenmp 编译可以启用 OpenMP。在我的四核机器上，运行时间始终约为 10,247 微秒。禁用 OpenMP 后，运行时间约为 2,561 微秒。
开启和管理线程团队是昂贵的。
但有一种真正的方法可以加速您的代码：改善内存布局。
使用
std::vector< std::vector<T> > 设计意味着每个
vector<T> 可以位于内存中的任何位置。相反，我们希望所有内存都是连续的。我们可以通过使用平坦数组索引来实现这一点，如下所示：
（请注意，下面代码的早期版本使用了例如
mat.at(y*width+x)。这意味着范围检查导致了与现在使用的
mat[y*width+x] 相比显着的速度损失。已相应地更新时间。）
#include <vector>
#include <cstdlib>
#include <chrono>
#include <iostream>
using namespace std;
class Matrix {
public:
vector<unsigned char> mat;
int width;
int height;
Matrix(int width0, int height0){
width = width0;
height = height0;
for(int i=0;i<width*height;i++)
mat.emplace_back(rand()%255);
}
unsigned char& operator()(int x, int y){
return mat[y*width+x];
}
unsigned char operator()(int x, int y) const {
return mat[y*width+x];
}
unsigned char& operator()(int i){
return mat[i];
}
unsigned char operator()(int i) const {
return mat[i];
}
};
vector<float> average(const Matrix& original){
vector<float> average(original.width, 0.0);
#pragma omp parallel for
for(int y=0;y<original.height;y++)
for(int x=0;x<original.width;x++)
#pragma omp atomic
average[x] += original(x,y);
for (float& val : average)
val /= original.height;
return average;
}
int main(){
Matrix mat(1000,1000);
std::cerr<<mat.width<<" "<<mat.height<<std::endl;
std::chrono::steady_clock::time_point begin = std::chrono::steady_clock::now();
double dont_optimize = 0;
for(int i=0;i<100;i++){
auto ret = average(mat);
dont_optimize += ret[0];
}
std::chrono::steady_clock::time_point end = std::chrono::steady_clock::now();
std::cout<<"Time = "<<(std::chrono::duration_cast<std::chrono::microseconds>(end - begin).count()/100)<<std::endl;
return 0;
}
请注意，我使用的是
float而不是
double：这样可以将两倍的数字压缩到同样的空间中，这对于缓存来说很有好处。
这将在没有OpenMP的情况下运行292微秒，在OpenMP的情况下运行9426微秒。
总之，使用OpenMP/并行会使代码变慢，因为并行处理所需的时间比设置并行处理的时间更短，但使用更好的内存布局可以提高约90%的速度。此外，使用方便的Matrix类可以提高代码的可读性和可维护性。
编辑：
将其运行在10,000x10,000而不是1,000x1,000的矩阵上会得到类似的结果。对于向量的向量：没有OpenMP的情况下为7,449微秒，有OpenMP的情况下为156,316微秒。对于平面数组索引：没有OpenMP的情况下为32,668微秒，有OpenMP的情况下为145,470微秒。
性能可能与我的机器上可用的硬件指令集有关（特别是，如果我的机器缺少原子指令，那么OpenMP将不得不用互斥锁等模拟它们）。事实上，在平面数组示例中，使用
-march=native编译可以改善OpenMP的性能，尽管仍然不太好：没有OpenMP的情况下为33,079微秒，有OpenMP的情况下为127,841微秒。我稍后会尝试在更强大的机器上进行实验。
编辑：
虽然上述测试是在Intel(R) Core(TM) i5 CPU M 480 @ 2.67GHz上执行的，但我已经在超级厉害的Intel(R) Xeon(R) CPU E5-2680 v3 @ 2.50GHz上编译了这段代码（使用
-O3 -march=native）。结果类似：
- 1000x1000，向量的向量，没有OpenMP：145μs
- 1000x1000，向量的向量，有OpenMP：2,941μs
- 10000x10000，向量的向量，没有OpenMP：20,254μs
- 10000x10000，向量的向量，有OpenMP：85,703μs
- 1000x1000，平面数组，没有OpenMP：139μs
- 1000x1000，平面数组，有OpenMP：3,171μs
- 10000x10000，平面数组，没有OpenMP：18,712μs
- 10000x10000，平面数组，有OpenMP：89,097μs
这证实了我们之前的结果：即使你的硬件非常棒，使用OpenMP也往往会使事情变慢。实际上，两个处理器之间的大部分加速可能是由于Xeon的大型L3缓存大小：它的大小为30,720K，比i5上的3,720K缓存大10倍。
编辑
将Zulan的简化策略从下面的答案中纳入，可以有效地利用并行处理：
vector<float> average(const Matrix& original){
vector<float> average(original.width, 0.0);
auto average_data = average.data();
#pragma omp parallel for reduction(+ : average_data[ : original.width])
for(int y=0;y<original.height;y++){
for(int x=0;x<original.width;x++)
average_data[x] += original(x,y);
}
for (float& val : average)
val /= original.height;
return average;
}
对于24个线程，在10,000x10,000的数组上运行时间为2629微秒：相对于串行版本，提高了7.1倍。在您原始代码上使用Zulan的策略（不使用平坦的数组索引）需要3529微秒，因此通过使用更好的布局仍然可以获得25%的加速。
