Как ускорить сортировку коротких массивов?

Question

В проекте часто приходится сортировать массивы с такими характеристиками:

Размер 2 - 32 (чаще всего 2, 3, 4, 8)
Упорядочены случайно
Тип хранимых данных uint32_t

Стандартная сортировка работает с такими данными достаточно медленно.

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

Вот минимальная часть кода:

Вспомогательная функция, которая меняет второй элемент с первым, если второй меньше первого.

namespace mi { namespace impl {
  template<class ItTy, class PredicateTy>
  void swap_if(ItTy first, ItTy second, PredicateTy predicate) {
    if (predicate(*second, *first)) {
      std::swap(*second, *first);
    }
  }
}}  // namespace mi::impl

И реализация сортировки:

template<size_t N>
struct NetworkSort;

template<>
struct NetworkSort<30U> {
  public:
    template<class ItTy, class PredicateTy>
    void operator()(ItTy first, ItTy last, PredicateTy predicate) {
      mi::impl::swap_if(first + 0u, first + 2u, predicate);
      mi::impl::swap_if(first + 4u, first + 6u, predicate);
      mi::impl::swap_if(first + 8u, first + 10u, predicate);
      mi::impl::swap_if(first + 12u, first + 14u, predicate);
      mi::impl::swap_if(first + 16u, first + 18u, predicate);
      mi::impl::swap_if(first + 20u, first + 22u, predicate);
      mi::impl::swap_if(first + 24u, first + 26u, predicate);
      mi::impl::swap_if(first + 0u, first + 4u, predicate);
      mi::impl::swap_if(first + 8u, first + 12u, predicate);
      mi::impl::swap_if(first + 16u, first + 20u, predicate);
      mi::impl::swap_if(first + 24u, first + 28u, predicate);
      mi::impl::swap_if(first + 2u, first + 6u, predicate);
      mi::impl::swap_if(first + 10u, first + 14u, predicate);
      mi::impl::swap_if(first + 18u, first + 22u, predicate);
      mi::impl::swap_if(first + 0u, first + 8u, predicate);
      mi::impl::swap_if(first + 16u, first + 24u, predicate);
      mi::impl::swap_if(first + 2u, first + 10u, predicate);
      mi::impl::swap_if(first + 18u, first + 26u, predicate);
      mi::impl::swap_if(first + 4u, first + 12u, predicate);
      mi::impl::swap_if(first + 20u, first + 28u, predicate);
      mi::impl::swap_if(first + 6u, first + 14u, predicate);
      mi::impl::swap_if(first + 0u, first + 16u, predicate);
      mi::impl::swap_if(first + 2u, first + 18u, predicate);
      mi::impl::swap_if(first + 4u, first + 20u, predicate);
      mi::impl::swap_if(first + 6u, first + 22u, predicate);
      mi::impl::swap_if(first + 8u, first + 24u, predicate);
      mi::impl::swap_if(first + 10u, first + 26u, predicate);
      mi::impl::swap_if(first + 12u, first + 28u, predicate);
      mi::impl::swap_if(first + 10u, first + 20u, predicate);
      mi::impl::swap_if(first + 12u, first + 18u, predicate);
      mi::impl::swap_if(first + 6u, first + 24u, predicate);
      mi::impl::swap_if(first + 26u, first + 28u, predicate);
      mi::impl::swap_if(first + 14u, first + 22u, predicate);
      mi::impl::swap_if(first + 2u, first + 4u, predicate);
      mi::impl::swap_if(first + 8u, first + 16u, predicate);
      mi::impl::swap_if(first + 2u, first + 8u, predicate);
      mi::impl::swap_if(first + 14u, first + 26u, predicate);
      mi::impl::swap_if(first + 4u, first + 16u, predicate);
      mi::impl::swap_if(first + 22u, first + 28u, predicate);
      mi::impl::swap_if(first + 4u, first + 8u, predicate);
      mi::impl::swap_if(first + 10u, first + 12u, predicate);
      mi::impl::swap_if(first + 18u, first + 20u, predicate);
      mi::impl::swap_if(first + 22u, first + 26u, predicate);
      mi::impl::swap_if(first + 6u, first + 16u, predicate);
      mi::impl::swap_if(first + 14u, first + 24u, predicate);
      mi::impl::swap_if(first + 12u, first + 16u, predicate);
      mi::impl::swap_if(first + 20u, first + 24u, predicate);
      mi::impl::swap_if(first + 6u, first + 10u, predicate);
      mi::impl::swap_if(first + 14u, first + 18u, predicate);
      mi::impl::swap_if(first + 6u, first + 8u, predicate);
      mi::impl::swap_if(first + 10u, first + 12u, predicate);
      mi::impl::swap_if(first + 14u, first + 16u, predicate);
      mi::impl::swap_if(first + 18u, first + 20u, predicate);
      mi::impl::swap_if(first + 22u, first + 24u, predicate);
      mi::impl::swap_if(first + 12u, first + 14u, predicate);
      mi::impl::swap_if(first + 16u, first + 18u, predicate);
      mi::impl::swap_if(first + 1u, first + 3u, predicate);
      mi::impl::swap_if(first + 5u, first + 7u, predicate);
      mi::impl::swap_if(first + 9u, first + 11u, predicate);
      mi::impl::swap_if(first + 13u, first + 15u, predicate);
      mi::impl::swap_if(first + 17u, first + 19u, predicate);
      mi::impl::swap_if(first + 21u, first + 23u, predicate);
      mi::impl::swap_if(first + 25u, first + 27u, predicate);
      mi::impl::swap_if(first + 1u, first + 5u, predicate);
      mi::impl::swap_if(first + 9u, first + 13u, predicate);
      mi::impl::swap_if(first + 17u, first + 21u, predicate);
      mi::impl::swap_if(first + 25u, first + 29u, predicate);
      mi::impl::swap_if(first + 3u, first + 7u, predicate);
      mi::impl::swap_if(first + 11u, first + 15u, predicate);
      mi::impl::swap_if(first + 19u, first + 23u, predicate);
      mi::impl::swap_if(first + 1u, first + 9u, predicate);
      mi::impl::swap_if(first + 17u, first + 25u, predicate);
      mi::impl::swap_if(first + 3u, first + 11u, predicate);
      mi::impl::swap_if(first + 19u, first + 27u, predicate);
      mi::impl::swap_if(first + 5u, first + 13u, predicate);
      mi::impl::swap_if(first + 21u, first + 29u, predicate);
      mi::impl::swap_if(first + 7u, first + 15u, predicate);
      mi::impl::swap_if(first + 1u, first + 17u, predicate);
      mi::impl::swap_if(first + 3u, first + 19u, predicate);
      mi::impl::swap_if(first + 5u, first + 21u, predicate);
      mi::impl::swap_if(first + 7u, first + 23u, predicate);
      mi::impl::swap_if(first + 9u, first + 25u, predicate);
      mi::impl::swap_if(first + 11u, first + 27u, predicate);
      mi::impl::swap_if(first + 13u, first + 29u, predicate);
      mi::impl::swap_if(first + 11u, first + 21u, predicate);
      mi::impl::swap_if(first + 13u, first + 19u, predicate);
      mi::impl::swap_if(first + 7u, first + 25u, predicate);
      mi::impl::swap_if(first + 27u, first + 29u, predicate);
      mi::impl::swap_if(first + 15u, first + 23u, predicate);
      mi::impl::swap_if(first + 3u, first + 5u, predicate);
      mi::impl::swap_if(first + 9u, first + 17u, predicate);
      mi::impl::swap_if(first + 3u, first + 9u, predicate);
      mi::impl::swap_if(first + 15u, first + 27u, predicate);
      mi::impl::swap_if(first + 5u, first + 17u, predicate);
      mi::impl::swap_if(first + 23u, first + 29u, predicate);
      mi::impl::swap_if(first + 5u, first + 9u, predicate);
      mi::impl::swap_if(first + 11u, first + 13u, predicate);
      mi::impl::swap_if(first + 19u, first + 21u, predicate);
      mi::impl::swap_if(first + 23u, first + 27u, predicate);
      mi::impl::swap_if(first + 7u, first + 17u, predicate);
      mi::impl::swap_if(first + 15u, first + 25u, predicate);
      mi::impl::swap_if(first + 13u, first + 17u, predicate);
      mi::impl::swap_if(first + 21u, first + 25u, predicate);
      mi::impl::swap_if(first + 7u, first + 11u, predicate);
      mi::impl::swap_if(first + 15u, first + 19u, predicate);
      mi::impl::swap_if(first + 7u, first + 9u, predicate);
      mi::impl::swap_if(first + 11u, first + 13u, predicate);
      mi::impl::swap_if(first + 15u, first + 17u, predicate);
      mi::impl::swap_if(first + 19u, first + 21u, predicate);
      mi::impl::swap_if(first + 23u, first + 25u, predicate);
      mi::impl::swap_if(first + 13u, first + 15u, predicate);
      mi::impl::swap_if(first + 17u, first + 19u, predicate);
      mi::impl::swap_if(first + 0u, first + 1u, predicate);
      mi::impl::swap_if(first + 2u, first + 3u, predicate);
      mi::impl::swap_if(first + 4u, first + 5u, predicate);
      mi::impl::swap_if(first + 6u, first + 7u, predicate);
      mi::impl::swap_if(first + 8u, first + 9u, predicate);
      mi::impl::swap_if(first + 10u, first + 11u, predicate);
      mi::impl::swap_if(first + 12u, first + 13u, predicate);
      mi::impl::swap_if(first + 14u, first + 15u, predicate);
      mi::impl::swap_if(first + 16u, first + 17u, predicate);
      mi::impl::swap_if(first + 18u, first + 19u, predicate);
      mi::impl::swap_if(first + 20u, first + 21u, predicate);
      mi::impl::swap_if(first + 22u, first + 23u, predicate);
      mi::impl::swap_if(first + 24u, first + 25u, predicate);
      mi::impl::swap_if(first + 26u, first + 27u, predicate);
      mi::impl::swap_if(first + 28u, first + 29u, predicate);
      mi::impl::swap_if(first + 1u, first + 16u, predicate);
      mi::impl::swap_if(first + 3u, first + 18u, predicate);
      mi::impl::swap_if(first + 5u, first + 20u, predicate);
      mi::impl::swap_if(first + 7u, first + 22u, predicate);
      mi::impl::swap_if(first + 9u, first + 24u, predicate);
      mi::impl::swap_if(first + 11u, first + 26u, predicate);
      mi::impl::swap_if(first + 13u, first + 28u, predicate);
      mi::impl::swap_if(first + 1u, first + 8u, predicate);
      mi::impl::swap_if(first + 3u, first + 10u, predicate);
      mi::impl::swap_if(first + 5u, first + 12u, predicate);
      mi::impl::swap_if(first + 7u, first + 14u, predicate);
      mi::impl::swap_if(first + 9u, first + 16u, predicate);
      mi::impl::swap_if(first + 11u, first + 18u, predicate);
      mi::impl::swap_if(first + 13u, first + 20u, predicate);
      mi::impl::swap_if(first + 15u, first + 22u, predicate);
      mi::impl::swap_if(first + 17u, first + 24u, predicate);
      mi::impl::swap_if(first + 19u, first + 26u, predicate);
      mi::impl::swap_if(first + 21u, first + 28u, predicate);
      mi::impl::swap_if(first + 1u, first + 4u, predicate);
      mi::impl::swap_if(first + 3u, first + 6u, predicate);
      mi::impl::swap_if(first + 5u, first + 8u, predicate);
      mi::impl::swap_if(first + 7u, first + 10u, predicate);
      mi::impl::swap_if(first + 9u, first + 12u, predicate);
      mi::impl::swap_if(first + 11u, first + 14u, predicate);
      mi::impl::swap_if(first + 13u, first + 16u, predicate);
      mi::impl::swap_if(first + 15u, first + 18u, predicate);
      mi::impl::swap_if(first + 17u, first + 20u, predicate);
      mi::impl::swap_if(first + 19u, first + 22u, predicate);
      mi::impl::swap_if(first + 21u, first + 24u, predicate);
      mi::impl::swap_if(first + 23u, first + 26u, predicate);
      mi::impl::swap_if(first + 25u, first + 28u, predicate);
      mi::impl::swap_if(first + 1u, first + 2u, predicate);
      mi::impl::swap_if(first + 3u, first + 4u, predicate);
      mi::impl::swap_if(first + 5u, first + 6u, predicate);
      mi::impl::swap_if(first + 7u, first + 8u, predicate);
      mi::impl::swap_if(first + 9u, first + 10u, predicate);
      mi::impl::swap_if(first + 11u, first + 12u, predicate);
      mi::impl::swap_if(first + 13u, first + 14u, predicate);
      mi::impl::swap_if(first + 15u, first + 16u, predicate);
      mi::impl::swap_if(first + 17u, first + 18u, predicate);
      mi::impl::swap_if(first + 19u, first + 20u, predicate);
      mi::impl::swap_if(first + 21u, first + 22u, predicate);
      mi::impl::swap_if(first + 23u, first + 24u, predicate);
      mi::impl::swap_if(first + 25u, first + 26u, predicate);
      mi::impl::swap_if(first + 27u, first + 28u, predicate);
    }
};

