Прогон в foreach с изменением количества элементов

#c_sharp #многопоточность #циклы #foreach #parallel

У меня есть список. Необходимо прогнать все его элементы, вызвать в каждой итерации
функцию расчета и при каком-то условии, удалить элемент из цикла и продолжить итерации.

Вначале думал про for и если удаляем, делать i--, т.к. уменьшилось количество элементов
на 1, но это не сделать при параллельном прогоне, поэтому думаю сделать подобное в
foreach, чтобы не привязываться к индексу элемента. Но при изменение foreach выскакивает
исключение, что логично.

Можно ли как-нибудь прогнать параллельно по циклу и по условию, если что удалить
элемент. Чтобы это не сказывалась на других элементах в параллельных потоках?
    

Ответы

Ответ 1

Если удалять элемент из коллекции сразу же на месте не требуется — заведите еще одну
(потокобезопасную) коллекцию, элементы подлежащие удалению складывайте в нее, после
завершения первого цикла просто удалите из первой коллекции все элементы, присутствующие
во второй:

// заводим коллекцию
var forRemoving = new ConcurrentBag();
foreach (var item in items)
{
    ...
    // складываем в нее элементы подлежащие удалению
    forRemoving.Add(item);
}
// удаляем
foreach (var item in forRemoving)
    items.Remove(item);




Если для вашей задачи не принципиально удаление элементов из входной коллекции, а
подойдет также вариант с формированием новой коллекцией — можно воспользоваться функционалом
Parallel Linq:

var result = items.AsParallel()
                  .Where(item => condition(item))
                  .ToList();


Ответ 2

У меня есть список.


Если список - это именно List, то используй обычный цикл for.

Обращаю внимание, что удаление нескольких элементов через Remove неэффективно, поскольку
приведёт к квадратичной асимптотике (хвост всё время сдвигается). Правильным вариантом
было бы использовать метод RemoveAll. Ну либо написать обраотку списка в один проход
с ручным сдвигом конца по мере порверки.

Помогите разобраться с замедлением многопоточной (c++ threads) реализацией игры “Жизнь”

#cpp #многопоточность #cpp11 #parallel

Делая задание по реализации игры "Жизнь" Конвея, столкнулся с непонятной мне проблемой.
Ниже приведен листинг части программы. 

std::vector< std::thread > threads;
unsigned int NUM_THREADS = std::thread::hardware_concurrency();
std::cout << std::endl << "Avaliable threads: " << NUM_THREADS << std::endl;
std::cout << "C++ threads programm starts to work..." << std::endl;

std::vector changes(NUM_THREADS, 0);
std::vector borders;
for (unsigned int i = 0; i < NUM_THREADS; ++i)
    borders.push_back(i * field_height / NUM_THREADS);
borders.push_back(field_height);
threads.reserve(NUM_THREADS);
t1 = GetTickCount();
do
{
    std::fill(changes.begin(), changes.end(), 0);
    isChange = 0;
    {
        for (unsigned int k = 0; k < NUM_THREADS; ++k) {
            threads.emplace_back(
                [&borders, &changes, k, flag]() {
                //std::cerr << borders[k] << " - " << borders[k + 1] << std::endl;
                    for (unsigned int i = borders[k]; i < borders[k + 1]; ++i)
                    {
                        for (unsigned int j = 0; j < field_width; ++j)
                        {
                            changes[k] += changeNeibours(flag, i, j, changes[k]);
                        }
                    }
                }
            );
        }
        for (auto &thread : threads) {
            thread.join();
        }
        threads.clear();
    }
    if (flag == 0) 
        flag++;
    else 
        flag--;
    toCount++;
    for (auto change : changes)
        isChange += change;
    if (isChange == 0 && isSteady())
    {
        std::cout << "Steady state reached!" << std::endl;
        break;
    }
} while (toCount != generations);
t2 = GetTickCount();
parallel_time = t2 - t1;
std::cout << "C++ threads programm time: " << (double)parallel_time / 1000 << " seconds"
<< std::endl;
std::cout << "Amount of iterations done: " << iter << std::endl;
if (parallel_time)
    std::cout << "Acceleration: " << (double)consistent_time / parallel_time << std::endl;


Поле замкнуто в тор. Функция changeNeibours проверяет соседей и записывает результат
проверки. На каждом итерации мира (в каждом поколении) одна матрица является текущим
поколением (только читается для проверки), другая - будущим (только записывается результат
проверки каждой ячейки), они смсеняются по изменению переменной flag. Вектор changes
содержит информацию о количестве изменений состояния клеток в каждом потоке, после
завершения просчета нового поколения они складываются и проверяется на достижение стационарного
состояния.

