c# - 在 .NET 中与 TPL 并发。 Parallel Generator 在一段时间后阻塞并卡住

Question

我正在 CodeGolf.stackexchange.com 上运行一个游戏，玩家可以提交机器人在游戏中相互竞争。

在这个阶段有 70 个机器人，并且有 (N*(N+1))/2 场比赛，锦标赛的运行速度变得非常慢，所以现在我希望将其并行化。游戏规则之一是机器人可以写入它自己的数据目录，所以我想确保我没有一个机器人实例同时玩 2 个游戏。

我写了一个IEnuemable<T>生成器返回有效的匹配(有效的是两个玩家当前都没有参与另一场比赛)但是我有某种并发/阻塞问题导致 Enumerable 无限循环。

在任意点，while(any matches)调用将继续循环，因为剩下的比赛只涉及 activePlayers 中的玩家。列表，所以它会一直循环下去。这将暗示正在进行中的事件匹配。

但是 IsCompleted事件将永远不会被再次调用，所以无论正在运行的比赛是否以某种方式被阻止，或者它们已经完成，但我在 _activePlayers.TryRemove() 中遇到了并发错误。代码。

public interface IMatchGenerator
{
    void CompleteMatch(Match match);
}

public class MatchGenerator : IMatchGenerator
{
    private static object SYNC_LOCK = new object();
    private ConcurrentDictionary<Player, Player> _activePlayers; //tracks players actively playing
    private ConcurrentQueue<Match> _allMatches;

    public MatchGenerator()
    {
        _activePlayers = new ConcurrentDictionary<Player, Player>();
    }

    public IEnumerable<Match> Generate(IList<Match> matches)
    {
        //take the list of matches passed in and stick them in a queue.
        _allMatches = new ConcurrentQueue<Match>(matches);

        //keep looping while there are matches to play
        while (_allMatches.Any())
        {
            Match nextMatch;

            lock (SYNC_LOCK)
            {
                 _allMatches.TryDequeue(out nextMatch); //Grab from front of queue

                if (!_activePlayers.ContainsKey(nextMatch.Player1) &&
                     !_activePlayers.ContainsKey(nextMatch.Player2))
                {
                    //If neither player is in the active player list, then this is a
                    //good match so add both players
                    _activePlayers.TryAdd(nextMatch.Player1, nextMatch.Player1);
                    _activePlayers.TryAdd(nextMatch.Player2, nextMatch.Player2);

                }
                else
                {
                    //Otherwise push this match back in to the start of the queue...
                    //FIFO should move on to next;
                    _allMatches.Enqueue(nextMatch);
                    nextMatch = null;
                }
            }
            if (nextMatch != null)
                yield return nextMatch;
        }
    }

    public void CompleteMatch(Match match)
    {
        //Matches in progress have the generator attached to them and will call 
        //home when they are complete to remove players from the active list
        Player junk1, junk2;
        lock (SYNC_LOCK)
        {
            _activePlayers.TryRemove(match.Player1, out junk1);
            _activePlayers.TryRemove(match.Player2, out junk2);
        }
        if (junk1 == null || junk2 == null)
        {
            Debug.WriteLine("Uhoh! a match came in for completion but on of the players who should have been in the active list didn't get removed");
        }
    }
}

以及使用它的代码。

var mg = new MatchGenerator();
//Code to generate IList<Match> or all player combinations and attach mg

Parallel.ForEach(mg.Generate(matches),
                    new ParallelOptions() {MaxDegreeOfParallelism = 8},
                    match =>
                    {
                        var localMatch = match;
                        try
                        {
                            PlayMatch(localMatch, gameLogDirectory, results);
                        }
                        finally
                        {
                            localMatch.IsCompleted();
                        }
                    });

从这里开始，它变得有点冗长，但并没有发生太多事情。 PlayMatch(...)调用方法调用 Play并且里面有一些 stringbuilder 代码。 Plays根据正在播放的机器人(即 ruby​​/python 等)调用几个外部进程...它还会流式写入每个播放器日志文件，但假设一次只有一个播放器机器人在运行，这里应该有任何冲突.

整个控制程序在 GitHub @ 上可用

