Скорости разработки и исполнения не достижимые на С

В продолжении статьи о кроссплатформенной и кросс-аппаратной оптимизации, на примере задачи поиска полным проходом по таблице из 5 полей и 10 000 000 строк, и неизбежности этой задачи даже при индексном поиске, я покажу как ускорить такой поиск в 3.5 – 5.3 раза с использованием C++ независимо от аппаратной платформы.

В предыдущей статье нам удалось ускорить поиск в 1.3 раза: GitHub.com

Мы не будем банально описывать конструкции языка, а покажем преимущества C++ при решении одного из этапов реальной задачи.

Мы по-прежнему пишем кроссплатформенно под MSVC11(MSVS2012) и GCC 4.7.2, и используем в них C и частично реализованный стандарт C++11.

Для упрощения понимания мы все ещё пишем без индексного поиска, но это решение в дальнейшем будет использоваться при индексном поиске.

1. Hardcore-ное решение на C++

Действительно хардкорным решением это выглядело бы на C. Это потому, что к несчастью для C-разработчиков, вам придется создать 3840 функций и нигде в них не ошибиться. Пока C-разработчики их пишут, мы отнесемся к ним с большим уважением, а я покажу, как эти 3840 функции может написать за вас компилятор C++. Сделает он это не быстро, но на порядки быстрее и точнее, чем их напишут C-разработчики.

Все очень просто, в теории СУБД есть такое понятие, как селективность предиката (selectivity) – процент подходящих строк таблицы по одному из условий(предиката). Чем меньше селективность условия, тем раньше по нему надо сравнивать, и тем раньше мы поймем, что строка не подходит. Т.е. избавимся от сравнений по следующим условиям.

В реальных СУБД для этого оптимизатором собирается статистика и подбирается нужная формула для расчета селективности в зависимости от графика распределения данных. Думаю, дискретной математики и комбинаторики с перебором полей на сегодня вам достаточно, поэтому для простоты мы будем считать селективность по кардинальности из расчета равномерного распределения данных. (Кардинальность колонки (cardinality) – процент уникальных значений в колонке.)

Селективность (количество строк) будет равна:

selectivity = c_array_size * (1 + field.end – field.begin) / cardinality;

Размер массива и кардинальность для равномерно распределенных данных нам известна на этапе компиляции. А границы условий поиска для каждого поля мы получаем во время исполнения и рассчитываем селективность.

Затем выбираем поля в порядке возрастания селективности начиная с наименьшей. И в зависимости от порядка выбранных номеров полей, мы формируем числовой индекс наиболее оптимальной функции, которую затем вызываем для поиска.

Оптимизированные функции на каждый вариант условий поиска, а точнее полиморфные классы, мы генерируем двумя вложенными раскрутками шаблонов:

1. Перебор всех вариантов наличия условий поиска (как это было в предыдущем решении на C++)


2. Перебор всех вариантов порядка условий поиска

Итого создается и заполняется массив из (2^5)*5! = 32*120 = 3840 объектов с разной реализацией, но с общим базовым абстрактным предком.

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

Теперь разберем, как будет выглядеть код.

2. Уменьшение зависимостей

Сначала для реализации нашего решения придется реализовать доступ к полям строки через шаблонные compile-time-геттеры. Это так же добавит приятный бонус – уменьшение зависимостей – теперь любые изменения типов и числа полей в коде будут затрагивать только структуру строки T_cash_account_row.

Сам интерфейс нашего поискового класса будет оставаться неизменным, а реализация поиска по произвольной таблице будет осуществляться инстанцированием шаблонного поискового класса.

Так будет выглядеть код структуры строки с полями:

/* Row */
struct T_cash_account_row {

        // Fields:
        enum T_field_enum { amount_of_money_e, gender_e, age_e, code_e, height_e,   /*<<<<<- add fields here (with index) */
                last_e  /*<<<<<- add included fields here (without index) */
        };
        static_assert(last_e > 0, "Number of indexed fields in enum of T_cash_account_row must be greater than 0!");

        unsigned code:20;                       // 0 - 1000000
        unsigned gender:1;                      // 0 - 1
        unsigned age:7;                 // 0 - 100      
        unsigned amount_of_money:20;    // 0 - 1000000
        unsigned height:9;                      // 0 – 300