Программа работает функционально правильно, то есть расчет N поколений проходит верно
(проверял, сравнивая со сторонними реализациями и просчетом на бумажке). Последовательная
реализация (полный код программы приложен) отрабатывает также верно и показывает аналогичный
результат.
Проблема заключается в том что на процессоре Intel Core i5 6600K, используя 4 потока,
параллельная реализация работает в 23 раза медленнее, нежели последовательная при одинаковых
входных данных:


Размер поля 100х100
Количество поколений 100000
В поле добавляется фигура Blinker для избежания стационарного состояния
Число потоков 4


При изменении входных данных на:


Размер поля 10000х10000
Количество поколений 5
В поле добавляется фигура Blinker для избежания стационарного состояния
Число потоков 4


Праллельная реализация замедляется всего в 2 раза. Для наглядности приведу скриншоты
моих замеров.




Вопрос заключается в том, что я написал такого, что программа так сильно замедляется
в параллельном исполнении.

На всякий случай приведу полный код программы:

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

int checkBorders(int index, int size);
void printField(int flag, std::string name);
int changeNeibours(int flag, int i, int j, int isChange);
bool isSteady();
void randomStartFilling();
void startFilling();


const unsigned int field_width = 10000;
const unsigned int field_height = 10000;
const unsigned int generations = 5;

bool fields[field_width][field_height];
bool new_fields[field_width][field_height];
bool fields2[field_width][field_height];
bool new_fields2[field_width][field_height];
bool temp_fields[field_width][field_height];

/*
* FUNCTION FOR CHECKING BORDERS
*/
int checkBorders(int index, int size)
{
    int result;
    if (index == -1)
    {
        result = size - 1;
        return result;
    }
    if (index == size)
    {
        result = 0;
        return result;
    }
    return index;
}

/*
* FUNCTION FOR CHECKING NEIGHBOURS
*/
int changeNeibours(int flag, int i, int j, int isChange)
{
    int neib = 0;
    switch (flag)
    {
    case 0: {
        if (fields[checkBorders(i + 1, field_height)][checkBorders(j - 1, field_width)]
== 1) //down left
            neib++;
        if (fields[checkBorders(i + 1, field_height)][j] == 1) //down
            neib++;
        if (fields[checkBorders(i + 1, field_height)][checkBorders(j + 1, field_width)]
== 1) //down right
            neib++;
        if (fields[checkBorders(i - 1, field_height)][checkBorders(j - 1, field_width)]
== 1) //up left
            neib++;
        if (fields[checkBorders(i - 1, field_height)][j] == 1) //up
            neib++;
        if (fields[checkBorders(i - 1, field_height)][checkBorders(j + 1, field_width)]
== 1) //up right
            neib++;
        if (fields[i][checkBorders(j - 1, field_width)] == 1) //left
            neib++;
        if (fields[i][checkBorders(j + 1, field_width)] == 1) //right
            neib++;
        if (fields[i][j] == 0)
        {
            if (neib == 3)
            {
                new_fields[i][j] = 1;
                isChange++;
            }
        }
        else
        {
            if ((neib < 2) || (neib > 3))
            {
                new_fields[i][j] = 0;
                isChange++;
            }
        }
    }
            break;
    case 1: {
        if (new_fields[checkBorders(i + 1, field_height)][checkBorders(j - 1, field_width)]
== 1) //down left
            neib++;
        if (new_fields[checkBorders(i + 1, field_height)][j] == 1) //down
            neib++;
        if (new_fields[checkBorders(i + 1, field_height)][checkBorders(j + 1, field_width)]
== 1) //down right
            neib++;
        if (new_fields[checkBorders(i - 1, field_height)][checkBorders(j - 1, field_width)]
== 1) //up left
            neib++;
        if (new_fields[checkBorders(i - 1, field_height)][j] == 1) //up
            neib++;
        if (new_fields[checkBorders(i - 1, field_height)][checkBorders(j + 1, field_width)]
== 1) //up right
            neib++;
        if (new_fields[i][checkBorders(j - 1, field_width)] == 1) //left
            neib++;
        if (new_fields[i][checkBorders(j + 1, field_width)] == 1) //right
            neib++;
        if (new_fields[i][j] == 0)
        {
            if (neib == 3)
            {
                fields[i][j] = 1;
                isChange++;
            }
        }
        else
        {
            if ((neib < 2) || (neib > 3))
            {
                fields[i][j] = 0;
                isChange++;
            }
        }
    }
            break;
    }
    return isChange;
}

