De / Interleave массив быстро в C #

Question

Я ищу самый быстрый способ де / чередования буфера. Если говорить более конкретно, я имею дело с аудиоданными, поэтому я пытаюсь оптимизировать время, которое я трачу на разделение / объединение каналов и буферов FFT.

В настоящее время я использую цикл for с 2 индексными переменными для каждого массива, поэтому только операции плюс, но все проверки управляемого массива не будут сравниваться с методом указателя C.

Мне нравятся методы Buffer.BlockCopy и Array.Copy, которые сокращают много времени при обработке каналов, но у массива нет возможности иметь собственный индексатор.

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

Я не хочу небезопасного решения, хотя, судя по всему, это может быть единственным способом оптимизировать этот тип операций.

Спасибо.

Вот что я сейчас делаю:

private float[][] DeInterleave(float[] buffer, int channels)
{
    float[][] tempbuf = new float[channels][];
    int length = buffer.Length / channels;
    for (int c = 0; c < channels; c++)
    {
        tempbuf[c] = new float[length];
        for (int i = 0, offset = c; i < tempbuf[c].Length; i++, offset += channels)
            tempbuf[c][i] = buffer[offset];
    }
    return tempbuf;
}

Answer 1

Я провел несколько тестов, и вот код, который я тестировал:

delegate(float[] inout)
{ // My Original Code
    float[][] tempbuf = new float[2][];
    int length = inout.Length / 2;
    for (int c = 0; c < 2; c++)
    {
        tempbuf[c] = new float[length];
        for (int i = 0, offset = c; i < tempbuf[c].Length; i++, offset += 2)
            tempbuf[c][i] = inout[offset];
    }
}
delegate(float[] inout)
{ // jerryjvl's recommendation: loop unrolling
    float[][] tempbuf = new float[2][];
    int length = inout.Length / 2;
    for (int c = 0; c < 2; c++)
        tempbuf[c] = new float[length];
    for (int ix = 0, i = 0; ix < length; ix++)
    {
        tempbuf[0][ix] = inout[i++];
        tempbuf[1][ix] = inout[i++];
    }

}
delegate(float[] inout)
{ // Unsafe Code
    unsafe
    {
        float[][] tempbuf = new float[2][];
        int length = inout.Length / 2;
        fixed (float* buffer = inout)
            for (int c = 0; c < 2; c++)
            {
                tempbuf[c] = new float[length];
                float* offset = buffer + c;
                fixed (float* buffer2 = tempbuf[c])
                {
                    float* p = buffer2;
                    for (int i = 0; i < length; i++, offset += 2)
                        *p++ = *offset;
                }
            }
    }
}
delegate(float[] inout)
{ // Modifying my original code to see if the compiler is not as smart as i think it is.
    float[][] tempbuf = new float[2][];
    int length = inout.Length / 2;
    for (int c = 0; c < 2; c++)
    {
        float[] buf = tempbuf[c] = new float[length];
        for (int i = 0, offset = c; i < buf.Length; i++, offset += 2)
            buf[i] = inout[offset];
    }
}

и результаты: (размер буфера = 2 ^ 17, число итераций по времени на тест = 200)

Average for test #1:      0.001286 seconds +/- 0.000026
Average for test #2:      0.001193 seconds +/- 0.000025
Average for test #3:      0.000686 seconds +/- 0.000009
Average for test #4:      0.000847 seconds +/- 0.000008

Average for test #1:      0.001210 seconds +/- 0.000012
Average for test #2:      0.001048 seconds +/- 0.000012
Average for test #3:      0.000690 seconds +/- 0.000009
Average for test #4:      0.000883 seconds +/- 0.000011

Average for test #1:      0.001209 seconds +/- 0.000015
Average for test #2:      0.001060 seconds +/- 0.000013
Average for test #3:      0.000695 seconds +/- 0.000010
Average for test #4:      0.000861 seconds +/- 0.000009

Я получал похожие результаты при каждом тесте. Очевидно, что небезопасный код - самый быстрый, но я был удивлен, увидев, что CLS не может понять, что он может отбросить проверки индекса при работе с неровным массивом. Может быть, кто-то может придумать больше способов оптимизировать мои тесты.

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

delegate(float[] inout)
{
    float[][] tempbuf = new float[2][];
    int length = inout.Length / 2;
    float[] tempbuf0 = tempbuf[0] = new float[length];
    float[] tempbuf1 = tempbuf[1] = new float[length];

    for (int ix = 0, i = 0; ix < length; ix++)
    {
        tempbuf0[ix] = inout[i++];
        tempbuf1[ix] = inout[i++];
    }
}

Результаты также представляют собой сравнительный тест № 4 с разницей в 1%. До сих пор тест №4 - мой лучший путь.

