diff --git a/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java index 107ab10c9..209b0110a 100644 --- a/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java @@ -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. Chronicle hash container configurations. + * This interface defines the meaning of configurations, common to {@link ChronicleMapBuilder} and + * {@link ChronicleSetBuilder}, i. e. Chronicle hash container configurations. * *

{@code ChronicleHashBuilder} is mutable. Configuration methods mutate the builder and return * the builder itself 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 the type of keys in hash containers, created by this builder - * @param the container type, created by this builder, i. e. {@link ChronicleMap} - * or {@link ChronicleSet} - * @param the concrete builder type, i. e. {@link ChronicleMapBuilder} - * or {@link ChronicleSetBuilder} + * @param the container type, created by this builder, i. e. {@link ChronicleMap} or {@link + * ChronicleSet} + * @param the concrete builder type, i. e. {@link ChronicleMapBuilder} or {@link + * ChronicleSetBuilder} */ public interface ChronicleHashBuilder> 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 ChronicleHashBuilderIf 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, + *

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 ChronicleHashBuilderIf 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. + *

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. * *

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{@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 entry size - * 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 entry size 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. * - *

There are three major patterns of this configuration usage: - *

    - *
  1. 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.
    2. - *
  2. 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.
    3. - *
  3. Key (and/or value size, in {@code ChronicleMap} case) varies in a wide range. - * Then it's best to use entry size configuration in chunk mode. 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. + *

    There are three major patterns of this configuration usage:

    1. 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.
    2. 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.
    3. Key (and/or value size, in {@code + * ChronicleMap} case) varies in a wide range. Then it's best to use entry size configuration in + * chunk mode. 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. * - *

      However, remember that - *

        - *
      • 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.
        • - *
      • 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.
        • - *
      • The number of "entries" {@linkplain #entries(long) in the whole map} and - * {@linkplain #actualEntriesPerSegment(int) per segment} is actually the number - * "chunks".
        • - *
      + *

      However, remember that

      • 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.
      • 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.
        • + *
      • The number of "entries" {@linkplain #entries(long) in the whole map} and {@linkplain + * #actualEntriesPerSegment(int) per segment} is actually the number "chunks".
      * - *

      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: 

      {@code
+     * 
      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: 
      {@code
      * Map 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
-     *     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%).
-     *
      4. + * 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%). *
    * * @param entrySize the "chunk size" in bytes @@ -164,29 +152,26 @@ public interface ChronicleHashBuilder - *
  4. 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).
    5. - *
  5. 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.
    6. - *
  6. 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.
    7. - *
+ *
  1. 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).
  2. 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.
    3. + *
  3. 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.
* *

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. * *

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 ChronicleHashBuilderDefault 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 ChronicleHashBuilderThis 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 ChronicleHashBuilderthe section - * about Channels and ChannelProvider in ChronicleMap manual for more information. + * Configures replication of the hash containers, created by this builder, via so called + * "channels". See the + * section about Channels and ChannelProvider in ChronicleMap manual for more information. * *

Another way to establish replication if {@link #replicators(byte, ReplicationConfig[])} * method. @@ -269,8 +253,8 @@ public interface ChronicleHashBuilderBy default replication is not configured, so this method is useful only to ensure * that hash containers, created by some inherited or passed builder, wouldn't replicate. I. e. @@ -287,6 +271,7 @@ public interface ChronicleHashBuilderDefault 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 ChronicleHashBuilderDefault {@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 ChronicleHashBuilderDefault 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 ChronicleHashBuilderthe section about serialization - * in ChronicleMap manual for more information. + * memory in hash containers, created by this builder. See the + * section about serialization in ChronicleMap manual for more information. * * @param keyMarshaller the marshaller used to serialize keys * @return this builder back @@ -347,40 +330,58 @@ public interface ChronicleHashBuilder 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. * - *

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. + *

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. * - *

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. + *

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. * - *

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 + *

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 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 { diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index b2d016bfc..ff1b6efff 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -130,6 +130,7 @@ public class ChronicleMapBuilder implements Cloneable, private V defaultValue = null; private DefaultValueProvider defaultValueProvider = null; private byte identifier = -1; + private boolean isStateless; ChronicleMapBuilder(Class keyClass, Class valueClass) { keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); @@ -731,6 +732,14 @@ public class ChronicleMapBuilder implements Cloneable, return this; } + + + @Override + public ChronicleMapBuilder 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 implements Cloneable, return establishReplication(map); } + private VanillaChronicleMap newMap() throws IOException { preMapConstruction(); if (!useReplication()) { diff --git a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java index 8ca2af5f3..c986ee1ce 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java @@ -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 implements ChronicleMap { +class StatelessMapClient implements ChronicleMap, 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 implements ChronicleMap { 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 implements ChronicleMap { } 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 implements ChronicleMap { 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 implements ChronicleMap { writeValue(newValue); // get the data back from the server - return blockingFetch(sizeLocation).readBoolean(); + return blockingFetch(sizeLocation, 20000).readBoolean(); } @@ -166,7 +169,7 @@ class StatelessMapClient implements ChronicleMap { 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 implements ChronicleMap { 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 implements ChronicleMap { 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 implements ChronicleMap { 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 implements ChronicleMap { long sizeLocation = writeEvent(CLEAR); // get the data back from the server - - blockingFetch(sizeLocation); - + blockingFetch(sizeLocation, 20000); } @@ -287,7 +288,7 @@ class StatelessMapClient implements ChronicleMap { 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 implements ChronicleMap { 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 implements ChronicleMap { 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 implements ChronicleMap { } } - 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 implements ChronicleMap { // 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 implements ChronicleMap { 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 implements ChronicleMap { } 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() { + + } + +} \ No newline at end of file diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 2b3a506e4..ca5ed10d7 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -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); diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index 30fa67ca6..1988dd27a 100644 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -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. * *

{@code ChronicleMapBuilder} is mutable, see a note in {@link ChronicleHashBuilder} interface * documentation. @@ -97,9 +97,9 @@ public class ChronicleSetBuilder * .keySize(10) * .create();} * - *

(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}.) + *

(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 /** * {@inheritDoc} * - *

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. + *

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. * *

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 entrySize(int entrySize) { @@ -253,6 +253,12 @@ public class ChronicleSetBuilder return this; } + @Override + public ChronicleSetBuilder stateless(boolean isStateless) { + chronicleMapBuilder.stateless(isStateless); + return this; + } + @Override public ChronicleSet create(File file) throws IOException { final ChronicleMap map = chronicleMapBuilder.create(file);