/*
* FUNCTION FOR PRINTING FIELD
*/
void printField(int flag, std::string name) {
    std::fstream f;
    f.open(name + ".txt", std::ios::out);
    std::cout << std::endl;
    for (int i = 0; i < field_height; ++i) {
        for (int j = 0; j < field_width; ++j) {
            if (flag)
                f << fields[i][j];
            else
                f << new_fields[i][j];
        }
        f << std::endl;
    }
}

/*
* FUNCTION FOR CHECKING FOR STEADY STATE
*/
bool isSteady() {
    for (int i = 0; i < field_height; ++i) {
        for (int j = 0; j < field_width; ++j) {
            if (fields[i][j] != new_fields[i][j])
                return false;
        }
    }
    return true;
}

/*
* FUNCTION FOR RANDOM FILLING START WORLD
*/
void randomStartFilling() {
    for (int i = 0; i < field_width; i++)
    {
        for (int j = 0; j < field_height; j++)
        {
            fields[i][j] = rand() % 2 + 0;
            temp_fields[i][j] = fields[i][j];
            new_fields[i][j] = 0;
        }
    }
}

/*
* FUNCTION FOR FILLING START WORLD WITH FLICKER
*/
void startFilling() {
    for (int i = 0; i < 3; i++)
    {
        for (int j = 0; j < 3; j++)
        {
            if (i == 0 || i == 2)
                fields[i][j] = 0;
            else
                fields[i][j] = 1;
            temp_fields[i][j] = fields[i][j];
            new_fields[i][j] = 0;
        }
    }
    for (int i = 3; i < field_width; i++)
    {
        for (int j = 3; j < field_height; j++)
        {
            fields[i][j] = rand() % 2 + 0;
            temp_fields[i][j] = fields[i][j];
            new_fields[i][j] = 0;
        }
    }
}

/*
* MAIN
*/
int main(int argc, char *argv[])
{
    setlocale(LC_ALL, "Russian");
    int isChange;
    int toCount;
    int flag = 0;
    long t1 = 0;
    long t2 = 0;
    long consistent_time = 0, parallel_time = 0;
    int iter = 0;
    srand((unsigned int)time(NULL));
    //randomStartFilling();
    startFilling();

    /*
    * BEGIN OF CONSISTENT PART
    */
    system("cls");
    std::cout << "Amount of generations to calculate: " << generations << std::endl;
    std::cout << "World size: " << field_width << "x" << field_height << std::endl;
    std::cout << "Amount of iterations to be done: " << generations * field_width
* field_height << std::endl;
    std::cout << std::endl << "Consistent programm starts to work..." << std::endl;

    toCount = 0;
    t1 = GetTickCount();
    do
    {
        isChange = 0;
        {
            for (int i = 0; i < field_height; i++)
            {
                for (int j = 0; j < field_width; j++)
                {
                    isChange += changeNeibours(flag, i, j, isChange);
                    ++iter;
                }
            }
        }
        if (flag == 0) flag++;
        else flag--;
        toCount++;
        if (isChange == 0 && isSteady())
        {
            std::cout << "Steady state reached!" << std::endl;
            break;
        }
    } while (toCount != generations);
    t2 = GetTickCount();
    consistent_time = t2 - t1;
    std::cout << "Consistent programm time: " << (double)consistent_time / 1000 <<
" seconds" << std::endl;
    std::cout << "Amount of iterations done: " << iter << std::endl;

    //printField(flag, "consistent");

    /*
    * END OF CONSISTENT PART
    */

    for (int i = 0; i < field_height; i++)
    {
        for (int j = 0; j < field_width; j++)
        {
            fields2[i][j] = fields[i][j];
            fields[i][j] = temp_fields[i][j];
            new_fields[i][j] = 0;
        }
    }
    flag = 0;
    toCount = 0;
    iter = 0;

    /*
    * BEGIN OF C++ threads PART
    */

    std::vector< std::thread > threads;
    unsigned int NUM_THREADS = std::thread::hardware_concurrency();
    //NUM_THREADS = 2;
    std::cout << std::endl << "Avaliable threads: " << NUM_THREADS << std::endl;
    std::cout << "C++ threads programm starts to work..." << std::endl;

    std::vector iterations(NUM_THREADS, 0);
    std::vector changes(NUM_THREADS, 0);
    std::vector borders;
    for (unsigned int i = 0; i < NUM_THREADS; ++i)
        borders.push_back(i * field_height / NUM_THREADS);
    borders.push_back(field_height);
    threads.reserve(NUM_THREADS);
    t1 = GetTickCount();
    do
    {
        std::fill(changes.begin(), changes.end(), 0);
        std::fill(iterations.begin(), iterations.end(), 0);
        isChange = 0;
        {
            for (unsigned int k = 0; k < NUM_THREADS; ++k) {
                threads.emplace_back(
                    [&borders, &changes, &iterations, k, flag]() {
                    //std::cerr << borders[k] << " - " << borders[k + 1] << std::endl;
                        for (unsigned int i = borders[k]; i < borders[k + 1]; ++i)
                        {
                            for (unsigned int j = 0; j < field_width; ++j)
                            {
                                changes[k] += changeNeibours(flag, i, j, changes[k]);
                                ++iterations[k];
                            }
                        }
                    }
                );
            }
            for (auto &thread : threads) {
                thread.join();
            }
            threads.clear();
        }
        if (flag == 0) 
            flag++;
        else 
            flag--;
        toCount++;
        for (auto iteration : iterations)
            iter += iteration;
        for (auto change : changes)
            isChange += change;
        if (isChange == 0 && isSteady())
        {
            std::cout << "Steady state reached!" << std::endl;
            break;
        }
    } while (toCount != generations);
    t2 = GetTickCount();
    parallel_time = t2 - t1;
    std::cout << "C++ threads programm time: " << (double)parallel_time / 1000 <<
" seconds" << std::endl;
    std::cout << "Amount of iterations done: " << iter << std::endl;
    if (parallel_time)
        std::cout << "Acceleration: " << (double)consistent_time / parallel_time
<< std::endl;

    //printField(flag, "cppThreads");

    /*
    * END OF C++ threads PART
    */
    _getch();
    return 0;
}

    

