20

I'm trying to design a key-based locking facility: something like a normal reentrant lock, but instead of lock() and unlock(), you lock(key) and unlock(key), with the contract that no-one will be able to lock(key1) simultaneously if key.equals(key1).

Will this code work? Are there more efficients solutions? I particularly don't like the while loop while trying to put the lock in the map...

package luca;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantLock;

public class KeyedReentrantLock<K> {
    private ConcurrentMap<K, ReentrantLock> lockMap = new ConcurrentHashMap<K, ReentrantLock>();

    public void lock(K key) {
        ReentrantLock oldLock = lockMap.get(key);
        if (oldLock != null && oldLock.isHeldByCurrentThread()){
            // increase lock count and return.
            oldLock.lock();
            return;
        }
        ReentrantLock newLock = new ReentrantLock();
        newLock.lock();
        while ((oldLock = lockMap.putIfAbsent(key, newLock)) != null){
            // wait for the old lock to be released;
            oldLock.lock();
            oldLock.unlock();
        }
        return;
    }

    public void unlock(K key){
        ReentrantLock lock = lockMap.get(key);
        if (lock == null) throw new IllegalMonitorStateException("There was no lock for this key!");
        if (lock.getHoldCount() == 1){
            lockMap.remove(key);
        }
        lock.unlock();
    }

}
8
  • You can simplify your code if you don't need arbitrary keys for locks. If you could use simple array of locks instead of map, code would be much simpler, as locks will be present in array from the start. This is simple form of "lock-striping". Dec 2, 2011 at 10:32
  • Additionally, your code seems to produce new lock every time other trade locked given key. Is it as designed? Seems like lot of unnecessary work here. Dec 2, 2011 at 10:45
  • @VictorSorokin Yes, this is by design: every thread will get a new lock. The cost of instantiating a new lock seemed acceptable in this context and it's done outside of any synchronization block. Dec 2, 2011 at 11:26
  • I can't understand why your threads can't share a single lock... What are advantages of your approach? Dec 2, 2011 at 11:54
  • @VictorSorokin I want to use the keyed lock in a multiton (more specifically in stackoverflow.com/questions/8189148/… where it would help me to simplify the interface (the concrete class will need just to provide a factory method, without the separate initializer and the awkward isNullObject) and hopefully improve the liveliness). There a striped locking wouldn't help me, while obviously more useful than this somewhat complicated approach in most of the common uses. Dec 2, 2011 at 14:33

4 Answers 4

6

Why don't just use simple striped locking, like:

/**
 * Striped locks holder, contains array of {@link java.util.concurrent.locks.ReentrantLock}, on which lock/unlock
 * operations are performed. Purpose of this is to decrease lock contention.
 * <p>When client requests lock, it gives an integer argument, from which target lock is derived as follows:
 * index of lock in array equals to <code>id & (locks.length - 1)</code>.
 * Since <code>locks.length</code> is the power of 2, <code>locks.length - 1</code> is string of '1' bits,
 * and this means that all lower bits of argument are taken into account.
 * <p>Number of locks it can hold is bounded: it can be from set {2, 4, 8, 16, 32, 64}.
  */
public class StripedLock {
    private final ReentrantLock[] locks;

    /**
     * Default ctor, creates 16 locks
     */
    public StripedLock() {
        this(4);
    }

    /**
     * Creates array of locks, size of array may be any from set {2, 4, 8, 16, 32, 64} 
     * @param storagePower size of array will be equal to <code>Math.pow(2, storagePower)</code>
     */
    public StripedLock(int storagePower) {
        if (!(storagePower >= 1 && storagePower <= 6)) { throw new IllegalArgumentException("storage power must be in [1..6]"); }

        int lockSize = (int) Math.pow(2, storagePower);
        locks = new ReentrantLock[lockSize];
        for (int i = 0; i < locks.length; i++)
            locks[i] = new ReentrantLock();
    }

