5
\$\begingroup\$

My application has a thread for handling database inserts and updates. Data is added by other threads to a generic queue in the DB thread. The contents of this queue is retrieved and is sequentially inserted to the database by the thread.

To improve performance, I am planning to bring in a priority mechanism so that messages with higher priority should get inserted in the database before those with lower priority.

I implemented a class with a SortedDictionary for handling the priority. An enum indicating the priority would be the key and a Queue containing the messages would be the value part of the dictionary.

Please review the following code and provide suggestions on how to improve it. Also if there are any readily available solutions please provide the details.

public class PriorityQueue
{
    private Object lockObj;
    private SortedDictionary<PQMsgPriority, Queue<PQMessage>> messageDictionary;

    public PriorityQueue()
    {
        lockObj = new object();
        messageDictionary = new SortedDictionary<PQMsgPriority, Queue<PQMessage>>();
    }

    public void Enqueue(PQMessage item)
    {
        lock (lockObj)
        {
            if (item != null)
            {
                if (messageDictionary.ContainsKey(item.MsgPriority))
                {
                    Queue<PQMessage> dataList = messageDictionary[item.MsgPriority];
                    dataList.Enqueue(item);
                }
                else
                {
                    Queue<PQMessage> dataList = new Queue<PQMessage>();
                    dataList.Enqueue(item);
                    messageDictionary.Add(item.MsgPriority, dataList);
                }
            }
        }
    }

    public PQMessage Dequeue()
    {
        lock (lockObj)
        {
            PQMessage messageData = null;
            PQMsgPriority prioKeyDeleteFlag = PQMsgPriority.None;

            //If no data available, throw an exception
            if (messageDictionary.Count == 0)
                throw new InvalidOperationException();

            foreach (KeyValuePair<PQMsgPriority, Queue<PQMessage>> item in messageDictionary)
            {
                Queue<PQMessage> dataList = item.Value;
                if (dataList.Count > 0)
                    messageData = dataList.Dequeue();
                else
                {
                    prioKeyDeleteFlag = item.Key;
                    continue;
                }
                break;
            }

            if (prioKeyDeleteFlag != PQMsgPriority.None)
                messageDictionary.Remove(prioKeyDeleteFlag);

            return messageData;
        }
    }

    public int Count()
    {
        lock (lockObj)
        {
            int totalCount = 0;
            foreach (KeyValuePair<PQMsgPriority, Queue<PQMessage>> item in messageDictionary)
            {
                if (item.Value != null)
                {
                    if (item.Value.Count == 0)
                        continue;
                    else
                    {
                        totalCount = totalCount + item.Value.Count;
                        continue;
                    }
                }
                break;
            }
            return totalCount;
        }
    }

    public PQMessage Peek()
    {
        lock (lockObj)
        {
            PQMessage messageData = null;

            //If no data available, throw an exception
            if (messageDictionary.Count == 0)
                throw new InvalidOperationException();

            foreach (KeyValuePair<PQMsgPriority, Queue<PQMessage>> item in messageDictionary)
            {
                Queue<PQMessage> dataList = item.Value;
                if (dataList.Count > 0)
                    messageData = dataList.Peek();
                break;
            }

            return messageData;
        }
    }
}

The priority enum is defined as follows:

 public enum PQMsgPriority
 {
     High = 0,
     Medium = 1,
     Low = 2,
     None = 3
 }

The priority message is defined as follows:

public class PQMessage
{
    private PQMsgPriority msgPriority;
    private Object message;

    #region Properties
    public PQMsgPriority MsgPriority
    {
        get { return msgPriority; }
       set { msgPriority = value; }
    }
    public Object Message
    {
        get { return message; }
        set { message = value; }
    }
    #endregion

    public PQMessage(PQMsgPriority msgPriority, Object message)
    {
        this.msgPriority = msgPriority;
        this.message = message;
    }
}
\$\endgroup\$
8
\$\begingroup\$
  1. I don't like that you're treat PQMsgPriority.None in a special way in Dequeue(). If somebody tried to enqueue something with this priority, the code wouldn't work as expected.
  2. I think your Peek() wouldn't work correctly in some cases. Specifically, Dequeue() can leave the queue with the highest priority empty, while other queues still have items. In this case Peek() would incorrectly return null.
  3. I think you're overusing break and continue. They can be useful, but I think the way you're using them makes your code harder to read. For example your (incorrect) code in Peek() could be simply rewritten using First().
  4. Your Count() could be simplified to messageDictionary.Values.Sum(q => q.Count). Especially the check for 0 seems completely useless here (possibly an attempt at nano-optimization?). The null check also seems useless, if the dictionary contained null values, that's a bug you're hiding by this.
  5. It would make sense to make this class generic in both the priority type and the message type.
