diff --git a/pom.xml b/pom.xml index 5b0008a48..140b910e1 100755 --- a/pom.xml +++ b/pom.xml @@ -44,7 +44,7 @@ net.openhft chronicle-bom - 1.10.35 + 1.10.37-SNAPSHOT pom import diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 04641f37f..6d4e7db7b 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -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); } diff --git a/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java b/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java index b36d8bf6c..e3011b5ec 100755 --- a/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java @@ -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); } diff --git a/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java b/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java index eb62a1271..c90b60f9a 100755 --- a/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java @@ -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); } diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListener.java b/src/main/java/net/openhft/chronicle/map/MapEventListener.java index 48e1c9275..c57d37b4a 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListener.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListener.java @@ -97,34 +97,6 @@ public abstract class MapEventListener 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 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); + } + } } diff --git a/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java b/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java index e378c6cdd..e6e1dc408 100644 --- a/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java +++ b/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java @@ -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) { diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 47d8e8223..7ab35e8a4 100755 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -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(127 + RESERVED_MOD_ITER); transient Set 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= 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 diff --git a/src/main/java/net/openhft/chronicle/map/ShortShortMultiMap.java b/src/main/java/net/openhft/chronicle/map/ShortShortMultiMap.java index 310ccf9d7..05e617ca6 100755 --- a/src/main/java/net/openhft/chronicle/map/ShortShortMultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/ShortShortMultiMap.java @@ -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); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 8b248e7a1..a7df7e140 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -99,11 +99,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo private long selectorTimeout; - enum State { - CONNECTED, DISCONNECTED; - } - - /** * @throws IOException on an io error. */ @@ -346,7 +341,6 @@ final class TcpReplicator 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 extends AbstractChannelReplicator implements Clo return result; } + enum State { + CONNECTED, DISCONNECTED; + } + static class StatelessClientParameters { VanillaChronicleMap map; SerializationBuilder keySerializationBuilder; @@ -896,7 +894,7 @@ final class TcpReplicator 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() { diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 90cbdc13e..8d9e08853 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -1395,11 +1395,10 @@ class VanillaChronicleMap, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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);