Parallel.ForEach missing/ignoring/jumping over/not adding objects

user_9692251 0
Viewed 68

My _baseBlockContainer.GetBaseBlocks(); returns a ConcurrentQueue with 15317 objects. For further processing I want to sort them by Type. However, it always "misses" some objects.

It appears that my Parallel.ForEach is not thread-safe because the amount of objects within the ConcurrentQueue for a Type is sometimes less (off by 1 to 250 objects for a Type) than when sorted by the synchronous foreach; but I do not see where/why.

var baseBlocks = _baseBlockContainer.GetBaseBlocks();

var baseBlocksByTypeConcurrent = new ConcurrentDictionary<Type, ConcurrentQueue<BaseBlock>>();
// results of this always differ
Parallel.ForEach(baseBlocks, bb =>
{
  if (!baseBlocksByTypeConcurrent.ContainsKey(bb.GetType()))
  {
    baseBlocksByTypeConcurrent[bb.GetType()] = new ConcurrentQueue<BaseBlock>();
  }
  baseBlocksByTypeConcurrent[bb.GetType()].Enqueue(bb);
});

var baseBlocksByType = new ConcurrentDictionary<Type, ConcurrentQueue<BaseBlock>>();
// results of this are always the same
foreach (var bb in baseBlocks)
{
  if (!baseBlocksByType.ContainsKey(bb.GetType()))
  {
     baseBlocksByType[bb.GetType()] = new ConcurrentQueue<BaseBlock>();
  }
  baseBlocksByType[bb.GetType()].Enqueue(bb);
}
parallel.foreach parallel-processing concurrency c#
2 Answers

Replace this:

if (!baseBlocksByTypeConcurrent.ContainsKey(bb.GetType()))
{
    baseBlocksByTypeConcurrent[bb.GetType()] = new ConcurrentQueue<BaseBlock>();
}
baseBlocksByTypeConcurrent[bb.GetType()].Enqueue(bb);

with this:

baseBlocksByTypeConcurrent.TryAdd(bb.GetType(), new ConcurrentQueue<BaseBlock>());
baseBlocksByTypeConcurrent[bb.GetType()].Enqueue(bb);

The problem with your existing code is that if .ContainsKey evaluates to false in multiple threads at the same time for the same block type, then they will all set the value corresponding to the type to a new queue, erasing any existing queue for that type. That is to say: ContainsKey and the indexer are, by themselves, thread safe, but not when used separately in the way you are doing it.

TryAdd is thread safe and will only add that key once, rather than rewriting it as assigning to the indexer would.

user_31819330

Your code is suffering from what is known as race condition. Using concurrent collections alone does not prevent race condition to occur. You also have to use them correctly, by utilizing their special atomic APIs. In your case the appropriate API to use is the GetOrAdd method:

Adds a key/value pair to the ConcurrentDictionary<TKey,TValue> if the key does not already exist. Returns the new value, or the existing value if the key already exists.

Usage:

ParallelOptions options = new()
{
    MaxDegreeOfParallelism = Environment.ProcessorCount
};

Parallel.ForEach(baseBlocks, options, bb =>
{
    baseBlocksByTypeConcurrent
        .GetOrAdd(bb.GetType(), _ => new ConcurrentQueue<BaseBlock>())
        .Enqueue(bb);
});

As a side note, whenever you use the Parallel.ForEach method it is advisable¹ to specify explicitly the MaxDegreeOfParallelism. The default MaxDegreeOfParallelism is -1, which means unbounded parallelism, which in practice saturates the ThreadPool.

¹ It is advisable by me, not Microsoft. The general advice offered in the official documentation is that modifying the MaxDegreeOfParallelism setting is not needed. Microsoft's argument is that configuring by default this option to Environment.ProcessorCount might result in performance regression, and also that it opens the possibility of deadlock in case one iteration depends on another executing iteration. Apparently Microsoft does not consider the ThreadPool saturation to be a serious problem.

user_111785490
Related