CODEWITHSUNDEEP
c#.netconcurrencyqueue
bmt22033
bmt22033

Reputation: 7240

Consumer/producer pattern using a List

I have a WinForms app with one consumer and one producer task. My producer task periodically connects to a web service and retrieves a specified number of strings which then need to be placed into some kind of concurrent fixed-size FIFO queue. My consumer task then processes these strings and sends then out as SMS messages (one string per message). The SOAP function that my producer task calls requires a parameter to specify the number of strings that I want to get. This number will be determined by the space available in my queue. So if I have a max queue size of 100 strings and I have 60 strings in the queue the next time my producer polls the web service, I need it to ask for 40 strings since that's all that I can fit in my queue at that moment.

Here's the code that I'm using to represent my fixed-size FIFO queue:

public class FixedSizeQueue<T>
{
    private readonly List<T> queue = new List<T>();
    private readonly object syncObj = new object();

    public int Size { get; private set; }

    public FixedSizeQueue(int size)
    {
        Size = size;
    }

    public void Enqueue(T obj)
    {
        lock (syncObj)
        {
            queue.Insert(0, obj);

            if (queue.Count > Size)
            {
                queue.RemoveRange(Size, queue.Count - Size);
            }
        }
    }

    public T[] Dequeue()
    {
        lock (syncObj)
        {
            var result = queue.ToArray();
            queue.Clear();
            return result;
        }
    }

    public T Peek()
    {
        lock (syncObj)
        {
            var result = queue[0];
            return result;
        }
    }

    public int GetCount()
    {
        lock (syncObj)
        {
            return queue.Count;
        }
    }

My producer task doesn't currently specify the number of strings that I need from the web service but it seems like it could be as simple as getting the current item count in my queue (q.GetCount()) and then subtracting it from my max queue size. However, even though GetCount() uses a lock, isn't it possible that as soon as GetCount() exits, my consumer task could process 10 strings in the queue meaning that I'll never actually be able to keep the queue 100% full?

Also, my consumer task basically needs to "peek" at the first string in the queue before trying to sent it in an SMS message. In the event that the message can't be sent, I need to leave the string in it's original position in the queue. My first thought about accomplishing this is to "peek" at the first string in the queue, try to send it in an SMS message and then remove it from the queue if the send was successful. This way, if the send fails, the string is still in the queue at its original position. Does that sound reasonable?

Upvotes: 0

Views: 635

Answers (1)

cdhowie
cdhowie

Reputation: 169028

This is a broad question, so there really is no definitive answer, but here are my thoughts.

However, even though GetCount() uses a lock, isn't it possible that as soon as GetCount() exits, my consumer task could process 10 strings in the queue meaning that I'll never actually be able to keep the queue 100% full?

Yes, it is, unless you lock on syncObj for the entire duration of your query to the web service. But the point of producer/consumer is to allow the consumer to process items while the producer is fetching more. There's really not much you can do about this; at some point, the queue will not be 100% full. If it always was 100% full then that would mean that the consumer isn't doing anything at all.

This way, if the send fails, the string is still in the queue at its original position. Does that sound reasonable?

Perhaps, but the way you have this coded, a Dequeue() operation returns the entire state of the queue and clears it. Your only option given this interface is to re-queue failed items to be processed later, which is a perfectly reasonable technique.

I would also consider adding a way for the consumer to block itself until there are items to be processed. For example:

public T[] WaitForItemAndDequeue(TimeSpan timeout)
{
    lock (syncObj) {
        if (queue.Count == 0 && !Monitor.Wait(syncObj, timeout)) {
            return null; // Timeout expired
        }

        return Dequeue();
    }
}

public T[] WaitForItem()
{
    lock (syncObj) {
        while (queue.Count != 0) {
            Monitor.Wait(syncObj);
        }

        return Dequeue();
    }
}

Then you have to change Enqueue() to call Monitor.Pulse(syncObj) after it has manipulated the list (so at the end of the method, but inside of the lock block).

Upvotes: 1

Related Questions