Как я могу ускорить этот цикл (в C)?

Question

Я пытаюсь распараллелить функцию свертки в C. Вот оригинальная функция, которая сворачивает два массива с 64-битными числами с плавающей запятой:

void convolve(const Float64 *in1,
              UInt32 in1Len,
              const Float64 *in2,
              UInt32 in2Len,
              Float64 *results)
{
    UInt32 i, j;

    for (i = 0; i < in1Len; i++) {
        for (j = 0; j < in2Len; j++) {
            results[i+j] += in1[i] * in2[j];
        }
    }
}

Для обеспечения параллелизма (без семафоров) я создал функцию, которая вычисляет результат для конкретной позиции в массиве results:

void convolveHelper(const Float64 *in1,
                    UInt32 in1Len,
                    const Float64 *in2,
                    UInt32 in2Len,
                    Float64 *result,
                    UInt32 outPosition)
{
    UInt32 i, j;

    for (i = 0; i < in1Len; i++) {
        if (i > outPosition)
            break;
        j = outPosition - i;
        if (j >= in2Len)
            continue;
        *result += in1[i] * in2[j];
    }
}

Проблема в том, что convolveHelper замедляет код примерно в 3,5 раза (при работе в одном потоке).

Есть идеи о том, как я могу ускорить convolveHelper, сохраняя при этом безопасность потоков?

Answer 1

Свертки во временной области становятся умножениями в области Фурье. Я предлагаю вам взять быструю библиотеку FFT (например, FFTW ) и использовать ее. Вы перейдете от O (n ^ 2) к O (n log n).

Алгоритмические оптимизации почти всегда превосходят микрооптимизации.

Answer 2

Самая очевидная вещь, которая могла бы помочь, - это предварительно вычислить начальный и конечный индексы цикла и удалить дополнительные тесты для i и j (и связанных с ними переходов). Это:

for (i = 0; i < in1Len; i++) {
   if (i > outPosition)
     break;
   j = outPosition - i;
   if (j >= in2Len)
     continue;
   *result += in1[i] * in2[j];
}

можно переписать как:

UInt32 start_i = (in2Len < outPosition) ? outPosition - in2Len + 1 : 0;
UInt32 end_i = (in1Len < outPosition) ? in1Len : outPosition + 1;

for (i = start_i; i < end_i; i++) {
   j = outPosition - i;
   *result += in1[i] * in2[j];
}

Таким образом, условие j >= in2Len никогда не выполняется, и циклический тест по сути является комбинацией тестов i < in1Len и i < outPosition.

Теоретически вы также можете избавиться от присвоения j и превратить i++ в ++i, но компилятор, вероятно, уже выполняет эти оптимизации для вас.

Answer 3

• Вместо двух операторов if в цикле можно вычислить правильные минимальные / максимальные значения для i перед циклом.

• Вы рассчитываете каждую позицию результата отдельно. Вместо этого вы можете разбить массив results на блоки и заставить каждый поток вычислять блок. Расчет для блока будет выглядеть как функция convolve.

Answer 4

Я наконец-то понял, как правильно рассчитать начальный / конечный индексы (совет, данный как Тайлер МакГенри , так и межджайный ):

if (in1Len > in2Len) {
    if (outPosition < in2Len - 1) {
        start = 0;
        end = outPosition + 1;
    } else if (outPosition >= in1Len) {
        start = 1 + outPosition - in2Len;
        end = in1Len;
    } else {
        start = 1 + outPosition - in2Len;
        end = outPosition + 1;
    }
} else {
    if (outPosition < in1Len - 1) {
        start = 0;
        end = outPosition + 1;
    } else if (outPosition >= in2Len) {
        start = 1 + outPosition - in2Len;
        end = in1Len;
    } else {
        start = 0;
        end = in1Len;
    }
}

for (i = start; i < end; i++) {
    *result = in1[i] * in2[outPosition - i];
}

К сожалению, предварительный расчет индексов не дает заметного уменьшения времени выполнения : (

Answer 5

Ответ лежит в простой математике и НЕ многопоточности (ОБНОВЛЕНО)

---

Вот почему ...

рассмотрим a b + a c

U можно оптимизировать как a * (b + c) (один умножение меньше)

В вашем случае in2Len ненужные умножения во внутреннем цикле . Который может быть устранен.

Следовательно, изменение кода следующим образом должно дать нам сверточное требование:

( ПРИМЕЧАНИЕ: Следующий код возвращает циклическую свертку , которую необходимо развернуть, чтобы получить линейную свертку .

void convolve(const Float64 *in1,
              UInt32 in1Len,
              const Float64 *in2,
              UInt32 in2Len,
              Float64 *results)
{
    UInt32 i, j;

    for (i = 0; i < in1Len; i++) {

        for (j = 0; j < in2Len; j++) {
            results[i+j] += in2[j];
        }

        results[i] = results[i] * in1[i];

    }
}

Это должно дать U скачок максимальной производительности больше, чем что-либо еще. Попробуйте наш и посмотрите !!

GOODLUCK !!

CVS @ 2600Hertz

Answer 6

Если ваши массивы не очень большие, использование потока вряд ли действительно поможет, так как затраты на запуск потока будут больше, чем стоимость циклов. Тем не менее, давайте предположим, что ваши массивы большие, а многопоточность - это чистый выигрыш. В этом случае я бы сделал следующее:

• Забудьте свой текущий convolveHelper, который слишком сложен и мало поможет.

• Разделить внутреннюю часть цикла на функцию резьбы. То есть просто сделай

  for (j = 0; j < in2Len; j++) {
      results[i+j] += in1[i] * in2[j];
  }
  

в свою собственную функцию, которая принимает i в качестве параметра наряду со всем остальным.

• У тела convolve просто запускаются темы. Для максимальной эффективности используйте семафор, чтобы быть уверенным, что вы никогда не создадите больше потоков, чем имеете ядра.

Answer 7

Позвольте помощнику сверток работать на больших наборах, вычисляя несколько результатов, используя короткий внешний цикл.

Ключ в распараллеливании - найти хороший баланс между распределением работы между потоками. Не используйте больше потоков, чем количество ядер ЦП.

Разделить работу равномерно между всеми потоками. С такой проблемой сложность работы каждого потока должна быть одинаковой.