Бла-бла-бла

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

В последнее время мои мысли заняты вопросом как увеличить скорость поиска рекламных кампаний по заданным параметрам. Рекламная кампания — это, фактически, набор ограничений, которые надо проверить, чтобы выбрать подходящую под запрос. Пытаясь ускорить полный перебор я дошёл до интринсиков, но достижения нельзя было назвать выдающимися. При этом постоянно было чувство, что я иду не совсем верными путём, что где-то что-то надо перевернуть вверх ногами и скорость увеличится на порядок.

Вчера меня озарило после разговора с Пашей Бейзманом, выступавшим на днях с докладом на HighLoad++. Битовые операции!!! Да, я прекрасно понимаю мощь битовых операций, но я их неправильно готовил ;) Результаты тестов превзошли мои самые оптимистичные ожидания.

Идея

Один из типов ограничений при выборе кампании — это проверка по вхождению значения в набор. Например, показ кампании может быть ограничен по стране, области, городу, категории и т.п. Первое, что приходит в голову — это дерево со всеми значениями на уровне кампании. Второе — массив значений, если значений в наборе немного (около 10), прямой перебор будет быстре, чем использование деревьев. Для выбора кампании необходимо перебрать все и в каждой из них проверить вхождение заданного значения в набор.

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

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

Тесты

Куда же без тестов производительности… Я тестировал обработку одного ограничения для 1 миллиона кампаний. Для каждой из кампаний было сгенерировано случайным образом ограничение с набором в 10 значений от 0 до 256 (проверки на ошибки выделения памяти и т.п. из кода убраны):

    int nels = 10;
    int campaigns = 1000000;
    int *values = (int *) malloc(sizeof(int) * campaigns * nels);
    srand(13);
    for (int i = 0; i < campaigns; i++) {
        for (int j = 0; j < nels; j++) {
            values[i * nels + j] = rand() % 256;
        }
    }

Перебор со значениями в массиве

Пробуем просто перебрать все кампании со всеми ограничениями. Здесь и далее компилируем с опцией -O3.

Тестируем.

    clock_t cs, ce;
    int found_campaigns = 0;
    int reference = 13;
    cs = clock();
    for (int i = 0; i < campaigns; i++) {
        for (int j = 0; j < nels; j++) {
            if (values[i * nels + j] == reference) {
                found_campaigns++;
                break;
            }
        }
    }
    ce = clock();
    printf("test #1 bruteforce, found: %d, clocks: %ld, sec: %f\n", found_campaigns, (long) (ce - cs), (double) (ce - cs) / CLOCKS_PER_SEC);

Результат: clocks: 7034, sec: 0.007034.

Перебор со значениями в дереве

Пробуем вместо массива значений ограничения использовать дерево (Judy Arrays).

Pvoid_t *sets = (Pvoid_t) malloc(sizeof(Pvoid_t) * campaigns);
memset(sets, 0, sizeof(Pvoid_t) * campaigns);
for (int i = 0; i < campaigns; i++) {
    for (int j = 0; j < nels; j++) {
        PWord_t pw;
        JLI(pw, sets[i], values[i * nels + j]);
    }
}

Тестируем.

    found_campaigns = 0;
    cs = clock();
    for (int i = 0; i < campaigns; i++) {
        PWord_t pw;
        JLG(pw, sets[i], reference);
        if (pw) {
            found_campaigns++;
        }
    }
    ce = clock();
    printf("test #2 set, found: %d, clocks: %ld, sec: %f\n", found_campaigns, (long) (ce - cs), (double) (ce - cs) / CLOCKS_PER_SEC);

Результат: clocks: 7930, sec: 0.007930.

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

Используем битовые маски

    #define SET_BIT(_n, _i) _n[_i / 64] |= 1ULL << (_i % 64)
    #define CHECK_BIT(_n, _i) _n[_i / 64] & (1ULL << (_i % 64))

    PWord_t pw;
    Pvoid_t pv = (Pvoid_t) 0;
    int mask_size = sizeof(uint64_t) * (campaigns / 64 + 1);
    for (int i = 0; i < campaigns; i++) {
        for (int j = 0; j < nels; j++) {
            int id = values[i * nels + j];
            JLI(pw, pv, id);
            if (*pw) {
                uint64_t *mask = (uint64_t *) *pw;
                SET_BIT(mask, i);
            } else {
                uint64_t *mask = (uint64_t *) malloc(mask_size);
                memset(mask, 0, mask_size);
                SET_BIT(mask, i);
                *pw = (Word_t) mask;
            }
        }
    }

Тестируем.

    found_campaigns = 0;
    cs = clock();
    JLG(pw, pv, reference);
    if (pw) {
        uint64_t *mask = (uint64_t *) *pw;
        for (int i = 0; i < campaigns; i++) {
            if (CHECK_BIT(mask, i)) {
                found_campaigns++;
            }
        }
    }
    ce = clock();
    printf("test #3 bitmaps, found: %d, clocks: %ld, sec: %f\n", found_campaigns, (long) (ce - cs), (double) (ce - cs) / CLOCKS_PER_SEC);

Результат: clocks: 1393, sec: 0.001393.

В 5 раз быстрее чем перебор. Неплохо, но можно лучше.

Ускоряем использование битовых масок

Для ускорения будет использовать встроенную функцию gcc __builtin_ffsll, которая возвращает индекс первого установленого бита.

Тестируем.

    found_campaigns = 0;
    cs = clock();
    JLG(pw, pv, reference);
    if (pw) {
        uint64_t *mask = (uint64_t *) *pw;
        for (int i = 0; i < (campaigns / 64 + 1); i++) {
            uint64_t m = mask[i];
            while (m) {
                int n = __builtin_ffsll(m);
                m &= ~(1ULL << (n - 1));
                found_campaigns++;
            }
        }
    }
    ce = clock();
    printf("test #4 bitmaps 2, found: %d, clocks: %ld, sec: %f\n", found_campaigns, (long) (ce - cs), (double) (ce - cs) / CLOCKS_PER_SEC);

Результат: clocks: 28, sec: 0.000028.

В этом месте мои глаза полезли на лоб. Ускорение в 250 раз по сравнению с полным перебором.

Как работает процессор, какие есть SSE, AVX и тому подобные инструкции, как устроен кеш — эти знания позволяют ускорить работу программы, часто существенно. Но красивый алгоритм может ускорить работу на порядки.

Текст программы можно посмотреть здесь: https://github.com/saterenko/labs/tree/master/bit_limits

P.S.

Спасибо PkXwmpgN за найденную ошибку в инициализации, за-за неё заполнялся не весь массив тестовых данных, а только его десятая часть. После исправления ошибки время выполнения последнего теста увеличилось на порядок, т.е. ускорение по сравнению с полным перебором — 25 раз, глаза на лоб больше не лезут, но тоже впечатляет ;)