Java Concurrency: Paired locks with shared access

Question

I am looking for a Java implementation of the following concurrency semantics. I want something similar to ReadWriteLock except symmetrical, i.e. both the read and write sides can be shared amongst many threads, but read excludes write and vice versa.

1. There are two locks, let's call them A and B.
2. Lock A is shared, i.e. there may be multiple threads holding it concurrently. Lock B is also shared, there may be multiple threads holding it concurrently.
3. If any thread holds lock A then no thread may take B – threads attempting to take B shall block until all threads holding A have released A.
4. If any thread holds lock B then no thread may take A – threads attempting to take A shall block until all threads holding B have released B.

Is there an existing library class that achieves this? At the moment I have approximated the desired functionality with a ReadWriteLock because fortunately the tasks done in the context of lock B are somewhat rarer. It feels like a hack though, and it could affect the performance of my program under heavy load.

Answer 1

Short answer:

In the standard library, there is nothing like what you need.

Long answer:

To easily implement a custom Lock you should subclass or delegate to an AbstractQueuedSynchronizer.

The following code is an example of a non-fair lock that implements what you need, including some (non exhausting) test. I called it LeftRightLock because of the binary nature of your requirements.

The concept is pretty straightforward:

AbstractQueuedSynchronizer exposes a method to atomically set the state of an int using the Compare and swap idiom ( compareAndSetState(int expect, int update) ), we can use the exposed state keep the count of the threads holding the lock, setting it to a positive value in case the Right lock is being held or a negative value in case the Left lock is being held.

Than we just make sure of the following conditions: - you can lock Left only if the state of the internal AbstractQueuedSynchronizer is zero or negative - you can lock Right only if the state of the internal AbstractQueuedSynchronizer is zero or positive

LeftRightLock.java

---

import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Lock;

/**
 * A binary mutex with the following properties:
 *
 * Exposes two different {@link Lock}s: LEFT, RIGHT.
 *
 * When LEFT is held other threads can acquire LEFT but thread trying to acquire RIGHT will be
 * blocked. When RIGHT is held other threads can acquire RIGHT but thread trying to acquire LEFT
 * will be blocked.
 */
public class LeftRightLock {

    public static final int ACQUISITION_FAILED = -1;
    public static final int ACQUISITION_SUCCEEDED = 1;

    private final LeftRightSync sync = new LeftRightSync();

    public void lockLeft() {
        sync.acquireShared(LockSide.LEFT.getV());
    }

    public void lockRight() {
        sync.acquireShared(LockSide.RIGHT.getV());
    }

    public void releaseLeft() {
        sync.releaseShared(LockSide.LEFT.getV());
    }

    public void releaseRight() {
        sync.releaseShared(LockSide.RIGHT.getV());
    }

    public boolean tryLockLeft() {
        return sync.tryAcquireShared(LockSide.LEFT) == ACQUISITION_SUCCEEDED;
    }

    public boolean tryLockRight() {
        return sync.tryAcquireShared(LockSide.RIGHT) == ACQUISITION_SUCCEEDED;
    }

    private enum LockSide {
        LEFT(-1), NONE(0), RIGHT(1);

        private final int v;

        LockSide(int v) {
            this.v = v;
        }

        public int getV() {
            return v;
        }
    }

    /**
     * <p>
     * Keep count the count of threads holding either the LEFT or the RIGHT lock.
     * </p>
     *
     * <li>A state ({@link AbstractQueuedSynchronizer#getState()}) greater than 0 means one or more threads are holding RIGHT lock. </li>
     * <li>A state ({@link AbstractQueuedSynchronizer#getState()}) lower than 0 means one or more threads are holding LEFT lock.</li>
     * <li>A state ({@link AbstractQueuedSynchronizer#getState()}) equal to zero means no thread is holding any lock.</li>
     */
    private static final class LeftRightSync extends AbstractQueuedSynchronizer {


        @Override
        protected int tryAcquireShared(int requiredSide) {
            return (tryChangeThreadCountHoldingCurrentLock(requiredSide, ChangeType.ADD) ? ACQUISITION_SUCCEEDED : ACQUISITION_FAILED);
        }    

        @Override
        protected boolean tryReleaseShared(int requiredSide) {
            return tryChangeThreadCountHoldingCurrentLock(requiredSide, ChangeType.REMOVE);
        }