\$\endgroup\$
  • \$\begingroup\$ Btw, Dictionary cannot contain null keys - it will throw an exception. \$\endgroup\$ – ANeves May 17 '12 at 13:26
5
\$\begingroup\$

Enqueue

When item is null, it won't be inserted. I suggest that you check at the start, before the lock, whether item is null, and if it is, throw an ArgumentNullException.

Your code for enqueuing an item is the same in both cases of the if-statement. Move that code out of it.

Queue<PQMessage> dataList;
if (messageDictionary.ContainsKey(item.MsgPriority))
{
    dataList = messageDictionary[item.MsgPriority];
}
else
{
    dataList = new Queue<PQMessage>();
    messageDictionary.Add(item.MsgPriority, dataList);
}

dataList.Enqueue(item);

Dequeue

I'd move the check whether the number if items in the dictionary is zero to the top of the method. It does not need to be inside the lock.

If you want, you may use foreach(var item in messageDictionary) instead, as it is cleaner (to me).

Rethink how you delete an empty entry from the dictionary. The way it is now, if the dictionary contains two empty queues, only the last one gets deleted. Also note that you may use nullable types (e.g. PQMsgPriority? prioKeyDeleteFlag = null;) so that you don't need to assign special behavior to PQMsgPriority.None.

Try to use the fact that the foreach will continue on its own. Remove the continue keyword, and this requires you to move the break closer to the code that requires it.

Count

You made Count a method, which is not common in collection classes. On the other hand, the rule is to make something a method when it performs many calculations. In this case you can go either way, altough I would prefer for Count to be a property. If you are concerned about performance, you could adjust the count on each enqueue or dequeue.

For the foreach, reorder the break to be where you want it:

foreach (KeyValuePair<PQMsgPriority, Queue<PQMessage>> item in messageDictionary)
{
    if (item.Value == null)
        break;
    else
    {
        if (item.Value.Count == 0)
            continue;
        else
        {
            totalCount = totalCount + item.Value.Count;
            continue;
        }
    }
}

Again, use in your implementation the fact that foreach will continue on its own:

foreach (KeyValuePair<PQMsgPriority, Queue<PQMessage>> item in messageDictionary)
{
    if (item.Value == null)
        break;
    else
    {
        if (item.Value.Count != 0)
            totalCount = totalCount + item.Value.Count;
    }
}

And note that in your code it is not possible for item.Value to be null, so you don't have to check for that:

foreach (var item in messageDictionary)
{
    if (item.Value.Count != 0)
        totalCount += item.Value.Count;
}

Adding 0 is not a problem:

foreach (var item in messageDictionary)
{
    totalCount += item.Value.Count;
}

And did you know that LINQ is very useful for this kind of looping over collections?

public int Count
{
    get 
    {
        return messageDictionary
            .Sum(item => item.Value.Count);
    }
}

Peek

Again, move your Count == 0 check to the top of the method.

Your foreach is incorrect. It will only every iterate to the first item in the dictionary, as break is a statement that it will always encounter. You probably wanted to continue iterating the dictionary when you encounter an empty queue in the current item. For example:

foreach (var item in messageDictionary)
{
    Queue<PQMessage> dataList = item.Value;
    if (dataList.Count > 0)
    {
        messageData = dataList.Peek();
        break;
    }
}

LINQ again may make your life easier here:

var messageData = messageDictionary     // From each entry in the dictionary
    .Select(item => item.Value)         // select the queue
    .First(queue => queue.Count > 0)    // and take the first queue with Count > 0
    .Peek();                            // and call Peek() on that.

In general: check all preconditions at the start of the method. Use the inherent qualities of the looping constructs instead of explicit break and continue. Especially continue is something you should almost never need. Try reading up on LINQ, it may solve many problems you'll encounter while programming.

\$\endgroup\$
1
\$\begingroup\$

Your locking scope to too large. I only see one or two places in each method where locking is needed, yet the entire method is locked.

Is there any reason a SortedDictionary is used? If there is not, have you considered a ConcurrentDictionary? From what I can tell this would eliminate your need to lock all together.

\$\endgroup\$
0
\$\begingroup\$

That is actually a limited priority queue - at least according to the fact that you have an enum with a discrete amount of values. Due to the fact that it's limited you are wasting both CPU cycles and memory because you are dealing with a Dictionary - when an array will do:

private Queue<PQMessage>[] priorityQueue;
private int maxPriority;