https://github.com/eoincampbell/big-bang-game/blob/master/BigBang.Orchestrator/Program.cs

    public static Result Play(Player p1, Player p2, string gameLogDirectory)
    {
        var dir = Player.PlayerDirectory;

        var result = new Result() { P1 = p1, P2 = p2, P1Score = 0, P2Score = 0 };
        string player1ParamList = string.Empty, player2ParamList = string.Empty;
        List<long> p1Times = new List<long>(), p2Times = new List<long>();
        Stopwatch sw1 = new Stopwatch(), sw2 = new Stopwatch(), swGame = new Stopwatch();
        var sb = new StringBuilder();

        var proc = new Process
        {
            StartInfo =
            {
                UseShellExecute = false, RedirectStandardOutput = true, WorkingDirectory = dir
            }
        };

        swGame.Start();
        sb.AppendLine("+--------------------------------------------------------------------------------------------+");
        sb.AppendFormat("| Starting Game between {0} & {1} \n", p1.Name, p2.Name);
        sb.AppendLine("| ");
        for (var i = 0; i < 1; i++)
        {
            sw1.Reset();
            sw1.Start();
            var o1 = ProcessRunner.RunPlayerProcess(ref proc, player1ParamList, player2ParamList, p1, dir);
            sw1.Stop();
            p1Times.Add(sw1.ElapsedMilliseconds);

            //System.Threading.Thread.Sleep(1);

            sw2.Reset();
            sw2.Start();
            var o2 = ProcessRunner.RunPlayerProcess(ref proc, player2ParamList, player1ParamList, p2, dir);
            sw2.Stop();
            p2Times.Add(sw2.ElapsedMilliseconds);


            var whoWon = GetWinner(o1, o2, ref player1ParamList, ref player2ParamList);
            var whoWonMessage = "Draw Match";
            if (whoWon == "P1")
            {
                result.P1Score++;
                whoWonMessage = string.Format("{0} wins", p1.Name);
            }
            else if (whoWon == "P2")
            {
                result.P2Score++;
                whoWonMessage = string.Format("{0} wins", p2.Name);
            }

            sb.AppendFormat("| {0} plays {1} | {2} plays {3} | {4}\n", p1.Name, o1, p2.Name, o2, whoWonMessage);

        }
        swGame.Stop();
        sb.AppendLine("| ");
        sb.AppendFormat("| Game Time: {0}", swGame.Elapsed);

        result.WriteLine(sb.ToString());

        var resultMessage = string.Format("Result: {0} vs {1}: {2} - {3}",
                        result.P1,
                        result.P2,
                        result.P1Score,
                        result.P2Score);

        sb.AppendLine("| ");
        sb.AppendFormat("| {0}", resultMessage);



        using (var p1sw = new StreamWriter(Path.Combine(gameLogDirectory, p1.Name + ".log"), true))
        {
            p1sw.WriteLine(sb.ToString());
        }
        using (var p2sw = new StreamWriter(Path.Combine(gameLogDirectory, p2.Name + ".log"), true))
        {
            p2sw.WriteLine(sb.ToString());
        }

        result.P1AvgTimeMs = p1Times.Average();
        result.P2AvgTimeMs = p2Times.Average();

        return result;
    }

Answer 1

我相信你的问题是由使用 Parallel.ForEach() 引起的在 IEnumerable<T> 上可能需要无限长的时间才能产生下一个元素。这基本上与 using Parallel.ForEach() on BlockingCollection.GetConsumingEnumerable() 相同的问题:

the partitioning algorithm employed by default by both Parallel.ForEach and PLINQ use chunking in order to minimize synchronization costs: rather than taking the lock once per element, it'll take the lock, grab a group of elements (a chunk), and then release the lock.

但由于在您的情况下，在处理前一个元素之前不会生成序列中的下一个元素，这将导致无限期阻塞。

我认为这里正确的解决方案是使用 BlockingCollection<T>而不是 yield return : 替换 yield return nextMatch与 blockingCollection.Add(nextMatch) ，然后运行 ​​Generate()在单独的线程上使用 blockingCollection.GetConsumingPartitioner()来自上面提到的博客文章 Parallel.ForEach() .

我也不喜欢你的 Generate()在没有有效匹配的情况下浪费整个 CPU 内核，基本上什么都不做，但这是一个单独的问题。