        // Get field value
template<int field_enum> inline 
        typename std::enable_if<field_enum == code_e, decltype(code)>::type get_field_value() const { return code; }

template<int field_enum> inline 
        typename std::enable_if<field_enum == gender_e, decltype(gender)>::type get_field_value() const { return gender; }

template<int field_enum> inline 
        typename std::enable_if<field_enum == age_e, decltype(age)>::type get_field_value() const { return age; }

template<int field_enum> inline 
        typename std::enable_if<field_enum == amount_of_money_e, decltype(amount_of_money)>::type get_field_value() const { return amount_of_money; }

template<int field_enum> inline 
        typename std::enable_if<field_enum == height_e, decltype(height)>::type get_field_value() const { return height; }

template<int field_enum> inline 
        typename std::enable_if<field_enum == last_e, bool>::type get_field_value() const { return true; }

        static_assert(5 == last_e, "Add/delete new field-case and correct this assert!");
};
/* ----------------------------------------------------------------------- */

Перечисления enum T_field_enum по умолчанию всегда имеют значения в порядке возрастания начиная с 0 (C++03 7.2 Enumeration declarations).

Шаблон std::enable_if<> из библиотеки #include <type_traits> позволяет получать экземпляры функций/классов только для определенных значений параметра шаблона. В противном случае происходит ошибка подстановки при инстанцировании, и по принципу SFINAE (substitution-failure-is-not-an-error) – при ошибке подстановки ошибка компиляции не возникает, а экземпляр функции просто не создается. Подробней в стандарте (С++03 14.8.2 Template argument deduction).

Это необходимо для обращения через параметры шаблона геттера к нужным полям разного типа, например так:

auto code = row->get_field_value<T_cash_account_row::code_e>();

В нашем случае все 5 полей имеют одинаковый тип и можно было бы обойтись без std::enable_if<>, но мы же думаем наперёд. В моем примере вы можете изменить тип любого из 5 полей и геттеры отлично скомпилируются и сработают. Допустим, если бы мы сделали их через специализации шаблонной функции, то получали бы ошибки при обращении к разным типам, например такой код не скомпилируется:

/* Row */
struct T_cash_account_row {

    // Fields:
        enum T_field_enum { amount_of_money_e, gender_e, age_e, code_e, height_e,   /*<<<<<- add fields here (with index) */
                last_e /*<<<<<- add included fields here (without index) */
        };
        static_assert(last_e > 0, "Number of indexed fields in enum of T_user_row must be greater than 0!");

        unsigned code:20;                       // 0 - 1000000
        unsigned gender:1;                      // 0 - 1
        unsigned age:7;                 // 0 - 100      
        unsigned amount_of_money:20;    // 0 - 1000000
        int height;                             // 0 – 300

        // Get field value
        template<int field_enum> inline unsigned get_field_value();
        template<int field_enum> inline int get_field_value();

        static_assert(5 == last_e, "Add/delete new field-case and correct this assert!");
};

template<> inline unsigned T_cash_account_row::get_field_value<T_cash_account_row::code_e>() { return code; }
template<> inline unsigned T_cash_account_row::get_field_value<T_cash_account_row::gender_e>() { return gender; }
template<> inline unsigned T_cash_account_row::get_field_value<T_cash_account_row::age_e>() { return age; }
template<> inline unsigned T_cash_account_row::get_field_value<T_cash_account_row::amount_of_money_e>() { return amount_of_money; }
template<> inline int T_cash_account_row::get_field_value<T_cash_account_row::height_e>() { return height; }
template<> inline unsigned T_cash_account_row::get_field_value<T_cash_account_row::last_e>() { return 1; }
/* ----------------------------------------------------------------------- */

int main() {
    T_cash_account_row *row = new T_cash_account_row;
    auto code = row->get_field_value<T_cash_account_row::code_e>();
    
    return 0;
}

Компилятор естественно выдаст ошибку о не однозначном вызове функции get_field_value<>(). Ideone.com

Наш же вариант отлично скомпилируется: ideone.com

С геттерами разобрались, теперь применим раскрутку шаблонов (unrolling of template) похожую на раскрутку из прошлой статьи, но не для конструктора, а для функтора (функтор – это класс с перегруженным оператором вызова operator()(…)):

