Как оптимизировать количество обращений к памяти / пропуски кеша для этого массива программы decimate/downsample?
Недавно меня спросили о куске кода для децимации / уменьшения выборки массива "на месте". Эта функция "decimation" принимает массив целых и сохраняет запись по четному индексу i в массиве по индексу i/2, Это делает это для всех записей в массиве.
Это переместит все четные записи в исходном массиве в первую половину массива. Затем остальная часть массива может быть инициализирована равной 0. Общий результат - это массив, который сохранил все четные записи индекса в исходном массиве (путем перемещения их в первую половину), а вторая половина массива равна 0. Это по-видимому, используется для уменьшения частоты сигналов при обработке сигналов.
Код выглядит примерно так:
void decimate (vector<int>& a) {
int sz = a.size();
for (int i =0; i < sz; i++) {
if (i%2 == 0) {
a[i/2] = a[i];
}
}
for (int i =(sz-1)/2; i < sz; i++) a[i] = 0;
}
После предложения базовых улучшений, которые хранят определенные переменные в регистрах, я не могу найти дальнейшего пути его оптимизации, но не уверен, что это невозможно сделать.
Есть ли способы оптимизировать схему доступа к памяти в цикле для повышения производительности кэша? Или какие-либо другие способы оптимизировать основные операции копирования сжатия / понижающей дискретизации массива в первой половине? (например, путем векторизации для платформ, которые его поддерживают)
for (int i =0; i < sz; i++) {
if (i%2 == 0) {
a[i/2] = a[i];
}
}
Существуют ли какие-либо преобразования цикла (такие как разбиение на листы / майнинг), которые могут привести к высокоэффективному коду для такого десятичного цикла?
РЕДАКТИРОВАТЬ: Есть несколько различных способов, предложенных в ответах ниже, которые, кажется, используют преимущества арифметики memset / fill или указателя для повышения эффективности скорости. Этот вопрос в основном сосредоточен на том, существуют ли четко определенные преобразования циклов, которые могут значительно улучшить ошибки локальности или кэширования (например, если бы это было гнездо цикла с двумя циклами, можно было бы изучить разбиение на циклы для оптимизации ошибок кэширования)
5 ответов
У вас есть такой массив:
0 1 2 3 4 5 6 7 8 9
Вы хотите закончить с этим:
0 2 4 6 8 0 0 0 0 0
Я бы сделал это так:
void decimate (vector<int>& a) {
size_t slow = 1, fast = 2;
// read the first half, write the first quarter
size_t stop = (a.size()+1)/2;
while (fast < stop) {
a[slow++] = a[fast];
fast += 2;
}
// read and clear the second half, write the second quarter
stop = a.size();
while (fast < stop) {
a[slow++] = a[fast];
a[fast++] = 0;
a[fast++] = 0;
}
// clean up (only really needed when length is even)
a[slow] = 0;
}
В моей системе это примерно на 20% быстрее, чем в исходной версии.
Теперь вам нужно протестировать и сообщить нам, работает ли он быстрее в вашей системе!
Вот версия, использующая арифметику указателей и размещение новых, которая использует тот факт, что std:: vector использует внутреннюю непрерывную разметку памяти:
void down_sample(std::vector<int> & v){
int * begin = &v[0];
int * stop = begin + v.size();
int * position = begin + 2;
int * half_position = begin +1;
while( position < stop){
*half_position = *position;
++half_position;
position += 2;
}
size_t size = v.size()/2;
int * a = new (half_position) int[size]();
}
На моей машине этот код работает в 3 раза быстрее, чем ваш с отключенными оптимизациями, и примерно на 30 % быстрее, чем ваша версия, когда она скомпилирована с -o3 на gcc7.2. Я проверил это с размером вектора 20 000 000 элементов.
И я думаю, что в вашей версии строки:
for (int i =(sz-1)/2; i < sz; i++) a[i] = 0;
должно быть
for (int i =(sz-1)/2 + 1; i < sz; i++) a[i] = 0;
в противном случае будет установлено слишком много элементов в ноль.
Принимая во внимание вопрос Джона Цвинка, я сделал несколько быстрых тестов с memset и std:: fill вместо размещения new.
Вот результаты:
n = 20000000
compiled with -o0
orginal 0.111396 seconds
mine 0.0327938 seconds
memset 0.0303007 seconds
fill 0.0507268 seconds
compiled with -o3
orginal 0.0181994 seconds
mine 0.014135 seconds
memset 0.0141561 seconds
fill 0.0138893 seconds
n = 2000
compiled with -o0
orginal 3.0119e-05 seconds
mine 9.171e-06 seconds
memset 9.612e-06 seconds
fill 1.3868e-05 seconds
compiled with -o3
orginal 5.404e-06 seconds
mine 2.105e-06 seconds
memset 2.04e-06 seconds
fill 1.955e-06 seconds
n= 500000000 (with -o3)
mine= 0,350732
memeset = 0.349054
fill = 0.352398
Кажется, что memset немного быстрее для больших векторов, а std:: fill немного быстрее для маленьких векторов. Но разница очень маленькая.
Моя версия одного прохода decimate():
void decimate (std::vector<int>& a) {
const std::size_t sz = a.size();
const std::size_t half = sz / 2;
bool size_even = ((sz % 2) == 0);
std::size_t index = 2;
for (; index < half; index += 2) {
a[index/2] = a[index];
}
for (; index < sz; ++index) {
a[(index+1)/2] = a[index];
a[index] = 0;
}
if (size_even && (half < sz)) {
a[half] = 0;
}
}
и тесты для него:
#include <vector>
#include <iostream>
#include <cstddef>
void decimate(std::vector<int> &v);
void print(std::vector<int> &a) {
std::cout << "{";
bool f = false;
for(auto i:a) {
if (f) std::cout << ", ";
std::cout << i;
f = true;
}
std::cout << "}" << std::endl;
}
void test(std::vector<int> v1, std::vector<int> v2) {
auto v = v1;
decimate(v1);
bool ok = true;
for(std::size_t i = 0; i < v1.size(); ++i) {
ok = (ok && (v1[i] == v2[i]));
}
if (ok) {
print(v);
print(v1);
} else {
print(v);
print(v1);
print(v2);
}
std::cout << "--------- " << (ok?"ok":"fail") << "\n" << std::endl;
}
int main(int, char**)
{
test({},
{});
test({1},
{1});
test({1, 2},
{1, 0});
test({1, 2, 3},
{1, 3, 0});
test({1, 2, 3, 4},
{1, 3, 0, 0});
test({1, 2, 3, 4, 5},
{1, 3, 5, 0, 0});
test({1, 2, 3, 4, 5, 6},
{1, 3, 5, 0, 0, 0});
test({1, 2, 3, 4, 5, 6, 7},
{1, 3, 5, 7, 0, 0, 0});
test({1, 2, 3, 4, 5, 6, 7, 8},
{1, 3, 5, 7, 0, 0, 0, 0});
test({1, 2, 3, 4, 5, 6, 7, 8, 9},
{1, 3, 5, 7, 9, 0, 0, 0, 0});
test({1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
{1, 3, 5, 7, 9, 0, 0, 0, 0, 0});
test({1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
{1, 3, 5, 7, 9, 11, 0, 0, 0, 0, 0});
return 0;
}
Не подходите к sz, если впоследствии вы установите его на ноль.
Если sz есть даже goto sz/2, если нет - (sz-1)/2.
for (int i =0; i < sz_half; i++)
a[i] = a[2*i];
Я сравнил все ответы, приведенные здесь. Я использовал компилятор Intel ICC версии 15.0.3. Был использован уровень оптимизации O3.
Orig: Time difference [micro s] = 79506
JohnZwinck: Time difference [micro s] = 69127
Hatatister: Time difference [micro s] = 79838
user2807083: Time difference [micro s] = 80000
Schorsch312: Time difference [micro s] = 84491
Все времена относятся к вектору с длиной 100000000.
#include <vector>
#include <cstddef>
#include <iostream>
#include <chrono>
const int MAX = 100000000;
void setup(std::vector<int> & v){
for (int i = 0 ; i< MAX; i++) {
v.push_back(i);
}
}
void checkResult(std::vector<int> & v) {
int half_length;
if (MAX%2==0)
half_length = MAX/2;
else
half_length = MAX-1/2;
for (int i = 0 ; i< half_length; i++) {
if (v[i] != i*2)
std::cout << "Error: v[i]=" << v[i] << " but should be " << 2*i << "\n";
}
for (int i = half_length+1; i< MAX; i++) {
if (v[i] != 0)
std::cout << "Error: v[i]=" << v[i] << " but should be 0 \n";
}
}
void down_sample(){
std::vector<int> v;
setup(v);
auto start_time = std::chrono::steady_clock::now();
int * begin = &v[0];
int * stop = begin + v.size();
int * position = begin + 2;
int * half_position = begin +1;
while( position < stop){
*half_position = *position;
++half_position;
position += 2;
}
size_t size = v.size()/2;
int * a = new (half_position) int[size]();
auto duration = std::chrono::steady_clock::now() - start_time;
std::cout << "Orig: Time difference [micro s] = " << std::chrono::duration_cast<std::chrono::microseconds>(duration).count() <<std::endl;
checkResult(v);
}
void down_sample_JohnZwinck () {
std::vector<int> v;
setup(v);
auto start_time = std::chrono::steady_clock::now();
size_t slow = 1, fast = 2;
// read the first half, write the first quarter
size_t stop = (v.size()+1)/2;
while (fast < stop) {
v[slow++] = v[fast];
fast += 2;
}
// read and clear the second half, write the second quarter
stop = v.size();
while (fast < stop) {
v[slow++] = v[fast];
v[fast++] = 0;
v[fast++] = 0;
}
// clean up (only really needed when length is even)
v[slow] = 0;
auto duration = std::chrono::steady_clock::now() - start_time;
std::cout << "JohnZwinck: Time difference [micro s] = " << std::chrono::duration_cast<std::chrono::microseconds>(duration).count() <<std::endl;
checkResult(v);
}
void down_sample_Schorsch312(){
std::vector<int> v;
setup(v);
auto start_time = std::chrono::steady_clock::now();
int half_length;
if (v.size()%2==0)
half_length = MAX/2;
else
half_length = MAX-1/2;
for (int i=0; i < half_length; i++)
v[i] = v[2*i];
for (int i=half_length+1; i< MAX; i++)
v[i]=0;
auto duration = std::chrono::steady_clock::now() - start_time;
std::cout << "Schorsch312: Time difference [micro s] = " << std::chrono::duration_cast<std::chrono::microseconds>(duration).count() <<std::endl;
}
void down_sample_Hatatister(){
std::vector<int> v;
setup(v);
auto start_time = std::chrono::steady_clock::now();
int * begin = &v[0];
int * stop = begin + v.size();
int * position = begin + 2;
int * half_position = begin +1;
while( position < stop){
*half_position = *position;
++half_position;
position += 2;
}
size_t size = v.size()/2;
int * a = new (half_position) int[size]();
auto duration = std::chrono::steady_clock::now() - start_time;
std::cout << "Hatatister: Time difference [micro s] = " << std::chrono::duration_cast<std::chrono::microseconds>(duration).count() <<std::endl;
checkResult(v);
}
void down_sample_user2807083 () {
std::vector<int> v;
setup(v);
auto start_time = std::chrono::steady_clock::now();
const std::size_t sz = v.size();
const std::size_t half = sz / 2;
bool size_even = ((sz % 2) == 0);
std::size_t index = 2;
for (; index < half; index += 2) {
v[index/2] = v[index];
}
for (; index < sz; ++index) {
v[(index+1)/2] = v[index];
v[index] = 0;
}
if (size_even && (half < sz)) {
v[half] = 0;
}
auto duration = std::chrono::steady_clock::now() - start_time;
std::cout << "user2807083: Time difference [micro s] = " << std::chrono::duration_cast<std::chrono::microseconds>(duration).count() <<std::endl;
checkResult(v);
}
int main () {
down_sample();
down_sample_JohnZwinck ();
down_sample_Schorsch312();
down_sample_Hatatister();
down_sample_user2807083();
}