diff --git a/pom.xml b/pom.xml index 7414b52e0..35c5c7a2b 100755 --- a/pom.xml +++ b/pom.xml @@ -147,7 +147,7 @@ net.openhft stage-compiler - 1.3 + 1.4 provided diff --git a/src/main/java/net/openhft/chronicle/hash/Data.java b/src/main/java/net/openhft/chronicle/hash/Data.java index 291b59eef..d3982f328 100644 --- a/src/main/java/net/openhft/chronicle/hash/Data.java +++ b/src/main/java/net/openhft/chronicle/hash/Data.java @@ -59,7 +59,7 @@ public interface Data { return BytesUtil.bytesEqual(source, sourceOffset, bytes(), offset(), size()); } - default void writeTo(RandomDataOutput target, long targetOffset) { + default void writeTo(RandomDataOutput target, long targetOffset) { target.write(targetOffset, bytes(), offset(), size()); } 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 ca4320bcb..a0006af6a 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 @@ -28,6 +28,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.hash.locks.InterProcessLock; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; +import net.openhft.chronicle.map.impl.IterationContext; import net.openhft.lang.collection.DirectBitSet; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; @@ -258,6 +259,7 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { } else { closeNestedLocks(); } + deregisterIterationContextLockedInThisThread(); localLockState = null; rootContextOnThisSegment = null; } @@ -379,9 +381,36 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { @Stage("Locks") public void setLocalLockState(LocalLockState newState) { + boolean isLocked = localLockState != UNLOCKED && localLockState != null; + boolean goingToLock = newState != UNLOCKED && newState != null; + if (isLocked) { + if (!goingToLock) + deregisterIterationContextLockedInThisThread(); + } else if (goingToLock) { + registerIterationContextLockedInThisThread(); + } localLockState = newState; } + public void checkIterationContextNotLockedInThisThread() { + if (chaining.rootContextInThisThread.iterationContextLockedInThisThread) { + throw new IllegalStateException("Update or Write locking is forbidden in the context" + + "of locked iteration context"); + } + } + + private void registerIterationContextLockedInThisThread() { + if (this instanceof IterationContext) { + chaining.rootContextInThisThread.iterationContextLockedInThisThread = true; + } + } + + private void deregisterIterationContextLockedInThisThread() { + if (this instanceof IterationContext) { + chaining.rootContextInThisThread.iterationContextLockedInThisThread = false; + } + } + @Stage("Locks") public String debugContextsAndLocks() { String message = ""; 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 39f8e79fd..37291bd7b 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 @@ -42,6 +42,7 @@ public class UpdateLock implements InterProcessLock { public void lock() { switch (s.localLockState) { case UNLOCKED: + s.checkIterationContextNotLockedInThisThread(); if (s.updateZero() && s.writeZero()) { if (!s.readZero()) throw forbiddenUpdateLockWhenOuterContextReadLocked(); @@ -85,6 +86,7 @@ public class UpdateLock implements InterProcessLock { public void lockInterruptibly() throws InterruptedException { switch (s.localLockState) { case UNLOCKED: + s.checkIterationContextNotLockedInThisThread(); if (s.updateZero() && s.writeZero()) { if (!s.readZero()) throw forbiddenUpdateLockWhenOuterContextReadLocked(); @@ -105,6 +107,7 @@ public class UpdateLock implements InterProcessLock { public boolean tryLock() { switch (s.localLockState) { case UNLOCKED: + s.checkIterationContextNotLockedInThisThread(); if (s.updateZero() && s.writeZero()) { if (!s.readZero()) throw forbiddenUpdateLockWhenOuterContextReadLocked(); @@ -133,6 +136,7 @@ public class UpdateLock implements InterProcessLock { public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException { switch (s.localLockState) { case UNLOCKED: + s.checkIterationContextNotLockedInThisThread(); if (s.updateZero() && s.writeZero()) { if (!s.readZero()) throw forbiddenUpdateLockWhenOuterContextReadLocked(); 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 c8989c0d0..ca4ec62e3 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 @@ -40,6 +40,7 @@ public class WriteLock implements InterProcessLock { public void lock() { switch (s.localLockState) { case UNLOCKED: + s.checkIterationContextNotLockedInThisThread(); if (s.writeZero()) { if (!s.updateZero()) { s.segmentHeader.upgradeUpdateToWriteLock(s.segmentHeaderAddress); @@ -89,6 +90,7 @@ public class WriteLock implements InterProcessLock { public void lockInterruptibly() throws InterruptedException { switch (s.localLockState) { case UNLOCKED: + s.checkIterationContextNotLockedInThisThread(); if (s.writeZero()) { if (!s.updateZero()) { s.segmentHeader.upgradeUpdateToWriteLockInterruptibly( @@ -121,6 +123,7 @@ public class WriteLock implements InterProcessLock { public boolean tryLock() { switch (s.localLockState) { case UNLOCKED: + s.checkIterationContextNotLockedInThisThread(); if (s.writeZero()) { if (!s.updateZero()) { if (s.segmentHeader.tryUpgradeUpdateToWriteLock(s.segmentHeaderAddress)) { @@ -175,6 +178,7 @@ public class WriteLock implements InterProcessLock { public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException { switch (s.localLockState) { case UNLOCKED: + s.checkIterationContextNotLockedInThisThread(); if (s.writeZero()) { if (!s.updateZero()) { if (s.segmentHeader.tryUpgradeUpdateToWriteLock( diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java index 05b52137d..06aacddd4 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/Chaining.java @@ -23,21 +23,24 @@ import java.util.List; import java.util.function.Function; @Staged -public class Chaining implements ChainingInterface { +public class Chaining extends ChainingInterface { public final List contextChain; public final int indexInContextChain; - + public final ChainingInterface rootContextInThisThread; + public Chaining() { contextChain = new ArrayList<>(); contextChain.add(this); indexInContextChain = 0; + rootContextInThisThread = this; } - public Chaining(ChainingInterface root) { - contextChain = root.getContextChain(); + public Chaining(ChainingInterface rootContextInThisThread) { + contextChain = rootContextInThisThread.getContextChain(); indexInContextChain = contextChain.size(); contextChain.add(this); + this.rootContextInThisThread = rootContextInThisThread; } @Override diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ChainingInterface.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ChainingInterface.java index b94068db5..3a229df02 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ChainingInterface.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ChainingInterface.java @@ -19,14 +19,14 @@ package net.openhft.chronicle.hash.impl.stage.hash; import java.util.List; import java.util.function.Function; -public interface ChainingInterface { +public abstract class ChainingInterface extends ThreadLocalState { - List getContextChain(); + public abstract List getContextChain(); - void initUsed(boolean used); + public abstract void initUsed(boolean used); - boolean usedInit(); + public abstract boolean usedInit(); - T getContext( + public abstract T getContext( Class contextClass, Function createChaining); } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalState.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalState.java new file mode 100644 index 000000000..0478b0677 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/ThreadLocalState.java @@ -0,0 +1,21 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.hash.impl.stage.hash; + +public class ThreadLocalState { + public boolean iterationContextLockedInThisThread; +} diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 9e02bc70f..3dac111ec 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -317,7 +317,7 @@ public class VanillaChronicleMap iterationContext() { + public IterationContext iterationContext() { return i().getContext(CompiledMapIterationContext.class, ci -> new CompiledMapIterationContext<>(ci, this)); } @@ -342,7 +342,7 @@ public class VanillaChronicleMap segmentContext(int segmentIndex) { - IterationContextInterface c = iterationContext(); + IterationContext c = iterationContext(); c.initSegmentIndex(segmentIndex); return c; } diff --git a/src/main/java/net/openhft/chronicle/map/impl/IterationContextInterface.java b/src/main/java/net/openhft/chronicle/map/impl/IterationContext.java similarity index 89% rename from src/main/java/net/openhft/chronicle/map/impl/IterationContextInterface.java rename to src/main/java/net/openhft/chronicle/map/impl/IterationContext.java index 6c2d01b05..340f65720 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/IterationContextInterface.java +++ b/src/main/java/net/openhft/chronicle/map/impl/IterationContext.java @@ -19,8 +19,7 @@ package net.openhft.chronicle.map.impl; import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.MapSegmentContext; -public interface IterationContextInterface extends MapEntry, - MapSegmentContext { +public interface IterationContext extends MapEntry, MapSegmentContext { long pos(); void initSegmentIndex(int segmentIndex); diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedIterationContextInterface.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedIterationContext.java similarity index 87% rename from src/main/java/net/openhft/chronicle/map/impl/ReplicatedIterationContextInterface.java rename to src/main/java/net/openhft/chronicle/map/impl/ReplicatedIterationContext.java index e643b06fa..06362dd26 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedIterationContextInterface.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedIterationContext.java @@ -16,8 +16,7 @@ package net.openhft.chronicle.map.impl; -public interface ReplicatedIterationContextInterface - extends IterationContextInterface { +public interface ReplicatedIterationContext extends IterationContext { void readExistingEntry(long pos); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java index 9a7d9b411..0da4ebaba 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java @@ -20,7 +20,7 @@ import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.iter.HashSegmentIteration; import net.openhft.chronicle.map.MapContext; import net.openhft.chronicle.map.MapEntry; -import net.openhft.chronicle.map.impl.IterationContextInterface; +import net.openhft.chronicle.map.impl.IterationContext; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; import net.openhft.chronicle.map.impl.stage.map.WrappedValueInstanceValueHolder; import net.openhft.sg.StageRef; @@ -29,7 +29,7 @@ import org.jetbrains.annotations.NotNull; @Staged public abstract class MapSegmentIteration extends HashSegmentIteration> - implements MapEntry, IterationContextInterface { + implements MapEntry, IterationContext { @StageRef MapEntryStages entry; @StageRef WrappedValueInstanceValueHolder wrappedValueInstanceValueHolder; diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java index bfb74824b..976d1c5f7 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java @@ -23,7 +23,7 @@ import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.map.MapAbsentEntry; import net.openhft.chronicle.map.MapEntry; -import net.openhft.chronicle.map.impl.ReplicatedIterationContextInterface; +import net.openhft.chronicle.map.impl.ReplicatedIterationContext; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; @@ -35,7 +35,7 @@ import java.util.function.Predicate; @Staged public abstract class ReplicatedMapSegmentIteration extends MapSegmentIteration - implements ReplicatedIterationContextInterface, ReplicableEntry, + implements ReplicatedIterationContext, ReplicableEntry, ReplicatedHashSegmentContext>, MapAbsentEntry { @StageRef VanillaChronicleHashHolder hh; diff --git a/src/test/java/net/openhft/chronicle/map/NestedContextsInIterationContextTest.java b/src/test/java/net/openhft/chronicle/map/NestedContextsInIterationContextTest.java new file mode 100644 index 000000000..81519375e --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/NestedContextsInIterationContextTest.java @@ -0,0 +1,173 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.map; + +import org.junit.Test; + +import static org.junit.Assert.assertNull; + +public class NestedContextsInIterationContextTest { + + @Test + public void testNestedGetsInIterationContextAllowed() { + try (ChronicleMap map = ChronicleMapBuilder + .of(Integer.class, Integer.class) + .entries(100) + .create()) { + map.put(42, 42); + map.forEachEntry(e -> assertNull(map.get(0))); + } + } + + @Test(expected = IllegalStateException.class) + public void testNestedPutsInIterationContextForbidden() { + try (ChronicleMap map = ChronicleMapBuilder + .of(Integer.class, Integer.class) + .entries(100) + .create()) { + map.put(42, 42); + map.forEachEntry(e -> map.put(0, 0)); + } + } + + @Test + public void testNestedUpdatesDifferentSegmentInIterationContextForbidden() { + try (ChronicleMap map = ChronicleMapBuilder + .of(Integer.class, Integer.class) + .entries(100) + .actualSegments(2) + .create()) { + map.put(42, 42); + int keySegmentIndex = map.queryContext(42).segmentIndex(); + try (MapSegmentContext cxt = map.segmentContext(keySegmentIndex)) { + cxt.forEachSegmentEntry(e -> { + int k = 0; + ExternalMapQueryContext nestedCxt; + // try lock different segment + while ((nestedCxt = map.queryContext(k)).segmentIndex() == keySegmentIndex) { + k++; + } + try (ExternalMapQueryContext nestedC = nestedCxt) { + nestedC.updateLock().lock(); + } catch (IllegalStateException ex) { + // expected + + // try lock the same segment + while ((nestedCxt = map.queryContext(k)).segmentIndex() != + keySegmentIndex) { + k++; + } + try (ExternalMapQueryContext nestedC2 = nestedCxt) { + nestedC2.updateLock().lock(); + } catch (IllegalStateException ex2) { + // expected + return; + } + } + throw new AssertionError("expected " + IllegalStateException.class); + }); + } + } + } + + @Test + public void testNestedWritesDifferentSegmentInIterationContextForbidden() { + try (ChronicleMap map = ChronicleMapBuilder + .of(Integer.class, Integer.class) + .entries(100) + .actualSegments(2) + .create()) { + map.put(42, 42); + int keySegmentIndex = map.queryContext(42).segmentIndex(); + try (MapSegmentContext cxt = map.segmentContext(keySegmentIndex)) { + cxt.forEachSegmentEntry(e -> { + int k = 0; + ExternalMapQueryContext nestedCxt; + // try lock different segment + while ((nestedCxt = map.queryContext(k)).segmentIndex() == keySegmentIndex) { + k++; + } + try (ExternalMapQueryContext nestedC = nestedCxt) { + nestedC.writeLock().lock(); + } catch (IllegalStateException ex) { + // expected + + // try lock the same segment + while ((nestedCxt = map.queryContext(k)).segmentIndex() != + keySegmentIndex) { + k++; + } + try (ExternalMapQueryContext nestedC2 = nestedCxt) { + nestedC2.writeLock().lock(); + } catch (IllegalStateException ex2) { + // expected + return; + } + } + throw new AssertionError("expected " + IllegalStateException.class); + }); + } + } + } + + @Test + public void testNestedReadDifferentSegmentInIterationContextAllowed() { + try (ChronicleMap map = ChronicleMapBuilder + .of(Integer.class, Integer.class) + .entries(100) + .actualSegments(2) + .create()) { + map.put(42, 42); + int keySegmentIndex = map.queryContext(42).segmentIndex(); + try (MapSegmentContext cxt = map.segmentContext(keySegmentIndex)) { + cxt.forEachSegmentEntry(e -> { + int k = 0; + ExternalMapQueryContext nestedCxt; + // try lock different segment + while ((nestedCxt = map.queryContext(k)).segmentIndex() == keySegmentIndex) { + k++; + } + try (ExternalMapQueryContext nestedC = nestedCxt) { + nestedC.readLock().lock(); + } + + // try lock the same segment + while ((nestedCxt = map.queryContext(k)).segmentIndex() != + keySegmentIndex) { + k++; + } + try (ExternalMapQueryContext nestedC = nestedCxt) { + nestedC.readLock().lock(); + } + }); + } + } + } + + @Test(expected = IllegalStateException.class) + public void testNestedIterationInIterationContextForbidden() { + try (ChronicleMap map = ChronicleMapBuilder + .of(Integer.class, Integer.class) + .entries(100) + .create()) { + map.put(42, 42); + try (MapSegmentContext cxt = map.segmentContext(0)) { + cxt.forEachSegmentEntry(e -> map.segmentContext(1).forEachSegmentEntry(e2 -> {})); + } + } + } +}