# Closable BlockingQueue

I am working on legacy code, specifically a sort of BoundedBlockingQueue (mainly used as a pipe between different threads). As it is heavily used in the system and the current implementation features fully synchronized methods and a wait/notify mechanism, I attempted to rewrite it, using the java 5 concurrency utilities. Below is my result, that is considerably faster in (naive) testing and I haven't hit obvious threading issues (yet... (: ).

As this is legacy code I cannot simply switch to a BlockingQueue implementation, but must support blocking read, write and peek methods. An additional complication is that closing the pipe is required, i.e. the writer or reader may decide to close it. The reader should then be able to empty the pipe, while the writer should not write more.

I would appreciate any constructive critique, especially regarding the correctness of my approach and hints at optimizations.


public class ConcurrentBufferedPipe implements Pipe {
/** Possible states of a pipe. ERROR and CLOSED are final states. */
private enum State {
OPEN, CLOSED, ERROR;
}

/*
* Relies on the thread-safety of the used BlockingQueue, the volatile
* semantics on the state variable and state invariants of the Pipe
* Interface, namely:
*  - a closed or erroneous pipe will never be reopened
*  - as long as blocks are available, readers are permitted to continue
*      reading - even if the pipe was closed or set to error state
*  - it is acceptable that a write happens while another thread closes the
*      pipe
*  Access to the blocking queue is controlled by two semaphores, one for
*  writers and one for readers. They essentially represent the currently
*  available blocks or space.
*/

/* waiting times above this timeout are unlikely and indicate starvation */
private static final long TIMEOUT = 60;
private static final TimeUnit UNIT = TimeUnit.SECONDS;

private final String name;
private final BlockingQueue buffer;
private final int size;

/* concurrency tools */
private volatile State state;
private final Semaphore availableBlocks;
private final Semaphore availableSpace;

public ConcurrentBufferedPipe(final String name, final int size) {
super();
this.name = name;
this.size = size;
this.buffer = new LinkedBlockingQueue(size);
this.state = State.OPEN;
this.availableBlocks = new Semaphore(size);
this.availableBlocks.drainPermits();
this.availableSpace = new Semaphore(size);
}

@Override
public Object read() throws PipeIOException, PipeTerminatedException,
DataError {
aquireOrFail(this.availableBlocks);
final Object head = buffer.poll();
if (head == null) { // indicates a closed or error state
assert state != State.OPEN;
this.availableBlocks.release();
return closedMarkerOrError();
} else {
this.availableSpace.release();
}
assert head != null;
}

/**
* {@inheritDoc}
*
* @throws DataError
*             if the pipe is empty and was closed due to an error
*/
@Override
public Object peek() throws PipeIOException, PipeTerminatedException,
DataError {
aquireOrFail(this.availableBlocks);
final Object head = buffer.peek();
this.availableBlocks.release();
if (head == null) {
assert state != State.OPEN;
return closedMarkerOrError();
}
assert head != null;
}

/**
* {@inheritDoc}
*
* This implementation will also fail with a {@link PipeClosedException} if
* the pipe was closed by a writer.
*
*/
@Override
public void write(final Object block) throws PipeClosedException,
PipeIOException, PipeTerminatedException {
aquireOrFail(this.availableSpace);
boolean hasWroteBlock = false;
if (state == State.OPEN) {
hasWroteBlock = buffer.offer(block);
} else {
this.availableSpace.release();
throw new PipeClosedException();
}
this.availableBlocks.release();
assert hasWroteBlock;
}

@Override
public void closeForReading() {
state = State.CLOSED;
wakeAll();
buffer.clear();
}

@Override
public void closeForWriting() {
state = State.CLOSED;
wakeAll();
}

@Override
public void closeForWritingDueToError() {
state = State.ERROR;
wakeAll();
}

/**
* Safely tries to acquire a permission from a semaphore.
*
* @param resource
*            holds permissions
* @throws PipeTerminatedException
*             if the current thread is interrupted before or while
*             acquiring the permission or acquisition times out
*/
private void aquireOrFail(final Semaphore resource)
throws PipeTerminatedException {
try {
final boolean aquired = resource.tryAcquire(TIMEOUT, UNIT);
if (!aquired) { // indicates time out
throw new PipeTerminatedException(name);
}
} catch (final InterruptedException e) {
throw new PipeTerminatedException(name);
}
}

/**
* Depending on final state of pipe returns appropriate marker value. May
* only be called if this pipe is NOT open.
*
* @return NO_MORE_DATA marker if pipe is closed
* @throws DataError
*             if pipe is in error
*/
private Object closedMarkerOrError() throws DataError {
final State state = this.state;
if (state == State.ERROR) {
throw new DataError();
}
assert state == State.CLOSED;
return ControlBlock.NO_MORE_DATA;
}

/**
* Releases all reader / writer limits. May only be called after setting the
* pipe to a final state (ERROR or CLOSED), as it ultimately corrupts the
* invariants guarded by the used semaphores.
*/
private void wakeAll() {
assert this.state != State.OPEN;
this.availableBlocks.release(size);
this.availableSpace.release(size);
}
}


Thank you for your input !

-
The present perfect of write is has written, so I'd use that instead of hasWrote. Apart from that: wouldn't this be a good opportunity to update the queue to use generics? Legacy code should still work against a generic version of your queue, if my understanding of Java generics is correct. You could then gradually remove the then-unnecessary casts from your system as you stumble across them. –  codesparkle Nov 10 '12 at 15:48
hasWrote also sounded fishy to me and I thought some time about a better name. But as English is not my native language I missed the grammatical error, thank you! I agree with you on the use of generics, but unfortunately I cannot touch the Pipe Interface (especially the write method signature). Furthermore, the usage of special markers (e.g. ControlBlock.NO_MORE_DATA) in the system prevent the use of generics here. –  Pyranja Nov 10 '12 at 17:40

1. read or write methods requires 3 synchronization operations (2 semaphore operations and one access to the LinkedBlockingQueue. If you use synchronized methods (or locking with ReentrantLock) and non-synchronized underlying queue, then only one synchronization operation is needed per read/write/peek method.

2. LinkedBlockingQueue creates additional wrapper object (link) for each item put into the queue. Use java.util.ArrayDeque(size) to avoid redundant object creation.

-
Thank you for your thoughts. It's true that there are more syncs, but they are more fine grained and should hopefully allow interleaved read/write operations, upping the throughput (e.g. LBQ uses distinct locks for putting and offering). My basic testing confirmed a speed up by roughly 30%, compared with the former fully synchronized version, under high contention. I hadn't thought about the impact of your second point, but it could certainly be an issue. I'll roll another version using the 2-condition-lock idiom with a backing ArrayDeque and compare the results. Thanks again for your insight! –  Pyranja Nov 10 '12 at 14:26
I tested the ArrayDeque version with a single ReentrantLock and separate conditions (as in ArrayBlockingQueue) and it is really fast. Much better than both of the alternatives. Thanks again for pointing me into the right direction! –  Pyranja Nov 10 '12 at 19:17
this.availableSpace.release();