Более быстрый способ преобразования вектора векторов в один непрерывный вектор с противоположным порядком хранения
У меня есть std::vector<std::vector<double>> который я пытаюсь преобразовать в один непрерывный вектор как можно быстрее. Мой вектор имеет форму примерно 4000 x 50.
Проблема в том, что иногда мне нужен мой выходной вектор в смежном главном столбце (просто конкатенация внутренних векторов моего 2-го входного вектора), а иногда мне нужен мой выходной вектор в смежном основном ряду, что фактически требует транспонирования.
Я обнаружил, что наивный цикл for довольно быстр для преобразования в главный вектор-столбец:
auto to_dense_column_major_naive(std::vector<std::vector<double>> const & vec)
-> std::vector<double>
{
auto n_col = vec.size();
auto n_row = vec[0].size();
std::vector<double> out_vec(n_col * n_row);
for (size_t i = 0; i < n_col; ++i)
for (size_t j = 0; j < n_row; ++j)
out_vec[i * n_row + j] = vec[i][j];
return out_vec;
}
Но очевидно, что подобный подход очень медленный для построчного преобразования из-за всех ошибок кеша. Так что для построчного преобразования я подумал, что стратегия блокировки для продвижения локальности кэша может быть моей лучшей ставкой:
auto to_dense_row_major_blocking(std::vector<std::vector<double>> const & vec)
-> std::vector<double>
{
auto n_col = vec.size();
auto n_row = vec[0].size();
std::vector<double> out_vec(n_col * n_row);
size_t block_side = 8;
for (size_t l = 0; l < n_col; l += block_side) {
for (size_t k = 0; k < n_row; k += block_side) {
for (size_t j = l; j < l + block_side && j < n_col; ++j) {
auto const &column = vec[j];
for (size_t i = k; i < k + block_side && i < n_row; ++i)
out_vec[i * n_col + j] = column[i];
}
}
}
return out_vec;
}
Это значительно быстрее, чем простой цикл для преобразования основной строки, но все же почти на порядок медленнее, чем простой цикл главной колонки для моего входного размера.
![enter image description here]()
Мой вопрос заключается в том, существует ли более быстрый подход к преобразованию вектора (основного столбца) векторов двойных чисел в один смежный основной вектор-строку? Я пытаюсь понять, каким должен быть предел скорости этого кода, и поэтому задаюсь вопросом, не упускаю ли я что-то очевидное. Мое предположение состояло в том, что блокирование даст мне гораздо большее ускорение, чем, по-видимому, на самом деле.
Диаграмма была сгенерирована с использованием QuickBench (и несколько проверена локально на моем компьютере с помощью GBench) с этим кодом: (Clang 7, С++ 20, -O3)
auto to_dense_column_major_naive(std::vector<std::vector<double>> const & vec)
-> std::vector<double>
{
auto n_col = vec.size();
auto n_row = vec[0].size();
std::vector<double> out_vec(n_col * n_row);
for (size_t i = 0; i < n_col; ++i)
for (size_t j = 0; j < n_row; ++j)
out_vec[i * n_row + j] = vec[i][j];
return out_vec;
}
auto to_dense_row_major_naive(std::vector<std::vector<double>> const & vec)
-> std::vector<double>
{
auto n_col = vec.size();
auto n_row = vec[0].size();
std::vector<double> out_vec(n_col * n_row);
for (size_t i = 0; i < n_col; ++i)
for (size_t j = 0; j < n_row; ++j)
out_vec[j * n_col + i] = vec[i][j];
return out_vec;
}
auto to_dense_row_major_blocking(std::vector<std::vector<double>> const & vec)
-> std::vector<double>
{
auto n_col = vec.size();
auto n_row = vec[0].size();
std::vector<double> out_vec(n_col * n_row);
size_t block_side = 8;
for (size_t l = 0; l < n_col; l += block_side) {
for (size_t k = 0; k < n_row; k += block_side) {
for (size_t j = l; j < l + block_side && j < n_col; ++j) {
auto const &column = vec[j];
for (size_t i = k; i < k + block_side && i < n_row; ++i)
out_vec[i * n_col + j] = column[i];
}
}
}
return out_vec;
}
auto to_dense_column_major_blocking(std::vector<std::vector<double>> const & vec)
-> std::vector<double>
{
auto n_col = vec.size();
auto n_row = vec[0].size();
std::vector<double> out_vec(n_col * n_row);
size_t block_side = 8;
for (size_t l = 0; l < n_col; l += block_side) {
for (size_t k = 0; k < n_row; k += block_side) {
for (size_t j = l; j < l + block_side && j < n_col; ++j) {
auto const &column = vec[j];
for (size_t i = k; i < k + block_side && i < n_row; ++i)
out_vec[j * n_row + i] = column[i];
}
}
}
return out_vec;
}
auto make_vecvec() -> std::vector<std::vector<double>>
{
std::vector<std::vector<double>> vecvec(50, std::vector<double>(4000));
std::mt19937 mersenne {2019};
std::uniform_real_distribution<double> dist(-1000, 1000);
for (auto &vec: vecvec)
for (auto &val: vec)
val = dist(mersenne);
return vecvec;
}
static void NaiveColumnMajor(benchmark::State& state) {
// Code before the loop is not measured
auto vecvec = make_vecvec();
for (auto _ : state) {
benchmark::DoNotOptimize(to_dense_column_major_naive(vecvec));
}
}
BENCHMARK(NaiveColumnMajor);
static void NaiveRowMajor(benchmark::State& state) {
// Code before the loop is not measured
auto vecvec = make_vecvec();
for (auto _ : state) {
benchmark::DoNotOptimize(to_dense_row_major_naive(vecvec));
}
}
BENCHMARK(NaiveRowMajor);
static void BlockingRowMajor(benchmark::State& state) {
// Code before the loop is not measured
auto vecvec = make_vecvec();
for (auto _ : state) {
benchmark::DoNotOptimize(to_dense_row_major_blocking(vecvec));
}
}
BENCHMARK(BlockingRowMajor);
static void BlockingColumnMajor(benchmark::State& state) {
// Code before the loop is not measured
auto vecvec = make_vecvec();
for (auto _ : state) {
benchmark::DoNotOptimize(to_dense_column_major_blocking(vecvec));
}
}
BENCHMARK(BlockingColumnMajor);
Ответы
Ответ 1
Прежде всего, я съеживаюсь всякий раз, когда что-то квалифицируется как "очевидно". Это слово часто используется, чтобы скрыть недостатки в одном вычете.
Но очевидно, что подобный подход очень медленный для построчного преобразования из-за всех ошибок кеша.
Я не уверен, что должно быть очевидным: будет показано покадровое преобразование или оно будет медленным из-за пропадания кэша. В любом случае, я нахожу это не очевидным. В конце концов, здесь есть два аспекта кэширования, не так ли? Один для чтения и один для письма? Давайте посмотрим на код с точки зрения чтения:
row_major_naive
for (size_t i = 0; i < n_col; ++i)
for (size_t j = 0; j < n_row; ++j)
out_vec[j * n_col + i] = vec[i][j];
Последовательные чтения из vec - это чтение непрерывной памяти: vec[i][0] затем vec[i][1] и т.д. Очень хорошо для кэширования. Итак... кеш отсутствует? Медленный? :) Может быть, не так очевидно.
Тем не менее, есть что почерпнуть из этого. Требование только неправильно, утверждая "очевидно". Есть не локальные проблемы, но они возникают в конце письма. (Последовательные записи компенсируются пробелом в 50 double значений.) И эмпирическое тестирование подтверждает медлительность. Так может быть решение состоит в том, чтобы переключиться на то, что считается "очевидным"?
перевернутый ряд
for (size_t j = 0; j < n_row; ++j)
for (size_t i = 0; i < n_col; ++i)
out_vec[j * n_col + i] = vec[i][j];
Все, что я сделал здесь, это повернул циклы. Буквально поменяйте местами порядок этих двух строк кода, затем измените отступ. Теперь последовательные чтения потенциально повсюду, так как они читают из разных векторов. Тем не менее, последовательные записи теперь в смежные блоки памяти. В каком-то смысле мы находимся в той же ситуации, что и раньше. Но так же, как и раньше, нужно измерить производительность, прежде чем предполагать "быстрый" или "медленный".
NaiveColumnMajor: 3,4 секунды
NaiveRowMajor: 7,7 секунд
FlippedRowMajor: 4,2 секунды
BlockingRowMajor: 4,4 секунды
BlockingColumnMajor: 3,9 секунды
По-прежнему медленнее, чем наивная колонна основных преобразований. Тем не менее, этот подход не только быстрее наивного майора строк, но и быстрее, чем блокирование мажора строк. По крайней мере, на моем компьютере (с использованием gcc -O3 и, очевидно: повторение тысяч раз). Пробег может отличаться. Я не знаю, что скажут необычные инструменты профилирования. Дело в том, что иногда проще, тем лучше.
Для забавы я сделал тест, где размеры меняются местами (изменяясь от 50 векторов 4000 элементов до 4000 векторов 50 элементов). Все методы пострадали таким образом, но "NaiveRowMajor" получил самый большой удар. Стоит отметить, что "перевернутая мажорная строка" отстала от версии блокировки. Так что, как и следовало ожидать, лучший инструмент для работы зависит от того, что именно работа.
NaiveColumnMajor: 3,7 секунды
NaiveRowMajor: 16 секунд
FlippedRowMajor: 5,6 секунд
BlockingRowMajor: 4,9 секунды
BlockingColumnMajor: 4,5 секунды
(Кстати, я также попробовал трюк с переворотом на версии с блокировкой. Изменение было небольшим - около 0,2 - и противоположно изменению наивной версии. То есть "перевернутая блокировка" была медленнее, чем "блокировка" для вопроса Векторы 50 на 4000, но быстрее для моего варианта 4000 на 50. Точная настройка может улучшить результаты.)
Обновление: я провел немного больше тестирования с помощью трюка с переворотом на блокирующей версии. Эта версия имеет четыре цикла, поэтому "переворот" не так прост, как в случае только двух циклов. Похоже, что замена порядка двух внешних циклов плохо влияет на производительность, а замена внутренних двух циклов - это хорошо. (Изначально я сделал и то и другое и получил смешанные результаты.) Когда я поменял местами только внутренние циклы, я измерил 3,8 секунды (и 4,1 секунды в сценарии 4000-из-50), что сделало эту опцию лучшим в ряду в моих тестах,
рядный гибрид
for (size_t l = 0; l < n_col; l += block_side)
for (size_t i = 0; i < n_row; ++i)
for (size_t j = l; j < l + block_side && j < n_col; ++j)
out_vec[i * n_col + j] = vec[j][i];
(После замены внутренних петель я объединил средние циклы.)
Что касается теории, стоящей за этим, я бы предположил, что это равносильно попытке записать один блок кеша за раз. Как только блок записан, попробуйте повторно использовать векторы (vec[j]), прежде чем они будут извлечены из кэша. После того, как вы исчерпали эти исходные векторы, переходите к новой группе исходных векторов, снова записывая полные блоки за раз.
Ответ 2
Я только что добавил две функции параллельной версии вещей
#include <ppl.h>
auto ppl_to_dense_column_major_naive(std::vector<std::vector<double>> const & vec)
-> std::vector<double>
{
auto n_col = vec.size();
auto n_row = vec[0].size();
std::vector<double> out_vec(n_col * n_row);
size_t vecLen = out_vec.size();
concurrency::parallel_for(size_t(0), vecLen, [&](size_t i)
{
size_t row = i / n_row;
size_t column = i % n_row;
out_vec[i] = vec[row][column];
});
return out_vec;
}
auto ppl_to_dense_row_major_naive(std::vector<std::vector<double>> const & vec)
-> std::vector<double>
{
auto n_col = vec.size();
auto n_row = vec[0].size();
std::vector<double> out_vec(n_col * n_row);
size_t vecLen = out_vec.size();
concurrency::parallel_for(size_t(0), vecLen, [&](size_t i)
{
size_t column = i / n_col;
size_t row = i % n_col;
out_vec[i] = vec[row][column];
});
return out_vec;
}
и дополнительные коды для всех из них
template< class _Fn, class ... Args >
auto callFncWithPerformance( std::string strFnName, _Fn toCall, Args&& ...args )
{
auto start = std::chrono::high_resolution_clock::now();
auto toRet = toCall( std::forward<Args>(args)... );
auto end = std::chrono::high_resolution_clock::now();
std::chrono::duration<double> diff = end - start;
std::cout << strFnName << ": " << diff.count() << " s" << std::endl;
return toRet;
}
template< class _Fn, class ... Args >
auto second_callFncWithPerformance(_Fn toCall, Args&& ...args)
{
std::string strFnName(typeid(toCall).name());
auto start = std::chrono::high_resolution_clock::now();
auto toRet = toCall(std::forward<Args>(args)...);
auto end = std::chrono::high_resolution_clock::now();
std::chrono::duration<double> diff = end - start;
std::cout << strFnName << ": " << diff.count() << " s";
return toRet;
}
#define MAKEVEC( FN, ... ) callFncWithPerformance( std::string( #FN ) , FN , __VA_ARGS__ )
int main()
{
//prepare vector
auto vec = make_vecvec();
std::vector< double > vecs[]
{
std::vector<double>(MAKEVEC(to_dense_column_major_naive, vec)),
std::vector<double>(MAKEVEC(to_dense_row_major_naive, vec)),
std::vector<double>(MAKEVEC(ppl_to_dense_column_major_naive, vec)),
std::vector<double>(MAKEVEC(ppl_to_dense_row_major_naive, vec)),
std::vector<double>(MAKEVEC(to_dense_row_major_blocking, vec)),
std::vector<double>(MAKEVEC(to_dense_column_major_blocking, vec)),
};
//system("pause");
return 0;
}
а вот ниже результат этих
Отладка x64
to_dense_column_major_naive: 0,166859 с
to_dense_row_major_naive: 0,192488 с
ppl_to_dense_column_major_naive: 0,0557423 с
ppl_to_dense_row_major_naive: 0,0514017 с
to_dense_column_major_blocking: 0,118465 с
to_dense_row_major_blocking: 0.117732 с
Отладка x86
to_dense_column_major_naive: 0,15242 с
to_dense_row_major_naive: 0,158746 с
ppl_to_dense_column_major_naive: 0,0534966 с
ppl_to_dense_row_major_naive: 0,0484076 с
to_dense_column_major_blocking: 0,111217 с
to_dense_row_major_blocking: 0.107727 с
Выпуск x64
to_dense_column_major_naive: 0,000874 с
to_dense_row_major_naive: 0,0011973 с
ppl_to_dense_column_major_naive: 0,0054639 с
ppl_to_dense_row_major_naive: 0,0012034 с
to_dense_column_major_blocking: 0.0008023 с
to_dense_row_major_blocking: 0,0010282 с
Выпуск x86
to_dense_column_major_naive: 0,0007156 с
to_dense_row_major_naive: 0,0012538 с
ppl_to_dense_column_major_naive: 0,0053351 с
ppl_to_dense_row_major_naive: 0,0013022 с
to_dense_column_major_blocking: 0.0008761 с
to_dense_row_major_blocking: 0,0012404 с
Вы совершенно правы, для параллелизма это слишком маленький набор данных.
А также это слишком маленькие работы.
Хотя я буду пост для кого-то еще, чтобы ссылаться на эти функции.