        public boolean tryChangeThreadCountHoldingCurrentLock(int requiredSide, ChangeType changeType) {
            if (requiredSide != 1 && requiredSide != -1)
                throw new AssertionError("You can either lock LEFT or RIGHT (-1 or +1)");

            int curState;
            int newState;
            do {
                curState = this.getState();
                if (!sameSide(curState, requiredSide)) {
                    return false;
                }

                if (changeType == ChangeType.ADD) {
                    newState = curState + requiredSide;
                } else {
                    newState = curState - requiredSide;
                }
                //TODO: protect against int overflow (hopefully you won't have so many threads)
            } while (!this.compareAndSetState(curState, newState));
            return true;
        }    

        final int tryAcquireShared(LockSide lockSide) {
            return this.tryAcquireShared(lockSide.getV());
        }

        final boolean tryReleaseShared(LockSide lockSide) {
            return this.tryReleaseShared(lockSide.getV());
        }

        private boolean sameSide(int curState, int requiredSide) {
            return curState == 0 || sameSign(curState, requiredSide);
        }

        private boolean sameSign(int a, int b) {
            return (a >= 0) ^ (b < 0);
        }

        public enum ChangeType {
            ADD, REMOVE
        }
    }
}

LeftRightLockTest.java

---

import org.junit.Test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

public class LeftRightLockTest {


    int logLineSequenceNumber = 0;
    private LeftRightLock sut = new LeftRightLock();

    @Test(timeout = 2000)
    public void acquiringLeftLockExcludeAcquiringRightLock() throws Exception {
        sut.lockLeft();


        Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> sut.tryLockRight());
        assertFalse("I shouldn't be able to acquire the RIGHT lock!", task.get());
    }

    @Test(timeout = 2000)
    public void acquiringRightLockExcludeAcquiringLeftLock() throws Exception {
        sut.lockRight();
        Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> sut.tryLockLeft());
        assertFalse("I shouldn't be able to acquire the LEFT lock!", task.get());
    }

    @Test(timeout = 2000)
    public void theLockShouldBeReentrant() throws Exception {
        sut.lockLeft();
        assertTrue(sut.tryLockLeft());
    }

    @Test(timeout = 2000)
    public void multipleThreadShouldBeAbleToAcquireTheSameLock_Right() throws Exception {
        sut.lockRight();
        Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> sut.tryLockRight());
        assertTrue(task.get());
    }

    @Test(timeout = 2000)
    public void multipleThreadShouldBeAbleToAcquireTheSameLock_left() throws Exception {
        sut.lockLeft();
        Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> sut.tryLockLeft());
        assertTrue(task.get());
    }

    @Test(timeout = 2000)
    public void shouldKeepCountOfAllTheThreadsHoldingTheSide() throws Exception {

        CountDownLatch latchA = new CountDownLatch(1);
        CountDownLatch latchB = new CountDownLatch(1);


        Thread threadA = spawnThreadToAcquireLeftLock(latchA, sut);
        Thread threadB = spawnThreadToAcquireLeftLock(latchB, sut);

        System.out.println("Both threads have acquired the left lock.");

        try {
            latchA.countDown();
            threadA.join();
            boolean acqStatus = sut.tryLockRight();
            System.out.println("The right lock was " + (acqStatus ? "" : "not") + " acquired");
            assertFalse("There is still a thread holding the left lock. This shouldn't succeed.", acqStatus);
        } finally {
            latchB.countDown();
            threadB.join();
        }

    }

    @Test(timeout = 5000)
    public void shouldBlockThreadsTryingToAcquireLeftIfRightIsHeld() throws Exception {
        sut.lockLeft();

        CountDownLatch taskStartedLatch = new CountDownLatch(1);

        final Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> {
            taskStartedLatch.countDown();
            sut.lockRight();
            return false;
        });

        taskStartedLatch.await();
        Thread.sleep(100);

        assertFalse(task.isDone());
    }

    @Test
    public void shouldBeFreeAfterRelease() throws Exception {
        sut.lockLeft();
        sut.releaseLeft();
        assertTrue(sut.tryLockRight());
    }

    @Test
    public void shouldBeFreeAfterMultipleThreadsReleaseIt() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);

        final Thread thread1 = spawnThreadToAcquireLeftLock(latch, sut);
        final Thread thread2 = spawnThreadToAcquireLeftLock(latch, sut);

        latch.countDown();

        thread1.join();
        thread2.join();

        assertTrue(sut.tryLockRight());

    }

    @Test(timeout = 2000)
    public void lockShouldBeReleasedIfNoThreadIsHoldingIt() throws Exception {
        CountDownLatch releaseLeftLatch = new CountDownLatch(1);
        CountDownLatch rightLockTaskIsRunning = new CountDownLatch(1);

        Thread leftLockThread1 = spawnThreadToAcquireLeftLock(releaseLeftLatch, sut);
        Thread leftLockThread2 = spawnThreadToAcquireLeftLock(releaseLeftLatch, sut);

        Future<Boolean> acquireRightLockTask = Executors.newSingleThreadExecutor().submit(() -> {
            if (sut.tryLockRight())
                throw new AssertionError("The left lock should be still held, I shouldn't be able to acquire right a this point.");
            printSynchronously("Going to be blocked on right lock");
            rightLockTaskIsRunning.countDown();
            sut.lockRight();
            printSynchronously("Lock acquired!");
            return true;
        });

        rightLockTaskIsRunning.await();

        releaseLeftLatch.countDown();
        leftLockThread1.join();
        leftLockThread2.join();

        assertTrue(acquireRightLockTask.get());
    }

    private synchronized void printSynchronously(String str) {

        System.out.println(logLineSequenceNumber++ + ")" + str);
        System.out.flush();
    }

    private Thread spawnThreadToAcquireLeftLock(CountDownLatch releaseLockLatch, LeftRightLock lock) throws InterruptedException {
        CountDownLatch lockAcquiredLatch = new CountDownLatch(1);
        final Thread thread = spawnThreadToAcquireLeftLock(releaseLockLatch, lockAcquiredLatch, lock);
        lockAcquiredLatch.await();
        return thread;
    }

    private Thread spawnThreadToAcquireLeftLock(CountDownLatch releaseLockLatch, CountDownLatch lockAcquiredLatch, LeftRightLock lock) {
        final Thread thread = new Thread(() -> {
            lock.lockLeft();
            printSynchronously("Thread " + Thread.currentThread() + " Acquired left lock");
            try {
                lockAcquiredLatch.countDown();
                releaseLockLatch.await();
            } catch (InterruptedException ignore) {
            } finally {
                lock.releaseLeft();
            }

            printSynchronously("Thread " + Thread.currentThread() + " RELEASED left lock");
        });
        thread.start();
        return thread;
    }
}

