Иногда вам нужно сделать шаг назад и снова посмотреть на него.Первый вопрос, очевидно, вам это нужно?Может ли быть альтернативный алгоритм, который работал бы лучше?
При этом, и, предположив, ради этого вопроса, что вы уже остановились на этом алгоритме, мы можем попытаться рассуждать о том, что у нас есть на самом деле.
Отказ от ответственности: метод, который я описываю, вдохновлен успешным методом, который Тим Питерс использовал для улучшения традиционной реализации интросорта, что привело к TimSort .Поэтому, пожалуйста, потерпите меня;)
1.Извлечение свойств
Основная проблема, которую я вижу, - это зависимость между итерациями, которая предотвратит большую часть возможных оптимизаций и помешает многим попыткам распараллеливания.
int64_t v = in[i];
max += v;
if (v > max) max = v;
out[i] = max;
Давайте переделаемэтот код в функциональном виде:
max = calc(in[i], max);
out[i] = max;
Где:
int64_t calc(int64_t const in, int64_t const max) {
int64_t const bumped = max + in;
return in > bumped ? in : bumped;
}
Вернее, упрощенная версия (исключая переполнение, поскольку она не определена):
int64_t calc(int64_t const in, int64_t const max) {
return 0 > max ? in : max + in;
}
Вы замечаете точку наконечника?Поведение меняется в зависимости от того, является ли плохо названный (*) max положительным или отрицательным.
Этот переломный момент делает интересным более пристальное наблюдение за значениями в in, особенно в зависимости от эффекта, который ониможет иметь значения max:
max < 0 и in[i] < 0, затем out[i] = in[i] < 0max < 0 и in[i] > 0, затем out[i] = in[i] > 0max > 0 и in[i] < 0, затем out[i] = (max + in[i]) ?? 0max > 0 и in[i] > 0, а затем out[i] = (max + in[i]) > 0
(*) под псевдонимом, поскольку он также является аккумулятором,который имя скрывает.У меня нет лучшего предложения, хотя.
2.Оптимизация операций
Это приводит нас к обнаружению интересных случаев:
- , если у нас есть срез
[i, j) массива, содержащий только отрицательные значения (которые мы называем отрицательным срезом), тогда мы могли бы сделать std::copy(in + i, in + j, out + i) и max = out[j-1] - , если у нас есть срез
[i, j) массива, содержащий только положительные значения, то это чистый код накопления (который можно легко развернуть) max становится положительным, как только in[i] становится положительным
Поэтому может быть интересно (но, возможно, нет, я не даю обещаний) установить профиль ввода передна самом деле работает с этим.Обратите внимание, что для больших входов профиль может быть выполнен по частям, например, путем настройки размера чанка на основе размера строки кэша.
Для справки: 3 подпрограммы:
void copy(int64_t const in[], int64_t out[],
size_t const begin, size_t const end)
{
std::copy(in + begin, in + end, out + begin);
} // copy
void accumulate(int64_t const in[], int64_t out[],
size_t const begin, size_t const end)
{
assert(begin != 0);
int64_t max = out[begin-1];
for (size_t i = begin; i != end; ++i) {
max += in[i];
out[i] = max;
}
} // accumulate
void regular(int64_t const in[], int64_t out[],
size_t const begin, size_t const end)
{
assert(begin != 0);
int64_t max = out[begin - 1];
for (size_t i = begin; i != end; ++i)
{
max = 0 > max ? in[i] : max + in[i];
out[i] = max;
}
}
СейчасПредположим, что мы можем как-то охарактеризовать входные данные, используя простую структуру:
struct Slice {
enum class Type { Negative, Neutral, Positive };
Type type;
size_t begin;
size_t end;
};
typedef void (*Func)(int64_t const[], int64_t[], size_t, size_t);
Func select(Type t) {
switch(t) {
case Type::Negative: return ©
case Type::Neutral: return ®ular;
case Type::Positive: return &accumulate;
}
}
void theLoop(std::vector<Slice> const& slices, int64_t const in[], int64_t out[]) {
for (Slice const& slice: slices) {
Func const f = select(slice.type);
(*f)(in, out, slice.begin, slice.end);
}
}
Теперь, если интросорт не работает в цикле, минимален, поэтому вычисление характеристик может быть слишком дорогостоящим, как есть ... однако этохорошо ведет себя к распараллеливанию .
3.Простое распараллеливание
Обратите внимание, что характеристика является чистой функцией ввода.Следовательно, предположим, что вы работаете в режиме «чанк на чанк», можно параллельно иметь:
- Slice Producer: поток характеризатора, который вычисляет значение
Slice::Type - Slice Consumer: рабочий поток, который на самом деле выполняет код
Даже если ввод является по существу случайным, при условии, что порция достаточно мала (например, строка кэша L1 ЦП), возможно, существуеткуски, для которых это работает.Синхронизация между двумя потоками может быть выполнена с помощью простой потокобезопасной очереди Slice (производитель / потребитель) и добавления атрибута bool last для прекращения потребления или путем создания Slice в векторе с типом Unknownи имея потребительский блок до тех пор, пока он не станет известен (с использованием атомики).
Примечание: поскольку характеристика чистая, она смущающе параллельна.
4.Подробнее Распараллеливание: умозрительная работа
Помните это невинное замечание: max становится положительным, как только in[i] становится положительным .
Предположим, что мы можем предположить (надежно), что Slice[j-1] даст отрицательное значение max, тогда вычисления Slice[j] не зависят от того, что им предшествовало, и мы можем начать работу прямо сейчас!
Конечно, это предположение, поэтому мы можем ошибаться ... но как только мы полностью охарактеризовали все кусочки, у нас появятся свободные ядра, так что мы могли бы также использовать их для умозрительной работы!А если мы не правы?Что ж, поток потребителя просто аккуратно сотрет нашу ошибку и заменит ее на правильное значение.
Эвристика умозрительного вычисления Slice должна быть простой и должна быть настроена.Он также может быть адаптивным ... но это может быть сложнее!
Заключение
Проанализируйте ваш набор данных и попытайтесь выяснить, можно ли нарушить зависимости.Если это так, вы, вероятно, можете воспользоваться этим, даже не переходя к многопоточности.