Ответы

Ответ 1

Первое, что бросается в глаза - это создание и уничтожение потоков на каждой итерации.
Создавайте все потоки при старте программы.

Далее, при каждом вычислении все потоки пишут и читают элементы в массивах changes
и iterations. Несмотря на то, что они не конфликтуют между собой и обращаются исключительно
к элементу, соответствующему индексу потока, это все равно приводит к резкому замедлению
работы. Дело в том, что эти массивы имеют маленький размер и помещаются в одну (или
пару соседних) кэш-линий. Соответственно при изменении любого элемента будет производится
процедура синхронизации всей кэш линии на всех ядрах. Для устранения этой проблемы
нужно улучшить локальность данных, например сделать локальные переменные-счетчики и
записывать их финальное значение в общий массив при окончании итерации:

unsigned int rows_start{borders[k]};
unsigned int rows_end{borders[k + 1]};
unsigned long long changes_count{};
unsigned long long iterations_count{};
for(unsigned int row_index{rows_start}; row_index < rows_end; ++row_index)
{
    for (unsigned int col_index{}; col_index < field_width; ++col_index)
    {
        changes_count = changeNeibours(flag, row_index, col_index, changes_count);
        ++iterations_count;
    }
}
changes[k] = changes_count;
iterations[k] = iterations_count;


В таком варианте второй тест дает практически линейное увеличение производительности
с увеличением количества потоков.

Прогон в foreach с изменением количества элементов

У меня есть список. Необходимо прогнать все его элементы, вызвать в каждой итерации функцию расчета и при каком-то условии, удалить элемент из цикла и продолжить итерации.
Вначале думал про for и если удаляем, делать i--, т.к. уменьшилось количество элементов на 1, но это не сделать при параллельном прогоне, поэтому думаю сделать подобное в foreach, чтобы не привязываться к индексу элемента. Но при изменение foreach выскакивает исключение, что логично.
Можно ли как-нибудь прогнать параллельно по циклу и по условию, если что удалить элемент. Чтобы это не сказывалась на других элементах в параллельных потоках?

Ответ

Если удалять элемент из коллекции сразу же на месте не требуется — заведите еще одну (потокобезопасную) коллекцию, элементы подлежащие удалению складывайте в нее, после завершения первого цикла просто удалите из первой коллекции все элементы, присутствующие во второй:
// заводим коллекцию var forRemoving = new ConcurrentBag(); foreach (var item in items) { ... // складываем в нее элементы подлежащие удалению forRemoving.Add(item); } // удаляем foreach (var item in forRemoving) items.Remove(item);

Если для вашей задачи не принципиально удаление элементов из входной коллекции, а подойдет также вариант с формированием новой коллекцией — можно воспользоваться функционалом Parallel Linq:
var result = items.AsParallel() .Where(item => condition(item)) .ToList();