Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

added in a fix for net.openhft.chronicle.map.ReplicateProxyMapUpdatedViaSharedMemoryTest.testReplicateProxyMapUpdatedViaSharedMemory

Browse Source
This commit is contained in:
Rob Austin
2015-05-28 15:34:27 +01:00
parent 163e8bbbe1
commit 2943d2ec6a
7 changed files with 139 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+37 -11
View File
@@ -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<Bytes> payloads = new LinkedTransferQueue<Bytes>();
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();
}
};
src/main/java/net/openhft/chronicle/map/Replica.java
+11 -4
View File
@@ -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);
}
/**
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+22 -19
View File
@@ -351,8 +351,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
@Override
public Replica.ModificationIterator acquireModificationIterator(
byte remoteIdentifier, @NotNull final ModificationNotifier modificationNotifier) {
public ModificationIterator acquireModificationIterator(byte remoteIdentifier) {
ModificationIterator modificationIterator = modificationIterators.get(remoteIdentifier);
if (modificationIterator != null)
return modificationIterator;
@@ -368,31 +367,21 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
modIterBitSetSizeInBytes());
final ModificationIterator newModificationIterator = new ModificationIterator(
bytes, modificationNotifier);
bytes);
modificationIterators.set(remoteIdentifier, newModificationIterator);
System.out.println("Thread=" + Thread.currentThread() + " acquireModificationIterator id=" + remoteIdentifier);
modIterSet.set(remoteIdentifier);
return newModificationIterator;
}
}
@Override
void onPut(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
for (long next = modIterSet.nextSetBit(0L); next > 0L;
next = modIterSet.nextSetBit(next + 1L)) {
try {
ModificationIterator modificationIterator;
modificationIterator = modificationIterators.get((int) next);
if (modificationIterator == null)
continue;
System.out.println("thread=" + Thread.currentThread() + " onPut() - acquireModificationIterator id=" + next);
modificationIterator.onPut(pos, segment);
acquireModificationIterator((byte) next).onPut(pos, segment);
} catch (Exception e) {
LOG.error("", e);
}
@@ -409,7 +398,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
for (long next = modIterSet.nextSetBit(0L); next > 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<K, KI, MKI extends MetaBytesInterop<K, ? supe
for (long next = modIterSet.nextSetBit(0L); next > 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<K, KI, MKI extends MetaBytesInterop<K, ? supe
* @author Rob Austin.
*/
class ModificationIterator implements Replica.ModificationIterator {
private final ModificationNotifier modificationNotifier;
private ModificationNotifier modificationNotifier;
private final ATSDirectBitSet changes;
private final int segmentIndexShift;
private final long posMask;
@@ -1423,13 +1413,26 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
*/
public ModificationIterator(@NotNull final Bytes bytes,
@NotNull final ModificationNotifier modificationNotifier) {
this.modificationNotifier = modificationNotifier;
setModificationNotifier(modificationNotifier);
long bitsPerSegment = bitsPerSegmentInModIterBitSet();
segmentIndexShift = Long.numberOfTrailingZeros(bitsPerSegment);
posMask = bitsPerSegment - 1L;
changes = new ATSDirectBitSet(bytes);
}
public void setModificationNotifier(ModificationNotifier modificationNotifier) {
this.modificationNotifier = modificationNotifier;
}
public ModificationIterator(@NotNull final Bytes bytes) {
long bitsPerSegment = bitsPerSegmentInModIterBitSet();
segmentIndexShift = Long.numberOfTrailingZeros(bitsPerSegment);
posMask = bitsPerSegment - 1L;
changes = new ATSDirectBitSet(bytes);
}
/**
* used to merge multiple segments and positions into a single index used by the bit map
*
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+3 -1
View File
@@ -641,7 +641,9 @@ final class TcpReplicator<K, V> 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);
src/main/java/net/openhft/chronicle/map/UdpReplicator.java
+2 -2
View File
@@ -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(),
src/test/java/net/openhft/chronicle/TestReplication.java
-44
View File
@@ -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<Short, Short> 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<Short, Short> 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);
}
}
}
src/test/java/net/openhft/chronicle/map/ReplicateProxyMapUpdatedViaSharedMemoryTest.java
+64
View File
@@ -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<String, String> server1 = null;
ChronicleMap<String, String> server2 = null;
ChronicleMap<String, String> 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();
}
}