    /**
     * Locks lock associated with given id.
     * @param id value, from which lock is derived
     */
    public void lock(int id) {
        getLock(id).lock();
    }

    /**
     * Unlocks lock associated with given id.
     * @param id value, from which lock is derived 
     */
    public void unlock(int id) {
        getLock(id).unlock();
    }

    /**
     * Map function between integer and lock from locks array
     * @param id argument
     * @return lock which is result of function 
     */
    private ReentrantLock getLock(int id) {
        return locks[id & (locks.length - 1)];
    }
}
2
2
public class KeyLock<K> {

    private /*static*/ final Set<K> lockedKeys = new HashSet<>();

    public /*static*/ void lock(K key) throws InterruptedException {
        synchronized (lockedKeys) {
            while (!lockedKeys.add(key)) {
                lockedKeys.wait();
            }
        }
    }

    public /*static*/ void unlock(K key) {
        synchronized (lockedKeys) {
            lockedKeys.remove(key);
            lockedKeys.notifyAll();
        }
    }

}

class Test {
    private KeyLock<String> keyLock = new KeyLock<>();

    public void doSynchronouslyOnlyForEqualKeys(String key) throws InterruptedException {
        try {
            keyLock.lock(key);

            //Do what you need with your key.
            //For different keys this part is executed in parallel.
            //For equal keys this part is executed synchronously.

        } finally {
            keyLock.unlock(key);
        }
    }
}
  • Class K must correctly override methods 'equals' and 'hashCode'.
  • try-finally - is very important - you must guarantee to unlock waiting threads after your operation even if your operation threw exception.
  • This approach will not work if your back-end is distributed across multiple servers/JVMs.
1
  • I down voted this answer and realize I was wrong, now I want to take it back, however in order to do so, it says the answer needs to be edited first. I am sorry @anton-fil May 7, 2020 at 16:51
1

Called it DynamicKeyLock. It's a single process lock for any object as key (equals+hashcode` for uniqueness).

Implementation for Java 8:

public class DynamicKeyLock<T> implements Lock
{
    private final static ConcurrentHashMap<Object, LockAndCounter> locksMap = new ConcurrentHashMap<>();

    private final T key;

    public DynamicKeyLock(T lockKey)
    {
        this.key = lockKey;
    }

    private static class LockAndCounter
    {
        private final Lock lock = new ReentrantLock();
        private final AtomicInteger counter = new AtomicInteger(0);
    }

    private LockAndCounter getLock()
    {
        return locksMap.compute(key, (key, lockAndCounterInner) ->
        {
            if (lockAndCounterInner == null) {
                lockAndCounterInner = new LockAndCounter();
            }
            lockAndCounterInner.counter.incrementAndGet();
            return lockAndCounterInner;
        });
    }

    private void cleanupLock(LockAndCounter lockAndCounterOuter)
    {
        if (lockAndCounterOuter.counter.decrementAndGet() == 0)
        {
            locksMap.compute(key, (key, lockAndCounterInner) ->
            {
                if (lockAndCounterInner == null || lockAndCounterInner.counter.get() == 0) {
                    return null;
                }
                return lockAndCounterInner;
            });
        }
    }

    @Override
    public void lock()
    {
        LockAndCounter lockAndCounter = getLock();

        lockAndCounter.lock.lock();
    }

    @Override
    public void unlock()
    {
        LockAndCounter lockAndCounter = locksMap.get(key);
        lockAndCounter.lock.unlock();

        cleanupLock(lockAndCounter);
    }


    @Override
    public void lockInterruptibly() throws InterruptedException
    {
        LockAndCounter lockAndCounter = getLock();

        try
        {
            lockAndCounter.lock.lockInterruptibly();
        }
        catch (InterruptedException e)
        {
            cleanupLock(lockAndCounter);
            throw e;
        }
    }