Answer 2

I don't know any library that does that you want. Even if there is such a library it possess little value because every time your request changes the library stops doing the magic.

The actual question here is "How to I implement my own lock with custom specification?"

Java provides tool for that named AbstractQueuedSynchronizer. It has extensive documentation. Apart from docs one would possibly like to look at CountDownLatch and ReentrantLock sources and use them as examples.

For your particular request see code below, but beware that it is 1) not fair 2) not tested

public class MultiReadWriteLock implements ReadWriteLock {

    private final Sync sync;
    private final Lock readLock;
    private final Lock writeLock;

    public MultiReadWriteLock() {
        this.sync = new Sync();
        this.readLock = new MultiLock(Sync.READ, sync);
        this.writeLock = new MultiLock(Sync.WRITE, sync);
    }

    @Override
    public Lock readLock() {
        return readLock;
    }

    @Override
    public Lock writeLock() {
        return writeLock;
    }

    private static final class Sync extends AbstractQueuedSynchronizer {

        private static final int READ = 1;
        private static final int WRITE = -1;

        @Override
        public int tryAcquireShared(int arg) {
            int state, result;
            do {
                state = getState();
                if (state >= 0 && arg == READ) {
                    // new read
                    result = state + 1;
                } else if (state <= 0 && arg == WRITE) {
                    // new write
                    result = state - 1;
                } else {
                    // blocked
                    return -1;
                }
            } while (!compareAndSetState(state, result));
            return 1;
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
            int state, result;
            do {
                state = getState();
                if (state == 0) {
                    return false;
                }
                if (state > 0 && arg == READ) {
                    result = state - 1;
                } else if (state < 0 && arg == WRITE) {
                    result = state + 1;
                } else {
                    throw new IllegalMonitorStateException();
                }
            } while (!compareAndSetState(state, result));
            return result == 0;
        }
    }

    private static class MultiLock implements Lock {

        private final int parameter;
        private final Sync sync;

        public MultiLock(int parameter, Sync sync) {
            this.parameter = parameter;
            this.sync = sync;
        }

        @Override
        public void lock() {
            sync.acquireShared(parameter);
        }

        @Override
        public void lockInterruptibly() throws InterruptedException {
            sync.acquireSharedInterruptibly(parameter);
        }

        @Override
        public boolean tryLock() {
            return sync.tryAcquireShared(parameter) > 0;
        }

        @Override
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return sync.tryAcquireSharedNanos(parameter, unit.toNanos(time));
        }

        @Override
        public void unlock() {
            sync.releaseShared(parameter);
        }

        @Override
        public Condition newCondition() {
            throw new UnsupportedOperationException(
                "Conditions are unsupported as there are no exclusive access"
            );
        }
    }
}