Как улучшить производительность std :: set_intersection в C ++? - PullRequest
0 голосов
/ 19 февраля 2019

Во время экспериментов с std :: set в C ++ и set () в Python я столкнулся с проблемой производительности, которую я не могу объяснить.Установите пересечение в C ++ по крайней мере в 3 раза медленнее, чем в Python.

Так кто-нибудь может указать мне на оптимизацию, которая может быть сделана для кода C ++, и / или объяснить, как Python делает это намного быстрее?

Я ожидаю, что оба они используют похожий алгоритм со сложностью O (n), пока множество упорядочено.Но, вероятно, Python делает некоторые оптимизации, чтобы достичь меньшего коэффициента.

set_bench.cc

#include <iostream>
#include <set>
#include <algorithm>
#include <iterator>
#include <chrono>
#include <functional>
#include <thread>

void elapsed(std::function<void()> f, const std::string& s)
{
    auto start = std::chrono::steady_clock::now();
    f();
    std::chrono::duration<double> elapsed = std::chrono::steady_clock::now() - start;
    std::cout << s << " " << elapsed.count() << " seconds" << std::endl;
}

template <typename T>
void fill_set(std::set<T>& s, T start, T end, T step)
{
    for (T i = start; i < end; i += step) {
        s.emplace(i);
    }
}

template <typename T>
void intersect(const std::set<T>& s1, const std::set<T>& s2, std::set<T>& result)
{
    std::set_intersection(s1.begin(), s1.end(),
                            s2.begin(), s2.end(),
                            std::inserter(result, result.begin()));
}

