diff --git a/pom.xml b/pom.xml index f6df24acd..bdb2af24c 100755 --- a/pom.xml +++ b/pom.xml @@ -46,7 +46,7 @@ net.openhft chronicle-bom pom - 1.10.19 + 1.10.28-SNAPSHOT import diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 9dfb27848..6346cf068 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -443,8 +443,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap remoteOpContext = mapContext()) { - remoteOpContext.initReplicationInput(source); - remoteOpContext.processReplicatedEvent(); + remoteOpContext.processReplicatedEvent(source); } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java index e9b8fa226..79531547e 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java @@ -28,8 +28,6 @@ import net.openhft.chronicle.hash.impl.stage.query.QuerySegmentStages; import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; -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.data.bytes.WrappedValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDataHolder; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; @@ -93,9 +91,6 @@ nested = { ReplicatedMapAbsentDelegating.class, - ReplicatedInputKeyBytesData.class, - ReplicatedInputValueBytesData.class, - DummyValueZeroData.class, HashEntryChecksumStrategy.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/ReplicatedInputKeyBytesData.java b/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/ReplicatedInputKeyBytesData.java deleted file mode 100644 index dab33ea18..000000000 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/ReplicatedInputKeyBytesData.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * Copyright (C) 2015 higherfrequencytrading.com - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Lesser General Public License as published by - * the Free Software Foundation, either version 3 of the License. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public License - * along with this program. If not, see . - */ - -package net.openhft.chronicle.map.impl.stage.data.bytes; - -import net.openhft.chronicle.bytes.Bytes; -import net.openhft.chronicle.bytes.RandomDataInput; -import net.openhft.chronicle.hash.AbstractData; -import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; -import net.openhft.chronicle.map.impl.stage.input.ReplicatedInput; -import net.openhft.sg.Stage; -import net.openhft.sg.StageRef; -import net.openhft.sg.Staged; - - -@Staged -public class ReplicatedInputKeyBytesData extends AbstractData { - @StageRef ReplicatedInput in; - @StageRef KeyBytesInterop ki; - - @Stage("CachedBytesReplicatedInputKey") private K cachedBytesReplicatedInputKey; - @Stage("CachedBytesReplicatedInputKey") private boolean cachedBytesReplicatedInputKeyRead = - false; - - private void initCachedBytesReplicatedInputKey() { - cachedBytesReplicatedInputKey = getUsing(cachedBytesReplicatedInputKey); - cachedBytesReplicatedInputKeyRead = true; - } - - @Override - public RandomDataInput bytes() { - return in.replicatedInputBytes; - } - - @Override - public long offset() { - return in.riKeyOffset; - } - - @Override - public long size() { - return in.riKeySize; - } - - @Override - public K get() { - return cachedBytesReplicatedInputKey; - } - - @Override - public K getUsing(K using) { - Bytes inputBytes = in.replicatedInputBytes; - inputBytes.readPosition(in.riKeyOffset); - return ki.keyReader.read(inputBytes, size(), using); - } -} diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/ReplicatedInputValueBytesData.java b/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/ReplicatedInputValueBytesData.java deleted file mode 100644 index 9158d9651..000000000 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/ReplicatedInputValueBytesData.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * Copyright (C) 2015 higherfrequencytrading.com - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Lesser General Public License as published by - * the Free Software Foundation, either version 3 of the License. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public License - * along with this program. If not, see . - */ - -package net.openhft.chronicle.map.impl.stage.data.bytes; - -import net.openhft.chronicle.bytes.Bytes; -import net.openhft.chronicle.bytes.RandomDataInput; -import net.openhft.chronicle.hash.AbstractData; -import net.openhft.chronicle.map.impl.stage.input.ReplicatedInput; -import net.openhft.chronicle.map.impl.stage.map.ValueBytesInterop; -import net.openhft.sg.Stage; -import net.openhft.sg.StageRef; -import net.openhft.sg.Staged; - - -@Staged -public class ReplicatedInputValueBytesData extends AbstractData { - @StageRef ReplicatedInput in; - @StageRef ValueBytesInterop vi; - - @Stage("CachedBytesReplicatedInputValue") private V cachedBytesReplicatedInputValue; - @Stage("CachedBytesReplicatedInputValue") private boolean cachedBytesReplicatedInputValueRead = - false; - - private void initCachedBytesReplicatedInputValue() { - cachedBytesReplicatedInputValue = getUsing(cachedBytesReplicatedInputValue); - cachedBytesReplicatedInputValueRead = true; - } - - @Override - public RandomDataInput bytes() { - return in.replicatedInputBytes; - } - - @Override - public long offset() { - return in.riValueOffset; - } - - @Override - public long size() { - return in.riValueSize; - } - - @Override - public V get() { - return cachedBytesReplicatedInputValue; - } - - @Override - public V getUsing(V using) { - Bytes inputBytes = in.replicatedInputBytes; - inputBytes.readPosition(in.riValueOffset); - return vi.valueReader.read(inputBytes, size(), using); - } -} diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java b/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java index 1c4279ae2..e742104e6 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java @@ -19,16 +19,14 @@ 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.CheckOnEachPublicOperation; import net.openhft.chronicle.hash.replication.RemoteOperationContext; import net.openhft.chronicle.map.Replica; 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.query.ReplicatedMapQuery; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.chronicle.map.replication.MapRemoteQueryContext; -import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -37,66 +35,44 @@ import net.openhft.sg.Staged; public abstract class ReplicatedInput implements RemoteOperationContext, MapRemoteQueryContext, Replica.QueryContext { + @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; @StageRef ReplicatedChronicleMapHolder mh; @StageRef ReplicationUpdate ru; - @StageRef ReplicatedInputKeyBytesData replicatedInputKeyBytesValue; - @StageRef ReplicatedInputValueBytesData replicatedInputValueBytesValue; @StageRef ReplicatedMapQuery q; @StageRef SegmentStages s; @StageRef DummyValueZeroData dummyValue; @Override public Data dummyZeroValue() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); return dummyValue; } - public Bytes replicatedInputBytes = null; + public void processReplicatedEvent(Bytes replicatedInputBytes) { + long bootstrapTimestamp = replicatedInputBytes.readLong(); - public void initReplicatedInputBytes(Bytes replicatedInputBytes) { - this.replicatedInputBytes = replicatedInputBytes; - } + long timestamp = replicatedInputBytes.readStopBit(); + byte identifier = replicatedInputBytes.readByte(); + ru.initReplicationUpdate(identifier, timestamp); - // ri for "replication input" - @Stage("ReplicationInput") public long bootstrapTimestamp; - @Stage("ReplicationInput") public long riKeySize = -1; - @Stage("ReplicationInput") public long riValueSize; + boolean isDeleted = replicatedInputBytes.readBoolean(); + long keySize = mh.m().keySizeMarshaller.readSize(replicatedInputBytes); + long keyOffset = replicatedInputBytes.readPosition(); - @Stage("ReplicationInput") public long riKeyOffset; - @Stage("ReplicationInput") public long riValueOffset; + mh.m().setLastModificationTime(identifier, bootstrapTimestamp); - @Stage("ReplicationInput") public long riTimestamp; - @Stage("ReplicationInput") public byte riId; - @Stage("ReplicationInput") public boolean isDeleted; - - - public void initReplicationInput(Bytes replicatedInputBytes) { - initReplicatedInputBytes(replicatedInputBytes); - bootstrapTimestamp = replicatedInputBytes.readLong(); - riTimestamp = replicatedInputBytes.readStopBit(); - riId = replicatedInputBytes.readByte(); - ru.initReplicationUpdate(riId, riTimestamp); - - isDeleted = replicatedInputBytes.readBoolean(); - riKeySize = mh.m().keySizeMarshaller.readSize(replicatedInputBytes); - riKeyOffset = replicatedInputBytes.readPosition(); - if (!isDeleted) { - replicatedInputBytes.readSkip(riKeySize); - riValueSize = mh.m().valueSizeMarshaller.readSize(replicatedInputBytes); - riValueOffset = replicatedInputBytes.readPosition(); - } - } - - public void processReplicatedEvent() { - - mh.m().setLastModificationTime(riId, bootstrapTimestamp); - - q.initInputKey(replicatedInputKeyBytesValue); + q.initInputKey(q.getInputKeyBytesAsData(replicatedInputBytes, keyOffset, keySize)); if (isDeleted) { s.innerUpdateLock.lock(); mh.m().remoteOperations.remove(this); } else { + replicatedInputBytes.readSkip(keySize); + long valueSize = mh.m().valueSizeMarshaller.readSize(replicatedInputBytes); + long valueOffset = replicatedInputBytes.readPosition(); + Data value = q.wrapValueBytesAsData(replicatedInputBytes, valueOffset, valueSize); + s.innerWriteLock.lock(); - mh.m().remoteOperations.put(this, replicatedInputValueBytesValue); + mh.m().remoteOperations.put(this, value); } }