Support read/write locking.

2021-09-13 01:26:12 +00:00 · 2014-10-24 11:23:15 +01:00
parent 3a3c634a19
commit c44aae129f
3 changed files with 79 additions and 40 deletions
@@ -607,7 +607,10 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         */
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                    }
                }
            } finally {
-                unlock();
+                if (create)
+                    writeUnlock();
+                else
+                    readUnlock();
            }
        }

@@ -654,7 +660,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         */
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                    LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() + " was not " +
                            "found");
            } finally {
-                unlock();
+                writeUnlock();
            }
        }

@@ -704,7 +710,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                incrementSize();

            } finally {
-                unlock();
+                writeUnlock();
            }
        }

@@ -795,7 +801,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                notifyPut(offset, true, key, value, posFromOffset(offset));
                return null;
            } finally {
-                unlock();
+                writeUnlock();
            }
        }

@@ -961,7 +967,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
                 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                // key is not found
                return null;
            } finally {
-                unlock();
+                writeUnlock();
            }
        }

@@ -1029,7 +1035,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         */
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                // key is not found
                return null;
            } finally {
-                unlock();
+                writeUnlock();
            }
        }

@@ -1068,7 +1074,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                });

            } finally {
-                unlock();
+                readUnlock();
            }
        }

@@ -1452,7 +1458,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,

                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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                    // 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();
                }
            }
        }
@@ -26,11 +26,15 @@ import java.util.Map;
 */
 interface SharedSegment<K, V> {

-    void lock();
+    void readLock();
+
+    void writeLock();

    Map.Entry<K, V> getEntry(long pos);

-    void unlock();
+    void readUnlock();
+
+    void writeUnlock();

    NativeBytes entry(long pos);

@@ -627,9 +627,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        }


-        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
            }
        }

-        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
         */
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                    }
                }
            } finally {
-                unlock();
+                if (create)
+                    writeUnlock();
+                else
+                    readUnlock();
            }
        }

@@ -788,7 +815,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,

        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                notifyPut(offset, true, key, value, posFromOffset(offset));
                return null;
            } finally {
-                unlock();
+                writeUnlock();
            }
        }

@@ -980,7 +1007,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         */
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                // 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                }
                return false;
            } finally {
-                unlock();
+                readUnlock();
            }
        }

@@ -1041,7 +1068,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         */
        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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                // key is not found
                return null;
            } finally {
-                unlock();
+                writeUnlock();
            }
        }

@@ -1199,14 +1226,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        }

        void clear() {
-            lock();
+            writeLock();
            try {
                hashLookup.clear();
                freeList.clear();
                nextPosToSearchFrom = 0;
                resetSize();
            } finally {
-                unlock();
+                writeUnlock();
            }
        }

@@ -1229,7 +1256,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         * 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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                    pos += entrySizeInBlocks;
                }
            } finally {
-                unlock();
+                readUnlock();
            }
        }

@@ -1318,7 +1345,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
                    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<K, KI, MKI extends MetaBytesInterop<K, KI>,
                    advance(returnedSeg = segIndex, returnedPos = pos);
                    return returnedEntry = segment.getEntry(pos);
                } finally {
-                    segment.unlock();
+                    segment.readUnlock();
                }
            }
        }
@@ -1340,21 +1367,23 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
            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();
            }
        }