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

Replace outdated (pre-tiers) implementation of ReplicatedMapSegmentIteration.forEachSegmentReplicableEntryWhile() by shared code with HashSegmentIteration.forEachSegmentEntryWhile()

Browse Source
This commit is contained in:
leventov
2015-12-06 15:19:49 +03:00
parent dbba27660f
commit 82988e7435
3 changed files with 59 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/HashSegmentContext.java
+5 -5
View File
@@ -31,16 +31,16 @@ import java.util.function.Predicate;
public interface HashSegmentContext<K, E extends HashEntry<K>> extends HashContext<K>, 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 <i>present</i> 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<? super E> action);
/**
* Checks the given predicate on each entry in this segment until all entries
* Checks the given predicate on each <i>present</i> 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<K, E extends HashEntry<K>> extends HashConte
boolean forEachSegmentEntryWhile(Predicate<? super E> predicate);
/**
* Returns the number of entries in this segment.
* Returns the number of <i>present</i> entries in this segment.
*/
long size();
}
src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
+26 -16
View File
@@ -39,10 +39,14 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
@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<K, E extends HashEntry<K>>
abstract void closeHashLookupEntry();
@Override
public boolean forEachSegmentEntryWhile(Predicate<? super E> action) {
public boolean forEachSegmentEntryWhile(Predicate<? super E> predicate) {
checkOnEachPublicOperation.checkOnEachPublicOperation();
s.innerUpdateLock.lock();
return innerForEachSegmentEntryWhile(predicate, s.size());
}
public <T> boolean innerForEachSegmentEntryWhile(
Predicate<? super T> 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<K, E extends HashEntry<K>>
}
}
private long forEachTierWhile(
Predicate<? super E> action, long size,
private <T> long forEachTierWhile(
Predicate<? super T> predicate, long leftEntries,
int currentTier, long currentTierBaseAddr, long tierIndex) {
boolean interrupted = false;
long startPos = 0L;
@@ -107,7 +117,7 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
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<K, E extends HashEntry<K>>
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<K, E extends HashEntry<K>>
// 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
src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java
+28 -36
View File
@@ -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<K, V, R> extends MapSegmentIteration<K, V, R>
implements ReplicatedIterationContext<K, V, R>, ReplicableEntry,
@@ -47,9 +49,21 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> extends MapSegmentI
@StageRef ReplicatedMapAbsentDelegatingForIteration<K, V> absentEntryDelegating;
@StageRef ReplicatedMapEntryDelegating<K, V> 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<K, V, R> extends MapSegmentI
}
}
@Override
public boolean forEachSegmentEntryWhile(Predicate<? super MapEntry<K, V>> predicate) {
checkOnEachPublicOperation.checkOnEachPublicOperation();
initEntriesToTest(PRESENT);
s.innerUpdateLock.lock();
return innerForEachSegmentEntryWhile(predicate, s.size());
}
@Override
public boolean forEachSegmentReplicableEntryWhile(
Predicate<? super ReplicableEntry> 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