Add snapshottable collection lock

src/main/java/org/warp/commonutils/locks/SnapshottableCollectionLock.java

@@ -0,0 +1,372 @@
+package org.warp.commonutils.locks;
+
+import it.cavallium.concurrentlocks.ReadWriteUpdateLock;
+import java.util.Objects;
+import java.util.Set;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Fine-grained locking on data or on the entire collection for a snapshottable collection
+ * @param <T>
+ */
+public class SnapshottableCollectionLock<T> {
+
+	private final TransitReadWriteLock fullLock;
+	private final TransitReadWriteLock dataLock;
+	private final Striped<ReadWriteUpdateLock> transactionLock;
+
+	public SnapshottableCollectionLock(int stripedSize) {
+		this.transactionLock = Striped.readWriteUpdateLock(stripedSize);
+		this.fullLock = new TransitReadWriteLock();
+		this.dataLock = new TransitReadWriteLock();
+	}
+
+	public enum DataLockType {
+		READ,
+		/**
+		 * "Update" allow reads but blocks writes and other updates.
+		 */
+		UPDATE,
+		WRITE
+	}
+
+	public enum FullLockType {
+		/** <p>This lock is a read lock for "full", nothing for "data"</p>
+		 * <p>Useful for coarse iterators</p>
+		 *
+		 * <ul>
+		 * <li>full<ul>
+		 *     <li>read: allow</li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * <li>data<ul>
+		 *     <li>read/update: allow</li>
+		 *     <li>write: allow</li>
+		 * </ul></li>
+		 * </ul>
+		 */
+		DENY_ONLY_FULL_WRITE,
+		/** <p>This lock is a read lock for "full" and a read lock for "data"</p>
+		 *
+		 * <ul>
+		 * <li>full<ul>
+		 *     <li>read: allow</li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * <li>data<ul>
+		 *     <li>read/update: allow</li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * </ul>
+		 */
+		ALLOW_ONLY_READ,
+		/** <p>This lock is a write lock for "full" and a read lock for "data"</p>
+		 *
+		 * <ul>
+		 * <li>full<ul>
+		 *     <li>read: <u>deny</u></li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * <li>data<ul>
+		 *     <li>read/update: allow</li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * </ul>
+		 */
+		ALLOW_ONLY_ITEMS_READ,
+		/** <p>This lock is a read lock for "full" and a write lock for "data"</p>
+		 *
+		 * <ul>
+		 * <li>full<ul>
+		 *     <li>read: allow</li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * <li>data<ul>
+		 *     <li>read/update: <u>deny</u></li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * </ul>
+		 */
+		ALLOW_ONLY_FULL_READ,
+		/** <p>This lock is a write lock for "full" and a write lock for "data"</p>
+		 *
+		 * <ul>
+		 * <li>full<ul>
+		 *     <li>read: <u>deny</u></li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * <li>data<ul>
+		 *     <li>read/update: <u>deny</u></li>
+		 *     <li>write: <u>deny</u></li>
+		 * </ul></li>
+		 * </ul>
+		 */
+		EXCLUSIVE;
+
+		public static final Set<FullLockType> READ_FULL_TYPES = Set.of(
+				DENY_ONLY_FULL_WRITE,
+				ALLOW_ONLY_READ,
+				ALLOW_ONLY_FULL_READ
+		);
+		public static final Set<FullLockType> WRITE_FULL_TYPES = Set.of(ALLOW_ONLY_ITEMS_READ, EXCLUSIVE);
+		public static final Set<FullLockType> READ_DATA_TYPES = Set.of(ALLOW_ONLY_READ, ALLOW_ONLY_ITEMS_READ);
+		public static final Set<FullLockType> WRITE_DATA_TYPES = Set.of(ALLOW_ONLY_FULL_READ, EXCLUSIVE);
+		public static final Set<FullLockType> NOOP_DATA_TYPES = Set.of(DENY_ONLY_FULL_WRITE);
+	}
+
+	public void fullLock(Object snapshot, FullLockType fullLockMode) {
+		if (snapshot == null) {
+			// Transit on full locks
+			switch (fullLockMode) {
+				case DENY_ONLY_FULL_WRITE:
+				case ALLOW_ONLY_READ:
+				case ALLOW_ONLY_FULL_READ:
+					fullLock.startTransitRead();
+					break;
+				case ALLOW_ONLY_ITEMS_READ:
+				case EXCLUSIVE:
+					break;
+				default:
+					throw new IllegalStateException("Unexpected value: " + fullLockMode);
+			}
+			// Manage full locks
+			switch (fullLockMode) {
+				case DENY_ONLY_FULL_WRITE:
+				case ALLOW_ONLY_READ:
+				case ALLOW_ONLY_FULL_READ:
+					fullLock.disallowTransitWrite();
+					break;
+				case ALLOW_ONLY_ITEMS_READ:
+				case EXCLUSIVE:
+					fullLock.disallowTransit();
+					break;
+				default:
+					throw new IllegalStateException("Unexpected value: " + fullLockMode);
+			}
+			// Manage data locks
+			switch (fullLockMode) {
+				case DENY_ONLY_FULL_WRITE:
+					break;
+				case ALLOW_ONLY_READ:
+				case ALLOW_ONLY_ITEMS_READ:
+					dataLock.disallowTransitWrite();
+					break;
+				case ALLOW_ONLY_FULL_READ:
+				case EXCLUSIVE:
+					dataLock.disallowTransit();
+					break;
+				default:
+					throw new IllegalStateException("Unexpected value: " + fullLockMode);
+			}
+		}
+	}
+
+	public void fullUnlock(Object snapshot, FullLockType fullLockMode) {
+		if (snapshot == null) {
+			// Manage full locks
+			switch (fullLockMode) {
+				case DENY_ONLY_FULL_WRITE:
+				case ALLOW_ONLY_READ:
+				case ALLOW_ONLY_FULL_READ:
+					fullLock.reAllowTransitWrite();
+					break;
+				case ALLOW_ONLY_ITEMS_READ:
+				case EXCLUSIVE:
+					fullLock.reAllowTransit();
+					break;
+				default:
+					throw new IllegalStateException("Unexpected value: " + fullLockMode);
+			}
+			// Transit on full locks
+			switch (fullLockMode) {
+				case DENY_ONLY_FULL_WRITE:
+				case ALLOW_ONLY_READ:
+				case ALLOW_ONLY_FULL_READ:
+					fullLock.endTransitRead();
+					break;
+				case ALLOW_ONLY_ITEMS_READ:
+				case EXCLUSIVE:
+					break;
+				default:
+					throw new IllegalStateException("Unexpected value: " + fullLockMode);
+			}
+			// Manage data locks
+			switch (fullLockMode) {
+				case DENY_ONLY_FULL_WRITE:
+					break;
+				case ALLOW_ONLY_READ:
+				case ALLOW_ONLY_ITEMS_READ:
+					dataLock.reAllowTransitWrite();
+					break;
+				case ALLOW_ONLY_FULL_READ:
+				case EXCLUSIVE:
+					dataLock.reAllowTransit();
+					break;
+				default:
+					throw new IllegalStateException("Unexpected value: " + fullLockMode);
+			}
+		}
+	}
+
+	private void partialLockKey(DataLockType dataLockMode, ReadWriteUpdateLock lock) {
+		switch (dataLockMode) {
+			case READ:
+				dataLock.startTransitRead();
+				lock.readLock().lock();
+				break;
+//todo: check carefully if an update lock can lock dataLock only for data read.
+// If it can, replace startTransitWrite with startTransitRead here.
+			case UPDATE:
+				dataLock.startTransitWrite();
+				lock.updateLock().lock();
+				break;
+			case WRITE:
+				dataLock.startTransitWrite();
+				lock.writeLock().lock();
+				break;
+		}
+	}
+
+	private void partialUnlockKey(DataLockType dataLockMode, ReadWriteUpdateLock lock) {
+		switch (dataLockMode) {
+			case READ:
+				lock.readLock().unlock();
+				dataLock.endTransitRead();
+				break;
+//todo: check carefully if an update lock can lock dataLock only for data read.
+// If it can, replace startTransitWrite with startTransitRead here.
+			case UPDATE:
+				lock.updateLock().unlock();
+				dataLock.endTransitWrite();
+				break;
+			case WRITE:
+				lock.writeLock().unlock();
+				dataLock.endTransitWrite();
+				break;
+		}
+	}
+
+	private void partialLockKeyUpgrade(DataLockType dataLockMode, ReadWriteUpdateLock lock) {
+		switch (dataLockMode) {
+			case READ:
+				lock.readLock().lock();
+				break;
+			case UPDATE:
+				lock.updateLock().lock();
+				break;
+			case WRITE:
+				lock.writeLock().lock();
+				break;
+		}
+	}
+
+	private void partialUnlockKeyUpgrade(DataLockType dataLockMode, ReadWriteUpdateLock lock) {
+		switch (dataLockMode) {
+			case READ:
+				lock.readLock().unlock();
+				break;
+			case UPDATE:
+				lock.writeLock().unlock();
+				break;
+			case WRITE:
+				lock.writeLock().unlock();
+				break;
+		}
+	}
+
+	public void dataLock(Object snapshot, DataLockType dataLockMode, @NotNull T key) {
+		if (snapshot == null) {
+			Objects.requireNonNull(key, () -> "Lock key must not be null");
+			var lock = transactionLock.get(key);
+			partialLockKey(dataLockMode, lock);
+		}
+	}
+
+	public void dataLockAt(Object snapshot, DataLockType dataLockMode, int index) {
+		if (snapshot == null) {
+			var lock = transactionLock.getAt(index);
+			partialLockKey(dataLockMode, lock);
+		}
+	}
+
+	public void dataLockUpgrade(Object snapshot, DataLockType dataLockMode, @NotNull T key) {
+		if (snapshot == null) {
+			Objects.requireNonNull(key, () -> "Lock key must not be null");
+			var lock = transactionLock.get(key);
+			partialLockKeyUpgrade(dataLockMode, lock);
+		}
+	}
+
+	public void dataLockUpgradeAt(Object snapshot, DataLockType dataLockMode, int index) {
+		if (snapshot == null) {
+			var lock = transactionLock.getAt(index);
+			partialLockKeyUpgrade(dataLockMode, lock);
+		}
+	}
+
+	public void dataUnlock(Object snapshot, DataLockType dataLockMode, @NotNull T key) {
+		if (snapshot == null) {
+			Objects.requireNonNull(key, () -> "Lock key must not be null");
+			var lock = transactionLock.get(key);
+			partialUnlockKey(dataLockMode, lock);
+		}
+	}
+
+	public void dataUnlockAt(Object snapshot, DataLockType dataLockMode, int index) {
+		if (snapshot == null) {
+			var lock = transactionLock.getAt(index);
+			partialUnlockKey(dataLockMode, lock);
+		}
+	}
+
+	public void dataUnlockUpgrade(Object snapshot, DataLockType dataLockMode, @NotNull T key) {
+		if (snapshot == null) {
+			Objects.requireNonNull(key, () -> "Lock key must not be null");
+			var lock = transactionLock.get(key);
+			partialUnlockKeyUpgrade(dataLockMode, lock);
+		}
+	}
+
+	public void dataUnlockUpgradeAt(Object snapshot, DataLockType dataLockMode, @NotNull int index) {
+		if (snapshot == null) {
+			var lock = transactionLock.getAt(index);
+			partialUnlockKeyUpgrade(dataLockMode, lock);
+		}
+	}
+
+	public void dataLockMulti(Object snapshot, DataLockType dataLockMode, @NotNull Iterable<? extends T> key) {
+		if (snapshot == null) {
+			Objects.requireNonNull(key, () -> "Lock key must not be null");
+			for (ReadWriteUpdateLock lock : transactionLock.bulkGet(key)) {
+				partialLockKey(dataLockMode, lock);
+			}
+		}
+	}
+
+	public void dataLockAtMulti(Object snapshot, DataLockType dataLockMode, @NotNull Iterable<Integer> key) {
+		if (snapshot == null) {
+			Objects.requireNonNull(key, () -> "Lock key must not be null");
+			for (ReadWriteUpdateLock lock : transactionLock.bulkGetAt(key)) {
+				partialLockKey(dataLockMode, lock);
+			}
+		}
+	}
+
+	public void dataUnlockMulti(Object snapshot, DataLockType dataLockMode, @NotNull Iterable<? extends T> key) {
+		if (snapshot == null) {
+			Objects.requireNonNull(key, () -> "Lock key must not be null");
+			for (ReadWriteUpdateLock lock : transactionLock.bulkGetInverse(key)) {
+				partialUnlockKey(dataLockMode, lock);
+			}
+		}
+	}
+
+	public void dataUnlockAtMulti(Object snapshot, DataLockType dataLockMode, @NotNull Iterable<Integer> key) {
+		if (snapshot == null) {
+			Objects.requireNonNull(key, () -> "Lock key must not be null");
+			for (ReadWriteUpdateLock lock : transactionLock.bulkGetAtInverse(key)) {
+				partialUnlockKey(dataLockMode, lock);
+			}
+		}
+	}
+}