// The templated functor of unrolling of the input templated functor (reusable)
template<unsigned unroll_count, template<unsigned> class T>
struct T_unroll_functor {
        T_unroll_functor<unroll_count-1, T> next_unroll;
        T<unroll_count-1> functor;
        template<typename T1, typename T2> inline bool operator()(T1 const& val1, T2 const& val2) {
                return next_unroll(val1, val2) && functor(val1, val2);
        }
};
// End of unroll
template<template<unsigned> class T>
struct T_unroll_functor<0, T> { 
        template<typename T1, typename T2> inline bool operator()(T1 const&, T2 const&) { return true; } 
};
// -------------------------------------------------------------------------

Такой раскруткой мы развернем сравнения для каждого поля, основываясь на значении перечисления last_e и доступе к полям через наши шаблонные геттеры.

А разворачивать будем функтор struct T_test_pred следующим образом:

// The filters for each search variant (range_filters)
template<typename T_row, unsigned index_pred>
struct T_custom_filter : T_filter<T_row> {

        // Test predicates functor for unrolling
        template<unsigned field_num>
        struct T_test_pred {
                bool inline operator()(T_row *const __restrict row, T_range_filters *const __restrict range_filters) const 
                {
                        typedef typename T_row::T_field_enum T_field_enum;
                        // Without fields where use_filter==0
                        return ( T_filter<T_row>::template T_get_use_filter<index_pred, field_num>::value ||
                                (row->template get_field_value<static_cast<T_field_enum>(field_num)>() >= 
                                range_filters->begin.template get_field_value<static_cast<T_field_enum>(field_num)>()&& 
                                row->template get_field_value<static_cast<T_field_enum>(field_num)>() <= 
                                range_filters->end.template get_field_value<static_cast<T_field_enum>(field_num)>()) 
                        );
                }                       
        };
        // -----------------------------------------------------------------------
        
        // search
        virtual size_t search(T_row *const __restrict array_ptr, const size_t c_array_size, T_row *const __restrict result_ptr, T_range_filters *const __restrict range_filters) final
        {
                size_t result_size = 0;
                size_t i; // loop index
                T_unroll_functor<T_row::last_e, T_test_pred> test_predicate;
                for(i = 0; i < c_array_size; ++i) {
                        if(test_predicate(array_ptr + i, range_filters)) 
                                result_ptr[result_size] = array_ptr[i], ++result_size;
                }
                return result_size;
        }
};
// -------------------------------------------------------------------------

Замечу ещё, что теперь в классах T_filter<>, T_custom_filter<> и T_optimized_search<> появился параметр шаблона T_row, в который передается тип строки таблицы – в нашем случае T_cash_account_row.

Теперь все изменения количества, названий и типов полей таблицы сосредоточены исключительно внутри T_cash_account_row, а весь код поиска генерируется автоматически из шаблонных классов, достаточно только инстанцировать шаблон типом нужной строки:

T_optimized_search<T_cash_account_row> optimized_search; // C++ optimized search

Вот ссылка на получившийся рабочий код: GitHub.com

В этом примере наглядно показано почему в критически важных участках кода team-lead и performance-architect должен уделять внимание не только внешним интерфейсам модулей, но и типам объектов передаваемых через эти интерфейсы. В нашем случае это: T_cash_account_row и T_range_filters. Задав возможность в T_cash_account_row передавать параметры через шаблонные геттеры, мы предопределяем более гибкую реализацию разработчиком модуля через раскрутку шаблонов, а так же более гибкую реализацию тестов при использовании техники разработки через тестирование TDD, что я покажу в следующей статье.

На компиляторе MSVC11(MSVS2012) и одном ядре CPU Core i5 K750 результат такой:

Generated rows: 10000000

C++-Searching…

C++-optimized search took 0.056000 seconds.

Found rows: 38

C-Searching…

C-search took 0.074000 seconds.

The C++ faster than C: 1.321429 times

Found rows: 38

Как видим, код на C++ по прежнему быстрее в 1.3 раза, чем код на C. Получаемый ассемблерный код функции test_predicate по прежнему идентичен предыдущему варианту на C++, test_pradicate_enum_cpp.asm.

«Картинка disasm из TortoiseDiff»



Но теперь мы сосредоточили все изменения в одном месте, что значительно облегчит использование этого модуля поиска.

Кроме того, мы подготовили код к другой, ещё более производительной оптимизации.

3. Ускорение в 3.5 – 5.3 раза

