Java queue with dynamic priority flag

Question

I need to build a queue where the elements will be added and removed in chronological order by default. But if the client sets the priority flag for the queue I need to be able to pull the elements based on the priority order of the elements.

I currently have this implemented using two priority queues, however I do not like the approach as it seems like an overkill.

Please let me know if there is a better way of doing this or if there is an existing implementation that exists.

import javax.naming.OperationNotSupportedException;
import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
public class DynamicPriorityQueue<ComparableQueueElement> implements IQueue<ComparableQueueElement> {
 private static final int CONSTANT_HUNDRED = 100;
 private boolean fetchByCustomPriority = false;
 private final ReentrantLock lock;
 private final PriorityQueue<ComparableQueueElement> queue;
 private final PriorityQueue<ComparableQueueElement> customPriorityQueue;
 public DynamicPriorityQueue() {
 this(null);
 }
 public DynamicPriorityQueue(Comparator<ComparableQueueElement> comparator) {
 this.lock = new ReentrantLock();
 this.queue = new PriorityQueue<>(CONSTANT_HUNDRED);
 if (comparator != null)
 this.customPriorityQueue = new PriorityQueue<ComparableQueueElement>(CONSTANT_HUNDRED, comparator);
 else
 this.customPriorityQueue = null;
 }
 public void setFetchByCustomPriority(boolean fetchByCustomPriority) throws OperationNotSupportedException {
 if (this.customPriorityQueue == null)
 throw new OperationNotSupportedException("Object was created without a custom comparator.");
 this.fetchByCustomPriority = fetchByCustomPriority;
 }
 public void push(ComparableQueueElement t) throws InterruptedException {
 if (this.lock.tryLock(CONSTANT_HUNDRED, TimeUnit.MILLISECONDS)) {
 try {
 this.queue.offer(t);
 if (this.customPriorityQueue != null)
 this.customPriorityQueue.offer(t);
 } finally {
 this.lock.unlock();
 }
 }
 }
 public ComparableQueueElement peek() {
 return this.fetchByCustomPriority ? this.queue.peek()
 : (this.customPriorityQueue != null ? this.customPriorityQueue.peek() : null);
 }
 public ComparableQueueElement pop() throws InterruptedException {
 ComparableQueueElement returnElement = null;
 if (this.lock.tryLock(CONSTANT_HUNDRED, TimeUnit.MILLISECONDS)) {
 try {
 if (this.fetchByCustomPriority && this.customPriorityQueue != null) {
 returnElement = this.customPriorityQueue.poll();
 this.queue.remove(returnElement);
 }
 else {
 returnElement = this.queue.poll();
 if (this.customPriorityQueue != null) {
 this.customPriorityQueue.remove(returnElement);
 }
 }
 } finally {
 this.lock.unlock();
 }
 }
 return returnElement;
 }
}

Answer 1

Constants naming

CONSTANT_HUNDRED is a bad name. It tells us its value, but not its role. Its value could be changed, making the whole thing useless, and in addition it is used in completely different contexts (queue size, and time out delay) so a user wishing to change one, might unknowingly alter the other.

You should declare two distinct constants:

private static final int QUEUE_DEFAULT_SIZE = 100;
private static final int LOCK_TIMEOUT_MILLIS = 100;

