并行执行多对并发任务

Question

详情：

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

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

代码：

我目前的做法如下。

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

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

        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 方法的签名，使其 returns 成为 Task<Individual> 而不是 Individual，并替换对 [=15] 的调用=] 与 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, and instead project each individual to a Task, bundle all tasks in an array, and wait them all to complete with the Task.WaitAll method (or with the await Task.WhenAll if you want to go 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 限制并发异步操作数量的方法。

并行执行多对并发任务

Execute multiple pairs of concurrent tasks in parallel

.net

c#

multithreading

task