Достаточно часто встречаются статьи по C++, где рассматриваются compile-time вычисления или например поиск простых чисел на шаблонах. Статьи интересные, но как замечают сами авторы: «я расскажу, как сделать совершенно бесполезную вещь».

Я же расскажу, как получить заметное ускорение поиска от compile-time вычислений и оптимизаций с использованием шаблонов.

3.1 Простые вычисления на шаблонах

Начнем с самого простого из нужного нам, реализуем compile-time вычисление факториала:

template <unsigned N> 
        struct factorial : std::integral_constant<unsigned, N * factorial<N-1>::value> {};
template <> struct factorial<0> : std::integral_constant<unsigned, 1> {};

Вычисление факториала нам понадобиться для создания статичного массива std::array<> с числом элементов равным (2^5)*5! = 3840.

Из дискретной математики и комбинаторики известно, что число различных перестановок уникальных элементов равно факториалу их количества. А число вариантов наличия элементов равно 2 в степени количества элементов. (Конечно, нам абсолютно не важно, как переставлены отсутствующие элементы, и отсутствие лишних вариантов могло бы ускорить компиляцию, но я пытаюсь как можно сильнее упростить эту статью.)

Теперь посмотрим какие оптимизации будут внутри структуры

struct T_custom_filter при её инстанцировании.

Во-первых, она теперь ещё принимает параметр index_order. А во-вторых в ней изменился поисковый раскручиваемый функтор.

// The filters for each search variant (range_filters)
template<typename T_row, unsigned index_pred, unsigned index_order = 0>
struct T_custom_filter : T_filter<T_row> {

        // Test predicates functor for unrolling
        template<unsigned field_num>
        struct T_test_pred {
                bool inline operator()(T_row *const __restrict row, T_range_filters *const __restrict range_filters) const 
                {
                        typedef typename T_row::T_field_enum T_field_enum;

                        // Without fields where use_filter==0
                        enum { ordered_field_number = T_filter<T_row>::template T_number_field<index_order, field_num>::value };

                        return ( T_filter<T_row>::template T_get_use_filter<index_pred, ordered_field_number>::value ||
                                (row->template get_field_value<static_cast<T_field_enum>(ordered_field_number)>() >= 
                                range_filters->begin.template get_field_value<static_cast<T_field_enum>(ordered_field_number)>() && 
                                row->template get_field_value<static_cast<T_field_enum>(ordered_field_number)>() <= 
                                range_filters->end.template  get_field_value<static_cast<T_field_enum>(ordered_field_number)>()) 
                        );
                }                       
        };
        // -----------------------------------------------------------------------
        // search
        virtual size_t search(T_row *const __restrict array_ptr, const size_t c_array_size, T_row *const __restrict result_ptr, T_range_filters *const __restrict range_filters) final;
};
// -------------------------------------------------------------------------

Как и прежде, шаблон T_get_use_filter возвращает флаг использования (или не использования) фильтра по полю number_filter для данного индекса index_pred.

// whether to use the filter for a given predicate? (strictly at compile time)
template <unsigned index_pred, unsigned number_filter> 
struct T_get_use_filter : std::integral_constant<bool, !(index_pred & 1<<number_filter)> {};

Из шаблона T_number_field<> на основании index_order – индекса порядка сравнения полей и порядкового номера поля field_num, мы получаем номер поля ordered_field_number для получения его значения из геттера get_field_value<static_cast<T_field_enum>(ordered_field_number)>().

3.2 Более сложные вычисления на шаблонах

Теперь сложнее. Реализуем вычисление порядка поля в шаблонном классе T_number_field<>. Индекс порядка полей index_order формируется следующим образом – например, индекс для 5 полей составляется по следующей формуле:

index_order = field0 + 5*field1 + 5*4*field2 + 5*4*3*field3 + 5*4*3*2*field4;

где field1 – номер поля от 0 до 4-их (из 5-ти имеющихся), идущий первым при сравнении; field2 – номер поля от 0 до 3 (из 4-ех оставшихся), идущий вторым при сравнении; …; а field5 – номер поля от 0 до 0 (из 1-го оставшегося) идущий последним, т.к. он всегда 0, то его можно опустить:

index_order = field0 + 5*field1 + 5*4*field2 + 5*4*3*field3;

или, что тоже самое:

index_order = field0 + 5*(field1 + 4*(field2 + 3*(field3)));

