mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
HCOLL-329 fixing the bootstrapping timestamp issue
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 src/main/java/net/openhft/chronicle/map/SharedSegment.java src/main/java/net/openhft/chronicle/map/TcpReplicator.java src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
This commit is contained in:
Rob Austin
leventov
@@ -484,7 +484,6 @@ abstract class AbstractChannelReplicator implements Closeable {
|
||||
out = result;
|
||||
|
||||
assert out.capacity() == in.capacity();
|
||||
|
||||
assert out.capacity() == size;
|
||||
assert out.capacity() == in.capacity();
|
||||
assert in.limit() == in.capacity();
|
||||
@@ -498,7 +497,9 @@ abstract class AbstractChannelReplicator implements Closeable {
|
||||
|
||||
|
||||
@Override
|
||||
public boolean onEntry(final Bytes entry, final int chronicleId) {
|
||||
public boolean onEntry(@NotNull final Bytes entry,
|
||||
final int chronicleId,
|
||||
final long bootstrapTime) {
|
||||
|
||||
long startOfEntry = entry.position();
|
||||
long pos0 = in.position();
|
||||
@@ -514,7 +515,7 @@ abstract class AbstractChannelReplicator implements Closeable {
|
||||
|
||||
start = in.position();
|
||||
|
||||
externalizable.writeExternalEntry(entry, in, chronicleId);
|
||||
externalizable.writeExternalEntry(entry, in, chronicleId, bootstrapTime);
|
||||
|
||||
if (in.position() == start) {
|
||||
in.position(pos0);
|
||||
@@ -558,7 +559,8 @@ abstract class AbstractChannelReplicator implements Closeable {
|
||||
|
||||
in.position(pos0);
|
||||
entry.position(startOfEntry);
|
||||
return onEntry(entry, chronicleId);
|
||||
|
||||
return onEntry(entry, chronicleId, bootstrapTime);
|
||||
} else
|
||||
throw e;
|
||||
|
||||
|
||||
@@ -18,7 +18,10 @@
|
||||
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.chronicle.hash.replication.*;
|
||||
import net.openhft.chronicle.hash.replication.ReplicableEntry;
|
||||
import net.openhft.chronicle.hash.replication.ReplicationHub;
|
||||
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
|
||||
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
|
||||
import net.openhft.lang.collection.DirectBitSet;
|
||||
import net.openhft.lang.collection.SingleThreadedDirectBitSet;
|
||||
import net.openhft.lang.io.ByteBufferBytes;
|
||||
@@ -111,7 +114,6 @@ public final class ChannelProvider implements Closeable {
|
||||
|
||||
/**
|
||||
* writes the entry to the chronicle channel provided
|
||||
*
|
||||
* @param entry the byte location of the entry to be stored
|
||||
* @param destination a buffer the entry will be written to, the segment may reject
|
||||
* this operation and add zeroBytes, if the identifier in the entry
|
||||
@@ -119,21 +121,21 @@ public final class ChannelProvider implements Closeable {
|
||||
* @param chronicleChannel used in cluster into identify the canonical map or queue
|
||||
*/
|
||||
@Override
|
||||
public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
|
||||
int chronicleChannel) {
|
||||
public void writeExternalEntry(@NotNull Bytes entry,
|
||||
@NotNull Bytes destination,
|
||||
int chronicleChannel, long bootstrapTime) {
|
||||
channelDataReadLock();
|
||||
try {
|
||||
destination.writeStopBit(chronicleChannel);
|
||||
channelEntryExternalizables[chronicleChannel]
|
||||
.writeExternalEntry(entry, destination, chronicleChannel);
|
||||
.writeExternalEntry(entry, destination, chronicleChannel, bootstrapTime);
|
||||
} finally {
|
||||
channelDataLock.readLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readExternalEntry(
|
||||
@NotNull Bytes source) {
|
||||
public void readExternalEntry(@NotNull Bytes source) {
|
||||
channelDataReadLock();
|
||||
try {
|
||||
final int chronicleId = (int) source.readStopBit();
|
||||
@@ -263,6 +265,20 @@ public final class ChannelProvider implements Closeable {
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setLastModificationTime(byte identifier, long timestamp) {
|
||||
channelDataReadLock();
|
||||
try {
|
||||
// not including the SystemQueue at index 0
|
||||
for (Replica channel : chronicleChannelMap.values()) {
|
||||
channel.setLastModificationTime(identifier, timestamp);
|
||||
}
|
||||
|
||||
} finally {
|
||||
channelDataLock.readLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
ChannelProvider.this.close();
|
||||
@@ -483,7 +499,7 @@ public final class ChannelProvider implements Closeable {
|
||||
final Bytes bytes = payloads.poll();
|
||||
if (bytes == null)
|
||||
return false;
|
||||
callback.onEntry(bytes, 0);
|
||||
callback.onEntry(bytes, 0, System.currentTimeMillis());
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -519,6 +535,11 @@ public final class ChannelProvider implements Closeable {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setLastModificationTime(byte identifier, long timestamp) {
|
||||
// do nothing
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
// do nothing
|
||||
@@ -537,7 +558,7 @@ public final class ChannelProvider implements Closeable {
|
||||
|
||||
@Override
|
||||
public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
|
||||
int na) {
|
||||
int na, long bootstrapTime) {
|
||||
destination.write(entry, entry.position(), entry.remaining());
|
||||
}
|
||||
|
||||
|
||||
@@ -69,6 +69,8 @@ public interface Replica extends Closeable {
|
||||
*/
|
||||
long lastModificationTime(byte remoteIdentifier);
|
||||
|
||||
void setLastModificationTime(byte identifier, long timestamp);
|
||||
|
||||
/**
|
||||
* notifies when there is a changed to the modification iterator
|
||||
*/
|
||||
@@ -156,8 +158,8 @@ public interface Replica extends Closeable {
|
||||
* match the maps local
|
||||
* @param chronicleId is the channel id used to identify the canonical map or queue
|
||||
*/
|
||||
void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, int
|
||||
chronicleId);
|
||||
void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
|
||||
int chronicleId, long bootstrapTime);
|
||||
|
||||
/**
|
||||
* The map implements this method to restore its contents. This method must read the values
|
||||
@@ -188,16 +190,19 @@ public interface Replica extends Closeable {
|
||||
* source node is not from one of our changes, WARNING even though we check the identifier
|
||||
* in the ModificationIterator the entry may have been updated.
|
||||
*/
|
||||
public abstract boolean onEntry(final Bytes entry, final int chronicleId);
|
||||
public abstract boolean onEntry(
|
||||
final Bytes entry, final int chronicleId, long bootstrapTime);
|
||||
|
||||
/**
|
||||
* Called just after {@link #onEntry(net.openhft.lang.io.Bytes, int)}. No-op by default.
|
||||
* Called just after {@link #onEntry(net.openhft.lang.io.Bytes, int, long)}.
|
||||
* No-op by default.
|
||||
*/
|
||||
public void onAfterEntry() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Called just before {@link #onEntry(net.openhft.lang.io.Bytes, int)}. No-op by default.
|
||||
* Called just before {@link #onEntry(net.openhft.lang.io.Bytes, int, long)}.
|
||||
* No-op by default.
|
||||
*/
|
||||
public void onBeforeEntry() {
|
||||
}
|
||||
|
||||
@@ -39,6 +39,8 @@ import java.io.IOException;
|
||||
import java.io.ObjectInputStream;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CopyOnWriteArraySet;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.atomic.AtomicReferenceArray;
|
||||
|
||||
import static net.openhft.lang.MemoryUnit.*;
|
||||
@@ -92,6 +94,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
private static final long serialVersionUID = 0L;
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ReplicatedChronicleMap.class);
|
||||
private static final long LAST_UPDATED_HEADER_SIZE = 128L * 8L;
|
||||
private static final int SIZE_OF_BOOTSTRAP_TIME_STAMP = 8;
|
||||
|
||||
public final TimeProvider timeProvider;
|
||||
private final byte localIdentifier;
|
||||
transient Set<Closeable> closeables;
|
||||
@@ -266,11 +270,11 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
}
|
||||
}
|
||||
|
||||
public void raiseChange(long segmentIndex, long pos) {
|
||||
public void raiseChange(long segmentIndex, long pos, long timestamp) {
|
||||
for (long next = modIterSet.nextSetBit(0L); next > 0L;
|
||||
next = modIterSet.nextSetBit(next + 1L)) {
|
||||
try {
|
||||
acquireModificationIterator((byte) next).raiseChange(segmentIndex, pos);
|
||||
acquireModificationIterator((byte) next).raiseChange(segmentIndex, pos, timestamp);
|
||||
} catch (Exception e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
@@ -288,14 +292,14 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
}
|
||||
}
|
||||
|
||||
public void moveChange(long segmentIndex, long oldPos, long newPos) {
|
||||
public void moveChange(long segmentIndex, long oldPos, long newPos, long timestamp) {
|
||||
for (long next = modIterSet.nextSetBit(0L); next > 0L;
|
||||
next = modIterSet.nextSetBit(next + 1L)) {
|
||||
try {
|
||||
ModificationIterator modificationIterator =
|
||||
acquireModificationIterator((byte) next);
|
||||
if (modificationIterator.dropChange(segmentIndex, oldPos))
|
||||
modificationIterator.raiseChange(segmentIndex, newPos);
|
||||
modificationIterator.raiseChange(segmentIndex, newPos, timestamp);
|
||||
} catch (Exception e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
@@ -332,7 +336,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
// make a byte buffer
|
||||
assert result < Integer.MAX_VALUE;
|
||||
|
||||
return (int) result;
|
||||
return (int) result + SIZE_OF_BOOTSTRAP_TIME_STAMP;
|
||||
} finally {
|
||||
entry.position(start);
|
||||
}
|
||||
@@ -346,7 +350,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
@Override
|
||||
public void writeExternalEntry(@NotNull Bytes entry,
|
||||
@NotNull Bytes destination,
|
||||
int chronicleId) {
|
||||
int chronicleId,
|
||||
long bootstrapTime) {
|
||||
|
||||
final long keySize = keySizeMarshaller.readSize(entry);
|
||||
|
||||
@@ -369,7 +374,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
}
|
||||
|
||||
final long valuePosition = entry.position();
|
||||
|
||||
destination.writeLong(bootstrapTime);
|
||||
keySizeMarshaller.writeSize(destination, keySize);
|
||||
valueSizeMarshaller.writeSize(destination, valueSize);
|
||||
destination.writeStopBit(timeStamp);
|
||||
@@ -492,6 +497,12 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
private final int segmentIndexShift;
|
||||
private final long posMask;
|
||||
|
||||
// when a bootstrap is send this is the time stamp that the client will bootstrap up to
|
||||
// if it is set as ZERO then the onPut() will set it to the current time, once the
|
||||
// consumer has cycled through the bit set the timestamp will be set back to zero.
|
||||
private AtomicLong bootStrapTimeStamp = new AtomicLong();
|
||||
private long lastBootStrapTimeStamp = timeProvider.currentTime();
|
||||
|
||||
// records the current position of the cursor in the bitset
|
||||
private volatile long position = -1L;
|
||||
|
||||
@@ -524,10 +535,15 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
return (segmentIndex << segmentIndexShift) | pos;
|
||||
}
|
||||
|
||||
void raiseChange(long segmentIndex, long pos) {
|
||||
void raiseChange(long segmentIndex, long pos, long timestamp) {
|
||||
LOG.debug("raise change: id {}, segment {}, pos {}",
|
||||
localIdentifier, segmentIndex, pos);
|
||||
changesForUpdates.set(combine(segmentIndex, pos));
|
||||
assert timestamp > timeProvider.currentTime() - TimeUnit.SECONDS.toMillis(1) &&
|
||||
timestamp <= timeProvider.currentTime() : "timeStamp=" + timestamp + ", " +
|
||||
"currentTime=" + timeProvider.currentTime();
|
||||
// todo improve this - use the timestamp from the entry its self
|
||||
bootStrapTimeStamp.compareAndSet(0, timestamp);
|
||||
if (modificationNotifier != null)
|
||||
modificationNotifier.onChange();
|
||||
}
|
||||
@@ -568,6 +584,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
this.position = NOT_FOUND;
|
||||
return false;
|
||||
}
|
||||
|
||||
bootStrapTimeStamp.set(0);
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -594,7 +612,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
// process it later, by NOT clearing the changes.clear(position)
|
||||
context.entryBytes().limit(context.valueOffset() + context.valueSize());
|
||||
context.entryBytes().position(context.keySizeOffset());
|
||||
boolean success = entryCallback.onEntry(context.entryBytes(), chronicleId);
|
||||
boolean success = entryCallback.onEntry(
|
||||
context.entryBytes(), chronicleId, bootStrapTimeStamp());
|
||||
entryCallback.onAfterEntry();
|
||||
|
||||
if (success)
|
||||
@@ -609,6 +628,18 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the timestamp that the remote client should bootstrap from when there has been a
|
||||
* disconnection, this time maybe later than the message time as event are not send in
|
||||
* chronological order from the bit set.
|
||||
*/
|
||||
private long bootStrapTimeStamp() {
|
||||
final long timeStamp = bootStrapTimeStamp.get();
|
||||
long result = (timeStamp == 0) ? this.lastBootStrapTimeStamp : timeStamp;
|
||||
this.lastBootStrapTimeStamp = result;
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void dirtyEntries(long fromTimeStamp) {
|
||||
try (CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R, ?> c =
|
||||
@@ -636,7 +667,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
|
||||
if (re.originTimestamp() >= fromTimeStamp &&
|
||||
(!bootstrapOnlyLocalEntries ||
|
||||
re.originIdentifier() == localIdentifier)) {
|
||||
raiseChange(segmentIndex, c.pos());
|
||||
raiseChange(segmentIndex, c.pos(), c.timestamp());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1282,6 +1282,7 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
|
||||
*/
|
||||
class TcpSocketChannelEntryReader {
|
||||
public static final int HEADROOM = 1024;
|
||||
public static final int SIZE_OF_BOOTSTRAP_TIMESTAMP = 8;
|
||||
public long lastHeartBeatReceived = System.currentTimeMillis();
|
||||
ByteBuffer in;
|
||||
ByteBufferBytes out;
|
||||
@@ -1367,7 +1368,8 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
|
||||
|
||||
// if the buffer is too small to read this payload we will have to grow the
|
||||
// size of the buffer
|
||||
long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1;
|
||||
long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1 +
|
||||
SIZE_OF_BOOTSTRAP_TIMESTAMP;
|
||||
if (out.capacity() < requiredSize) {
|
||||
attached.entryReader.resizeBuffer(requiredSize + HEADROOM);
|
||||
}
|
||||
|
||||
@@ -46,7 +46,8 @@ public abstract class ReplicatedInput<K, V, R>
|
||||
this.replicatedInputBytes = replicatedInputBytes;
|
||||
replicatedInputStore.setBytes(replicatedInputBytes);
|
||||
}
|
||||
|
||||
|
||||
@Stage("ReplicationInput") public long bootstrapTimestamp;
|
||||
@Stage("ReplicationInput") public long riKeySize = -1;
|
||||
@Stage("ReplicationInput") public long riValueSize;
|
||||
|
||||
@@ -60,6 +61,7 @@ public abstract class ReplicatedInput<K, V, R>
|
||||
|
||||
public void initReplicationInput(Bytes replicatedInputBytes) {
|
||||
initReplicatedInputBytes(replicatedInputBytes);
|
||||
bootstrapTimestamp = replicatedInputBytes.readLong();
|
||||
riKeySize = mh.m().keySizeMarshaller.readSize(replicatedInputBytes);
|
||||
riValueSize = mh.m().valueSizeMarshaller.readSize(replicatedInputBytes);
|
||||
|
||||
@@ -79,6 +81,8 @@ public abstract class ReplicatedInput<K, V, R>
|
||||
return;
|
||||
}
|
||||
|
||||
mh.m().setLastModificationTime(riId, bootstrapTimestamp);
|
||||
|
||||
q.initInputKey(replicatedInputKeyBytesValue);
|
||||
|
||||
boolean debugEnabled = lh.LOG.isDebugEnabled();
|
||||
@@ -90,7 +94,6 @@ public abstract class ReplicatedInput<K, V, R>
|
||||
mh.m().identifier(), riId, q.inputKey);
|
||||
}
|
||||
mh.m().remoteOperations.remove(this);
|
||||
mh.m().setLastModificationTime(riId, riTimestamp);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -103,7 +106,7 @@ public abstract class ReplicatedInput<K, V, R>
|
||||
|
||||
|
||||
mh.m().remoteOperations.put(this, replicatedInputValueBytesValue);
|
||||
mh.m().setLastModificationTime(riId, riTimestamp);
|
||||
|
||||
|
||||
if (debugEnabled) {
|
||||
lh.LOG.debug(message + "value=" + replicatedInputValueBytesValue + ")");
|
||||
|
||||
+2
-2
@@ -51,7 +51,7 @@ public abstract class ReplicationUpdate<K> implements RemoteOperationContext<K>
|
||||
}
|
||||
|
||||
public void moveChange(long oldPos, long newPos) {
|
||||
mh.m().moveChange(s.segmentIndex, oldPos, newPos);
|
||||
mh.m().moveChange(s.segmentIndex, oldPos, newPos, e.timestamp());
|
||||
}
|
||||
|
||||
public void updateChange() {
|
||||
@@ -61,7 +61,7 @@ public abstract class ReplicationUpdate<K> implements RemoteOperationContext<K>
|
||||
}
|
||||
|
||||
public void raiseChange() {
|
||||
mh.m().raiseChange(s.segmentIndex, e.pos);
|
||||
mh.m().raiseChange(s.segmentIndex, e.pos, e.timestamp());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
Reference in New Issue
Block a user