Starting a job in a thread reads from queue from other thread

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, Feb 06 '15 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, Feb 06 '15 at 12:38
It's been answered a couple of times already, it's just a different context. I'll post a duplicate. — Dirk, Feb 06 '15 at 12:39

score 0 · Answer 1 · answered Feb 06 '15 at 12:58

0

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().

answered Feb 06 '15 at 12:58

SverreN

177
1
9

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. – Halfgaar Feb 06 '15 at 13:31

Starting a job in a thread reads from queue from other thread

1 Answers1