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.

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<T> {

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 < 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 > 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;
}

}

• @jamal thanks for editing, post looks better now – veritas Apr 17 '15 at 2:42
• Hi. Have you checked the Javadoc for Condition? It has a straightforward implementation of a BoundedBuffer, which I believe is what you're looking for (just need to parameterize the capacity). – Elegie Apr 17 '15 at 6:39
• @Elegie Hi, now the example in javadoc for condition is again using a single lock. whereas my implementation is specifically using separate locks for put and take i.e. is the main thing i wanted to verify – veritas Apr 17 '15 at 6:45
• If you haven't already done so, you should look at Disruptor; which similarly uses a shared ring buffer as a queue (but tuned for a different set of use cases). github.com/LMAX-Exchange/disruptor – VoiceOfUnreason Apr 17 '15 at 19:55
• @VoiceOfUnreason thanks,yes in my current project we are already using Disruptor – veritas Apr 17 '15 at 20:20

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<T> implements BlockQueue<T>
{
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

• its not a typo, please see LinkedBlockingQueue Implementation. There are no unnecessary signaling, instead its a good design where consumer thread only signals the producer thread only when the queue is full, else unblocking of threads is clearly taken care by producer thread. Also AtomicInteger takes care memory ordering especially happens before relationship so no visibility problem there. See LinkedBlockingQueue impl u will understand what i am talking – veritas Apr 17 '15 at 7:47
• So as such all three of your observations are wrong. anyways thanks for reviewing it carefully – veritas Apr 17 '15 at 7:49
• @veritas It is a typo. Read the LinkedBlockingQueue put function. You will see. I'm referring to the empty vs full. You may be right about the signalling being ok. – JS1 Apr 17 '15 at 8:04
• @yahh am sorry i was looking at the add function – veritas Apr 17 '15 at 8:07
• @veritas Sorry for the incorrect review. I have now updated the answer to account for the things that you pointed out. Seems like I have a lot to learn about Java! – JS1 Apr 17 '15 at 8:53

Simplified code

In order to understand your code, I copied it out, and simplified it by:

1. removing redundant variables (why do you take copies of private-final fields in your methods?) The following code is particularly bizarre:

final AtomicInteger count = this.count;

2. ... there was going to be another thing, but then I realized I was wrong...

OK, is your code thread safe? Yes, I believe so. It is horribly complicated, though....

You have tagged your question as ... it obviously is not lock-free. You have multiple locks. What's more, is that it is overly-locked.

Additionally, I predict the performance in a concurrent situation will be much worse than you expect. Note that, for the typical put/take, you do 4 locking operations, and two atomic operations

1. lock the put/tack Lock
2. atomic get on the count
3. atomic update the count
4. unlock the lock
5. lock the other lock
6. signal the other lock
7. unlock the other lock.

Atomic operations are fast, but they are not free. Additionally, since you are using the same atomic count in both the put and get, and also, since both put, and take lock both putLock and takeLock (and access count), you are not actually accomplishing the stated goal of "i.e using two locks so that my take should not block my put."

In my experience, the value of ReentrantLock is not related to the fact that it is in the concurrent package and "new", it is in the fact that the cost of a double-barrier (lock/unlock) when amortized over a long-running piece of code (which your code is not), is worth it for the cool features (like multiple signals on one lock instance, or in-the-middle-of-execution unlocking). You do neither of those things. Your use of the ReentrantLock is basically just slowing you down, and making your code complicated.

Additionally, what you see as an 'optimization' for only notifying the other method if there is something new, is actually just slowing things down too.

Your code would be simpler, and faster, and more concurrent with traditional synchronization...

public class ArrayBlockingSyncQueue<T> {

private final Object [] items ;
private int takeIndex;
private int size;

public ArrayBlockingSyncQueue(int capacity) {
items = new Object[capacity];
}

public void add(T t) throws InterruptedException {
synchronized(items) {
while (size == items.length) {
items.wait();
}
int putIndex = (takeIndex + size) % items.length;
items[putIndex] = t;
size++;
items.notifyAll();
}
}

public T take() throws InterruptedException {
synchronized(items) {
while (size == 0) {
items.wait();
}
@SuppressWarnings("unchecked")
T t = (T)items[takeIndex];
items[takeIndex] = null; //for GC
takeIndex++;
size--;
if (takeIndex >= items.length) {
takeIndex = 0;
}
items.notifyAll();
return t;
}
}

}


Not Fast Enough - BENCHMARKS

UPDATE: After benchmarking your code, against my suggestion above, I found your code was faster than I expected. This runs against my experience with Locks, which I have, in the past, found to be slower than synchronization due to the double-lock process (lock barrier going in, and again when coming out).

I was expecting the plain-jane synchronzation to out-perform the complciated locking, and I was wrong. So, I investigated, and have the following thoughts on the results:

1. your strategy is faster than the single-sync because the access-loop is so tight in my testing, that all threads are locked on a single monitor. Your locking is locked on 2, so, even though your individual locking is slower than synchronization, it's still doing 2 threads at once, not one.
2. double-entry synchronization should still be faster than the locks, in a tight loop.
3. the test is essentially invalid, because there is no job out there which does nothing except feed, or empty a queue. All real jobs do something before giving, or after taking the data, and in a typical situation, the actual time in the queue should be almost negligible.

So, based on the above, I implemented a collection of benchmarks using the following strategies:

1. build an interface of the add/take methods on your queue called BlockingQueue

2. build a tight-loop putter-thread, and a tight-loop taker thread. They look like:

private final int runPuts(int id) throws InterruptedException {
long waiting = 0;
int base = id * count;
for (int i = 0; i < count; i++) {
}
return count;
}

private final long runGets() {
long sum = 0;
try {
while (true) {
Integer got = queue.take();
if (got == POISON) {
break;
}
sum += got;
}
} catch (InterruptedException ie) {
ie.printStackTrace();
}
return sum;
}

3. Have multiple threads putting values, and taking values. Each value put on to the queue is unique (it is the thread-ID * count-per-thread + i) The if all the values are removed from the queue, then added up, the sum should be predictable, and is a good test to ensure that every value added to the queue is also removed from it. Any error should be immediately obvious if the sum is wrong.

4. Run the same test for multiple implementations of BlockingQueue, at multiple different combinations of Producers and Consumers.

What were the results? Comparing the above single-monitor synchronization with your double-locked implementation, using 20-capacity queues, and passing 12,000 values through, the results are:

               1x1,   2x2,   3x3,   4x4,   6x6, 12x12,  1x12,   2x6,   3x4,   4x3,   6x2,  12x1
Veritas, 2.267, 5.659, 3.349, 3.884, 3.599, 4.256, 4.931, 4.970, 4.216, 3.490, 6.699, 6.582
SingleSync, 3.773, 3.111, 4.057, 4.551, 5.440, 6.894, 5.813, 3.276, 4.435, 5.012, 4.519, 4.691


Note a few things in there.... my Computer is a 4-core i7, and it has slow performance on your locking when there's 4 threads at 100% (two producers, 2 consumers). Something 'odd' happens then. The rest of the time it is fairly predictable.

The single-sync code is slower than yours for simple cases, but starts performing better than yours under high thread imbalance - where there are many more consumers than producers, or the other way.

Double-ended Synchronization

Still, the results are interesting for their variance.... but, what if I used a similar double-ended synchronization system to your double-ended locks? What would happen then?

import java.util.concurrent.atomic.AtomicInteger;

public class ArrayBlockingSyncDeQueue<T> implements BlockQueue<T>{

private final AtomicInteger count = new AtomicInteger(0);
private final Object putLock = new Object();
private final Object takeLock = new Object();
private final Object [] items;
int takeIndex;
int putIndex;

public ArrayBlockingSyncDeQueue(int capacity) {
items = new Object[capacity];
}

@Override
public void add(T t) throws InterruptedException {
synchronized(putLock) {
while (count.get() == items.length) {
putLock.wait();
}
items[putIndex] = t;
putIndex++;
putIndex %= items.length;
int used = count.incrementAndGet();
if (used < items.length) {
putLock.notify();
}
if (used > 1) {
// no need to notify takers.
// just exit.
return;
}

}
synchronized(takeLock) {
takeLock.notify();
}
}

@Override
public T take() throws InterruptedException {
boolean notify = false;
try {
synchronized(takeLock) {
while (count.get() == 0) {
takeLock.wait();
}
@SuppressWarnings("unchecked")
T t = (T) items[takeIndex];
items[takeIndex] = null;
takeIndex++;
takeIndex %= items.length;
int used = count.getAndDecrement();
if (used == items.length) {
notify = true;
}
if (used > 1) {
takeLock.notify();
}
return t;
}
} finally {
if (notify) {
synchronized (putLock) {
putLock.notify();
}
}
}
}

}


Well,t hat is essentially always faster than the Reentrant lock system of yours:

               1x1,   2x2,   3x3,   4x4,   6x6, 12x12,  1x12,   2x6,   3x4,   4x3,   6x2,  12x1
Veritas, 2.267, 5.659, 3.349, 3.884, 3.599, 4.256, 4.931, 4.970, 4.216, 3.490, 6.699, 6.582
DoubleSync, 1.680, 3.497, 3.451, 3.585, 5.465, 3.627, 4.738, 3.398, 4.432, 4.030, 4.241, 4.071
SingleSync, 3.773, 3.111, 4.057, 4.551, 5.440, 6.894, 5.813, 3.276, 4.435, 5.012, 4.519, 4.691


Java library

But, how would this compare to the standard Java Concurrent library (which has the BlockingQueues too)?

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ConcBlockingArrayQueue<T> implements BlockQueue<T> {

private final BlockingQueue<T> q;

public ConcBlockingArrayQueue(int capacity) {
q = new ArrayBlockingQueue<>(capacity);
}

@Override
public void add(T t) throws InterruptedException {
q.put(t);
}

@Override
public T take() throws InterruptedException {
return q.take();
}

}


and

import java.util.concurrent.BlockingQueue;

public class ConcBlockingLinkedQueue<T> implements BlockQueue<T> {

private final BlockingQueue<T> q;

}

@Override
public void add(T t) throws InterruptedException {
q.put(t);
}

@Override
public T take() throws InterruptedException {
return q.take();
}

}


