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

Reorder fields in replication message, to avoid writing value size, when deleted event is sent

Browse Source
This commit is contained in:
leventov
2015-12-17 21:25:47 +03:00
parent f0bcd784b7
commit 268ba7f569
2 changed files with 20 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+14 -21
View File
@@ -372,34 +372,25 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
ReplicableEntry entry, Bytes payload, @NotNull Bytes destination,
int chronicleId, long bootstrapTime) {
Data key;
Data value;
long valueSize;
boolean isDeleted;
if (entry instanceof MapEntry) {
isDeleted = false;
MapEntry mapEntry = (MapEntry) entry;
key = mapEntry.key();
value = mapEntry.value();
valueSize = value.size();
} else {
isDeleted = true;
MapAbsentEntry mapAbsentEntry = (MapAbsentEntry) entry;
key = mapAbsentEntry.absentKey();
value = null;
valueSize = Math.max(0, valueSizeMarshaller.minStorableSize());
}
destination.writeLong(bootstrapTime);
keySizeMarshaller.writeSize(destination, key.size());
valueSizeMarshaller.writeSize(destination, valueSize);
destination.writeStopBit(entry.originTimestamp());
if (entry.originIdentifier() == 0)
throw new IllegalStateException("Identifier can't be 0");
destination.writeByte(entry.originIdentifier());
Data key;
boolean isDeleted;
if (entry instanceof MapEntry) {
isDeleted = false;
key = ((MapEntry) entry).key();
} else {
isDeleted = true;
key = ((MapAbsentEntry) entry).absentKey();
}
destination.writeBoolean(isDeleted);
keySizeMarshaller.writeSize(destination, key.size());
key.writeTo(destination, destination.writePosition());
destination.writeSkip(key.size());
@@ -419,6 +410,8 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
if (isDeleted)
return;
Data value = ((MapEntry) entry).value();
valueSizeMarshaller.writeSize(destination, value.size());
value.writeTo(destination, destination.writePosition());
destination.writeSkip(value.size());
src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java
+6 -11
View File
@@ -19,14 +19,11 @@ package net.openhft.chronicle.map.impl.stage.input;
import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
import net.openhft.chronicle.hash.impl.stage.hash.LogHolder;
import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
import net.openhft.chronicle.hash.replication.RemoteOperationContext;
import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder;
import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData;
import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputKeyBytesData;
import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputValueBytesData;
import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages;
import net.openhft.chronicle.map.impl.stage.query.ReplicatedMapQuery;
import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate;
import net.openhft.chronicle.map.replication.MapRemoteQueryContext;
@@ -41,13 +38,10 @@ public abstract class ReplicatedInput<K, V, R>
@StageRef ReplicatedChronicleMapHolder<K, V, R> mh;
@StageRef ReplicationUpdate<K> ru;
@StageRef LogHolder lh;
@StageRef ReplicatedInputKeyBytesData<K> replicatedInputKeyBytesValue;
@StageRef ReplicatedInputValueBytesData<V> replicatedInputValueBytesValue;
@StageRef KeySearch<?> ks;
@StageRef ReplicatedMapQuery<K, V, ?> q;
@StageRef SegmentStages s;
@StageRef ReplicatedMapEntryStages<K, V> e;
@StageRef DummyValueZeroData<V> dummyValue;
@Override
@@ -77,17 +71,18 @@ 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);
riTimestamp = replicatedInputBytes.readStopBit();
riId = replicatedInputBytes.readByte();
ru.initReplicationUpdate(riTimestamp, riId);
isDeleted = replicatedInputBytes.readBoolean();
riKeySize = mh.m().keySizeMarshaller.readSize(replicatedInputBytes);
riKeyOffset = replicatedInputBytes.readPosition();
riValueOffset = riKeyOffset + riKeySize;
if (!isDeleted) {
replicatedInputBytes.readSkip(riKeySize);
riValueSize = mh.m().valueSizeMarshaller.readSize(replicatedInputBytes);
riValueOffset = replicatedInputBytes.readPosition();
}
}
public void processReplicatedEvent() {