Java blocking queue

Question

public class BQueue<T> {

    private Queue<T> q = new LinkedList<T>();
    private int limit;

    public BQueue(int limit) {
        this.limit = limit;
    }

    public synchronized void put (T t) throws InterruptedException {
        while (isFull()) {
            wait();
        }
        boolean e = isEmpty();
        q.add(t);
        if (e)
            notifyAll();
    }


    public synchronized T get () throws InterruptedException {
        while (isEmpty()) {
            wait();
        }
        boolean f = isFull();
        T t = q.poll();
        if (f)
            notifyAll();
        return t;
    }

    private boolean isEmpty() {
        return q.size() == 0;
    }
    private boolean isFull() {
        return q.size() == limit;
    }
}

Is this implementation thread-safe?

Answer 1

Yes, it's thread-safe. (Or at least I haven't found any issue.) Please note, that the Java Collections Framework already has a BlockingQueue interface which has some implementations, for example a LinkedBlockingQueue (source). It's probably well tested add has better performance, so try not to reinvent the wheel if it's not necessary.

Some other notes:

1. Try using longer variable names:

   boolean f = isFull();
   

   It could be isFull which results more readable code. The same is true for q and t. (I'd rename it to queue and item.)

2. Check your input: What happens when limit is 0 or less than zero? (You should throw an IllegalArgumentException.)

3. The limit and q fields could be marked final. It would improve code readability since readers don't have to check whether their values have changed somewhere in the class or not. It also would prevent accidental value modifications.

4. The used Queue has an isEmpty method (by implementing Collection.isEmpty()), you could use that instead of your own.

Answer 2

This looks thread-safe to me. However, I'm puzzled by the logic driving the notifications.

In put(), you wait until the queue is non-full (i.e., at least one slot open), then you add element t to the end of the queue. Cool, but then you only notify other threads if the queue was empty prior to adding t. Similarly, in get(), you wait until the queue is non-empty (i.e., has at least one item), then you fetch element t from the head of the queue. Again, cool, but then you only notify other threads if the queue was full prior to fetching t. This will have the effect of requiring the queue to be completely filled before it can be emptied, and conversely for it to be completely emptied before it can be refilled.

Maybe this was your intention, but this is not documented in the behavior, nor is it typical for a blocking queue.

I think what you might want is this instead:

public synchronized void put (T t) throws InterruptedException {
    while (isFull())
        wait();
    q.add(t);
    notifyAll();
}

and:

public synchronized T get () throws InterruptedException {
    while (isEmpty())
        wait();
    T t = q.poll();
    notifyAll();
    return t;
}

This version of put() will notify waiting threads immediately after a new element is added. This will wake up any waiting get() calls and immediately fetch the value you just added. This is usually what you want. If any other waiting put() call wakes up and finds that the queue is full (since you just added an element after possibly having waited for it to be non-full), it will simply cycle around the while loop and begin waiting again; no harm done.

Similarly, this version of get() will notify waiting threads immediately after an element is removed. This will wake up any waiting put() calls immediately and allow them to store a new value. This is also usually what you want. And if any other waiting get() call wakes up and finds that the queue is empty (since you may have just removed the last element), it will simply cycle around the while loop and begin waiting again until someone else adds another element.

Other than that, it looks good to me!

p.s. — Since get() and put() are synchronized, there's no harm in calling notifyAll() before you actually remove the item from the queue, because other waiting threads can't actually run again until you exit from get() (because you're not waiting). So, it can be simplified even further:

public synchronized T get () throws InterruptedException {
    while (isEmpty())
        wait();
    notifyAll();
    return q.poll();
}

Answer 3

It looks to me like it is thread-safe. But don't take my word for it; you better consider each answer as a vote, and go with what the majority says. (Multi-threading is tricky!) Note: comments should not be considered as answers.

Answer 4

I think its thread safe.. Also there can be multiple consumers as the method get is synchronized so only 1 consumer would be able to acquire lock on get.

When the first consumer releases the lock, then the second consumer would check the while loop that if its empty and would again enter the wait state.

So it seems fine to me.

Answer 5

This is thread safe if there are only one producer and one consumer. The get() may return null or throw exceptions if there are two consumers. For instance if queue is empty and 2 consumers calling get(), then both consumers will be suspended at the wait(). When the producer pushes an object into queue and notify both consumers, both consumers will break the while loop and call get(). Unfortunately one consumer will get the object but the other consumer will receive null object. The fix should extend the while loop to cover this case