Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Prefix Read/Update/WriteLock methods with checkOnEachLockOperation()

Browse Source
This commit is contained in:
leventov
2015-12-11 18:50:01 +03:00
parent 7a5c799a28
commit 4907af142b
6 changed files with 53 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java
+9 -1
View File
@@ -16,6 +16,7 @@
package net.openhft.chronicle.hash.impl.stage.entry;
import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
import net.openhft.chronicle.hash.locks.InterProcessLock;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
@@ -29,18 +30,21 @@ import static net.openhft.chronicle.hash.impl.LocalLockState.UNLOCKED;
@Staged
public class ReadLock implements InterProcessLock {
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef SegmentStages s;
@StageRef HashEntryStages entry;
@StageRef HashLookupPos hlp;
@Override
public boolean isHeldByCurrentThread() {
checkOnEachPublicOperation.checkOnEachLockOperation();
return s.localLockState.read;
}
@Override
public void lock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
if (s.localLockState == UNLOCKED) {
if (s.readZero() && s.updateZero() && s.writeZero())
s.segmentHeader.readLock(s.segmentHeaderAddress);
@@ -51,6 +55,7 @@ public class ReadLock implements InterProcessLock {
@Override
public void lockInterruptibly() throws InterruptedException {
checkOnEachPublicOperation.checkOnEachLockOperation();
if (s.localLockState == UNLOCKED) {
if (s.readZero() && s.updateZero() && s.writeZero())
s.segmentHeader.readLockInterruptibly(s.segmentHeaderAddress);
@@ -61,6 +66,7 @@ public class ReadLock implements InterProcessLock {
@Override
public boolean tryLock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
if (s.localLockState == UNLOCKED) {
if (!s.readZero() || !s.updateZero() || !s.writeZero() ||
s.segmentHeader.tryReadLock(s.segmentHeaderAddress)) {
@@ -77,6 +83,7 @@ public class ReadLock implements InterProcessLock {
@Override
public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
checkOnEachPublicOperation.checkOnEachLockOperation();
if (s.localLockState == UNLOCKED) {
if (!s.readZero() || !s.updateZero() || !s.writeZero() ||
s.segmentHeader.tryReadLock(s.segmentHeaderAddress, time, unit)) {
@@ -93,6 +100,7 @@ public class ReadLock implements InterProcessLock {
@Override
public void unlock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
// TODO what should close here?
hlp.closeHashLookupPos();
entry.closePos();
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
+6 -12
View File
@@ -181,6 +181,7 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
@Override
@Stage("Locks")
public int changeAndGetTotalReadLockCount(int change) {
assert totalReadLockCount + change >= 0 : "read underflow";
return totalReadLockCount += change;
}
@@ -192,6 +193,7 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
@Override
@Stage("Locks")
public int changeAndGetTotalUpdateLockCount(int change) {
assert totalUpdateLockCount + change >= 0 : "update underflow";
return totalUpdateLockCount += change;
}
@@ -203,6 +205,7 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
@Override
@Stage("Locks")
public int changeAndGetTotalWriteLockCount(int change) {
assert totalWriteLockCount + change >= 0 : "write underflow";
return totalWriteLockCount += change;
}
@@ -301,17 +304,13 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
case UNLOCKED:
return;
case READ_LOCKED:
int newTotalReadLockCount = decrementRead();
if (newTotalReadLockCount == 0) {
if (decrementRead() == 0) {
if (updateZero() && writeZero())
segmentHeader.readUnlock(segmentHeaderAddress);
} else {
assert newTotalReadLockCount > 0 : "read underflow";
}
return;
case UPDATE_LOCKED:
int newTotalUpdateLockCount = decrementUpdate();
if (newTotalUpdateLockCount == 0) {
if (decrementUpdate() == 0) {
if (writeZero()) {
if (readZero()) {
segmentHeader.updateUnlock(segmentHeaderAddress);
@@ -319,13 +318,10 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
segmentHeader.downgradeUpdateToReadLock(segmentHeaderAddress);
}
}
} else {
assert newTotalUpdateLockCount > 0 : "update underflow";
}
return;
case WRITE_LOCKED:
int newTotalWriteLockCount = decrementWrite();
if (newTotalWriteLockCount == 0) {
if (decrementWrite() == 0) {
if (!updateZero()) {
segmentHeader.downgradeWriteToUpdateLock(segmentHeaderAddress);
} else {
@@ -335,8 +331,6 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
segmentHeader.writeUnlock(segmentHeaderAddress);
}
}
} else {
assert newTotalWriteLockCount > 0 : "write underflow";
}
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java
+10 -8
View File
@@ -16,6 +16,7 @@
package net.openhft.chronicle.hash.impl.stage.entry;
import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
import net.openhft.chronicle.hash.impl.stage.hash.LogHolder;
import net.openhft.chronicle.hash.locks.InterProcessLock;
import net.openhft.sg.StageRef;
@@ -30,17 +31,20 @@ import static net.openhft.chronicle.hash.impl.LocalLockState.UPDATE_LOCKED;
@Staged
public class UpdateLock implements InterProcessLock {
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef LogHolder logHolder;
@StageRef SegmentStages s;
@StageRef HashEntryStages<?> entry;
@Override
public boolean isHeldByCurrentThread() {
checkOnEachPublicOperation.checkOnEachLockOperation();
return s.localLockState.update;
}
@Override
public void lock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
s.checkIterationContextNotLockedInThisThread();
@@ -85,6 +89,7 @@ public class UpdateLock implements InterProcessLock {
@Override
public void lockInterruptibly() throws InterruptedException {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
s.checkIterationContextNotLockedInThisThread();
@@ -106,6 +111,7 @@ public class UpdateLock implements InterProcessLock {
@Override
public boolean tryLock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
s.checkIterationContextNotLockedInThisThread();
@@ -135,6 +141,7 @@ public class UpdateLock implements InterProcessLock {
@Override
public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
s.checkIterationContextNotLockedInThisThread();
@@ -164,30 +171,25 @@ public class UpdateLock implements InterProcessLock {
@Override
public void unlock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
entry.closeDelayedUpdateChecksum();
switch (s.localLockState) {
case UNLOCKED:
case READ_LOCKED:
return;
case UPDATE_LOCKED:
int newTotalUpdateLockCount = s.decrementUpdate();
if (newTotalUpdateLockCount == 0) {
if (s.decrementUpdate() == 0) {
if (s.writeZero())
s.segmentHeader.downgradeUpdateToReadLock(s.segmentHeaderAddress);
} else {
assert newTotalUpdateLockCount > 0 : "update underflow";
}
break;
case WRITE_LOCKED:
int newTotalWriteLockCount = s.decrementWrite();
if (newTotalWriteLockCount == 0) {
if (s.decrementWrite() == 0) {
if (!s.updateZero()) {
s.segmentHeader.downgradeWriteToUpdateLock(s.segmentHeaderAddress);
} else {
s.segmentHeader.downgradeWriteToReadLock(s.segmentHeaderAddress);
}
} else {
assert newTotalWriteLockCount > 0 : "write underflow";
}
}
s.incrementRead();
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/WriteLock.java
+11 -7
View File
@@ -16,6 +16,7 @@
package net.openhft.chronicle.hash.impl.stage.entry;
import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
import net.openhft.chronicle.hash.locks.InterProcessLock;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
@@ -29,15 +30,18 @@ import static net.openhft.chronicle.hash.impl.LocalLockState.WRITE_LOCKED;
@Staged
public class WriteLock implements InterProcessLock {
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef SegmentStages s;
@Override
public boolean isHeldByCurrentThread() {
checkOnEachPublicOperation.checkOnEachLockOperation();
return s.localLockState.write;
}
@Override
public void lock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
s.checkIterationContextNotLockedInThisThread();
@@ -88,6 +92,7 @@ public class WriteLock implements InterProcessLock {
@Override
public void lockInterruptibly() throws InterruptedException {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
s.checkIterationContextNotLockedInThisThread();
@@ -121,6 +126,7 @@ public class WriteLock implements InterProcessLock {
@Override
public boolean tryLock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
s.checkIterationContextNotLockedInThisThread();
@@ -176,6 +182,7 @@ public class WriteLock implements InterProcessLock {
@Override
public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
s.checkIterationContextNotLockedInThisThread();
@@ -233,20 +240,17 @@ public class WriteLock implements InterProcessLock {
@Override
public void unlock() {
checkOnEachPublicOperation.checkOnEachLockOperation();
switch (s.localLockState) {
case UNLOCKED:
case READ_LOCKED:
case UPDATE_LOCKED:
return;
case WRITE_LOCKED:
int newTotalWriteLockCount = s.decrementWrite();
if (newTotalWriteLockCount == 0) {
if (s.decrementWrite() == 0)
s.segmentHeader.downgradeWriteToUpdateLock(s.segmentHeaderAddress);
} else {
assert newTotalWriteLockCount > 0 : "write underflow";
}
s.incrementUpdate();
s.setLocalLockState(UPDATE_LOCKED);
}
s.incrementUpdate();
s.setLocalLockState(UPDATE_LOCKED);
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java
+14
View File
@@ -17,6 +17,10 @@
package net.openhft.chronicle.hash.impl.stage.hash;
import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder;
import net.openhft.chronicle.hash.impl.stage.entry.ReadLock;
import net.openhft.chronicle.hash.impl.stage.entry.UpdateLock;
import net.openhft.chronicle.hash.impl.stage.entry.WriteLock;
import net.openhft.chronicle.map.impl.stage.query.QueryCheckOnEachPublicOperation;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
@@ -27,6 +31,16 @@ public abstract class CheckOnEachPublicOperation {
@StageRef VanillaChronicleHashHolder<?> hh;
public void checkOnEachPublicOperation() {
checkOnEachLockOperation();
}
/**
* This method call prefix methods in {@link ReadLock}, {@link UpdateLock}, {@link WriteLock}.
* Distinction from {@link #checkOnEachPublicOperation()} is needed because {@link
* QueryCheckOnEachPublicOperation#checkOnEachPublicOperation()} depends on some stages that
* depend on locking methods, closing a dependency cycle.
*/
public void checkOnEachLockOperation() {
holder.checkAccessingFromOwnerThread();
if (!hh.h().isOpen())
throw new IllegalStateException("Access to Chronicle Hash after close()");
src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java
+3 -1
View File
@@ -110,7 +110,9 @@ public interface InterProcessLock extends Lock {
boolean tryLock();
/**
* Releases the lock, if the lock is not held by the current thread, returns immediately.
* Releases the lock (and all stronger-level lock, in the context of {@link
* InterProcessReadWriteUpdateLock}, if the lock is not held by the current thread, returns
* immediately.
*
* @throws IllegalMonitorStateException if this method call observes illegal lock state
*/