并行执行多对并发任务

Question

详细信息：

我有一场游戏，两个独立的认可机构互相比赛。每个人工智能都有自己的任务。这两个任务都需要同时开始，需要接受一些参数并返回一个值。现在我想并行运行100-200个游戏(每两个任务)。

我现在遇到的问题是这两个任务不是一起启动的。只要有一些免费的资源，它们都是完全随机启动的。

代码：

我目前的做法如下所示。

• 我有一个输入对象列表，其中包括一些参数。
• 使用Parallel.ForEach，我为每个输入创建一个游戏，为游戏创建两个AI。
• 无论哪个AI完成游戏，首先用CancellationToken停止另一个AI，玩相同的游戏。
• 所有返回的值都保存在ConcurrentBag中。

因为每个游戏的两个人工智能任务不是一起开始的，所以我添加了一个AutoResetEvent。我希望我可以等待一个任务，直到第二个任务开始，但是AutoResetEvent.WaitOne却阻塞了所有的资源。因此，使用AutoResetEvent的结果是，第一个AI-任务正在启动并等待第二个任务的启动，但是由于它们不再释放线程，所以它们将永远等待。

        private ConcurrentBag<Individual> TrainKis(List<Individual> population) {
            ConcurrentBag<Individual> resultCollection = new ConcurrentBag<Individual>();
            ConcurrentBag<Individual> referenceCollection = new ConcurrentBag<Individual>();

            Parallel.ForEach(population, individual =>
            {
                GameManager gm = new GameManager();

                CancellationTokenSource c = new CancellationTokenSource();
                CancellationToken token = c.Token;
                AutoResetEvent waitHandle = new AutoResetEvent(false);

                KI_base eaKI = new KI_Stupid(gm, individual.number, "KI-" + individual.number, Color.FromArgb(255, 255, 255));
                KI_base referenceKI = new KI_Stupid(gm, 999, "REF-" + individual.number, Color.FromArgb(0, 0, 0));
                Individual referenceIndividual = CreateIndividual(individual.number, 400, 2000);

                var t1 = referenceKI.Start(token, waitHandle, referenceIndividual).ContinueWith(taskInfo => {
                    c.Cancel();
                    return taskInfo.Result;
                }).Result;
                var t2 = eaKI.Start(token, waitHandle, individual).ContinueWith(taskInfo => { 
                    c.Cancel(); 
                    return taskInfo.Result; 
                }).Result;

                referenceCollection.Add(t1);
                resultCollection.Add(t2);
            });

            return resultCollection;
        }

这是人工智能的开始方法，在这里我等待第二个AI发挥：

            public Task<Individual> Start(CancellationToken _ct, AutoResetEvent _are, Individual _i) {
                i = _i;
                gm.game.kis.Add(this);
                if (gm.game.kis.Count > 1) {
                    _are.Set();
                    return Task.Run(() => Play(_ct));
                }
                else {
                    _are.WaitOne();
                    return Task.Run(() => Play(_ct));
                }
            }

和简化的游戏方法

public override Individual Play(CancellationToken ct) {
            Console.WriteLine($"{player.username} started.");
            while (Constants.TOWN_NUMBER*0.8 > player.towns.Count || player.towns.Count == 0) {
                try {
                    Thread.Sleep((int)(Constants.TOWN_GROTH_SECONDS * 1000 + 10));
                }
                catch (Exception _ex) {
                    Console.WriteLine($"{player.username} error: {_ex}");
                }
                
                //here are the actions of the AI (I removed them for better overview)

                if (ct.IsCancellationRequested) {
                    return i;
                }
            }
            if (Constants.TOWN_NUMBER * 0.8 <= player.towns.Count) {
                winner = true;
                return i;
            }
            return i;
        }

有没有一个更好的方法来做到这一点，保持所有的东西，但确保两个基-任务在每个游戏是在同一时间开始？

Answer 1

我的建议是更改Play方法的签名，以便它返回一个Task<Individual>而不是Individual，并用await Task.Delay替换对Thread.Sleep的调用。这个小小的改变应该对AI玩家的响应有显著的积极影响，因为没有线程会被他们阻塞，并且ThreadPool线程的小池将被优化地利用。

public override async Task<Individual> Play(CancellationToken ct)
{
    Console.WriteLine($"{player.username} started.");
    while (Constants.TOWN_NUMBER * 0.8 > player.towns.Count || player.towns.Count == 0)
    {
        //...
        await Task.Delay((int)(Constants.TOWN_GROTH_SECONDS * 1000 + 10));
        //...
    }
}

您还可以考虑将方法的名称从Play更改为PlayAsync，以符合指导方针。

然后，您应该刮掉Parallel.ForEach方法，因为它不是异步友好的。，而是将每个individual投影到一个Task中，将所有任务捆绑在一个数组中，然后等待它们全部使用Task.WaitAll方法完成(如果您想使用全异步，则使用await Task.WhenAll )。

Task[] tasks = population.Select(async individual =>
{
    GameManager gm = new GameManager();
    CancellationTokenSource c = new CancellationTokenSource();

    //...

    var task1 = referenceKI.Start(token, waitHandle, referenceIndividual);
    var task2 = eaKI.Start(token, waitHandle, individual);

    await Task.WhenAny(task1, task2);
    c.Cancel();
    await Task.WhenAll(task1, task2);
    var result1 = await task1;
    var result2 = await task2;
    referenceCollection.Add(result1);
    resultCollection.Add(result2);
}).ToArray();

Task.WaitAll(tasks);

这样，您将有最大的并发性，这可能不是理想的，因为您的计算机的CPU或<=>或CPU内存带宽可能会饱和。您可以查看这里，寻找限制并发异步操作数量的方法。