Подтвердить что ты не робот

Как разбить массив на блоки

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

int getCellIndex(int ix, int iy, int iz) {
    return ix + (iy * numCellsX) + (iz * numCellsX * numCellsY);
}

Число ячеек в домене:

numCells = (numX + 2) * (numY + 2) * (numZ + 2)

Где numX/numY/numZ - количество ячеек в направлении X/Y/Z. +2 в каждом направлении - создавать прописные ячейки вокруг внешней стороны домена. Количество ячеек в каждом направлении определяется следующим образом:

numX = 5 * numY
numZ = numY/2
numY = userInput

Для каждой ячейки я хочу рассчитать новое значение для этой ячейки на основе ее значения соседей (т.е. трафарета), где соседи расположены выше, ниже, слева, справа, спереди и сзади. Тем не менее, я только хочу сделать этот расчет для ячеек, которые неплохо. У меня есть логический массив, который отслеживает, является ли ячейка плохим. Вот что выглядит в настоящее время вычисление:

for(int z = 1; z < numZ+1; z++) {
    for(int y = 1; y < numY+1; y++) {
        for(int x = 1; x < numX+1; x++) {
            if(!isBadCell[ getCellIndex(x,y,z) ] {
                // Do stencil Computation
            }
        }
    }
}

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

for(block : blocks) {
    if(isBadBlock[block]) {
        slowProcessBlock(block) // As above
    } else {
        fastVectorizedProcessBlock(block)
    }
}

ПРИМЕЧАНИЕ. Для того, чтобы блоки физически существовали, нет необходимости, то есть это может быть достигнуто путем изменения функции индексирования и использования разных индексов для цикла по массиву. Я открыт для всех, кто работает лучше всего.

Функция fastVectorizedProcessBlock() будет похожа на функцию slowProcessBlock(), но с оператором if remove (поскольку мы знаем, что он не содержит плохих ячеек) и прагмой векторизации.

Как я могу разделить свой домен на блоки, чтобы я мог это сделать? Кажется сложным, потому что: а) количество ячеек в каждом направлении не равно, б) нам нужно учитывать ячейки заполнения, так как мы никогда не должны пытаться вычислить их значение, так как это приведет к отсутствию доступа к памяти границ.

Как я могу обработать блоки, которые не содержат плохие ячейки, не используя оператор if?

EDIT:

Это идея, которую я изначально имел:

