mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
Merge remote-tracking branch 'origin/master'
This commit is contained in:
Peter Lawrey
@@ -0,0 +1,64 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle;
|
||||
|
||||
import net.openhft.chronicle.map.ChronicleMap;
|
||||
import net.openhft.chronicle.set.ChronicleSet;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* This interface defines common {@link ChronicleMap} and {@link ChronicleSet}, related to off-heap
|
||||
* memory management and file-mapping. Not usable by itself.
|
||||
*/
|
||||
public interface ChronicleHash extends Closeable {
|
||||
/**
|
||||
* Returns the file this hash container mapped to, i. e. when it is created by
|
||||
* {@link ChronicleHashBuilder#create()} call, or {@code null} if it is purely in-memory,
|
||||
* i. e. if it is created by {@link ChronicleHashBuilder#create()} call.
|
||||
*
|
||||
* @return the file this {@link ChronicleMap} or {@link ChronicleSet} is mapped to,
|
||||
* or {@code null} if it is not mapped to any file
|
||||
* @see ChronicleHashBuilder#create(File)
|
||||
* @see ChronicleHashBuilder#create()
|
||||
*/
|
||||
File file();
|
||||
|
||||
/**
|
||||
* Releases the off-heap memory, used by this hash container and resources, used by replication,
|
||||
* if any. However, if hash container (hence off-heap memory, used by it) is mapped to the file
|
||||
* and there are other instances mapping the same data on the server across JVMs, the memory
|
||||
* won't be actually freed on operation system level. I. e. this method call doesn't affect
|
||||
* other {@link ChronicleMap} or {@link ChronicleSet} instances mapping the same data.
|
||||
*
|
||||
* <p>If you won't call this method, memory would be held at least until next garbage
|
||||
* collection. This could be a problem if, for example, you target rare garbage collections,
|
||||
* but load and drop {@code ChronicleHash}es regularly.
|
||||
*
|
||||
* <p>
|
||||
* TODO what about commit guarantees, when ChronicleMap is used with memory-mapped files, if
|
||||
* {@code ChronicleMap}/{@code ChronicleSet} closed/not?
|
||||
*
|
||||
* <p>After this method call behaviour of <i>all</i> methods of {@code ChronicleMap}
|
||||
* or {@code ChronicleSet} is undefined. <i>Any</i> method call on the map might throw
|
||||
* <i>any</i> exception, or even JVM crash. Shortly speaking, don't call use map closing.
|
||||
*/
|
||||
@Override
|
||||
void close() throws IOException;
|
||||
}
|
||||
@@ -1,11 +1,16 @@
|
||||
package net.openhft.chronicle;
|
||||
|
||||
import net.openhft.chronicle.map.*;
|
||||
import net.openhft.chronicle.map.ChannelProvider;
|
||||
import net.openhft.chronicle.map.ChronicleMap;
|
||||
import net.openhft.chronicle.map.ChronicleMapBuilder;
|
||||
import net.openhft.chronicle.map.ReplicationConfig;
|
||||
import net.openhft.chronicle.set.ChronicleSet;
|
||||
import net.openhft.chronicle.set.ChronicleSetBuilder;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
import net.openhft.lang.io.serialization.*;
|
||||
import net.openhft.lang.io.serialization.BytesMarshaller;
|
||||
import net.openhft.lang.io.serialization.BytesMarshallerFactory;
|
||||
import net.openhft.lang.io.serialization.ObjectSerializer;
|
||||
import net.openhft.lang.model.Byteable;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
@@ -17,17 +22,15 @@ 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.
|
||||
*
|
||||
* @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, B extends ChronicleHashBuilder<K, C, B>>
|
||||
extends Cloneable {
|
||||
public interface ChronicleHashBuilder<K, C extends ChronicleHash,
|
||||
B extends ChronicleHashBuilder<K, C, B>> extends Cloneable {
|
||||
|
||||
B clone();
|
||||
|
||||
@@ -40,14 +43,13 @@ public interface ChronicleHashBuilder<K, C, B extends ChronicleHashBuilder<K, C,
|
||||
B minSegments(int minSegments);
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* 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.
|
||||
*
|
||||
* <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.
|
||||
* <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.
|
||||
*
|
||||
* @param keySize number of bytes, taken by serialized form of keys
|
||||
* @return this builder back
|
||||
@@ -57,15 +59,15 @@ public interface ChronicleHashBuilder<K, C, B extends ChronicleHashBuilder<K, C,
|
||||
B keySize(int keySize);
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* 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.
|
||||
*
|
||||
* <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.
|
||||
* <p>If key size varies, method {@link #keySize(int)} or {@link #entrySize(int)} should be called instead
|
||||
* of this one.
|
||||
*
|
||||
* @param sampleKey the sample key
|
||||
* @return this builder back
|
||||
@@ -74,55 +76,43 @@ public interface ChronicleHashBuilder<K, C, B extends ChronicleHashBuilder<K, C,
|
||||
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.
|
||||
* <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.
|
||||
*
|
||||
* <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
|
||||
@@ -130,11 +120,8 @@ public interface ChronicleHashBuilder<K, C, B extends ChronicleHashBuilder<K, C,
|
||||
* .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>
|
||||
* </ol>
|
||||
* 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
|
||||
* @return this builder back
|
||||
@@ -143,35 +130,28 @@ public interface ChronicleHashBuilder<K, C, B extends ChronicleHashBuilder<K, C,
|
||||
B entrySize(int entrySize);
|
||||
|
||||
/**
|
||||
* Configures the maximum number of {@linkplain #entrySize(int) "entry size chunks"}, which
|
||||
* could be taken by the maximum number of entries, inserted into the hash containers, created
|
||||
* by this builder. If you try to insert more data, {@link IllegalStateException} might be
|
||||
* thrown, because currently {@link ChronicleMap} and {@link ChronicleSet} doesn't support
|
||||
* resizing.
|
||||
* Configures the maximum number of {@linkplain #entrySize(int) "entry size chunks"}, which could be taken
|
||||
* by the maximum number of entries, inserted into the hash containers, created by this builder. If you
|
||||
* try to insert more data, {@link IllegalStateException} might be 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.
|
||||
* <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.
|
||||
*
|
||||
* <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)
|
||||
*/
|
||||
@@ -189,7 +169,9 @@ public interface ChronicleHashBuilder<K, C, B extends ChronicleHashBuilder<K, C,
|
||||
|
||||
B metaDataBytes(int metaDataBytes);
|
||||
|
||||
B replicators(byte identifier, Replicator... replicators);
|
||||
B replicators(byte identifier, ReplicationConfig... replicators);
|
||||
|
||||
B channel(ChannelProvider.ChronicleChannel chronicleChannel);
|
||||
|
||||
B timeProvider(TimeProvider timeProvider);
|
||||
|
||||
@@ -200,15 +182,15 @@ public interface ChronicleHashBuilder<K, C, B extends ChronicleHashBuilder<K, C,
|
||||
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
|
||||
*/
|
||||
|
||||
@@ -17,7 +17,7 @@
|
||||
package net.openhft.chronicle;
|
||||
|
||||
import net.openhft.chronicle.map.ChronicleMap;
|
||||
import net.openhft.chronicle.map.ChronicleSet;
|
||||
import net.openhft.chronicle.set.ChronicleSet;
|
||||
|
||||
import java.io.ObjectStreamException;
|
||||
import java.io.Serializable;
|
||||
|
||||
@@ -197,8 +197,9 @@ abstract class AbstractChannelReplicator implements Closeable {
|
||||
|
||||
|
||||
/**
|
||||
* checks the number of bytes written in this interval, if this number of bytes exceeds a threshold,
|
||||
* the selected will de-register the socket that is being written to, until the interval is finished.
|
||||
* checks the number of bytes written in this interval, if this number of bytes exceeds a
|
||||
* threshold, the selected will de-register the socket that is being written to, until the
|
||||
* interval is finished.
|
||||
*
|
||||
* @throws ClosedChannelException
|
||||
*/
|
||||
@@ -261,12 +262,22 @@ abstract class AbstractChannelReplicator implements Closeable {
|
||||
|
||||
@Override
|
||||
public boolean onEntry(final Bytes entry, final int chronicleId) {
|
||||
|
||||
long pos0 = in.position();
|
||||
|
||||
// used to denote that this is not a stateless map event
|
||||
in.writeByte(StatelessMapClient.EventId.STATEFUL_UPDATE.ordinal());
|
||||
|
||||
long sizeLocation = in.position();
|
||||
|
||||
// this is where we will store the size of the entry
|
||||
in.skip(SIZE_OF_SHORT);
|
||||
|
||||
final long start = in.position();
|
||||
externalizable.writeExternalEntry(entry, in, chronicleId);
|
||||
|
||||
if (in.position() == start) {
|
||||
in.position(in.position() - SIZE_OF_SHORT);
|
||||
in.position(pos0);
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -276,7 +287,10 @@ abstract class AbstractChannelReplicator implements Closeable {
|
||||
if (entrySize > MAX_UNSIGNED_SHORT)
|
||||
throw new IllegalStateException("entry too large, the entry size=" + entrySize + ", " +
|
||||
"entries are limited to a size of " + MAX_UNSIGNED_SHORT);
|
||||
in.writeUnsignedShort(start - SIZE_OF_SHORT, entrySize);
|
||||
|
||||
if (LOG.isDebugEnabled())
|
||||
LOG.debug("sending entry of entrySize=" + entrySize);
|
||||
in.writeUnsignedShort(sizeLocation, entrySize);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
+5
-5
@@ -45,8 +45,8 @@ import static net.openhft.chronicle.map.Replica.ModificationNotifier.NOP;
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public final class ReplicatingChannel implements Closeable {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ReplicatingChannel.class.getName());
|
||||
public final class ChannelProvider implements Closeable {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ChannelProvider.class.getName());
|
||||
|
||||
private static final byte BOOTSTRAP_MESSAGE = 'B';
|
||||
final EntryExternalizable asEntryExternalizable = new EntryExternalizable() {
|
||||
@@ -152,7 +152,7 @@ public final class ReplicatingChannel implements Closeable {
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
ReplicatingChannel.this.close();
|
||||
ChannelProvider.this.close();
|
||||
}
|
||||
};
|
||||
private final int maxEntrySize;
|
||||
@@ -169,7 +169,7 @@ public final class ReplicatingChannel implements Closeable {
|
||||
private final Set<AbstractChannelReplicator> replicators =
|
||||
new CopyOnWriteArraySet<AbstractChannelReplicator>();
|
||||
|
||||
ReplicatingChannel(ReplicatingChannelBuilder builder) {
|
||||
ChannelProvider(ChannelProviderBuilder builder) {
|
||||
localIdentifier = builder.identifier;
|
||||
maxEntrySize = builder.maxEntrySize;
|
||||
chronicleChannels = new Replica[builder.maxNumberOfChronicles];
|
||||
@@ -407,7 +407,7 @@ public final class ReplicatingChannel implements Closeable {
|
||||
|
||||
@Override
|
||||
protected Closeable applyTo(ChronicleMapBuilder builder,
|
||||
Replica map, EntryExternalizable entryExternalizable) {
|
||||
Replica map, EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) {
|
||||
if (builder.entrySize() > maxEntrySize) {
|
||||
throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " +
|
||||
"maxEntrySize=" + maxEntrySize + " was specified, but map with " +
|
||||
+29
-21
@@ -21,46 +21,54 @@ import java.io.IOException;
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public final class ReplicatingChannelBuilder {
|
||||
public final class ChannelProviderBuilder {
|
||||
|
||||
final byte identifier;
|
||||
final int maxEntrySize;
|
||||
byte identifier;
|
||||
int maxEntrySize = 1024;
|
||||
int maxNumberOfChronicles = 128;
|
||||
private UdpReplicationConfig udpReplicationConfig = null;
|
||||
private TcpReplicationConfig tcpReplicationConfig = null;
|
||||
|
||||
ReplicatingChannelBuilder(byte identifier, final int maxEntrySize) {
|
||||
this.identifier = identifier;
|
||||
|
||||
|
||||
public int maxEntrySize() {
|
||||
return maxEntrySize;
|
||||
}
|
||||
|
||||
public ChannelProviderBuilder maxEntrySize(int maxEntrySize) {
|
||||
this.maxEntrySize = maxEntrySize;
|
||||
if (identifier <= 0) {
|
||||
throw new IllegalArgumentException("Identifier must be positive, identifier=" +
|
||||
identifier);
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChannelProviderBuilder replicators(byte identifier, ReplicationConfig... replicationConfigs) {
|
||||
this.identifier = identifier;
|
||||
|
||||
|
||||
for (ReplicationConfig replicationConfig : replicationConfigs) {
|
||||
|
||||
if (replicationConfig instanceof TcpReplicationConfig) {
|
||||
this.tcpReplicationConfig = (TcpReplicationConfig) replicationConfig;
|
||||
} else if (replicationConfig instanceof UdpReplicationConfig) {
|
||||
this.udpReplicationConfig = (UdpReplicationConfig) replicationConfig;
|
||||
} else
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
}
|
||||
|
||||
public ReplicatingChannelBuilder udpReplication(UdpReplicationConfig replicationConfig) {
|
||||
this.udpReplicationConfig = replicationConfig;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ReplicatingChannelBuilder tcpReplication(TcpReplicationConfig replicationConfig) {
|
||||
this.tcpReplicationConfig = replicationConfig;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ReplicatingChannelBuilder maxNumberOfChronicles(int maxNumberOfChronicles) {
|
||||
public ChannelProviderBuilder maxNumberOfChronicles(int maxNumberOfChronicles) {
|
||||
this.maxNumberOfChronicles = maxNumberOfChronicles;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ReplicatingChannel create() throws IOException {
|
||||
final ReplicatingChannel replicatingCluster = new ReplicatingChannel(this);
|
||||
public ChannelProvider create() throws IOException {
|
||||
final ChannelProvider replicatingCluster = new ChannelProvider(this);
|
||||
if (tcpReplicationConfig != null) {
|
||||
final TcpReplicator tcpReplicator = new TcpReplicator(
|
||||
replicatingCluster.asReplica,
|
||||
replicatingCluster.asEntryExternalizable,
|
||||
tcpReplicationConfig,
|
||||
maxEntrySize);
|
||||
maxEntrySize, null);
|
||||
replicatingCluster.add(tcpReplicator);
|
||||
}
|
||||
|
||||
@@ -16,24 +16,103 @@
|
||||
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.chronicle.ChronicleHash;
|
||||
import net.openhft.chronicle.map.serialization.BytesReader;
|
||||
import net.openhft.lang.io.Bytes;
|
||||
import net.openhft.lang.io.serialization.BytesMarshaller;
|
||||
import net.openhft.lang.io.serialization.ObjectFactory;
|
||||
import net.openhft.lang.model.Byteable;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, Closeable {
|
||||
/**
|
||||
* Extension of {@link ConcurrentMap} interface, stores the data off-heap.
|
||||
*
|
||||
* <p>For information on
|
||||
* <ul>
|
||||
* <li>how to construct a {@code ChronicleMap}</li>
|
||||
* <li>{@code ChronicleMap} flavors and properties</li>
|
||||
* <li>available configurations</li>
|
||||
* </ul>
|
||||
* see {@link ChronicleMapBuilder} documentation.
|
||||
*
|
||||
* <p>Functionally this interface defines some methods supporting garbage-free off-heap programming:
|
||||
* {@link #getUsing(Object, Object)}, {@link #acquireUsing(Object, Object)}.
|
||||
*
|
||||
* <p>Note that {@code ChronicleMap} extends {@link Closeable}, don't forget
|
||||
* to {@linkplain #close() close} map when it is no longer needed.
|
||||
*
|
||||
* @param <K> the map key type
|
||||
* @param <V> the map value type
|
||||
*/
|
||||
public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
|
||||
/**
|
||||
* @return size as a long.
|
||||
* @see java.util.Map size()
|
||||
* Returns the number of entries in this map.
|
||||
*
|
||||
* @return number of entries in this map
|
||||
* @see Map#size()
|
||||
*/
|
||||
long longSize();
|
||||
|
||||
/**
|
||||
* Get a value for a key if available. If the value is Byteable, it will be assigned to
|
||||
* reference the value, instead of copying the data for zero copy access to the collection.
|
||||
* Returns the value to which the specified key is mapped, or {@code null} if this map contains
|
||||
* no mapping for the key.
|
||||
*
|
||||
* @param key to lookup.
|
||||
* @param value to reuse if possible. If null, a new object will be created.
|
||||
* @return value found or null if not.
|
||||
* <p>If the specified key if absent in the map, {@linkplain
|
||||
* ChronicleMapBuilder#defaultValue(Object) default value} is taken or {@linkplain
|
||||
* ChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) default value provider}
|
||||
* is called (depending on what was configured last in {@code ChronicleMapBuilder}).
|
||||
* If it results to something different from {@code null}, this object is put to this map for
|
||||
* the specified key, and then returned from this {@code get()} call. Otherwise, if default
|
||||
* value is {@code null} and default value provider returns {@code null}, the map remains
|
||||
* unchanged and {@code null} is returned.
|
||||
*
|
||||
* <p>If the value class allows reusing, particularly if it is a {@link Byteable} subclass,
|
||||
* consider {@link #getUsing(Object, Object)} method instead of this to reduce garbage creation.
|
||||
*
|
||||
* @param key the key whose associated value is to be returned
|
||||
* @return the value to which the specified key is mapped after this method call,
|
||||
* or {@code null} if no value is mapped
|
||||
* @see #getUsing(Object, Object)
|
||||
*/
|
||||
@Override
|
||||
V get(Object key);
|
||||
|
||||
/**
|
||||
* Returns the value to which the specified key is mapped, read to the provided {@code value}
|
||||
* object, if possible, or returns {@code null}, if this map contains no mapping for the key.
|
||||
*
|
||||
* <p>If the specified key is present in the map, the value data is read to the provided
|
||||
* {@code value} object via value marshaller's {@link BytesMarshaller#read(Bytes, Object)
|
||||
* read(Bytes, value)} or value reader's {@link BytesReader#read(Bytes, long, Object)
|
||||
* read(Bytes, size, value)} method, depending on what deserialization strategy is configured
|
||||
* on this map builder. If the value deserializer is able to reuse the given {@code value}
|
||||
* object, calling this method instead of {@link #get(Object)} could help to reduce garbage
|
||||
* creation.
|
||||
*
|
||||
* <p>The provided {@code value} object is allowed to be {@code null}, in this case
|
||||
* {@code map.getUsing(key, null)} call is semantically equivalent to simple
|
||||
* {@code map.get(key)} call.
|
||||
*
|
||||
* <p>If the specified key if absent in the map, {@linkplain
|
||||
* ChronicleMapBuilder#defaultValue(Object) default value} is taken or {@linkplain
|
||||
* ChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) default value provider}
|
||||
* is called. If it results to something different from {@code null}, this object is put
|
||||
* to this map for the specified key, and then a new value is returned from this
|
||||
* {@code getUsing()} call. Otherwise, if default value is {@code null} and default value
|
||||
* provider, if specified, returns {@code null}, the map remains unchanged and {@code null}
|
||||
* is returned. The provided {@code value} is untouched anyway in this case.
|
||||
*
|
||||
* @param key the key whose associated value is to be returned
|
||||
* @param value the object to read value data in, if possible
|
||||
* @return the value mapped to the specified key after this method call, or {@code null} if no
|
||||
* value is mapped
|
||||
* @see #get(Object)
|
||||
* @see ChronicleMapBuilder#valueMarshallerAndFactory(BytesMarshaller, ObjectFactory)
|
||||
*/
|
||||
V getUsing(K key, V value);
|
||||
|
||||
@@ -41,15 +120,10 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, Closeable {
|
||||
* Acquire a value for a key, creating if absent. If the value is Byteable, it will be assigned
|
||||
* to reference the value, instead of copying the data.
|
||||
*
|
||||
* @param key to lookup.
|
||||
* @param key the key whose associated value is to be returned
|
||||
* @param value to reuse if possible. If null, a new object will be created.
|
||||
* @return value created or found.
|
||||
*/
|
||||
V acquireUsing(K key, V value);
|
||||
|
||||
/**
|
||||
* @return The file or directory for this ChronicleMap
|
||||
*/
|
||||
File file();
|
||||
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ import net.openhft.chronicle.ChronicleHashBuilder;
|
||||
import net.openhft.chronicle.ChronicleHashErrorListener;
|
||||
import net.openhft.chronicle.ChronicleHashErrorListeners;
|
||||
import net.openhft.chronicle.TimeProvider;
|
||||
import net.openhft.chronicle.map.serialization.AgileBytesMarshaller;
|
||||
import net.openhft.chronicle.map.serialization.MetaBytesInterop;
|
||||
import net.openhft.chronicle.map.serialization.MetaBytesWriter;
|
||||
import net.openhft.chronicle.map.serialization.MetaProvider;
|
||||
@@ -555,11 +556,21 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
|
||||
return toString().hashCode();
|
||||
}
|
||||
|
||||
public ChronicleMapBuilder<K, V> replicators(byte identifier, Replicator... replicators) {
|
||||
public ChronicleMapBuilder<K, V> replicators(byte identifier, ReplicationConfig... replicationConfigs) {
|
||||
|
||||
this.identifier = identifier;
|
||||
this.replicators.clear();
|
||||
for (Replicator replicator : replicators) {
|
||||
for (ReplicationConfig replicationConfig : replicationConfigs) {
|
||||
|
||||
Replicator replicator;
|
||||
if (replicationConfig instanceof TcpReplicationConfig) {
|
||||
replicator = Replicators.tcp((TcpReplicationConfig) replicationConfig);
|
||||
} else if (replicationConfig instanceof UdpReplicationConfig) {
|
||||
replicator = Replicators.udp((UdpReplicationConfig) replicationConfig);
|
||||
} else
|
||||
throw new UnsupportedOperationException();
|
||||
|
||||
|
||||
this.replicators.put(replicator.getClass(), replicator);
|
||||
}
|
||||
|
||||
@@ -567,10 +578,10 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
|
||||
}
|
||||
|
||||
|
||||
public ChronicleMapBuilder<K, V> channel(ReplicatingChannel.ChronicleChannel replicator) {
|
||||
this.identifier = replicator.identifier();
|
||||
public ChronicleMapBuilder<K, V> channel(ChannelProvider.ChronicleChannel chronicleChannel) {
|
||||
this.identifier = chronicleChannel.identifier();
|
||||
this.replicators.clear();
|
||||
replicators.put(replicator.getClass(), replicator);
|
||||
replicators.put(chronicleChannel.getClass(), chronicleChannel);
|
||||
|
||||
return this;
|
||||
}
|
||||
@@ -629,6 +640,13 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChronicleMapBuilder<K, V> valueMarshallerAndFactory(
|
||||
@NotNull AgileBytesMarshaller<V> valueMarshaller,
|
||||
@NotNull ObjectFactory<V> valueFactory) {
|
||||
valueBuilder.agileMarshaller(valueMarshaller, valueFactory);
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChronicleMapBuilder<K, V> valueFactory(@NotNull ObjectFactory<V> valueFactory) {
|
||||
valueBuilder.factory(valueFactory);
|
||||
return this;
|
||||
@@ -773,7 +791,7 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
|
||||
return identifier != -1;
|
||||
}
|
||||
|
||||
private void preMapConstruction() {
|
||||
void preMapConstruction() {
|
||||
keyBuilder.objectSerializer(objectSerializer());
|
||||
valueBuilder.objectSerializer(objectSerializer());
|
||||
|
||||
@@ -792,7 +810,7 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
|
||||
if (map instanceof ReplicatedChronicleMap) {
|
||||
ReplicatedChronicleMap result = (ReplicatedChronicleMap) map;
|
||||
for (Replicator replicator : replicators.values()) {
|
||||
Closeable token = replicator.applyTo(this, result, result);
|
||||
Closeable token = replicator.applyTo(this, result, result, map);
|
||||
if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) {
|
||||
LOG.warn(
|
||||
"MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " +
|
||||
|
||||
@@ -1,8 +0,0 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.util.Set;
|
||||
|
||||
public interface ChronicleSet<E> extends Set<E>, Closeable {
|
||||
public long longSize();
|
||||
}
|
||||
@@ -25,7 +25,6 @@ import static net.openhft.chronicle.map.ConcurrentExpiryMap.getDefaultAddress;
|
||||
import static net.openhft.chronicle.map.DiscoveryNodeBytesMarshallable.ProposedNodes;
|
||||
import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES;
|
||||
import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
|
||||
import static net.openhft.chronicle.map.Replicators.tcp;
|
||||
import static net.openhft.chronicle.map.UdpReplicationConfig.simple;
|
||||
|
||||
/**
|
||||
@@ -212,7 +211,7 @@ public class NodeDiscovery {
|
||||
return ChronicleMapBuilder.of(Integer.class,
|
||||
CharSequence.class)
|
||||
.entries(20000L)
|
||||
.replicators(identifier, tcp(tcpConfig)).create();
|
||||
.replicators(identifier, tcpConfig).create();
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public interface ReplicationConfig<B> {
|
||||
|
||||
|
||||
}
|
||||
@@ -41,6 +41,7 @@ public abstract class Replicator {
|
||||
* map configurations, not accessible via the map instance.
|
||||
* @param map a replicated map instance. Provides basic tools for replication implementation.
|
||||
* @param entryExternalizable the callback for ser/deser implementation in the replicator
|
||||
* @param chronicleMap
|
||||
* @return a {@code Closeable} token to control replication resources.
|
||||
* It should be closed on closing the replicated map.
|
||||
* @throws java.io.IOException if an io error occurred during the replicator setup
|
||||
@@ -49,6 +50,6 @@ public abstract class Replicator {
|
||||
* applied to a map (or the specified number of maps)
|
||||
*/
|
||||
protected abstract Closeable applyTo(ChronicleMapBuilder builder,
|
||||
Replica map, Replica.EntryExternalizable entryExternalizable)
|
||||
Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap)
|
||||
throws IOException;
|
||||
}
|
||||
|
||||
@@ -20,31 +20,39 @@ package net.openhft.chronicle.map;
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
|
||||
public final class Replicators {
|
||||
final class Replicators {
|
||||
|
||||
private Replicators() {
|
||||
}
|
||||
|
||||
public static Replicator tcp(final TcpReplicationConfig replicationConfig) {
|
||||
static Replicator tcp(final TcpReplicationConfig replicationConfig) {
|
||||
return new Replicator() {
|
||||
|
||||
@Override
|
||||
protected Closeable applyTo(ChronicleMapBuilder builder,
|
||||
Replica map, Replica.EntryExternalizable entryExternalizable)
|
||||
Replica replica, Replica.EntryExternalizable entryExternalizable,
|
||||
final ChronicleMap chronicleMap)
|
||||
throws IOException {
|
||||
return new TcpReplicator(map, entryExternalizable, replicationConfig,
|
||||
builder.entrySize());
|
||||
|
||||
final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder
|
||||
.keyBuilder, builder.valueBuilder);
|
||||
|
||||
StatelessServerConnector statelessServerConnector = new StatelessServerConnector
|
||||
(keyValueSerializer, chronicleMap);
|
||||
|
||||
return new TcpReplicator(replica, entryExternalizable, replicationConfig,
|
||||
builder.entrySize(), statelessServerConnector);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
public static Replicator udp(
|
||||
static Replicator udp(
|
||||
final UdpReplicationConfig replicationConfig) {
|
||||
return new Replicator() {
|
||||
|
||||
@Override
|
||||
protected Closeable applyTo(ChronicleMapBuilder builder,
|
||||
Replica map, Replica.EntryExternalizable entryExternalizable)
|
||||
Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap)
|
||||
throws IOException {
|
||||
return new UdpReplicator(map, entryExternalizable, replicationConfig,
|
||||
builder.entrySize());
|
||||
|
||||
@@ -80,7 +80,7 @@ final class SerializationBuilder<E> implements Cloneable {
|
||||
if (role == Role.VALUE) {
|
||||
factory = concreteClass(eClass) && marshallerUseFactory(eClass) ?
|
||||
new NewInstanceObjectFactory<E>(eClass) :
|
||||
NullObjectFactory.INSTANCE;
|
||||
(ObjectFactory<E>) NullObjectFactory.<E>of();
|
||||
}
|
||||
|
||||
if (concreteClass(eClass) && Byteable.class.isAssignableFrom(eClass)) {
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.chronicle.map.serialization.*;
|
||||
import net.openhft.chronicle.map.threadlocal.Provider;
|
||||
import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies;
|
||||
import net.openhft.lang.io.Bytes;
|
||||
import net.openhft.lang.model.Byteable;
|
||||
import net.openhft.lang.model.constraints.NotNull;
|
||||
|
||||
/**
|
||||
* uses a {@code SerializationBuilder} to read and write object(s) as bytes
|
||||
*
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
class Serializer<O> {
|
||||
|
||||
final SizeMarshaller sizeMarshaller;
|
||||
final BytesReader<O> originalReader;
|
||||
transient Provider<BytesReader<O>> readerProvider;
|
||||
final Object originalWriter;
|
||||
transient Provider writerProvider;
|
||||
final MetaBytesInterop originalMetaValueWriter;
|
||||
final MetaProvider metaValueWriterProvider;
|
||||
|
||||
Serializer(final SerializationBuilder serializationBuilder) {
|
||||
|
||||
sizeMarshaller = serializationBuilder.sizeMarshaller();
|
||||
originalMetaValueWriter = serializationBuilder.metaInterop();
|
||||
metaValueWriterProvider = serializationBuilder.metaInteropProvider();
|
||||
|
||||
originalReader = serializationBuilder.reader();
|
||||
originalWriter = serializationBuilder.interop();
|
||||
|
||||
readerProvider = Provider.of((Class) originalReader.getClass());
|
||||
writerProvider = Provider.of((Class) originalWriter.getClass());
|
||||
}
|
||||
|
||||
public O readMarshallable(@NotNull Bytes in) throws IllegalStateException {
|
||||
|
||||
final ThreadLocalCopies copies = readerProvider.getCopies(null);
|
||||
final BytesReader<O> valueReader = readerProvider.get(copies, originalReader);
|
||||
|
||||
try {
|
||||
long valueSize = sizeMarshaller.readSize(in);
|
||||
return valueReader.read(in, valueSize, null);
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
public void writeMarshallable(O value, @NotNull Bytes out) {
|
||||
|
||||
ThreadLocalCopies copies = readerProvider.getCopies(null);
|
||||
Object valueWriter = writerProvider.get(copies, originalWriter);
|
||||
copies = writerProvider.getCopies(copies);
|
||||
|
||||
final long valueSize;
|
||||
|
||||
MetaBytesWriter metaValueWriter = null;
|
||||
|
||||
if ((value instanceof Byteable)) {
|
||||
valueSize = ((Byteable) value).maxSize();
|
||||
} else {
|
||||
copies = writerProvider.getCopies(copies);
|
||||
valueWriter = writerProvider.get(copies, originalWriter);
|
||||
copies = metaValueWriterProvider.getCopies(copies);
|
||||
metaValueWriter = metaValueWriterProvider.get(
|
||||
copies, originalMetaValueWriter, valueWriter, value);
|
||||
valueSize = metaValueWriter.size(valueWriter, value);
|
||||
}
|
||||
|
||||
sizeMarshaller.writeSize(out, valueSize);
|
||||
|
||||
if (metaValueWriter != null)
|
||||
metaValueWriter.write(valueWriter, out, value);
|
||||
else
|
||||
throw new UnsupportedOperationException("");
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,533 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.lang.io.ByteBufferBytes;
|
||||
import net.openhft.lang.io.Bytes;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.SocketChannel;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
import static net.openhft.chronicle.map.StatelessMapClient.EventId.*;
|
||||
|
||||
/**
|
||||
* **** THIS IS VERY MUCH IN DRAFT ***
|
||||
*
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
class StatelessMapClient<K, V> {
|
||||
|
||||
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});
|
||||
|
||||
public static enum EventId {
|
||||
HEARTBEAT,
|
||||
STATEFUL_UPDATE,
|
||||
LONG_SIZE, SIZE,
|
||||
IS_EMPTY,
|
||||
CONTAINS_KEY,
|
||||
CONTAINS_VALUE,
|
||||
GET, PUT,
|
||||
REMOVE,
|
||||
CLEAR,
|
||||
KEY_SET,
|
||||
VALUES,
|
||||
ENTRY_SET,
|
||||
REPLACE,
|
||||
REPLACE_WITH_OLD_AND_NEW_VALUE,
|
||||
PUT_IF_ABSENT,
|
||||
REMOVE_WITH_VALUE
|
||||
}
|
||||
|
||||
private final ThreadLocal<Buffer> sourceBuffer = new ThreadLocal<Buffer>();
|
||||
private final Map<Long, Object> transactionIDs = new HashMap<Long, Object>();
|
||||
|
||||
private final AtomicLong transactionID = new AtomicLong();
|
||||
|
||||
private long timeoutMs = TimeUnit.SECONDS.toMillis(20);
|
||||
|
||||
|
||||
public StatelessMapClient(final KeyValueSerializer<K,
|
||||
V> keyValueSerializer, final InetSocketAddress remote) throws IOException {
|
||||
|
||||
this.keyValueSerializer = keyValueSerializer;
|
||||
|
||||
clientChannel = SocketChannel.open();
|
||||
clientChannel.connect(remote);
|
||||
|
||||
doHandShaking();
|
||||
}
|
||||
|
||||
private void doHandShaking() throws IOException {
|
||||
|
||||
|
||||
while (STATELESS_CLIENT_IDENTIFER.remaining() > 0) {
|
||||
clientChannel.write(STATELESS_CLIENT_IDENTIFER);
|
||||
}
|
||||
|
||||
in.buffer().clear();
|
||||
|
||||
while (in.buffer().position() <= 0) {
|
||||
clientChannel.read(in.buffer());
|
||||
}
|
||||
|
||||
in.limit(in.buffer().position());
|
||||
in.position(0);
|
||||
byte remoteIdentifier = in.readByte();
|
||||
|
||||
LOG.info("Attached to a map with a remote identifier=" + remoteIdentifier);
|
||||
|
||||
}
|
||||
|
||||
|
||||
public File file() {
|
||||
return null;
|
||||
}
|
||||
|
||||
public void close() throws IOException {
|
||||
throw new UnsupportedOperationException("This is not supported in the " + this.getClass()
|
||||
.getSimpleName());
|
||||
}
|
||||
|
||||
long nextUniqueTransaction() {
|
||||
long time = System.currentTimeMillis();
|
||||
|
||||
long l = transactionID.get();
|
||||
if (time > l) {
|
||||
boolean b = transactionID.compareAndSet(l, time);
|
||||
if (b) return time;
|
||||
}
|
||||
|
||||
return transactionID.incrementAndGet();
|
||||
}
|
||||
|
||||
|
||||
private final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocate(1024));
|
||||
private final ByteBufferBytes in = new ByteBufferBytes(ByteBuffer.allocate(1024));
|
||||
private final KeyValueSerializer<K, V> keyValueSerializer;
|
||||
|
||||
|
||||
public synchronized V putIfAbsent(K key, V value) throws IOException {
|
||||
|
||||
long sizeLocation = writeEvent(PUT_IF_ABSENT);
|
||||
writeKey(key);
|
||||
writeValue(value);
|
||||
return readKey(sizeLocation);
|
||||
}
|
||||
|
||||
|
||||
public synchronized boolean remove(Object key, Object value) throws IOException {
|
||||
long sizeLocation = writeEvent(REMOVE_WITH_VALUE);
|
||||
writeKey((K) key);
|
||||
writeValue((V) value);
|
||||
|
||||
// get the data back from the server
|
||||
return blockingFetch(sizeLocation).readBoolean();
|
||||
|
||||
}
|
||||
|
||||
|
||||
public synchronized boolean replace(K key, V oldValue, V newValue) throws IOException {
|
||||
long sizeLocation = writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE);
|
||||
writeKey(key);
|
||||
writeValue(oldValue);
|
||||
writeValue(newValue);
|
||||
|
||||
// get the data back from the server
|
||||
return blockingFetch(sizeLocation).readBoolean();
|
||||
}
|
||||
|
||||
|
||||
public synchronized V replace(K key, V value) throws IOException {
|
||||
long sizeLocation = writeEvent(REPLACE);
|
||||
writeKey(key);
|
||||
writeValue(value);
|
||||
|
||||
// get the data back from the server
|
||||
return readKey(sizeLocation);
|
||||
}
|
||||
|
||||
|
||||
public synchronized int size() throws IOException {
|
||||
|
||||
long sizeLocation = writeEvent(SIZE);
|
||||
|
||||
// get the data back from the server
|
||||
return blockingFetch(sizeLocation).readInt();
|
||||
}
|
||||
|
||||
|
||||
public synchronized boolean isEmpty() throws IOException {
|
||||
long sizeLocation = writeEvent(IS_EMPTY);
|
||||
|
||||
// get the data back from the server
|
||||
return blockingFetch(sizeLocation).readBoolean();
|
||||
}
|
||||
|
||||
|
||||
public synchronized boolean containsKey(Object key) throws IOException {
|
||||
long sizeLocation = writeEvent(CONTAINS_KEY);
|
||||
writeKey((K) key);
|
||||
|
||||
// get the data back from the server
|
||||
return blockingFetch(sizeLocation).readBoolean();
|
||||
|
||||
}
|
||||
|
||||
|
||||
public synchronized boolean containsValue(Object value) throws IOException {
|
||||
long sizeLocation = writeEvent(CONTAINS_VALUE);
|
||||
writeValue((V) value);
|
||||
|
||||
// get the data back from the server
|
||||
return blockingFetch(sizeLocation).readBoolean();
|
||||
}
|
||||
|
||||
|
||||
public synchronized long longSize() throws IOException {
|
||||
long sizeLocation = writeEvent(LONG_SIZE);
|
||||
// get the data back from the server
|
||||
return blockingFetch(sizeLocation).readLong();
|
||||
}
|
||||
|
||||
|
||||
public synchronized V get(Object key) throws IOException {
|
||||
long sizeLocation = writeEvent(GET);
|
||||
writeKey((K) key);
|
||||
|
||||
// get the data back from the server
|
||||
return readKey(sizeLocation);
|
||||
|
||||
}
|
||||
|
||||
|
||||
public synchronized V getUsing(K key, V value) {
|
||||
throw new UnsupportedOperationException("getUsing is not supported for stateless clients");
|
||||
}
|
||||
|
||||
|
||||
public synchronized V acquireUsing(K key, V value) {
|
||||
throw new UnsupportedOperationException("getUsing is not supported for stateless clients");
|
||||
}
|
||||
|
||||
|
||||
public synchronized V put(K key, V value) throws IOException {
|
||||
|
||||
long sizeLocation = writeEvent(PUT);
|
||||
writeKey(key);
|
||||
writeValue(value);
|
||||
|
||||
// get the data back from the server
|
||||
return readKey(sizeLocation);
|
||||
|
||||
}
|
||||
|
||||
|
||||
public synchronized V remove(Object key) throws IOException {
|
||||
long sizeLocation = writeEvent(REMOVE);
|
||||
writeKey((K) key);
|
||||
|
||||
// get the data back from the server
|
||||
return readKey(sizeLocation);
|
||||
|
||||
}
|
||||
|
||||
|
||||
public synchronized void putAll(Map<? extends K, ? extends V> map) throws IOException {
|
||||
|
||||
for (Map.Entry<? extends K, ? extends V> e : map.entrySet()) {
|
||||
put(e.getKey(), e.getValue());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private void writeEntries(Map<? extends K, ? extends V> map) {
|
||||
|
||||
final HashMap<K, V> safeCopy = new HashMap<K, V>(map);
|
||||
|
||||
out.writeStopBit(safeCopy.size());
|
||||
|
||||
final Set<Entry> entries = (Set) safeCopy.entrySet();
|
||||
|
||||
for (Entry e : entries) {
|
||||
writeKey(e.getKey());
|
||||
writeValue(e.getValue());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public synchronized void clear() throws IOException {
|
||||
long sizeLocation = writeEvent(CLEAR);
|
||||
|
||||
// get the data back from the server
|
||||
blockingFetch(sizeLocation);
|
||||
|
||||
}
|
||||
|
||||
@NotNull
|
||||
|
||||
public synchronized Set<K> keySet() throws IOException {
|
||||
long sizeLocation = writeEvent(KEY_SET);
|
||||
|
||||
// get the data back from the server
|
||||
return readKeySet(blockingFetch(sizeLocation));
|
||||
}
|
||||
|
||||
|
||||
@NotNull
|
||||
|
||||
public synchronized Collection<V> values() throws IOException {
|
||||
long sizeLocation = writeEvent(VALUES);
|
||||
|
||||
// get the data back from the server
|
||||
final Bytes in = blockingFetch(sizeLocation);
|
||||
|
||||
final long size = in.readStopBit();
|
||||
|
||||
if (size > Integer.MAX_VALUE)
|
||||
throw new IllegalStateException("size=" + size + " is too large.");
|
||||
|
||||
final ArrayList<V> result = new ArrayList<V>((int) size);
|
||||
|
||||
for (int i = 0; i < size; i++) {
|
||||
result.add(keyValueSerializer.readValue(in));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
@NotNull
|
||||
|
||||
public synchronized Set<Map.Entry<K, V>> entrySet() throws IOException {
|
||||
long sizeLocation = writeEvent(ENTRY_SET);
|
||||
|
||||
// get the data back from the server
|
||||
Bytes in = blockingFetch(sizeLocation);
|
||||
|
||||
long size = in.readStopBit();
|
||||
|
||||
Set<Map.Entry<K, V>> result = new HashSet<Map.Entry<K, V>>();
|
||||
|
||||
for (int i = 0; i < size; i++) {
|
||||
K k = keyValueSerializer.readKey(in);
|
||||
V v = keyValueSerializer.readValue(in);
|
||||
result.add(new Entry(k, v));
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
private long writeEvent(EventId event) {
|
||||
out.clear();
|
||||
out.buffer().clear();
|
||||
|
||||
out.write((byte) event.ordinal());
|
||||
long sizeLocation = markSizeLocation();
|
||||
return sizeLocation;
|
||||
}
|
||||
|
||||
public Buffer buffer() {
|
||||
Buffer result = sourceBuffer.get();
|
||||
|
||||
|
||||
if (result != null) {
|
||||
return result;
|
||||
}
|
||||
|
||||
result = new Buffer() {
|
||||
|
||||
volatile ByteBufferBytes buffer = null;
|
||||
|
||||
|
||||
public void set(ByteBufferBytes source) {
|
||||
buffer = source;
|
||||
}
|
||||
|
||||
|
||||
public ByteBufferBytes get() {
|
||||
return buffer;
|
||||
}
|
||||
};
|
||||
|
||||
sourceBuffer.set(result);
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* blocks until a message is received from the server or a timeout is reached
|
||||
*
|
||||
* @param transactionId
|
||||
* @return
|
||||
*/
|
||||
private Bytes blockingFetchWithTransaction(long transactionId) {
|
||||
final Buffer buffer = buffer();
|
||||
buffer.set(null);
|
||||
|
||||
synchronized (this) {
|
||||
transactionIDs.put(transactionId, buffer);
|
||||
try {
|
||||
buffer.wait(timeoutMs);
|
||||
} catch (InterruptedException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
final ByteBufferBytes bufferBytes = buffer.get();
|
||||
|
||||
if (bufferBytes == null)
|
||||
throw new RuntimeException("Timed-out", new TimeoutException());
|
||||
|
||||
return bufferBytes;
|
||||
}
|
||||
|
||||
|
||||
class Entry implements Map.Entry<K, V> {
|
||||
|
||||
final K key;
|
||||
final V value;
|
||||
|
||||
/**
|
||||
* Creates new entry.
|
||||
*/
|
||||
Entry(K k1, V v) {
|
||||
value = v;
|
||||
key = k1;
|
||||
}
|
||||
|
||||
public final K getKey() {
|
||||
return key;
|
||||
}
|
||||
|
||||
public final V getValue() {
|
||||
return value;
|
||||
}
|
||||
|
||||
public final V setValue(V newValue) {
|
||||
V oldValue = value;
|
||||
try {
|
||||
StatelessMapClient.this.put((K) getKey(), (V) newValue);
|
||||
} catch (IOException e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
return oldValue;
|
||||
}
|
||||
|
||||
public final boolean equals(Object o) {
|
||||
if (!(o instanceof Map.Entry))
|
||||
return false;
|
||||
Map.Entry e = (Map.Entry) o;
|
||||
Object k1 = getKey();
|
||||
Object k2 = e.getKey();
|
||||
if (k1 == k2 || (k1 != null && k1.equals(k2))) {
|
||||
Object v1 = getValue();
|
||||
Object v2 = e.getValue();
|
||||
if (v1 == v2 || (v1 != null && v1.equals(v2)))
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public final int hashCode() {
|
||||
return (key == null ? 0 : key.hashCode()) ^
|
||||
(value == null ? 0 : value.hashCode());
|
||||
}
|
||||
|
||||
public final String toString() {
|
||||
return getKey() + "=" + getValue();
|
||||
}
|
||||
}
|
||||
|
||||
private Bytes blockingFetch(long sizeLocation) throws IOException {
|
||||
long transactionId = nextUniqueTransaction();
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
in.clear();
|
||||
in.buffer().clear();
|
||||
|
||||
// read size
|
||||
while (in.buffer().position() < 2) {
|
||||
clientChannel.read(in.buffer());
|
||||
}
|
||||
|
||||
int size = in.readUnsignedShort();
|
||||
|
||||
while (in.buffer().position() < size + 2) {
|
||||
clientChannel.read(in.buffer());
|
||||
}
|
||||
|
||||
long inTransactionId = in.readLong();
|
||||
|
||||
if (inTransactionId != transactionId) {
|
||||
throw new IllegalStateException("the received transaction-id=" + inTransactionId +
|
||||
", does not match the expected transaction-id=" + transactionId);
|
||||
}
|
||||
|
||||
return in;
|
||||
}
|
||||
|
||||
private void writeSizeAt(long locationOfSize) {
|
||||
long size = out.position() - locationOfSize;
|
||||
out.writeUnsignedShort(locationOfSize, (int) size - 2);
|
||||
}
|
||||
|
||||
private long markSizeLocation() {
|
||||
long position = out.position();
|
||||
out.skip(2);
|
||||
return position;
|
||||
}
|
||||
|
||||
private void writeKey(K key) {
|
||||
keyValueSerializer.writeKey(key, out);
|
||||
}
|
||||
|
||||
private V readKey(final long sizeLocation) throws IOException {
|
||||
return keyValueSerializer.readValue(blockingFetch(sizeLocation));
|
||||
}
|
||||
|
||||
private void writeValue(V value) {
|
||||
keyValueSerializer.writeValue(value, out);
|
||||
}
|
||||
|
||||
|
||||
private Set<K> readKeySet(Bytes in) {
|
||||
long size = in.readStopBit();
|
||||
final HashSet<K> result = new HashSet<>();
|
||||
|
||||
for (long i = 0; i < size; i++) {
|
||||
result.add(keyValueSerializer.readKey(in));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
interface Buffer {
|
||||
|
||||
void set(ByteBufferBytes source);
|
||||
|
||||
ByteBufferBytes get();
|
||||
}
|
||||
@@ -0,0 +1,317 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.lang.io.Bytes;
|
||||
import net.openhft.lang.model.constraints.NotNull;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
class StatelessServerConnector<K, V> {
|
||||
|
||||
|
||||
private final KeyValueSerializer<K, V> keyValueSerializer;
|
||||
private final ChronicleMap<K, V> map;
|
||||
|
||||
|
||||
public StatelessServerConnector(KeyValueSerializer<K, V> keyValueSerializer, ChronicleMap<K, V> map) {
|
||||
this.keyValueSerializer = keyValueSerializer;
|
||||
this.map = map;
|
||||
}
|
||||
|
||||
|
||||
public void processStatelessEvent(byte eventId, @NotNull Bytes in, @NotNull Bytes out) {
|
||||
|
||||
|
||||
StatelessMapClient.EventId event = StatelessMapClient.EventId.values()[eventId];
|
||||
switch (event) {
|
||||
|
||||
case LONG_SIZE:
|
||||
longSize(in, out);
|
||||
break;
|
||||
|
||||
case IS_EMPTY:
|
||||
isEmpty(in, out);
|
||||
break;
|
||||
|
||||
case CONTAINS_KEY:
|
||||
containsKey(in, out);
|
||||
break;
|
||||
case CONTAINS_VALUE:
|
||||
containsValue(in, out);
|
||||
break;
|
||||
case GET:
|
||||
get(in, out);
|
||||
break;
|
||||
case PUT:
|
||||
put(in, out);
|
||||
break;
|
||||
case REMOVE:
|
||||
remove(in, out);
|
||||
break;
|
||||
case CLEAR:
|
||||
clear(in, out);
|
||||
break;
|
||||
case KEY_SET:
|
||||
keySet(in, out);
|
||||
break;
|
||||
case VALUES:
|
||||
values(in, out);
|
||||
break;
|
||||
case ENTRY_SET:
|
||||
entrySet(in, out);
|
||||
break;
|
||||
case REPLACE:
|
||||
replace(in, out);
|
||||
break;
|
||||
case REPLACE_WITH_OLD_AND_NEW_VALUE:
|
||||
replaceWithOldAndNew(in, out);
|
||||
break;
|
||||
case PUT_IF_ABSENT:
|
||||
putIfAbsent(in, out);
|
||||
break;
|
||||
case REMOVE_WITH_VALUE:
|
||||
removeWithValue(in, out);
|
||||
break;
|
||||
|
||||
case SIZE:
|
||||
size(in, out);
|
||||
break;
|
||||
|
||||
default:
|
||||
throw new IllegalStateException("unsupported event=" + event);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private void removeWithValue(Bytes in, Bytes out) {
|
||||
boolean result = map.remove(readKey(in), readValue(in));
|
||||
out.writeBoolean(result);
|
||||
}
|
||||
|
||||
private void replaceWithOldAndNew(Bytes in, Bytes out) {
|
||||
|
||||
final K key = readKey(in);
|
||||
V oldValue = readValue(in);
|
||||
V newValue = readValue(in);
|
||||
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
map.replace(key, oldValue, newValue);
|
||||
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
|
||||
public void longSize(Bytes in, Bytes out) {
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
out.writeLong(map.longSize());
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void size(Bytes in, Bytes out) {
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
out.writeInt(map.size());
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void isEmpty(Bytes in, Bytes out) {
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
out.writeBoolean(map.isEmpty());
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void containsKey(Bytes in, Bytes out) {
|
||||
K k = readKey(in);
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
out.writeBoolean(map.containsKey(k));
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void containsValue(Bytes in, Bytes out) {
|
||||
V v = readValue(in);
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
out.writeBoolean(map.containsValue(v));
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void get(Bytes in, Bytes out) {
|
||||
K k = readKey(in);
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
writeValue(map.get(k), out);
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void put(Bytes in, Bytes out) {
|
||||
K k = readKey(in);
|
||||
V v = readValue(in);
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
writeValue(map.put(k, v), out);
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void remove(Bytes in, Bytes out) {
|
||||
final V value = map.remove(readKey(in));
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
writeValue(value, out);
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void putAll(Bytes in, Bytes out) {
|
||||
map.putAll(readEntries(in));
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
private Map<K, V> readEntries(Bytes in) {
|
||||
|
||||
long size = in.readStopBit();
|
||||
final HashMap<K, V> result = new HashMap<K, V>();
|
||||
|
||||
for (long i = 0; i < size; i++) {
|
||||
result.put(readKey(in), readValue(in));
|
||||
}
|
||||
return result;
|
||||
|
||||
}
|
||||
|
||||
public void clear(Bytes in, Bytes out) {
|
||||
map.clear();
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
|
||||
public void keySet(Bytes in, Bytes out) {
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
|
||||
final Set<K> ks = map.keySet();
|
||||
out.writeStopBit(ks.size());
|
||||
for (K key : ks) {
|
||||
writeKey(key, out);
|
||||
}
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void values(Bytes in, Bytes out) {
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
final Collection<V> values = map.values();
|
||||
out.writeStopBit(values.size());
|
||||
for (final V value : values) {
|
||||
writeValue(value, out);
|
||||
}
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
public void entrySet(Bytes in, Bytes out) {
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
|
||||
final Set<Map.Entry<K, V>> entries = map.entrySet();
|
||||
out.writeStopBit(entries.size());
|
||||
for (Map.Entry<K, V> e : entries) {
|
||||
writeKey(e.getKey(), out);
|
||||
writeValue(e.getValue(), out);
|
||||
}
|
||||
}
|
||||
|
||||
public void putIfAbsent(Bytes in, Bytes out) {
|
||||
K key = readKey(in);
|
||||
V v = readValue(in);
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
writeValue(map.putIfAbsent(key, v), out);
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
|
||||
public void replace(Bytes in, Bytes out) {
|
||||
K k = readKey(in);
|
||||
V v = readValue(in);
|
||||
|
||||
long sizeLocation = reflectTransactionId(in, out);
|
||||
map.replace(k, v);
|
||||
writeSizeAt(sizeLocation, out);
|
||||
}
|
||||
|
||||
/**
|
||||
* write the keysize and the key to the the {@code target} buffer
|
||||
*
|
||||
* @param key the key of the map
|
||||
*/
|
||||
private void writeKey(K key, Bytes out) {
|
||||
keyValueSerializer.writeKey(key, out);
|
||||
}
|
||||
|
||||
private long reflectTransactionId(Bytes in, Bytes out) {
|
||||
long sizeLocation = out.position();
|
||||
out.skip(2);
|
||||
long transactionId = in.readLong();
|
||||
out.writeLong(transactionId);
|
||||
return sizeLocation;
|
||||
}
|
||||
|
||||
private void writeValue(V value, final Bytes out) {
|
||||
keyValueSerializer.writeValue(value, out);
|
||||
}
|
||||
|
||||
private K readKey(Bytes in) {
|
||||
return keyValueSerializer.readKey(in);
|
||||
}
|
||||
|
||||
private V readValue(Bytes in) {
|
||||
return keyValueSerializer.readValue(in);
|
||||
}
|
||||
|
||||
private void writeSizeAt(long locationOfSize, Bytes out) {
|
||||
long size = out.position() - locationOfSize;
|
||||
out.writeUnsignedShort((int) size);
|
||||
}
|
||||
}
|
||||
|
||||
class KeyValueSerializer<K, V> {
|
||||
|
||||
private final Serializer<V> valueSerializer;
|
||||
private final Serializer<K> keySerializer;
|
||||
|
||||
KeyValueSerializer(SerializationBuilder<K> keySerializer,
|
||||
SerializationBuilder<V> valueSerializer) {
|
||||
this.keySerializer = new Serializer<K>(keySerializer);
|
||||
this.valueSerializer = new Serializer<V>(valueSerializer);
|
||||
}
|
||||
|
||||
V readValue(Bytes in) {
|
||||
if (in.readBoolean())
|
||||
return null;
|
||||
return valueSerializer.readMarshallable(in);
|
||||
}
|
||||
|
||||
K readKey(Bytes in) {
|
||||
if (in.readBoolean())
|
||||
return null;
|
||||
|
||||
return keySerializer.readMarshallable(in);
|
||||
}
|
||||
|
||||
/**
|
||||
* write the keysize and the key to the the {@code target} buffer
|
||||
*
|
||||
* @param key the key of the map
|
||||
*/
|
||||
void writeKey(K key, Bytes out) {
|
||||
|
||||
out.writeBoolean(key == null);
|
||||
if (key != null)
|
||||
keySerializer.writeMarshallable(key, out);
|
||||
}
|
||||
|
||||
void writeValue(V value, Bytes out) {
|
||||
out.writeBoolean(value == null);
|
||||
if (value != null)
|
||||
valueSerializer.writeMarshallable(value, out);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -29,7 +29,7 @@ import static java.util.Collections.unmodifiableSet;
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
|
||||
|
||||
public class TcpReplicationConfig {
|
||||
public class TcpReplicationConfig implements ReplicationConfig {
|
||||
|
||||
private static final int DEFAULT_PACKET_SIZE = 1024 * 8;
|
||||
private static final long DEFAULT_HEART_BEAT_INTERVAL = 20;
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.lang.io.ByteBufferBytes;
|
||||
import net.openhft.lang.io.Bytes;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
@@ -32,10 +33,11 @@ import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import static java.nio.channels.SelectionKey.*;
|
||||
import static net.openhft.chronicle.map.StatelessMapClient.EventId.HEARTBEAT;
|
||||
|
||||
/**
|
||||
* Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a socket
|
||||
* connection <p/> {@see net.openhft.map.OutSocketReplicator}
|
||||
* Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a
|
||||
* socket connection <p/> {@see net.openhft.map.OutSocketReplicator}
|
||||
*
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
@@ -43,6 +45,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName());
|
||||
private static final int BUFFER_SIZE = 0x100000; // 1MB
|
||||
public static final int STATELESS_CLIENT = -127;
|
||||
public static final byte NOT_SET = (byte) HEARTBEAT.ordinal();
|
||||
|
||||
private final SelectionKey[] selectionKeysStore = new SelectionKey[Byte.MAX_VALUE + 1];
|
||||
// used to instruct the selector thread to set OP_WRITE on a key correlated by the bit index in the
|
||||
@@ -57,22 +61,28 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
private final Replica.EntryExternalizable externalizable;
|
||||
private final TcpReplicationConfig replicationConfig;
|
||||
|
||||
private final StatelessServerConnector statelessServerConnector;
|
||||
|
||||
private long selectorTimeout;
|
||||
|
||||
/**
|
||||
* @param maxEntrySizeBytes used to check that the last entry will fit into the buffer, it can not be
|
||||
* smaller than the size of and entry, if it is set smaller the buffer will over
|
||||
* flow, it can be larger then the entry, but setting it too large reduces the
|
||||
* workable space in the buffer.
|
||||
* @param maxEntrySizeBytes used to check that the last entry will fit into the buffer,
|
||||
* it can not be smaller than the size of and entry, if it is
|
||||
* set smaller the buffer will over flow, it can be larger then
|
||||
* the entry, but setting it too large reduces the workable
|
||||
* space in the buffer.
|
||||
* @param statelessServerConnector
|
||||
* @throws IOException
|
||||
*/
|
||||
TcpReplicator(@NotNull final Replica replica,
|
||||
@NotNull final Replica.EntryExternalizable externalizable,
|
||||
@NotNull final TcpReplicationConfig replicationConfig,
|
||||
final int maxEntrySizeBytes) throws IOException {
|
||||
final int maxEntrySizeBytes,
|
||||
StatelessServerConnector statelessServerConnector) throws IOException {
|
||||
|
||||
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(),
|
||||
maxEntrySizeBytes);
|
||||
this.statelessServerConnector = statelessServerConnector;
|
||||
|
||||
final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig();
|
||||
long throttleBucketInterval = TimeUnit.MILLISECONDS.convert(throttlingConfig
|
||||
@@ -251,7 +261,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
}
|
||||
|
||||
/**
|
||||
* check to see if we have lost connection with the remote node and if we have attempts a reconnect.
|
||||
* check to see if we have lost connection with the remote node and if we have attempts a
|
||||
* reconnect.
|
||||
*
|
||||
* @param key the key relating to the heartbeat that we are checking
|
||||
* @param approxTimeOutTime the approximate time in milliseconds
|
||||
@@ -374,8 +385,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
}
|
||||
|
||||
/**
|
||||
* this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear all the
|
||||
* old bootstrap information
|
||||
* this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear
|
||||
* all the old bootstrap information
|
||||
*
|
||||
* @param key the nio SelectionKey
|
||||
*/
|
||||
@@ -387,7 +398,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
}
|
||||
|
||||
/**
|
||||
* used to exchange identifiers and timestamps and heartbeat intervals between the server and client
|
||||
* used to exchange identifiers and timestamps and heartbeat intervals between the server and
|
||||
* client
|
||||
*
|
||||
* @param key the SelectionKey relating to the this cha
|
||||
* @param socketChannel
|
||||
@@ -404,6 +416,11 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
|
||||
final byte remoteIdentifier = reader.identifierFromBuffer();
|
||||
|
||||
if (remoteIdentifier == STATELESS_CLIENT) {
|
||||
attached.handShakingComplete = true;
|
||||
return;
|
||||
}
|
||||
|
||||
if (remoteIdentifier == Byte.MIN_VALUE)
|
||||
return;
|
||||
|
||||
@@ -425,7 +442,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
final SocketAddress remoteAddress = socketChannel.getRemoteAddress();
|
||||
|
||||
|
||||
if ( (identifierListener != null && !identifierListener.isIdentifierUnique(remoteIdentifier,
|
||||
if ((identifierListener != null && !identifierListener.isIdentifierUnique(remoteIdentifier,
|
||||
remoteAddress)) || remoteIdentifier == localIdentifier)
|
||||
|
||||
// throwing this exception will cause us to disconnect, both the client and server
|
||||
@@ -477,7 +494,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
// now we're finished we can get on with reading the entries
|
||||
attached.handShakingComplete = true;
|
||||
attached.remoteModificationIterator.dirtyEntries(attached.remoteBootstrapTimestamp);
|
||||
reader.entriesFromBuffer();
|
||||
reader.entriesFromBuffer(attached);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -531,17 +548,17 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
attached.entryReader.lastHeartBeatReceived = approxTime;
|
||||
|
||||
if (attached.isHandShakingComplete()) {
|
||||
attached.entryReader.entriesFromBuffer();
|
||||
attached.entryReader.entriesFromBuffer(attached);
|
||||
} else {
|
||||
doHandShaking(key, socketChannel);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* sets interestOps to "selector keys",The change to interestOps much be on the same thread as the
|
||||
* selector. This class, allows via {@link AbstractChannelReplicator .KeyInterestUpdater#set(int)} to
|
||||
* holds a pending change in interestOps ( via a bitset ), this change is processed later on the same
|
||||
* thread as the selector
|
||||
* sets interestOps to "selector keys",The change to interestOps much be on the same thread as
|
||||
* the selector. This class, allows via {@link AbstractChannelReplicator
|
||||
* .KeyInterestUpdater#set(int)} to holds a pending change in interestOps ( via a bitset ),
|
||||
* this change is processed later on the same thread as the selector
|
||||
*/
|
||||
private static class KeyInterestUpdater {
|
||||
|
||||
@@ -572,8 +589,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
}
|
||||
|
||||
/**
|
||||
* @param keyIndex the index of the key that has changed, the list of keys is provided by the
|
||||
* constructor {@link KeyInterestUpdater(int, SelectionKey[])}
|
||||
* @param keyIndex the index of the key that has changed, the list of keys is provided by
|
||||
* the constructor {@link KeyInterestUpdater(int, SelectionKey[])}
|
||||
*/
|
||||
public void set(int keyIndex) {
|
||||
changeOfOpWriteRequired.set(keyIndex);
|
||||
@@ -773,12 +790,19 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @return the buffer messages can be written to
|
||||
*/
|
||||
Bytes buffer() {
|
||||
return in;
|
||||
}
|
||||
|
||||
/**
|
||||
* writes the timestamp into the buffer
|
||||
*
|
||||
* @param localIdentifier the current nodes identifier
|
||||
*/
|
||||
void identifierToBuffer(final int localIdentifier) {
|
||||
void identifierToBuffer(final byte localIdentifier) {
|
||||
in.writeByte(localIdentifier);
|
||||
}
|
||||
|
||||
@@ -792,7 +816,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
}
|
||||
|
||||
/**
|
||||
* writes all the entries that have changed, to the buffer which will later be written to TCP/IP
|
||||
* writes all the entries that have changed, to the buffer which will later be written to
|
||||
* TCP/IP
|
||||
*
|
||||
* @param modificationIterator a record of which entries have modification
|
||||
* @param selectionKey
|
||||
@@ -804,7 +829,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
final SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
|
||||
final Attached attached = (Attached) selectionKey.attachment();
|
||||
|
||||
// this can occur when new CHM's are added to a cluster
|
||||
// this can occur when new map are added to a channel
|
||||
final boolean handShakingComplete = attached.isHandShakingComplete();
|
||||
|
||||
for (; ; ) {
|
||||
@@ -867,9 +892,15 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
}
|
||||
|
||||
/**
|
||||
* used to send an single zero byte if we have not send any data for up to the localHeartbeatInterval
|
||||
* used to send an single zero byte if we have not send any data for up to the
|
||||
* localHeartbeatInterval
|
||||
*/
|
||||
private void writeHeartbeatToBuffer() {
|
||||
|
||||
// denotes the state - 0 for a heartbeat
|
||||
in.writeByte(HEARTBEAT.ordinal());
|
||||
|
||||
// denotes the size in bytes
|
||||
in.writeUnsignedShort(0);
|
||||
}
|
||||
|
||||
@@ -879,8 +910,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
|
||||
|
||||
/**
|
||||
* removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE will get
|
||||
* added back in as soon as we have data to write
|
||||
* removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE
|
||||
* will get added back in as soon as we have data to write
|
||||
*
|
||||
* @param socketChannel the socketChannel we wish to stop writing to
|
||||
* @param attached data associated with the socketChannels key
|
||||
@@ -914,8 +945,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
private final ByteBuffer in;
|
||||
private final ByteBufferBytes out;
|
||||
public long lastHeartBeatReceived = System.currentTimeMillis();
|
||||
// we use Integer.MIN_VALUE as N/A
|
||||
private int sizeOfNextEntry = Integer.MIN_VALUE;
|
||||
|
||||
private int sizeInBytes;
|
||||
private byte state;
|
||||
|
||||
private TcpSocketChannelEntryReader() {
|
||||
in = ByteBuffer.allocateDirect(replicationConfig.packetSize() + maxEntrySizeBytes);
|
||||
@@ -943,45 +975,79 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
/**
|
||||
* reads entries from the buffer till empty
|
||||
*
|
||||
* @param attached
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
private void entriesFromBuffer() throws InterruptedException, IOException {
|
||||
|
||||
private void entriesFromBuffer(Attached attached) throws InterruptedException, IOException {
|
||||
for (; ; ) {
|
||||
|
||||
out.limit(in.position());
|
||||
|
||||
// its set to MIN_VALUE when it should be read again
|
||||
if (sizeOfNextEntry == Integer.MIN_VALUE) {
|
||||
if (out.remaining() < SIZE_OF_SHORT) {
|
||||
if (state == NOT_SET) {
|
||||
if (out.remaining() < SIZE_OF_SHORT + 1 ) {
|
||||
return;
|
||||
}
|
||||
|
||||
int value = out.readUnsignedShort();
|
||||
// state is used for both heartbeat and stateless
|
||||
state = out.readByte();
|
||||
|
||||
// this is the heartbeat
|
||||
if (value == 0)
|
||||
|
||||
|
||||
sizeInBytes = out.readUnsignedShort();
|
||||
|
||||
if (state == 0 && sizeInBytes != 0) {
|
||||
int i = 1;
|
||||
}
|
||||
|
||||
// this is the :
|
||||
// -- heartbeat if its 0
|
||||
// -- stateful update if its 1
|
||||
// -- the id of the stateful event
|
||||
if (state == NOT_SET)
|
||||
continue;
|
||||
|
||||
sizeOfNextEntry = value;
|
||||
}
|
||||
|
||||
|
||||
if (out.remaining() < sizeOfNextEntry) {
|
||||
|
||||
if (out.remaining() < sizeInBytes) {
|
||||
return;
|
||||
}
|
||||
|
||||
final long nextEntryPos = out.position() + sizeOfNextEntry;
|
||||
if (sizeInBytes == 0) {
|
||||
int j = 1;
|
||||
}
|
||||
|
||||
|
||||
final long nextEntryPos = out.position() + sizeInBytes;
|
||||
final long limit = out.limit();
|
||||
out.limit(nextEntryPos);
|
||||
externalizable.readExternalEntry(out);
|
||||
|
||||
boolean isStateless = (state != 1);
|
||||
|
||||
if (isStateless) {
|
||||
if (statelessServerConnector == null) {
|
||||
|
||||
LOG.error("", new IllegalArgumentException("received an event " +
|
||||
"from a stateless map, stateless maps are not " +
|
||||
"currently supported when using Chronicle Channels"));
|
||||
} else
|
||||
|
||||
statelessServerConnector.processStatelessEvent(state,
|
||||
out, attached.entryWriter.buffer());
|
||||
// to allow the sizeInBytes to be read the next time around
|
||||
|
||||
|
||||
} else
|
||||
externalizable.readExternalEntry(out);
|
||||
|
||||
out.limit(limit);
|
||||
// skip onto the next entry
|
||||
out.position(nextEntryPos);
|
||||
|
||||
// to allow the sizeOfNextEntry to be read the next time around
|
||||
sizeOfNextEntry = Integer.MIN_VALUE;
|
||||
|
||||
state = NOT_SET;
|
||||
sizeInBytes = 0;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1016,7 +1082,12 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
* @return the timestamp or -1 if unsuccessful
|
||||
*/
|
||||
long remoteBootstrapTimestamp() {
|
||||
return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE;
|
||||
if (out.remaining() >= 8)
|
||||
return out.readLong();
|
||||
else
|
||||
return Long.MIN_VALUE;
|
||||
|
||||
// return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE;
|
||||
}
|
||||
|
||||
public long remoteHeartbeatIntervalFromBuffer() {
|
||||
|
||||
@@ -23,7 +23,7 @@ import java.net.InetAddress;
|
||||
import java.net.NetworkInterface;
|
||||
|
||||
|
||||
public abstract class UdpReplicationConfig {
|
||||
public abstract class UdpReplicationConfig implements ReplicationConfig {
|
||||
|
||||
/**
|
||||
* Package-private constructor forbids subclassing from outside of the package
|
||||
@@ -37,7 +37,6 @@ public abstract class UdpReplicationConfig {
|
||||
throw new IllegalArgumentException();
|
||||
|
||||
|
||||
|
||||
return create(address, port, null, ThrottlingConfig.noThrottling());
|
||||
}
|
||||
|
||||
|
||||
@@ -133,7 +133,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
this.removeReturnsNull = builder.removeReturnsNull();
|
||||
|
||||
this.actualSegments = builder.actualSegments();
|
||||
this.entriesPerSegment = builder.actualEntriesPerSegment();
|
||||
// align by 8 because otherwise sizeOfBitSets() might address less blocks than specified
|
||||
this.entriesPerSegment = align8(builder.actualEntriesPerSegment());
|
||||
this.metaDataBytes = builder.metaDataBytes();
|
||||
this.eventListener = builder.eventListener();
|
||||
|
||||
@@ -158,6 +159,10 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
return (l + 63) & ~63;
|
||||
}
|
||||
|
||||
static int align8(int n) {
|
||||
return (n + 7) & ~7;
|
||||
}
|
||||
|
||||
void initTransients() {
|
||||
keyReaderProvider = Provider.of((Class) originalKeyReader.getClass());
|
||||
keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass());
|
||||
@@ -555,8 +560,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
long start = bytes.startAddr() + SEGMENT_HEADER;
|
||||
createHashLookups(start);
|
||||
start += align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment();
|
||||
final NativeBytes bsBytes = new NativeBytes(
|
||||
tmpBytes.objectSerializer(), start, start + sizeOfBitSets(), null);
|
||||
final NativeBytes bsBytes = new NativeBytes(tmpBytes.objectSerializer(),
|
||||
start, start + ((entriesPerSegment + 7) / 8), null);
|
||||
freeList = new SingleThreadedDirectBitSet(bsBytes);
|
||||
start += numberOfBitSets() * sizeOfBitSets();
|
||||
entriesOffset = start - bytes.startAddr();
|
||||
@@ -914,15 +919,23 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
);
|
||||
}
|
||||
}
|
||||
updateNextPosToSearchFrom(ret, blocks);
|
||||
} else {
|
||||
// if bit at nextPosToSearchFrom is clear, it was skipped because
|
||||
// more than 1 block was requested. Don't move nextPosToSearchFrom
|
||||
// in this case. blocks == 1 clause is just a fast path.
|
||||
if (blocks == 1 || freeList.isSet(nextPosToSearchFrom)) {
|
||||
updateNextPosToSearchFrom(ret, blocks);
|
||||
}
|
||||
}
|
||||
// if bit at nextPosToSearchFrom is clear, it was skipped because
|
||||
// more than 1 block was requested. Don't move nextPosToSearchFrom
|
||||
// in this case. blocks == 1 clause is just a fast path.
|
||||
if (blocks == 1 || freeList.isSet(nextPosToSearchFrom))
|
||||
nextPosToSearchFrom = ret + blocks;
|
||||
return ret;
|
||||
}
|
||||
|
||||
private void updateNextPosToSearchFrom(int allocated, int blocks) {
|
||||
if ((nextPosToSearchFrom = allocated + blocks) >= freeList.size())
|
||||
nextPosToSearchFrom = 0;
|
||||
}
|
||||
|
||||
private boolean realloc(int fromPos, int oldBlocks, int newBlocks) {
|
||||
if (freeList.allClear(fromPos + oldBlocks, fromPos + newBlocks)) {
|
||||
freeList.set(fromPos + oldBlocks, fromPos + newBlocks);
|
||||
@@ -1188,6 +1201,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
|
||||
try {
|
||||
hashLookup.clear();
|
||||
freeList.clear();
|
||||
nextPosToSearchFrom = 0;
|
||||
resetSize();
|
||||
} finally {
|
||||
unlock();
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
package net.openhft.chronicle.set;
|
||||
|
||||
import net.openhft.chronicle.ChronicleHash;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.util.Set;
|
||||
|
||||
public interface ChronicleSet<E> extends Set<E>, ChronicleHash {
|
||||
public long longSize();
|
||||
}
|
||||
@@ -23,26 +23,24 @@ import net.openhft.chronicle.map.*;
|
||||
import net.openhft.lang.io.serialization.BytesMarshaller;
|
||||
import net.openhft.lang.io.serialization.BytesMarshallerFactory;
|
||||
import net.openhft.lang.io.serialization.ObjectSerializer;
|
||||
import net.openhft.lang.io.serialization.impl.NullObjectFactory;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.io.Serializable;
|
||||
import java.util.AbstractSet;
|
||||
import java.util.Collection;
|
||||
import java.util.Iterator;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
|
||||
public class ChronicleSetBuilder<E>
|
||||
implements ChronicleHashBuilder<E, ChronicleSet<E>, ChronicleSetBuilder<E>> {
|
||||
|
||||
private ChronicleMapBuilder<E, Void> chronicleMapBuilder;
|
||||
private ChronicleMapBuilder<E, DummyValue> chronicleMapBuilder;
|
||||
|
||||
ChronicleSetBuilder(Class<E> keyClass) {
|
||||
chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, Void.class)
|
||||
.entryAndValueAlignment(Alignment.NO_ALIGNMENT);
|
||||
chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, DummyValue.class)
|
||||
.entryAndValueAlignment(Alignment.NO_ALIGNMENT)
|
||||
.valueMarshallerAndFactory(DummyValueMarshaller.INSTANCE,
|
||||
NullObjectFactory.<DummyValue>of());
|
||||
}
|
||||
|
||||
public static <K> ChronicleSetBuilder<K> of(Class<K> keyClass) {
|
||||
@@ -88,7 +86,8 @@ public class ChronicleSetBuilder<E>
|
||||
* .entries(50000)
|
||||
* .keySize(10)
|
||||
* .create();}</pre>
|
||||
* (Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded (and each
|
||||
*
|
||||
* <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}.)
|
||||
*
|
||||
@@ -165,11 +164,18 @@ public class ChronicleSetBuilder<E>
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChronicleSetBuilder<E> replicators(byte identifier, Replicator... replicators) {
|
||||
public ChronicleSetBuilder<E> replicators(byte identifier, ReplicationConfig... replicators) {
|
||||
chronicleMapBuilder.replicators(identifier, replicators);
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChronicleSetBuilder<E> channel(ChannelProvider.ChronicleChannel chronicleChannel) {
|
||||
chronicleMapBuilder.channel(chronicleChannel);
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return " ChronicleSetBuilder{" +
|
||||
@@ -190,12 +196,6 @@ public class ChronicleSetBuilder<E>
|
||||
return toString().hashCode();
|
||||
}
|
||||
|
||||
public ChronicleSetBuilder<E> replicators(byte identifier, Replicator replicator) {
|
||||
chronicleMapBuilder.replicators(identifier, replicator);
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public ChronicleSetBuilder<E> timeProvider(TimeProvider timeProvider) {
|
||||
chronicleMapBuilder.timeProvider(timeProvider);
|
||||
@@ -229,108 +229,15 @@ public class ChronicleSetBuilder<E>
|
||||
|
||||
@Override
|
||||
public ChronicleSet<E> create(File file) throws IOException {
|
||||
final ChronicleMap<E, Void> map = chronicleMapBuilder.create(file);
|
||||
final ChronicleMap<E, DummyValue> map = chronicleMapBuilder.create(file);
|
||||
return new SetFromMap<E>(map);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChronicleSet<E> create() throws IOException {
|
||||
final ChronicleMap<E, Void> map = chronicleMapBuilder.create();
|
||||
final ChronicleMap<E, DummyValue> map = chronicleMapBuilder.create();
|
||||
return new SetFromMap<E>(map);
|
||||
}
|
||||
|
||||
private static class SetFromMap<E> extends AbstractSet<E>
|
||||
implements ChronicleSet<E>, Serializable {
|
||||
|
||||
private final ChronicleMap<E, Void> m; // The backing map
|
||||
private transient Set<E> s; // Its keySet
|
||||
|
||||
SetFromMap(ChronicleMap<E, Void> map) {
|
||||
if (!map.isEmpty())
|
||||
throw new IllegalArgumentException("Map is non-empty");
|
||||
m = map;
|
||||
s = map.keySet();
|
||||
}
|
||||
|
||||
public void clear() {
|
||||
m.clear();
|
||||
}
|
||||
|
||||
public int size() {
|
||||
return m.size();
|
||||
}
|
||||
|
||||
public boolean isEmpty() {
|
||||
return m.isEmpty();
|
||||
}
|
||||
|
||||
public boolean contains(Object o) {
|
||||
return m.containsKey(o);
|
||||
}
|
||||
|
||||
public boolean remove(Object o) {
|
||||
return m.remove(o) != null;
|
||||
}
|
||||
|
||||
public boolean add(E e) {
|
||||
return m.put(e, (Void) null) == null;
|
||||
}
|
||||
|
||||
public Iterator<E> iterator() {
|
||||
return s.iterator();
|
||||
}
|
||||
|
||||
public Object[] toArray() {
|
||||
return s.toArray();
|
||||
}
|
||||
|
||||
public <T> T[] toArray(T[] a) {
|
||||
return s.toArray(a);
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return s.toString();
|
||||
}
|
||||
|
||||
public int hashCode() {
|
||||
return s.hashCode();
|
||||
}
|
||||
|
||||
public boolean equals(Object o) {
|
||||
return o == this || s.equals(o);
|
||||
}
|
||||
|
||||
public boolean containsAll(Collection<?> c) {
|
||||
return s.containsAll(c);
|
||||
}
|
||||
|
||||
public boolean removeAll(Collection<?> c) {
|
||||
return s.removeAll(c);
|
||||
}
|
||||
|
||||
public boolean retainAll(Collection<?> c) {
|
||||
return s.retainAll(c);
|
||||
}
|
||||
// addAll is the only inherited implementation
|
||||
|
||||
private static final long serialVersionUID = 2454657854757543876L;
|
||||
|
||||
private void readObject(java.io.ObjectInputStream stream)
|
||||
throws IOException, ClassNotFoundException {
|
||||
stream.defaultReadObject();
|
||||
s = m.keySet();
|
||||
}
|
||||
|
||||
@Override
|
||||
public long longSize() {
|
||||
return m.longSize();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
m.close();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.set;
|
||||
|
||||
enum DummyValue {
|
||||
DUMMY_VALUE
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.set;
|
||||
|
||||
import net.openhft.chronicle.map.serialization.AgileBytesMarshaller;
|
||||
import net.openhft.lang.io.Bytes;
|
||||
|
||||
import static net.openhft.chronicle.set.DummyValue.DUMMY_VALUE;
|
||||
|
||||
enum DummyValueMarshaller implements AgileBytesMarshaller<DummyValue> {
|
||||
INSTANCE;
|
||||
|
||||
@Override
|
||||
public boolean startsWith(Bytes bytes, DummyValue dummyValue) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public long hash(DummyValue dummyValue) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public DummyValue read(Bytes bytes, long size) {
|
||||
return DUMMY_VALUE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public DummyValue read(Bytes bytes, long size, DummyValue dummyValue) {
|
||||
return DUMMY_VALUE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long size(DummyValue dummyValue) {
|
||||
return 0L;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void write(Bytes bytes, DummyValue dummyValue) {
|
||||
// do nothing
|
||||
}
|
||||
|
||||
@Override
|
||||
public int sizeEncodingSize(long size) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeSize(Bytes bytes, long size) {
|
||||
// do nothing
|
||||
}
|
||||
|
||||
@Override
|
||||
public long readSize(Bytes bytes) {
|
||||
return 0L;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.set;
|
||||
|
||||
import net.openhft.chronicle.map.ChronicleMap;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.io.Serializable;
|
||||
import java.util.AbstractSet;
|
||||
import java.util.Collection;
|
||||
import java.util.Iterator;
|
||||
import java.util.Set;
|
||||
|
||||
import static net.openhft.chronicle.set.DummyValue.DUMMY_VALUE;
|
||||
|
||||
class SetFromMap<E> extends AbstractSet<E>
|
||||
implements ChronicleSet<E>, Serializable {
|
||||
|
||||
private final ChronicleMap<E, DummyValue> m; // The backing map
|
||||
private transient Set<E> s; // Its keySet
|
||||
|
||||
SetFromMap(ChronicleMap<E, DummyValue> map) {
|
||||
if (!map.isEmpty())
|
||||
throw new IllegalArgumentException("Map is non-empty");
|
||||
m = map;
|
||||
s = map.keySet();
|
||||
}
|
||||
|
||||
public void clear() {
|
||||
m.clear();
|
||||
}
|
||||
|
||||
public int size() {
|
||||
return m.size();
|
||||
}
|
||||
|
||||
public boolean isEmpty() {
|
||||
return m.isEmpty();
|
||||
}
|
||||
|
||||
public boolean contains(Object o) {
|
||||
return m.containsKey(o);
|
||||
}
|
||||
|
||||
public boolean remove(Object o) {
|
||||
return m.remove(o) == DUMMY_VALUE;
|
||||
}
|
||||
|
||||
public boolean add(E e) {
|
||||
return m.putIfAbsent(e, DUMMY_VALUE) == null;
|
||||
}
|
||||
|
||||
public Iterator<E> iterator() {
|
||||
return s.iterator();
|
||||
}
|
||||
|
||||
public Object[] toArray() {
|
||||
return s.toArray();
|
||||
}
|
||||
|
||||
public <T> T[] toArray(T[] a) {
|
||||
return s.toArray(a);
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return s.toString();
|
||||
}
|
||||
|
||||
public int hashCode() {
|
||||
return s.hashCode();
|
||||
}
|
||||
|
||||
public boolean equals(Object o) {
|
||||
return o == this || s.equals(o);
|
||||
}
|
||||
|
||||
public boolean containsAll(Collection<?> c) {
|
||||
return s.containsAll(c);
|
||||
}
|
||||
|
||||
public boolean removeAll(Collection<?> c) {
|
||||
return s.removeAll(c);
|
||||
}
|
||||
|
||||
public boolean retainAll(Collection<?> c) {
|
||||
return s.retainAll(c);
|
||||
}
|
||||
// addAll is the only inherited implementation
|
||||
|
||||
private static final long serialVersionUID = 2454657854757543876L;
|
||||
|
||||
private void readObject(java.io.ObjectInputStream stream)
|
||||
throws IOException, ClassNotFoundException {
|
||||
stream.defaultReadObject();
|
||||
s = m.keySet();
|
||||
}
|
||||
|
||||
@Override
|
||||
public long longSize() {
|
||||
return m.longSize();
|
||||
}
|
||||
|
||||
@Override
|
||||
public File file() {
|
||||
return m.file();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
m.close();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user