c# - 并行执行多对并发任务

Question

细节:
我有一个游戏，有两个独立的 AI 互相玩。
每个人工智能都有自己的任务。
两个任务需要同时启动，需要带一些参数并返回一个值。
现在我想并行运行 100-200 个游戏(每两个任务)。
问题我现在的情况是这两个任务没有一起开始。只要有一些免费资源，它们就完全随机启动。
代码:
我目前的方法如下。

我有一个包含一些参数的输入对象列表。

使用 Parallel.ForEach，我为每个输入对象创建了一个游戏和两个游戏 AI。

无论哪个 AI 先完成游戏，都会使用 CancellationToken 停止另一个玩相同游戏的 AI。

所有返回值都保存在 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 任务同时启动？

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方法(或者使用 await Task.WhenAll 如果你想去 async-all-the-way )。

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用于限制并发异步操作数量的方法。