cluster tuning

This commit is contained in:
Peter Lawrey
2016-01-01 09:04:31 +00:00
parent 77e24b9b1c
commit e0cb22f22b
10 changed files with 117 additions and 111 deletions
+1 -1
View File
@@ -44,7 +44,7 @@
<dependency>
<groupId>net.openhft</groupId>
<artifactId>chronicle-bom</artifactId>
<version>1.10.35</version>
<version>1.10.37-SNAPSHOT</version>
<type>pom</type>
<scope>import</scope>
</dependency>
@@ -446,7 +446,7 @@ abstract class AbstractChannelReplicator implements Closeable {
EntryCallback(@NotNull final Replica.EntryExternalizable externalizable,
final int tcpBufferSize) {
this.externalizable = externalizable;
out = ByteBuffer.allocateDirect(tcpBufferSize);
out = ByteBuffer.allocateDirect(tcpBufferSize).order(ByteOrder.nativeOrder());
in = new ByteBufferBytes(out);
}
@@ -60,7 +60,7 @@ class Int24Int24MultiMap implements MultiMap {
capacityMask = capacity - 1L;
capacityInBytes = indexToPos(capacity);
this.bytes = multiMapBytes;
positions = new ATSDirectBitSet(multiMapBitSetBytes);
positions = ATSDirectBitSet.wrap(multiMapBitSetBytes);
}
/**
@@ -80,16 +80,6 @@ class Int24Int24MultiMap implements MultiMap {
return (key &= MASK) != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY;
}
private void checkValueForPut(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isClear(value) : "Shouldn't put existing value";
}
private void checkValueForRemove(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isSet(value) : "Shouldn't remove absent value";
}
private static long key(long entry) {
return entry >>> 24;
}
@@ -102,6 +92,16 @@ class Int24Int24MultiMap implements MultiMap {
return (key << 24) | (value & MASK);
}
private void checkValueForPut(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isClear(value) : "Shouldn't put existing value";
}
private void checkValueForRemove(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isSet(value) : "Shouldn't remove absent value";
}
private long pos(long key) {
return indexToPos(key & capacityMask);
}
@@ -61,7 +61,7 @@ class IntIntMultiMap implements MultiMap {
capacityMask = capacity - 1L;
capacityMask2 = indexToPos(capacityMask);
this.bytes = multiMapBytes;
positions = new ATSDirectBitSet(multiMapBitSetBytes);
positions = ATSDirectBitSet.wrap(multiMapBitSetBytes);
}
/**
@@ -81,16 +81,6 @@ class IntIntMultiMap implements MultiMap {
return (key &= MASK) != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY;
}
private void checkValueForPut(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isClear(value) : "Shouldn't put existing value";
}
private void checkValueForRemove(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isSet(value) : "Shouldn't remove absent value";
}
private static long key(long entry) {
return entry >>> 32;
}
@@ -103,6 +93,16 @@ class IntIntMultiMap implements MultiMap {
return (key << 32) | value;
}
private void checkValueForPut(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isClear(value) : "Shouldn't put existing value";
}
private void checkValueForRemove(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isSet(value) : "Shouldn't remove absent value";
}
private long pos(long key) {
return indexToPos(key & capacityMask);
}
@@ -97,34 +97,6 @@ public abstract class MapEventListener<K, V> implements Serializable {
}
private static class LoggingMapEventListener extends MapEventListener {
private static final long serialVersionUID = 0L;
public final static Logger LOGGER = LoggerFactory.getLogger(LoggingMapEventListener.class);
private final String prefix;
private LoggingMapEventListener(String prefix) {
this.prefix = prefix;
}
@Override
public void onGetFound(Object key, Object value) {
LOGGER.info("get {} => {}", prefix, key, value);
}
@Override
public void onPut(Object key, Object value, Object replacedValue, boolean
replicationEvent, boolean updateResult, boolean hasValueChanged, byte identifier,
byte replacedIdentifier, long timeStamp, long replacedTimeStamp) {
LOGGER.info("{} put {} => {}", prefix, key, value);
}
@Override
public void onRemove(Object key, Object value, boolean replicationEvent, byte identifier, byte replacedIdentifier, long timestamp, long replacedTimeStamp) {
LOGGER.info("{} remove {} was {}", prefix, key, value);
}
}
/**
* This method is called if the key is found in the map during {@link ChronicleMap#get get},
* {@link ChronicleMap#getUsing getUsing} or {@link ChronicleMap#acquireUsing acquireUsing}
@@ -155,5 +127,41 @@ public abstract class MapEventListener<K, V> implements Serializable {
byte replacedIdentifier, long timestamp, long replacedTimeStamp) {
// do nothing
}
/**
* This performs an early check to see if the events will be consumed or discarded.
*
* @return true unless you know the event won't be needed.
*/
public boolean isActive() {
return true;
}
private static class LoggingMapEventListener extends MapEventListener {
public final static Logger LOGGER = LoggerFactory.getLogger(LoggingMapEventListener.class);
private static final long serialVersionUID = 0L;
private final String prefix;
private LoggingMapEventListener(String prefix) {
this.prefix = prefix;
}
@Override
public void onGetFound(Object key, Object value) {
LOGGER.info("get {} => {}", prefix, key, value);
}
@Override
public void onPut(Object key, Object value, Object replacedValue, boolean
replicationEvent, boolean updateResult, boolean hasValueChanged, byte identifier,
byte replacedIdentifier, long timeStamp, long replacedTimeStamp) {
LOGGER.info("{} put {} => {}", prefix, key, value);
}
@Override
public void onRemove(Object key, Object value, boolean replicationEvent, byte identifier, byte replacedIdentifier, long timestamp, long replacedTimeStamp) {
LOGGER.info("{} remove {} was {}", prefix, key, value);
}
}
}
@@ -18,6 +18,7 @@ package net.openhft.chronicle.map;
import net.openhft.lang.Maths;
import net.openhft.lang.collection.ATSDirectBitSet;
import net.openhft.lang.collection.DirectBitSet;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.DirectStore;
@@ -82,11 +83,11 @@ enum MultiMapFactory {
return LONGS.align(BYTES.alignAndConvert(actualChunksPerSegment, BITS), BYTES);
}
public static ATSDirectBitSet newPositions(long capacity) {
public static DirectBitSet newPositions(long capacity) {
if (!isPowerOf2(capacity))
throw new AssertionError("capacity should be a power of 2");
capacity = LONGS.align(capacity, BITS);
return new ATSDirectBitSet(DirectStore.allocateLazy(BYTES.convert(capacity, BITS)).bytes());
return ATSDirectBitSet.wrap(DirectStore.allocateLazy(BYTES.convert(capacity, BITS)).bytes());
}
static long multiMapCapacity(long minCapacity) {
@@ -30,6 +30,7 @@ import net.openhft.chronicle.map.Replica.EntryResolver;
import net.openhft.chronicle.map.ReplicatedChronicleMap.ModificationIterator.EntryModifiableCallback;
import net.openhft.lang.Maths;
import net.openhft.lang.collection.ATSDirectBitSet;
import net.openhft.lang.collection.DirectBitSet;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.CheckedBytes;
import net.openhft.lang.io.MultiStoreBytes;
@@ -107,7 +108,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
new AtomicReferenceArray<ModificationIterator>(127 + RESERVED_MOD_ITER);
transient Set<Closeable> closeables;
private transient Bytes identifierUpdatedBytes;
private transient ATSDirectBitSet modIterSet;
private transient DirectBitSet modIterSet;
private transient long startOfModificationIterators;
private boolean bootstrapOnlyLocalEntries;
@@ -199,7 +200,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
Bytes modDelBytes = ms.bytes(offset, assignedModIterBitSetSizeInBytes()).zeroOut();
offset += assignedModIterBitSetSizeInBytes();
startOfModificationIterators = offset;
modIterSet = new ATSDirectBitSet(new CheckedBytes(modDelBytes));
modIterSet = ATSDirectBitSet.wrap(new CheckedBytes(modDelBytes));
}
/**
@@ -961,7 +962,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
true, localIdentifier, replacedIdentifier, timestamp,
replacedTimestamp, this);
}
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
V valueInstance = toValue.toInstance(copies, v, valueSize);
writeUnlock();
// TODO unlocking is dangerous, because this method is a part of
@@ -1185,7 +1186,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
identifier, replacedIdentifier,
timeStamp, replacedTimestamp, this);
}
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
final K key1 = toKey.toInstance(copies, key, keySize);
final V newValue = toValue.toInstance(copies, value, valueSize);
@@ -1226,7 +1227,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
segmentState.valueSizePos, true, false, true, identifier,
replacedIdentifier, timeStamp, replacedTimestamp, this);
}
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
byte replacedIdentifier = 0;
long replacedTimestamp = 0;
final K key1 = toKey.toInstance(copies, key, keySize);
@@ -1351,7 +1352,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
replacedIdentifier, timestamp,
replacedTimestamp, this);
}
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
byte replacedIdentifier = 0;
long replacedTimestamp = 0;
final K key1 = toKey.toInstance(copies, key, keySize);
@@ -1761,13 +1762,14 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
*/
class ModificationIterator implements Replica.ModificationIterator {
private final ATSDirectBitSet changes;
private final DirectBitSet changes;
private final int segmentIndexShift;
private final long posMask;
private final EntryModifiableCallback entryModifiableCallback =
new EntryModifiableCallback();
// todo get rid of this
private final MultiStoreBytes tmpBytes = new MultiStoreBytes();
volatile Thread thread = null;
private ModificationNotifier modificationNotifier;
// when a bootstrap is send this is the time stamp that the client will bootstrap up to
// if it is set as ZERO then the onPut() will set it to the current time, once the
@@ -1787,14 +1789,14 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
long bitsPerSegment = bitsPerSegmentInModIterBitSet();
segmentIndexShift = Long.numberOfTrailingZeros(bitsPerSegment);
posMask = bitsPerSegment - 1L;
changes = new ATSDirectBitSet(bytes);
changes = ATSDirectBitSet.wrap(bytes);
}
public ModificationIterator(@NotNull final Bytes bytes) {
long bitsPerSegment = bitsPerSegmentInModIterBitSet();
segmentIndexShift = Long.numberOfTrailingZeros(bitsPerSegment);
posMask = bitsPerSegment - 1L;
changes = new ATSDirectBitSet(bytes);
changes = ATSDirectBitSet.wrap(bytes);
}
public void setModificationNotifier(ModificationNotifier modificationNotifier) {
@@ -1868,8 +1870,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
(position >= 0L && changes.nextSetBit(0L) != NOT_FOUND);
}
volatile Thread thread = null;
/**
* @param entryCallback call this to get an entry, this class will take care of the locking
* @return true if an entry was processed
@@ -42,7 +42,7 @@ class ShortShortMultiMap implements MultiMap {
private final int capacityMask;
private final int capacityMask2;
private final Bytes bytes;
private ATSDirectBitSet positions;
private DirectBitSet positions;
public ShortShortMultiMap(long minCapacity) {
assert minCapacity <= MAX_CAPACITY;
@@ -64,7 +64,7 @@ class ShortShortMultiMap implements MultiMap {
capacityMask = (int) (capacity - 1L);
capacityMask2 = (int) (capacityMask * ENTRY_SIZE);
this.bytes = multiMapBytes;
positions = new ATSDirectBitSet(multiMapBitSetBytes);
positions = ATSDirectBitSet.wrap(multiMapBitSetBytes);
}
/**
@@ -76,32 +76,10 @@ class ShortShortMultiMap implements MultiMap {
return multiMapCapacity(minCapacity) * ENTRY_SIZE;
}
private void checkValueForPut(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isClear(value) : "Shouldn't put existing value";
}
private void checkValueForRemove(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isSet(value) : "Shouldn't remove absent value";
}
private static long maskUnsetKey(long key) {
return (key &= MASK) != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY;
}
private int step(int pos) {
return (pos + ENTRY_SIZE) & capacityMask2;
}
private long stepBack(long pos) {
return (pos - ENTRY_SIZE) & capacityMask2;
}
private int pos(long key) {
return ((int) key & capacityMask) << ENTRY_SIZE_SHIFT;
}
private static int entry(long key, long value) {
return (int) ((key << 16) | value);
}
@@ -114,6 +92,28 @@ class ShortShortMultiMap implements MultiMap {
return (long) (entry & 0xFFFF);
}
private void checkValueForPut(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isClear(value) : "Shouldn't put existing value";
}
private void checkValueForRemove(long value) {
assert (value & ~MASK) == 0L : "Value out of range, was " + value;
assert positions.isSet(value) : "Shouldn't remove absent value";
}
private int step(int pos) {
return (pos + ENTRY_SIZE) & capacityMask2;
}
private long stepBack(long pos) {
return (pos - ENTRY_SIZE) & capacityMask2;
}
private int pos(long key) {
return ((int) key & capacityMask) << ENTRY_SIZE_SHIFT;
}
@Override
public void putPosition(long value) {
checkValueForPut(value);
@@ -99,11 +99,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
private long selectorTimeout;
enum State {
CONNECTED, DISCONNECTED;
}
/**
* @throws IOException on an io error.
*/
@@ -346,7 +341,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
}
}
/**
* check to see if its time to send a heartbeat, and send one if required
*
@@ -864,6 +858,10 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
return result;
}
enum State {
CONNECTED, DISCONNECTED;
}
static class StatelessClientParameters<K, V> {
VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
SerializationBuilder<K> keySerializationBuilder;
@@ -896,7 +894,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
this.op = op;
this.selectionKeys = selectionKeys;
long bitSetSize = LONGS.align(BYTES.alignAndConvert(selectionKeys.length, BITS), BYTES);
changeOfOpWriteRequired = new ATSDirectBitSet(DirectStore.allocate(bitSetSize).bytes());
changeOfOpWriteRequired = ATSDirectBitSet.wrap(DirectStore.allocate(bitSetSize).bytes());
}
public void applyUpdates() {
@@ -1395,11 +1395,10 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
private final HeapWriteLocked heapWriteLocked = new HeapWriteLocked(this);
long pos;
long valueSizePos;
private boolean used;
private SegmentState next;
byte identifier = 0;
long timestamp = 0;
private boolean used;
private SegmentState next;
private SegmentState(int depth) {
if (depth > (1 << 10))
@@ -2124,7 +2123,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
segmentState.identifier, replacedIdentifier,
segmentState.timestamp, replacedTimeStamp, this);
}
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
byte replacedIdentifier = (byte) 0;
long replacedTimeStamp = 0;
final K key1 = toKey.toInstance(copies, key, keySize);
@@ -2164,7 +2163,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
if (bytesEventListener != null) {
bytesEventListener.onGetFound(entry, 0L, metaDataBytes, valuePos);
}
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
eventListener.onGetFound(toKey.toInstance(copies, key, keySize),
toValue.toInstance(copies, v, valueSize));
}
@@ -2237,7 +2236,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
segmentState.identifier, (byte) 0,
segmentState.timestamp, 0, this
);
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
final K key1 = toKey.toInstance(copies, key, keySize);
final V newValue = toValue.toInstance(copies, value, valueSize);
writeUnlock();
@@ -2271,7 +2270,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
segmentState.identifier, replacedIdentifier,
segmentState.timestamp, replacedTimeStamp, this);
}
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
byte replacedIdentifier = 0;
long replacedTimeStamp = 0;
final K key1 = toKey.toInstance(copies, key, keySize);
@@ -2386,7 +2385,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
bytesEventListener.onPut(entry, 0L, metaDataBytes,
segmentState.valueSizePos, true, false, true,
segmentState.identifier, (byte) 0, segmentState.timestamp, 0, this);
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
final K key1 = toKey.toInstance(copies, key, keySize);
final V newValue = toValue.toInstance(copies, value, valueSize);
writeUnlock();
@@ -2429,7 +2428,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
if (!resultUnused && !entryIsDeleted)
prevValue = readValue.readValue(copies, entry, null, prevValueSize);
// todo optimize -- prevValue could be read twice
if (eventListener != null && !entryIsDeleted) {
if (eventListener != null && eventListener.isActive() && !entryIsDeleted) {
entry.positionAddr(valueAddr);
prevValueInstance = null;
if (!putReturnsNull)
@@ -2438,7 +2437,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
boolean doPutValue;
boolean hasValueChanged = false;
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
entry.positionAddr(valueAddr);
hasValueChanged = prevValueSize != valueSize ||
!metaValueInterop.startsWith(valueInterop, entry, value);
@@ -2469,7 +2468,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
hasValueChanged,
segmentState.identifier, replacedIdentifier,
segmentState.timestamp, replacedTimestamp, this);
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
final K key1 = toKey.toInstance(copies, key, keySize);
final V newValue = toValue.toInstance(copies, value, valueSize);
@@ -2717,7 +2716,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
boolean booleanResult, byte replacedIdentifier, long replacedTimestamp) {
// get the removed value, if needed
RV removedValue = null;
if ((!booleanResult && !resultUnused) || eventListener != null) {
if ((!booleanResult && !resultUnused) || (eventListener != null && eventListener.isActive())) {
// todo reuse some value
removedValue = readValue.readValue(copies, entry, null, valueSize);
entry.position(entry.position() - valueSize);
@@ -2739,7 +2738,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
bytesEventListener.onRemove(entry, 0L, metaDataBytes, valueSizePos, remote,
segmentState.identifier, replacedIdentifier,
segmentState.timestamp, replacedTimestamp, this);
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
V removedValueForEventListener =
toValue.toInstance(copies, removedValue, valueSize);
@@ -2884,7 +2883,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
long newValueSize = metaValueInterop.size(valueInterop, newValue);
boolean doPutValue;
boolean hasValueChanged = false;
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
entry.positionAddr(valueAddr);
hasValueChanged = valueSize != newValueSize ||
!metaValueInterop.startsWith(valueInterop, entry, newValue);
@@ -2916,7 +2915,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
segmentState.timestamp, replacedTimestamp, this);
}
if (eventListener != null) {
if (eventListener != null && eventListener.isActive()) {
final K key1 = toKey.toInstance(copies, key, keySize);
final V newValue1 = toValue.toInstance(copies, newValue, newValueSize);
final V replacedValue = toValue.toInstance(copies, prevValue, valueSize);