I make first steps in parallel programming. I wrote a simplest code, but it result confused. This code takes top 10 recent items with producer-consumer pattern. Only 1 consumer and 1 producer. With two consumers in works better, but incorrect too.
public static void ProducerConsumer(string path)
{
var capacity = 50000;
var collection = new BlockingCollection<string>(capacity);
var dict = new Dictionary<string, int>();
var tasks = new List<Task<Dictionary<string, int>>>();
var producer = Task.Factory.StartNew(() =>
{
Parallel.ForEach(File.ReadLines(path), (line) =>
{
collection.Add(line);
});
collection.CompleteAdding();
});
for (int i = 0; i < 1; i++)
{
var consumer = Task.Factory.StartNew<Dictionary<string, int>>(() =>
{
var localDict = new Dictionary<string, int>();
while (!collection.IsCompleted)
{
string line;
if (collection.TryTake(out line))
{
Map(line, localDict);
}
}
return localDict;
});
tasks.Add(consumer);
}
int count = 0;
while (tasks.Count > 0)
{
var id = Task.WaitAny(tasks.ToArray());
var res = tasks[id].Result;
count += res.Sum(k => k.Value);
Aggregate(res, dict);
tasks.RemoveAt(id);
}
Console.WriteLine($"Values sum : {count}");
ShowResult(dict);
ShowTotal(dict, "End");
}
public static void Map(string line, Dictionary<string, int> dict)
{
var parts = line.Split(new string[] { ";" }, StringSplitOptions.RemoveEmptyEntries);
var streetName = parts[3];
if (dict.Keys.Contains(streetName))
{
dict[streetName]++;
}
else
{
dict.Add(streetName, 1);
}
}
public static void ShowResult(Dictionary<string, int> dict)
{
var res = dict.OrderByDescending(r => r.Value).Take(10).ToList();
foreach (var key in res)
{
Console.WriteLine($"{key.Key} - {key.Value}");
}
}
public static void Aggregate(Dictionary<string, int> part, Dictionary<string, int> main)
{
foreach (var key in part.Keys)
{
if (main.Keys.Contains(key))
{
main[key] += part[key];
}
else
{
main.Add(key, 1);
}
}
}
public static void ShowTotal(Dictionary<string, int> dict, string mark = null)
{
Console.WriteLine($"{mark ?? ""} Keys: {dict.Keys.Count} - Values:{dict.Sum(s => s.Value)}");
}
While debugging it shows correct steps, but result shows only one hit per item and incorrect total.
If I understand your algorithm, it should be:
main.Add(key, part[key])