Сортировки

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

Мы в EDISON часто разрабатываем умные алгоритмы в наших проектах.


Например, для одного заказчика был разработан специальный способ передачи данных с помощью светодиода. Изначально заказчик планировал поиск и анализ конкретного светового пятная в видеопотоке, однако мы предложили более элегантное и надёжное решение.

Мы очень любим computer science ;-)

Основная идея заключается в использовании относительно небольшой (по сравнению с основным массивом) вспомогательной кучи, которая выполняет роль приоритетной очереди. В этой куче сравниваются элементы на нижних уровнях, в результате чего меньшие элементы (в данном случае дерево у нас MIN-HEAP) поднимаются наверх, а в корень всплывает текущий минимум из той порции элементов массива, которые попали в эту кучу. Минимум переносится в дополнительный массив победителей, в результате чего в куче происходит сравнение/перемещение оставшихся элементов и вот уже в корне дерева новый минимум. Отметим, что при такой системе очередной минимум по значению больше чем предыдущий тогда легко собирается новый упорядоченный массив победителей, где новые минимумы просто добавляются в конец дополнительного массива.

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

Главной тонкостью является то, что кроме победителей есть ещё и проигравшие. Если в массив победителей уже перенесены какие-то элементы, то если необработанные элементы меньше, чем эти победители то просеивать их через турнирное дерево на этом этапе уже нет смысла вставка этих элементов в массив победителей будет слишком дорогой (в конец их уже не поставишь, а чтобы поставить в начало придётся сдвинуть уже вставленные раннее минимумы). Такие элементы, не успевшие вписаться в массив победителей отправляются в массив проигравших они пройдут через турнирное дерево в одной из следующих итерации, когда все действия повторяются для оставшихся неудачников.

В Интернете непросто найти реализацию этого алгоритма, однако в Ютубе обнаружилась понятная и симпатичная реализация на Руби. В разделе Ссылки ещё упомянут код на Java, но там выглядит не настолько читабельно.

  # Библиотека для работы с приоритетной очередью в виде кучи  require_relative "pqueue"  # Максимальный размер для приоритетной очереди  MAX_SIZE = 16  def tournament_sort(array)    # Если массив маленький, то упрощённый "турнир"    return optimal_tourney_sort(array) if array.size <= MAX_SIZE    bracketize(array) # Если массив большой, то стандартный "турнир"  end  # Пропускаем элементы через турнирную сетку  def bracketize(array)    size = array.size    pq = PriorityQueue.new    # Заполняем очередь первыми элементами    pq.add(array.shift) until pq.size == MAX_SIZE    winners = [] # Сбор "победителей"    losers = [] # Сбор "проигравших"    # Пока в основном массиве есть элементы    until array.empty?      # Массив "победителей" пока пуст?      if winners.empty?        # Из корня переносим в массив "победителей"        winners << pq.peek        # Восстановление кучи после переноса корня        pq.remove       end      # Если очередной элемент из массива больше, чем последний "победитель"      if array.first > winners.last        pq.add(array.shift) # Переносим элемент в очередь на вакантное место      else # Если очередной элемент меньше или равен последнему "победителю"        losers << array.shift # Переносим элемент в массив "проигравших"      end      # Если куча пока не пуста, корень переносим в массив "победителей"      winners << pk.peek unless pk.peek.nil?      pq.remove # Восстановление кучи после переноса корня    end    # Основной массив пуст, но может в куче ещё что-то осталось?    until pq.heap.empty?      winners << pq.peek # Корень переносим в массив "победителей"      pq.remove # Восстановление кучи после переноса корня    end    # Если массив "проигравших" остался пуст, то, значит,    # массив "победителей" - итоговый отсортированный массив    return winners if losers.empty?    # Если ещё не закончили, то к массиву "проигравших"    # припишем массив "победителей"    array = losers + winners    array.pop while array.size > size    bracketize(array) # Запускаем процесс снова  end  # Упрощённый турнир если массив меньше чем очередь  def optimal_tourney_sort(array)    sorted_array = [] # Заготовка для отсортированного массива    pq = PriorityQueue.new    array.each { |num| pq.add(num) } # Переносим все элементы в мини-кучу    until pq.heap.empty? # Пока мини-куча не пуста      sorted_array << pq.heap[0]       pq.remove # Восстановление кучи после переноса корня    end    sorted_array # Итоговый массив  end  # Тестирование  if $PROGRAM_NAME == __FILE__    # Генерируем массив    shuffled_array = Array.new(30) { rand(-100 ... 100) }    # Печать необработанного массива    puts "Random Array: #{shuffled_array}"    # Печать обработанного массива    puts "Sorted Array: #{tournament_sort(shuffled_array)}"  end

Это наивная реализация, в которой после каждого разделения на проигравших и победителей эти массивы объединяются и затем для объединённого массива все действия повторяются снова. Если в объединённом массиве будут сначала идти проигравшие элементы, то просейка через турнирную кучу упорядочит их совместно с победителями.


