mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
HCOLL-134 added the following method - bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries)
This commit is contained in:
@@ -142,6 +142,7 @@ abstract class AbstractChronicleMapBuilder<K, V,
|
||||
private PrepareValueBytes<K, V> prepareValueBytes = null;
|
||||
|
||||
private SingleChronicleHashReplication singleHashReplication = null;
|
||||
public boolean bootstapOnlyLocalEntries = false;
|
||||
|
||||
AbstractChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
|
||||
keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
|
||||
@@ -962,6 +963,13 @@ abstract class AbstractChronicleMapBuilder<K, V,
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public ChronicleMapBuilderI<K, V> bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries) {
|
||||
this.bootstapOnlyLocalEntries = bootstapOnlyLocalEntries;
|
||||
return this;
|
||||
}
|
||||
|
||||
public String name() {
|
||||
return this.name;
|
||||
}
|
||||
|
||||
@@ -402,9 +402,6 @@ final class ChannelProvider implements Closeable {
|
||||
|
||||
final PayloadProvider iterator = new PayloadProvider() {
|
||||
|
||||
|
||||
// todo it maybe possibel to replace this with a single atomic refernce,
|
||||
// todo this way no aditional objects will be created
|
||||
final Queue<Bytes> payloads = new LinkedTransferQueue<Bytes>();
|
||||
|
||||
@Override
|
||||
|
||||
@@ -492,7 +492,21 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
|
||||
|
||||
@Override
|
||||
public ChronicleMap<K, V> createPersistedTo(File file) throws IOException {
|
||||
return delegate.createPersistedTo(file);
|
||||
return delegate.createPersistedTo(file);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param bootstapOnlyLocalEntries if set to true - when a new node joins a TCP replication grid, the new
|
||||
* node will be populated with data, only for the nodes that created that
|
||||
* data. Otherwise, all the nodes will publish all the data they have (
|
||||
* potentially swamping the new node with duplicates ) however this does
|
||||
* guarantee that all the data is replicated over to the new node, and is
|
||||
* useful especially in the case that the originating node is not
|
||||
* currently running.
|
||||
*/
|
||||
public ChronicleMapBuilder<K, V> bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries) {
|
||||
delegate.bootstapOnlyLocalEntries(bootstapOnlyLocalEntries);
|
||||
return this;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -296,6 +296,17 @@ interface ChronicleMapBuilderI<K, V> extends Serializable {
|
||||
|
||||
ChronicleMapBuilderI<K, V> name(String name);
|
||||
|
||||
/**
|
||||
* @param bootstapOnlyLocalEntries if set to true - when a new node joins a TCP replication grid, the new
|
||||
* node will be populated with data, only for the nodes that created that
|
||||
* data. Otherwise, all the nodes will publish all the data they have (
|
||||
* potentially swamping the new node with duplicates ) however this does
|
||||
* guarantee that all the data is replicated over to the new node, and is
|
||||
* useful especially in the case that the originating node is not
|
||||
* currently running.
|
||||
*/
|
||||
ChronicleMapBuilderI<K, V> bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries);
|
||||
|
||||
String name();
|
||||
|
||||
|
||||
|
||||
@@ -248,6 +248,7 @@ public class NodeDiscovery {
|
||||
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
|
||||
.of(ourAddressAndPort.getPort(),
|
||||
toInetSocketCollection(knownHostPorts0))
|
||||
|
||||
.heartBeatInterval(1, SECONDS);
|
||||
|
||||
|
||||
@@ -257,6 +258,7 @@ public class NodeDiscovery {
|
||||
final ReplicationHub replicationHub = ReplicationHub.builder()
|
||||
.maxEntrySize(10 * 1024)
|
||||
.tcpTransportAndNetwork(tcpConfig)
|
||||
.remoteNodeValidator(remoteNodeValidator)
|
||||
.createWithId(identifier);
|
||||
|
||||
return new ReplicationHubFindByName(replicationHub);
|
||||
|
||||
@@ -163,6 +163,8 @@ final class OffHeapUpdatableChronicleMapBuilder<K, V>
|
||||
throw new UnsupportedOperationException("not supported for this combination of key/value type");
|
||||
}
|
||||
|
||||
|
||||
|
||||
@Override
|
||||
public OffHeapUpdatableChronicleMapBuilder<K, V> constantValueSizeBySample(V sampleValue) {
|
||||
throw new UnsupportedOperationException("not supported for this combination of key/value type");
|
||||
|
||||
@@ -97,6 +97,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
|
||||
private transient ModificationDelegator modificationDelegator;
|
||||
private transient long startOfModificationIterators;
|
||||
private boolean bootstapOnlyLocalEntries;
|
||||
|
||||
public ReplicatedChronicleMap(@NotNull AbstractChronicleMapBuilder<K, V, ?> builder,
|
||||
byte identifier)
|
||||
@@ -105,6 +106,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
this.timeProvider = builder.timeProvider();
|
||||
|
||||
this.localIdentifier = identifier;
|
||||
this.bootstapOnlyLocalEntries = builder.bootstapOnlyLocalEntries;
|
||||
|
||||
if (localIdentifier == -1) {
|
||||
throw new IllegalStateException("localIdentifier should not be -1");
|
||||
@@ -821,8 +823,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
* called from a remote node when it wishes to propagate a remove event
|
||||
*/
|
||||
void remotePut(@NotNull final Bytes inBytes, long hash2,
|
||||
final byte identifier, final long timestamp,
|
||||
long valuePos, long valueLimit) {
|
||||
final byte identifier, final long timestamp,
|
||||
long valuePos, long valueLimit) {
|
||||
|
||||
|
||||
BytesReader<V> valueReader = null;
|
||||
@@ -955,10 +957,10 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
return keySize == keySizeMarshaller.readSize(entry) && entry.startsWith(keyBytes);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Override
|
||||
V putWithoutLock(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V value, long hash2, boolean replaceIfPresent) {
|
||||
return putWithoutLock(copies, metaKeyInterop, keyInterop, key, value, hash2, replaceIfPresent,
|
||||
localIdentifier, timeProvider.currentTimeMillis());
|
||||
return putWithoutLock(copies, metaKeyInterop, keyInterop, key, value, hash2, replaceIfPresent,
|
||||
localIdentifier, timeProvider.currentTimeMillis());
|
||||
}
|
||||
|
||||
V put(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter,
|
||||
@@ -1231,8 +1233,16 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
anotherLookup.replace(hash, prevPos, pos);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @param timeStamp
|
||||
* @param callback
|
||||
* @param bootstapOnlyLocalEntries if true - only entries that have been create from this node will be
|
||||
* published during a bootstrap
|
||||
*/
|
||||
public void dirtyEntries(final long timeStamp,
|
||||
final ModificationIterator.EntryModifiableCallback callback) {
|
||||
final ModificationIterator.EntryModifiableCallback callback,
|
||||
final boolean bootstapOnlyLocalEntries) {
|
||||
|
||||
this.readLock();
|
||||
try {
|
||||
@@ -1247,8 +1257,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
|
||||
final long entryTimestamp = entry.readLong();
|
||||
|
||||
if (entryTimestamp >= timeStamp &&
|
||||
entry.readByte() == ReplicatedChronicleMap.this.identifier())
|
||||
if (entryTimestamp >= timeStamp && (!bootstapOnlyLocalEntries ||
|
||||
entry.readByte() == ReplicatedChronicleMap.this.identifier()))
|
||||
callback.set(index, pos);
|
||||
}
|
||||
});
|
||||
@@ -1655,7 +1665,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
// iterate over all the segments and mark bit in the modification iterator
|
||||
// that correspond to entries with an older timestamp
|
||||
for (final Segment segment : (Segment[]) segments) {
|
||||
segment.dirtyEntries(fromTimeStamp, entryModifiableCallback);
|
||||
segment.dirtyEntries(fromTimeStamp, entryModifiableCallback, bootstapOnlyLocalEntries);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -43,7 +43,7 @@ class ReplicationHubFindByName implements FindByName {
|
||||
|
||||
public static final int MAP_BY_NAME_CHANNEL = 1;
|
||||
|
||||
private final AtomicInteger nextFreeChannel = new AtomicInteger(1);
|
||||
private final AtomicInteger nextFreeChannel = new AtomicInteger(2);
|
||||
private final Map<String, ChronicleMapBuilderWithChannelId> map;
|
||||
private final ReplicationHub replicationHub;
|
||||
|
||||
@@ -73,7 +73,7 @@ class ReplicationHubFindByName implements FindByName {
|
||||
};
|
||||
|
||||
this.map = (Map) of(CharSequence.class, ChronicleMapBuilderWithChannelId.class)
|
||||
.entrySize(2500)
|
||||
.entrySize(3000)
|
||||
.entries(128)
|
||||
.eventListener(listener)
|
||||
.instance()
|
||||
|
||||
Reference in New Issue
Block a user