Задать вопрос
Egorian
@Egorian

Как ускорить этот код(клеточный автомат)?

Делаю движок для воспроизведения клеточных автоматов. Всё было достаточно хорошо, пока я не начал обрабатывать большие карты, где проседает скорость обсчета. И я без понятия как увеличить её.
Кода многовато, но он чрезвычайно простой.
Программа работает так: я создаю 2Д массив наполненный объектами класса Cell. Для массива есть класс Grid. Приведу код, а значение некоторых переменных и методов объясню позже.
// Grid.h
private:
    Rule* rule;
    int ** _ranges;
    int _width;
    int _height;
    /// обычный массив, только с перегруженым оператором []
    OverflowArray<OverflowArray<Cell>> _grid;
    std::vector<Cell*> _cells_to_draw;
    ThreadPool * pool;
    std::vector<Cell*> * storage;

    static int ** DivideGridIntoZones(int zones, int length);
    void CalculateZone(int id,const int range[2]);
    void UpdateCellsStates(const int * range);
public:
    Grid(int width, int height, int threadsAmount,sf::RenderTarget& screen);
    ~Grid();
    Cell& GetCell(int x, int y);
    void CalculateCells(int threads=4);
    void DisplayCells();

// Cell.h
enum CellState{
    Empty,
    Alive,
};
class Cell{
private:
    CellState _state;
    // сначала программа считает следующее состояние клетки, а в следующем цикле их обновляет
   // для этого и нужен _nextState
    CellState _nextState;
public:
    Cell();
    Cell(int x,int y,CellState state=CellState::Empty){...};
    bool IsAlive(){ return _state==CellState::Alive;}
    CellState & GetNextState()  { return _nextState;}
    CellState & GetState()  { return _state;}
    void SetNextState(CellState state){ _nextState = state;}
};


Запускаю бесконечный цикл, где постоянно запускается CalculateCells, в котором ведутся все обсчеты.
Теперь объясню зачем _ranges и DivideGridIntoZones(int zones, int length)
Grid::Grid(...){
 // DivideGridIntoZones делит ширину на промежутки, количество которых равно threadsAmount
 // код многопоточный, поэтому каждый поток обрабатывает свою зону
// _width=100, threadsAmount = 2
// _ranges = [[0,50],[50,100]]
_ranges = DivideGridIntoZones(threadsAmount,_width);
}

// Grid.cpp
// _width=1000, _height=1000, threadsAmount = 8
// с такими параметрами на большой карте с большим количеством живых клеток выжимаю 8 кадров
void Grid::CalculateCells(const int threadsAmount) {
    // этот код более-менее держит нагрузку и при таких параметрах выполняется за 0.01с
   // тут и происходит замена значения _state на значение _nextState
    for(int i=0;i<threadsAmount;i++){
       pool->AddJob( [this,i](){this->UpdateCellsStates(_ranges[i]);});
    }
    pool->WaitAll();
    //
    // моя главная проблема. При очень большом паттерне выполняется за 0.032с
    for(int i=0;i<threadsAmount;i++){
        pool->AddJob( [this,i]() {this->CalculateZone(i,_ranges[i]);});
    }
    pool->WaitAll();

 // some code
}

Сейчас приведу код CalculateZone, но сначала объясню значение storage. В этот вектор потоки записывают указатели на клетки, которые нужно нарисовать после обсчета, Затем я соединяю storage в _cells_to_draw.
void Grid::CalculateZone(int id,const int range[2]){
    std::vector<Cell*> temp;
    Cell * cell;
    // вот  и использование range
    for(int y=0;y<_height;y++){
        for(int x=range[0];x<range[1];x++){
            cell = &GetCell(x,y);
            // сохраняю в переменную temp, чтобы её сохранить в storage
            if(cell->GetState()==CellState::Alive){
                temp.emplace_back(cell);
            }
            if(!_isPaused) {
                    // тут считается следующее состояние клетки и изменяется
                   // код приводить не буду, но скажу, что тут есть 9 обращений к массиву
                    rule->Execute(cell,x,y);
            }
            }
        }
    // сохраняю storage
    storage[id] = temp;
}

