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

Factor Alloc stage abstraction, make alloc which fails to allocate in the current tier search from the first tier in chain

Browse Source
This commit is contained in:
leventov
2015-10-06 18:59:45 +03:00
parent e774c9f826
commit b967ad14d4
13 changed files with 242 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/Alloc.java
+28
View File
@@ -0,0 +1,28 @@
/*
* 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 <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.hash.impl.stage.entry;
public interface Alloc {
/**
* Allocates a position in a segment tier.
*
* @param chunks chunks to allocate
* @return the position, if allocation is successful, or -1
*/
long alloc(int chunks);
}
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/AllocatedChunks.java
+3 -2
View File
@@ -26,6 +26,7 @@ public class AllocatedChunks {
@StageRef public VanillaChronicleHashHolder<?, ?, ?> hh;
@StageRef public SegmentStages s;
@StageRef public HashEntryStages<?> entry;
@StageRef public Alloc alloc;
public int allocatedChunks = 0;
@@ -50,8 +51,8 @@ public class AllocatedChunks {
long oldKeySizeAddr = oldSegmentTierBaseAddr + entry.keySizeOffset;
long oldKeyAddr = oldSegmentTierBaseAddr + entry.keyOffset;
int tierBeforeAllocation = s.segmentTier;
long pos = s.alloc(allocatedChunks);
long pos = alloc.alloc(allocatedChunks);
entry.copyExistingEntry(pos, bytesToCopy, oldKeyAddr, oldKeySizeAddr);
return s.segmentTier > tierBeforeAllocation;
return s.segmentTier != tierBeforeAllocation;
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
+12 -11
View File
@@ -524,6 +524,18 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
throw new IllegalStateException("first tier doesn't have previous");
initSegmentTier(segmentTier - 1, prevTierIndex());
}
public void goToLastTier() {
while (hasNextTier()) {
nextTier();
}
}
public void goToFirstTier() {
while (segmentTier != 0) {
prevTier();
}
}
@Stage("Segment") public final PublicMultiStoreBytes segmentBytes = new PublicMultiStoreBytes();
@Stage("Segment") public final PointerBytesStore segmentBS = new PointerBytesStore();
@@ -589,17 +601,6 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface {
return ret;
}
public long alloc(int chunks) {
while (true) {
long ret = allocReturnCode(chunks);
if (ret >= 0) {
return ret;
} else {
nextTier();
}
}
}
public void free(long fromPos, int chunks) {
freeList.clearRange(fromPos, fromPos + chunks);
if (fromPos < nextPosToSearchFrom())
src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
+1 -7
View File
@@ -62,12 +62,6 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
abstract void closeHashLookupEntry();
private void goToLastTier() {
while (s.hasNextTier()) {
s.nextTier();
}
}
@Override
public boolean forEachSegmentEntryWhile(Predicate<? super E> action) {
s.innerUpdateLock.lock();
@@ -75,7 +69,7 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>>
long size = s.size();
if (size == 0)
return true;
goToLastTier();
s.goToLastTier();
while (true) {
int currentTier = s.segmentTier;
long currentTierBaseAddr = s.segmentBaseAddr;
src/main/java/net/openhft/chronicle/hash/impl/stage/iter/IterationAlloc.java
+43
View File
@@ -0,0 +1,43 @@
/*
* 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 <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.hash.impl.stage.iter;
import net.openhft.chronicle.hash.impl.stage.entry.Alloc;
import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
@Staged
public class IterationAlloc implements Alloc {
@StageRef public SegmentStages s;
/**
* Move only to next tiers, to avoid double visiting of relocated entries during iteration
*/
@Override
public long alloc(int chunks) {
while (true) {
long ret = s.allocReturnCode(chunks);
if (ret >= 0) {
return ret;
} else {
s.nextTier();
}
}
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/query/QueryAlloc.java
+45
View File
@@ -0,0 +1,45 @@
/*
* 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 <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.hash.impl.stage.query;
import net.openhft.chronicle.hash.impl.stage.entry.Alloc;
import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
@Staged
public class QueryAlloc implements Alloc {
@StageRef public SegmentStages s;
@Override
public long alloc(int chunks) {
long ret = s.allocReturnCode(chunks);
if (ret >= 0)
return ret;
int alreadyAttemptedTier = s.segmentTier;
s.goToFirstTier();
while (true) {
if (s.segmentTier != alreadyAttemptedTier) {
ret = s.allocReturnCode(chunks);
if (ret >= 0)
return ret;
}
s.nextTier();
}
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/query/SearchAllocatedChunks.java
+2 -2
View File
@@ -44,8 +44,8 @@ public class SearchAllocatedChunks extends AllocatedChunks {
// clears out searchState, and it performs the search again, but in inconsistent state
incrementSegmentEntriesIfNeeded();
int tierBeforeAllocation = s.segmentTier;
long pos = s.alloc(allocatedChunks);
long pos = alloc.alloc(allocatedChunks);
entry.writeNewEntry(pos, ks.inputKey);
return s.segmentTier > tierBeforeAllocation;
return s.segmentTier != tierBeforeAllocation;
}
}
src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java
+3
View File
@@ -22,6 +22,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
import net.openhft.chronicle.hash.impl.stage.hash.LogHolder;
import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder;
import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder;
import net.openhft.chronicle.hash.impl.stage.iter.IterationAlloc;
import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode;
import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages;
import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
@@ -61,6 +62,8 @@ import net.openhft.sg.Staged;
MapEntryStages.class,
ValueBytesInterop.class,
MapEntryOperationsDelegation.class,
IterationAlloc.class,
}, nested = {
ReadLock.class,
UpdateLock.class,
src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java
+2
View File
@@ -70,6 +70,8 @@ import net.openhft.sg.Staged;
HashInputBytes.class,
MapInputBytesValues.class,
QueryAlloc.class,
},
nested = {
ReadLock.class,
src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java
+3
View File
@@ -22,6 +22,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
import net.openhft.chronicle.hash.impl.stage.hash.LogHolder;
import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder;
import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder;
import net.openhft.chronicle.hash.impl.stage.iter.IterationAlloc;
import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode;
import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages;
import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
@@ -65,6 +66,8 @@ import net.openhft.sg.Staged;
ReplicationUpdate.class,
DefaultValue.class,
IterationAlloc.class,
}, nested = {
ReadLock.class,
UpdateLock.class,
src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java
+5 -5
View File
@@ -25,10 +25,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.LogHolder;
import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder;
import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder;
import net.openhft.chronicle.hash.impl.stage.input.HashInputBytes;
import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
import net.openhft.chronicle.hash.impl.stage.query.QueryHashLookupPos;
import net.openhft.chronicle.hash.impl.stage.query.QueryHashLookupSearch;
import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks;
import net.openhft.chronicle.hash.impl.stage.query.*;
import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData;
import net.openhft.chronicle.map.impl.stage.data.bytes.*;
import net.openhft.chronicle.map.impl.stage.data.instance.InputValueInstanceData;
@@ -38,6 +35,7 @@ import net.openhft.chronicle.map.impl.stage.input.MapInputBytesValues;
import net.openhft.chronicle.map.impl.stage.input.ReplicatedInput;
import net.openhft.chronicle.map.impl.stage.map.*;
import net.openhft.chronicle.map.impl.stage.query.*;
import net.openhft.chronicle.map.impl.stage.replication.ReplicatedQueryAlloc;
import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate;
import net.openhft.chronicle.map.impl.stage.ret.BytesReturnValue;
import net.openhft.chronicle.map.impl.stage.ret.DefaultReturnValue;
@@ -56,7 +54,7 @@ import net.openhft.sg.Staged;
ReplicatedChronicleMapHolderImpl.class,
KeyBytesInterop.class,
ReplicatedQuerySegmentStages.class,
QuerySegmentStages.class,
KeySearch.class,
InputKeyHashCode.class,
QueryHashLookupSearch.class,
@@ -78,6 +76,8 @@ import net.openhft.sg.Staged;
HashInputBytes.class,
MapInputBytesValues.class,
ReplicatedQueryAlloc.class,
},
nested = {
ReadLock.class,
src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java
-67
View File
@@ -1,67 +0,0 @@
/*
* 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 <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.map.impl.stage.query;
import net.openhft.chronicle.hash.ReplicatedHashSegmentContext;
import net.openhft.chronicle.hash.impl.stage.query.QuerySegmentStages;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.map.MapAbsentEntry;
import net.openhft.chronicle.map.MapSegmentContext;
import net.openhft.chronicle.map.ReplicatedChronicleMap;
import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
import java.util.function.Consumer;
@Staged
public abstract class ReplicatedQuerySegmentStages extends QuerySegmentStages {
@StageRef
ReplicatedChronicleMapHolder<?, ?, ?, ?, ?, ?, ?> mh;
final Consumer<ReplicableEntry> cleanupAction = e -> {
ReplicatedChronicleMap<?, ?, ?, ?, ?, ?, ?> map = mh.m();
if (e instanceof MapAbsentEntry) {
long deleteTimeout = map.timeProvider.currentTime() - e.originTimestamp();
if (deleteTimeout > map.cleanupTimeout && !e.isChanged())
e.doRemoveCompletely();
}
};
@Override
public long alloc(int chunks) {
while (true) {
long ret = allocReturnCode(chunks);
if (ret >= 0) {
return ret;
} else {
ReplicatedChronicleMap<?, ?, ?, ?, ?, ?, ?> map = mh.m();
try (MapSegmentContext<?, ?, ?> sc = map.segmentContext(segmentIndex)) {
((ReplicatedHashSegmentContext<?, ?>) sc)
.forEachSegmentReplicableEntry(cleanupAction);
}
ret = allocReturnCode(chunks);
if (ret >= 0) {
return ret;
} else {
nextTier();
}
}
}
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicatedQueryAlloc.java
+95
View File
@@ -0,0 +1,95 @@
/*
* 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 <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.map.impl.stage.replication;
import net.openhft.chronicle.hash.ReplicatedHashSegmentContext;
import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
import net.openhft.chronicle.hash.impl.stage.query.QueryAlloc;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.map.MapAbsentEntry;
import net.openhft.chronicle.map.MapSegmentContext;
import net.openhft.chronicle.map.ReplicatedChronicleMap;
import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
import java.util.function.Consumer;
@Staged
public class ReplicatedQueryAlloc extends QueryAlloc {
@StageRef ReplicatedChronicleMapHolder<?, ?, ?, ?, ?, ?, ?> mh;
@StageRef SegmentStages s;
final CleanupAction cleanupAction = new CleanupAction();
/**
* Returns {@code true} if at least one old deleted entry was removed.
*/
public boolean forcedOldDeletedEntriesCleanup() {
ReplicatedChronicleMap<?, ?, ?, ?, ?, ?, ?> map = mh.m();
try (MapSegmentContext<?, ?, ?> sc = map.segmentContext(s.segmentIndex)) {
cleanupAction.removedCompletely = 0;
((ReplicatedHashSegmentContext<?, ?>) sc)
.forEachSegmentReplicableEntry(cleanupAction);
return cleanupAction.removedCompletely > 0;
}
}
private class CleanupAction implements Consumer<ReplicableEntry> {
int removedCompletely;
@Override
public void accept(ReplicableEntry e) {
ReplicatedChronicleMap<?, ?, ?, ?, ?, ?, ?> map = mh.m();
if (e instanceof MapAbsentEntry) {
long deleteTimeout = map.timeProvider.currentTime() - e.originTimestamp();
map.timeProvider.systemTimeIntervalBetween(e.originTimestamp(),
map.timeProvider.currentTime(), map.cleanupTimeoutUnit);
if (deleteTimeout > map.cleanupTimeout && !e.isChanged()) {
e.doRemoveCompletely();
removedCompletely++;
}
}
}
}
@Override
public long alloc(int chunks) {
long ret = s.allocReturnCode(chunks);
if (ret >= 0)
return ret;
int alreadyAttemptedTier;
if (!forcedOldDeletedEntriesCleanup()) {
alreadyAttemptedTier = s.segmentTier;
} else {
// If forced old deleted entries cleanup deleted something, don't skip the first queried
// tier, because it could have new empty chunks now. segment tier is >= 0, so set to -1
// to make the condition with alreadyAttemptedTier always false
alreadyAttemptedTier = -1;
}
s.goToFirstTier();
while (true) {
if (s.segmentTier != alreadyAttemptedTier) {
ret = s.allocReturnCode(chunks);
if (ret >= 0)
return ret;
}
s.nextTier();
}
}
}