src/main/java/org/warp/commonutils/locks/Striped.java

@@ -134,6 +134,16 @@ public abstract class Striped<L> {
 	 * #get(Object)}; may contain duplicates), in an increasing index order.
 	 */
 	public Iterable<L> bulkGet(Iterable<?> keys) {
+		return Collections.unmodifiableList(bulkGet_(keys));
+	}
+
+	public Iterable<L> bulkGetInverse(Iterable<?> keys) {
+		var list = bulkGet_(keys);
+		Collections.reverse(list);
+		return Collections.unmodifiableList(list);
+	}
+
+	private List<L> bulkGet_(Iterable<?> keys) {
 		// Initially using the array to store the keys, then reusing it to store the respective L's
 		final Object[] array = Iterables.toArray(keys, Object.class);
 		if (array.length == 0) {
@@ -174,8 +184,62 @@ public abstract class Striped<L> {
 		 * be garbage collected after locking them, ending up in a huge mess.
 		 */
 		@SuppressWarnings("unchecked") // we carefully replaced all keys with their respective L's
-				List<L> asList = (List<L>) Arrays.asList(array);
+		List<L> asList = (List<L>) Arrays.asList(array);
-		return Collections.unmodifiableList(asList);
+		return asList;
+	}
+	public Iterable<L> bulkGetAt(Iterable<Integer> keys) {
+		return Collections.unmodifiableList(bulkGetAt_(keys));
+	}
+
+	public Iterable<L> bulkGetAtInverse(Iterable<Integer> keys) {
+		var list = bulkGetAt_(keys);
+		Collections.reverse(list);
+		return Collections.unmodifiableList(list);
+	}
+
+	private List<L> bulkGetAt_(Iterable<Integer> keys) {
+		// Initially using the array to store the keys, then reusing it to store the respective L's
+		final Object[] array = Iterables.toArray(keys, Object.class);
+		if (array.length == 0) {
+			return ImmutableList.of();
+		}
+		int[] stripes = new int[array.length];
+		for (int i = 0; i < array.length; i++) {
+			stripes[i] = (int) array[i];
+		}
+		Arrays.sort(stripes);
+		// optimize for runs of identical stripes
+		int previousStripe = stripes[0];
+		array[0] = getAt(previousStripe);
+		for (int i = 1; i < array.length; i++) {
+			int currentStripe = stripes[i];
+			if (currentStripe == previousStripe) {
+				array[i] = array[i - 1];
+			} else {
+				array[i] = getAt(currentStripe);
+				previousStripe = currentStripe;
+			}
+		}
+		/*
+		 * Note that the returned Iterable holds references to the returned stripes, to avoid
+		 * error-prone code like:
+		 *
+		 * Striped<Lock> stripedLock = Striped.lazyWeakXXX(...)'
+		 * Iterable<Lock> locks = stripedLock.bulkGet(keys);
+		 * for (Lock lock : locks) {
+		 *   lock.lock();
+		 * }
+		 * operation();
+		 * for (Lock lock : locks) {
+		 *   lock.unlock();
+		 * }
+		 *
+		 * If we only held the int[] stripes, translating it on the fly to L's, the original locks might
+		 * be garbage collected after locking them, ending up in a huge mess.
+		 */
+		@SuppressWarnings("unchecked") // we carefully replaced all keys with their respective L's
+		List<L> asList = (List<L>) Arrays.asList(array);
+		return asList;
 	}
 
 	// Static factories

src/test/java/org/warp/commonutils/locks/LeftRightLockTest.java

@@ -1,4 +1,4 @@
-package org.warp.commonutils.functional.org.warp.commonutils.locks;
+package org.warp.commonutils.locks;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -8,7 +8,6 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
-import org.warp.commonutils.locks.LeftRightLock;
 
 public class LeftRightLockTest {
 

src/test/java/org/warp/commonutils/locks/SnapshottableCollectionLockTest.java

@@ -0,0 +1,129 @@
+package org.warp.commonutils.locks;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.warp.commonutils.locks.SnapshottableCollectionLock.DataLockType;
+import org.warp.commonutils.locks.SnapshottableCollectionLock.FullLockType;
+
+public class SnapshottableCollectionLockTest {
+
+	@Test
+	public void testNoop() {
+		var lock = new SnapshottableCollectionLock<Long>(50);
+		checkSituation(lock, true, true, true, true);
+	}
+
+	@Test
+	public void testDenyOnlyFullWrite() {
+		var lock = new SnapshottableCollectionLock<Long>(50);
+		lock.fullLock(null, FullLockType.DENY_ONLY_FULL_WRITE);
+		checkSituation(lock, true, false, true, true);
+		lock.fullUnlock(null, FullLockType.DENY_ONLY_FULL_WRITE);
+	}
+
+	@Test
+	public void testAllowOnlyRead() {
+		var lock = new SnapshottableCollectionLock<Long>(50);
+		lock.fullLock(null, FullLockType.ALLOW_ONLY_READ);
+		checkSituation(lock, true, false, true, false);
+		lock.fullUnlock(null, FullLockType.ALLOW_ONLY_READ);
+	}
+
+	@Test
+	public void testAllowOnlyItemsRead() {
+		var lock = new SnapshottableCollectionLock<Long>(50);
+		lock.fullLock(null, FullLockType.ALLOW_ONLY_ITEMS_READ);
+		checkSituation(lock, false, false, true, false);
+		lock.fullUnlock(null, FullLockType.ALLOW_ONLY_ITEMS_READ);
+	}
+
+	@Test
+	public void testAllowOnlyFullRead() {
+		var lock = new SnapshottableCollectionLock<Long>(50);
+		lock.fullLock(null, FullLockType.ALLOW_ONLY_FULL_READ);
+		checkSituation(lock, true, false, false, false);
+		lock.fullUnlock(null, FullLockType.ALLOW_ONLY_FULL_READ);
+	}
+
+	@Test
+	public void testExclusive() {
+		var lock = new SnapshottableCollectionLock<Long>(50);
+		lock.fullLock(null, FullLockType.EXCLUSIVE);
+		checkSituation(lock, false, false, false, false);
+		lock.fullUnlock(null, FullLockType.EXCLUSIVE);
+	}
+
+	private void checkSituation(SnapshottableCollectionLock<Long> lock, boolean fullReadAllow, boolean fullWriteAllow, boolean dataReadAllow, boolean dataWriteAllow) {
+
+		for (FullLockType readFullType : FullLockType.READ_FULL_TYPES) {
+			if (fullReadAllow) {
+				Assertions.assertDoesNotThrow(() -> {
+					CompletableFuture.runAsync(() -> {
+						lock.fullLock(null, readFullType);
+						lock.fullUnlock(null, readFullType);
+					}).get(1, TimeUnit.SECONDS);
+				});
+			} else {
+				Assertions.assertThrows(TimeoutException.class, () -> {
+					CompletableFuture.runAsync(() -> {
+						lock.fullLock(null, readFullType);
+						lock.fullUnlock(null, readFullType);
+					}).get(50, TimeUnit.MILLISECONDS);
+				});
+			}
+		}
+
+		for (FullLockType writeFullType : FullLockType.WRITE_FULL_TYPES) {
+			if (fullWriteAllow) {
+				Assertions.assertDoesNotThrow(() -> {
+					CompletableFuture.runAsync(() -> {
+						lock.fullLock(null, writeFullType);
+						lock.fullUnlock(null, writeFullType);
+					}).get(1, TimeUnit.SECONDS);
+				});
+			} else {
+				Assertions.assertThrows(TimeoutException.class, () -> {
+					CompletableFuture.runAsync(() -> {
+						lock.fullLock(null, writeFullType);
+						lock.fullUnlock(null, writeFullType);
+					}).get(50, TimeUnit.MILLISECONDS);
+				});
+			}
+		}
+
+		if (dataReadAllow) {
+			Assertions.assertDoesNotThrow(() -> {
+				CompletableFuture.runAsync(() -> {
+					lock.dataLock(null, DataLockType.READ, 1L);
+					lock.dataUnlock(null, DataLockType.READ, 1L);
+				}).get(1, TimeUnit.SECONDS);
+			});
+		} else {
+			Assertions.assertThrows(TimeoutException.class, () -> {
+				CompletableFuture.runAsync(() -> {
+					lock.dataLock(null, DataLockType.READ, 1L);
+					lock.dataUnlock(null, DataLockType.READ, 1L);
+				}).get(50, TimeUnit.MILLISECONDS);
+			});
+		}
+
+		if (dataWriteAllow) {
+			Assertions.assertDoesNotThrow(() -> {
+				CompletableFuture.runAsync(() -> {
+					lock.dataLock(null, DataLockType.WRITE, 1L);
+					lock.dataUnlock(null, DataLockType.WRITE, 1L);
+				}).get(1, TimeUnit.SECONDS);
+			});
+		} else {
+			Assertions.assertThrows(TimeoutException.class, () -> {
+				CompletableFuture.runAsync(() -> {
+					lock.dataLock(null, DataLockType.WRITE, 1L);
+					lock.dataUnlock(null, DataLockType.WRITE, 1L);
+				}).get(50, TimeUnit.MILLISECONDS);
+			});
+		}
+	}
+}