public PriorityQueue()
{
    maxPriority = Enum.GetValues(typeof(PQMsgPriority)).Cast<int>().Max();
    priorityQueue = new Queue<PQMessage>[maxPriority + 1];
    for (var i = 0; i <= maxPriority; i++)
        priorityQueue[i] = new Queue<PQMessage>();
}

Now you can grab a queue by merely casting your enum value to an int. We never have to alter our array again we don't have to lock it (we are only retrieving elements from it) - so we only need to lock the queue in question.

public void Enqueue(PQMessage item)
{
    if (item != null)
    {
        if ((int)item.MsgPriority < 0 || (int)item.MsgPriority > maxPriority)
            throw new ArgumentOutOfRangeException("item.MsgPriority");

        var queue = priorityQueue[(int)item.MsgPriority];
        lock (queue) 
            queue.Enqueue(item);
    }
}

public PQMessage Dequeue()
{
    for (var i = 0; i <= maxPriority; i++)
    {
        var queue = priorityQueue[i];
        lock (queue)
        {
            if (queue.Count > 0)
                return queue.Dequeue();
        }
    }
    return null;
}

If you are using .Net 4.0 you can use the ConcurrentQueue (System.Collections.Concurrent) to eliminate locks altogether; as such:

class PriorityQueue
{
    private ConcurrentQueue<PQMessage>[] priorityQueue;
    private int maxPriority;

    public PriorityQueue()
    {
        maxPriority = Enum.GetValues(typeof(PQMsgPriority)).Cast<int>().Max();
        priorityQueue = new ConcurrentQueue<PQMessage>[maxPriority + 1];
        for (var i = 0; i <= maxPriority; i++)
            priorityQueue[i] = new ConcurrentQueue<PQMessage>();
    }

    public void Enqueue(PQMessage item)
    {
        if (item != null)
        {
            if ((int)item.MsgPriority < 0 || (int)item.MsgPriority > maxPriority)
                throw new ArgumentOutOfRangeException("item.MsgPriority");

            var queue = priorityQueue[(int)item.MsgPriority];
            queue.Enqueue(item);
        }
    }

    public PQMessage Dequeue()
    {
        PQMessage result = null;
        for (var i = 0; i <= maxPriority; i++)
        {
            var queue = priorityQueue[i];
            if (queue.TryDequeue(out result))
                break;
        }
        return result;
    }
}

As a final step I would make it more re-usable:

class PQMsgPriorityQueue : PriorityQueue<PQMsgPriority, PQMessage>
{
    public PQMsgPriorityQueue()
        : base(x => x.MsgPriority)
    {

    }
}

class PriorityQueue<TKey, TItem>
    // struct (ValueType) so we can ignore null checks, and all enums are value types
    // IConvertible so that we can convert values to ints, and all enums are IConvertible
    where TKey : struct, IConvertible
{
    private ConcurrentQueue<TItem>[] priorityQueue;
    private int maxPriority;
    private Func<TItem, TKey> extractKey;
    private int count;

    /// <summary>
    /// Gets the approximate amount of items in the queue.
    /// </summary>
    public int Count
    {
        get { return Thread.VolatileRead(ref count); }
    }

    public PriorityQueue(Func<TItem, TKey> extractKey)
    {
        if (extractKey == null)
            throw new ArgumentNullException("extractKey");

        this.extractKey = extractKey;
        // This will validate that the key type is an enum.
        maxPriority = Enum.GetValues(typeof(TKey)).Cast<IConvertible>().Select(x => x.ToInt32(null)).Max();
        priorityQueue = new ConcurrentQueue<TItem>[maxPriority + 1];
        for (var i = 0; i <= maxPriority; i++)
            priorityQueue[i] = new ConcurrentQueue<TItem>();
    }

    public void Enqueue(TItem item)
    {
        if (item != null)
        {
            var key = extractKey(item).ToInt32(null);
            if (key < 0 || key > maxPriority)
                throw new ArgumentOutOfRangeException("item>key");

            var queue = priorityQueue[key];
            queue.Enqueue(item);
            Interlocked.Increment(ref count);
        }
    }

    public TItem Dequeue()
    {
        TItem result;
        TryDequeue(out result);
        return result;
    }

    public bool TryDequeue(out TItem result)
    {
        for (var i = 0; i <= maxPriority; i++)
        {
            var queue = priorityQueue[i];
            if (queue.TryDequeue(out result))
            {
                Interlocked.Decrement(ref count);
                return true;
            }
        }
        result = default(TItem);
        return false;
    }

    // The presence of peek is indicative of race conditions
    // as such I have not implemented it.
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.