次のコードをスレッドセーフにしたいと思います。残念ながら、このコード内でさまざまなレベルでロックを試みましたが、成功しませんでした。スレッドセーフを達成できると思われる唯一の例は、ループ全体にロックを配置することです。これにより、Parallel.ForEach は foreach を使用するよりも速くなりません (おそらくさらに遅くなります)。コードはロックなしで比較的/ほぼ安全です。約 20 回の実行ごとに、geneTokens.Value[-1] キーと gtCandidates.Value[-1] キーの合計にわずかな変動が見られるだけです。
Dictionary はスレッドセーフではないことを認識しています。ただし、この特定のオブジェクトを ConcurrentDictionary に変更すると、ダウンストリームのパフォーマンスが大幅に低下します。特定のオブジェクトを変更するよりも、コードのこの部分を通常の foreach で実行したいと思います。ただし、個々の Dictionary オブジェクトを保持するために ConcurrentDictionary を使用しています。私もこの変更を試みましたが、人種の問題は解決しません。
これが私のクラスレベルの変数です:
//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]));
}
}
}
});