Well, the Linked one is fast, the Array one is slow....

               1x1,   2x2,   3x3,   4x4,   6x6, 12x12,  1x12,   2x6,   3x4,   4x3,   6x2,  12x1
Veritas, 2.267, 5.659, 3.349, 3.884, 3.599, 4.256, 4.931, 4.970, 4.216, 3.490, 6.699, 6.582
DoubleSync, 1.680, 3.497, 3.451, 3.585, 5.465, 3.627, 4.738, 3.398, 4.432, 4.030, 4.241, 4.071
SingleSync, 3.773, 3.111, 4.057, 4.551, 5.440, 6.894, 5.813, 3.276, 4.435, 5.012, 4.519, 4.691
ConcLinked, 2.982, 5.078, 3.463, 3.634, 3.658, 3.966, 4.353, 5.867, 3.362, 3.536, 3.572, 4.375
ConcArray, 3.045, 3.246, 6.018, 5.377, 5.586, 6.207,23.079, 6.781, 6.544, 6.924, 6.609,12.136


Lock Free (for real this time).

But, finally, what if we actually did use a Lock-free implementation? Using no locks, just semaphores, and atomics?

Here's the preview:

               1x1,   2x2,   3x3,   4x4,   6x6, 12x12,  1x12,   2x6,   3x4,   4x3,   6x2,  12x1