Главная функция, может кому пригодится:

#define N 30

static void std_sort(benchmark::State& state) {
  auto v = generate_random_vector(state.range(0));

  STD array<int, N> a;

  for (size_t i = 0; i < N; ++i) { a[i] = (rand() % 1000000); }

  for (auto _: state) {
    benchmark::DoNotOptimize(a[0]);
    std::sort(std::begin(a), std::end(a));
  }
}

static void network_sort(benchmark::State& state) {
  auto              v = generate_random_vector(state.range(0));
  MI NetworkSort<N> sorter;

  STD array<int, N> a;

  for (size_t i = 0; i < N; ++i) { a[i] = (rand() % 1000000); }

  for (auto _: state) {
    benchmark::DoNotOptimize(a[0]);
    sorter(a.begin(), a.end(), STD less<int>{});
  }
}

BENCHMARK(network_sort)->Arg(1);
BENCHMARK(std_sort)->Arg(1);

Но почему-то моя реализация работает намного медленнее стандартной сортировки. Хотя у других, вроде как, получается быстрее: Тык#1, Тык#2, Тык#3. Я протестировал их код и он у меня работает медленнее, чем стандартная сортировка.

Вот результаты теста:

Run on (4 X 3600 MHz CPU s)
CPU Caches:
  L1 Data 32 KiB (x4)
  L1 Instruction 32 KiB (x4)
  L2 Unified 256 KiB (x4)
  L3 Unified 6144 KiB (x1)
