使用任务进行重复操作

Question

我有 50 个机器学习代理。每一帧，他们都会得到一些输入并计算神经网络。因为每个代理都是独立的，我想让每个代理计算网络作为一个单独的任务。

如果我要为每个代理、每个帧创建一个任务，它会使我的程序变慢。我尝试将我的代理分为 2 个任务（25 和 25），但这仍然是开销。

我看到它的方式是在开始时为 n 组代理创建 n 个线程，并以某种方式在每一帧查询这些线程。一个线程会为一组代理计算网络，然后等待下一个查询。

我已阅读有关此主题的一些文章，但我发现我无法重用任务。那么，有什么解决方法可行呢？

基本上，我对 50 个代理有一个重复的操作，每帧运行大约一分钟，如果不将它们并行化将是一种浪费。

我还是多线程和任务的新手，所以我依赖你的帮助。

---

旁注：我在 Unity 中使用遗传算法。

---

这是我尝试将代理分成 n 个组并在 n 个任务中计算它们的网络的代码。

public async Task EvaluateAsync(int groupSize = 10)
{
    var groups = genomes.Select((g, i) => new { Value = g, Index = i })
                        .GroupBy(x => x.Index / groupSize)
                        .Select(x => x.Select(v => v.Value));

    var tasks = groups.Select(g =>
    {
        return Task.Run(() =>
        {
            foreach (var element in g)
                element.Fitness += ComputeFitness(element as NeuralGenome);
        });
    }).ToArray();

    for (var i = 0; i < tasks.Length; i++)
        await tasks[i];
}

在我调用的Update() 函数中：

EvaluateAsync(25).Wait();

网络很大的时候会快一点，但是只有10个神经元的时候会慢很多。

使组更小，只有在网络非常庞大的情况下才会产生更好的性能。

这里我为每个代理创建一个任务：

public async Task EvaluateAsyncEach()
{
    var tasks = genomes.Select(x => Task.Run(() => x.Fitness += ComputeFitness(x as NeuralGenome)))
                       .ToArray();
    foreach (var task in tasks)
        await task;
}

以下测量是针对 10 帧进行的。意思是，t/10 将是完成一项任务的时间。

正常运行时间：

00:00:00.3791190
00:00:00.3758430
00:00:00.3697020
00:00:00.3743900
00:00:00.3764850

每帧每个代理一个任务：

00:00:01.1288240
00:00:01.0761770
00:00:00.9311210
00:00:01.0122570
00:00:00.8938200

25 人一组：

00:00:00.5401100
00:00:00.5629660
00:00:00.5640470
00:00:00.5932220
00:00:00.6053940
00:00:00.5828170

Answer 1

您应该为此使用 Microsoft 的响应式框架。它非常适合这种处理。

代码如下：

var query =
    from genome in genomes.ToObservable()
    from fitness in Observable.Start(() => ComputeFitness(genome as NeuralGenome))
    select new { genome, fitness };

IDisposable subscription =
    query.Subscribe(x => x.genome.Fitness += x.fitness);

它在后台进行所有自己的线程/任务管理。它还会在计算结果后尽快生成结果。

如果你希望能够await得到结果你可以这样做：

var query =
    from genome in genomes.ToObservable()
    from fitness in Observable.Start(() => ComputeFitness(genome as NeuralGenome))
    select new { genome, fitness };

var results = await query.ToArray();

foreach (var x in results)
{
    x.genome.Fitness += x.fitness;
}

只需 NuGet "System.Reactive" 并将 using System.Reactive.Linq; 添加到您的查询中。

---

根据您评论中的代码，我认为您应该看看这个：

private async Task ComputingNetworksAsync()
{
    var query =
        from a in agents.ToObservable()
        let i = a.GenerateNetworkInputs()
        from n in Observable.Start(() => a.ComputeNetwork(i))
        select n;

    await query.ToArray();
}

这直接等同于您的代码（.ToArray() 除外）。

但是，您可以更进一步，这样做：

private async Task ComputingNetworksAsync()
{
    var query =
        from a in agents.ToObservable()
        from i in Observable.Start(() => a.GenerateNetworkInputs())
        from n in Observable.Start(() => a.ComputeNetwork(i))
        select n;

    await query.ToArray();
}

Answer 2

这是一篇好文章。

http://fintechexplained.blogspot.com/2018/05/top-ten-tips-for-implementing-multi.html?m=1

您的解决方案是 PLINQ。避免创建新任务