From 82988e743554b30f2b2cb575994acd422ba7614a Mon Sep 17 00:00:00 2001 From: leventov Date: Sun, 6 Dec 2015 15:19:49 +0300 Subject: [PATCH] Replace outdated (pre-tiers) implementation of ReplicatedMapSegmentIteration.forEachSegmentReplicableEntryWhile() by shared code with HashSegmentIteration.forEachSegmentEntryWhile() --- .../chronicle/hash/HashSegmentContext.java | 10 +-- .../impl/stage/iter/HashSegmentIteration.java | 42 +++++++----- .../iter/ReplicatedMapSegmentIteration.java | 64 ++++++++----------- 3 files changed, 59 insertions(+), 57 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/HashSegmentContext.java b/src/main/java/net/openhft/chronicle/hash/HashSegmentContext.java index 9ea6011b5..2c8dae8fc 100644 --- a/src/main/java/net/openhft/chronicle/hash/HashSegmentContext.java +++ b/src/main/java/net/openhft/chronicle/hash/HashSegmentContext.java @@ -31,16 +31,16 @@ import java.util.function.Predicate; public interface HashSegmentContext> extends HashContext, Closeable { /** - * Performs the given action for each entry in this segment until all entries have - * been processed or the action throws an {@code Exception}. Exceptions thrown by the action are - * relayed to the caller. + * Performs the given action for each present entry in this segment until all entries + * have been processed or the action throws an {@code Exception}. Exceptions thrown by the + * action are relayed to the caller. * * @param action the action to be performed for each entry in this segment */ void forEachSegmentEntry(Consumer action); /** - * Checks the given predicate on each entry in this segment until all entries + * Checks the given predicate on each present entry in this segment until all entries * have been processed or the predicate returns {@code false} for some entry, or throws * an {@code Exception}. Exceptions thrown by the predicate are relayed to the caller. * @@ -54,7 +54,7 @@ public interface HashSegmentContext> extends HashConte boolean forEachSegmentEntryWhile(Predicate predicate); /** - * Returns the number of entries in this segment. + * Returns the number of present entries in this segment. */ long size(); } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java index a212b66fc..efb90ed71 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java @@ -39,10 +39,14 @@ public abstract class HashSegmentIteration> @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation; @StageRef protected HashLookupPos hlp; - public boolean entryIsPresent() { + public boolean shouldTestEntry() { return true; } + public Object entryForIteration() { + return this; + } + public boolean entryRemovedOnThisIteration = false; abstract boolean entryRemovedOnThisIterationInit(); @@ -62,27 +66,33 @@ public abstract class HashSegmentIteration> abstract void closeHashLookupEntry(); @Override - public boolean forEachSegmentEntryWhile(Predicate action) { + public boolean forEachSegmentEntryWhile(Predicate predicate) { + checkOnEachPublicOperation.checkOnEachPublicOperation(); s.innerUpdateLock.lock(); + return innerForEachSegmentEntryWhile(predicate, s.size()); + } + + public boolean innerForEachSegmentEntryWhile( + Predicate predicate, long expectedEntriesToIterate) { try { - long size = s.size(); - if (size == 0) + long leftEntries = expectedEntriesToIterate; + if (leftEntries == 0) return true; s.goToLastTier(); while (true) { int currentTier = s.tier; long currentTierBaseAddr = s.tierBaseAddr; long currentTierIndex = s.tierIndex; - size = forEachTierWhile(action, size, + leftEntries = forEachTierWhile(predicate, leftEntries, currentTier, currentTierBaseAddr, currentTierIndex); - if (size < 0) // interrupted + if (leftEntries < 0) // interrupted return false; - if (size == 0) + if (leftEntries == 0) return true; if (currentTier == 0) throw new IllegalStateException("We iterated all tiers without interruption, " + - "but according to size counter there should be " + size + " more " + - "entries. Size diverged?"); + "but according to segment counters there should be " + leftEntries + + " more entries. Size diverged?"); s.prevTier(); } } finally { @@ -92,8 +102,8 @@ public abstract class HashSegmentIteration> } } - private long forEachTierWhile( - Predicate action, long size, + private long forEachTierWhile( + Predicate predicate, long leftEntries, int currentTier, long currentTierBaseAddr, long tierIndex) { boolean interrupted = false; long startPos = 0L; @@ -107,7 +117,7 @@ public abstract class HashSegmentIteration> do { // Step from hlp.hashLookupPos, not currentHashLookupPos (with additional initialization // of this local variable to startPos outside the loop), because if e.remove() is - // called in the `action`, hlp.hashLookupPos is stepped back in doRemove(), and + // called in the `predicate`, hlp.hashLookupPos is stepped back in doRemove(), and // currentHashLookupPos become invalid currentHashLookupPos = hashLookup.step(hlp.hashLookupPos); steps++; @@ -116,14 +126,14 @@ public abstract class HashSegmentIteration> initHashLookupEntry(entry); if (!hashLookup.empty(entry)) { e.readExistingEntry(hashLookup.value(entry)); - if (entryIsPresent()) { + if (shouldTestEntry()) { initEntryRemovedOnThisIteration(false); try { - if (!action.test((E) this)) { + if (!predicate.test((T) entryForIteration())) { interrupted = true; break; } else { - if (--size == 0) + if (--leftEntries == 0) break; } } finally { @@ -150,7 +160,7 @@ public abstract class HashSegmentIteration> // becomes equal to start pos without making the whole loop, but only visiting a single // entry } while (currentHashLookupPos != startPos || steps == 0); - return interrupted ? ~size : size; + return interrupted ? ~leftEntries : leftEntries; } @Override 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 0b4d4b2ac..2c585c453 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 @@ -18,7 +18,6 @@ package net.openhft.chronicle.map.impl.stage.iter; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.ReplicatedHashSegmentContext; -import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.map.MapAbsentEntry; import net.openhft.chronicle.map.MapEntry; @@ -35,6 +34,9 @@ import net.openhft.sg.Staged; import java.util.function.Consumer; import java.util.function.Predicate; +import static net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapSegmentIteration.EntriesToTest.ALL; +import static net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapSegmentIteration.EntriesToTest.PRESENT; + @Staged public abstract class ReplicatedMapSegmentIteration extends MapSegmentIteration implements ReplicatedIterationContext, ReplicableEntry, @@ -47,9 +49,21 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI @StageRef ReplicatedMapAbsentDelegatingForIteration absentEntryDelegating; @StageRef ReplicatedMapEntryDelegating entryDelegating; + enum EntriesToTest {PRESENT, ALL} + EntriesToTest entriesToTest = null; + + void initEntriesToTest(EntriesToTest entriesToTest) { + this.entriesToTest = entriesToTest; + } + @Override - public boolean entryIsPresent() { - return !e.entryDeleted(); + public boolean shouldTestEntry() { + return entriesToTest == ALL || !e.entryDeleted(); + } + + @Override + public Object entryForIteration() { + return !e.entryDeleted() ? entryDelegating : absentEntryDelegating; } @Override @@ -64,43 +78,21 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI } } + @Override + public boolean forEachSegmentEntryWhile(Predicate> predicate) { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + initEntriesToTest(PRESENT); + s.innerUpdateLock.lock(); + return innerForEachSegmentEntryWhile(predicate, s.size()); + } + @Override public boolean forEachSegmentReplicableEntryWhile( Predicate predicate) { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + initEntriesToTest(ALL); s.innerUpdateLock.lock(); - try { - long entries = s.entries(); - if (entries == 0) - return true; - boolean interrupted = false; - long startPos = 0L; - CompactOffHeapLinearHashTable hashLookup = mh.h().hashLookup; - while (!hashLookup.empty(hashLookup.readEntry(s.tierBaseAddr, startPos))) { - startPos = hashLookup.step(startPos); - } - hlp.initHashLookupPos(startPos); - do { - hlp.setHashLookupPos(hashLookup.step(hlp.hashLookupPos)); - long entry = hashLookup.readEntry(s.tierBaseAddr, hlp.hashLookupPos); - if (!hashLookup.empty(entry)) { - e.readExistingEntry(hashLookup.value(entry)); - ReplicableEntry e = !this.e.entryDeleted() ? entryDelegating : - absentEntryDelegating; - initEntryRemovedOnThisIteration(false); - if (!predicate.test(e)) { - interrupted = true; - break; - } else { - if (--entries == 0) - break; - } - } - } while (hlp.hashLookupPos != startPos); - return !interrupted; - } finally { - s.innerReadLock.unlock(); - initEntryRemovedOnThisIteration(false); - } + return innerForEachSegmentEntryWhile(predicate, s.entries()); } @Override