Как я сказал jerryjvl, проблема в том, что CLS не индексирует проверку входного буфера, поскольку добавление второй проверки (&& offset

Редактировать 2: Я запускал тесты раньше в IDE, так что результаты за пределами:

2^17 items, repeated 200 times
******************************************
Average for test #1:      0.000533 seconds +/- 0.000017
Average for test #2:      0.000527 seconds +/- 0.000016
Average for test #3:      0.000407 seconds +/- 0.000008
Average for test #4:      0.000374 seconds +/- 0.000008
Average for test #5:      0.000424 seconds +/- 0.000009

2^17 items, repeated 200 times
******************************************
Average for test #1:      0.000547 seconds +/- 0.000016
Average for test #2:      0.000732 seconds +/- 0.000020
Average for test #3:      0.000423 seconds +/- 0.000009
Average for test #4:      0.000360 seconds +/- 0.000008
Average for test #5:      0.000406 seconds +/- 0.000008


2^18 items, repeated 200 times
******************************************
Average for test #1:      0.001295 seconds +/- 0.000036
Average for test #2:      0.001283 seconds +/- 0.000020
Average for test #3:      0.001085 seconds +/- 0.000027
Average for test #4:      0.001035 seconds +/- 0.000025
Average for test #5:      0.001130 seconds +/- 0.000025

2^18 items, repeated 200 times
******************************************
Average for test #1:      0.001234 seconds +/- 0.000026
Average for test #2:      0.001319 seconds +/- 0.000023
Average for test #3:      0.001309 seconds +/- 0.000025
Average for test #4:      0.001191 seconds +/- 0.000026
Average for test #5:      0.001196 seconds +/- 0.000022

Test#1 = My Original Code
Test#2 = Optimized safe loop unrolling
Test#3 = Unsafe code - loop unrolling
Test#4 = Unsafe code
Test#5 = My Optimized Code

Похоже, что развертывание петли не выгодно. Мой оптимизированный код по-прежнему мой лучший способ, с разницей всего в 10% по сравнению с небезопасным кодом. Если бы я только мог сказать компилятору, что (i

Answer 2

Может быть, вы раскроете свой лучший ответ:

delegate(float[] inout)
{
    unsafe
    {
        float[][] tempbuf = new float[2][];
        int length = inout.Length / 2;

        fixed (float* buffer = inout)
        {
            float* pbuffer = buffer;

            tempbuf[0] = new float[length];
            tempbuf[1] = new float[length];

            fixed (float* buffer0 = tempbuf[0])
            fixed (float* buffer1 = tempbuf[1])
            {
                float* pbuffer0 = buffer0;
                float* pbuffer1 = buffer1;

                for (int i = 0; i < length; i++)
                {
                    *pbuffer0++ = *pbuffer++;
                    *pbuffer1++ = *pbuffer++;
                }
            }
        }
    }
}

Это может привести к еще большей производительности.

Answer 3

Это не даст вам значительного прироста производительности (я приблизительно измерил 20% на моей машине), но вы могли бы рассмотреть развертывание цикла для обычных случаев. Если в большинстве случаев у вас относительно ограниченное количество каналов:

static private float[][] Alternative(float[] buffer, int channels)
{
    float[][] result = new float[channels][];
    int length = buffer.Length / channels;
    for (int c = 0; c < channels; c++)
        result[c] = new float[length];

    int i = 0;
    if (channels == 8)
    {
        for (int ix = 0; ix < length; ix++)
        {
            result[0][ix] = buffer[i++];
            result[1][ix] = buffer[i++];
            result[2][ix] = buffer[i++];
            result[3][ix] = buffer[i++];
            result[4][ix] = buffer[i++];
            result[5][ix] = buffer[i++];
            result[6][ix] = buffer[i++];
            result[7][ix] = buffer[i++];
        }
    }
    else
        for (int ix = 0; ix < length; ix++)
            for (int ch = 0; ch < channels; ch++)
                result[ch][ix] = buffer[i++];


    return result;
}

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

Answer 4

Поскольку для этого нет встроенной функции, использование индексов массива - самая быстрая операция, о которой вы только могли подумать. Индексаторы и подобные решения только усугубляют ситуацию, вводя вызовы методов и не позволяя оптимизатору JIT оптимизировать связанные проверки.

В любом случае, я думаю, что ваш текущий метод является самым быстрым, не unsafe решением, которое вы могли бы использовать. Если производительность действительно имеет значение для вас (что обычно имеет значение в приложениях обработки сигналов), вы можете сделать все это в unsafe C # (что достаточно быстро, вероятно, сравнимо с C) и обернуть ее в метод, который вы вызываете из ваши безопасные методы.

Answer 5

Я думаю, что многие читатели зададутся вопросом, почему вы не хотите небезопасного решения для обработки звука. Это тот тип вещей, который напрашивается на горячую оптимизацию, и я бы очень огорчился, зная, что это происходит через виртуальную машину.