c - 如何在 C 中有效地累积数据数组

Question

问题是我有一个巨大的矩阵 A，并给定一个(相当大的)整数数组，例如，假设我的矩阵是: [0,0,0,0,0,0,0,0, 1,1,1,1,1,1,1,1, 2,2,2,2,2,2,2,2, 3,3,3,3,3,3,3,3, 4,4,4,4,4,4,4,4, .........]

整数数组为[0, 2, 4]

然后通过累加 [0,0,0,0,0,0,0,0], [2,2, 2,2,2,2,2,2],[4,4,4,4,4,4,4,4]

这是一个简单的问题，但是一个简单的 C 实现似乎很慢。累积大量行时尤其如此。

手动 loop_unrolling 似乎没有帮助。我不熟悉内联汇编，有什么建议吗？我想知道是否也有用于此类操作的已知库。

下面是我目前的实现:

void accumulateRows(int* js, int num_j, Dtype* B, int nrow, int ncol, int incRowB, Dtype* buffer){

int i = 0;
int num_accumulated_rows = (num_j / 8) * 8;
int remaining_rows = num_j - num_accumulated_rows;

// unrolling factor of 8, each time, accumulate 8 rows  
for(; i < num_accumulated_rows; i+=8){
    int r1 = js[i];
    int r2 = js[i+1];
    int r3 = js[i+2];
    int r4 = js[i+3];
    int r5 = js[i+4];
    int r6 = js[i+5];
    int r7 = js[i+6];
    int r8 = js[i+7];
    register Dtype* B1_row = &B[r1*incRowB];
    register Dtype* B2_row = &B[r2*incRowB];
    register Dtype* B3_row = &B[r3*incRowB];
    register Dtype* B4_row = &B[r4*incRowB];
    register Dtype* B5_row = &B[r5*incRowB];
    register Dtype* B6_row = &B[r6*incRowB];
    register Dtype* B7_row = &B[r7*incRowB];
    register Dtype* B8_row = &B[r8*incRowB];
    for(int j = 0; j < ncol; j+=1){
        register Dtype temp = B1_row[j] + B2_row[j] + B3_row[j] + B4_row[j];
        temp += B5_row[j] + B6_row[j] + B7_row[j] + B8_row[j];
        buffer[j] += temp;
    }
}

// left_over from the loop unrolling
for(; i < remaining_rows; i++){
    int r = js[i];
    Dtype* B_row = &B[r*incRowB];
    for(int i = 0; i < n; i++){
        buffer[i] += B_row[i];
    }
}

编辑 我认为这种累积在数据库中很常见，例如当我们要查询任何星期一、星期二等的总销售额时。

我知道 gcc 支持 Intel SSE，我希望了解如何将其应用到这个问题中，因为这是非常多的 SIMD

Answer 1

这是实现该功能的一种方法，以及一些关于进一步加速的建议

#include <stdlib.h> // size_t

typedef int Dtype;

// Note:
// following function assumes a 'contract' with the caller
//    that no entry in 'whichRows[]'
//    is larger than (number of rows in 'baseArray[][]' -1)

void accumulateRows(
    // describe source 2d array
    /* size_t numRows */ size_t numCols, Dtype BaseArray[][ numCols ],

    // describe row selector array
    size_t numSelectRows, size_t whichRows[ numSelectRows ],

    // describe result array
    Dtype resultArray[ numCols ] )
{
    size_t colIndex;
    size_t selectorIndex;

    // initialize resultArray to all 0
    for( colIndex = 0; colIndex < numCols; colIndex++ )
    {
        resultArray[colIndex] = 0;
    }

    // accumulate totals for each column of selected rows
    for( selectorIndex = 0; selectorIndex < numSelectRows; selectorIndex++ )
    {
        for( colIndex = 0; colIndex < numCols; colIndex++ )
        {
            resultArray[colIndex] += BaseArray[ whichRows[selectorIndex] ][colIndex];
        } // end for each column
    } // end for each selected row
}

#if 0
// you might want to unroll the "initialize resultArray" loop
//    by replacing the loop with
    resultArray[0] = 0;
    resultArray[1] = 0;
    resultArray[2] = 0;
    resultArray[3] = 0;
    resultArray[4] = 0;
    resultArray[5] = 0;
    resultArray[6] = 0;
    resultArray[7] = 0;
// however, that puts a constraint on the number of columns always being 8
#endif

#if 0
// you might want to unroll the 'sum of columns' loop by replacing the loop with
    resultArray[0] += BaseArray[ whichRows[selectorIndex] ][0];
    resultArray[1] += BaseArray[ whichRows[selectorIndex] ][1];
    resultArray[2] += BaseArray[ whichRows[selectorIndex] ][2];
    resultArray[3] += BaseArray[ whichRows[selectorIndex] ][3];
    resultArray[4] += BaseArray[ whichRows[selectorIndex] ][4];
    resultArray[5] += BaseArray[ whichRows[selectorIndex] ][5];
    resultArray[6] += BaseArray[ whichRows[selectorIndex] ][6];
    resultArray[7] += BaseArray[ whichRows[selectorIndex] ][7];
// however, that puts a constraint on the number of columns always being 8
#endif

#if 0
// on Texas Instrument DSPs ,
//    could use a #pragma to unroll the loop
//    or (better)
//    make use of the built-in loop table
//    to massively speed up the execution of the loop(s)
#endif