for(int i = 0; i < numBlocks; i++) { // use blocks of 4x4x4 = 64
    if(!isBadBlock[i]) {
        // vectorization pragma here
        for(int z = 0; z < 4; z++) {
            for(int y = 0; y < 4; y++) {
                for(int x = 0; x < 4; x++) {
                    // calculate stencil using getCellIndex(x,y,z)*i
                }
             }
         }
     } else {
         for(int z = 0; z < 4; z++) {
            for(int y = 0; y < 4; y++) {
                for(int x = 0; x < 4; x++) {
                    if(!isBadCell[i*getCellIndex(x,y,z)]) {    
                    // calculate stencil using getCellIndex(x,y,z)*i
                }
             }
         }
     }
 }

Теперь ячейки будут храниться в блоках, то есть все ячейки в первом блоке 4x4x4 будут сохранены в позиции 0-63, тогда все ячейки во втором блоке будут сохранены в позиции 64-127 и т.д.

Однако, я не думаю, что это будет работать, если значения numX/numY/numZ не являются добрыми. Например, что, если numY = 2, numZ = 1 и numX = 10? Для циклов for ожидается, что направление z будет по крайней мере на 4 ячейки. Есть ли хороший способ преодолеть это?

ОБНОВЛЕНИЕ 2 - Вот как выглядит трафаретное вычисление:

if ( isBadCell[ getCellIndex(x,y,z) ] ) {
  double temp = someOtherArray[ getCellIndex(x,y,z) ] +
                    1.0/CONSTANT/CONSTANT*
                    (
                      - 1.0 * cells[ getCellIndex(x-1,y,z) ]
                      - 1.0 * cells[ getCellIndex(x+1,y,z) ]
                      - 1.0 * cells[ getCellIndex(x,y-1,z) ]
                      - 1.0 * cells[ getCellIndex(x,y+1,z) ]
                      - 1.0 * cells[ getCellIndex(x,y,z-1) ]
                      - 1.0 * cells[ getCellIndex(x,y,z+1) ]
                      + 6.0 * cells[ getCellIndex(x,y,z) ]
                      );
  globalTemp += temp * temp;
  cells[ getCellIndex(x,y,z) ] += -omega * temp / 6.0 * CONSTANT * CONSTANT;
}
4b9b3361

Ответ 1

Где getCellIndex() извлекать значения numCellX и numCellY? Было бы лучше передать их как аргументы вместо того, чтобы полагаться на глобальные переменные, и сделать эту функцию static inline, чтобы позволить компилятору оптимизировать.

static line int getCellIndex(int ix, int iy, int iz, int numCellsX, numCellsY) {
    return ix + (iy * numCellsX) + (iz * numCellsX * numCellsY);
}

for (int z = 1; z <= numZ; z++) {
    for (int y = 1; y <= numY; y++) {
        for (int x = 1; x <= numX; x++) {
            if (!isBadCell[getCellIndex(x, y, z, numX + 2, numY + 2)] {
                // Do stencil Computation
            }
        }
    }
}

Вы также можете удалить все умножения с помощью некоторых локальных переменных:

int index = (numY + 2) * (numX + 2);  // skip top padding plane
for (int z = 1; z <= numZ; z++) {
    index += numX + 2;  // skip first padding row
    for (int y = 1; y <= numY; y++) {
        index += 1;   // skip first padding col
        for (int x = 1; x <= numX; x++, index++) {
            if (!isBadCell[index] {
                // Do stencil Computation
            }
        }
        index += 1;   // skip last padding col
    }
    index += numX + 2;   // skip last padding row
}

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

Если вы можете изменить формат логического массива для плохих ячеек, было бы полезно проложить строки до кратного 8 и использовать горизонтальное заполнение из 8 столбцов для улучшения выравнивания. Создание логического массива из массива битов позволяет проверять 8, 16, 32 или даже 64 ячейки за один раз с помощью одного теста.

Вы можете настроить указатель массива на использование координат, основанных на 0.

Вот как это работает:

int numCellsX = 8 + ((numX + 7) & ~7) + 8;
int numCellsY = 1 + numY + 1;
int numCellsXY = numCellsX * numCellsY;
// adjusted array_pointer
array_pointer = allocated_pointer + 8 + numCellsX + numCellsXY;
// assuming the isBadCell array is 0 based too.
for (int z = 0, indexZ = 0; z < numZ; z++, indexZ += numCellsXY) {
    for (int y = 0, indexY = indexZ; y < numY; y++, indexY += numCellsX) {
        for (int x = 0, index = indexY; x <= numX - 8; x += 8, index += 8) {
            int mask = isBadCell[index >> 3];
            if (mask == 0) {
                // let the compiler unroll computation for 8 pixels with
                for (int i = 0; i < 8; i++) {
                   // compute stencil value for x+i,y,z at index+i
                }
            } else {
                for (int i = 0; i < 8; i++, mask >>= 1) {
                    if (!(mask & 1)) {
                       // compute stencil value for x+i,y,z at index+i
                    }
                }
            }
        }
        int mask = isBadCell[index >> 3];
        for (; x < numX; x++, index++, mask >>= 1) {
            if (!(mask & 1)) {
                // compute stencil value for x,y,z at index
            }
        }
    }
}

EDIT:

Функция трафарета использует слишком много вызовов для getCellIndex. Вот как оптимизировать его, используя значение индекса, вычисленное в приведенном выше коде:

// index is the offset of cell x,y,z
// numCellsX, numCellsY are the dimensions of the plane
// numCellsXY is the offset between planes: numCellsX * numCellsY

if (isBadCell[index]) {
    double temp = someOtherArray[index] +
                1.0 / CONSTANT / CONSTANT *
                ( - 1.0 * cells[index - 1]
                  - 1.0 * cells[index + 1]
                  - 1.0 * cells[index - numCellsX]
                  - 1.0 * cells[index + numCellsX]
                  - 1.0 * cells[index - numCellsXY]
                  - 1.0 * cells[index + numCellsXY]
                  + 6.0 * cells[index]
                );
    cells[index] += -omega * temp / 6.0 * CONSTANT * CONSTANT;
    globalTemp += temp * temp;
}

precomputing &cells[index], поскольку указатель может улучшить код, но компилятор должен иметь возможность обнаруживать это общее подвыражение и генерировать эффективный код уже.

EDIT2:

Вот черепичный подход: вы можете добавить отсутствующие аргументы, большинство размеров предполагается глобальным, но вы, вероятно, должны передать указатель на структуру контекста со всеми этими значениями. Он использует isBadTile[] и isGoodTile[]: массивы булевых сообщений, если у данной плитки все ячейки плохие, а все ячейки - соответственно.

void handle_tile(int x, int y, int z, int nx, int ny, int nz) {
    int index0 = x + y * numCellsX + z * numCellsXY;
    // skipping a tile with all cells bad.
    if (isBadTile[index0] && nx == 4 && ny == 4 && nz == 4)
        return;
    // handling a 4x4x4 tile with all cells OK.
    if (isGoodTile[index0] && nx == 4 && ny == 4 && nz == 4) {
        for (int iz = 0; iz < 4; iz++) {
            for (int iy = 0; iy < 4; iy++) {
                for (int ix = 0; ix < 4; ix++) {
                    int index = index0 + ix + iy * numCellsX + iz + numCellsXY;
                    // Do stencil computation using `index`
                }
            }
        }
    } else {
        for (int iz = 0; iz < nz; iz++) {
            for (int iy = 0; iy < ny; iy++) {
                for (int ix = 0; ix < nx; ix++) {
                    int index = index0 + ix + iy * numCellsX + iz + numCellsXY;
                    if (!isBadCell[index] {
                        // Do stencil computation using `index`
                }
            }
        }
    }
}

void handle_cells() {
    int x, y, z;
    for (z = 1; z <= numZ; z += 4) {
        int nz = min(numZ + 1 - z, 4);
        for (y = 1; y <= numY; y += 4) {
            int ny = min(numY + 1 - y, 4);
            for (x = 1; x <= numX; x += 4) {
                int nx = min(numX + 1 - x, 4);
                handle_tile(x, y, z, nx, ny, nz);
            }
        }
    }
}

Вот функция для вычисления массива isGoodTile[]. Единственные смещения, правильно рассчитанные, соответствуют значениям x кратных 4 + 1, y и z меньше 3 от их максимальных значений.

Эта реализация является неоптимальной, так как может быть вычислено меньшее количество элементов. Неполные граничные плитки (менее 4 от края) могут быть помечены как нехорошие, чтобы пропустить хороший случай с одним случаем. Тест на плохие плитки мог бы работать для этих граничных плит, если массив isBadTile был правильно рассчитан для граничных плит, что в настоящее время не так.

void computeGoodTiles() {
    int start = 1 + numCellsX + numCellsXY;
    int stop = numCellsXY * numCellsZ - 1 - numCellsX - numCellsXY;

    memset(isGoodTile, 0, sizeof(*isGoodTile) * numCellsXY * numCellsZ);
    for (int i = start; i < stop; i += 4) {
        isGoodTile[i] = (isBadCell[i + 0] | isBadCell[i + 1] |
                         isBadCell[i + 2] | isBadCell[i + 3]) ^ 1;
    }
    for (int i = start; i < stop - 3 * numCellsX; i += 4) {
        isGoodTile[i] = isGoodTile[i + 0 * numCellsX] &
                        isGoodTile[i + 1 * numCellsX] &
                        isGoodTile[i + 2 * numCellsX] &
                        isGoodTile[i + 3 * numCellsX];
    }
    for (int i = start; i < stop - 3 * numCellsXY; i += 4) {
        isGoodTile[i] = isGoodTile[i + 0 * numCellsXY] &
                        isGoodTile[i + 1 * numCellsXY] &
                        isGoodTile[i + 2 * numCellsXY] &
                        isGoodTile[i + 3 * numCellsXY];
    }
}

void computeBadTiles() {
    int start = 1 + numCellsX + numCellsXY;
    int stop = numCellsXY * numCellsZ - 1 - numCellsX - numCellsXY;

    memset(isBadTile, 0, sizeof(*isBadTile) * numCellsXY * numCellsZ);
    for (int i = start; i < stop; i += 4) {
        isBadTile[i] = isBadCell[i + 0] & isBadCell[i + 1] &
                       isBadCell[i + 2] & isBadCell[i + 3];
    }
    for (int i = start; i < stop - 3 * numCellsX; i += 4) {
        isBadTile[i] = isBadTile[i + 0 * numCellsX] &
                       isBadTile[i + 1 * numCellsX] &
                       isBadTile[i + 2 * numCellsX] &
                       isBadTile[i + 3 * numCellsX];
    }
    for (int i = start; i < stop - 3 * numCellsXY; i += 4) {
        isBadTile[i] = isBadTile[i + 0 * numCellsXY] &
                       isBadTile[i + 1 * numCellsXY] &
                       isBadTile[i + 2 * numCellsXY] &
                       isBadTile[i + 3 * numCellsXY];
    }
}

Ответ 2

Хотя OP требует подхода с использованием блокировки, я бы предложил против него.

Вы видите, что каждая последовательная последовательность ячеек (1D-ячейки вдоль оси X) уже является таким блоком. Вместо того, чтобы упростить задачу, блокировка просто заменяет исходную проблему меньшими копиями фиксированного размера, повторяется снова и снова.

Проще говоря, блокирование вообще не помогает с реальной проблемой. Это не должно быть необходимой функцией решения.

Вместо этого я бы предложил избегать проблемы с корнем в целом - совсем по-другому.

Вы видите, вместо того, чтобы иметь флажок "плохая ячейка" для каждой ячейки, которую нужно протестировать (один раз для каждой ячейки, не менее), вы можете сохранить (отсортированный) список индексов плохих ячеек. Затем вы можете обрабатывать весь набор данных сразу, а затем цикл исправления по ячейкам, указанным в списке индексов плохих ячеек.

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

Итак, вот мое предложение:

#include <stdlib.h>
#include <errno.h>

typedef struct {
    /* Core cells in the state, excludes border cells */
    size_t   xsize;
    size_t   ysize;
    size_t   zsize;

    /* Index calculation: x + y * ystride + z * zstride */
    /* x is always linear in memory; xstride = 1 */
    size_t   ystride; /* = xsize + 2 */
    size_t   zstride; /* = ystride * (ysize + 2) */

    /* Cell data, points to cell (0,0,0) */
    double  *current;
    double  *previous;

    /* Bad cells */
    size_t   fixup_cells;  /* Number of bad cells */
    size_t  *fixup_index;  /* Array of bad cells' indexes */

    /* Dynamically allocated memory */
    void    *mem[3];
} lattice;

void lattice_free(lattice *const ref)
{
    if (ref) {
        /* Free dynamically allocated memory, */
        free(ref->mem[0]);
        free(ref->mem[1]);
        free(ref->mem[2]);
        /* then initialize/poison the contents. */
        ref->xsize = 0;
        ref->ysize = 0;
        ref->zsize = 0;
        ref->ystride = 0;
        ref->zstride = 0;
        ref->previous = NULL;
        ref->current = NULL;
        ref->fixup_cells = 0;
        ref->fixup_index = NULL;
        ref->mem[0] = NULL;
        ref->mem[1] = NULL;
        ref->mem[2] = NULL;
    }
}


int lattice_init(lattice *const ref, const size_t xsize, const size_t ysize, const size_t zsize)
{
    const size_t  xtotal = xsize + 2;
    const size_t  ytotal = ysize + 2;
    const size_t  ztotal = zsize + 2;
    const size_t  ntotal = xtotal * ytotal * ztotal;
    const size_t  double_bytes = ntotal * sizeof (double);
    const size_t  size_bytes = xsize * ysize * zsize * sizeof (size_t);

    /* NULL reference to the variable to initialize? */
    if (!ref)
        return EINVAL;

    /* Initialize/poison the lattice variable. */
    ref->xsize = 0;
    ref->ysize = 0;
    ref->zsize = 0;
    ref->ystride = 0;
    ref->zstride = 0;
    ref->previous = NULL;
    ref->current = NULL;
    ref->fixup_cells = 0;
    ref->fixup_index = NULL;
    ref->mem[0] = NULL;
    ref->mem[1] = NULL;
    ref->mem[2] = NULL;

    /* Verify size is nonzero */
    if (xsize < 1 || ysize < 1 || zsize < 1)
        return EINVAL;        

    /* Verify size is not too large */
    if (xtotal <= xsize || ytotal <= ysize || ztotal <= zsize ||
        ntotal / xtotal / ytotal != ztotal ||
        ntotal / xtotal / ztotal != ytotal ||
        ntotal / ytotal / ztotal != xtotal ||
        double_bytes / ntotal != sizeof (double) ||
        size_bytes / ntotal != sizeof (size_t))
        return ENOMEM;

    /* Allocate the dynamic memory needed. */
    ref->mem[0] = malloc(double_bytes);
    ref->mem[1] = malloc(double_bytes);
    ref->mem[2] = malloc(size_bytes);
    if (!ref->mem[0] || !ref->mem[1] || !ref->mem[2]) {
        free(ref->mem[2]);
        ref->mem[2] = NULL;
        free(ref->mem[1]);
        ref->mem[1] = NULL;
        free(ref->mem[0]);
        ref->mem[0] = NULL;
        return ENOMEM;
    }

    ref->xsize = xsize;
    ref->ysize = ysize;
    ref->zsize = zsize;

    ref->ystride = xtotal;
    ref->zstride = xtotal * ytotal;

    ref->current = (double *)ref->mem[0] + 1 + xtotal;
    ref->previous = (double *)ref->mem[1] + 1 + xtotal;

    ref->fixup_cells = 0;
    ref->fixup_index = (size_t *)ref->mem[2];

    return 0;
}

Обратите внимание, что я предпочитаю форму расчета индекса x + ystride * y + zstride * z по сравнению с x + xtotal * (y + ytotal * z), потому что два умножения в первом могут быть выполнены параллельно (в суперскалярном конвейере, на архитектурах, которые могут одновременно выполнять два несвязанных целочисленных умножения на одном ядре процессора), тогда как в последнем умножения должны быть последовательными.

Обратите внимание, что ref->current[-1 - ystride - zstride] относится к текущему значению ячейки в ячейке (-1, -1, -1), то есть диагональ пограничной ячейки из исходной ячейки (0, 0, 0). Другими словами, если у вас есть ячейка (x, y, z) в индексе i, то
  i-1 - ячейка at (x -1, y, z)
  i+1 - ячейка at (x +1, y, z)
  i-ystride - ячейка at (x, y -1, z)
  i+ystride - ячейка at (x, y +1, z)
  i-zstride - это ячейка at ( x, y, z -1)
  i+zstride - это ячейка at ( x, y, z -1)
  i-ystride - ячейка at (x, y -1, z)
  i-1-ystride-zstride - ячейка at ( x -1, y -1, z -1)
  i+1+ystride+zstride является ячейкой при ( x +1, y +1, z +1) и т.д.

Массив ref->fixup_index достаточно велик, чтобы отображать все ячейки, за исключением пограничных ячеек. Рекомендуется сохранять сортировку (или сортировать ее после ее создания), поскольку это помогает с локальностью кэша.

Если ваша решетка имеет периодические граничные условия, вы можете использовать шесть 2D-циклов, двенадцать 1D-петель и восемь копий для копирования первой и последней допустимых ячеек на границу, прежде чем запускать новое обновление.

Таким образом, цикл обновления:

  • Вычислить или заполнить границы в ->current.

  • Сменить ->current и ->previous.

  • Вычислить все ячейки для ->current, используя данные из ->previous.

  • Прокрутите индексы ->fixup_cells в ->fixup_index и пересчитайте соответствующие ячейки ->current.

Обратите внимание, что на шаге 3 вы можете сделать это линейно для всех индексов между 0 и xsize-1 + (ysize-1)*ystride + (zsize-1)*zstride, включительно; т.е. около 67% пограничных ячеек. Они относительно немного по сравнению со всем объемом, и наличие одной линейной петли, скорее всего, быстрее, чем пропуск через пограничные ячейки, особенно если вы можете векторизовать вычисление. (Что в этом случае нетривиально.)

Вы можете даже разделить работу на несколько потоков, предоставляя каждому потоку непрерывный набор индексов для работы. Поскольку вы читаете из ->previous и пишите в ->current, потоки не будут топтать друг друга, хотя может быть некоторый пинг-понг в кешетинге, если поток достигает конца его области, а другой находится в начале его область; из-за того, как данные ориентированы (и кэш-строки - это всего лишь несколько - обычно 2, 4 или 8 - ячейки в размерах), что пинг-понг не должен быть проблемой на практике. (Очевидно, никаких блокировок не требуется.)

Эта конкретная проблема никоим образом не является новой. Моделирование Conway Game of Life или модель Изинга с квадратной или кубической решеткой, а также реализация многих других моделей решетки включают ту же проблему (но часто с булевыми данными, а не с удвоениями и без "плохих ячеек" ).

Ответ 3

Я думаю, вы можете вложить пару подобных наборов петель. Что-то вроде этого:

for(int z = 1; z < numZ+1; z+=4) {
    for(int y = 1; y < numY+1; y+=4) {
        for(int x = 1; x < numX+1; x+=4) {
            if(!isBadBlock[ getBlockIndex(x>>2,y>>2,z>>2) ]) {
                for(int zz = z; zz < z + 4 && zz < numZ+1; zz++) {
                   for(int yy = y; yy < y + 4 && yy < numY+1; yy++) {
                      for(int xx = z; xx < x + 4 && xx < numX+1; xx++) {
                         if(!isBadCell[ getCellIndex(xx,yy,zz) ]) {
                             // Do stencil Computation
                            }
                        }
                    }
                }
            }
        }
    }
}

Ответ 4

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

#include <sys/types.h>
#define numX 256
#define numY 128
#define numZ 64
//Note the use of powers of 2 - it will simplify things a lot

int cells[numX][numY][numZ];

size_t getindex(size_t x, size_t y,size_t z){
  return (int*)&cells[x][y][z]-(int*)&cells[0][0][0];
}

Это выведет ячейки как:

[0,0,0][0,0,1][0,0,2]...[0,0,numZ-1]
[0,1,0][0,1,1][0,1,2]...[0,1,numZ-1]
...
[0,numY-1,0][0,numY-1,1]...[0,1,numZ-1]
...
[1,0,0][1,0,1][0,0,2]...[1,0,numZ-1]
[1,1,0][1,1,1][1,1,2]...[1,1,numZ-1]
...
[numX-1,numY-1,0][numX-1,numY-1,1]...[numX-1,numY-1,numZ-1]

So efficient loops would look like:

for(size_t x=0;x<numX;x++)
  for(size_t y=0;y<numY;y++)
    for(size_t z=0;z<numZ;z++)
      //vector operations on z values

Но если вы хотите разбить его на блоки 4x4x4, вы можете просто использовать 3d-массив из блоков 4x4x4, например:

#include <sys/types.h>
#define numX 256 
#define numY 128
#define numZ 64

typedef int block[4][4][4];
block blocks[numX][numY][numZ];
//add a compiler specific 64 byte alignment to  help with cache misses?

size_t getblockindex(size_t x, size_t y,size_t z){
  return (block *)&blocks[x][y][z]-(block *)&blocks[0][0][0];
}

Я переупорядочил индексы до x, y, z, чтобы я мог держать их прямо в моей голове, но убедитесь, что вы заказываете их так, чтобы последний был тем, который вы используете в серии из ваших самых внутренних для петель.