I have seen many examples which the results of a Producer/Consumer queue implemented using BlockingCollection<T> are independent, e.g. each result can be consumed independently in parallel. For example, here the produces calculate a square root and consumers grab each result to update the UI.
I wonder whether explicit locking can be avoided if the consumers can NOT independently process each result calculated by producers. I created a demo example of this where each result has to be processed taking into account previous processed results in order to either insert a row to the data grid or update an existing data grid; it seems to work ok on some small data.
My questions are:
Can explicit locking be avoided?
How to encapsulate consumer and producer within a single object e.g.
ProducerConsumerQueuewhich currently is really a producer queue.Currently I use a separate task to process each result especially given it ultimately has to post back to the UI thread; is this a waste?
Below is the code behind a simple WinForm which only has a DataGridView and a button.
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Forms;
namespace PCQueueDemo
{
public partial class Form1 : Form
{
// to consume results that are depedent
private ConcurrentDictionary<int, WorkItem> _dict = new ConcurrentDictionary<int, WorkItem>();
private readonly object _sync = new object();
private BindingSource _bindingSource = new BindingSource();
private TaskScheduler _ui;
private ProducerConsumerQueue _producers;
private BlockingCollection<WorkItem> _workQueue;
public Form1()
{
InitializeComponent();
_workQueue = new BlockingCollection<WorkItem>();
// create 3 producers which will produce work item to the work queue.
_producers = new ProducerConsumerQueue(3, _workQueue);
// capture the UI context in order to update data grid.
_ui = TaskScheduler.FromCurrentSynchronizationContext();
// bind the data grid
dataGridView.DataSource = _bindingSource;
dataGridView.AutoSize = true;
dataGridView.Dock = DockStyle.Fill;
}
// Run button click handler
private void button1_Click(object sender, EventArgs e)
{
_producers.Produce(); // start producing on each producer thread.
var ignore = Task.Run(() => updateGrid()); // consume the producer result to update the grid.
}
private void updateGrid()
{
foreach (var item in _workQueue.GetConsumingEnumerable())
{
WorkItem copy = item;
//? Is it a waste to use a thread for each result?
Task.Factory.StartNew(() =>
{
WorkItem localItem;
if (_dict.TryGetValue(item.Number, out localItem))
{
//? how to avoid locking if the update here takes longer than simply incrementing the counter?
lock (_sync) localItem.Counter++;
}
else if (_dict.TryAdd(copy.Number, copy))
{
//? Is this negligible since the lock should be very short?
lock(_sync) _bindingSource.Add(copy);
}
dataGridView.Refresh();
}, CancellationToken.None, TaskCreationOptions.None, _ui);
}
}
}
// a demo work item
public class WorkItem
{
public WorkItem(int number)
{
Number = number;
Counter = 1; // 1st appearance
}
public int Number { get; private set; }
public int Counter { get; set; }
}
//? This is just a producer queue; how to encapsulate the consumer together with producer while allowing it to update the UI?
public class ProducerConsumerQueue
{
private readonly int _producerCount;
private Task[] _producers;
private BlockingCollection<WorkItem> _workQueue;
public ProducerConsumerQueue(int producerCount, BlockingCollection<WorkItem> workQueue)
{
_workQueue = workQueue;
_producerCount = producerCount;
_producers = new Task[_producerCount];
}
public void Produce()
{
for (int i = 0; i < _producerCount; i++)
_producers[i] = Task.Run(() => runProducer());
Task.WhenAll(_producers).ContinueWith(_ => _workQueue.CompleteAdding());
}
private void runProducer()
{
// to demo just run for 5 seconds on each producer
var fiveSecondsLater = DateTime.Now.AddSeconds(15);
var random = new Random();
while (!_workQueue.IsAddingCompleted && DateTime.Now <= fiveSecondsLater)
{
_workQueue.Add(new WorkItem(random.Next(21)));
Thread.Sleep(100);
}
}
}
}
