Как выполнить итерацию одновременно двух массивов, которые не находятся на равном расстоянии друг от друга, оптимизированным образом?

Скажем, мне нужно умножить два массива, такие как A[MAX_BUFFER] и B[MAX_BUFFER]MAX_BUFFER = 256).

По какой-то причине каждое значение B[MAX_BUFFER] вычисляется с фиксированной скоростью управления (например, 8), поскольку каждое значение будет подвергаться тяжелой обработке.

Позже мне нужно умножить друг друга на C[MAX_BUFFER], учитывая (введенный) другой интервал. Итак, с A с 256 значениями, я получу B с переменным размером (32 в этом примере, поскольку скорость управления равна 8).

Вот пример кода:

#include <iostream>
#include <math.h>

#define MAX_BUFFER 256

double HeavyFunction(double value) {
    if (value == 0) return 0.0;

    return pow(10.0, value); // heavy operations on value...
}

int main()
{    
    int blockSize = 256;
    int controlRate = 8;

    double A[MAX_BUFFER];
    double B[MAX_BUFFER];
    double C[MAX_BUFFER];

    // fill A
    for (int sampleIndex = 0; sampleIndex < blockSize; sampleIndex++) {
        A[sampleIndex] = sampleIndex;
    }

    // fill B (control rated)
    int index = 0;
    for (int sampleIndex = 0; sampleIndex < blockSize; sampleIndex += controlRate, index++) {
        B[index] = HeavyFunction(index);
    }

    // calculate C
    for (int sampleIndex = 0; sampleIndex < blockSize; sampleIndex++) {     
        C[sampleIndex] = A[sampleIndex] + B[sampleIndex / 8];

        std::cout << C[sampleIndex] << std::endl;
    }
}

Мне нужна производительность, поскольку я буду обрабатывать множество этих операций параллельно, отправляя много данных за 1 секунду (например, 44100 сэмплов, разделенных в blockSize <= MAX_BUFFER).

Я бы хотел избежать ветвления (например, if) и разделения (как в приведенном выше примере), которые не являются операциями, подобными процессору (обработка большого количества данных).

В предыдущем примере это представит "бесполезную" операцию N для sampleIndex / 8 * N; вещи, если я назову эту процедуру для миллионов образцов ...

Как бы вы реорганизовали этот код необычным и легким способом для процессора?

Любой достойный оптимизирующий компилятор заменит /8 сдвигом вправо. Посмотрите на сгенерированную сборку, прежде чем беспокоиться о «накладных расходах» индексации массива, вполне возможно, что их вообще нет.

Mat 26.10.2018 09:21

@Mat: Я бы хотел узнать, насколько компилятор будет умнее меня :) Пример этого сдвига вправо? Так я могу реализовать вручную?

markzzz 26.10.2018 09:23

Не похоже, что вам нужен этот уровень оптимизации (наивное выполнение 44100 умножений за 1 секунду тривиально, если у вас нет серьезных аппаратных ограничений), но вы, вероятно, выиграете больше от использования инструкций SSE / AVX для выполнения операции, чем от беспокоясь о ветвлении.

Collin Dauphinee 26.10.2018 09:36

@markzzz, вам не нужно делать правильный сдвиг самостоятельно, компилятор, скорее всего, достаточно умен, чтобы сделать это за вас.

Jabberwocky 26.10.2018 09:36

Опять же: я знаю, что компилятор умный! Хотелось бы увидеть, как может быть код без «деления» и «если» :)

markzzz 26.10.2018 09:42

@CollinDauphinee: это если я назову это на 44100 * nVoices * nParam * nFilter.

markzzz 26.10.2018 09:43

@markzzz посмотрите сгенерированный код сборки, деления на 8 нет, деление сдвигом, посмотрите инструкцию sar eax, 3.

Jabberwocky 26.10.2018 09:47

почему HeavyFunction (0) = 0?

UmNyobe 26.10.2018 09:55

Это всего лишь пример, не беспокойтесь об этом :)

markzzz 26.10.2018 09:58

@Jabberwocky: это зависит от обстоятельств. Если controlRate можно переключить как «const», да, но если он динамический, он будет использовать не sar, а деление: godbolt.org/z/CJrsQr

markzzz 26.10.2018 10:31
1
10
76
2

Ответы 2

Я думаю, что оптимизатор может выполнить эту работу в одиночку, но вы можете развернуть цикл, чтобы избежать разделения:

// calculate C
const max = blockSize / 8;
int j = 0;
for (int i = 0; i != max; ++i) {
    const auto b = B[i];
    C[j] = A[j] + b; std::cout << C[j] << std::endl; ++j;
    C[j] = A[j] + b; std::cout << C[j] << std::endl; ++j;
    C[j] = A[j] + b; std::cout << C[j] << std::endl; ++j;
    C[j] = A[j] + b; std::cout << C[j] << std::endl; ++j;
    C[j] = A[j] + b; std::cout << C[j] << std::endl; ++j;
    C[j] = A[j] + b; std::cout << C[j] << std::endl; ++j;
    C[j] = A[j] + b; std::cout << C[j] << std::endl; ++j;
    C[j] = A[j] + b; std::cout << C[j] << std::endl; ++j;
}

две проблемы: 1 - целочисленное деление, которое будет сдвигом, это последнее, что вы хотите здесь оптимизировать. 2 - Компилятор не будет оптимизировать чтение из A и B или запись в C. Учитывая, что мы используем многопоточную парадигму, это может вызвать побочные эффекты. Итак, вы действительно хотите сохранить B [i] в ​​локальной константной переменной, если она вам понадобится 8 раз.

UmNyobe 26.10.2018 10:29

@UmNyobe: (1) - это то, что хочет OP, я тоже не думаю, что это проблема ;-). (2) Для многопоточности мы должны иметь синхронизацию с мьютексом или использовать атомарную переменную. Теперь я использую локальную переменную для B[i] для возможного сглаживания, если используется в функции.

Jarod42 26.10.2018 10:41

How do you iterate simultaneously two array that are not equally spaced in a optimized way?

Краткий ответ: сосредоточьтесь на HeavyFunction, избегайте обмена ненужным между потоками.

К сожалению, ваш пример не подходит под вопрос. Массивы

double A[MAX_BUFFER];
double B[MAX_BUFFER];
double C[MAX_BUFFER];

выделяются в стеке простым перемещением указателя стека, поэтому можно сказать, что они очень похожи на один непрерывный массив.

Даже если бы это было не так, современные кеши настолько сложны, что попытка микрооптимизации может привести к снижению производительности.

Предполагая, что у вас есть

BUFFER_SIZE = 1024 * 1024 * 1024;
std::vector<double> A(MAX_BUFFER);
std::vector<double> B(MAX_BUFFER);

Хорошее улучшение - это

std::vector<double> C{A};
for (int i = 0; i < blockSize/controlRate; i++) { 
     const double b = B[i];
     int indexStart = i*controlRate;
     for(int j = 0 ; j < controlRate; ++j){
        Cprime[indexStart+j] += b;
     }

}

Вы читаете A один раз (блоками), B один раз (по одному двойному за раз) и обращаетесь к C столько же времени.

controlRate и blockSize могут отличаться. Не могу гарантировать, что они всегда точно разделяются. Например, что если у меня blockSize 250 и контрольная скорость 8? Он пропустит некоторые шаги ...
markzzz 29.10.2018 19:21

Другие вопросы по теме