diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index e02f7c2df..6b7afb3f5 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -44,7 +44,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. @@ -153,6 +152,8 @@ final class ChannelProvider implements Closeable { } } }; + + private final byte localIdentifier; final Replica asReplica = new Replica() { @Override @@ -162,7 +163,8 @@ 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); @@ -170,6 +172,9 @@ final class ChannelProvider implements Closeable { return result; final ModificationIterator result0 = new ModificationIterator() { + + volatile Replica.ModificationNotifier notifier0; + @Override public boolean hasNext() { channelDataLock.readLock().lock(); @@ -179,7 +184,8 @@ final class ChannelProvider implements Closeable { for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { final ModificationIterator modificationIterator = chronicleChannelList.get(i).acquireModificationIterator( - remoteIdentifier, notifier); + remoteIdentifier); + if (modificationIterator.hasNext()) return true; } @@ -197,7 +203,8 @@ final class ChannelProvider implements Closeable { for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { Replica chronicleChannel = chronicleChannelList.get(i); final ModificationIterator modificationIterator = chronicleChannel - .acquireModificationIterator(remoteIdentifier, notifier); + .acquireModificationIterator(remoteIdentifier); + if (modificationIterator .nextEntry(callback, chronicleChannelIds.get(i))) return true; @@ -214,18 +221,30 @@ final class ChannelProvider implements Closeable { try { for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { try { - chronicleChannelList.get(i) - .acquireModificationIterator(remoteIdentifier, notifier) - .dirtyEntries(fromTimeStamp); + ModificationIterator mi = chronicleChannelList.get(i) + .acquireModificationIterator(remoteIdentifier + ); + + mi.dirtyEntries(fromTimeStamp); } catch (InterruptedException e) { throw new RuntimeException(e); } - notifier.onChange(); + notifier0.onChange(); } } finally { channelDataLock.readLock().unlock(); } } + + @Override + public void setModificationNotifier(@NotNull ModificationNotifier modificationNotifier) { + for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { + Replica chronicleChannel = chronicleChannelList.get(i); + chronicleChannel.acquireModificationIterator(remoteIdentifier) + .setModificationNotifier(modificationNotifier); + } + notifier0 = modificationNotifier; + } }; modificationIterator.set((int) remoteIdentifier, result0); @@ -340,7 +359,7 @@ final class ChannelProvider implements Closeable { // this could be null if one node has a chronicle channel before the other if (chronicleChannels[chronicleChannel] != null) { try { - chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier, NOP) + chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier) .dirtyEntries(lastModificationTime); } catch (InterruptedException e) { throw new RuntimeException(e); @@ -443,14 +462,16 @@ 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() { @@ -471,6 +492,11 @@ 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) @@ -478,7 +504,7 @@ 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 9ae880c6c..c8cc23777 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -50,15 +50,14 @@ interface Replica extends Closeable { * Gets (if it does not exist, creates) an instance of ModificationIterator associated with a * 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 + * @param remoteIdentifier the identifier of the remote node * @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,7 +75,7 @@ interface Replica extends Closeable { * notifies when there is a changed to the modification iterator */ interface ModificationNotifier { - public static final ModificationNotifier NOP = new ModificationNotifier() { + ModificationNotifier NOP = new ModificationNotifier() { @Override public void onChange() { } @@ -120,6 +119,14 @@ interface Replica extends Closeable { * @param fromTimeStamp the timestamp from which all entries should be dirty */ void dirtyEntries(long fromTimeStamp) throws InterruptedException; + + /** + * 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 a5af8ed3a..9cfe79488 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -351,8 +351,7 @@ final class ReplicatedChronicleMap 0L; next = modIterSet.nextSetBit(next + 1L)) { try { - modificationIterators.get((int) next).onRemove(pos, segment); + acquireModificationIterator((byte) next).onRemove(pos, segment); } catch (Exception e) { LOG.error("", e); } @@ -426,7 +415,7 @@ final class ReplicatedChronicleMap 0L; next = modIterSet.nextSetBit(next + 1L)) { try { - modificationIterators.get((int) next).onRelocation(pos, segment); + acquireModificationIterator((byte) next).onRelocation(pos, segment); } catch (Exception e) { LOG.error("", e); } @@ -1403,7 +1392,8 @@ final class ReplicatedChronicleMap extends AbstractChannelReplicator implements Clo "remoteIdentifier=" + remoteIdentifier); attached.remoteModificationIterator = - replica.acquireModificationIterator(remoteIdentifier, attached); + replica.acquireModificationIterator(remoteIdentifier); + + attached.remoteModificationIterator.setModificationNotifier(attached); long timeStampOfLastMessage = replica.lastModificationTime(remoteIdentifier); writer.writeRemoteBootstrapTimestamp(timeStampOfLastMessage); diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 51e0432a2..7a9058159 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(); + } +}