Java synchronous queue implementation

Question

I just want to make sure there aren't any deadlocks or inconsistencies (the code is also available on GitHub).

Disclaimer - In real life, I would not implement a queue myself, but use an existing implementation. This is just preparations for job interviews.

package org.basic.concurrent;

import java.util.ArrayList;
import java.util.List;

public class SynchedQueue<T> {
    private final Object lock = new Object();
    private final Object full = new Object();
    private final Object free = new Object();

    private final List<T> buffer;
    private int head;
    private int tail;
    private int size;
    private final int capacity;

    public SynchedQueue(int capacity) {
        // http://stackoverflow.com/questions/4912088/how-to-create-a-fixed-size-generic-buffer-in-java
        buffer = new ArrayList<T>(capacity);
        for (int i = 0; i < capacity; ++i)
            buffer.add(null);
        this.capacity = capacity;
    }

    public void enqueue(T x) {
        if (x == null)
            throw new RuntimeException("Doesn't allow storing nulls");

        synchronized (lock) {
            while (!tryEnqueue(x)) {
                try {
                    free.wait();
                } catch (InterruptedException ignored) {
                }
            }
            full.notify();
        }
    }

    public T dequeue() {
        synchronized (lock) {
            while (true) {
                T item = tryDequeue();
                if (item != null)
                {
                    free.notify();
                    return item;
                }
                try {
                    full.wait();
                }
                catch (InterruptedException ignored) {}
            }
        }
    }

    private boolean tryEnqueue(T x) {
        assert size <= capacity;
        if (size >= capacity) {
            return false;
        }

        buffer.set(tail, x);
        tail = (tail + 1) % capacity;
        ++size;
        return true;
    }

    private T tryDequeue() {
        assert size >= 0;
        if (size == 0)
            return null;
        T item = buffer.get(head);
        head = (head + 1) % capacity;
        --size;
        return item;
    }
}

Answer 1

I have not coded in Java for a few years so it might be good to get another opinion, though some points did stand out while reading your code.

• It seems odd to me that you would use an ArrayList with a fixed size, personally I would just use an array for the buffer.

• when you dequeue, you do not overwrite the value until you enqueue over the top of it. This might not seem like an issue (and if you only queue primative types like int, its not) but if you use reference types and the queue cycles slowly then it means you hold a reference to the object for longer than needed and it cannot be collected.

• I'm a bit worried about catching the InterruptedException inside a while(true) loops without breaking out of the loop. The most common reason I have seen for one thread to interrupt another is when the interrupting thread wants to give the interrupted thread the opportunity to terminate gratefully, but this will prevent that option. I thought about what can be returned in such a case and eventually decided it might be best to let this exception flow through and be handled by the caller.

Answer 2

Have you tried running this implementation? It fails immediately with a java.lang.IllegalMonitorStateException which according to the Javadoc is:

Thrown to indicate that a thread has attempted to wait on an object's monitor or to notify other threads waiting on an object's monitor without owning the specified monitor.

The problem is that the implementation synchronizes on the field lock but calls notify() and wait() on full and free without holding their locks. When you fix this, it's important to keep in mind that calling wait() automatically releases the lock of the object you are waiting on, but does not release locks on other objects. If you don't take that into account, it's quite easy to create deadlocks.

For readability, I'd recommend using java.util.concurrent.locks.Condition instead of wait() and notify() which are fairly low level and difficult to reason about. In fact, the example usages in the Javadoc for Condition come from the implementation of a bounded buffer.

I also have to echo Brian's concern: it's important that you don't silently swallow InterruptedException. You have two choices on how to handle interruption. If you want to handle the exception yourself, then JCIP says you need to set the interrupted status back.

catch (InterruptedException e)
{
  Thread.interrupt();
  // Handle the interruption
}

The other choice, which is better in my opinion, is to just propagate the exception. Java's built in libraries use this strategy, for an example see BlockingQueue.take().

Answer 3

1. It looks like a BlockingQueue not just SynchQueue

2. throw new RuntimeException("Doesn't allow storing nulls"); - there is IllegalArgumnetException for this.

3. It would better to add throws InterruptedException to method declaration rather than swallowing it.

4. You are going to add/remove from the tail/head of the internal buffer - LinkedList looks like better options for internal storage. You don't need tons of Array.Copy in this case. It would also remove those tryDequeue() and tryEnqueue() and greatly simplify the code.

5. This code doesn't work in a single thread scenario. Consider the situation when I call enqueue() for the first time on queue then full.notify() will fail immediately.