ArrayBlockingQueue: concurrent put and take I have implemented an ArrayBlockingQueue (put and take) on the lines of LinkedBlockingQueue i.e using two locks so that my take should not block my put.

For more information please see this Stack Overflow question.  The code is also available at GitHub.

Please review the code for any race conditions.

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ArrayBlockingQueue&lt;T&gt; {

    private final int capacity;
    private final AtomicInteger count = new AtomicInteger(0); 
    private final Lock putLock = new ReentrantLock();
    private final Lock takeLock = new ReentrantLock();
    private final Condition notEmpty = takeLock.newCondition(); // putLock will signal that its not empty
    private final Condition notFull = putLock.newCondition(); // signal when queue is notFull
    private final Object [] items ;
    int takeIndex;
    int putIndex;

    public ArrayBlockingQueue(int capacity) {
        this.capacity = capacity;
        items = new Object[capacity];

    }

    public void add(T t) throws InterruptedException {
        int c = -1;
        final Lock lock = this.putLock;
        final AtomicInteger count = this.count;
        lock.lock();
        try {
            while(count.get() == capacity) {
                    notFull.await();
            }
            enqueue(t);
            c = count.getAndIncrement();
            if(c+1 &lt; capacity) {
                notFull.signal();
            }
        }finally {
            lock.unlock();
        }
        if(c == 0) {
            signalNotEmpty();
        }
    }

    public T take() throws InterruptedException {
        int c = -1;
        final Lock lock = this.takeLock;
        final AtomicInteger count = this.count;
        T t = null;
        lock.lock();
        try {
            while(count.get() == 0) {
                notEmpty.await();
            }
            t = dequeue();
            c = count.getAndDecrement();
            if(c &gt; 1) {
                notEmpty.signal();
            }
        }finally {
            lock.unlock();
        }
        if(c == capacity) {
            signalNotFull();
        }
        return t;
    }

    private void signalNotEmpty() {
        final Lock lock = this.takeLock;
        lock.lock();
        try {
            notEmpty.signal();
        }finally {
            lock.unlock();
        }
    }

    private void signalNotFull() {
        final Lock lock = this.putLock;
        lock.lock();
        try {
            notFull.signal();
        }finally {
            lock.unlock();
        }
    }

    private T dequeue() {
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        T x = (T) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        return x;
    }


    private void enqueue(T t) {
        final Object[] items = this.items;
        items[putIndex] = t;
        if (++putIndex == items.length)
            putIndex = 0;
    }


}
Typo

In your take() function, you wrote this:

            notFull.signal();


But I'm pretty sure you meant to write this:

            notEmpty.signal();


Unnecessary Signalling? Nope

Originally, I thought that the extra signalling in the add() and take() functions were unnecessary.  In fact, they are not required, but when they are added, it improves performance.  It is a concept called "cascading notifies".  The LinkedBlockingQueue sources  explain it like this:


  Also, to minimize need for puts to get takeLock and vice-versa, cascading notifies are used. When a put notices that it has enabled at least one take, it signals taker. That taker in turn signals others if more items have been entered since the signal. And symmetrically for takes signalling puts.


Race condition? Nope

Originally, I thought that the code had a race condition involving memory reordering when accessing the shared queue.  However, it was pointed out that AtomicInteger acts as a volatile variable and provides a memory barrier.  So therefore the accesses are safe.

Update

After reading @rolfl's answer with his multiple variants that attempt to improve upon the original function, I was inspired to come up with my own variant.  I came up with one that is simpler than any of the variants listed here so far.  It basically uses two semaphores, normal synchronization, and an array:

import java.util.concurrent.Semaphore;

public class ArrayBlockingSemQueue&lt;T&gt; implements BlockQueue&lt;T&gt;
{
    private final int capacity;
    private final Semaphore notEmpty;
    private final Semaphore notFull;
    private final Object [] items;
    int takeIndex;
    int putIndex;

    public ArrayBlockingSemQueue(int capacity) {
        this.capacity = capacity;
        notEmpty = new Semaphore(0);
        notFull  = new Semaphore(capacity);
        items = new Object[capacity];
    }

    public void add(T t) throws InterruptedException {
        notFull.acquire();

        synchronized(notFull) {
            items[putIndex++] = t;
            if (putIndex == capacity)
                putIndex = 0;
        }

        notEmpty.release();
    }

    @SuppressWarnings("unchecked")
    public T take() throws InterruptedException {
        T ret;

        notEmpty.acquire();

        synchronized(notEmpty) {
            ret = (T) items[takeIndex++];
            if (takeIndex == capacity)
                takeIndex = 0;
        }

        notFull.release();
        return ret;
    }
}


Benchmarks

I used the same code that @rolfl listed in his answer to test the timings of the various Queues.  Here are my results:

           1x1   2x2   3x3   4x4   6x6   12x12 1x12  2x6   3x4   4x3   6x2   12x1
Veritas    3.32  3.39  3.54  3.59  3.54  3.62  6.63  3.95  3.67  3.68  3.88  6.16
DoubleSync 3.72  4.97  4.87  4.94  5.07  4.78  5.43  5.01  4.96  4.93  5.01  5.41
CDLatch    2.89  2.56  4.62  5.02  5.83  8.01  8.25  5.95  5.24  3.60  2.58  2.65
JS1        3.05  3.00  3.18  3.08  3.04  3.09  3.47  3.13  3.26  3.07  3.00  3.15