----------------------------------------------------------
Benchmark                Time             CPU   Iterations
----------------------------------------------------------
network_sort/1        94.5 ns         80.2 ns      8960000
std_sort/1            45.8 ns         36.8 ns     18666667

Меня смущает то, что количество сравнений у std::sort при длине массива n = 30 составляет примерно n*log2(n) ~ 150 В СРЕДНЕМ случае, а у NetworkSort 170 В ЛЮБОМ случае.

Хм.. Может оно и не может быстрее работать?

Мне кажется, что компилятор должен оптимизировать этот код так, чтобы элементы массива менялись местами по несколько за раз. И у меня есть уверенность что какой-нибудь GCC так бы и сделал (я использую VS).

Можно как-то ускорить это? Стараться разделить свапы на несколько потоков, наверное, только замедлит этот код, а больше у меня нет идей.

Answer 1

Проверка на VC++ для массивов 4,8,16,32 показывает, что вставка и при 16 бьет стандартную сортировку. На g++ — и того круче, при 32 :)

Тут данные для VC++ 2019; полный код и результаты для g++ - см. https://ideone.com/EduCM3

Так что считаю свой совет использовать алгоритм сортировки вставками не просто обоснованным, а и доказанным :)

stl sort    : N =  4 : 19972097628447 for   2236 mks
stable sort : N =  4 : 19972097628447 for   2198 mks
Bubble sort : N =  4 : 19972097628447 for   2091 mks
EBubble sort: N =  4 : 19972097628447 for   2242 mks
Insert sort : N =  4 : 19972097628447 for   1792 mks
Select sort : N =  4 : 19972097628447 for   2635 mks


