HCOLL-167 added isStateless

This commit is contained in:
Rob Austin
2014-10-20 15:02:05 +01:00
parent 023d12a6d0
commit c047e59d56
6 changed files with 222 additions and 181 deletions
@@ -19,35 +19,35 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
* This interface defines the meaning of configurations, common to {@link ChronicleMapBuilder}
* and {@link ChronicleSetBuilder}, i. e. <i>Chronicle hash container</i> configurations.
* This interface defines the meaning of configurations, common to {@link ChronicleMapBuilder} and
* {@link ChronicleSetBuilder}, i. e. <i>Chronicle hash container</i> configurations.
*
* <p>{@code ChronicleHashBuilder} is mutable. Configuration methods mutate the builder and return
* <i>the builder itself</i> back to support chaining pattern, rather than the builder copies with
* the corresponding configuration changed. To make an independent configuration,
* {@linkplain #clone} the builder.
* the corresponding configuration changed. To make an independent configuration, {@linkplain
* #clone} the builder.
*
* @param <K> the type of keys in hash containers, created by this builder
* @param <C> the container type, created by this builder, i. e. {@link ChronicleMap}
* or {@link ChronicleSet}
* @param <B> the concrete builder type, i. e. {@link ChronicleMapBuilder}
* or {@link ChronicleSetBuilder}
* @param <C> the container type, created by this builder, i. e. {@link ChronicleMap} or {@link
* ChronicleSet}
* @param <B> the concrete builder type, i. e. {@link ChronicleMapBuilder} or {@link
* ChronicleSetBuilder}
*/
public interface ChronicleHashBuilder<K, C extends ChronicleHash,
B extends ChronicleHashBuilder<K, C, B>> extends Cloneable {
/**
* Clones this builder. Useful for configuration persisting, because
* {@code ChronicleHashBuilder}s are mutable and changed on each configuration method call.
* Original and cloned builders are independent.
* Clones this builder. Useful for configuration persisting, because {@code
* ChronicleHashBuilder}s are mutable and changed on each configuration method call. Original
* and cloned builders are independent.
*
* @return a new clone of this builder
*/
B clone();
/**
* Set minimum number of segments in hash containers, constructed by this builder.
* See concurrencyLevel in {@link ConcurrentHashMap}.
* Set minimum number of segments in hash containers, constructed by this builder. See
* concurrencyLevel in {@link ConcurrentHashMap}.
*
* @param minSegments the minimum number of segments in containers, constructed by this builder
* @return this builder object back
@@ -56,11 +56,11 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
/**
* Configures the optimal number of bytes, taken by serialized form of keys, put into hash
* containers, created by this builder. If key size is always the same, call
* {@link #constantKeySizeBySample(Object)} method instead of this one.
* containers, created by this builder. If key size is always the same, call {@link
* #constantKeySizeBySample(Object)} method instead of this one.
*
* <p>If key size varies moderately, specify the size higher than average, but lower than
* the maximum possible, to minimize average memory overuse. If key size varies in a wide range,
* <p>If key size varies moderately, specify the size higher than average, but lower than the
* maximum possible, to minimize average memory overuse. If key size varies in a wide range,
* it's better to use {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it
* directly.
*
@@ -73,11 +73,11 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
/**
* Configures the constant number of bytes, taken by serialized form of keys, put into hash
* containers, created by this builder. This is done by providing the {@code sampleKey},
* all keys should take the same number of bytes in serialized form, as this sample object.
* containers, created by this builder. This is done by providing the {@code sampleKey}, all
* keys should take the same number of bytes in serialized form, as this sample object.
*
* <p>If keys are of boxed primitive type or {@link Byteable} subclass, i. e. if key size
* is known statically, it is automatically accounted and this method shouldn't be called.
* <p>If keys are of boxed primitive type or {@link Byteable} subclass, i. e. if key size is
* known statically, it is automatically accounted and this method shouldn't be called.
*
* <p>If key size varies, method {@link #keySize(int)} or {@link #entrySize(int)} should be
* called instead of this one.
@@ -89,55 +89,45 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
B constantKeySizeBySample(K sampleKey);
/**
* Configures the size in bytes of allocation unit of hash container instances, created by
* this builder.
* Configures the size in bytes of allocation unit of hash container instances, created by this
* builder.
*
* <p>{@link ChronicleMap} and {@link ChronicleSet} store their data off-heap, so it is required
* to serialize key (and values, in {@code ChronicleMap} case) (unless they are direct
* {@link Byteable} instances). Serialized key bytes (+ serialized value bytes,
* in {@code ChronicleMap} case) + some metadata bytes comprise "entry space", which
* {@code ChronicleMap} or {@code ChronicleSet} should allocate. So <i>entry size</i>
* is a minimum allocation portion in the hash containers, created by this builder.
* E. g. if entry size is 100, the created container could only allocate 100, 200, 300... bytes
* for an entry. If say 150 bytes of entry space are required by the entry, 200 bytes
* will be allocated, 150 used and 50 wasted. To minimize memory overuse and improve speed, you
* should pay decent attention to this configuration.
* to serialize key (and values, in {@code ChronicleMap} case) (unless they are direct {@link
* Byteable} instances). Serialized key bytes (+ serialized value bytes, in {@code ChronicleMap}
* case) + some metadata bytes comprise "entry space", which {@code ChronicleMap} or {@code
* ChronicleSet} should allocate. So <i>entry size</i> is a minimum allocation portion in the
* hash containers, created by this builder. E. g. if entry size is 100, the created container
* could only allocate 100, 200, 300... bytes for an entry. If say 150 bytes of entry space are
* required by the entry, 200 bytes will be allocated, 150 used and 50 wasted. To minimize
* memory overuse and improve speed, you should pay decent attention to this configuration.
*
* <p>There are three major patterns of this configuration usage:
* <ol>
* <li>Key (and value, in {@code ChronicleMap} case) sizes are constant. Configure them
* via {@link #constantKeySizeBySample(Object)} and
* {@link ChronicleMapBuilder#constantValueSizeBySample(Object)} methods, and you will
* experience no memory waste at all.</li>
* <li>Key (and/or value size, in {@code ChronicleMap} case) varies moderately. Specify them
* using corresponding methods, or specify entry size directly by calling this method,
* by sizes somewhere between average and maximum possible. The idea is to have most
* (90% or more) entries to fit a single "entry size" with moderate memory waste
* (10-20% on average), rest 10% or less of entries should take 2 "entry sizes", thus with
* ~50% memory overuse.</li>
* <li>Key (and/or value size, in {@code ChronicleMap} case) varies in a wide range.
* Then it's best to use entry size configuration in <i>chunk mode</i>. Specify entry size
* so that most entries should take from 5 to several dozens of "chunks". With this
* approach, average memory waste should be very low.
* <p>There are three major patterns of this configuration usage: <ol> <li>Key (and value, in
* {@code ChronicleMap} case) sizes are constant. Configure them via {@link
* #constantKeySizeBySample(Object)} and {@link ChronicleMapBuilder#constantValueSizeBySample(Object)}
* methods, and you will experience no memory waste at all.</li> <li>Key (and/or value size, in
* {@code ChronicleMap} case) varies moderately. Specify them using corresponding methods, or
* specify entry size directly by calling this method, by sizes somewhere between average and
* maximum possible. The idea is to have most (90% or more) entries to fit a single "entry size"
* with moderate memory waste (10-20% on average), rest 10% or less of entries should take 2
* "entry sizes", thus with ~50% memory overuse.</li> <li>Key (and/or value size, in {@code
* ChronicleMap} case) varies in a wide range. Then it's best to use entry size configuration in
* <i>chunk mode</i>. Specify entry size so that most entries should take from 5 to several
* dozens of "chunks". With this approach, average memory waste should be very low.
*
* <p>However, remember that
* <ul>
* <li>Operations with entries that span several "entry sizes" are a bit slower, than
* entries which take a single "entry size". That is why "chunk" approach
* is not recommended, when key (and/or value size) varies moderately.</li>
* <li>The maximum number of chunks could be taken by an entry is 64.
* {@link IllegalArgumentException} is thrown on attempt to insert too large entry,
* compared to the configured or computed entry size.</li>
* <li>The number of "entries" {@linkplain #entries(long) in the whole map} and
* {@linkplain #actualEntriesPerSegment(int) per segment} is actually the number
* "chunks".</li>
* </ul>
* <p>However, remember that <ul> <li>Operations with entries that span several "entry sizes"
* are a bit slower, than entries which take a single "entry size". That is why "chunk" approach
* is not recommended, when key (and/or value size) varies moderately.</li> <li>The maximum
* number of chunks could be taken by an entry is 64. {@link IllegalArgumentException} is thrown
* on attempt to insert too large entry, compared to the configured or computed entry size.</li>
* <li>The number of "entries" {@linkplain #entries(long) in the whole map} and {@linkplain
* #actualEntriesPerSegment(int) per segment} is actually the number "chunks".</li> </ul>
*
* <p>Example: if values in your {@code ChronicleMap} are adjacency lists of some social
* graph, where nodes are represented as {@code long} ids, and adjacency lists
* are serialized in efficient manner, for example as {@code long[]} arrays. Typical number
* of connections is 100-300, maximum is 3000. In this case entry size of
* 50 * (8 bytes for each id) = 400 bytes would be a good choice: <pre>{@code
* <p>Example: if values in your {@code ChronicleMap} are adjacency lists of some social graph,
* where nodes are represented as {@code long} ids, and adjacency lists are serialized in
* efficient manner, for example as {@code long[]} arrays. Typical number of connections is
* 100-300, maximum is 3000. In this case entry size of
* 50 * (8 bytes for each id) = 400 bytes would be a good choice: <pre>{@code
* Map<Long, long[]> socialGraph = ChronicleMapBuilder
* .of(Long.class, long[].class)
* // given that graph should have of 1 billion nodes, and 150 average adjacency list size
@@ -145,10 +135,8 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* .entries(1_000_000_000L * (150 / 50))
* .entrySize(50 * 8)
* .create();}</pre>
* It is minimum possible (because 3000 friends / 50 friends = 60 is close to 64 "max
* chunks by single entry" limit, and ensures moderate average memory overuse (not more
* than 20%).
* </li>
* It is minimum possible (because 3000 friends / 50 friends = 60 is close to 64 "max chunks by
* single entry" limit, and ensures moderate average memory overuse (not more than 20%). </li>
* </ol>
*
* @param entrySize the "chunk size" in bytes
@@ -164,29 +152,26 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* thrown, because currently {@link ChronicleMap} and {@link ChronicleSet} doesn't support
* resizing.
*
* <ol>
* <li>If key size (and value size, in {@code ChronicleMap} case) is constant, this number
* is equal to the maximum number of entries (because each entry takes exactly one
* "entry size" memory unit).</li>
* <li>If key (and/or value, in {@code ChronicleMap} case) size varies moderately, you
* should pass to this method the maximum number of entries + 5-25%, depending on your data
* properties and configured {@linkplain #keySize(int) key}/{@linkplain
* ChronicleMapBuilder#valueSize(int) value}/{@linkplain #entrySize(int) entry} sizes.</li>
* <li>If your data size varies in a wide range, pass the maximum number of entries
* multiplied by average data size and divided by the configured "entry size" (i. e. chunk
* size). See an example in the documentation to {@link #entrySize(int)} method.</li>
* </ol>
* <ol> <li>If key size (and value size, in {@code ChronicleMap} case) is constant, this number
* is equal to the maximum number of entries (because each entry takes exactly one "entry size"
* memory unit).</li> <li>If key (and/or value, in {@code ChronicleMap} case) size varies
* moderately, you should pass to this method the maximum number of entries + 5-25%, depending
* on your data properties and configured {@linkplain #keySize(int) key}/{@linkplain
* ChronicleMapBuilder#valueSize(int) value}/{@linkplain #entrySize(int) entry} sizes.</li>
* <li>If your data size varies in a wide range, pass the maximum number of entries multiplied
* by average data size and divided by the configured "entry size" (i. e. chunk size). See an
* example in the documentation to {@link #entrySize(int)} method.</li> </ol>
*
* <p>You shouldn't put additional margin over the number, computed according the rules above.
* This bad practice was popularized by {@link HashMap#HashMap(int)}
* and {@link HashSet#HashSet(int)} constructors, which accept "capacity", that should
* be multiplied by "load factor" to obtain actual maximum expected number of entries.
* {@code ChronicleMap} and {@code ChronicleSet} don't have a notion of load factor.
* This bad practice was popularized by {@link HashMap#HashMap(int)} and {@link
* HashSet#HashSet(int)} constructors, which accept "capacity", that should be multiplied by
* "load factor" to obtain actual maximum expected number of entries. {@code ChronicleMap} and
* {@code ChronicleSet} don't have a notion of load factor.
*
* <p>Default value is 2^20 (~ 1 million).
*
* @param entries maximum size of the created maps, in memory allocation units,
* so-called "entry size"
* @param entries maximum size of the created maps, in memory allocation units, so-called "entry
* size"
* @return this builder back
* @see #entrySize(int)
*/
@@ -199,17 +184,17 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
/**
* Configures timeout of locking on {@linkplain #actualSegments(int) segments} of hash
* containers, created by this builder, when performing any queries, as well as bulk operations
* like iteration. If timeout expires, {@link ChronicleHashErrorListener#onLockTimeout(long)}
* is called, and then thread tries to obtain the segment lock one more time, and so in a loop,
* like iteration. If timeout expires, {@link ChronicleHashErrorListener#onLockTimeout(long)} is
* called, and then thread tries to obtain the segment lock one more time, and so in a loop,
* until thread is interrupted. However, you can configure {@linkplain #errorListener(
* ChronicleHashErrorListener) error listener} to throw an exception on the first (or n-th) lock
*ChronicleHashErrorListener) error listener} to throw an exception on the first (or n-th) lock
* acquisition fail.
*
* <p>Default lock time out is 2 seconds.
*
* @param lockTimeOut new lock timeout for segments of containers created by this builder,
* in the given time units
* @param unit time unit of the given lock timeout
* @param lockTimeOut new lock timeout for segments of containers created by this builder, in
* the given time units
* @param unit time unit of the given lock timeout
* @return this builder back
*/
B lockTimeOut(long lockTimeOut, TimeUnit unit);
@@ -233,8 +218,8 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* <p>This method call overrides all previous replication configurations of this builder, made
* either by this method or {@link #channel(ChannelProvider.ChronicleChannel)} method.
*
* @param identifier the new {@linkplain Replica#identifier() identifier} of the containers,
* created by this builder
* @param identifier the new {@linkplain Replica#identifier() identifier} of the containers,
* created by this builder
* @param replicators several replicators, used to sync the data across replicating nodes
* independently
* @return this builder back
@@ -244,10 +229,9 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
B replicators(byte identifier, ReplicationConfig... replicators);
/**
* Configures replication of the hash containers, created by this builder,
* via so called "channels". See <a
* href="https://github.com/OpenHFT/Chronicle-Map#channels-and-channelprovider">the section
* about Channels and ChannelProvider in ChronicleMap manual</a> for more information.
* Configures replication of the hash containers, created by this builder, via so called
* "channels". See <a href="https://github.com/OpenHFT/Chronicle-Map#channels-and-channelprovider">the
* section about Channels and ChannelProvider in ChronicleMap manual</a> for more information.
*
* <p>Another way to establish replication if {@link #replicators(byte, ReplicationConfig[])}
* method.
@@ -269,8 +253,8 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
/**
* Specifies that hash containers, created by this builder, shouldn't replicate their data over
* network. This method call clears any replication configurations, made via {@link
* #replicators(byte, ReplicationConfig[])} or {@link #channel(
* ChannelProvider.ChronicleChannel)} method call.
* #replicators(byte, ReplicationConfig[])} or {@link #channel(ChannelProvider.ChronicleChannel)}
* method call.
*
* <p>By default replication is not configured, so this method is useful <i>only</i> to ensure
* that hash containers, created by some inherited or passed builder, wouldn't replicate. I. e.
@@ -287,6 +271,7 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* replication consensus protocol (conflicting data updates resolution).
*
* <p>Default time provider is {@link TimeProvider#SYSTEM}.
*
* @param timeProvider a new time provider for replication needs
* @return this builder back
* @see #replicators(byte, ReplicationConfig[])
@@ -295,20 +280,20 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
B timeProvider(TimeProvider timeProvider);
/**
* Configures a {@link BytesMarshallerFactory} to be used with
* {@link BytesMarshallableSerializer}, which is a default {@link #objectSerializer
* ObjectSerializer}, to serialize/deserialize data to/from off-heap memory in hash containers,
* created by this builder.
* Configures a {@link BytesMarshallerFactory} to be used with {@link
* BytesMarshallableSerializer}, which is a default {@link #objectSerializer ObjectSerializer},
* to serialize/deserialize data to/from off-heap memory in hash containers, created by this
* builder.
*
* <p>Default {@code BytesMarshallerFactory} is an instance of {@link
* VanillaBytesMarshallerFactory}. This is a convenience configuration method, it has no effect
* on the resulting hash containers, if {@linkplain #keyMarshaller(BytesMarshaller) custom data
* marshallers} are configured, data types extends one of specific serialization interfaces,
* recognized by this builder (e. g. {@code Externalizable} or {@code BytesMarshallable}),
* or {@code ObjectSerializer} is configured.
* recognized by this builder (e. g. {@code Externalizable} or {@code BytesMarshallable}), or
* {@code ObjectSerializer} is configured.
*
* @param bytesMarshallerFactory the marshaller factory to be used with the default
* {@code ObjectSerializer}, i. e. {@code BytesMarshallableSerializer}
* @param bytesMarshallerFactory the marshaller factory to be used with the default {@code
* ObjectSerializer}, i. e. {@code BytesMarshallableSerializer}
* @return this builder back
* @see #objectSerializer(ObjectSerializer)
*/
@@ -318,16 +303,15 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* Configures the serializer used to serialize/deserialize data to/from off-heap memory, when
* specified class doesn't implement a specific serialization interface like {@link
* Externalizable} or {@link BytesMarshallable} (for example, if data is loosely typed and just
* {@code Object} is specified as the data class), or nullable data, and if custom marshaller
* is not {@linkplain #keyMarshaller(BytesMarshaller) configured}, in hash containers, created
* by this builder. Please read {@link ObjectSerializer} docs for more info and available
* options.
* {@code Object} is specified as the data class), or nullable data, and if custom marshaller is
* not {@linkplain #keyMarshaller(BytesMarshaller) configured}, in hash containers, created by
* this builder. Please read {@link ObjectSerializer} docs for more info and available options.
*
* <p>Default serializer is {@link BytesMarshallableSerializer}, configured with the specified
* or default {@link #bytesMarshallerFactory(BytesMarshallerFactory) BytesMarshallerFactory}.
*
* @param objectSerializer the serializer used to serialize loosely typed or nullable data
* if custom marshaller is not configured
* @param objectSerializer the serializer used to serialize loosely typed or nullable data if
* custom marshaller is not configured
* @return this builder back
* @see #bytesMarshallerFactory(BytesMarshallerFactory)
* @see #keyMarshaller(BytesMarshaller)
@@ -336,9 +320,8 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
/**
* Configures the {@code BytesMarshaller} used to serialize/deserialize keys to/from off-heap
* memory in hash containers, created by this builder. See <a
* href="https://github.com/OpenHFT/Chronicle-Map#serialization">the section about serialization
* in ChronicleMap manual</a> for more information.
* memory in hash containers, created by this builder. See <a href="https://github.com/OpenHFT/Chronicle-Map#serialization">the
* section about serialization in ChronicleMap manual</a> for more information.
*
* @param keyMarshaller the marshaller used to serialize keys
* @return this builder back
@@ -347,40 +330,58 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
B keyMarshaller(@NotNull BytesMarshaller<K> keyMarshaller);
/**
* Specifies that key objects, queried with the hash containers, created by this builder,
* are inherently immutable. Keys in {@link ChronicleMap} or {@link ChronicleSet}
* are not required to be immutable, as in ordinary {@link Map} or {@link Set} implementations,
* because they are serialized off-heap. However, {@code ChronicleMap} and {@code ChronicleSet}
* implementations can benefit from the knowledge that keys are not mutated between queries.
* Specifies that key objects, queried with the hash containers, created by this builder, are
* inherently immutable. Keys in {@link ChronicleMap} or {@link ChronicleSet} are not required
* to be immutable, as in ordinary {@link Map} or {@link Set} implementations, because they are
* serialized off-heap. However, {@code ChronicleMap} and {@code ChronicleSet} implementations
* can benefit from the knowledge that keys are not mutated between queries.
*
* <p>By default, {@code ChronicleHashBuilder}s detects immutability automatically
* only for very few standard JDK types (for example, for {@link String}), it is not recommended
* to rely on {@code ChronicleHashBuilder} to be smart enough about this.
* <p>By default, {@code ChronicleHashBuilder}s detects immutability automatically only for very
* few standard JDK types (for example, for {@link String}), it is not recommended to rely on
* {@code ChronicleHashBuilder} to be smart enough about this.
*
* @return this builder back
*/
B immutableKeys();
/**
* Specifies that hash containers, created by this builder, should be a stateless
* implementation, this stateless implementation will be referred to as a stateless client as it
* will not hold any of its own data locally, the stateless client will perform Remote Procedure
* Calls ( RPC ) to another {@link ChronicleMap} or {@link ChronicleSet} which we will refer to
* as the server. The server will hold all your data, the server can not it’s self be a
* stateless client. Your stateless client must be connected to the server via TCP/IP. The
* stateless client will delegate all your method calls to the remote sever. The stateless
* client operations will block, in other words the stateless client will wait for the server to
* send a response before continuing to the next operation. The stateless client could be
* consider to be a ClientProxy to {@link ChronicleMap} or {@link ChronicleSet} running on
* another host
*
* @param isStateless true if you want your container to hold no data
* @return this builder back
*/
B stateless(boolean isStateless);
/**
* Opens a hash container residing the specified file, or creates a new one if the file not yet
* exists and maps its off-heap memory to the file. All changes to the map are persisted to disk
* (this is an operating system guarantee) independently from JVM process lifecycle.
*
* <p>Multiple containers could give access to the same data simultaneously, either inside
* a single JVM or across processes. Access is synchronized correctly across all instances,
* i. e. hash container mapping the data from the first JVM isn't able to modify the data,
* concurrently accessed from the second JVM by another hash container instance, mapping
* the same data.
* <p>Multiple containers could give access to the same data simultaneously, either inside a
* single JVM or across processes. Access is synchronized correctly across all instances, i. e.
* hash container mapping the data from the first JVM isn't able to modify the data,
* concurrently accessed from the second JVM by another hash container instance, mapping the
* same data.
*
* <p>On container's {@link ChronicleHash#close() close()} the data isn't removed, it remains
* on disk and available to be opened again (given the same file name) or during different JVM
* <p>On container's {@link ChronicleHash#close() close()} the data isn't removed, it remains on
* disk and available to be opened again (given the same file name) or during different JVM
* run.
*
* @param file the file with existing hash container or a desired location of a new off-heap
* persisted hash container
* @return a hash container mapped to the given file
* @throws IOException if any IO error, related to off-heap memory allocation or file mapping
* occurs
* occurs
* @see ChronicleHash#file()
* @see ChronicleHash#close()
* @see #create()
@@ -398,4 +399,6 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* @see #create()
*/
C create() throws IOException;
}
@@ -62,7 +62,7 @@ abstract class AbstractChannelReplicator implements Closeable {
throws IOException {
executorService = Executors.newSingleThreadExecutor(
new NamedThreadFactory(name, true));
selector = openSelector();
selector = openSelector(closeables);
throttler = throttlingConfig.throttling(DAYS) > 0 ?
new Throttler(selector,
@@ -70,7 +70,7 @@ abstract class AbstractChannelReplicator implements Closeable {
maxEntrySizeBytes, throttlingConfig.throttling(DAYS)) : null;
}
Selector openSelector() throws IOException {
static Selector openSelector(final CloseablesManager closeables) throws IOException {
Selector result = null;
try {
result = Selector.open();
@@ -81,7 +81,7 @@ abstract class AbstractChannelReplicator implements Closeable {
return result;
}
SocketChannel openSocketChannel() throws IOException {
static SocketChannel openSocketChannel(final CloseablesManager closeables) throws IOException {
SocketChannel result = null;
try {
@@ -130,6 +130,7 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
private V defaultValue = null;
private DefaultValueProvider<K, V> defaultValueProvider = null;
private byte identifier = -1;
private boolean isStateless;
ChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
@@ -731,6 +732,14 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
return this;
}
@Override
public ChronicleMapBuilder<K, V> stateless(boolean isStateless) {
this.isStateless = isStateless;
return this;
}
/**
* Non-public because should be called only after {@link #preMapConstruction()}
*/
@@ -806,6 +815,7 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
return establishReplication(map);
}
private VanillaChronicleMap<K, ?, ?, V, ?, ?> newMap() throws IOException {
preMapConstruction();
if (!useReplication()) {
@@ -6,6 +6,7 @@ import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -23,13 +24,14 @@ import static net.openhft.chronicle.map.StatelessMapClient.EventId.*;
*
* @author Rob Austin.
*/
class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
class StatelessMapClient<K, V> implements ChronicleMap<K, V>, Closeable {
private static final Logger LOG = LoggerFactory.getLogger(StatelessMapClient.class);
private final SocketChannel clientChannel;
public static final ByteBuffer STATELESS_CLIENT_IDENTIFER = ByteBuffer.wrap(new byte[]{-127});
private final CloseablesManager closeables = new CloseablesManager();
public static enum EventId {
HEARTBEAT,
@@ -63,16 +65,19 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
this.keyValueSerializer = keyValueSerializer;
clientChannel = SocketChannel.open();
try {
clientChannel = AbstractChannelReplicator.openSocketChannel(closeables);
clientChannel.connect(remote);
clientChannel.connect(remote);
doHandShaking();
doHandShaking();
} catch (IOException e) {
closeables.closeQuietly();
throw e;
}
}
private void doHandShaking() throws IOException {
private void doHandShaking() throws IOException {
while (STATELESS_CLIENT_IDENTIFER.remaining() > 0) {
clientChannel.write(STATELESS_CLIENT_IDENTIFER);
@@ -98,12 +103,10 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
}
public void close() {
throw new UnsupportedOperationException("This is not supported in the " + this.getClass()
.getSimpleName());
closeables.closeQuietly();
}
long nextUniqueTransaction() {
long time = System.currentTimeMillis();
long nextUniqueTransaction(long time) {
long l = transactionID.get();
if (time > l) {
@@ -135,7 +138,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
writeValue((V) value);
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
return blockingFetch(sizeLocation, 20000).readBoolean();
}
@@ -147,7 +150,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
writeValue(newValue);
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
return blockingFetch(sizeLocation, 20000).readBoolean();
}
@@ -166,7 +169,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
long sizeLocation = writeEvent(SIZE);
// get the data back from the server
return blockingFetch(sizeLocation).readInt();
return blockingFetch(sizeLocation, 20000).readInt();
}
@@ -174,7 +177,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
long sizeLocation = writeEvent(IS_EMPTY);
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
return blockingFetch(sizeLocation, 20000).readBoolean();
}
@@ -183,7 +186,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
writeKey((K) key);
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
return blockingFetch(sizeLocation, 20000).readBoolean();
}
@@ -193,14 +196,14 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
writeValue((V) value);
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
return blockingFetch(sizeLocation, 20000).readBoolean();
}
public synchronized long longSize() {
long sizeLocation = writeEvent(LONG_SIZE);
// get the data back from the server
return blockingFetch(sizeLocation).readLong();
return blockingFetch(sizeLocation, 20000).readLong();
}
@@ -275,9 +278,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
long sizeLocation = writeEvent(CLEAR);
// get the data back from the server
blockingFetch(sizeLocation);
blockingFetch(sizeLocation, 20000);
}
@@ -287,7 +288,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
long sizeLocation = writeEvent(KEY_SET);
// get the data back from the server
return readKeySet(blockingFetch(sizeLocation));
return readKeySet(blockingFetch(sizeLocation, 20000));
}
@@ -297,7 +298,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
long sizeLocation = writeEvent(VALUES);
// get the data back from the server
final Bytes in = blockingFetch(sizeLocation);
final Bytes in = blockingFetch(sizeLocation, 20000);
final long size = in.readStopBit();
@@ -319,7 +320,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
long sizeLocation = writeEvent(ENTRY_SET);
// get the data back from the server
Bytes in = blockingFetch(sizeLocation);
Bytes in = blockingFetch(sizeLocation, 20000);
long size = in.readStopBit();
@@ -454,28 +455,34 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
}
}
private Bytes blockingFetch(long sizeLocation) {
private Bytes blockingFetch(long sizeLocation, long timeOutMs) {
try {
return blockingFetchThrowable(sizeLocation);
return blockingFetchThrowable(sizeLocation, timeOutMs);
} catch (IOException e) {
closeables.closeQuietly();
throw new RuntimeIOException(e);
} catch (Exception e) {
closeables.closeQuietly();
throw e;
}
}
private Bytes blockingFetchThrowable(long sizeLocation) throws IOException {
long transactionId = nextUniqueTransaction();
private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs) throws IOException {
long startTime = System.currentTimeMillis();
long transactionId = nextUniqueTransaction(startTime);
long timeoutTime = startTime + timeOutMs;
LOG.info("sending data with transactionId=" + transactionId);
out.writeLong(transactionId);
writeSizeAt(sizeLocation);
out.buffer().limit((int) out.position());
while (out.buffer().remaining() > 0) {
int write = clientChannel.write(out.buffer());
System.out.println(write);
clientChannel.write(out.buffer());
checkTimeout(timeoutTime);
}
in.clear();
@@ -484,12 +491,14 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
// read size
while (in.buffer().position() < 2) {
clientChannel.read(in.buffer());
checkTimeout(timeoutTime);
}
int size = in.readUnsignedShort();
while (in.buffer().position() < size) {
clientChannel.read(in.buffer());
checkTimeout(timeoutTime);
}
boolean isException = in.readBoolean();
@@ -511,10 +520,14 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
throw runtimeException;
}
return in;
}
private void checkTimeout(long timeoutTime) {
if (timeoutTime < System.currentTimeMillis())
throw new RuntimeTimeOutException();
}
private void writeSizeAt(long locationOfSize) {
long size = out.position() - locationOfSize;
out.writeUnsignedShort(locationOfSize, (int) size - 2);
@@ -531,7 +544,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
}
private V readKey(final long sizeLocation) {
return keyValueSerializer.readValue(blockingFetch(sizeLocation));
return keyValueSerializer.readValue(blockingFetch(sizeLocation, 20000));
}
private void writeValue(V value) {
@@ -564,4 +577,13 @@ class RuntimeIOException extends RuntimeException {
public RuntimeIOException(IOException e) {
super(e);
}
}
class RuntimeTimeOutException extends RuntimeException {
public RuntimeTimeOutException() {
}
}
@@ -729,7 +729,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
SelectableChannel doConnect() throws IOException, InterruptedException {
boolean success = false;
final SocketChannel socketChannel = openSocketChannel();
final SocketChannel socketChannel = openSocketChannel(TcpReplicator.this.closeables);
try {
socketChannel.configureBlocking(false);
@@ -31,8 +31,8 @@ import java.io.IOException;
import java.util.concurrent.TimeUnit;
/**
* {@code ChronicleSetBuilder} manages the whole set of {@link ChronicleSet} configurations,
* could be used as a classic builder and/or factory.
* {@code ChronicleSetBuilder} manages the whole set of {@link ChronicleSet} configurations, could
* be used as a classic builder and/or factory.
*
* <p>{@code ChronicleMapBuilder} is mutable, see a note in {@link ChronicleHashBuilder} interface
* documentation.
@@ -97,9 +97,9 @@ public class ChronicleSetBuilder<E>
* .keySize(10)
* .create();}</pre>
*
* <p>(Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded (and each
* character takes 2 bytes on-heap), because default off-heap {@link String} encoding is UTF-8 in {@code
* ChronicleSet}.)
* <p>(Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded
* (and each character takes 2 bytes on-heap), because default off-heap {@link String} encoding
* is UTF-8 in {@code ChronicleSet}.)
*
* @see #constantKeySizeBySample(Object)
* @see #entrySize(int)
@@ -129,13 +129,13 @@ public class ChronicleSetBuilder<E>
/**
* {@inheritDoc}
*
* <p>In fully default case you can expect entry size to be about 120-130 bytes. But it is strongly
* recommended always to configure {@linkplain #keySize(int) key size}, if they couldn't be derived
* statically.
* <p>In fully default case you can expect entry size to be about 120-130 bytes. But it is
* strongly recommended always to configure {@linkplain #keySize(int) key size}, if they
* couldn't be derived statically.
*
* <p>If entry size is not configured explicitly by calling this method, it is computed based on
* {@linkplain #metaDataBytes(int) meta data bytes}, plus {@linkplain #keySize(int) key size}, plus a few
* bytes required by implementations.
* {@linkplain #metaDataBytes(int) meta data bytes}, plus {@linkplain #keySize(int) key size},
* plus a few bytes required by implementations.
*/
@Override
public ChronicleSetBuilder<E> entrySize(int entrySize) {
@@ -253,6 +253,12 @@ public class ChronicleSetBuilder<E>
return this;
}
@Override
public ChronicleSetBuilder<E> stateless(boolean isStateless) {
chronicleMapBuilder.stateless(isStateless);
return this;
}
@Override
public ChronicleSet<E> create(File file) throws IOException {
final ChronicleMap<E, DummyValue> map = chronicleMapBuilder.create(file);