    @Override
    public boolean tryLock()
    {
        LockAndCounter lockAndCounter = getLock();

        boolean acquired = lockAndCounter.lock.tryLock();

        if (!acquired)
        {
            cleanupLock(lockAndCounter);
        }

        return acquired;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException
    {
        LockAndCounter lockAndCounter = getLock();

        boolean acquired;
        try
        {
            acquired = lockAndCounter.lock.tryLock(time, unit);
        }
        catch (InterruptedException e)
        {
            cleanupLock(lockAndCounter);
            throw e;
        }

        if (!acquired)
        {
            cleanupLock(lockAndCounter);
        }

        return acquired;
    }

    @Override
    public Condition newCondition()
    {
        LockAndCounter lockAndCounter = locksMap.get(key);

        return lockAndCounter.lock.newCondition();
    }
}

Implementation for Java 6:

public class DynamicKeyLock<T> implements Lock
{
    private final static ConcurrentHashMap<Object, LockAndCounter> locksMap = new ConcurrentHashMap<Object, LockAndCounter>();
    private final T key;

    public DynamicKeyLock(T lockKey) {
        this.key = lockKey;
    }

    private static class LockAndCounter {
        private final Lock lock = new ReentrantLock();
        private final AtomicInteger counter = new AtomicInteger(0);
    }

    private LockAndCounter getLock()
    {
        while (true) // Try to init lock
        {
            LockAndCounter lockAndCounter = locksMap.get(key);

            if (lockAndCounter == null)
            {
                LockAndCounter newLock = new LockAndCounter();
                lockAndCounter = locksMap.putIfAbsent(key, newLock);

                if (lockAndCounter == null)
                {
                    lockAndCounter = newLock;
                }
            }

            lockAndCounter.counter.incrementAndGet();

            synchronized (lockAndCounter)
            {
                LockAndCounter lastLockAndCounter = locksMap.get(key);
                if (lockAndCounter == lastLockAndCounter)
                {
                    return lockAndCounter;
                }
                // else some other thread beat us to it, thus try again.
            }
        }
    }

    private void cleanupLock(LockAndCounter lockAndCounter)
    {
        if (lockAndCounter.counter.decrementAndGet() == 0)
        {
            synchronized (lockAndCounter)
            {
                if (lockAndCounter.counter.get() == 0)
                {
                    locksMap.remove(key);
                }
            }
        }
    }

    @Override
    public void lock()
    {
        LockAndCounter lockAndCounter = getLock();

        lockAndCounter.lock.lock();
    }

    @Override
    public void unlock()
    {
        LockAndCounter lockAndCounter = locksMap.get(key);
        lockAndCounter.lock.unlock();

        cleanupLock(lockAndCounter);
    }


    @Override
    public void lockInterruptibly() throws InterruptedException
    {
        LockAndCounter lockAndCounter = getLock();

        try
        {
            lockAndCounter.lock.lockInterruptibly();
        }
        catch (InterruptedException e)
        {
            cleanupLock(lockAndCounter);
            throw e;
        }
    }

    @Override
    public boolean tryLock()
    {
        LockAndCounter lockAndCounter = getLock();

        boolean acquired = lockAndCounter.lock.tryLock();

        if (!acquired)
        {
            cleanupLock(lockAndCounter);
        }

        return acquired;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException
    {
        LockAndCounter lockAndCounter = getLock();

        boolean acquired;
        try
        {
            acquired = lockAndCounter.lock.tryLock(time, unit);
        }
        catch (InterruptedException e)
        {
            cleanupLock(lockAndCounter);
            throw e;
        }

        if (!acquired)
        {
            cleanupLock(lockAndCounter);
        }

        return acquired;
    }

    @Override
    public Condition newCondition()
    {
        LockAndCounter lockAndCounter = locksMap.get(key);

        return lockAndCounter.lock.newCondition();
    }
}

Simple test:

public class ProcessLockTest
{
    @Test
    public void testDifferentKeysDontLock() throws InterruptedException
    {
        ProcessLock<Object> lock = new ProcessLock<>(new Object());
        lock.lock();
        AtomicBoolean anotherThreadWasExecuted = new AtomicBoolean(false);
        try
        {
            new Thread(() ->
            {
                ProcessLock<Object> anotherLock = new ProcessLock<>(new Object());
                anotherLock.lock();
                try
                {
                    anotherThreadWasExecuted.set(true);
                }
                finally
                {
                    anotherLock.unlock();
                }
            }).start();
            Thread.sleep(100);
        }
        finally
        {
            Assert.assertTrue(anotherThreadWasExecuted.get());
            lock.unlock();
        }
    }

