c# - 并行长时间运行任务的时间优化

Question

简介

我正在使用一个复杂的外部库，我试图在其中对大量项目执行它的功能。该库没有公开良好的异步接口(interface)，所以我只能使用一些非常老式的代码。

我的目标是优化完成一批处理所需的时间，并在不包含实际的第 3 方库的情况下演示问题我已经创建了下面问题的近似值

问题

给定一个非异步操作，您可以提前知道操作的“大小”(即复杂性):

public interface IAction
{
    int Size { get; }
    void Execute();
}

鉴于此操作有 3 个变体:

public class LongAction : IAction
{
    public int Size => 10000;
    public void Execute()
    {
        Thread.Sleep(10000);
    }
}

public class MediumAction : IAction
{

    public int Size => 1000;
    public void Execute()
    {
        Thread.Sleep(1000);
    }
}

public class ShortAction : IAction
{
    public int Size => 100;
    public void Execute()
    {
        Thread.Sleep(100);
    }
}

您如何优化一长串这些操作，以便在以某种并行方式运行时，整个批处理尽快完成？

天真地，您可以将整个批处理都扔到一个 Parallel.ForEach 中，具有相当高的并行度，这当然有效 - 但必须有一种方法来优化它们，以便一些最大的首先开始。

为了进一步说明问题，如果我们举一个 super 简化的例子

• 1 个大小为 10 的任务
• 5 个大小为 2 的任务
• 10 个大小为 1 的任务

还有 2 个可用线程。我可以想出 2 种(很多)方法来安排这些任务(黑条是死时间 - 没有什么可安排的):

很明显，第一个比第二个完成得早。

最小完整且可验证的代码

完整的测试代码，如果有人喜欢 bash(尝试让它比我下面的天真实现更快):

class Program
{
    static void Main(string[] args)
    {
        MainAsync().GetAwaiter().GetResult();
        Console.ReadLine();
    }

    static async Task MainAsync()
    {
        var list = new List<IAction>();
        for (var i = 0; i < 200; i++) list.Add(new LongAction());
        for (var i = 0; i < 200; i++) list.Add(new MediumAction());
        for (var i = 0; i < 200; i++) list.Add(new ShortAction());


        var swSync = Stopwatch.StartNew();
        Parallel.ForEach(list, new ParallelOptions { MaxDegreeOfParallelism = 20 }, action =>
        {
            Console.WriteLine($"{DateTime.Now:HH:mm:ss}: Starting action {action.GetType().Name} on thread {Thread.CurrentThread.ManagedThreadId}");
            var sw = Stopwatch.StartNew();
            action.Execute();
            sw.Stop();
            Console.WriteLine($"{DateTime.Now:HH:mm:ss}: Finished action {action.GetType().Name} in {sw.ElapsedMilliseconds}ms on thread {Thread.CurrentThread.ManagedThreadId}");
        });
        swSync.Stop();
        Console.WriteLine($"Done in {swSync.ElapsedMilliseconds}ms");
    }
}


public interface IAction
{
    int Size { get; }
    void Execute();
}

public class LongAction : IAction
{
    public int Size => 10000;
    public void Execute()
    {
        Thread.Sleep(10000);
    }
}

public class MediumAction : IAction
{

    public int Size => 1000;
    public void Execute()
    {
        Thread.Sleep(1000);
    }
}

public class ShortAction : IAction
{
    public int Size => 100;
    public void Execute()
    {
        Thread.Sleep(100);
    }
}

Answer 1

一个相对快速和肮脏的解决方案是使用 a load-balancing partitioner在按大小递减排序的 Action 列表之上

var sorted = list.OrderByDescending(a => a.Size).ToArray();
var partitioner=Partitioner.Create(sorted, loadBalance:true);

Parallel.ForEach(partitioner, options, action =>...);

与其他答案一样，仅使用这两行，性能就提高了约 30%。

PLINQ 对数据进行分区，并使用单独的任务一次处理整个分区。当输入大小已知时，就像 IList 派生数组和列表的情况一样，输入被分成大小相等的 block 并提供给每个工作任务。

当大小未知时，如迭代器方法、LINQ 查询等，PLINQ 使用 block 分区。一次检索一大块数据并将其提供给工作任务。

我忘记的另一个选项是在 top chunck 分区上的负载平衡。这将使用小块的 block 分区应用于数组和 IList 派生的输入。负载均衡Partitioner.Create重载返回 OrderablePartitioner 实例，因此 IAction 项的顺序得以保留

同样可以用 IEnumerable<T> 来实现通过指定 EnumerablePartitionerOptions.NoBuffering 来源选项:

var sorted = list.OrderByDescending(a => a.Size);
var partitioner=Partitioner.Create(sorted,EnumerablePartitionerOptions.NoBuffering);

这将创建一个使用 block 编码的 OrderablePartitioner