Veritas, 2.267, 5.659, 3.349, 3.884, 3.599, 4.256, 4.931, 4.970, 4.216, 3.490, 6.699, 6.582
LockFree, 2.797, 2.884, 2.951, 3.076, 3.391, 3.461, 3.186, 3.176, 2.901, 3.225, 3.602, 4.186
DoubleSync, 1.680, 3.497, 3.451, 3.585, 5.465, 3.627, 4.738, 3.398, 4.432, 4.030, 4.241, 4.071
SingleSync, 3.773, 3.111, 4.057, 4.551, 5.440, 6.894, 5.813, 3.276, 4.435, 5.012, 4.519, 4.691
ConcLinked, 2.982, 5.078, 3.463, 3.634, 3.658, 3.966, 4.353, 5.867, 3.362, 3.536, 3.572, 4.375
ConcArray, 3.045, 3.246, 6.018, 5.377, 5.586, 6.207,23.079, 6.781, 6.544, 6.924, 6.609,12.136


Lock Free is consistently performing faster than the alternatives when more than one thread is accessing either end. 25% faster would be about the right ballpark.

Here's the lock-free code. Note, instead of using an array, it uses a linked list (calling it an ArrayBlockingAtomQueue is a bad name - I agree):

import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class ArrayBlockingAtomQueue<T> implements BlockQueue<T>{

private static final class Node<U> {

private U val;
private AtomicReference<Node<U>> next = new AtomicReference<>();
private AtomicBoolean owned = new AtomicBoolean(false);

public Node(U val) {
super();
this.val = val;
}

@Override
public String toString() {
Node<U> n = next.get();
return String.format("Node Value %s next %s", val, n == null ? "---" : n.val);
}

}

private final AtomicReference<Node<T>> head = new AtomicReference<>();
private final AtomicReference<Node<T>> tail = new AtomicReference<>();
private final Semaphore slot;
private final Semaphore waiting;

private final Node<T> sentry = new Node<>(null);
//    private final Node<T> empty = new Node<>(null, null);

public ArrayBlockingAtomQueue(int capacity) {
slot = new Semaphore(capacity);
waiting = new Semaphore(0);
//        empty.next.set(empty);
sentry.owned.set(true);
tail.set(sentry);
}

@Override
public void add(T t) throws InterruptedException {

// space, or wait till there is.
slot.acquire();

final Node<T> node = new Node<>(t);

Node<T> back = tail.getAndSet(node);
if (!back.next.compareAndSet(null, node)) {
throw new IllegalStateException("Tail node's next was already set");
}
waiting.release();
}

@Override
public T take() throws InterruptedException {
// data, or wait till there is.
waiting.acquire();
// we know a bunch of read threads may be scanning the list.
// the head node may not yet be set with data, but we know it is coming.
Node<T> mine = null;
do {
while (h != null) {
if (h.owned.compareAndSet(false, true)) {
mine = h;
break;
}
Node<T> n = h.next.get();
if (n != null) {
}
h = h.next.get();
}
} while (mine == null);

//waiting.release();
slot.release();
T value = mine.val;
mine.val = null;
return value;
}

}


