并行长期运行任务的时间优化

Question

介绍

我正在使用一个复杂的外部库,我试图在一大堆项目上执行它的功能.该库没有公开一个好的异步接口,所以我坚持使用一些非常老式的代码.

我的目标是优化完成一批处理所需的时间,并演示问题而不必包含我在下面创建的实际第三方库的近似问题

问题

给定非异步操作,您可以提前知道操作的"大小"(即复杂性):

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,并且具有相当高的并行性并且肯定有效 - 但是必须有一种方法来优化它们,所以一些最大的首先开始.

为了进一步说明问题,如果我们采取一个超简化的例子

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

还有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

一个相对快速但肮脏的解决方案是在按大小递减排序的操作列表之上使用负载平衡分区器

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

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

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

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

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

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

IEnumerable<T>通过指定选项可以使用源实现相同的效果EnumerablePartitionerOptions.NoBuffering：

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

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