int main()
{
    std::set<int64_t> s1;
    std::set<int64_t> s2;
    std::set<int64_t> s3;

    elapsed(std::bind(fill_set<int64_t>, std::ref(s1), 8, 1000*1000*100, 13), "fill s1 took");
    elapsed(std::bind(fill_set<int64_t>, std::ref(s2), 0, 1000*1000*100, 7), "fill s2 took");

    std::cout << "s1 length = " << s1.size() << ", s2 length = " << s2.size() << std::endl;

    elapsed(std::bind(intersect<int64_t>, std::ref(s1), std::ref(s2), std::ref(s3)), "intersect s1 and s2 took");

    std::cout << "s3 length = " << s3.size() << std::endl;

    // sleep to let check memory consumption
    // while (true) std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}

set_bench.py ​​

#!/usr/bin/env python3

import time

def elapsed(f, s):
    start = time.monotonic()
    f()
    elapsed = time.monotonic() - start
    print(f'{s} {elapsed} seconds')

def fill_set(s, start, end, step=1):
    for i in range(start, end, step):
        s.add(i)

def intersect(s1, s2, result):
    result.update(s1 & s2)

s1 = set()
s2 = set()

elapsed(lambda : fill_set(s1, 8, 1000*1000*100, 13), 'fill s1 took')
elapsed(lambda : fill_set(s2, 0, 1000*1000*100, 7), 'fill s2 took')

print(f's1 length = {len(s1)}, s2 length = {len(s2)}')


s3 = set()

elapsed(lambda: intersect(s1, s2, s3), 'intersect s1 and s2 took')

print(f's3 length = {len(s3)}')

# sleep to let check memory consumption
# while True: time.sleep(1)

Вот результаты запуска этой программы вследующая среда:

  • clang версия 7.0.1
  • gcc 8.2.0
  • Python 3.7.2
  • i7-7700 CPU @ 3,60 ГГц
$ clang -lstdc++ -O0 set_bench.cc -o set_bench && ./set_bench
fill s1 took 5.38646 seconds
fill s2 took 10.5762 seconds
s1 length = 7692308, s2 length = 14285715
intersect s1 and s2 took 1.48387 seconds
s3 length = 1098901
$ clang -lstdc++ -O1 set_bench.cc -o set_bench && ./set_bench
fill s1 took 3.31435 seconds
fill s2 took 6.41415 seconds
s1 length = 7692308, s2 length = 14285715
intersect s1 and s2 took 1.01276 seconds
s3 length = 1098901
$ clang -lstdc++ -O2 set_bench.cc -o set_bench && ./set_bench
fill s1 took 1.90269 seconds
fill s2 took 3.85651 seconds
s1 length = 7692308, s2 length = 14285715
intersect s1 and s2 took 0.512727 seconds
s3 length = 1098901
$ clang -lstdc++ -O3 set_bench.cc -o set_bench && ./set_bench
fill s1 took 1.92473 seconds
fill s2 took 3.72621 seconds
s1 length = 7692308, s2 length = 14285715
intersect s1 and s2 took 0.523683 seconds
s3 length = 1098901
$ gcc -lstdc++ -O3 set_bench.cc -o set_bench && time ./set_bench
fill s1 took 1.72481 seconds
fill s2 took 3.3846 seconds
s1 length = 7692308, s2 length = 14285715
intersect s1 and s2 took 0.516702 seconds
s3 length = 1098901
$ python3.7 ./set_bench.py 
fill s1 took 0.9404696229612455 seconds
fill s2 took 1.082577683031559 seconds
s1 length = 7692308, s2 length = 14285715
intersect s1 and s2 took 0.17995300807524472 seconds
s3 length = 1098901

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

Кстати - программа RSS для C ++ стоит 1084896 кБ, а для Python - 1590400 кБ.

Ответы [ 3 ]

0 голосов
/ 20 февраля 2019

В этом посте есть два вопроса:

Q: Как улучшить std::set_intersection производительность в C ++?

Использовать отсортированный std::vector с вместо наборов, это гораздо более дружественно к кешу.Поскольку пересечение выполняется последовательно за один проход, оно будет максимально быстрым.В моей системе я получил 0,04 с время выполнения.Остановитесь здесь, если это все, что вам нужно.

Q: ... как Python делает это намного быстрее?

Илидругими словами ", почему Python установлен быстрее, чем набор C ++? ".Я сосредоточусь на этом вопросе до конца моего поста.

Прежде всего, Python set - это хеш-таблица , а std::set - это двоичное дерево.Поэтому используйте std::unordered_set, чтобы сравнить яблоки с яблоками (на данный момент мы отказываемся от двоичного дерева на основе сложности поиска O ( logN )).

Обратите внимание, что std::set_intersection просто алгоритм двух указателей ;он перебирает два отсортированных набора, сохраняя только совпадающие значения.Помимо его имени, в Python нет ничего общего с set_intersection, который сам по себе является простым циклом:

  • Перебираем меньшую хеш-таблицу
  • Для каждого элемента, если он существует в другой хеш-таблице, добавьте его к результату

Поэтому мы не можем использовать std::set_intersection для несортированных данных, и нам необходимо реализовать цикл:

    for (auto& v : set1) {
        if (set2.find(v) != set2.end()) {
            result.insert(v);
        }
    }

Ничего особенного здесь.К сожалению, хотя прямое применение этого алгоритма на std::unordered_set все еще медленнее в 3 раза. Как это может быть?

  1. Мы видим, что размер входных данных составляет> 100 МБ.Это не помещается в 8 МБ кэш-памяти i7-7700, что означает, что чем больше работы вы сможете уместить в пределах 8 МБ, тем быстрее будет работать ваша программа.

  2. Python используетспециальная форма «плотная хеш-таблица» аналогична PHP-хеш-таблице (обычно это класс с открытой адресацией хеш-таблиц), тогда как C ++ std::unordered_set обычно представляет собой наивную или вектор-списки , хеш-таблицу.Плотная структура намного более удобна для кэша и, следовательно, быстрее.Подробнее о реализации см. dictobject.c и setobject.c .

  3. Встроенный C ++ std::hash<long> слишком сложен для ужеуникальный входной набор данных, который вы генерируете.Python, с другой стороны, использует функцию хеширования тождества (без операции) для целых чисел до 2 30 (см. long_hash).Столкновения амортизируются LCG, встроенным в их хеш-таблицу реализации.Вы не можете соответствовать этому со стандартными функциями библиотеки C ++;хеш идентичности здесь, к сожалению, снова приведет к слишком разреженной хеш-таблице.

  4. Python использует специальный распределитель памяти pymalloc , который похож на jemalloc и оптимизирован для локальности данных.Как правило, он превосходит встроенный Linux tcmalloc, который обычно использует программа на C ++.

Обладая этими знаниями, мы можем придумать аналогичную версию C ++, чтобы продемонстрировать техническую осуществимость:

#include <iostream>
#include <unordered_set>
#include <algorithm>
#include <iterator>
#include <chrono>
#include <functional>
#include <thread>
#include <tuple>
#include <string>

using namespace std::chrono_literals;

void elapsed(std::function<void()> f, const std::string& s)
{
    auto start = std::chrono::steady_clock::now();
    f();
    auto end = std::chrono::steady_clock::now();
    std::cout << s << " " << (end - start) / 1.0s << " seconds" << std::endl;
}

template <typename T>
struct myhash {
    size_t operator()(T x) const {
        return x / 5; // cheating to improve data locality
    }
};

template <typename T>
using myset = std::unordered_set<T, myhash<T>>;

template <typename T>
void fill_set(myset<T>& s, T start, T end, T step)
{
    s.reserve((end - start) / step + 1);
    for (T i = start; i < end; i += step) {
        s.emplace(i);
    }
}

template <typename T>
void intersect(const myset<T>& s1, const myset<T>& s2, myset<T>& result)
{
    result.reserve(s1.size() / 4); // cheating to compete with a better memory allocator
    for (auto& v : s1)
    {
        if (s2.find(v) != s2.end())
            result.insert(v);
    }
}

int main()
{
    myset<int64_t> s1;
    myset<int64_t> s2;
    myset<int64_t> s3;

    elapsed(std::bind(fill_set<int64_t>, std::ref(s1), 8, 1000 * 1000 * 100, 13), "fill s1 took");
    elapsed(std::bind(fill_set<int64_t>, std::ref(s2), 0, 1000 * 1000 * 100, 7), "fill s2 took");

    std::cout << "s1 length = " << s1.size() << ", s2 length = " << s2.size() << std::endl;

    elapsed(std::bind(intersect<int64_t>, std::ref(s1), std::ref(s2), std::ref(s3)), "intersect s1 and s2 took");

    std::cout << "s3 length = " << s3.size() << std::endl;
}

С этим кодом я получил 0,28 с времени выполнения в версиях C ++ и Python.

Теперь, если мы хотим побить установленную производительность Python, мы можем удалитьвсе читы и использовать Google dense_hash_set, который реализует открытую адресацию с квадратичным зондированием, в качестве замены (просто нужно позвонить set_empty_object(0)).

С google::dense_hash_set и функцией хеширования без операции мы получаем:

fill s1 took 0.321397 seconds
fill s2 took 0.529518 seconds
s1 length = 7692308, s2 length = 14285714
intersect s1 and s2 took 0.0974416 seconds
s3 length = 1098901

или в 2,8 раза быстрее, чем Python, сохраняя функциональность хэш-набора!


PS Oneподумал бы - почему стандартная библиотека C ++ реализует такую ​​медленную хеш-таблицу?Здесь также применима теорема об отсутствии свободного обеда: решение на основе зондирования не всегда быстро;будучи оппортунистическим решением, оно иногда страдает от «слипания» (бесконечного исследования занятого пространства).И когда это происходит, производительность падает экспоненциально.Идея реализации стандартной библиотеки заключалась в том, чтобы гарантировать предсказуемую производительность для всех возможных входных данных.К сожалению, хотя эффект кеширования на современном оборудовании слишком велик, чтобы пренебрегать им, как объясняет Чэндлер Каррут в в своем выступлении .

0 голосов
/ 20 февраля 2019

Использование отсортированного vector значительно превзойдет set в этом тесте:

#include <iostream>
#include <vector>
#include <algorithm>
#include <iterator>
#include <chrono>
#include <functional>
#include <thread>

void elapsed(std::function<void()> f, const std::string& s)
{
    auto start = std::chrono::steady_clock::now();
    f();
    std::chrono::duration<double> elapsed = std::chrono::steady_clock::now() - start;
    std::cout << s << " " << elapsed.count() << " seconds" << std::endl;
}

template <typename T>
void fill_set(std::vector<T>& s, T start, T end, T step)
{
    for (T i = start; i < end; i += step) {
        s.emplace_back(i);
    }
    std::sort(s.begin(), s.end());
}

template <typename T>
void intersect(const std::vector<T>& s1, const std::vector<T>& s2, std::vector<T>& result)
{
    std::set_intersection(s1.begin(), s1.end(),
                            s2.begin(), s2.end(),
                            std::inserter(result, result.begin()));
}

int main()
{
    std::vector<int64_t> s1;
    std::vector<int64_t> s2;
    std::vector<int64_t> s3;

    elapsed(std::bind(fill_set<int64_t>, std::ref(s1), 8, 1000*1000*100, 13), "fill s1 took");
    elapsed(std::bind(fill_set<int64_t>, std::ref(s2), 0, 1000*1000*100, 7), "fill s2 took");

    std::cout << "s1 length = " << s1.size() << ", s2 length = " << s2.size() << std::endl;

    elapsed(std::bind(intersect<int64_t>, std::ref(s1), std::ref(s2), std::ref(s3)), "intersect s1 and s2 took");

    std::cout << "s3 length = " << s3.size() << std::endl;

    // sleep to let check memory consumption
    // while (true) std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}

Для меня (clang / libc ++ -O3) это взяло результаты из:

fill s1 took 2.01944 seconds
fill s2 took 3.98959 seconds
s1 length = 7692308, s2 length = 14285715
intersect s1 and s2 took 1.55453 seconds
s3 length = 1098901

до:

fill s1 took 0.143026 seconds
fill s2 took 0.20209 seconds
s1 length = 7692308, s2 length = 14285715
intersect s1 and s2 took 0.0548819 seconds
s3 length = 1098901

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

0 голосов
/ 19 февраля 2019

Вы не сравниваете подобное с подобным.

Наборы Python являются неупорядоченными (хэш) наборами.std::set<> - это упорядоченный набор (двоичное дерево).

Из документации по питону:

5.4.Наборы Python также включает в себя тип данных для наборов. Набор представляет собой неупорядоченную коллекцию без повторяющихся элементов.Основное использование включает тестирование членства и устранение дублирующихся записей.Объекты-наборы также поддерживают математические операции, такие как объединение, пересечение, разность и симметричная разность.

рефакторинг для сравнения как с подобным:

#include <iostream>
#include <unordered_set>
#include <algorithm>
#include <iterator>
#include <chrono>
#include <functional>
#include <thread>
#include <tuple>

void elapsed(std::function<void()> f, const std::string& s)
{
    auto start = std::chrono::steady_clock::now();
    f();
    std::chrono::duration<double> elapsed = std::chrono::steady_clock::now() - start;
    std::cout << s << " " << elapsed.count() << " seconds" << std::endl;
}

template <typename T>
void fill_set(std::unordered_set<T>& s, T start, T end, T step)
{
    for (T i = start; i < end; i += step) {
        s.emplace(i);
    }
}

template <typename T>
void intersect(const std::unordered_set<T>& s1, const std::unordered_set<T>& s2, std::unordered_set<T>& result)
{
    auto ordered_refs = [&]()
    {
        if (s1.size() <= s2.size())
            return std::tie(s1, s2);
        else
            return std::tie(s2, s1);
    };

    auto lr = ordered_refs();
    auto& l = std::get<0>(lr);
    auto& r = std::get<1>(lr);
    result.reserve(l.size());

    for (auto& v : l)
    {
        if (auto i = r.find(v) ; i != r.end())
            result.insert(v);
    }
}

int main()
{
    std::unordered_set<int64_t> s1;
    std::unordered_set<int64_t> s2;
    std::unordered_set<int64_t> s3;

    elapsed(std::bind(fill_set<int64_t>, std::ref(s1), 8, 1000*1000*100, 13), "fill s1 took");
    elapsed(std::bind(fill_set<int64_t>, std::ref(s2), 0, 1000*1000*100, 7), "fill s2 took");

    std::cout << "s1 length = " << s1.size() << ", s2 length = " << s2.size() << std::endl;

    elapsed(std::bind(intersect<int64_t>, std::ref(s1), std::ref(s2), std::ref(s3)), "intersect s1 and s2 took");

    std::cout << "s3 length = " << s3.size() << std::endl;

    // sleep to let check memory consumption
    // while (true) std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}

производительность будет зависеть от вашего комплекта.

Я подозреваю, что вы можете значительно увеличить производительность с помощью специального распределителя.По умолчанию это потокобезопасный и т. Д.

Сказав это, на моей машине я увидел только 20% ускорение с неупорядоченной версией.Я бы рискнул предположить, что код пересечения Python был оптимизирован вручную.

Для справки, исходный код Python находится здесь: https://github.com/python/cpython/blob/master/Objects/setobject.c

...