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 
Conflicts:
	src/main/java/net/openhft/chronicle/map/ChannelProvider.java
	src/main/java/net/openhft/chronicle/map/Replica.java
	src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
This commit is contained in:
Rob Austin
2015-07-25 19:00:44 +03:00
committed by leventov
parent dedaa11c5f
commit e3fc5c9aa8
7 changed files with 126 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+31 -10
View File
@@ -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<Integer, Replica> 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<Bytes> payloads = new LinkedTransferQueue<Bytes>();
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();
}
};
src/main/java/net/openhft/chronicle/map/Replica.java
+9 -9
View File
@@ -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);
}
/**
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+17 -14
View File
@@ -242,8 +242,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
}
@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;
@@ -259,7 +258,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
modIterBitSetSizeInBytes());
final ModificationIterator newModificationIterator = new ModificationIterator(
bytes, modificationNotifier);
bytes);
modificationIterators.set(remoteIdentifier, newModificationIterator);
modIterSet.set(remoteIdentifier);
@@ -271,7 +270,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
for (long next = modIterSet.nextSetBit(0L); next > 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<K, KI, MKI extends MetaBytesInterop<K, ? sup
for (long next = modIterSet.nextSetBit(0L); next > 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<K, KI, MKI extends MetaBytesInterop<K, ? sup
for (long next = modIterSet.nextSetBit(0L); next > 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<K, KI, MKI extends MetaBytesInterop<K, ? sup
* @author Rob Austin.
*/
class ModificationIterator implements Replica.ModificationIterator {
private final ModificationNotifier modificationNotifier;
private ModificationNotifier modificationNotifier;
private final SingleThreadedDirectBitSet changesForUpdates;
// to getVolatile when reading changes bits, because we iterate when without lock.
// hardly this is needed on x86, probably on other architectures too.
@@ -495,13 +495,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
// records the current position of the cursor in the bitset
private volatile long position = -1L;
/**
* @param bytes the back the bitset, used to mark which entries have changed
* @param modificationNotifier called when ever there is a change applied
*/
public ModificationIterator(@NotNull final Bytes bytes,
@NotNull final ModificationNotifier modificationNotifier) {
this.modificationNotifier = modificationNotifier;
public ModificationIterator(@NotNull final Bytes bytes) {
long bitsPerSegment = bitsPerSegmentInModIterBitSet();
segmentIndexShift = Long.numberOfTrailingZeros(bitsPerSegment);
posMask = bitsPerSegment - 1L;
@@ -509,6 +503,15 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
changesForIteration = new ATSDirectBitSet(bytes);
}
public ModificationIterator(@NotNull final Bytes bytes, ModificationNotifier notifier) {
this(bytes);
setModificationNotifier(notifier);
}
public void setModificationNotifier(ModificationNotifier modificationNotifier) {
this.modificationNotifier = modificationNotifier;
}
/**
* 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
@@ -647,7 +647,9 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
"remoteIdentifier=" + remoteIdentifier);
attached.remoteModificationIterator =
replica.acquireModificationIterator(remoteIdentifier, attached);
replica.acquireModificationIterator(remoteIdentifier);
attached.remoteModificationIterator.setModificationNotifier(attached);
writer.writeRemoteBootstrapTimestamp(replica.lastModificationTime(remoteIdentifier));
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();
}
}