Кому-то возможно будет легче представить значение index_order, как число с переменной основой системы исчисления, где field0 – первая цифра с основой 5 (может принимать 5 значений 0-4), в field1 – вторая цифра с основой 4 (может принимать 4 значения 0-3), и так далее. Таким образом в (5!) комбинаций мы можем закодировать любые перестановки полей. Для лучшего понимания желательно ознакомиться с основами теории вероятности или перебором из дискретной математики.

Визуально это можно изобразить так: круглыми скобками () обозначаем наименьшее значение приоритета из имеющихся (наиболее приоритетное), квадратным [] – его номер из имеющихся, а фигурными {} – реальный номер поля. Например, изначально имеем 5 полей с такими приоритетами: 5, 3, 1, 2, 4

Номер поля:         0 1  2  3 4
Приоритет сравнения:5 3 (1) 2 4 - [2] – текущий номер поля (field0) {2}
Приоритет сравнения:5 3 (2) 4   - [2] – текущий номер поля (field1) {3}
Приоритет сравнения:5 (3) 4     - [1] – текущий номер поля (field2) {1}
Приоритет сравнения:5 (4)       - [1] – текущий номер поля (field3) {4}
Приоритет сравнения:(5)         - [0] – текущий номер поля (field4) {0}

Подставив числа в: index_order = field0 + 5*(field1 + 4*(field2 + 3*(field3)));

Мы получаем значение индекса: index_order = 2 + 5*(2 + 4*(1 + 3*(1))) = 92;

Позже мы покажем, как это реализуется в run-time функции T_optimized_search::get_index_order().

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

Сначала научимся получать числа стоящие в квадратных скобках []. В нашем конкретном случае «текущие номера полей» находятся следующим образом, для полученного нами index_order = 2 + 5*(2 + 4*(1 + 3*(1))) = 92:

field0 = index_order % 5 = 92 % 5 = 2;
field1 = (index_order/5) % 4 = (92/5) % 4 = 2;
field2 = (index_order/(5*4)) % 3 = (92/5*4) % 3 = 1;
field3 = (index_order/(5*4*3)) % 2 = (92/5*4*3) % 2 = 1;
field4 = 0;

В compile-time это делает следующий шаблонный класс:

// Get the sequence number the remaining field for the number_filter, after removal of all previous 
template <unsigned index_order, unsigned number_filter, unsigned index = T_row::last_e> 
struct T_number_remaining_field : std::integral_constant<unsigned, T_number_remaining_field<index_order / index, number_filter - 1, index - 1>::value> {};

// End of unroll
template <unsigned index_order, unsigned index> 
struct T_number_remaining_field<index_order, 0, index> : std::integral_constant<unsigned, index_order % index> {};
// -------------------------------------------------------------------------

И получаем значение «текущего номера поля» равного number_filter, инстанцированием шаблона таким образом T_number_remaining_field<index_order, number_filter>::value.

Здесь есть момент, где некоторые могут запутаться: кто-то может подумать, что при (number_filter=0), т.е. T_number_remaining_field<index_order, 0>::value сразу вызывается частичная специализация с двумя параметрами:

template <unsigned index_order, unsigned index> struct T_number_remaining_field<index_order, 0, index>

Где index_order подставится в index_order, а 0 подставится в index. Но это не так!

По стандарту (С++03 14.8.2 Template argument deduction), сначала из общего шаблона подставляются значения по умолчанию index = T_row::last_e и фактически наше инстанцирование превращается в T_number_remaining_field<index_order, 0, T_row::last_e>::value. А затем уже идет обращение к специализации:

template <unsigned index_order, unsigned index> struct T_number_remaining_field<index_order, 0, index>

И для нулевого текущего номера поля из 5 полей (T_row::last_e=5) получаем index_order % 5, что и требовалось.

Мы реализовали способ нахождения «текущих номеров полей» тех, что в квадратных скобках. Теперь нам нужно реализовать получение реальных номеров полей тех, что в фигурных скобках {}.

Как мы видели из нашего визуального представления, «текущие номера полей» в квадратных скобках [] не всегда совпадают с реальными номерами полей в фигурных скобках {}. Если представить колоду пронумерованных карт лежащих в порядке возрастания снизу-вверх, то изначально их номера будут совпадать с их порядок в колоде. Но вынимая карты из середины, у тех карт, что лежат выше них, каждый раз будет уменьшаться на единицу порядковый номер в колоде относительно их реального номера.