Данная реализация достаточно проста и наглядна, но эффективной её не назовёшь. Отсортированные победители проходят через турнирное дерево неоднократно, что выглядит, очевидно, излишним. Предполагаю, что временная сложность здесь логарифмично-квадратичная, O(n log²n).

Вариант с многопутевым слиянием

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

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


В таком виде алгоритм может уже вполне найти полезное применение. Например, если приходится работать с big data, то по ходу процесса с помощью турнирной кучи будут выходить пакеты упорядоченных данных, с которыми уже можно что-нибудь делать, пока обрабатывается и всё остальное.

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

В финале сезона

Серия статей о сортировках кучей почти завершена. Осталось рассказать о самой эффективной из них.

Ссылки

Tournament sort

Priority queue

Tournament sort in Java

The Theory Behind the The Theory Behind the z/Architecture Sort Assist Instructions

Using Tournament Trees to Sort

Tournament Sort Demo Ruby

Tournament Sort Visualization

Tournament Sort Data Structure UGC NET DS

Tournament Sort Algorithm a Heapsort variant

Статьи серии:

• Excel-приложение AlgoLab.xlsm
• Сортировки обменами
• Сортировки вставками
• Сортировки выбором
  • Сортировки кучей: n-нарные пирамиды
  • Сортировки кучей: числа Леонардо
  • Сортировки кучей: слабая куча
  • Сортировки кучей: декартово дерево
  • Сортировки кучей: турнирное дерево
  • Сортировки кучей: просейка снизу-вверх
• Сортировки слиянием
• Сортировки распределением
• Гибридные сортировки

В приложение AlgoLab добавлена сегодняшняя сортировка, кто пользуется обновите excel-файл с макросами.

В комментариях к ячейке с название сортировки можно указать некоторые настройки.
Переменная way скольки-путевое турнирное дерево (просто на всякий случай предусмотрена возможность делать это дерево не только двоичным, но и троичным, четверичным и пятиричным).
Переменная queue это размер первоначальной очереди (количество узлов в самом нижнем уровне дерева). Так как деревья полные, то если, к примеру при way=2 указать queue=5, то размер очереди будет увеличен до ближайшей степени двойки (в данном случае до 8).
Переменная NWayMerge принимает значение 1 или 0 с помощью неё указывается, надо ли применять многопутевое слияние или нет.

Начнем с того, что данному вопросу уделяется мало времени и приходится гуглить данный вопрос.
Код программы используемый в данной статье, я переписывал пару раз. Всегда было интересно насколько одна сортировка будет быстрее другой. Их как бы все студенты проходят, но в основном как переписывание псевдоалгоритма на лекции в код на каком-нибудь языке. Может быть данная статья будет полезна для какого-нибудь начинающего программиста.
Рассмотрим 5 сортировок. Это пузырьковая(bubble), шейкерная(shake), пирамидальная(heap), вставками(insertion) и быстрая(quick).

Для анализа их скорости будет использоваться функция clock() до сортировки и она же после, потом берется их разность и мы узнаем время работы сортировки. Я использовал 100 итераций по 1000 значений заданных в векторах и одном листе для тестирования встроенной функции sort() из stl. Каждой сортировке даются одинаково разбросанные по массивам числа на каждой итерации. После чего время записывается в переменную mean каждой сортировки и делится по итогу на количество итераций. Так мы узнаем среднее время работы каждой сортировки и сможем в итоге их сравнить по скорости при одинаковых исходных данных. Данные вносятся в массивы функцией rand().
Файл Sorts.h:

#pragma once#include <iostream>#include <list>#include <vector>#include <iterator>template <typename T> class Sorts{public:std::list<T> arrayList;std::vector<T> bubbleArray,insertionArray,heapArray,shakeArray;float BubbleSort(){std::cout <<"Time to Bubble>" << std::endl;unsigned int start_time = clock(); // начальное времяint size = bubbleArray.size();for (int i = 1; i < size; i++)for (int j = size-1; j >=i; j--)if (bubbleArray[j-1] > bubbleArray[j])swap(&bubbleArray, j - 1, j);unsigned int end_time = clock(); // конечное времяunsigned int search_time = end_time - start_time; // искомое времяstd::cout << (float)search_time / CLOCKS_PER_SEC << std::endl;return (float)search_time / CLOCKS_PER_SEC;}float InsertionSort(){std::cout << "Time to Insertion>" << std::endl;unsigned int start_time = clock(); // начальное времяint size = insertionArray.size();for (int i = 1; i < size; i++){T tmp = insertionArray[i];int j = i;while (j > 0 && insertionArray[j - 1] > tmp){insertionArray[j] = insertionArray[j - 1];j = j - 1;}insertionArray[j] = tmp;}unsigned int end_time = clock(); // конечное времяunsigned int search_time = end_time - start_time; // искомое времяstd::cout << (float)search_time / CLOCKS_PER_SEC << std::endl;return (float)search_time / CLOCKS_PER_SEC;}void swap(std::vector<T> *v, int n, int m){T tmp = (*v)[n];(*v)[n] = (*v)[m];(*v)[m] = tmp;}float HeapSort(){std::cout << "Time to Heap>" << std::endl;unsigned int start_time = clock(); // начальное времяint size = heapArray.size();for (int j = 0; j < size; j++){for (int i = size / 2 - 1 - j / 2; i > -1; i--){if (2 * i + 2 <= size - 1 - j){if (heapArray[2 * i + 1] > heapArray[2 * i + 2]){if (heapArray[i] < heapArray[2 * i + 1]){swap(&heapArray, i, 2 * i + 1);}}elseif (heapArray[i] < heapArray[2 * i + 2]){swap(&heapArray, i, 2 * i + 2);}}elseif (2 * i + 1 <= size - 1 - j)if (heapArray[i] < heapArray[2 * i + 1])swap(&heapArray, i, 2 * i + 1);}swap(&heapArray, 0, size - 1 - j);}unsigned int end_time = clock(); // конечное времяunsigned int search_time = end_time - start_time; // искомое времяstd::cout << (float)search_time / CLOCKS_PER_SEC << std::endl;return (float)search_time / CLOCKS_PER_SEC;}float ShakeSort(){std::cout << "Time to Shake>" << std::endl;unsigned int start_time = clock(); // начальное времяint size = shakeArray.size();int left = 0;int right = size - 1;do {for (int i = left; i < right; i++) {if (shakeArray[i] > shakeArray[i + 1])swap(&shakeArray,i,i+1);}right--;for (int i = right; i > left; i--) {if (shakeArray[i] < shakeArray[i - 1])swap(&shakeArray, i-1, i);}left++;} while (left < right);unsigned int end_time = clock(); // конечное времяunsigned int search_time = end_time - start_time; // искомое времяstd::cout << (float)search_time / CLOCKS_PER_SEC << std::endl;return (float)search_time / CLOCKS_PER_SEC;}void PrintArray(int num){switch (num){case 0:for (typename std::list<T>::iterator it = arrayList.begin(); it != arrayList.end(); it++)std::cout << (*it) << " ";break;case 1:for (typename std::vector<T>::iterator it = bubbleArray.begin(); it != bubbleArray.end(); it++)std::cout << (*it) << " ";break;case 2:for (typename std::vector<T>::iterator it = shakeArray.begin(); it != shakeArray.end(); it++)std::cout << (*it) << " ";break;case 3:for (typename std::vector<T>::iterator it = heapArray.begin(); it != heapArray.end(); it++)std::cout << (*it) << " ";break;case 4:for (typename std::vector<T>::iterator it = insertionArray.begin(); it != insertionArray.end(); it++)std::cout << (*it) << " ";break;default:break;}std::cout << std::endl;}};

Замечу что можно использовать не только целые числа, но и вещественные и символы.
Файл основной программы:

#include "Sorts.h"int main(){std::vector<float> vq, vb, vs, vh, vi;float meanq = 0, meanb = 0, means = 0, meanh = 0, meani = 0;const int N = 100;srand(time(0));for (int i = 0; i < N; i++){std::cout << i+1 << " iteration" << std::endl;const int iSize = 1000;auto sort = new Sorts<int>();for (int i = 0; i < iSize; i++){int num = rand() % iSize;sort->arrayList.push_back(num);sort->bubbleArray.push_back(num);sort->shakeArray.push_back(num);sort->heapArray.push_back(num);sort->insertionArray.push_back(num);}std::cout << "Time to Quick sort from stl>" << std::endl;unsigned int start_time = clock(); // начальное времяsort->arrayList.sort();unsigned int end_time = clock(); // конечное времяunsigned int search_time = end_time - start_time; // искомое времяstd::cout << (float)search_time / CLOCKS_PER_SEC << std::endl;vq.push_back((float)search_time / CLOCKS_PER_SEC);vb.push_back(sort->BubbleSort());vs.push_back(sort->ShakeSort());vh.push_back(sort->HeapSort());vi.push_back(sort->InsertionSort());meanq += vq[i];meanb += vb[i];means += vs[i];meanh += vh[i];meani += vi[i];//sort->PrintArray(0);//sort->PrintArray(1);//sort->PrintArray(2);//sort->PrintArray(3);//sort->PrintArray(4);sort->arrayList.clear();sort->bubbleArray.clear();sort->shakeArray.clear();sort->heapArray.clear();sort->insertionArray.clear();std::cout << "end of "<< i + 1 <<" iteration" << std::endl;}std::cout << "Results:" << std::endl;std::cout << "Mean quick=" << (float)meanq / N << std::endl;std::cout << "Mean bubble=" << (float)meanb / N << std::endl;std::cout << "Mean shake=" << (float)means / N  << std::endl;std::cout << "Mean heap=" << (float)meanh / N  << std::endl;std::cout << "Mean insertion=" << (float)meani / N << std::endl;return 0;}

Каковы же результаты?
С большим отрывом идет sort из stl, потом вставками, пирамидальная, шейкерная и заканчивает пузырьковая.
Quick 0.00225 ms
Insertion 0.04482 ms
Heap 0.07025 ms
Shake 0.14186 ms
Bubble 0.14324 ms
В принципе слишком большие массивы данных долго сортируются, но quicksort справляется на порядки быстрее остальных.