Count Down Latch option

The previous atomic/link free code has an issue with the sentry node... it should not be a field variable, as a result, it is holding a reference to the entire chain of nodes... and that's a problem for the Garbage Collector. Fixing it made it significantly slower... for a reason I am unsure of. As a result, have worked it out yet another way, using a CountDownLatch:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

@SuppressWarnings("serial")
private static final class Node<U> extends AtomicBoolean{

private U val;
private final CountDownLatch latch = new CountDownLatch(1);
private Node<U> next = null;

public Node(U val) {
this.val = val;
}

}

private final AtomicReference<Node<T>> head = new AtomicReference<>();
private final AtomicReference<Node<T>> tail = new AtomicReference<>();
private final Semaphore putSlot;

putSlot = new Semaphore(capacity);

// prime the head with a 'done' node which we can
// append to in the put methods.
Node<T> sentry = new Node<>(null);
sentry.set(true);
tail.set(sentry);
}

@Override
public void add(T t) throws InterruptedException {

final Node<T> node = new Node<>(t);
// space, or wait till there is.
putSlot.acquire();
Node<T> back = tail.getAndSet(node);
back.next = node;
back.latch.countDown();
}

@Override
public T take() throws InterruptedException {
// we know a bunch of read threads may be scanning the list.
// the head node may not yet be set with data, but we know it is coming.
try {
while (!current.compareAndSet(false, true)) {
current.latch.await();
current = current.next;
}
T ret = current.val;
current.val = null;
return ret;
} finally {
putSlot.release();
}
}

}


Running the CountDownLatch code with the previous LockFree code, and the Veritas original, I get the following results (Note they were run at a different time to other results in this post, and are not comparable with them).

MILLISECONDS,   1x1,   2x2,   3x3,   4x4,   6x6, 12x12,  1x12,   2x6,   3x4,   4x3,   6x2,  12x1
Veritas, 1.648, 3.370, 3.042, 3.334, 3.459, 3.681, 1.513, 1.758, 3.299, 3.314, 3.495, 2.944
LockFree, 1.537, 2.420, 2.916, 2.960, 3.111, 3.396, 3.013, 2.996, 2.849, 2.816, 2.938, 2.497
CDLatch, 0.969, 2.082, 2.553, 2.956, 3.465, 5.619, 5.520, 3.019, 2.820, 2.628, 2.557, 2.107


Conclusion

The bottom line is that the double-ended synchronization is faster than the double-ended locking.... as I would expect. Faster yet is a lock-free system that's not using an array at all, but that may be a 'cheat'. Still, compared to the native Java library offerings, both the array-based, and link-based options can be beaten... handily.

• I agree with the redundancy point i too don't like it. but i try to code in a similar fashion to LBQ so that people don't have problem in comparison. also see some arguments in favour of this style here stackoverflow.com/questions/28975415/… – veritas Apr 17 '15 at 17:47
• signalling the other lock only when the queue is empty or full. not on any other attempt. you code on the other hand signals every time. and consider the worst case scenario .. when the consumer is very slow so that the queue is full almost every time and producer threads are blocked. Your notifyAll wakes all the thread each eating more cpu. – veritas Apr 17 '15 at 18:36
• think of this scenario one producer Thread and 10 consumer threads. A single producer thread had fight with other 10 threads to get the lock. Whereas in my code producer thread isn't contented at all except when the queue is empty. And Empty queue is always a happy scenario means you overall performance is anyways great. This tradeoff increases the throughput for the writer/producer threads. – veritas Apr 17 '15 at 18:37
• the code is not the same but the idea is please see research.ibm.com/people/m/michael/podc-1996.pdf – veritas Apr 17 '15 at 18:46
• Probably throughput of two-lock implementation will be better under high contention scenarios in comparison to single lock. But yes agreed there may be scenarios where the opposite is true – veritas Apr 17 '15 at 18:48