Using BlockingCollection to replace queue.

Question

0.00/5 (No votes)

See more:

Originally I used queue to solve my problem. The application is a concurrent calls. Each time it only can take 4 calls.(MaximumChannels)

C#

public static int m_ActiveCallCount = 0;
public static int MaximumChannels = 4;
public static object m_SyncVar = new object();
public static Queue<AppointmentReminder> m_Queue = new Queue<AppointmentReminder>();

public static void IncrementCallCounters()
{
    Interlocked.Increment(ref m_ActiveCallCount);
}

After adding tons of items to the queue, I had the code to run.

C#

while (true)
                {
                    AppointmentReminder reminder = null;
                    if (m_ActiveCallCount < MaximumChannels)
                    {
                        lock (m_SyncVar)
                        {
                            // Make sure we have values to read
                            if (m_Queue.Count > 0)
                            {
                                reminder = m_Queue.Dequeue();
                            }
                        }
                        if (reminder != null)
                        {
                            Dial dial = new Dial(tServer);
                            Thread oThread = new Thread(() => dial.RunScript(reminder));
                            IncrementCallCounters();
                            oThread.Start();
                        }
                    }

Now I want to use BlockingCollection http://msdn.microsoft.com/en-us/library/dd267312(v=vs.110).aspx[^] to replace queue. The problem I have is that I am not sure how to apply the boundary number herein referring to "MaximumChannels" with BlockingCollection class.

I need the code hint.
Thanks.

Posted 24-Sep-14 3:47am

zhshqzyc

Updated 24-Sep-14 4:46am

v3

Add a Solution

Comments

Sergey Alexandrovich Kryukov 24-Sep-14 11:03am

Not clear what do you mean by "replacing a queue". And what's the idea of that "replacement". Why?

I do understand that the problem of over-populating of the queue is the problem. This is a general problem of the producer-consumer pattern. It is addressed in "bounding" if the collection described in its documentation.

—SA

[no name] 24-Sep-14 11:54am

Replacement means: I want to use
public static BlockingCollection m_Queue = new BlockingCollection(); to replace queue class. Because it is thread-safe.

The question is if I apply the boundary to the collection.
public static BlockingCollection m_Queue = new BlockingCollection(MaximumChannels);

It only processes MaximumChannels items.

I heard about ActionBlock Class etc.

Sergey Alexandrovich Kryukov 24-Sep-14 11:59am

Right. This is what I mentioned in my answer. Before we can continue this discussion, please read it and then reply. Why bounding (by maximum number) is not good enough for you?
—SA

[no name] 24-Sep-14 13:52pm

Because right now I just test the project, I haven't used the while loop in producer method. If I add an upper boundary, the producer only has 4 items. If I don't use the boundary, I have to retrieve 4 items in the consumer method each time. The difficult thing to me is I how to apply the constraint in the consumer. I don't know how to do it.

Do I need still use IncrementCallCounters(); and so on?
Or use something like BufferBlock such as http://blog.stephencleary.com/2012/11/async-producerconsumer-queue-using.html

Sergey Alexandrovich Kryukov 24-Sep-14 15:19pm

I don't really understand why would you use the while loop in producer method. It is not assumed in the pattern. Generally, the pattern assumes that you "infinitely" adds items as they are created, typically, until the end of the lifetime of the application. And the consumer works in an "infinite" loop retrieving items as they become available in the queue. Hence the blocking character of this code: the consumer is put to a wait state (spending zero CPU time) if it requested for another item when it collection is empty. If you are doing something else, you may need to review the design.

The issue is the behavior of the producer when the collection is full. Usually, it needs to be put to wait state, too. What would you need? This is not clear.

Why 4 items? What's wrong with 4 items? If 4 always come together, it would also mean a different pattern. Doesn't it mean that you need to have only one item of different type, composing the 4? I don't know. I would really need to know your whole problem.

—SA

[no name] 24-Sep-14 15:50pm

@SA, Sorry for the unclear question. The reason is I thought there was a throttle issue in the application. The application is a phone call process. Yes, say there are 100 calls in the queue, however there are only 4 ports available. We only can take up to 4 calls at the same time. Until one call finishes, the released port then can take one call remained in the queue. Originally I thought to use semaphore but I failed, then I used queue class and succeeded as the very first code shown. But I had to use
m_ActiveCallCount < MaximumChannels
to make sure we had a port available.

I thought it was stupid so I wanted to switch queue class to BlockingCollection framework to avoid it.

Actually I already replaced queue class with BlockingCollection and it works out now. It seems no question at all. But the real question is that I still need to put the code snippet
m_ActiveCallCount < MaximumChannels
into my code.

I think that if we use BlockingCollection will eliminate the if condition as I used to do. I am looking the skill with the BlockingCollection plus this port boundary. So basically it is a question about "Create a Task Scheduler That Limits Concurrency". MSDN http://msdn.microsoft.com/en-us/library/ee789351%28v=vs.110%29.aspx has an example. I think that it is too complicated. For my particular case, I just want to modify my original "Queue" code to "BlockingCollection". I don't want to make too many changes.

Hopefully I clarified the question.

Sergey Alexandrovich Kryukov 24-Sep-14 16:18pm

First of all, number and/or even the sets of the ports should be a parameter of the problem. So far, I need just one detail: how the ports are identified and how can you distribute ports.

I'll explain: say, you have the set of ports 0... 3. (They may not be a range, could be arbitrary set some identities, addresses, anything.) Say, you got 2 elements in the queue, with ports 0 and 1. It means that not only you can add only two elements, but next element can use either port 2 or port 3, not port 0 or port 1. Did you get my question?

And, before you even answer, I can suggest some shot-off-the-heaps idea, a known one. You should have two queues, a queue of available ports and queue of items "eating each other". This is similar to the known problem of thread scheduling in the OS: you have a queue if waiting threads and a queue of available CPI cores, and these queues "eat each other" on the tip of the scheduler. I hope you can get the hint.

Please clarify. And if you need some custom blocking not available with BlockingCollection, you can use my own generic BlockingQueue, at least as a starting point:

Simple Blocking Queue for Thread Communication and Inter-thread Invocation.

—SA

[no name] 24-Sep-14 16:51pm

To distribute ports is not my problem. I use the third party product, it can pick up a port by the framework. Likewise your example, either port 2 or 3 is not my concern. The framework will automatically select one. As my original code, we had m_ActiveCallCount

I only need to deal with the BlockingCollection, perhaps I still need m_ActiveCallCount < MaximumChannels. But if so, I don't see any benefit of BlockingCollection over Queue. The code is almost same. I thought TPL had the advantage. Maybe I should use the example.

http://blog.stephencleary.com/2012/11/async-producerconsumer-queue-using.html

Sergey Alexandrovich Kryukov 24-Sep-14 17:00pm

Still, you need to explain what's going on with this distribution. As I can understand, this is the root of the problem. You return to the issue of 4 yourself...
—SA

[no name] 24-Sep-14 17:06pm

Okay. The framework has a method.

m_ChannelResource = m_TelephonyServer.GetChannel();

If no port available, an exception will be thrown. Otherwise(no exception), one port is got. We don't know which port, either 3 or 4. We don't need to care it, anyway one port is occupied then.

Sergey Alexandrovich Kryukov 24-Sep-14 17:50pm

All right. Is there the way to say: "give me any available port"? Or do you want to try out 0 to 3, blocking exception propagation?
—SA

[no name] 24-Sep-14 20:05pm

Perhaps the first one. I don't want to try exception then return pattern. I would like to make sure that there is a port available then do work, otherwise wait.

Maybe we need something like:

var semaphore = new SemaphoreSlim(NUMBER_OF_PORTS);
var pendingTasks = new HashSet<task>();
var syncLock = new Object();

Action<task> queueTaskAsync = async(task) =>
{
lock (syncLock)
pendingTasks.Add(task);

await semaphore.WaitAsync().ConfigureAwait(false);

Just a guess, I am not strong on it.

Sergey Alexandrovich Kryukov 24-Sep-14 20:12pm

Wait a second. In such problems, it's important to determine the original trigger of events. What is that? If this is a customer call, it's important to understand how it gets into the system, what waits for it. The ultimate criterion is to avoid any polling, exactly as in case of the blocking collection. You should not jump to implementation. And — no guesswork.
—SA

1 solution

Add a Solution

This content, along with any associated source code and files, is licensed under The Code Project Open License (CPOL)

Sergey Alexandrovich Kryukov · Answer 1 · 2014-09-24T05:57:00

Please see my comment to the question. Even though it is not 100% clear to me, but this looks like a general problem of the producer-consumer pattern: [Sorry, we are experiencing the problem of broken anchors in CodeProject posts; please use:
http://en.wikipedia.org/wiki/Producer%E2%80%93consumer_problem;
sorry for the inconvenience].

The class BlockingCollection<T> offers bounding (not to be confused with the notion of binding):

MSDN tells us:
BlockingCollection<t> is a thread-safe collection class that provides the following:
[…]
A bounded collection that blocks Add and Take operations when the collection is full or empty.

Please pay attention for these constructors defining bounding:
http://msdn.microsoft.com/en-us/library/dd267301(v=vs.110).aspx[^],
http://msdn.microsoft.com/en-us/library/dd267306(v=vs.110).aspx[^]
and for this member: http://msdn.microsoft.com/en-us/library/dd287159(v=vs.110).aspx[^].

See also: How to: Add Bounding and Blocking Functionality to a Collection.

As to the "replacing" (why?), consider adding and the methods named Skip*.

—SA

Using BlockingCollection to replace queue.

1 solution

Solution 1

Add your solution here

Preview 0

Using BlockingCollection to replace queue.

1 solution

Solution 1

Add your solution here

Preview 0

Existing Members

...or Join us