stl sort    : N =  8 : 40006459862344 for   6572 mks
stable sort : N =  8 : 40006459862344 for   6136 mks
Bubble sort : N =  8 : 40006459862344 for   7996 mks
EBubble sort: N =  8 : 40006459862344 for   8875 mks
Insert sort : N =  8 : 40006459862344 for   5301 mks
Select sort : N =  8 : 40006459862344 for   8386 mks


stl sort    : N = 16 : 80077500292087 for  14895 mks
stable sort : N = 16 : 80077500292087 for  13838 mks
Bubble sort : N = 16 : 80077500292087 for  35777 mks
EBubble sort: N = 16 : 80077500292087 for  32958 mks
Insert sort : N = 16 : 80077500292087 for  14141 mks
Select sort : N = 16 : 80077500292087 for  32009 mks


stl sort    : N = 32 : 159981299934791 for  32750 mks
stable sort : N = 32 : 159981299934791 for  33863 mks
Bubble sort : N = 32 : 159981299934791 for 126464 mks
EBubble sort: N = 32 : 159981299934791 for 120895 mks
Insert sort : N = 32 : 159981299934791 for  36501 mks
Select sort : N = 32 : 159981299934791 for 124016 mks

Update

В связи с новыми данными, что на сортировку может влиять диапазон значений — вот новые результаты: https://ideone.com/s2pFfu В смысле данного вопроса особых изменений нет.

БЛОГ НА HUSL

Как ускорить сортировку коротких массивов?

Ответы (1 шт):