并行执行多对并发任务

Question

详情：

我有一个游戏，两个独立的 AI 互相对战。 每个 AI 都有自己的任务。 两个任务需要同时启动，需要带一些参数并返回一个值。 现在我想并行运行 100-200 个游戏（每两个任务）。

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

代码：

我目前的做法如下。

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

因为每个游戏的两个 AI 任务没有一起启动，所以我添加了一个 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播放的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;
        }

有没有更好的方法来做到这一点，保留所有东西但确保每个游戏中的两个 KI-Tasks 同时启动？

Answer 1

我的建议是更改Play 方法的签名，使其返回Task<Individual> 而不是Individual，并将对Thread.Sleep 的调用替换为await Task.Delay。这个小改动应该会对 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，以符合guidelines。

然后你应该刮掉Parallel.ForEach方法，因为it is not async-friendly，而是将每个individual投影到Task，将所有任务捆绑在一个数组中，并等待它们全部使用Task.WaitAll方法完成（或者如果你想去async-all-the-way，可以使用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 或 RAM 或 CPU RAM 带宽可能会饱和。您可以查看 here 以了解限制并发异步操作数量的方法。