Лучший способ сделать мой Parallel.ForEach Thread безопасным?

Question

Я хотел бы сделать следующий код потокобезопасным. К сожалению, я попытался заблокировать на разных уровнях этого кода без успеха. Единственный экземпляр, который, как мне кажется, может обеспечить безопасность потоков, заключается в том, чтобы поместить блокировку вокруг всего цикла, что фактически делает Parallel.ForEach не быстрее (возможно, даже медленнее), чем просто использовать foreach. Код относительно/почти безопасен без блокировки. Похоже, что он показывает небольшие изменения в суммировании ключей geneTokens.Value [-1] и gtCandidates.Value [-1] примерно один раз из каждых 20 или около того.

Я понимаю, что Словарь не является потокобезопасным. Тем не менее, я не могу изменить этот конкретный объект на ConcurrentDictionary без серьезного снижения производительности. Я предпочел бы запустить эту часть кода с регулярным foreach, чем изменить этот конкретный объект. Однако я использую ConcurrentDictionary для хранения отдельных объектов Dictionary. Я также попытался сделать это изменение, и это не решает проблему моей гонки.

Вот мои переменные уровня Класс:

//Holds all tokens derived from each sequence chunk 
public static ConcurrentBag<sequenceItem> tokenBag = 
    new ConcurrentBag<sequenceItem>(); 
public BlockingCollection<sequenceItem> sequenceTokens = new 
    BlockingCollection<sequenceItem>(tokenBag); 
public ConcurrentDictionary<string, int> categories = new 
    ConcurrentDictionary<string, int>(); 
public ConcurrentDictionary<int, Dictionary<int, int>> gtStartingFrequencies = new 
    ConcurrentDictionary<int, Dictionary<int, int>>(); 
public ConcurrentDictionary<string, Dictionary<int, int>> gtCandidates = new 
    ConcurrentDictionary<string, Dictionary<int, int>>(); 
public ConcurrentDictionary<string, Dictionary<int, int>> geneTokens = new 
    ConcurrentDictionary<string, Dictionary<int, int>>(); 

Вот Parallel.ForEach:

Parallel.ForEach(sequenceTokens.GetConsumingEnumerable(), seqToken => 
{ 
    lock (locker) 
    { 
    //Check to see if the Sequence Token is a Gene Token 
    Dictionary<int, int> geneTokenFreqs; 
    if (geneTokens.TryGetValue(seqToken.text, out geneTokenFreqs)) 
    { //The Sequence Token is a Gene Token 


     *****************Race Issue Seems To Occur Here**************************** 
     //Increment or create category frequencies for each category provided 
     int frequency; 
     foreach (int category in seqToken.categories) 
     { 
     if (geneTokenFreqs.TryGetValue(category, out frequency)) 
     { //increment the category frequency, if it already exists 
      frequency++; 
      geneTokenFreqs[category] = frequency; 
     } 
     else 
     { //Create the category frequency, if it does not exist 
      geneTokenFreqs.Add(category, 1); 
     } 
     } 

     //Update the frequencies total [-1] by the total # of categories incremented. 
     geneTokenFreqs[-1] += seqToken.categories.Length; 
     ****************************************************************************** 
    } 
    else 
    { //The Sequence Token is NOT yet a Gene Token 
     //Check to see if the Sequence Token is a Gene Token Candidate yet 
     Dictionary<int, int> candidateTokenFreqs; 
     if (gtCandidates.TryGetValue(seqToken.text, out candidateTokenFreqs)) 
     { 
     *****************Race Issue Seems To Occur Here**************************** 
     //Increment or create category frequencies for each category provided 
     int frequency; 
     foreach (int category in seqToken.categories) 
     { 
      if (candidateTokenFreqs.TryGetValue(category, out frequency)) 
      { //increment the category frequency, if it already exists 
      frequency++; 
      candidateTokenFreqs[category] = frequency; 
      } 
      else 
      { //Create the category frequency, if it does not exist 
      candidateTokenFreqs.Add(category, 1); 
      } 
     } 

     //Update the frequencies total [-1] by the total # of categories incremented. 
     candidateTokenFreqs[-1] += seqToken.categories.Length; 
     ***************************************************************************** 

     //Only update the candidate sequence count once per sequence 
     if (candidateTokenFreqs[-3] != seqToken.sequenceId) 
     { 
      candidateTokenFreqs[-3] = seqToken.sequenceId; 
      candidateTokenFreqs[-2]++; 

      //Promote the Token Candidate to a Gene Token, if it has been found >= 
      //the user defined candidateThreshold 
      if (candidateTokenFreqs[-2] >= candidateThreshold) 
      { 
      Dictionary<int, int> deletedCandidate; 
      gtCandidates.TryRemove(seqToken.text, out deletedCandidate); 
      geneTokens.TryAdd(seqToken.text, candidateTokenFreqs); 
      } 
     } 
     } 
     else 
     { 
     //create a new token candidate frequencies dictionary by making 
     //a copy of the default dictionary from 
     gtCandidates.TryAdd(seqToken.text, new 
      Dictionary<int, int>(gtStartingFrequencies[seqToken.sequenceId])); 
     } 
    } 
    } 
}); 

Answer 1

Очевидно одна раса данных исходит из того, что некоторые потоки будут добавления пункты здесь:

geneTokens.TryAdd(seqToken.text, candidateTokenFreqs); 

и другие будут читать здесь:

if (geneTokens.TryGetValue(seqToken.text, out geneTokenFreqs)) 

Answer 2

Как я использовал Параллельный словарь в моем проекте:

я м положить флаг в словаре и проверку из другого потока, что если флаг там или не. Если флаг присутствует я делаю свою задачу соответственно ..

Для этого, что им делать это, как:

1) Декларирование Параллельное словарь 2) Добавление флага с использованием метода TryAdd 3) пытается получить квартиру с помощью TryGet Methid.

1) Объявление

Dim cd As ConcurrentDictionary(Of Integer, [String]) = New ConcurrentDictionary(Of Integer, String)() 

2) Добавление

If cd.TryAdd(1, "uno") Then 
     Console.WriteLine("CD.TryAdd() succeeded when it should have failed") 
     numFailures += 1 
    End If 

3) Получение

If cd.TryGetValue(1, "uno") Then 
     Console.WriteLine("CD.TryAdd() succeeded when it should have failed") 
     numFailures += 1 
    End If