В нашем случае это при водит к тому, что все поля имеющие меньший реальный номер {} и меньшее значение приоритета () (более приоритетное, которые раньше изымаются), уменьшают на единицу «текущее значение поля» [], чей реальный номер и значение приоритета больше. Такие смещения и рассчитывают следующие шаблонные классы:

 // Get 1 or 0 offset if current_field (for number_filter) affect to number_field
        template <unsigned index_order, unsigned number_filter, unsigned number_field>
        struct T_offset {
                enum { current_field = T_number_remaining_field<index_order, number_filter>::value,
                value = (current_field <= number_field)?1:0 };
        };
        
        // Get offset of number_field (enum in row-type) for number_filter
        template <unsigned index_order, unsigned number_filter, unsigned number_field>
        struct T_offset_number_field {
                enum { offset = T_offset<index_order, number_filter-1, number_field>::value,
                value = T_offset_number_field<index_order, number_filter-1, number_field + offset>::value + offset };
        };

        // End of unroll
template <unsigned index_order, unsigned number_field>
struct T_offset_number_field<index_order, 0, number_field> :      std::integral_constant<unsigned, 0> {};

И наконец, реальные номера полей мы получаем следующим шаблонным классом:

 // (Main) Get number of field (enum in row-type) for number_filter
        template <unsigned index_order, unsigned number_filter>
        struct T_number_field {
                enum { remaining_field = T_number_remaining_field<index_order, number_filter>::value,
                value = remaining_field + T_offset_number_field<index_order, number_filter, remaining_field>::value };
        };
        // -------------------------------------------------------------------------

Готово.

3.3 Вложенная раскрутка шаблонов

Теперь нам надо инстанцировать шаблон фильтра T_custom_filter<> по наличию полей поиска и по их порядку. Для этого мы применим вложенную раскрутку шаблонов:

template<typename T_row>
class T_optimized_search {
        // unroll tamplates
        template<unsigned index_pred>
        struct T_unroll_find {
                template<unsigned index_order>
                struct T_unroll_order {
                        template<typename T> T_unroll_order(T &filters) { 
                                filters[index_pred + index_order*(1<<T_row::last_e)].reset( new T_custom_filter<T_row, index_pred, index_order>() ); 
                        }
                };
                T_unroll_constructor<factorial<T_row::last_e>::value, T_unroll_order> fill_ordered_filter;

                template<typename T> T_unroll_find(T &filters) : fill_ordered_filter(filters)
                {}
        };
        // -------------------------------------------------------------------------

        std::array<std::unique_ptr<T_filter<T_row>>, (1<<T_row::last_e)*factorial<T_row::last_e>::value> filters;
        T_unroll_constructor< 1<<T_row::last_e, T_unroll_find> fill_filter;
public:
        T_optimized_search() : fill_filter(filters) {}

        // C++ optimized search
        inline size_t search(T_row *const __restrict array_ptr, const size_t c_array_size,
                T_row *const __restrict result_ptr, T_range_filters *const __restrict range_filters) 
        {
                auto const& filter = filters[T_filter<T_row>::get_index_pred(range_filters) + 
                        T_filter<T_row>::get_index_order(range_filters)*(1<<T_row::last_e)];
                return filter->search(array_ptr, c_array_size, result_ptr, range_filters);
        }
};
// -------------------------------------------------------------------------

Здесь мы использовали compile-time раскрутку наличия сравнения полей struct T_unroll_find<>, а внутри неё использовали раскрутку порядка сравнения полей struct T_unroll_order<>.

А в поисковой функции мы используем run-time функции get_index_pred() – получения индекса в зависимости от необходимости сравнения полей, и функции get_index_order() – получения индекса в зависимости от необходимого порядка сравнения полей.

Первая функция осталась той же:

 // Get index of T_test_pred version for current search range 
        static inline const unsigned get_index_pred(T_range_filters *const __restrict range_filters)    {
                unsigned result = 0;
                for(size_t i = 0; i < T_row::last_e; ++i) 
                        result |= range_filters->use_filter[i]?(1<<i):0;
                return result;
        }
        // -------------------------------------------------------------------------

И добавилась вторая get_index_order(). Сначала уточню, что в структуре строки struct T_cash_account_row появилась функция get_bitf_size() – получения кардинальности (именно в такой форме, а не через switch/case, она может выступать как constexpr, т.е. вычисляемой compile-time, но MSVC11 это пока не поддерживает):

