From c44aae129fd07f9ce7f51f6e98da74165f3a73ef Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 24 Oct 2014 11:23:15 +0100 Subject: [PATCH] Support read/write locking. --- .../chronicle/map/ReplicatedChronicleMap.java | 38 ++++++---- .../openhft/chronicle/map/SharedSegment.java | 8 +- .../chronicle/map/VanillaChronicleMap.java | 73 +++++++++++++------ 3 files changed, 79 insertions(+), 40 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 0f5f3ad69..e421278c4 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -607,7 +607,10 @@ class ReplicatedChronicleMap, */ V acquire(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, K key, V usingValue, int hash2, boolean create, long timestamp) { - lock(); + if (create) + writeLock(); + else + readLock(); try { long keySize = metaKeyWriter.size(keyWriter, key); MultiStoreBytes entry = tmpBytes; @@ -636,7 +639,10 @@ class ReplicatedChronicleMap, } } } finally { - unlock(); + if (create) + writeUnlock(); + else + readUnlock(); } } @@ -654,7 +660,7 @@ class ReplicatedChronicleMap, */ private void remoteRemove(Bytes keyBytes, int hash2, final long timestamp, final byte identifier) { - lock(); + writeLock(); try { long keySize = keyBytes.remaining(); hashLookupLiveAndDeleted.startSearch(hash2); @@ -693,7 +699,7 @@ class ReplicatedChronicleMap, LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() + " was not " + "found"); } finally { - unlock(); + writeUnlock(); } } @@ -704,7 +710,7 @@ class ReplicatedChronicleMap, private void remotePut(@NotNull final Bytes inBytes, int hash2, final byte identifier, final long timestamp, long valuePos, long valueLimit) { - lock(); + writeLock(); try { // inBytes position and limit correspond to the key final long keySize = inBytes.remaining(); @@ -784,7 +790,7 @@ class ReplicatedChronicleMap, incrementSize(); } finally { - unlock(); + writeUnlock(); } } @@ -795,7 +801,7 @@ class ReplicatedChronicleMap, V put(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, K key, V value, int hash2, boolean replaceIfPresent, final byte identifier, final long timestamp) { - lock(); + writeLock(); try { IntIntMultiMap hashLookup = hashLookupLiveAndDeleted; long keySize = metaKeyWriter.size(keyWriter, key); @@ -850,7 +856,7 @@ class ReplicatedChronicleMap, notifyPut(offset, true, key, value, posFromOffset(offset)); return null; } finally { - unlock(); + writeUnlock(); } } @@ -961,7 +967,7 @@ class ReplicatedChronicleMap, K key, V expectedValue, int hash2, final long timestamp, final byte identifier) { assert identifier > 0; - lock(); + writeLock(); try { long keySize = metaKeyWriter.size(keyWriter, key); final IntIntMultiMap multiMap = hashLookupLiveAndDeleted; @@ -1015,7 +1021,7 @@ class ReplicatedChronicleMap, // key is not found return null; } finally { - unlock(); + writeUnlock(); } } @@ -1029,7 +1035,7 @@ class ReplicatedChronicleMap, */ V replace(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, K key, V expectedValue, V newValue, int hash2, long timestamp) { - lock(); + writeLock(); try { long keySize = metaKeyWriter.size(keyWriter, key); hashLookupLiveOnly.startSearch(hash2); @@ -1052,7 +1058,7 @@ class ReplicatedChronicleMap, // key is not found return null; } finally { - unlock(); + writeUnlock(); } } @@ -1068,7 +1074,7 @@ class ReplicatedChronicleMap, public void dirtyEntries(final long timeStamp, final ModificationIterator.EntryModifiableCallback callback) { - this.lock(); + this.readLock(); try { final int index = Segment.this.getIndex(); hashLookupLiveAndDeleted.forEach(new IntIntMultiMap.EntryConsumer() { @@ -1088,7 +1094,7 @@ class ReplicatedChronicleMap, }); } finally { - unlock(); + readUnlock(); } } @@ -1452,7 +1458,7 @@ class ReplicatedChronicleMap, this.position = position; final SharedSegment segment = segment((int) (position >>> segmentIndexShift)); - segment.lock(); + segment.readLock(); try { if (changes.clearIfSet(position)) { @@ -1473,7 +1479,7 @@ class ReplicatedChronicleMap, // while we were trying to obtain segment lock (for example, in onReplication()), // go to pick up next (next iteration in while (true) loop) } finally { - segment.unlock(); + segment.readUnlock(); } } } diff --git a/src/main/java/net/openhft/chronicle/map/SharedSegment.java b/src/main/java/net/openhft/chronicle/map/SharedSegment.java index 03c05d430..5e87e98ef 100644 --- a/src/main/java/net/openhft/chronicle/map/SharedSegment.java +++ b/src/main/java/net/openhft/chronicle/map/SharedSegment.java @@ -26,11 +26,15 @@ import java.util.Map; */ interface SharedSegment { - void lock(); + void readLock(); + + void writeLock(); Map.Entry getEntry(long pos); - void unlock(); + void readUnlock(); + + void writeUnlock(); NativeBytes entry(long pos); diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index e6bb3a8b1..13775f0a5 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -627,9 +627,9 @@ class VanillaChronicleMap, } - public void lock() throws IllegalStateException { + public void readLock() throws IllegalStateException { while (true) { - final boolean success = bytes.tryLockNanosLong(LOCK_OFFSET, lockTimeOutNS); + final boolean success = bytes.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS); if (success) return; if (currentThread().isInterrupted()) { throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted")); @@ -640,9 +640,30 @@ class VanillaChronicleMap, } } - public void unlock() { + public void writeLock() throws IllegalStateException { + while (true) { + final boolean success = bytes.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS); + if (success) return; + if (currentThread().isInterrupted()) { + throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted")); + } else { + errorListener.onLockTimeout(bytes.threadIdForLockLong(LOCK_OFFSET)); + bytes.resetLockLong(LOCK_OFFSET); + } + } + } + + public void readUnlock() { try { - bytes.unlockLong(LOCK_OFFSET); + bytes.unlockRWReadLock(LOCK_OFFSET); + } catch (IllegalMonitorStateException e) { + errorListener.errorOnUnlock(e); + } + } + + public void writeUnlock() { + try { + bytes.unlockRWWriteLock(LOCK_OFFSET); } catch (IllegalMonitorStateException e) { errorListener.errorOnUnlock(e); } @@ -705,7 +726,10 @@ class VanillaChronicleMap, */ V acquire(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V usingValue, int hash2, boolean create) { - lock(); + if (create) + writeLock(); + else + readLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); MultiStoreBytes entry = tmpBytes; @@ -732,7 +756,10 @@ class VanillaChronicleMap, } } } finally { - unlock(); + if (create) + writeUnlock(); + else + readUnlock(); } } @@ -788,7 +815,7 @@ class VanillaChronicleMap, V put(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V value, int hash2, boolean replaceIfPresent) { - lock(); + writeLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); @@ -813,7 +840,7 @@ class VanillaChronicleMap, notifyPut(offset, true, key, value, posFromOffset(offset)); return null; } finally { - unlock(); + writeUnlock(); } } @@ -980,7 +1007,7 @@ class VanillaChronicleMap, */ V remove(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V expectedValue, int hash2) { - lock(); + writeLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); @@ -1006,12 +1033,12 @@ class VanillaChronicleMap, // key is not found return null; } finally { - unlock(); + writeUnlock(); } } boolean containsKey(KI keyInterop, MKI metaKeyInterop, K key, int hash2) { - lock(); + readLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); IntIntMultiMap hashLookup = containsKeyHashLookup(); @@ -1023,7 +1050,7 @@ class VanillaChronicleMap, } return false; } finally { - unlock(); + readUnlock(); } } @@ -1041,7 +1068,7 @@ class VanillaChronicleMap, */ V replace(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V expectedValue, V newValue, int hash2) { - lock(); + writeLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); hashLookup.startSearch(hash2); @@ -1058,7 +1085,7 @@ class VanillaChronicleMap, // key is not found return null; } finally { - unlock(); + writeUnlock(); } } @@ -1199,14 +1226,14 @@ class VanillaChronicleMap, } void clear() { - lock(); + writeLock(); try { hashLookup.clear(); freeList.clear(); nextPosToSearchFrom = 0; resetSize(); } finally { - unlock(); + writeUnlock(); } } @@ -1229,7 +1256,7 @@ class VanillaChronicleMap, * Check there is no garbage in freeList. */ void checkConsistency() { - lock(); + readLock(); try { IntIntMultiMap hashLookup = checkConsistencyHashLookup(); for (int pos = 0; (pos = (int) freeList.nextSetBit(pos)) >= 0; ) { @@ -1250,7 +1277,7 @@ class VanillaChronicleMap, pos += entrySizeInBlocks; } } finally { - unlock(); + readUnlock(); } } @@ -1318,7 +1345,7 @@ class VanillaChronicleMap, throw new NoSuchElementException(); final Segment segment = segments[segIndex]; try { - segment.lock(); + segment.readLock(); if (segment.getHashLookup().getPositions().isClear(pos)) { // the pos was removed after the previous advance advance(segIndex, pos); @@ -1327,7 +1354,7 @@ class VanillaChronicleMap, advance(returnedSeg = segIndex, returnedPos = pos); return returnedEntry = segment.getEntry(pos); } finally { - segment.unlock(); + segment.readUnlock(); } } } @@ -1340,21 +1367,23 @@ class VanillaChronicleMap, final Segment segment = segments[segIndex]; final int pos = (int) returnedPos; try { - segment.lock(); + segment.writeLock(); if (segment.getHashLookup().getPositions().isClear(pos)) { // The case: // 1. iterator.next() - thread 1 // 2. map.put() which cause relocation of the key, returned in above - thread 2 // OR map.remove() which remove this key - thread 2 // 3. iterator.remove() - thread 1 + segment.writeUnlock(); // not re-entrant. VanillaChronicleMap.this.remove(returnedEntry.getKey()); + segment.writeLock(); } else { removePresent(segment, pos); } returnedSeg = -1; returnedEntry = null; } finally { - segment.unlock(); + segment.writeUnlock(); } }