计算包含高维向量的两个矩阵之间的最小欧氏距离的最快方法

Fastest way to calculate minimum euclidean distance between two matrices containing high dimensional vectors

本文关键字:之间 欧氏距离 方法 向量 包含高 计算 两个      更新时间:2023-10-16

我在另一个线程上开始了一个类似的问题,但后来我专注于如何使用OpenCV。由于没有达到我最初想要的,我将在这里问我到底想要什么。

我有两个矩阵。矩阵a为2782x128,矩阵b为4000x128,均为无符号字符值。这些值存储在一个数组中。对于a中的每个向量,我需要b中具有最接近欧氏距离的向量的索引。

好的,现在我的代码来实现这个:

#include <windows.h>
#include <stdlib.h>
#include <stdio.h>
#include <cstdio>
#include <math.h>
#include <time.h>
#include <sys/timeb.h>
#include <iostream>
#include <fstream>
#include "main.h"
using namespace std;
void main(int argc, char* argv[])
{
int a_size;
unsigned char* a = NULL;
read_matrix(&a, a_size,"matrixa");
int b_size;
unsigned char* b = NULL;
read_matrix(&b, b_size,"matrixb");
LARGE_INTEGER liStart;
LARGE_INTEGER liEnd;
LARGE_INTEGER liPerfFreq;
QueryPerformanceFrequency( &liPerfFreq );
QueryPerformanceCounter( &liStart );
int* indexes = NULL;
min_distance_loop(&indexes, b, b_size, a, a_size);
QueryPerformanceCounter( &liEnd );
cout << "loop time: " << (liEnd.QuadPart - liStart.QuadPart) / long double(liPerfFreq.QuadPart) << "s." << endl;
if (a)
delete[]a;
if (b)
delete[]b;
if (indexes)
delete[]indexes;
return;
}
void read_matrix(unsigned char** matrix, int& matrix_size, char* matrixPath)
{
ofstream myfile;
float f;
FILE * pFile;
pFile = fopen (matrixPath,"r");
fscanf (pFile, "%d", &matrix_size);
*matrix = new unsigned char[matrix_size*128];
for (int i=0; i<matrix_size*128; ++i)
{
unsigned int matPtr;
fscanf (pFile, "%u", &matPtr);
matrix[i]=(unsigned char)matPtr;
}
fclose (pFile);
}
void min_distance_loop(int** indexes, unsigned char* b, int b_size, unsigned char* a, int a_size)
{
const int descrSize = 128;
*indexes = (int*)malloc(a_size*sizeof(int));
int dataIndex=0;
int vocIndex=0;
int min_distance;
int distance;
int multiply;
unsigned char* dataPtr;
unsigned char* vocPtr;
for (int i=0; i<a_size; ++i)
{
min_distance = LONG_MAX;
for (int j=0; j<b_size; ++j)
{
distance=0;
dataPtr = &a[dataIndex];
vocPtr = &b[vocIndex];
for (int k=0; k<descrSize; ++k)
{
multiply = *dataPtr++-*vocPtr++;
distance += multiply*multiply;
// If the distance is greater than the previously calculated, exit
if (distance>min_distance)
break;
}
// if distance smaller
if (distance<min_distance)
{
min_distance = distance;
(*indexes)[i] = j;
}
vocIndex+=descrSize;
}
dataIndex+=descrSize;
vocIndex=0;
}
}

附件是带有样本矩阵的文件。

matrixamatrixb

我使用windows.h只是为了计算消耗时间,所以如果你想在windows之外的另一个平台上测试代码,只需更改windows.h头并更改计算消耗时间的方式。

我电脑里的这个代码大约是0.5秒。问题是,我在Matlab中有另一个代码,它在0.05秒内完成了同样的事情。在我的实验中,我每秒都会收到几个矩阵,比如矩阵a,所以0.5秒太多了。

现在用matlab代码计算这个:

aa=sum(a.*a,2); bb=sum(b.*b,2); ab=a*b'; 
d = sqrt(abs(repmat(aa,[1 size(bb,1)]) + repmat(bb',[size(aa,1) 1]) - 2*ab));
[minz index]=min(d,[],2);

好的。Matlab代码使用的是(x-a)^2=x^2+a^2-2ab。

所以我的下一个尝试是做同样的事情。我删除了我自己的代码来进行同样的计算,但大约是1.2秒

然后,我尝试使用不同的外部库。第一次尝试是Eigen:

const int descrSize = 128;
MatrixXi a(a_size, descrSize);
MatrixXi b(b_size, descrSize);
MatrixXi ab(a_size, b_size);
unsigned char* dataPtr = matrixa;
for (int i=0; i<nframes; ++i)
{
for (int j=0; j<descrSize; ++j)
{
a(i,j)=(int)*dataPtr++;
}
}
unsigned char* vocPtr = matrixb;
for (int i=0; i<vocabulary_size; ++i)
{
for (int j=0; j<descrSize; ++j)
{
b(i,j)=(int)*vocPtr ++;
}
}
ab = a*b.transpose();
a.cwiseProduct(a);
b.cwiseProduct(b);
MatrixXi aa = a.rowwise().sum();
MatrixXi bb = b.rowwise().sum();
MatrixXi d = (aa.replicate(1,vocabulary_size) + bb.transpose().replicate(nframes,1) - 2*ab).cwiseAbs2();
int* index = NULL;
index = (int*)malloc(nframes*sizeof(int));
for (int i=0; i<nframes; ++i)
{
d.row(i).minCoeff(&index[i]);
}

这个特征码的成本约为1.2,仅用于表示:ab=a*b.transpose();

使用opencv的类似代码也被使用,并且ab的成本=A*b.transpose();为0.65秒。

所以,matlab能这么快做同样的事情,而我在C++中却做不到,这真的很烦人!当然,能够运行我的实验会很好,但我认为知识的缺乏才是真正让我烦恼的地方。我如何才能达到与Matlab相同的性能?欢迎任何形式的解决方案。我的意思是,任何外部库(如果可能的话是免费的),循环展开的东西,模板的东西,SSE入侵(我知道它们存在),缓存的东西。正如我所说,我的主要目的是增加我的知识,以便能够以更快的性能编写这样的想法。

提前感谢

编辑:David Hammen建议的更多代码。在进行任何计算之前,我将数组强制转换为int。这是代码:

void min_distance_loop(int** indexes, unsigned char* b, int b_size, unsigned char* a, int a_size)
{
const int descrSize = 128;
int* a_int;
int* b_int;
LARGE_INTEGER liStart;
LARGE_INTEGER liEnd;
LARGE_INTEGER liPerfFreq;
QueryPerformanceFrequency( &liPerfFreq );
QueryPerformanceCounter( &liStart );
a_int = (int*)malloc(a_size*descrSize*sizeof(int));
b_int = (int*)malloc(b_size*descrSize*sizeof(int));
for(int i=0; i<descrSize*a_size; ++i)
a_int[i]=(int)a[i];
for(int i=0; i<descrSize*b_size; ++i)
b_int[i]=(int)b[i];
QueryPerformanceCounter( &liEnd );
cout << "Casting time: " << (liEnd.QuadPart - liStart.QuadPart) / long double(liPerfFreq.QuadPart) << "s." << endl;
*indexes = (int*)malloc(a_size*sizeof(int));
int dataIndex=0;
int vocIndex=0;
int min_distance;
int distance;
int multiply;
/*unsigned char* dataPtr;
unsigned char* vocPtr;*/
int* dataPtr;
int* vocPtr;
for (int i=0; i<a_size; ++i)
{
min_distance = LONG_MAX;
for (int j=0; j<b_size; ++j)
{
distance=0;
dataPtr = &a_int[dataIndex];
vocPtr = &b_int[vocIndex];
for (int k=0; k<descrSize; ++k)
{
multiply = *dataPtr++-*vocPtr++;
distance += multiply*multiply;
// If the distance is greater than the previously calculated, exit
if (distance>min_distance)
break;
}
// if distance smaller
if (distance<min_distance)
{
min_distance = distance;
(*indexes)[i] = j;
}
vocIndex+=descrSize;
}
dataIndex+=descrSize;
vocIndex=0;
}
}

整个过程现在为0.6,开始时的铸造循环为0.001秒。也许我做错了什么?

第二版:艾根有什么事吗?当我寻找外部libs时,他们总是谈论Eigen和它们的速度。我做错了什么?这里有一个使用Eigen的简单代码,表明它不是那么快。也许我缺少一些配置或标志,或者。。。

MatrixXd A = MatrixXd::Random(1000, 1000);
MatrixXd B = MatrixXd::Random(1000, 500);
MatrixXd X;

此代码大约为0.9秒。

正如您所观察到的,您的代码主要由表示大约2.8e9算术运算的矩阵乘积组成。Yopu说,Matlab(或者更确切地说是高度优化的MKL)在大约0.05s内计算出它。这代表了57GFLOPS的速率,表明它不仅使用矢量化,还使用多线程。使用Eigen,您可以通过启用OpenMP进行编译来启用多线程(使用gcc的-fopenmp)。在我5年前的电脑(2.66Ghz Core2)上,使用浮点和4个线程,你的产品大约需要0.053秒,没有OpenMP需要0.16秒,所以你的编译标志一定有问题。总之,要获得最好的特征:

  • 以64位模式编译
  • 使用浮点(由于矢量化,双打的速度是原来的两倍)
  • 启用OpenMP
  • 如果你的CPU有超线程,那么要么禁用它,要么将OMP_NUM_THREADS环境变量定义为物理内核的数量(这非常重要,否则性能会非常糟糕!)
  • 如果有其他任务正在运行,则最好将OMP_NUM_THREADS减少为nb_cores-1
  • 使用最新的编译器,GCC、clang和ICC最好,MSVC通常较慢

在C++代码中,有一件事肯定会伤害到你,那就是它有大量的char到int转换。所谓船载,我的意思是多达2*2782*4000*128个字符到int的转换。那些charint的转换是缓慢的,非常缓慢。

您可以通过分配一对int数组(一个是2782*128,另一个是4000*128)来包含char* achar* b数组的转换为整数内容,从而将这种转换减少为(2782+4000)*128。使用这些int*阵列,而不是您的char*阵列。

另一个问题可能是您使用int而不是long。我不在windows上工作,所以这可能不适用。在我工作的机器上,int是32位,long现在是64位。32位就足够了,因为255*255*128<256*256*128=223

这显然不是问题所在

令人震惊的是,有问题的代码并没有计算出Matlab代码正在创建的2728乘4000的巨大数组。更引人注目的是,Matlab很可能使用doubles而不是int来实现这一点,而且它仍然在击败C/C++代码。

一个大问题是缓存。这个4000*128的数组对于一级缓存来说太大了,您要在这个大数组上迭代2782次。您的代码对内存的等待太多了。要克服这个问题,请使用b数组中较小的块,以便您的代码尽可能长时间地使用1级缓存。

另一个问题是优化CCD_ 17。我怀疑这实际上是一个dis优化。将if测试放在最内层的循环中通常是个坏主意。以尽可能快的速度通过内部产品。除了浪费计算之外,取消这个测试没有害处。有时,如果这样做可以删除最内部循环中的分支,那么最好进行明显不需要的计算。这就是其中之一只需取消此测试,您就可以解决问题试着那样做。

回到缓存问题,您需要去掉这个分支,以便可以将ab矩阵上的操作拆分为更小的块,一次不超过256行的块。这就是在两个现代英特尔芯片的L1缓存中,有多少行128个无符号字符可以放入其中。由于250将4000划分,因此考虑将b矩阵逻辑地划分为16个块。你可能想要形成2872乘4000的内积的大数组,但要分成小块。您可以重新添加if (distance>min_distance) break;,但要在块级别而不是逐字节级别添加。

您应该能够击败Matlab,因为它几乎可以肯定地使用doubles,但您可以使用无符号字符和int。

矩阵乘法通常对两个矩阵中的一个使用最坏的缓存访问模式,解决方案是转置其中一个矩阵,并使用专门的乘法算法来处理以这种方式存储的数据。

您的矩阵已被转置存储。通过将其转换为正常顺序,然后使用正常矩阵相乘,你的表现绝对是致命的。

编写您自己的矩阵乘法循环,将索引的顺序反转为第二个矩阵(这具有转换它的效果,而不会实际移动任何东西并破坏缓存行为)。并向编译器传递它为启用自动向量化提供的任何选项。