/* Row */
struct T_cash_account_row {
        ...
        // Get cardinality
        template<int field_enum>
        // constexpr // not supported in the MSVS2012(MVCC11)
        static inline const unsigned int get_bitf_size() { 
                return (field_enum == gender_e)?                1:
                        (field_enum == age_e)?          100:
                        (field_enum == height_e)?               300:
                        (field_enum == code_e)?         1000000:
                        (field_enum == amount_of_money_e)?1000000:
                        0;
                static_assert(5 == last_e, "Add/delete new field-case and correct this assert!");
        }
};

И ещё раз напомню формулу селективности (количество строк):

selectivity = c_array_size * (1 + field.end – field.begin) / cardinality;

А теперь и сама функция получения индекса в зависимости от порядка сравнения полей, использующая шаблонную раскрутку конструктора для вычисления селективности для каждого условия/поля:

 // The unrolling filling of selectivity in a compile-time
        template<unsigned field_num>
        struct T_selectivity_fill {
                T_selectivity_fill(std::map<unsigned, unsigned> *const __restrict ordered_filters, T_range_filters *const __restrict range_filters) 
                { 
                        // selectivity for each field-filter
                        const unsigned selectivity = range_filters->use_filter[field_num]?
                                ((1 + 
                                        range_filters->end.template get_field_value<field_num>() - 
                                        range_filters->begin.template get_field_value<field_num>()     )*c_array_size / 
                                T_row::template get_bitf_size<field_num>()) // cardinality
                                :c_array_size;
                        ordered_filters->insert(std::make_pair(field_num, selectivity));     
                }
        };
                
        // Get index of T_test_pred version for current search range 
        static inline const unsigned get_index_order(
                T_range_filters *const __restrict range_filters) 
        {
                unsigned result = 0;

                std::map<unsigned, unsigned> ordered_filters;     
                T_unroll_constructor<T_row::last_e, T_selectivity_fill> selectivity_fill(&ordered_filters, range_filters);

                unsigned multiplexor = 1;
                for(size_t i = 0; i < T_row::last_e; ++i) {
                        unsigned cur_index = 0, min_index = 0;
                        unsigned field_num = ordered_filters.cbegin()->first;
                        unsigned min = c_array_size;
                        for(auto const& it : ordered_filters) {
                                if (it.second < min) 
                                        min = it.second, field_num = it.first, min_index = cur_index;
                                ++cur_index;
                        }
                        ordered_filters.erase(field_num);
                        result += min_index*multiplexor;
                        multiplexor *= (T_row::last_e - i);
                }
                return result;
        }
        // -------------------------------------------------------------------------

Здесь ключ в std::map<> заполняется номерами полей, а значение заполняется предполагаемой селективностью. Затем в цикле находим минимальную селективность, берем из ключа соответствующий номер поля, по нему удаляем запись из std::map<>, и используем этот номер поля для составления индекса. И так в цикле для каждого поля. Мы уже описывали как получается этот индекс, возвращаемый функцией get_index_order, математически:

index_order = field1 + 5*(field2 + 4*(field3 + 3*(field4)));

Так мы смогли раскрутить поисковый шаблон 3840 раз, заполнить этими объектами статичный массив и получить в run-time индекс необходимого объекта с оптимизированной функцией поиска, для данных условий поиска.

Получаемый ассемблерный код функции test_predicate изменился по сравнению с предыдущим вариантом на C++ – два сравнения поменялись местами, первое из них теперь идет последним: github.com asm.

«Картинка из TortoiseDiff»

4. Результаты тестов

Вот полностью рабочий вариант этого решения на C++: GitHub.com

На GCC4.7.2 с ключом –O3 –march=native, CPU Core i5 K750 и размером exe-файла в 1.3МБ результат такой:

Generated rows: 10000000

C++-Searching…

C++-optimized search took 0.017000 seconds.

Found rows: 38

C-Searching…

C-search took 0.090000 seconds.

The C++ faster than C: 5.294118 times

Found rows: 38

А на MSVC11(MSVS2012) с ключом /O2 /Ob2 /Oi, CPU Core i5 K750 и размером exe-файла в 1.42МБ результат такой:

Generated rows: 10000000

C++-Searching…

C++-optimized search took 0.021000 seconds.

Found rows: 38

