@Test(timeout = 10000) public void lockAcrossSessions() throws Exception { BlockLockManager manager = new BlockLockManager(); manager.lockBlock(1, TEST_BLOCK_ID, BlockLockType.READ); manager.lockBlock(2, TEST_BLOCK_ID, BlockLockType.READ); }
public long lockBlock(long sessionId, long blockId, BlockLockType blockLockType) { ClientRWLock blockLock = getBlockLock(blockId); Lock lock; if (blockLockType == BlockLockType.READ) { lock = blockLock.readLock(); } else { if (sessionHoldsLock(sessionId, blockId)) { throw new IllegalStateException(String .format("Session %s attempted to take a write lock on block %s, but the session already" + " holds a lock on the block", sessionId, blockId)); } lock = blockLock.writeLock(); } lock.lock(); try { long lockId = LOCK_ID_GEN.getAndIncrement(); try (LockResource r = new LockResource(mSharedMapsLock.writeLock())) { mLockIdToRecordMap.put(lockId, new LockRecord(sessionId, blockId, lock)); Set<Long> sessionLockIds = mSessionIdToLockIdsMap.get(sessionId); if (sessionLockIds == null) { mSessionIdToLockIdsMap.put(sessionId, Sets.newHashSet(lockId)); } else { sessionLockIds.add(lockId); } } return lockId; } catch (RuntimeException e) { unlock(lock, blockId); throw e; } }
BlockLockManager { public long lockBlock(long sessionId, long blockId, BlockLockType blockLockType) { ClientRWLock blockLock = getBlockLock(blockId); Lock lock; if (blockLockType == BlockLockType.READ) { lock = blockLock.readLock(); } else { if (sessionHoldsLock(sessionId, blockId)) { throw new IllegalStateException(String .format("Session %s attempted to take a write lock on block %s, but the session already" + " holds a lock on the block", sessionId, blockId)); } lock = blockLock.writeLock(); } lock.lock(); try { long lockId = LOCK_ID_GEN.getAndIncrement(); try (LockResource r = new LockResource(mSharedMapsLock.writeLock())) { mLockIdToRecordMap.put(lockId, new LockRecord(sessionId, blockId, lock)); Set<Long> sessionLockIds = mSessionIdToLockIdsMap.get(sessionId); if (sessionLockIds == null) { mSessionIdToLockIdsMap.put(sessionId, Sets.newHashSet(lockId)); } else { sessionLockIds.add(lockId); } } return lockId; } catch (RuntimeException e) { unlock(lock, blockId); throw e; } } }
BlockLockManager { public long lockBlock(long sessionId, long blockId, BlockLockType blockLockType) { ClientRWLock blockLock = getBlockLock(blockId); Lock lock; if (blockLockType == BlockLockType.READ) { lock = blockLock.readLock(); } else { if (sessionHoldsLock(sessionId, blockId)) { throw new IllegalStateException(String .format("Session %s attempted to take a write lock on block %s, but the session already" + " holds a lock on the block", sessionId, blockId)); } lock = blockLock.writeLock(); } lock.lock(); try { long lockId = LOCK_ID_GEN.getAndIncrement(); try (LockResource r = new LockResource(mSharedMapsLock.writeLock())) { mLockIdToRecordMap.put(lockId, new LockRecord(sessionId, blockId, lock)); Set<Long> sessionLockIds = mSessionIdToLockIdsMap.get(sessionId); if (sessionLockIds == null) { mSessionIdToLockIdsMap.put(sessionId, Sets.newHashSet(lockId)); } else { sessionLockIds.add(lockId); } } return lockId; } catch (RuntimeException e) { unlock(lock, blockId); throw e; } } BlockLockManager(); }
BlockLockManager { public long lockBlock(long sessionId, long blockId, BlockLockType blockLockType) { ClientRWLock blockLock = getBlockLock(blockId); Lock lock; if (blockLockType == BlockLockType.READ) { lock = blockLock.readLock(); } else { if (sessionHoldsLock(sessionId, blockId)) { throw new IllegalStateException(String .format("Session %s attempted to take a write lock on block %s, but the session already" + " holds a lock on the block", sessionId, blockId)); } lock = blockLock.writeLock(); } lock.lock(); try { long lockId = LOCK_ID_GEN.getAndIncrement(); try (LockResource r = new LockResource(mSharedMapsLock.writeLock())) { mLockIdToRecordMap.put(lockId, new LockRecord(sessionId, blockId, lock)); Set<Long> sessionLockIds = mSessionIdToLockIdsMap.get(sessionId); if (sessionLockIds == null) { mSessionIdToLockIdsMap.put(sessionId, Sets.newHashSet(lockId)); } else { sessionLockIds.add(lockId); } } return lockId; } catch (RuntimeException e) { unlock(lock, blockId); throw e; } } BlockLockManager(); long lockBlock(long sessionId, long blockId, BlockLockType blockLockType); boolean unlockBlockNoException(long lockId); void unlockBlock(long lockId); boolean unlockBlock(long sessionId, long blockId); void validateLock(long sessionId, long blockId, long lockId); void cleanupSession(long sessionId); Set<Long> getLockedBlocks(); void validate(); }
BlockLockManager { public long lockBlock(long sessionId, long blockId, BlockLockType blockLockType) { ClientRWLock blockLock = getBlockLock(blockId); Lock lock; if (blockLockType == BlockLockType.READ) { lock = blockLock.readLock(); } else { if (sessionHoldsLock(sessionId, blockId)) { throw new IllegalStateException(String .format("Session %s attempted to take a write lock on block %s, but the session already" + " holds a lock on the block", sessionId, blockId)); } lock = blockLock.writeLock(); } lock.lock(); try { long lockId = LOCK_ID_GEN.getAndIncrement(); try (LockResource r = new LockResource(mSharedMapsLock.writeLock())) { mLockIdToRecordMap.put(lockId, new LockRecord(sessionId, blockId, lock)); Set<Long> sessionLockIds = mSessionIdToLockIdsMap.get(sessionId); if (sessionLockIds == null) { mSessionIdToLockIdsMap.put(sessionId, Sets.newHashSet(lockId)); } else { sessionLockIds.add(lockId); } } return lockId; } catch (RuntimeException e) { unlock(lock, blockId); throw e; } } BlockLockManager(); long lockBlock(long sessionId, long blockId, BlockLockType blockLockType); boolean unlockBlockNoException(long lockId); void unlockBlock(long lockId); boolean unlockBlock(long sessionId, long blockId); void validateLock(long sessionId, long blockId, long lockId); void cleanupSession(long sessionId); Set<Long> getLockedBlocks(); void validate(); static final long INVALID_LOCK_ID; }