Думаю, кода хватит.
Моя цель - достичь 20 кадров в секунду, но я не имею понятия, что для этого надо сделать.
На гитхабе нашёл хороший пример, который показывает отличную производительность
Вот его обработка правил. Автор использует таблицу указателей, сдвиг битов.
Неужели это настолько повышает производительность? Или у меня в коде есть закралась огромная ошибка, которая жрет производительность? Могу только предположить, что где-то я чересчур много копирую значения, хотя я следовал правилу "Больше указателей богу указателей", да и уже раз 10 код пересматривал
  • Вопрос задан
  • 116 просмотров
Подписаться 1 Простой 4 комментария
Решения вопроса 1
@MarkusD Куратор тега C++
все время мелю чепуху :)
В общем смысле, как я вижу по твоему коду, ты вляпался в True Sharing, попутно обмазавшись Cache Misses и окончательно убив свою производительность с помощью неоправданно огромного размера клеток.

8Б на клетку, состояние которой может поместиться в 1Б, это действительно огромный размер.
enum CellState : uint8_t уменьшит размер состояния с 4Б до 1Б. А еще этот тип стоит переименовать, т.к. это не CellState, а что-то относящееся к поведению клетки. А вот CellState будет выглядеть так:
// Renamed from `CellState`.
enum CellBehavior : uint8_t
{
    Empty,
    Alive,
};

struct CellState final
{
	CellBehavior	current_behavior : 4;
	CellBehavior	next_behavior : 4;
};

Это позволяет уменьшить размер клетки до 1 байта.

Данные оперативной памяти процессор подтягивает к себе во внутренний кэш. Кэшей у процессора много и все они связаны. Кэш процессора поделен на линии, работа с которыми синхронизируется между ядрами процессора. Вот именно тут появляется два термина: False cacheline sharing и True cacheline sharing. Если "False", то обрабатываемые разными ядрами данные разделены в разные кэш-линии. Когда "True" - требуемые разным ядрам данные находятся в одной кэш-линии и привет синхронизация. А это ой как медленно.

В каждом процессоре сегодня сидит гадалка, которая предсказывает какие тебе надо подтянуть данные из RAM в CPU Cache. Выборка из RAM - это довольно долгая процедура, поэтому нужна гадалка чтобы предсказать что судьбой твоего алгоритма предначертано выбрать на следующем этапе. Бывает что гадалка ошибается и тогда твой лагоритм встает в синхронизацию до завершения нужной выборки из памяти. А это - еще медленнее чем синхронизация по кэш-линиям. Это называется промахом по кэшу - cache miss.
К счастью, это не гадалка виновата в своей ошибке, а ты просто неправильно написал лагоритм. Вот чтобы из лагоритма сделать алгоритм, следует озаботиться чтобы он был более лоялен к гадалке и кэшу процессора.

Докину еще немного полезной информации.
Сходи к Адаму Мартину и к Unity, посмотри на парадигму ES/ESP/ECS. Изучи DOD. Попробуй реорганизацию из твоего текущего потока сущностей с полями в потоки полей сущностей. Переделай батчинг обработки клеток так, чтобы данные не синхронизировались между ядрами процессора.
Возможно тебе еще поможет понимание подхода Out of line, т.к. там хорошо объясняется почему очень большие объекты при их поточной обработке - это не очень дружественно кэшу процессора.
Еще сюда можно добавить информацию о автоматической векторизации. Это позволит задействовать SIMD инструкции для твоего кода. DOD очень элегантно ложится для обработки твоих клеток SIMD командами.

Я тут крайне сумбурно накидал, только чтобы дать тебе направления. Кое-чего я даже не написал, но ты обязательно зацепишь все неописанное когда будешь изучать то, что я описал. Думаю, ты уже видишь, в какой объем выльется весь этот материал, если писать его в удобном понятном формате и раскрывая каждую тему.
Ответ написан
Комментировать
Пригласить эксперта
Ваш ответ на вопрос

Войдите, чтобы написать ответ

Похожие вопросы