    @Test
    public void testSameKeysLock() throws InterruptedException
    {
        Object key = new Object();
        ProcessLock<Object> lock = new ProcessLock<>(key);
        lock.lock();
        AtomicBoolean anotherThreadWasExecuted = new AtomicBoolean(false);
        try
        {
            new Thread(() ->
            {
                ProcessLock<Object> anotherLock = new ProcessLock<>(key);
                anotherLock.lock();
                try
                {
                    anotherThreadWasExecuted.set(true);
                }
                finally
                {
                    anotherLock.unlock();
                }
            }).start();
            Thread.sleep(100);
        }
        finally
        {
            Assert.assertFalse(anotherThreadWasExecuted.get());
            lock.unlock();
        }
    }
}
0

Please refer the following sample code, I have created a new lock for every thread.

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;

public class Processor implements Runnable {

    final static ConcurrentHashMap<Integer, ReentrantReadWriteLock> CONCURRENT_HASH_MAP = new ConcurrentHashMap<Integer, ReentrantReadWriteLock>();

    final private Employee employee;

    public Processor(int id) {
        this.employee = new Employee(id);
    }

    public void run() {
        processDate(employee);
    }

    /**
     * Method to be shared
     * 
     * @param id
     */
    public void processDate(final Employee employee) {

        final int employeeId = employee.getId();
        ReentrantReadWriteLock monitoredObject = new ReentrantReadWriteLock();
        System.out.println("Before taking the lock"
                    + Thread.currentThread().getName());
        while (CONCURRENT_HASH_MAP.putIfAbsent(employeeId, monitoredObject) != null) {
        }
        ReadLock lock = monitoredObject.readLock();
        lock.lock();
        try {
            processXML(employee);
        } catch (Exception e) {
            e.printStackTrace();
        }
        CONCURRENT_HASH_MAP.remove(employeeId);
        lock.unlock();
    }

    /**
     * For similar id object this will run one by one but for different objects
     * this will run parallal.
     * 
     * This method will execute serially if called by multiple threads for
     * employee with same emp id
     * 
     * @throws Exception
     */
    public void processXML(final Employee employee) throws Exception {
        System.out.println("Process XML for " + employee.getId()
                + Thread.currentThread().getName());
        Thread.sleep(2000);
        System.out.println("Done XML Processing for " + employee.getId()
                + Thread.currentThread().getName());
        ReentrantReadWriteLock lock = CONCURRENT_HASH_MAP.get(employee.getId());
        System.out.println("lock object " + lock + "queue length "
                + lock.getQueueLength());
    }

    class Employee {
        private Integer id;

        public Employee(final int id) {
            this.id = id;
        }

        public void setId(Integer id) {
            this.id = id;
        }

        public Integer getId() {
            return id;
        }
    }

    public static void main(String[] args) {
        final ExecutorService executorService = Executors.newFixedThreadPool(10);
        long startTime = System.currentTimeMillis();
        /**
         * In Processors Constructor instead of i write 1 and see the
         * difference.
         */
        for (int i = 1; i <= 5; i++)
            executorService.submit(new Processor(i));
        executorService.shutdown();
        /*
         * Let the main thread wait till the executor service is terminated to
         * observe the total time taken
         */
        while (executorService.isTerminated() != true) {
        }
        long endTime = System.currentTimeMillis();
        long timeTaken = endTime - startTime;
        System.out.println("time taken.... " + timeTaken + " ms");

    }
}

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.