Usage:

 this.queue = new PriorityQueue<>(QUEUE_DEFAULT_SIZE);
 ...
 if (this.lock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {

---

Dubious Exception

When setting fetchByCustomPriority to false on a Queue which does not support custom priority, there will be an exception. There shouldn't be, because the user isn't asking for the unavailable functionality

---

Timing out breaks Queue contract

When timing out on pop(), you would return null. according to the Queue contract this null value means, "there are no Objects in the Queue now", but that may be wrong. It might just be that too many Threads are waiting to get their turn and this one got timed out.

You should Throw a custom TimedoutException to differentate between timing out, and having an empty queue.

Same goes for push().

---

Wrong usage of ternary operator ?

In peek:

return this.fetchByCustomPriority ? this.queue.peek() : (this.customPriorityQueue != null ? this.customPriorityQueue.peek() : null)

if the condition is true, the first member is returned, otherwise the second is returned. I suspect if fetchByCustomPriority is true, you'll want to use the customPriorityQueue, so you need to invert arguments 2 and 3.

---

Uneven popping

In push, you push messages on both queues.

In pop(), you either pop the queue, or the customPriorityQueue. This means both queues act differently depending on if fetchByCustomPriority is true or false. This makes your object inconsistent and breaks Queue interface contract:

ComparableQueueElement comparableQueue = new ComparableQueueElement(aComparator);
comparableQueue.push(element);
comparableQueue.setFetchByCustomPriority(true);
comparableQueue.pop(); // returns element
comparableQueue.setFetchByCustomPriority(false);
comparableQueue.pop(); // returns element AGAIN!!

Your objects should stay consistent.

You could fix this by popping both queues in pop() method.

Or you could design your functionality in a way that it cannot fail...

---

Infallible DynamicPriorityQueue

Your fields are too rich for your object. The two queues could become out of sync. The best way to fix this is to have your objects only allow states which make sense.

Since you're describing an Object wich is a queue, let the only field by a queue. I've changed to class to make it a bit more dynamic even, in that the comparator is no longer a fixed one. So now your can setFetchByCustomPriority(null) to fetch in-order, or setFetchByCustomPriority(comparatorA) and switch to setFetchByCustomPriority(comparatorB) on the fly.

import javax.naming.OperationNotSupportedException;
import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
public class DynamicPriorityQueue<ComparableQueueElement> implements IQueue<ComparableQueueElement> {
 private static final int CONSTANT_HUNDRED = 100;
 private final ReentrantLock lock;
 private PriorityQueue<ComparableQueueElement> queue;
 public DynamicPriorityQueue() {
 this(null);
 }
 public DynamicPriorityQueue(Comparator<ComparableQueueElement> comparator) {
 this.lock = new ReentrantLock();
 if (comparator != null) {
 this.queue = new PriorityQueue<ComparableQueueElement>(CONSTANT_HUNDRED, comparator);
 } else {
 this.queue = ew PriorityQueue<ComparableQueueElement>(CONSTANT_HUNDRED);
 }
 }
 public void setFetchByCustomPriority(Comparator<ComparableQueueElement> comparator) {
 // Just replace the existing queue
 PriorityQueue<ComparableQueueElement> newQueue = comparator == null ? new PriorityQueue(queue.size()) : new PriorityQueue(queue.size(), comparator);
 ComparableQueueElement element = queue.poll();
 while(element != null){
 newQueue.push(element); 
 element = queue.poll();
 }
 queue = newQueue;
 }
 public void push(ComparableQueueElement t) throws InterruptedException, TimedOutException {
 if (this.lock.tryLock(CONSTANT_HUNDRED, TimeUnit.MILLISECONDS)) {
 try {
 this.queue.offer(t);
 } finally {
 this.lock.unlock();
 }
 } else {
 throw new TimedOutException();
 }
 }
 public ComparableQueueElement peek() {
 return this.queue.peek();
 }
 public ComparableQueueElement pop() throws InterruptedException, TimedOutException {
 ComparableQueueElement returnElement = null;
 if (this.lock.tryLock(CONSTANT_HUNDRED, TimeUnit.MILLISECONDS)) {
 try {
 return this.queue.pop();
 } finally {
 this.lock.unlock();
 }
 }
 throw new TimedOutException();
 }
}

• dissatisfiedprogrammer: Thank you Sylvain for all the comments. I agree with all, but I do not think initializing a new queue at runtime will be scalable. It does allow flexibility but that comes at the cost of scalability.
• MrBrushy: It does indeed. I didn't know if it would be an issue, so I left it aside (you know what they say of premature optimization ^^). Sometimes we use queues to share and distribute amongst Threads, but the hard work is not the distribution, it's the message handling (small queues, big messages). If you need large queues, then I may review this part.
• dissatisfiedprogrammer: Thank you. Now that I had some more time to think about it I think having a linked list for chronological order and having priority based linked lists storing chronological index of primary linked list will work better for scalability. Let me know what do you think about that.
• MrBrushy: Sounds awesome. Chronological does not have to be LinkedList though if you store indices. But if you do use a LinkedList, you could store the LinkedNode directly instead of the index. Oh, I really like this. Would you propose an answer to your own question? I'd be interested in seeing this.

Answer 2

Looks like this works fine:

import entities.ComparableQueueElement;
import entities.EventPriority;
import javax.naming.OperationNotSupportedException;
import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.locks.ReentrantLock;
public class DynamicPriorityQueue implements IQueue {
 private boolean fetchByCustomPriority = false;
 private final ReentrantLock lock;
 private final LinkedList<ComparableQueueElement> queue;
 private final TreeMap<EventPriority, LinkedList<ComparableQueueElement>> eventPriorityLinkedListMap;
 public DynamicPriorityQueue() {
 this.lock = new ReentrantLock();
 this.queue = new LinkedList<>();
 this.eventPriorityLinkedListMap = new TreeMap<>();
 for(EventPriority eventPriority : EventPriority.values()) {
 this.eventPriorityLinkedListMap.put(eventPriority, new LinkedList<>());
 }
 }
 public void setFetchByCustomPriority(boolean fetchByCustomPriority) {
 this.fetchByCustomPriority = fetchByCustomPriority;
 }
 public void push(ComparableQueueElement element) {
 try {
 this.lock.lock();
 this.queue.offer(element);
 this.eventPriorityLinkedListMap.get(element.getPriority()).offer(element);
 } finally {
 if (this.lock.isLocked())
 this.lock.unlock();
 }
 }
 public ComparableQueueElement peek() {
 ComparableQueueElement element;
 if (this.fetchByCustomPriority) {
 element = this.queue.peek();
 }
 else {
 element = getPriorityElementsQueue().peek();
 }
 return element;
 }
 private LinkedList<ComparableQueueElement> getPriorityElementsQueue() {
 LinkedList<ComparableQueueElement> priorityElementQueue = null;
 for (Map.Entry<EventPriority, LinkedList<ComparableQueueElement>> entry :
 this.eventPriorityLinkedListMap.entrySet()) {
 if (!entry.getValue().isEmpty()) {
 priorityElementQueue = entry.getValue();
 break;
 }
 }
 return priorityElementQueue;
 }
 public ComparableQueueElement pop() {
 ComparableQueueElement returnElement = null;
 try {
 this.lock.lock();
 if (this.fetchByCustomPriority) {
 returnElement = getPriorityElementsQueue().poll();
 this.queue.remove(returnElement);
 }
 else {
 returnElement = this.queue.poll();
 this.eventPriorityLinkedListMap.get(returnElement.getPriority()).remove(returnElement);
 }
 } finally {
 if (this.lock.isLocked())
 this.lock.unlock();
 }
 return returnElement;
 }
}

• MrBrushy: Excellent work. I would rename getPriorityElementsQueue to getHighestPriorityQueue ('elements' add nothing to the name), but above all there is a potential bug, because if all queues are empty it'll return null! You'll throw a NPE. either return one of the empty queues (maybe init priorityElementQueue to the first element?) or perform null checks on the calling mehods. My advice is to never assign a variable to null to avoid those mistakes.
• dissatisfiedprogrammer: Thank you Sylvain. Will get that fixed so we can make the client wait until a valid entity is present to return.