The Clustered Lock

For a while now I’ve been inspecting the Clustered Lock in Infinispan. Here are some thoughts.

The Normal Lock

The interface of java’s lock looks like this:

lock(): void
lockInterrubtibly(): void
tryLock(): boolean
tryLock(long, TimeUnit): boolean
unlock(): void
newCondition(): boolean

Of note here is that you have lock() which will wait forever and tryLock which tries to lock either directly and succeeds or fails, or tries for a given time period.

The Clustered Lock

Now we have the interface for the Infinispan Clustered Locks:

lock(): CompletableFuture<Void>
tryLock(): CompletableFuture<Boolean>
tryLock(long, TimeUnit): CompletableFuture<Boolean>
unlock(): CompletableFuture<Boolean>
isLocked(): CompletableFuture<Boolean>
isLockedByMe(): CompletableFuture<Boolean>

Notable difference here is the usage of Futures everywhere. This makes sense because doing these lock operations require a remote call to whatever machine it is that stores the lock. This might be very quick, but if there’s a lot of load in the cluster, network issues or a rebalance happening it might take a while before the request finishes.

Fundamental Idea

Now, what extra problems do we need to solve here? The developers of the infinispan has built an amazing library, and because they have strong fundamentals building the clustered locks library is not as hard as it could have been. Some fundamentals they already have are: The have a distributed Map that survives node failure. They have support of performing atomic operations on this Map. They also support executing remote functions on entries in this map. They also support listeners on when data changes.

The fundamental idea of the clustered lock is that we store who owns a lock into a Map. When you try to acquire a lock we then check if this Map’s value already has someone else owning the lock. If it does we listen to when this lock value changes, or when the node owning the lock leaves the cluster.

Components

The basic components at play here are:

• ClusteredLockImpl - The composer of the whole flow
• LockRequestHolder - Container for the Future that gives the response
• TryLockRequestHolder - Same as above, but includes handling of timeouts.
• LockFunction - This is the code that is executed on the remote machine
• RequestExpirationScheduler - This class handles timing out pending requests
• LockReleasedListener - Clustered listener that sees when the value of a lock has changed
• ClusterChangeListener - Listens to when members leave to cluster to ensure that we unlock locks where the owner left.

Now most of the components there are operating within the scope of a specific lock key. This means for example that we have one clustered listener for each lock we define, or when rebalance listener for each lock. I think it would have been more efficient if all locks would re-use the same listener, but that’s a question of details.

The Function

Acquiring the lock looks like this:

   public Boolean apply(EntryView.ReadWriteEntryView<ClusteredLockKey, ClusteredLockValue> entryView) {
      ClusteredLockValue lock = entryView.find().orElseThrow(() -> log.lockDeleted());
      if (log.isTraceEnabled()) {
         log.tracef("LOCK[%s] lock request by reqId %s requestor %s", entryView.key().getName(), requestId, requestor);
      }
      if (lock.getState() == ClusteredLockState.RELEASED) {
         entryView.set(new ClusteredLockValue(requestId, requestor, ClusteredLockState.ACQUIRED));
         if (log.isTraceEnabled()) {
            log.tracef("LOCK[%s] lock acquired by %s %s", entryView.key().getName(), requestId, requestor);
         }
         return Boolean.TRUE;
      } else if (lock.getState() == ClusteredLockState.ACQUIRED && lock.getRequestId().equals(requestId) && lock.getOwner().equals(requestor)) {
         log.tracef("LOCK[%s] lock already acquired by %s %s", entryView.key().getName(), requestId, requestor);
         return Boolean.TRUE;
      }
      if (log.isTraceEnabled()) {
         log.tracef("LOCK[%s] lock not available, owned by %s %s", entryView.key().getName(), lock.getRequestId(), lock.getOwner());
      }
      return Boolean.FALSE;
   }

Where the stored information looks like this:

   public ClusteredLockValue(String requestId, Object owner, ClusteredLockState state) {
      this.requestId = requestId;
      this.owner = owner;
      this.state = state;
   }

Transaction

What code makes it guarantee that we have a write lock when running our lock function? Well on the creation of the lock we do the following operation:

   private ClusteredLockImpl createLock(String lockName) {
      ClusteredLockConfiguration configuration = getConfiguration(lockName);
      if (configuration == null) {
         throw new ClusteredLockException(String.format("Lock %s does not exist", lockName));
      }
      ClusteredLockKey key = new ClusteredLockKey(ByteString.fromString(lockName));
      cache().putIfAbsent(key, ClusteredLockValue.INITIAL_STATE);
      ClusteredLockImpl lock = new ClusteredLockImpl(lockName, key, cache(), this);
      return lock;
   }

here the putIfAbsent makes sure that we only add the lock if it doesn’t already exist. Here one might wonder if we couldn’t just have done putIfAbsent for lock and remove for unlock. I think that would work, but it wouldn’t have made a big difference. We would still have needed the clustered listener to see when elements are removed, but I think we could have done away with our lock function.

What guarantees that the LockFunction is run synchronously? Well it’s part of the ReadWriteMap:

   /**
    * Exposes read-write operations that can be executed against the functional map.
    * The read-write operations that can be applied per entry are exposed by
    * {@link ReadWriteEntryView}.
    *
    * <p>Read-write operations offer the possibility of writing values or
    * metadata parameters, and returning previously stored information.
    * Read-write operations are also crucial for implementing conditional,
    * compare-and-swap (CAS) like operations.
    *
    * <p>Locks are acquired before executing the read-write lambda.
    *
    * <p>Method parameters for read-write operations, including lambdas,
    * must be marshallable when running in a cluster.
    *
    * @since 8.0
    */
   @Experimental
   interface ReadWriteMap<K, V> extends FunctionalMap<K, V> {

I really like the remote function call, it’s a cool feature. I still think it could have been avoided by using simple putIfAbsent. You of course have the question of fairness. Now, if a lock is unlocked there’s no saying who will get the next lock if there are multiple instances waiting to grab the lock. It’s a bit tricky to implement because you would need to store information on who’s waiting into the cache, which would trigger entry updates etc. Having unfair locks only seems like a good choice.