预期AVX向量化的C ++手动展开条件和比常规代码慢

问题描述

test_euclid_ask.h（只需要阅读2个函数：euclid_slow，euclid_fast）

#pragma once
#include "included.h"

double
euclid_slow(int n,double* data1,double* data2,int* mask1,int* mask2,const double weight[])
{
    double result = 0.0;
    double totalWeight = 0;    
   
    for (int i = 0; i < n; i++) {
        if (mask1[i] && mask2[i]) {
            double term = data1[i] - data2[i];
            result += weight[i] * term * term;
            totalWeight += weight[i];
        }
    }
        
    if (totalWeight==0) return 0; 
    return result / totalWeight;
}

double
euclid_fast(int n,const double weight[])
{
    double result = 0.0;
    double totalWeight = 0;
    double subResult[4] = { 0. };
    double subTweight[4] = { 0. };
    double subDiff[4] = { 0. };
    double subWeight[4] = { 0. };
    double subMask[4] = { 0. };
    int nstep4 = n - n % 4;

    for (int i = 0; i < nstep4; i += 4) {
        subMask[0] = mask1[i] && mask2[i];
        subMask[1] = mask1[i + 1] && mask2[i + 1];
        subMask[2] = mask1[i + 2] && mask2[i + 2];
        subMask[3] = mask1[i + 3] && mask2[i + 3];
        if (!(subMask[0] || subMask[1] || subMask[2] || subMask[3])) continue;            

        subDiff[0] = data1[i] - data2[i];
        subDiff[1] = data1[i + 1] - data2[i + 1];
        subDiff[2] = data1[i + 2] - data2[i + 2];
        subDiff[3] = data1[i + 3] - data2[i + 3];

        subDiff[0] *= subDiff[0];
        subDiff[1] *= subDiff[1];
        subDiff[2] *= subDiff[2];
        subDiff[3] *= subDiff[3];

        subWeight[0] = weight[i] * subMask[0];
        subWeight[1] = weight[i + 1] * subMask[1];
        subWeight[2] = weight[i + 2] * subMask[2];
        subWeight[3] = weight[i + 3] * subMask[3];

        subTweight[0] += subWeight[0];
        subTweight[1] += subWeight[1];
        subTweight[2] += subWeight[2];
        subTweight[3] += subWeight[3];

        subResult[0] += subWeight[0] * subDiff[0];
        subResult[1] += subWeight[1] * subDiff[1];
        subResult[2] += subWeight[2] * subDiff[2];
        subResult[3] += subWeight[3] * subDiff[3];
    }

    for (int i = nstep4; i < n; i++) {
        if (mask1[i] && mask2[i]) {
            double term = data1[i] - data2[i];
            result += weight[i] * term * term;
            totalWeight += weight[i];
        }
    }
        
    result += subResult[0] + subResult[1] + subResult[2] + subResult[3];
    totalWeight += subTweight[0] + subTweight[1] + subTweight[2] + subTweight[3];
        
    //cout << "end fast\n";
    if (!totalWeight) return 0; 
    return result / totalWeight;
}

void test_euclid_ask()
{   
    const int MAXN = 10000000,MINN = 1000000;
    double* data1,* data2;
    int* mask1,* mask2;
    double* dataPro1,* dataPro2;
    int* maskPro1,* maskPro2;
    double *weight,* weightPro;

    //***********
    data1 = new double[MAXN + MINN + 1];
    data2 = new double[MAXN + MINN + 1];
    mask1 = new int[MAXN + MINN + 1];
    mask2 = new int[MAXN + MINN + 1];
    dataPro1 = new double[MAXN + MINN + 1];
    dataPro2 = new double[MAXN + MINN + 1];
    maskPro1 = new int[MAXN + MINN + 1];
    maskPro2 = new int[MAXN + MINN + 1];

    // ******
    weight = new double[MAXN + MINN + 1];
    weightPro = new double[MAXN + MINN + 1];
    MyTimer timer;
    int n;
    double guess1,guess2,tmp,total1 = 0,total2 = 0,prev1 = 0,prev2 = 0;
    
    for (int t = 5000; t < 6000; t++) {
        if (t <= 5000) n = t;
        else n = MINN + rand() % (MAXN - MINN);        
        cout << n << "\n";
        
        int index = 0;
        for (int i = 0; i < n; i++) {
            weight[i] = int64(randomed()) % 100;
            data1[i] = int64(randomed()) % 100;
            data2[i] = int64(randomed()) % 100;
            mask1[i] = rand() % 10;
            mask2[i] = rand() % 10;
        }
        memcpy(weightPro,weight,n * sizeof(double));
        memcpy(dataPro1,data1,n * sizeof(double));
        memcpy(dataPro2,data2,n * sizeof(double));        
        memcpy(maskPro1,mask1,n * sizeof(int));
        memcpy(maskPro2,mask2,n * sizeof(int));
                    
        //****
        int tmp = flush_cache();    // do something to ensure the cache does not contain test data
        cout << "ignore this " << tmp << "\n";
        
        timer.startCounter();
        guess1 = euclid_slow(n,weight);
        tmp = timer.getCounterMicro();
        total1 += tmp;
        cout << "time slow = " << tmp << " us\n";

        timer.startCounter();
        guess2 = euclid_fast(n,dataPro1,dataPro2,maskPro1,maskPro2,weightPro);
        tmp = timer.getCounterMicro();
        total2 += tmp;
        cout << "time fast = " << tmp << " us\n";

        bool ok = fabs(guess1 - guess2) <= 0.1;
        if (!ok) {
            cout << "error at N = " << n << "\n";
            exit(-1);
        }        
        cout << "\n";
    }

    cout << "slow speed = " << (total1 / 1000) << " ms\n";
    cout << "fast speed = " << (total2 / 1000) << " ms\n";
}

基本上，该函数计算两个数组之间的一种欧式距离：

result = sum(weight[i] * (data1[i] - data2[i])^2)

，但仅在两个值均可用的位置（mask1[i]==0表示将被忽略，与mask2相同）。普通代码在函数euclid_slow中。

因此，我试图通过一次处理4个元素来改进代码，希望SSE / AVX可以加快速度。但是，结果保持相同或更慢（使用 g ++ -O3 -march = native ）或变得慢40％（使用Visual Studio 2019编译器，发布模式（x64），-O2，AVX2启用） 。我尝试了-O2和-O3，结果相同。

编译器比我编写的代码进行了更好的优化。但是我怎样才能使其更快呢？

编辑：测试程序here

的代码

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）