From c2e85064d6f1137bce51884707315107b4d419d7 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 7 Nov 2014 18:46:26 +0000 Subject: [PATCH] HCOLL-192 added test case net.openhft.chronicle.map.EventListenerTestWithTCPSocketReplication.testRemoteNotifyRemove() - and implemented its functionality. --- .../openhft/chronicle/map/NodeDiscovery.java | 21 ++-- .../chronicle/map/ReplicatedChronicleMap.java | 101 +++++++++++++----- ...tListenerTestWithTCPSocketReplication.java | 66 +++++++++--- 3 files changed, 136 insertions(+), 52 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 10b2954c2..ecbb09cd2 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -259,8 +259,8 @@ public class NodeDiscovery { /** - * bitwise OR's the two bit sets or put another way, merges the source bitset into the - * destination bitset and returns the destination + * bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset + * and returns the destination * * @param source to or * @param destination to or @@ -326,8 +326,8 @@ public class NodeDiscovery { } /** - * Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote - * nodes just by name, this functionality requires UDP + * Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote nodes + * just by name, this functionality requires UDP * * @author Rob Austin. */ @@ -738,8 +738,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { } /** - * this is used to tell nodes that are connecting to us which host and ports are in our grid, - * along with all the identifiers. + * this is used to tell nodes that are connecting to us which host and ports are in our grid, along with + * all the identifiers. */ @Override public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { @@ -819,7 +819,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { if (bootstrap.addressAndPort().equals(this.ourAddressAndPort)) return; - LOG.info("Received Bootstrap from " + bootstrap); + if (LOG.isDebugEnabled()) + LOG.debug("Received Bootstrap from " + bootstrap); proposedIdentifiersWithHost.put(bootstrap.addressAndPort, bootstrap); @@ -839,7 +840,7 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { if (sourceAddressAndPort.equals(ourAddressAndPort)) return; - LOG.info("Received Proposal"); + LOG.debug("Received Proposal"); this.remoteNode.readMarshallable(in); this.proposedIdentifiersWithHost.readMarshallable(in); @@ -929,8 +930,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { /** - * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the - * host:port and perhaps even proposed identifier of the node that sent it. + * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the host:port + * and perhaps even proposed identifier of the node that sent it. * * @param proposedNodes */ diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 7dd3f8404..63953105b 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -209,7 +209,7 @@ final class ReplicatedChronicleMap, /** * @return true if the user has added an event listerner */ - boolean hasAttachedEventListener() { + boolean hasAttachedEventListeners() { return eventListener != null; } @@ -440,7 +440,8 @@ final class ReplicatedChronicleMap, // we store the isDeleted flag in the identifier // ( when the identifier is negative it is deleted ) - destination.writeByte(isDeleted ? -identifier : identifier); + destination.writeByte(identifier); + destination.writeBoolean(isDeleted); // write the key entry.position(keyPosition); @@ -488,14 +489,12 @@ final class ReplicatedChronicleMap, final long valueSize = valueSizeMarshaller.readSize(source); final long timeStamp = source.readStopBit(); final byte id = source.readByte(); - final byte remoteIdentifier; - final boolean isDeleted; + final boolean isDeleted = source.readBoolean(); - if (id < 0) { - isDeleted = true; - remoteIdentifier = (byte) -id; - } else if (id != 0) { - isDeleted = false; + final byte remoteIdentifier; + + if (id != 0) { + // isDeleted = false; remoteIdentifier = id; } else { throw new IllegalStateException("identifier can't be 0"); @@ -520,8 +519,7 @@ final class ReplicatedChronicleMap, if (isDeleted) { if (debugEnabled) { - LOG.debug( - "READING FROM SOURCE - into local-id={}, remote={}, remove(key={})", + LOG.debug("READING FROM SOURCE - into local-id={}, remote={}, remove(key={})", localIdentifier, remoteIdentifier, source.toString().trim() ); } @@ -685,11 +683,14 @@ final class ReplicatedChronicleMap, bytes.skip(ADDITIONAL_ENTRY_BYTES); } + /** * called from a remote node as part of replication */ private void remoteRemove(Bytes keyBytes, long hash2, final long timestamp, final byte identifier) { + + writeLock(); try { long keySize = keyBytes.remaining(); @@ -701,6 +702,8 @@ final class ReplicatedChronicleMap, continue; // key is found + + long keyPosition = entry.position(); entry.skip(keySize); long timeStampPos = entry.position(); @@ -709,7 +712,7 @@ final class ReplicatedChronicleMap, return; } - // skip the is deleted flag + // read the is deleted flag boolean wasDeleted = entry.readBoolean(); if (!wasDeleted) { @@ -721,9 +724,35 @@ final class ReplicatedChronicleMap, entry.writeLong(timestamp); assert identifier > 0; entry.writeByte(identifier); + // was deleted entry.writeBoolean(true); + + long valuePosition = entry.position(); + + if (hasAttachedEventListeners() && !wasDeleted) { + entry.position(keyPosition); + K key = null; + ThreadLocalCopies keyCopies = null; + + if (hasAttachedEventListeners()) { + keyCopies = keyInteropProvider.getCopies(null); + BytesReader keyReader = keyReaderProvider.get(keyCopies, originalKeyReader); + key = keyReader.read(entry, keySize); + } + + entry.position(valuePosition); + ThreadLocalCopies valueCopies = valueReaderProvider.getCopies(keyCopies); + BytesReader valueReader = valueReaderProvider.get(valueCopies, + originalValueReader); + V value = valueReader.read(entry, readValueSize(entry)); + + notifyRemoteRemove(offset, key, value, pos); + } + + return; } + // key is not found if (LOG.isDebugEnabled()) LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() + @@ -734,9 +763,20 @@ final class ReplicatedChronicleMap, } + void notifyRemoteRemove(long offset, K key, V value, final long pos) { + if (eventListener() != MapEventListeners.NOP) { + MultiStoreBytes b = acquireTmpBytes(); + b.storePositionAndSize(bytes, offset, entrySize); + attachedEventListener().onRemove(ReplicatedChronicleMap.this, b, metaDataBytes, + key, value, pos, this); + } + } + + void notifyRemotePut(long offset, boolean added, final long pos, long keySize, final Bytes inBytes, final long valueSize, final long valuePosition, - final long keyPosition, final V replacedValue) { + final long keyPosition, final V replacedValue, BytesReader valueReader, + BytesReader keyReader) { if (eventListener() != MapEventListeners.NOP) { final MultiStoreBytes b = acquireTmpBytes(); b.storePositionAndSize(bytes, offset, entrySize); @@ -744,16 +784,13 @@ final class ReplicatedChronicleMap, inBytes.limit(inBytes.capacity()); // deserialize the key - final ThreadLocalCopies copies = keyReaderProvider.getCopies(null); inBytes.position(keyPosition); - final K key = keyReaderProvider.get(copies, originalKeyReader).read(inBytes, keySize); + final K key = keyReader.read(inBytes, keySize); // deserialize the value - final ThreadLocalCopies copiesV = valueReaderProvider.getCopies(copies); - inBytes.limit(inBytes.capacity()); inBytes.position(valuePosition); - final V value = valueReaderProvider.get(copiesV, originalValueReader).read(inBytes, valueSize); + final V value = valueReader.read(inBytes, valueSize); attachedEventListener().onPut(ReplicatedChronicleMap.this, b, metaDataBytes, added, key, replacedValue, value, pos, this); @@ -769,17 +806,23 @@ final class ReplicatedChronicleMap, long valuePos, long valueLimit) { + BytesReader valueReader = null; + BytesReader keyReader = null; + + if (hasAttachedEventListeners()) { + ThreadLocalCopies keyCopies = keyInteropProvider.getCopies(null); + keyReader = keyReaderProvider.get(keyCopies, originalKeyReader); + + ThreadLocalCopies valueCopies = valueReaderProvider.getCopies(keyCopies); + valueReader = valueReaderProvider.get(valueCopies, originalValueReader); + } + writeLock(); try { // inBytes position and limit correspond to the key final long keySize = inBytes.remaining(); V replacedValue = null; - ThreadLocalCopies keyCopies = keyInteropProvider.getCopies(null); - BytesReader keyReader = keyReaderProvider.get(keyCopies, originalKeyReader); - - ThreadLocalCopies valueCopies = valueReaderProvider.getCopies(keyCopies); - BytesReader valueReader = valueReaderProvider.get(valueCopies, originalValueReader); hashLookupLiveAndDeleted.startSearch(hash2); for (long pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0L; ) { @@ -791,8 +834,6 @@ final class ReplicatedChronicleMap, continue; // key is found - - entry.skip(keySize); final long timeStampPos = entry.positionAddr(); @@ -819,7 +860,7 @@ final class ReplicatedChronicleMap, inBytes.position(valuePos); // if we have event listeners attached we have to read back the old value - if (hasAttachedEventListener()) + if (hasAttachedEventListeners()) replacedValue = valueReader.read(entry, replacedValueSize, null); putValue(pos, offset, entry, valueSizePos, entryEndAddr, null, null, inBytes, @@ -831,7 +872,7 @@ final class ReplicatedChronicleMap, incrementSize(); } - if (hasAttachedEventListener()) { + if (hasAttachedEventListeners()) { // this is just like notifyPut, but it won't trigger the modification iterators notifyRemotePut(offset, @@ -842,7 +883,9 @@ final class ReplicatedChronicleMap, valueLimit - valuePos, valuePos, keyPosition, - replacedValue + replacedValue, + valueReader, + keyReader ); } @@ -882,7 +925,7 @@ final class ReplicatedChronicleMap, // this is just like notifyPut, but it won't trigger the modification iterators notifyRemotePut(offset, true, posFromOffset(offset), keySize, inBytes, valueSize, valuePos, - kPosition, null); + kPosition, null, valueReader, keyReader); } finally { writeUnlock(); diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java index 0bf456639..d2a070ef5 100644 --- a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java +++ b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java @@ -37,7 +37,6 @@ import static net.openhft.chronicle.hash.replication.TcpConfig.forSendingNode; * * @author Rob Austin. */ - public class EventListenerTestWithTCPSocketReplication { @@ -50,6 +49,9 @@ public class EventListenerTestWithTCPSocketReplication { final AtomicReference valueRef = new AtomicReference(); final AtomicBoolean putWasCalled = new AtomicBoolean(false); + final AtomicBoolean wasRemoved = new AtomicBoolean(false); + final AtomicReference valueRemoved = new AtomicReference(); + final MapEventListener> eventListener = new MapEventListener>() { private static final long serialVersionUID = 1L; @@ -62,6 +64,12 @@ public class EventListenerTestWithTCPSocketReplication { valueRef.set(value); replacedValueRef.set(replacedValue); } + + @Override + public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes, Integer key, CharSequence value) { + wasRemoved.set(true); + valueRemoved.set(value); + } }; @@ -75,13 +83,12 @@ public class EventListenerTestWithTCPSocketReplication { e.printStackTrace(); } } - + wasRemoved.set(false); + valueRemoved.set(null); putWasCalled.set(false); valueRef.set(null); replacedValueRef.set(null); keyRef.set(null); - - } @@ -123,7 +130,7 @@ public class EventListenerTestWithTCPSocketReplication { @Test - public void testAddedOriginalValueToBeOverWritten() throws IOException, InterruptedException { + public void testRemoteNotifyPutLargerValue() throws IOException, InterruptedException { // ---------- SERVER1 1 ---------- { @@ -149,9 +156,6 @@ public class EventListenerTestWithTCPSocketReplication { map1.put(5, "EXAMPLE"); waitTillReplicated(); - Assert.assertEquals("Maps should be equal", map1, map2); - Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); - Assert.assertTrue("Map 2 should not be empty", !map2.isEmpty()); Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); Assert.assertEquals(Integer.valueOf(5), keyRef.get()); @@ -161,9 +165,8 @@ public class EventListenerTestWithTCPSocketReplication { } - @Test - public void testAlreadyAddedWithSmallInitialValue() throws IOException, InterruptedException { + public void testRemoteNotifyPutSmallerValue() throws IOException, InterruptedException { // ---------- SERVER1 1 ---------- @@ -188,9 +191,6 @@ public class EventListenerTestWithTCPSocketReplication { waitTillReplicated(); - Assert.assertEquals("Maps should be equal", map1, map2); - Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); - Assert.assertTrue("Map 2 should not be empty", !map2.isEmpty()); Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); Assert.assertEquals(Integer.valueOf(5), keyRef.get()); @@ -199,6 +199,46 @@ public class EventListenerTestWithTCPSocketReplication { } + + @Test + public void testRemoteNotifyRemove() throws IOException, InterruptedException { + + // ---------- SERVER1 1 ---------- + { + + // we connect the maps via a TCP socket connection on port 8077 + + ChronicleMapBuilder map1Builder = serverBuilder(); + + map1 = map1Builder.create(); + + + } + + // ---------- SERVER2 2 on the same server as ---------- + { + map2 = clientBuilder().create(); + map2.put(5, "WILL_GET_REMOVED"); + } + + // its we have to wait here for this entry to be removed - as removing an entry that you dont have + // on map 1 will have no effect + waitTillReplicated(); + + // no more puts after this point + putWasCalled.set(false); + + // we will stores some data into one map here + map1.remove(5); + + waitTillReplicated(); + + Assert.assertTrue(!putWasCalled.get()); + Assert.assertTrue(wasRemoved.get()); + Assert.assertEquals("WILL_GET_REMOVED", valueRemoved.get()); + Assert.assertEquals(null, map2.get(5)); + } + private ChronicleMapBuilder clientBuilder() { TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); return ChronicleMapBuilder.of(Integer.class, CharSequence.class)