Hide the remaining parts of the replication API from the public eye

This commit is contained in:
leventov
2016-01-17 17:08:13 +03:00
parent d8a8a3c319
commit 56e5e93388
16 changed files with 272 additions and 137 deletions
@@ -18,7 +18,6 @@ package net.openhft.chronicle.hash;
import net.openhft.chronicle.bytes.Byteable;
import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
@@ -32,7 +31,6 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
* This interface defines the meaning of configurations, common to {@link ChronicleMapBuilder} and
@@ -382,43 +380,6 @@ public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
*/
B actualSegments(int actualSegments);
/**
* Configures timeout after which entries, marked as removed in the Chronicle Hash, constructed
* by this builder, are allowed to be completely removed from the data store. In replicated
* Chronicle nodes, when {@code remove()} on the key is called, the corresponding entry
* is not immediately erased from the data store, to let the distributed system eventually
* converge on some value for this key (or converge on the fact, that this key is removed).
* Chronicle Hash watch in runtime after the entries, and if one is removed and not updated
* in any way for this {@code removedEntryCleanupTimeout}, Chronicle is allowed to remove this
* entry completely from the data store. This timeout should depend on your distributed
* system topology, and typical replication latencies, that should be determined experimentally.
*
* <p>Default timeout is 1 minute.
*
* @param removedEntryCleanupTimeout timeout, after which stale removed entries could be erased
* from Chronicle Hash data store completely
* @param unit time unit, in which the timeout is given
* @return this builder back
* @throws IllegalArgumentException is the specified timeout is less than 1 millisecond
* @see #cleanupRemovedEntries(boolean)
* @see ReplicableEntry#doRemoveCompletely()
*/
B removedEntryCleanupTimeout(long removedEntryCleanupTimeout, TimeUnit unit);
/**
* Configures if replicated Chronicle Hashes, constructed by this builder, should
* completely erase entries, removed some time ago. See {@link #removedEntryCleanupTimeout(
* long, TimeUnit)} for more details on this mechanism.
*
* <p>Default value is {@code true} -- old removed entries are erased with 1 second timeout.
*
* @param cleanupRemovedEntries if stale removed entries should be purged from Chronicle Hash
* @return this builder back
* @see #removedEntryCleanupTimeout(long, TimeUnit)
* @see ReplicableEntry#doRemoveCompletely()
*/
B cleanupRemovedEntries(boolean cleanupRemovedEntries);
/**
* Configures the {@code DataAccess} and {@code SizedReader} used to serialize and deserialize
* keys to and from off-heap memory in hash containers, created by this builder.
@@ -511,8 +472,6 @@ public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
*/
B checksumEntries(boolean checksumEntries);
B replication(byte identifier);
/**
* Creates a new hash container from this builder, storing it's data in off-heap memory, not
* mapped to any file. On {@link ChronicleHash#close()} called on the returned container, or
@@ -622,5 +581,5 @@ public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
* @deprecated don't use private API in the client code
*/
@Deprecated
ChronicleHashBuilderPrivateAPI<K> privateAPI();
Object privateAPI();
}
@@ -16,9 +16,12 @@
package net.openhft.chronicle.hash;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.hash.serialization.impl.SerializationBuilder;
public interface ChronicleHashBuilderPrivateAPI<K> {
import java.util.concurrent.TimeUnit;
public interface ChronicleHashBuilderPrivateAPI<K, RO> {
SerializationBuilder<K> keyBuilder();
@@ -43,4 +46,45 @@ public interface ChronicleHashBuilderPrivateAPI<K> {
boolean aligned64BitMemoryOperationsAtomic();
boolean checksumEntries();
void replication(byte identifier);
/**
* Configures if replicated Chronicle Hashes, constructed by this builder, should
* completely erase entries, removed some time ago. See {@link #removedEntryCleanupTimeout(
* long, TimeUnit)} for more details on this mechanism.
*
* <p>Default value is {@code true} -- old removed entries are erased with 1 second timeout.
*
* @param cleanupRemovedEntries if stale removed entries should be purged from Chronicle Hash
* @return this builder back
* @see #removedEntryCleanupTimeout(long, TimeUnit)
* @see ReplicableEntry#doRemoveCompletely()
*/
void cleanupRemovedEntries(boolean cleanupRemovedEntries);
/**
* Configures timeout after which entries, marked as removed in the Chronicle Hash, constructed
* by this builder, are allowed to be completely removed from the data store. In replicated
* Chronicle nodes, when {@code remove()} on the key is called, the corresponding entry
* is not immediately erased from the data store, to let the distributed system eventually
* converge on some value for this key (or converge on the fact, that this key is removed).
* Chronicle Hash watch in runtime after the entries, and if one is removed and not updated
* in any way for this {@code removedEntryCleanupTimeout}, Chronicle is allowed to remove this
* entry completely from the data store. This timeout should depend on your distributed
* system topology, and typical replication latencies, that should be determined experimentally.
*
* <p>Default timeout is 1 minute.
*
* @param removedEntryCleanupTimeout timeout, after which stale removed entries could be erased
* from Chronicle Hash data store completely
* @param unit time unit, in which the timeout is given
* @return this builder back
* @throws IllegalArgumentException is the specified timeout is less than 1 millisecond
* @see #cleanupRemovedEntries(boolean)
* @see ReplicableEntry#doRemoveCompletely()
*/
void removedEntryCleanupTimeout(long removedEntryCleanupTimeout, TimeUnit unit);
void remoteOperations(RO remoteOperations);
}
@@ -162,8 +162,9 @@ public abstract class VanillaChronicleHash<K,
// Version
dataFileVersion = BuildVersion.version();
@SuppressWarnings("deprecation")
ChronicleHashBuilderPrivateAPI<K> privateAPI = builder.privateAPI();
@SuppressWarnings({"deprecation", "unchecked"})
ChronicleHashBuilderPrivateAPI<K, ?> privateAPI =
(ChronicleHashBuilderPrivateAPI<K, ?>) builder.privateAPI();
// Data model
SerializationBuilder<K> keyBuilder = privateAPI.keyBuilder();
@@ -18,6 +18,7 @@ package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.set.replication.SetRemoteOperations;
@@ -25,8 +26,8 @@ import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyS
import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy.AcceptanceDecision.DISCARD;
/**
* Specifies the default eventual consistency strategy for {@link ChronicleHashBuilder#replication(
* byte) replicated} {@link ChronicleHash}es:
* Specifies the default eventual consistency strategy for {@link
* ChronicleHashBuilderPrivateAPI#replication(byte) replicated} {@link ChronicleHash}es:
* <i>last write wins</i>. If two writes to a single entry occurred simultaneously on different
* nodes, the write on the node with lower identifier wins.
*
@@ -18,6 +18,7 @@ package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
import net.openhft.chronicle.hash.HashAbsentEntry;
import net.openhft.chronicle.hash.HashEntry;
import net.openhft.chronicle.map.ChronicleMap;
@@ -121,8 +122,8 @@ public interface ReplicableEntry {
* method any subsequent call of any method of {@code ReplicableEntry} class throws {@code
* IllegalStateException}.
*
* @see ChronicleHashBuilder#cleanupRemovedEntries(boolean)
* @see ChronicleHashBuilder#removedEntryCleanupTimeout(long, TimeUnit)
* @see ChronicleHashBuilderPrivateAPI#cleanupRemovedEntries(boolean)
* @see ChronicleHashBuilderPrivateAPI#removedEntryCleanupTimeout(long, TimeUnit)
*/
void doRemoveCompletely();
}
@@ -18,6 +18,7 @@ package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -32,7 +33,7 @@ import static java.util.concurrent.TimeUnit.NANOSECONDS;
*
* @see ReplicableEntry#originTimestamp()
* @see RemoteOperationContext#remoteTimestamp()
* @see ChronicleHashBuilder#removedEntryCleanupTimeout(long, TimeUnit)
* @see ChronicleHashBuilderPrivateAPI#removedEntryCleanupTimeout(long, TimeUnit)
*/
public final class TimeProvider {
@@ -22,7 +22,6 @@ import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.bytes.BytesStore;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
import net.openhft.chronicle.hash.ChronicleHashRecoveryFailedException;
import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable;
import net.openhft.chronicle.hash.impl.stage.entry.ChecksumStrategy;
@@ -146,7 +145,8 @@ public final class ChronicleMapBuilder<K, V> implements
private static int MAX_BOOTSTRAPPING_HEADER_SIZE = (int) MemoryUnit.KILOBYTES.toBytes(16);
// not final because of cloning
private ChronicleMapBuilderPrivateAPI<K> privateAPI = new ChronicleMapBuilderPrivateAPI<>(this);
private ChronicleMapBuilderPrivateAPI<K, V> privateAPI =
new ChronicleMapBuilderPrivateAPI<>(this);
//////////////////////////////
// Configuration fields
@@ -267,7 +267,7 @@ public final class ChronicleMapBuilder<K, V> implements
*/
@Override
@Deprecated
public ChronicleHashBuilderPrivateAPI<K> privateAPI() {
public Object privateAPI() {
return privateAPI;
}
@@ -1152,8 +1152,7 @@ public final class ChronicleMapBuilder<K, V> implements
return toString().hashCode();
}
@Override
public ChronicleMapBuilder<K, V> removedEntryCleanupTimeout(
ChronicleMapBuilder<K, V> removedEntryCleanupTimeout(
long removedEntryCleanupTimeout, TimeUnit unit) {
if (unit.toMillis(removedEntryCleanupTimeout) < 1) {
throw new IllegalArgumentException("timeout should be >= 1 millisecond, " +
@@ -1164,8 +1163,7 @@ public final class ChronicleMapBuilder<K, V> implements
return this;
}
@Override
public ChronicleMapBuilder<K, V> cleanupRemovedEntries(boolean cleanupRemovedEntries) {
ChronicleMapBuilder<K, V> cleanupRemovedEntries(boolean cleanupRemovedEntries) {
this.cleanupRemovedEntries = cleanupRemovedEntries;
return this;
}
@@ -1342,8 +1340,7 @@ public final class ChronicleMapBuilder<K, V> implements
return this;
}
@Override
public ChronicleMapBuilder<K, V> replication(byte identifier) {
ChronicleMapBuilder<K, V> replication(byte identifier) {
if (identifier <= 0)
throw new IllegalArgumentException("Identifier must be positive, " + identifier +
" given");
@@ -1608,7 +1605,7 @@ public final class ChronicleMapBuilder<K, V> implements
return this;
}
public ChronicleMapBuilder<K, V> remoteOperations(
ChronicleMapBuilder<K, V> remoteOperations(
MapRemoteOperations<K, V, ?> remoteOperations) {
Objects.requireNonNull(remoteOperations);
this.remoteOperations = remoteOperations;
@@ -18,12 +18,16 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
import net.openhft.chronicle.hash.serialization.impl.SerializationBuilder;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
class ChronicleMapBuilderPrivateAPI<K> implements ChronicleHashBuilderPrivateAPI<K> {
import java.util.concurrent.TimeUnit;
private ChronicleMapBuilder<K, ?> b;
class ChronicleMapBuilderPrivateAPI<K, V>
implements ChronicleHashBuilderPrivateAPI<K, MapRemoteOperations<K, V, ?>> {
public ChronicleMapBuilderPrivateAPI(ChronicleMapBuilder<K, ?> b) {
private final ChronicleMapBuilder<K, V> b;
public ChronicleMapBuilderPrivateAPI(ChronicleMapBuilder<K, V> b) {
this.b = b;
}
@@ -86,4 +90,24 @@ class ChronicleMapBuilderPrivateAPI<K> implements ChronicleHashBuilderPrivateAPI
public boolean checksumEntries() {
return b.checksumEntries();
}
@Override
public void replication(byte identifier) {
b.replication(identifier);
}
@Override
public void cleanupRemovedEntries(boolean cleanupRemovedEntries) {
b.cleanupRemovedEntries(cleanupRemovedEntries);
}
@Override
public void removedEntryCleanupTimeout(long removedEntryCleanupTimeout, TimeUnit unit) {
b.removedEntryCleanupTimeout(removedEntryCleanupTimeout, unit);
}
@Override
public void remoteOperations(MapRemoteOperations<K, V, ?> remoteOperations) {
b.remoteOperations(remoteOperations);
}
}
@@ -64,7 +64,7 @@ import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyS
* @param <V> the map value type
* @param <R> the return type of {@link MapEntryOperations} specified fro the queried map
* @see DefaultEventualConsistencyStrategy
* @see ChronicleMapBuilder#remoteOperations(MapRemoteOperations)
* @see ChronicleMapBuilderPrivateAPI#remoteOperations(MapRemoteOperations)
*/
public interface MapRemoteOperations<K, V, R> {
@@ -22,14 +22,10 @@ import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.map.*;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.map.replication.MapRemoteQueryContext;
import net.openhft.chronicle.set.replication.SetRemoteOperations;
import net.openhft.chronicle.set.replication.SetRemoteQueryContext;
import org.jetbrains.annotations.NotNull;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
/**
* {@code ChronicleSetBuilder} manages the whole set of {@link ChronicleSet} configurations, could
@@ -46,12 +42,17 @@ public final class ChronicleSetBuilder<K>
implements ChronicleHashBuilder<K, ChronicleSet<K>, ChronicleSetBuilder<K>> {
private ChronicleMapBuilder<K, DummyValue> chronicleMapBuilder;
private ChronicleSetBuilderPrivateAPI<K> privateAPI;
ChronicleSetBuilder(Class<K> keyClass) {
chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, DummyValue.class)
.valueReaderAndDataAccess(
DummyValueMarshaller.INSTANCE, DummyValueMarshaller.INSTANCE)
.valueSizeMarshaller(SizeMarshaller.constant(0));
//noinspection deprecation,unchecked
privateAPI = new ChronicleSetBuilderPrivateAPI<>(
(ChronicleHashBuilderPrivateAPI<K, MapRemoteOperations<K, DummyValue, ?>>)
chronicleMapBuilder.privateAPI());
}
public static <K> ChronicleSetBuilder<K> of(Class<K> keyClass) {
@@ -195,19 +196,6 @@ public final class ChronicleSetBuilder<K>
return toString().hashCode();
}
@Override
public ChronicleSetBuilder<K> removedEntryCleanupTimeout(
long removedEntryCleanupTimeout, TimeUnit unit) {
chronicleMapBuilder.removedEntryCleanupTimeout(removedEntryCleanupTimeout, unit);
return this;
}
@Override
public ChronicleSetBuilder<K> cleanupRemovedEntries(boolean cleanupRemovedEntries) {
chronicleMapBuilder.cleanupRemovedEntries(cleanupRemovedEntries);
return this;
}
@Override
public ChronicleSetBuilder<K> keyReaderAndDataAccess(
SizedReader<K> keyReader, @NotNull DataAccess<K> keyDataAccess) {
@@ -262,44 +250,24 @@ public final class ChronicleSetBuilder<K>
return this;
}
@Override
public ChronicleSetBuilder<K> replication(byte identifier) {
chronicleMapBuilder.replication(identifier);
return this;
}
public ChronicleSetBuilder<K> entryOperations(SetEntryOperations<K, ?> entryOperations) {
chronicleMapBuilder.entryOperations(new MapEntryOperations<K, DummyValue, Object>() {
@Override
public Object remove(@NotNull MapEntry<K, DummyValue> entry) {
//noinspection unchecked
return entryOperations.remove((SetEntry<K>) entry);
}
@Override
public Object insert(
@NotNull MapAbsentEntry<K, DummyValue> absentEntry, Data<DummyValue> value) {
//noinspection unchecked
return entryOperations.insert((SetAbsentEntry<K>) absentEntry);
}
});
return this;
}
public ChronicleSetBuilder<K> remoteOperations(SetRemoteOperations<K, ?> remoteOperations) {
chronicleMapBuilder.remoteOperations(new MapRemoteOperations<K, DummyValue, Object>() {
@Override
public void remove(MapRemoteQueryContext<K, DummyValue, Object> q) {
remoteOperations.remove((SetRemoteQueryContext) q);
}
@Override
public void put(
MapRemoteQueryContext<K, DummyValue, Object> q, Data<DummyValue> newValue) {
remoteOperations.put((SetRemoteQueryContext) q);
}
});
return this;
}
@Override
public ChronicleSet<K> create() {
final ChronicleMap<K, DummyValue> map = chronicleMapBuilder.create();
@@ -331,9 +299,8 @@ public final class ChronicleSetBuilder<K>
*/
@Deprecated
@Override
public ChronicleHashBuilderPrivateAPI<K> privateAPI() {
//noinspection deprecation
return chronicleMapBuilder.privateAPI();
public Object privateAPI() {
return privateAPI;
}
}
@@ -0,0 +1,131 @@
/*
* 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 <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.set;
import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.serialization.impl.SerializationBuilder;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.map.replication.MapRemoteQueryContext;
import net.openhft.chronicle.set.replication.SetRemoteOperations;
import net.openhft.chronicle.set.replication.SetRemoteQueryContext;
import java.util.concurrent.TimeUnit;
class ChronicleSetBuilderPrivateAPI<K>
implements ChronicleHashBuilderPrivateAPI<K, SetRemoteOperations<K, ?>> {
private final ChronicleHashBuilderPrivateAPI<K, MapRemoteOperations<K, DummyValue, ?>> mapB;
public ChronicleSetBuilderPrivateAPI(
ChronicleHashBuilderPrivateAPI<K, MapRemoteOperations<K, DummyValue, ?>> mapB) {
this.mapB = mapB;
}
@Override
public SerializationBuilder<K> keyBuilder() {
return mapB.keyBuilder();
}
@Override
public int segmentEntrySpaceInnerOffset() {
return mapB.segmentEntrySpaceInnerOffset();
}
@Override
public long chunkSize() {
return mapB.chunkSize();
}
@Override
public int maxChunksPerEntry() {
return mapB.maxChunksPerEntry();
}
@Override
public long entriesPerSegment() {
return mapB.entriesPerSegment();
}
@Override
public long tierHashLookupCapacity() {
return mapB.tierHashLookupCapacity();
}
@Override
public long actualChunksPerSegmentTier() {
return mapB.actualChunksPerSegmentTier();
}
@Override
public int segmentHeaderSize() {
return mapB.segmentHeaderSize();
}
@Override
public int actualSegments() {
return mapB.actualSegments();
}
@Override
public long maxExtraTiers() {
return mapB.maxExtraTiers();
}
@Override
public boolean aligned64BitMemoryOperationsAtomic() {
return mapB.aligned64BitMemoryOperationsAtomic();
}
@Override
public boolean checksumEntries() {
return mapB.checksumEntries();
}
@Override
public void replication(byte identifier) {
mapB.replication(identifier);
}
@Override
public void cleanupRemovedEntries(boolean cleanupRemovedEntries) {
mapB.cleanupRemovedEntries(cleanupRemovedEntries);
}
@Override
public void removedEntryCleanupTimeout(long removedEntryCleanupTimeout, TimeUnit unit) {
mapB.removedEntryCleanupTimeout(removedEntryCleanupTimeout, unit);
}
@Override
public void remoteOperations(SetRemoteOperations<K, ?> remoteOperations) {
mapB.remoteOperations(new MapRemoteOperations<K, DummyValue, Object>() {
@Override
public void remove(MapRemoteQueryContext<K, DummyValue, Object> q) {
//noinspection unchecked
remoteOperations.remove((SetRemoteQueryContext) q);
}
@Override
public void put(
MapRemoteQueryContext<K, DummyValue, Object> q, Data<DummyValue> newValue) {
//noinspection unchecked
remoteOperations.put((SetRemoteQueryContext) q);
}
});
}
}