我需要尽可能高效地计算一个向量矩阵的积。具体来说,给定向量
目前使用的朴素方法是设计如下:
它运行良好,耗时近12000微秒。请注意,向量 s 具有499个元素,而A是499 x 15500。
下一步是尝试并行计算:如果我有N个线程,则可以将向量 s 的每个部分和矩阵 A 的“相应”行分配给每个线程。每个线程将计算一个大小为499的向量,最终结果将是它们的逐个条目之和。
首先,在类
最后,我修改了上述展示的
编辑 - 使用"轻量级"
然后有一种方法可以检索矩阵的特定条目:
s
和矩阵A
,我需要计算s * A
。我有一个名为Vector
的类,它包装了一个std::vector
,以及一个名为Matrix
的类,它也包装了一个std::vector
(为了提高效率)。目前使用的朴素方法是设计如下:
Vector<T> timesMatrix(Matrix<T>& matrix)
{
Vector<unsigned int> result(matrix.columns());
// constructor that does a resize on the underlying std::vector
for(unsigned int i = 0 ; i < vector.size() ; ++i)
{
for(unsigned int j = 0 ; j < matrix.columns() ; ++j)
{
result[j] += (vector[i] * matrix.getElementAt(i, j));
// getElementAt accesses the appropriate entry
// of the underlying std::vector
}
}
return result;
}
它运行良好,耗时近12000微秒。请注意,向量
下一步是尝试并行计算:如果我有N个线程,则可以将向量
首先,在类
Matrix
中,我添加了一个方法来从Matrix
中提取一些行并构建一个较小的矩阵:Matrix<T> extractSomeRows(unsigned int start, unsigned int end)
{
unsigned int rowsToExtract = end - start + 1;
std::vector<T> tmp;
tmp.reserve(rowsToExtract * numColumns);
for(unsigned int i = start * numColumns ; i < (end+1) * numColumns ; ++i)
{
tmp.push_back(matrix[i]);
}
return Matrix<T>(rowsToExtract, numColumns, tmp);
}
然后我定义了一个线程例程
void timesMatrixThreadRoutine
(Matrix<T>& matrix, unsigned int start, unsigned int end, Vector<T>& newRow)
{
// newRow is supposed to contain the partial result
// computed by a thread
newRow.resize(matrix.columns());
for(unsigned int i = start ; i < end + 1 ; ++i)
{
for(unsigned int j = 0 ; j < matrix.columns() ; ++j)
{
newRow[j] += vector[i] * matrix.getElementAt(i - start, j);
}
}
}
最后,我修改了上述展示的
timesMatrix
方法的代码:Vector<T> timesMatrix(Matrix<T>& matrix)
{
static const unsigned int NUM_THREADS = 4;
unsigned int matRows = matrix.rows();
unsigned int matColumns = matrix.columns();
unsigned int rowsEachThread = vector.size()/NUM_THREADS;
std::thread threads[NUM_THREADS];
Vector<T> tmp[NUM_THREADS];
unsigned int start, end;
// all but the last thread
for(unsigned int i = 0 ; i < NUM_THREADS - 1 ; ++i)
{
start = i*rowsEachThread;
end = (i+1)*rowsEachThread - 1;
threads[i] = std::thread(&Vector<T>::timesMatrixThreadRoutine, this,
matrix.extractSomeRows(start, end), start, end, std::ref(tmp[i]));
}
// last thread
start = (NUM_THREADS-1)*rowsEachThread;
end = matRows - 1;
threads[NUM_THREADS - 1] = std::thread(&Vector<T>::timesMatrixThreadRoutine, this,
matrix.extractSomeRows(start, end), start, end, std::ref(tmp[NUM_THREADS-1]));
for(unsigned int i = 0 ; i < NUM_THREADS ; ++i)
{
threads[i].join();
}
Vector<unsigned int> result(matColumns);
for(unsigned int i = 0 ; i < NUM_THREADS ; ++i)
{
result = result + tmp[i]; // the operator+ is overloaded
}
return result;
}
它仍然运行,但现在需要近30000微秒,几乎是之前的三倍。
我做错了什么吗?你认为有更好的方法吗?
编辑 - 使用"轻量级" VirtualMatrix
根据Ilya Ovodov的建议,我定义了一个类VirtualMatrix
,它包装一个T* matrixData
,并在构造函数中初始化为:
VirtualMatrix(Matrix<T>& m)
{
numRows = m.rows();
numColumns = m.columns();
matrixData = m.pointerToData();
// pointerToData() returns underlyingVector.data();
}
然后有一种方法可以检索矩阵的特定条目:
inline T getElementAt(unsigned int row, unsigned int column)
{
return *(matrixData + row*numColumns + column);
}
现在执行时间有所改善(大约8000微秒),但也许还有一些改进的空间。特别是线程例程现在
void timesMatrixThreadRoutine
(VirtualMatrix<T>& matrix, unsigned int startRow, unsigned int endRow, Vector<T>& newRow)
{
unsigned int matColumns = matrix.columns();
newRow.resize(matColumns);
for(unsigned int i = startRow ; i < endRow + 1 ; ++i)
{
for(unsigned int j = 0 ; j < matColumns ; ++j)
{
newRow[j] += (vector[i] * matrix.getElementAt(i, j));
}
}
}
真正缓慢的部分是嵌套for
循环。如果我将其删除,则结果显然是错误的,但“计算”时间不到500微秒。这就是说,现在传递参数几乎不需要时间,而重要的是计算。
根据您的看法,是否有任何方法可以使其更快?
return matrix[row*numColumns + column]
,其中matrix
是存储所有元素的std::vector
的名称。我可以尝试使用data()
来查看它是否更快。 - minomic