C-Searching…

C-search took 0.074000 seconds.

The C++ faster than C: 3.523809 times

Found rows: 38

Как мы видим, время исполнения упало с 74мс на C, до 21мс на C++, т.е. скорость выросла в 3.5 раза. Отлично. И как видно GCC показывает ещё более заметную разницу в скорости, в 5.3 раза. И это абсолютно кросс-аппаратное решение, в отличие от SIMD-инструкций и cache-prefetch. В отличие от PDO, который создает один оптимизированный путь исполнения программы, мы создали 3840 путей оптимизированных на все случаи входных данных.

Может быть это решение вам покажется сложным, зато теперь понятно, что значит действительно знать C++. Если вы не знаете хорошо C++ то у вас остается единственная альтернатива – писать на C.

Чтобы развеять сомнения, что это можно решить на C через массив указателей на функции, приведу пример. Т.к. я знаю, что хорошей оптимизации на C в данном случае не получится, то я не буду реализовывать механизм вычисления селективности и перестановки условий, а просто руками оптимально поменяю функции местами при заполнении массива указателей.

Вот рабочий вариант этого решения на C: GitHub.com

Чем он отличается от примера с C-оптимизацией из прошлой статьи можете посмотреть в diff с предыдущей версией: GitHub.com

На GCC4.7.2 с ключом –O3 –march=native, CPU Core i5 K750 результат такой:

Generated rows: 10000000

C-optimized-Searching…

C-optimized-search took 0.049000 seconds.

Found rows: 38

C-Searching…

C-search took 0.086000 seconds.

The C++ faster than C: 1.755102 times

Found rows: 38

А на MSVC11 с ключом /O2 /Ob2 /Oi, CPU Core i5 K750 результат такой:

Generated rows: 10000000

C-optimized-Searching…

C-optimized-search took 0.045000 seconds.

Found rows: 38

C-Searching…

C-search took 0.074000 seconds.

The C++ faster than C: 1.644444 times

Found rows: 38

Как мы видим результат с оптимизацией на C отстает от нашей оптимизации на C++ более чем в 2 — 3 раза.

Ускорение на GCC на C в 1.8 раза против 5.3 раза на C++, и ускорение на MSVC на C в 1.6 раза против 3.5 раза на C++. Причем оптимизация на C легко компилируется компилятором C++ без единого изменения, но не наоборот.

5. Заключение

Я показал, как с помощью C++–шаблонов реализовать вычисления и кросс-аппаратную оптимизацию во время компиляции для получения ускорения в 3.5 – 5.3 раза для реальной задачи.

Нет таких оптимизаций в C, которые бы нельзя было применить в C++, т.к. имеется почти полная обратная совместимость. А вот в C – единственная альтернатива C++–шаблонам для подобных оптимизаций – это писать десятки, сотни и тысячи функций – вот уж где действительно copy-paste.

Напомню, что эта оптимизация не применима в Java и C#, т.к. в generics невозможно в параметрах использовать значения, а возможно только типы. Java и C# имеют другие преимущества – это развитые библиотеки и защита от глупых ошибок.

В следующих статьях я опишу наиболее подходящий для данной задачи вариант индексного поиска, почему здесь не подойдут мульти-индексные соединения Index Hash Join и Bitmap AND, а необходимо применять Index Organized Tables и стратегию доступа к данным Index Only Scan. Опишу недостатки бинарного поиска, и почему в нашем случае подойдет упорядоченный хэш-индекс. Добьемся скорости от 0.15 до 3.6 миллиона запросов в секунду на одном ядре Core i5 K750, а затем покажем, как это реализуется в многопоточном варианте со слабым конкурентным доступом и редкими изменениями. Ускорим этот поиск на GPU и покажем почему будущее за многопоточным паттерном – distributed states, с одним барьером памяти на группу записей. Затем вплотную подойдем к задаче поиска в высоконагруженном web-проекте и сравним его реализации с использованием FastCGI и boost::asio. Но не все сразу.

This entry passed through the Full-Text RSS service — if this is your content and you're reading it on someone else's site, please read the FAQ at fivefilters.org/content-only/faq.php#publishers. Five Filters recommends: 'You Say What You Like, Because They Like What You Say' - http://www.medialens.org/index.php/alerts/alert-archive/alerts-2013/731-you-say-what-you-like-because-they-like-what-you-say.html