Быстрый сортировать частично отсортированный массив

Question

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

Я пишу сейчас простой метод, который сортирует данные. До сих пор я использовал Array.Sort, но PLINQOrderBy превосходит стандарт Array.Sort на больших массивах.

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

Теперь я делаюсь с разделения и сортировки:

public class PartitionSorter 
{ 
    public static void Sort(int[] arr) 
    { 
     var ranges = Range.FromArray(arr); 
     var allDone = new ManualResetEventSlim(false, ranges.Length*2); 
     int completed = 0; 
     foreach (var range in ranges) 
     { 
      ThreadPool.QueueUserWorkItem(r => 
      { 
       var rr = (Range) r; 
       Array.Sort(arr, rr.StartIndex, rr.Length); 
       if (Interlocked.Increment(ref completed) == ranges.Length) 
        allDone.Set(); 
      }, range); 
     } 
     allDone.Wait(); 
    } 
} 

public class Range 
{ 
    public int StartIndex { get; } 
    public int Length { get; } 

    public Range(int startIndex, int endIndex) 
    { 
     StartIndex = startIndex; 
     Length = endIndex; 
    } 

    public static Range[] FromArray<T>(T[] source) 
    { 
     int processorCount = Environment.ProcessorCount; 
     int partitionLength = (int) (source.Length/(double) processorCount); 
     var result = new Range[processorCount]; 
     int start = 0; 
     for (int i = 0; i < result.Length - 1; i++) 
     { 
      result[i] = new Range(start, partitionLength); 
      start += partitionLength; 
     } 
     result[result.Length - 1] = new Range(start, source.Length - start); 
     return result; 
    } 
} 

В результате я получаю массив с особой структурой, например

[1 3 5 | 2 4 7 | 6 8 9] 

Теперь, как я могу использовать эту информацию и закончить сортировку ? Сортировки вставки и другие не используют информацию о том, что данные в блоках уже отсортированы, и нам просто нужно объединить их вместе. Я попытался применить некоторые алгоритмы от Merge sort, но не смог.

Answer 1

Я провел некоторое тестирование с параллельной реализацией Quicksort.

Я протестировал следующий код с помощью сборки RELEASE на Windows x64 10, скомпилированный с C# 6 (Visual Studio 2015), .Net 4.61 и запущенный за пределами любого отладчика.

Мой процессор Quad Core с технологией HyperThreading (который, безусловно, будет помогать любой параллельной реализации!)

Размер массива 20000000 (так довольно большой массив).

я получил следующие результаты:

LINQ OrderBy() took 00:00:14.1328090 
PLINQ OrderBy() took 00:00:04.4484305 
Array.Sort() took 00:00:02.3695607 
Sequential  took 00:00:02.7274400 
Parallel  took 00:00:00.7874578 

PLINQ OrderBy() гораздо быстрее, чем LINQ OrderBy(), но медленнее, чем Array.Sort().

QuicksortSequential() вокруг одной и той же скоростью, как Array.Sort()

Но самое интересное здесь то, что QuicksortParallelOptimised() заметно быстрее на моей системе - так это, безусловно, эффективный способ сортировки, если у вас есть достаточное количество процессорных ядер.

Вот полное компилируемое консольное приложение. Не забудьте запустить его в режиме RELEASE - если вы запустите его в режиме DEBUG, результаты синхронизации будут ужасно неправильными.

using System; 
using System.Diagnostics; 
using System.Linq; 
using System.Threading.Tasks; 

namespace Demo 
{ 
    class Program 
    { 
     static void Main() 
     { 
      int n = 20000000; 
      int[] a = new int[n]; 
      var rng = new Random(937525); 

      for (int i = 0; i < n; ++i) 
       a[i] = rng.Next(); 

      var b = a.ToArray(); 
      var d = a.ToArray(); 

      var sw = new Stopwatch(); 

      sw.Restart(); 
      var c = a.OrderBy(x => x).ToArray(); // Need ToArray(), otherwise it does nothing. 
      Console.WriteLine("LINQ OrderBy() took " + sw.Elapsed); 

      sw.Restart(); 
      var e = a.AsParallel().OrderBy(x => x).ToArray(); // Need ToArray(), otherwise it does nothing. 
      Console.WriteLine("PLINQ OrderBy() took " + sw.Elapsed); 

      sw.Restart(); 
      Array.Sort(d); 
      Console.WriteLine("Array.Sort() took " + sw.Elapsed); 

      sw.Restart(); 
      QuicksortSequential(a, 0, a.Length-1); 
      Console.WriteLine("Sequential took " + sw.Elapsed); 

      sw.Restart(); 
      QuicksortParallelOptimised(b, 0, b.Length-1); 
      Console.WriteLine("Parallel took " + sw.Elapsed); 

      // Verify that our sort implementation is actually correct! 

      Trace.Assert(a.SequenceEqual(c)); 
      Trace.Assert(b.SequenceEqual(c)); 
     } 

     static void QuicksortSequential<T>(T[] arr, int left, int right) 
     where T : IComparable<T> 
     { 
      if (right > left) 
      { 
       int pivot = Partition(arr, left, right); 
       QuicksortSequential(arr, left, pivot - 1); 
       QuicksortSequential(arr, pivot + 1, right); 
      } 
     } 

     static void QuicksortParallelOptimised<T>(T[] arr, int left, int right) 
     where T : IComparable<T> 
     { 
      const int SEQUENTIAL_THRESHOLD = 2048; 
      if (right > left) 
      { 
       if (right - left < SEQUENTIAL_THRESHOLD) 
       { 
        QuicksortSequential(arr, left, right); 
       } 
       else 
       { 
        int pivot = Partition(arr, left, right); 
        Parallel.Invoke(
         () => QuicksortParallelOptimised(arr, left, pivot - 1), 
         () => QuicksortParallelOptimised(arr, pivot + 1, right)); 
       } 
      } 
     } 

     static int Partition<T>(T[] arr, int low, int high) where T : IComparable<T> 
     { 
      int pivotPos = (high + low)/2; 
      T pivot = arr[pivotPos]; 
      Swap(arr, low, pivotPos); 

      int left = low; 
      for (int i = low + 1; i <= high; i++) 
      { 
       if (arr[i].CompareTo(pivot) < 0) 
       { 
        left++; 
        Swap(arr, i, left); 
       } 
      } 

      Swap(arr, low, left); 
      return left; 
     } 

     static void Swap<T>(T[] arr, int i, int j) 
     { 
      T tmp = arr[i]; 
      arr[i] = arr[j]; 
      arr[j] = tmp; 
     } 
    } 
}