如何在跨度<T>上并行化合并排序

Answer 1

您可能应该避免使用 stackalloc，因为如果您的输入太大，这将失败。特别是因为你最终将它转换为常规数组，所以你应该在堆上分配内存开始。或者可能，使用可以重用的内存块池。如果您的输入足够小以适合堆栈，您可能不会从任何并行化中受益。

如果您考虑一下，使用堆栈内存进行并行操作几乎没有意义，因为工作线程需要引用属于另一个线程的堆栈内存，而且我认为没有任何方法可以使其既快速又安全.

一旦你有了实际的堆内存，你的问题就会消失。您的MergeLadder 可以将常规数组作为输入参数，并且您可以将数组转换为Merge 方法的跨度。这应该允许您的主循环转换为Parallel.For。

Answer 2

这是asynchronous version based on the advice of @JonasH.

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
                    
public class Program
{
    public static void Main()
    {
        Console.WriteLine(Sort.Merge("bonakidbonakidbonakid"));
    }
}

public static class Sort
{
    public async static ValueTask<string> Merge(
            string input,
            IComparer<char> comparer = null)
    {
        comparer ??= Comparer<char>.Default;
        
        // buffer to contain the original data.
        Memory<char> source = new char[input.Length];
        input.AsMemory().CopyTo(source);
        // other buffer, that starts as the output.
        Memory<char> buffer = new char[source.Length];
        
        // copy the buffer out as the result;
        return (await MergeLadder(
            source,
            buffer,
            comparer)).ToString();
    }
    
    public async static ValueTask<T[]> Merge<T>(
            T[] input,
            IComparer<T> comparer = null)
    {
        comparer ??= Comparer<T>.Default;
        
        // buffer to contain the orginal data.
        Memory<T> source = new T[input.Length];
        // intentionally copying input so it is not mutated.
        input.AsMemory().CopyTo(source);
        // other buffer, that starts as the output.
        Memory<T> buffer = new T[source.Length];
        
        // copy the buffer out as the result.
        return (await MergeLadder(
            source,
            buffer,
            comparer)).ToArray();
    }
    
    /// <remarks>
    /// Bottom up merge with alternating buffers.
    /// </remarks>
    private async static ValueTask<ReadOnlyMemory<T>> MergeLadder<T>(
        Memory<T> input,
        Memory<T> output,
        IComparer<T> comparer,
        int width = 2)
    {
        var half = width / 2;
        
        await Task.WhenAll(GetMergeTasks(
            input,
            output,
            comparer,
            width,
            half));
        
        // Did the last stride cover the whole input?
        if (width >= input.Length)
        {
            // Yes, everything is sorted
            return output;
        }
        else
        {
            // No, walk the array again with double the width.
            // Switch the buffers so we don't walk over the results.
            return await MergeLadder(output, input, comparer, width * 2);
        }
    }
    
    private static IEnumerable<Task> GetMergeTasks<T>(
        Memory<T> input,
        Memory<T> output,
        IComparer<T> comparer,
        int width,
        int half)
    {
        // walk the input sequence in mergeable strides
        // combine the left and right halves into the output
        for (int i = 0; i < input.Length; i += width)
        {
            Memory<T> left;
            Memory<T> right;
            Memory<T> merge;
            
            var remaining = input.Length - i;
            if (remaining <= half)
            {
                // not enough left for a right.
                left = input.Slice(i, remaining);
                right = Memory<T>.Empty;
                merge = output.Slice(i, remaining);
            }
            else if (remaining < width)
            {
                // not enough for a whole right.
                left = input.Slice(i, half);
                right = input.Slice(i + half, remaining - half);
                merge = output.Slice(i, remaining);
            }
            else
            {
                // the full stride.
                left = input.Slice(i, half);
                right = input.Slice(i + half, half);
                merge = output.Slice(i, width);
            }
            
            // Now merge the left and right for this stride.
            yield return Task.Run(() => Merge(left, right, merge, comparer));
        }
    }
            
    private static void Merge<T>(
        ReadOnlyMemory<T> leftMemory,
        ReadOnlyMemory<T> rightMemory,
        Memory<T> mergeMemory,
        IComparer<T> comparer)
    {
        var left = leftMemory.Span;
        var right = rightMemory.Span;
        var merge = mergeMemory.Span;
        
        //While either span has an element
        for(int m = 0, l = 0, r= 0; l < left.Length || r < right.Length; m++)
        {
            if (l < left.Length && r < right.Length)
            {
                //both sides have elements
                if (comparer.Compare(left[l], right[r]) <= 0)
                {
                    // left is less than right
                    merge[m] = left[l];
                    l++;
                }
                else
                {
                    // right is less than left
                    merge[m] = right[r];
                    r++;
                }
            }
            else if (l < left.Length)
            {
                // only left has some left
                merge[m] = left[l];
                l++;
            }
            else
            {
                // only right has some left
                merge[m] = right[r];
                r++;
            }
        }
    }
}

现在我想知道，并行版本需要多大的输入才能获胜？