From 4907af142be8cd1588537ff54bfe5318a254598a Mon Sep 17 00:00:00 2001 From: leventov Date: Fri, 11 Dec 2015 18:40:21 +0300 Subject: [PATCH] Prefix Read/Update/WriteLock methods with checkOnEachLockOperation() --- .../hash/impl/stage/entry/ReadLock.java | 10 +++++++++- .../hash/impl/stage/entry/SegmentStages.java | 18 ++++++------------ .../hash/impl/stage/entry/UpdateLock.java | 18 ++++++++++-------- .../hash/impl/stage/entry/WriteLock.java | 18 +++++++++++------- .../stage/hash/CheckOnEachPublicOperation.java | 14 ++++++++++++++ .../chronicle/hash/locks/InterProcessLock.java | 4 +++- 6 files changed, 53 insertions(+), 29 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java index 7a0077682..e7b498a26 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/ReadLock.java @@ -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(); diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java index 42acc754c..68c99bf5b 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java @@ -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"; } } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java index bca40d3b6..f5484c17e 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/UpdateLock.java @@ -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(); diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/WriteLock.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/WriteLock.java index 63545ce71..364815d1a 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/WriteLock.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/WriteLock.java @@ -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); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java index 5a7e56a88..8f5d9341f 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java @@ -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()"); diff --git a/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java index 2be9de5a4..3b56a79b1 100644 --- a/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java +++ b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java @@ -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 */