diff --git a/src/main/java/net/openhft/chronicle/hash/HashEntry.java b/src/main/java/net/openhft/chronicle/hash/HashEntry.java index 0b0fbdda3..91c3d7b66 100644 --- a/src/main/java/net/openhft/chronicle/hash/HashEntry.java +++ b/src/main/java/net/openhft/chronicle/hash/HashEntry.java @@ -18,7 +18,6 @@ package net.openhft.chronicle.hash; -import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.MapEntryOperations; import org.jetbrains.annotations.NotNull; @@ -40,14 +39,10 @@ public interface HashEntry { @NotNull Value key(); /** - * Removes the entry from the {@code ChronicleHash}. Returns {@code true} if the remove was - * successful, {@code false} if it failed for any reason. - * - *

This method if the default implementation for {@link MapEntryOperations#remove(MapEntry)}, - * which might be customized over the default. + * Removes the entry from the {@code ChronicleHash}. * * @throws IllegalStateException if some locking/state conditions required to perform remove * operation are not met */ - boolean defaultRemove(); + void doRemove(); } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java b/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java new file mode 100644 index 000000000..d0542d75e --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java @@ -0,0 +1,32 @@ +/* + * Copyright 2015 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash.replication; + +public final class DefaultEventualConsistencyStrategy { + + public static boolean shouldApplyRemoteModification( + ReplicatedEntry entry, ReplicationContext context) { + long remoteTimestamp = context.remoteTimestamp(); + long originTimestamp = entry.originTimestamp(); + return remoteTimestamp > originTimestamp || (remoteTimestamp == originTimestamp && + context.remoteIdentifier() <= entry.originIdentifier()); + } + + private DefaultEventualConsistencyStrategy() {} +} diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicatedEntry.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicatedEntry.java new file mode 100644 index 000000000..19d43391a --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicatedEntry.java @@ -0,0 +1,26 @@ +/* + * Copyright 2015 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash.replication; + +public interface ReplicatedEntry { + + byte originIdentifier(); + + long originTimestamp(); +} diff --git a/src/main/java/net/openhft/chronicle/hash/ReplicationContext.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationContext.java similarity index 80% rename from src/main/java/net/openhft/chronicle/hash/ReplicationContext.java rename to src/main/java/net/openhft/chronicle/hash/replication/ReplicationContext.java index dcb2adb45..fb47f2b45 100644 --- a/src/main/java/net/openhft/chronicle/hash/ReplicationContext.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationContext.java @@ -16,7 +16,10 @@ * limitations under the License. */ -package net.openhft.chronicle.hash; +package net.openhft.chronicle.hash.replication; + +import net.openhft.chronicle.hash.ChronicleHash; +import net.openhft.chronicle.hash.HashContext; /** * Context of internal replication operation. @@ -24,4 +27,8 @@ package net.openhft.chronicle.hash; * @param the key type of accessed {@link ChronicleHash} */ public interface ReplicationContext extends HashContext { + + byte remoteIdentifier(); + + long remoteTimestamp(); } diff --git a/src/main/java/net/openhft/chronicle/map/MapAbsentEntry.java b/src/main/java/net/openhft/chronicle/map/MapAbsentEntry.java index 62e9049da..70b2da8eb 100644 --- a/src/main/java/net/openhft/chronicle/map/MapAbsentEntry.java +++ b/src/main/java/net/openhft/chronicle/map/MapAbsentEntry.java @@ -45,16 +45,11 @@ public interface MapAbsentEntry { /** * Inserts the new entry into the map, of {@link #absentKey() the key} and the given value. - * Returns {@code true} if the insertion was successful, {@code false} if it failed for - * any reason. - * - *

This method if the default implementation for {@link MapAbsentEntryOperations#insert}, - * which might be customized over the default. * * @throws IllegalStateException if some locking/state conditions required to perform insertion * operation are not met */ - boolean defaultInsert(Value value); + void doInsert(Value value); /** * Returns the "nil" value, which should be inserted into the map in this context. This diff --git a/src/main/java/net/openhft/chronicle/map/MapAbsentEntryOperations.java b/src/main/java/net/openhft/chronicle/map/MapAbsentEntryOperations.java index dfbf63b12..0585c87e9 100644 --- a/src/main/java/net/openhft/chronicle/map/MapAbsentEntryOperations.java +++ b/src/main/java/net/openhft/chronicle/map/MapAbsentEntryOperations.java @@ -19,8 +19,12 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.Value; +import net.openhft.chronicle.hash.replication.ReplicatedEntry; +import net.openhft.chronicle.hash.replication.ReplicationContext; import org.jetbrains.annotations.NotNull; +import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy.shouldApplyRemoteModification; + /** * SPI interface for customizing "low-level" insertion of a new entry into {@link ChronicleMap}. * @@ -35,13 +39,21 @@ public interface MapAbsentEntryOperations { * Returns {@code true} if the insertion was successful, {@code false} if it failed * for any reason. * - * @implNote simply delegates to {@link MapAbsentEntry#defaultInsert(Value)}. - * * @throws IllegalStateException if some locking/state conditions required to perform insertion * operation are not met */ default boolean insert(@NotNull MapAbsentEntry absentEntry, Value value) { - return absentEntry.defaultInsert(value); + MapContext context = absentEntry.context(); + boolean shouldInsert = !(context instanceof ReplicationContext) || + !(absentEntry instanceof ReplicatedEntry) || + shouldApplyRemoteModification((ReplicatedEntry) absentEntry, + (ReplicationContext) context); + if (shouldInsert) { + absentEntry.doInsert(value); + return true; + } else { + return false; + } } /** diff --git a/src/main/java/net/openhft/chronicle/map/MapEntry.java b/src/main/java/net/openhft/chronicle/map/MapEntry.java index 4dfa6ddd3..b6ac09307 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEntry.java +++ b/src/main/java/net/openhft/chronicle/map/MapEntry.java @@ -42,15 +42,10 @@ public interface MapEntry extends HashEntry { @NotNull Value value(); /** - * Replaces the entry's value with the new one. Returns {@code true} if the replace operation - * was successful, {@code false} if it failed for any reason. - * - *

This method if the default implementation for {@link MapEntryOperations#replaceValue}, - * which might be customized over the default. + * Replaces the entry's value with the new one. * * @throws IllegalStateException if some locking/state conditions required to perform replace * operation are not met */ - boolean defaultReplaceValue(Value newValue); + void doReplaceValue(Value newValue); } - diff --git a/src/main/java/net/openhft/chronicle/map/MapEntryOperations.java b/src/main/java/net/openhft/chronicle/map/MapEntryOperations.java index dd7cd3695..155852f3e 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEntryOperations.java +++ b/src/main/java/net/openhft/chronicle/map/MapEntryOperations.java @@ -19,8 +19,12 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.Value; +import net.openhft.chronicle.hash.replication.ReplicatedEntry; +import net.openhft.chronicle.hash.replication.ReplicationContext; import org.jetbrains.annotations.NotNull; +import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy.shouldApplyRemoteModification; + /** * SPI interface for customizing "low-level" modification operations on the present * {@link ChronicleMap} entry. @@ -29,33 +33,45 @@ import org.jetbrains.annotations.NotNull; * @param type of the value in {@code ChronicleMap} */ public interface MapEntryOperations { + + static boolean shouldModify(@NotNull MapEntry entry) { + MapContext context = entry.context(); + return !(context instanceof ReplicationContext) || + shouldApplyRemoteModification((ReplicatedEntry) entry, + (ReplicationContext) context); + } + /** * Removes the given entry from the {@code ChronicleHash}. Returns {@code true} if the remove * was successful, {@code false} if it failed for any reason. - * - * @implNote simply delegates to {@link MapEntry#defaultRemove()}. * * @param entry the entry to remove * @throws IllegalStateException if some locking/state conditions required to perform remove * operation are not met */ default boolean remove(@NotNull MapEntry entry) { - return entry.defaultRemove(); + if (shouldModify(entry)) { + entry.doRemove(); + return true; + } else { + return false; + } } - - + /** * Replaces the given entry's value with the new one. Returns {@code true} if the replace * operation was successful, {@code false} if it failed for any reason. * - * @implNote simply delegates to {@link MapEntry#defaultReplaceValue(Value) - * entry.defaultReplace(newValue)}. - * * @param entry the entry to replace the value in * @throws IllegalStateException if some locking/state conditions required to perform replace * operation are not met */ default boolean replaceValue(@NotNull MapEntry entry, Value newValue) { - return entry.defaultReplaceValue(newValue); + if (shouldModify(entry)) { + entry.doReplaceValue(newValue); + return true; + } else { + return false; + } } } diff --git a/src/test/java/net/openhft/chronicle/map/LoggingMapEntryOperations.java b/src/test/java/net/openhft/chronicle/map/LoggingMapEntryOperations.java new file mode 100644 index 000000000..7101d4ffb --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/LoggingMapEntryOperations.java @@ -0,0 +1,41 @@ +/* + * Copyright 2015 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.Value; +import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +class LoggingMapEntryOperations implements MapEntryOperations { + + static final Logger LOG = LoggerFactory.getLogger(LoggingMapEntryOperations.class); + + @Override + public boolean replaceValue(@NotNull MapEntry entry, Value newValue) { + if (MapEntryOperations.shouldModify(entry)) { + LOG.info("replace: old key: {}, old value: {}, new value: {}", + entry.key(), entry.value(), newValue); + entry.doReplaceValue(newValue); + return true; + } else { + return false; + } + } +} diff --git a/src/test/java/net/openhft/chronicle/map/MaximizingValueMapEntryOperations.java b/src/test/java/net/openhft/chronicle/map/MaximizingValueMapEntryOperations.java new file mode 100644 index 000000000..bc0a833e6 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/MaximizingValueMapEntryOperations.java @@ -0,0 +1,43 @@ +/* + * Copyright 2015 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.Value; +import net.openhft.chronicle.hash.replication.ReplicationContext; +import org.jetbrains.annotations.NotNull; + +public class MaximizingValueMapEntryOperations> + implements MapEntryOperations { + + @Override + public boolean replaceValue(@NotNull MapEntry entry, Value newValue) { + if (entry.context() instanceof ReplicationContext) { + if (newValue.get().compareTo(entry.value().get()) > 0) { + // replace, if the new value is greater + entry.doReplaceValue(newValue); + return true; + } else { + // always fail, if the value came via replication is lesser than the current value. + return false; + } + } else { + return MapEntryOperations.super.replaceValue(entry, newValue); + } + } +}