From 5829cdbfbacbcd36eebc0ce3a4619dc9f74edd7c Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 18 May 2015 13:40:41 +0100 Subject: [PATCH 1/4] Updated SPI model and added examples --- .../net/openhft/chronicle/hash/HashEntry.java | 9 +--- .../DefaultEventualConsistencyStrategy.java | 32 ++++++++++++++ .../hash/replication/ReplicatedEntry.java | 26 +++++++++++ .../{ => replication}/ReplicationContext.java | 9 +++- .../openhft/chronicle/map/MapAbsentEntry.java | 7 +-- .../map/MapAbsentEntryOperations.java | 18 ++++++-- .../net/openhft/chronicle/map/MapEntry.java | 9 +--- .../chronicle/map/MapEntryOperations.java | 34 +++++++++++---- .../map/LoggingMapEntryOperations.java | 41 ++++++++++++++++++ .../MaximizingValueMapEntryOperations.java | 43 +++++++++++++++++++ 10 files changed, 195 insertions(+), 33 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java create mode 100644 src/main/java/net/openhft/chronicle/hash/replication/ReplicatedEntry.java rename src/main/java/net/openhft/chronicle/hash/{ => replication}/ReplicationContext.java (80%) create mode 100644 src/test/java/net/openhft/chronicle/map/LoggingMapEntryOperations.java create mode 100644 src/test/java/net/openhft/chronicle/map/MaximizingValueMapEntryOperations.java 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); + } + } +} From 4aa0af1acfd34220b49f48730e166176153ce8d3 Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 18 May 2015 13:49:50 +0100 Subject: [PATCH 2/4] Make MaximizingValueMapEntryOperations eventually consistent --- .../chronicle/map/MaximizingValueMapEntryOperations.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/map/MaximizingValueMapEntryOperations.java b/src/test/java/net/openhft/chronicle/map/MaximizingValueMapEntryOperations.java index bc0a833e6..f04f251e2 100644 --- a/src/test/java/net/openhft/chronicle/map/MaximizingValueMapEntryOperations.java +++ b/src/test/java/net/openhft/chronicle/map/MaximizingValueMapEntryOperations.java @@ -19,6 +19,7 @@ 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; @@ -27,8 +28,12 @@ public class MaximizingValueMapEntryOperations entry, Value newValue) { - if (entry.context() instanceof ReplicationContext) { - if (newValue.get().compareTo(entry.value().get()) > 0) { + MapContext context = entry.context(); + if (context instanceof ReplicationContext) { + int compareResult = newValue.get().compareTo(entry.value().get()); + if (compareResult > 0 || (compareResult == 0 && + ((ReplicationContext) context).remoteIdentifier() <= + ((ReplicatedEntry) entry).originIdentifier())) { // replace, if the new value is greater entry.doReplaceValue(newValue); return true; From 078068d8ffb9d865f63ac1070fc452879c445d1c Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 18 May 2015 13:51:03 +0100 Subject: [PATCH 3/4] Extract MapAbsentEntryOperations.shouldInsert() for subclassing ease --- .../chronicle/map/MapAbsentEntryOperations.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/MapAbsentEntryOperations.java b/src/main/java/net/openhft/chronicle/map/MapAbsentEntryOperations.java index 0585c87e9..10041ef06 100644 --- a/src/main/java/net/openhft/chronicle/map/MapAbsentEntryOperations.java +++ b/src/main/java/net/openhft/chronicle/map/MapAbsentEntryOperations.java @@ -32,6 +32,14 @@ import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyS * @param type of the value in {@code ChronicleMap} */ public interface MapAbsentEntryOperations { + + static boolean shoudInsert(@NotNull MapAbsentEntry absentEntry) { + MapContext context = absentEntry.context(); + return !(context instanceof ReplicationContext) || + !(absentEntry instanceof ReplicatedEntry) || + shouldApplyRemoteModification((ReplicatedEntry) absentEntry, + (ReplicationContext) context); + } /** * Inserts the new entry into the map, of {@link MapAbsentEntry#absentKey() the key} from @@ -43,12 +51,7 @@ public interface MapAbsentEntryOperations { * operation are not met */ default boolean insert(@NotNull MapAbsentEntry absentEntry, Value value) { - MapContext context = absentEntry.context(); - boolean shouldInsert = !(context instanceof ReplicationContext) || - !(absentEntry instanceof ReplicatedEntry) || - shouldApplyRemoteModification((ReplicatedEntry) absentEntry, - (ReplicationContext) context); - if (shouldInsert) { + if (shoudInsert(absentEntry)) { absentEntry.doInsert(value); return true; } else { From 9a228a6017e025696a97b9670e391fe027da0f9d Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 18 May 2015 13:51:59 +0100 Subject: [PATCH 4/4] Added InterProcessLock.newCondition default implementation --- .../net/openhft/chronicle/hash/locks/InterProcessLock.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java index e4f4284af..459d8fce8 100644 --- a/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java +++ b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java @@ -46,5 +46,8 @@ public interface InterProcessLock extends Lock { */ @NotNull @Override - Condition newCondition(); + default Condition newCondition() { + throw new UnsupportedOperationException( + "Conditions are not supported by inter-process locks"); + } }