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

Question

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);
}

No. When I call _baseBlockContainer.GetBaseBlocks(); after this method or somewhere else in the class the amount of objects within the Queue didn't change (obviously, since I didn't Dequeue() any anywhere in my code). I see now that my wording was pretty bad. The Parallel.ForEach is just missing / ignoring / not adding some objects to the Dictionary — MichaelS
– MichaelS, Commented May 31, 2022 at 6:36

ProgrammingLlama · Accepted Answer · 2022-05-31 06:38:44Z

4

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.

answered May 31, 2022 at 6:38

ProgrammingLlama

39.3k7 gold badges78 silver badges104 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

MichaelS Over a year ago

Thank you very much! Your solution did not only fix the problem but also reduces nesting! :-)

Theodor Zoulias · Accepted Answer · 2022-07-28 22:56:08Z

1

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.}

edited Jul 28, 2022 at 22:56

answered May 31, 2022 at 10:22

Theodor Zoulias

46.1k8 gold badges112 silver badges155 bronze badges

2 Comments

MichaelS Over a year ago

Thank you for your answer and your suggestions. One follow-up question: The program will run on a server for a few seconds and should be as fast as possible. Is ThreadPool saturation even a consideration for this task, since no other program will need resources during that time (or in other words I am fine if other stuff runs slower during the processing)

Theodor Zoulias Over a year ago

@MichaelS my opinion is that saturating the ThreadPool is wrong in principle. In the case that you describe, the impact of saturating the ThreadPool is likely to be non existent, but why do it? It will only serve as a bad habit for the next time that you'll need to use the Parallel.ForEach, in which case the impact might be significant.

Collectives™ on Stack Overflow

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

2 Answers 2

1 Comment

2 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

1 Comment

2 Comments

Your Answer

Sign up or log in

Post as a guest

Related