Starting a job in a thread reads from queue from other thread [duplicate]

Question

I have a problem that I can't figure out. It might not be easy to explain.

I have a singleton class with this private constructor:

private BarcodeMonitor() { processors[Machines.H1] = new BarcodeProcessor { Queue = new BlockingQueue<BarcodeData>("H1") }; processors[Machines.H2] = new BarcodeProcessor { Queue = new BlockingQueue<BarcodeData>("H2") }; processors[Machines.M] = new BarcodeProcessor { Queue = new BlockingQueue<BarcodeData>("M") }; processors[Machines.HP] = new BarcodeProcessor { Queue = new BlockingQueue<BarcodeData>("HP") }; foreach (KeyValuePair<Machines, BarcodeProcessor> pair in processors) { Thread t = new Thread(t1 => pair.Value.StartProccesingQueue()); t.Name = pair.Key.ToString() + "Processor"; t.Start(); threads.Add(t); } }

A new and unique BlockingQueue is given to the BarcodeProcessor and takes a name.

The BarcodeMonitor has this method:

public BlockingQueue<BarcodeData> GetQueue(Machines machine) { var processor = processors[machine]; return processor.Queue; }

so that incoming barcodes are put in the queue of the right machine. This works fine.

The barcodes are dequeued in StartProccesingQueue() (of which 4 instances are running). In the dequeue method, I have:

System.Console.WriteLine(string.Format("Thread {0} is taking from queue {1}", Thread.CurrentThread.Name, name));

Dequeue() uses Monitor.Wait(_internalQueue) when the queue is empty. The Enqueue() uses Monitor.PulseAll(_internalQueue) to continue the waiting dequeue.

What is happening, is that the StartProccesingQueue() method takes from other queues, even though it just access that Queue property that is assigned with a new BlockingQueue. In fact, thus far, I've only seen items from queue "M" being taken, and only the H1 and H2 threads are doing it.

I really don't get why this is happening.

You are accessing the foreach-iterator variable in a closure. That's dangerous and the behaviour is compiler-dependent. It might be the reason. Simply do a `var value = pair.Value;' and use that in the lambda and see if that helps. — Dirk
– Dirk, Commented Feb 6, 2015 at 12:27
@Dirk Wow, that did it. All the time it took me to find that out... Post as an answer for the acceptance :) — Halfgaar
– Halfgaar, Commented Feb 6, 2015 at 12:38
It's been answered a couple of times already, it's just a different context. I'll post a duplicate. — Dirk
– Dirk, Commented Feb 6, 2015 at 12:39

SverreN · Accepted Answer · 2015-02-06 12:58:21Z

Well, I'm not really sure about this but I think when you do the pulseAll(_internalQueue) all waiting threads are noticed that the _internalQueue is now ready to be processed, see MSDN PulseAll(). Maybe try just pulse(_internalQueue) this should notify only the Threads waiting for this _internalQueue, see MSDN pulse().

I indeed was confused why the original author of the BlockingQueue used PulseAll(), but in the end, the answer had to do with accessing the foreach variable in a closure. See the first comment to my question.

Collectives™ on Stack Overflow

Starting a job in a thread reads from queue from other thread [duplicate]

1 Answer 1

1 Comment

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Linked

Related