diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 33296bd79..126233fac 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -41,7 +41,6 @@ import static java.lang.Math.min; import static java.nio.ByteBuffer.wrap; import static net.openhft.chronicle.map.Replica.EntryExternalizable; import static net.openhft.chronicle.map.Replica.ModificationIterator; -import static net.openhft.chronicle.map.Replica.ModificationNotifier.NOP; /** * @author Rob Austin. @@ -150,6 +149,8 @@ public final class ChannelProvider implements Closeable { } } }; + + private final byte localIdentifier; final Replica asReplica = new Replica() { @Override @@ -159,7 +160,8 @@ public final class ChannelProvider implements Closeable { @Override public ModificationIterator acquireModificationIterator( - final byte remoteIdentifier, final ModificationNotifier notifier) { + final byte remoteIdentifier) { + channelDataLock.writeLock().lock(); try { final ModificationIterator result = modificationIterator.get(remoteIdentifier); @@ -167,6 +169,9 @@ public final class ChannelProvider implements Closeable { return result; final ModificationIterator result0 = new ModificationIterator() { + + volatile Replica.ModificationNotifier notifier0; + @Override public boolean hasNext() { channelDataReadLock(); @@ -176,7 +181,7 @@ public final class ChannelProvider implements Closeable { for (Replica chronicleChannel : chronicleChannelMap.values()) { final ModificationIterator modificationIterator = chronicleChannel.acquireModificationIterator( - remoteIdentifier, notifier); + remoteIdentifier); if (modificationIterator.hasNext()) return true; } @@ -191,12 +196,11 @@ public final class ChannelProvider implements Closeable { final int na) { channelDataReadLock(); try { - for (Map.Entry chronicleChannel : chronicleChannelMap.entrySet()) { final ModificationIterator modificationIterator = chronicleChannel.getValue() - .acquireModificationIterator(remoteIdentifier, notifier); + .acquireModificationIterator(remoteIdentifier); if (modificationIterator .nextEntry(callback, chronicleChannel.getKey())) return true; @@ -213,14 +217,24 @@ public final class ChannelProvider implements Closeable { try { for (Replica chronicleChannel : chronicleChannelMap.values()) { - chronicleChannel.acquireModificationIterator(remoteIdentifier, notifier) + chronicleChannel.acquireModificationIterator(remoteIdentifier) .dirtyEntries(fromTimeStamp); - notifier.onChange(); + notifier0.onChange(); } } finally { channelDataLock.readLock().unlock(); } } + + @Override + public void setModificationNotifier( + @NotNull ModificationNotifier modificationNotifier) { + for (Replica chronicleChannel : chronicleChannelMap.values()) { + chronicleChannel.acquireModificationIterator(remoteIdentifier) + .setModificationNotifier(modificationNotifier); + } + notifier0 = modificationNotifier; + } }; modificationIterator.set((int) remoteIdentifier, result0); @@ -344,7 +358,7 @@ public final class ChannelProvider implements Closeable { // this could be null if one node has a chronicle channel before the other if (chronicleChannels[chronicleChannel] != null) { - chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier, NOP) + chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier) .dirtyEntries(lastModificationTime); } } @@ -448,14 +462,16 @@ public final class ChannelProvider implements Closeable { @Override public ModificationIterator acquireModificationIterator( - final byte remoteIdentifier, final ModificationNotifier modificationNotifier) { + final byte remoteIdentifier) { final ModificationIterator result = systemModificationIterator.get(remoteIdentifier); if (result != null) return result; final PayloadProvider iterator = new PayloadProvider() { + final Queue payloads = new LinkedTransferQueue(); + ModificationNotifier modificationNotifier0; @Override public boolean hasNext() { @@ -476,6 +492,11 @@ public final class ChannelProvider implements Closeable { // do nothing } + @Override + public void setModificationNotifier(@NotNull ModificationNotifier modificationNotifier) { + modificationNotifier0 = modificationNotifier; + } + @Override public void addPayload(Bytes bytes) { if (bytes.remaining() == 0) @@ -483,7 +504,7 @@ public final class ChannelProvider implements Closeable { payloads.add(bytes); // notifies that a change has been made, this will nudge the OP_WRITE // selector to push this update out over the nio socket - modificationNotifier.onChange(); + modificationNotifier0.onChange(); } }; diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 8b2ae2b07..9c504bb1f 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -51,14 +51,11 @@ public interface Replica extends Closeable { * remote node, this weak associated is bound using the {@code identifier}. * * @param remoteIdentifier the identifier of the remote node - * @param modificationNotifier called when ever there is a change applied to the modification - * iterator * @return the ModificationIterator dedicated for replication to the remote node with the given * identifier * @see #identifier() */ - ModificationIterator acquireModificationIterator(byte remoteIdentifier, - ModificationNotifier modificationNotifier); + ModificationIterator acquireModificationIterator(byte remoteIdentifier); /** * Returns the timestamp of the last change from the specified remote node, already replicated @@ -76,11 +73,6 @@ public interface Replica extends Closeable { * notifies when there is a changed to the modification iterator */ interface ModificationNotifier { - public static final ModificationNotifier NOP = new ModificationNotifier() { - @Override - public void onChange() { - } - }; /** * called when ever there is a change applied to the modification iterator @@ -120,6 +112,14 @@ public interface Replica extends Closeable { * @param fromTimeStamp the timestamp from which all entries should be dirty */ void dirtyEntries(long fromTimeStamp); + + /** + * the {@code modificationNotifier} is called when ever there is a change applied to the + * modification iterator + * + * @param modificationNotifier gets notified when a change occurs + */ + void setModificationNotifier(@NotNull final ModificationNotifier modificationNotifier); } /** diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index acfb10686..995281ad2 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -242,8 +242,7 @@ public class ReplicatedChronicleMap 0L; next = modIterSet.nextSetBit(next + 1L)) { try { - modificationIterators.get((int) next).raiseChange(segmentIndex, pos); + acquireModificationIterator((byte) next).raiseChange(segmentIndex, pos); } catch (Exception e) { LOG.error("", e); } @@ -282,7 +281,7 @@ public class ReplicatedChronicleMap 0L; next = modIterSet.nextSetBit(next + 1L)) { try { - modificationIterators.get((int) next).dropChange(segmentIndex, pos); + acquireModificationIterator((byte) next).dropChange(segmentIndex, pos); } catch (Exception e) { LOG.error("", e); } @@ -293,7 +292,8 @@ public class ReplicatedChronicleMap 0L; next = modIterSet.nextSetBit(next + 1L)) { try { - ModificationIterator modificationIterator = modificationIterators.get((int) next); + ModificationIterator modificationIterator = + acquireModificationIterator((byte) next); if (modificationIterator.dropChange(segmentIndex, oldPos)) modificationIterator.raiseChange(segmentIndex, newPos); } catch (Exception e) { @@ -483,7 +483,7 @@ public class ReplicatedChronicleMap extends AbstractChannelReplicator impleme "remoteIdentifier=" + remoteIdentifier); attached.remoteModificationIterator = - replica.acquireModificationIterator(remoteIdentifier, attached); + replica.acquireModificationIterator(remoteIdentifier); + + attached.remoteModificationIterator.setModificationNotifier(attached); writer.writeRemoteBootstrapTimestamp(replica.lastModificationTime(remoteIdentifier)); diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 6f42ecca3..275cb5a86 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -49,8 +49,8 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi super(replicationConfig, replica.identifier()); Replica.ModificationIterator modificationIterator = replica.acquireModificationIterator( - ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this); - + ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID); + modificationIterator.setModificationNotifier(this); setReader(new UdpSocketChannelEntryReader(replicationConfig.udpBufferSize(), entryExternalizable)); setWriter(new UdpSocketChannelEntryWriter(replicationConfig.udpBufferSize(), diff --git a/src/test/java/net/openhft/chronicle/TestReplication.java b/src/test/java/net/openhft/chronicle/TestReplication.java index 1763f2972..2290061b7 100644 --- a/src/test/java/net/openhft/chronicle/TestReplication.java +++ b/src/test/java/net/openhft/chronicle/TestReplication.java @@ -67,50 +67,6 @@ public class TestReplication { private ReplicationHub hubOnServer1; - @Test - public void testAllDataGetsReplicated2() throws InterruptedException, IOException, ExecutionException { - - try - // server 1 with identifier = 1 - { - ChronicleMapBuilder builder = - ChronicleMapBuilder.of(Short.class, Short.class) - .entries(1000); - - byte identifier = (byte) 1; - - TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig - .of(8086); - - hubOnServer1 = ReplicationHub.builder() - .tcpTransportAndNetwork(tcpConfig) - .maxNumberOfChannels(SIZE) - .createWithId(identifier); - - for (short channel1 = 3; channel1 < SIZE; channel1++) { - ChronicleMap map = builder.instance() - .replicatedViaChannel(hubOnServer1.createChannel(channel1)).create(); - - for (int i = 0; i < 10; i++) { - map.put((short) i, (short) i); - } - - maps.put(channel1, map); - - System.out.println("" + channel1); - - - } - - } catch (Exception e) { - e.printStackTrace(); - } - - for (; ; ) { - Thread.sleep(1000); - } - - } } diff --git a/src/test/java/net/openhft/chronicle/map/ReplicateProxyMapUpdatedViaSharedMemoryTest.java b/src/test/java/net/openhft/chronicle/map/ReplicateProxyMapUpdatedViaSharedMemoryTest.java new file mode 100644 index 000000000..2ebb3331e --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/ReplicateProxyMapUpdatedViaSharedMemoryTest.java @@ -0,0 +1,64 @@ +package net.openhft.chronicle.map; + +import org.junit.Assert; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; + +/** + * Created by khoxsey on 5/22/15. + */ +public class ReplicateProxyMapUpdatedViaSharedMemoryTest { + + /** + * test 2 maps on one server ( one that replicates vai TCP/IP and one that just shares memory ) + * can connect to a remote map + * connect to a remote map + * @throws IOException + * @throws InterruptedException + */ + @Test + public void testReplicateProxyMapUpdatedViaSharedMemory() throws IOException, InterruptedException { + + ChronicleMap server1 = null; + ChronicleMap server2 = null; + ChronicleMap replicationMap = null; + + { + final File tempFile = File.createTempFile("temp", "chron"); + + server1 = ChronicleMapBuilder.of(String.class, String.class) + .replication((byte) 1).createPersistedTo(tempFile); + + TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(7077); + replicationMap = ChronicleMapBuilder.of(String.class, String.class) + .replication((byte) 1, serverConfig).createPersistedTo(tempFile); + } + + { + TcpTransportAndNetworkConfig server2Config = + TcpTransportAndNetworkConfig.of(6161, new InetSocketAddress("localhost", 7077)); + + server2 = ChronicleMapBuilder.of(String.class, String.class) + .replication((byte) 2, server2Config).create(); + } + + final String expected = "value"; + Thread.sleep(1000); + server1.put("key", expected); + + String actual; + do { + actual = server2.get("key"); + } while (actual == null); + + Assert.assertEquals(expected, actual); + + server1.close(); + server2.close(); + replicationMap.close(); + } +}