Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-213 - started to integrate MapByName and improved the stateless clients handling of large values

Browse Source
This commit is contained in:
Rob Austin
2014-11-17 19:31:56 +00:00
parent a76e1048f9
commit f854e96361
9 changed files with 288 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java
+13 -15
View File
@@ -22,33 +22,31 @@ import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import org.jetbrains.annotations.NotNull;
/**
* A token which should be created via {@link ReplicationHub#createChannel(short)} call and passed
* to {@link ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)} method, to
* establish a channel replication. See
* <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
* A token which should be created via {@link ReplicationHub#createChannel(int)} call and passed to {@link
* ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)} method, to establish a channel
* replication. See <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
* the corresponding section in ChronicleMap manual</a> for more information.
*
* <p>A {@code ReplicationChannel} could be used to replicate only one {@link ChronicleHash}
* instance.
* <p>A {@code ReplicationChannel} could be used to replicate only one {@link ChronicleHash} instance.
*
* @see ReplicationHub
* @see ReplicationHub#createChannel(short)
* @see ReplicationHub#createChannel(int)
*/
public final class ReplicationChannel {
private ReplicationHub hub;
private final short channelId;
private final int channelId;
ReplicationChannel(ReplicationHub hub, short channelId) {
ReplicationChannel(ReplicationHub hub, int channelId) {
this.hub = hub;
this.channelId = channelId;
}
/**
* Returns the {@link ReplicationHub} on which this {@code ReplicationChannel} was {@linkplain
* ReplicationHub#createChannel(short) created}.
* ReplicationHub#createChannel(int) created}.
*
* @return the {@code ReplicationHub} to which this {@code ReplicationChannel} belongs
* @see ReplicationHub#createChannel(short)
* @see ReplicationHub#createChannel(int)
*/
@NotNull
public ReplicationHub hub() {
@@ -56,13 +54,13 @@ public final class ReplicationChannel {
}
/**
* Returns the identifier of this channel, with which was {@linkplain
* ReplicationHub#createChannel(short) created}.
* Returns the identifier of this channel, with which was {@linkplain ReplicationHub#createChannel(int)
* created}.
*
* @return the identifier of this channel
* @see ReplicationHub#createChannel(short)
* @see ReplicationHub#createChannel(int)
*/
public short channelId() {
public int channelId() {
return channelId;
}
}
src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
+38 -40
View File
@@ -24,14 +24,12 @@ import net.openhft.chronicle.set.ChronicleSet;
import org.jetbrains.annotations.NotNull;
/**
* This class holds all configurations of
* <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
* This class holds all configurations of <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
* multicontainer replication</a>, which is usable, when you want to replicate several {@linkplain
* ChronicleHash ChronicleHashes} ({@linkplain ChronicleMap maps}, {@linkplain ChronicleSet sets},
* etc.) between same servers. {@code ReplicationHub} allows to share TCP/UDP connection, sockets,
* buffers, worker threads for that, considerably reducing resources usage and increasing cumulative
* (for all replicated containers) replication throughput. See
* <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
* ChronicleHash ChronicleHashes} ({@linkplain ChronicleMap maps}, {@linkplain ChronicleSet sets}, etc.)
* between same servers. {@code ReplicationHub} allows to share TCP/UDP connection, sockets, buffers, worker
* threads for that, considerably reducing resources usage and increasing cumulative (for all replicated
* containers) replication throughput. See <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
* the corresponding section in ChronicleMap manual</a> for more information.
*
* <p>Create instances of this class using this pattern: <pre>{@code
@@ -67,14 +65,14 @@ public final class ReplicationHub extends AbstractReplication {
private ReplicationHub(byte localIdentifier, Builder builder) {
super(localIdentifier, builder);
maxEntrySize = builder.maxEntrySize;
channels = new ReplicationChannel[builder.maxNumberOfChronicles];
channels = new ReplicationChannel[builder.maxNumberOfChannels];
}
/**
* Returns the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of {@link
* ChronicleHash}es, replicated by this {@code ReplicationHub}, could have. {@code
* RuntimeException} is thrown on attempt to replicate with this {@code ReplicationHub}
* a {@code ChronicleHash} with larger entry size.
* ChronicleHash}es, replicated by this {@code ReplicationHub}, could have. {@code RuntimeException} is
* thrown on attempt to replicate with this {@code ReplicationHub} a {@code ChronicleHash} with larger
* entry size.
*
* @return the maximum entry size
* @see Builder#maxEntrySize(int)
@@ -84,35 +82,34 @@ public final class ReplicationHub extends AbstractReplication {
}
/**
* Returns the maximum number of channels could be {@linkplain #createChannel(short) created}
* for this {@code ReplicationHub}.
* Returns the maximum number of channels could be {@linkplain #createChannel(int) created} for this
* {@code ReplicationHub}.
*
* @return the maximum number of channels
* @see Builder#maxNumberOfChannels()
* @see #createChannel(short)
* @see #createChannel(int)
*/
public int maxNumberOfChannels() {
return channels.length;
}
/**
* Creates a new {@link ReplicationChannel} in this {@code ReplicationHub} with the given
* identifier. Identifier shouldn't be lesser than zero and greater or equal to
* {@code maxNumberOfChannels() - 1}. On a {@code ReplicationHub} instance,
* {@code ReplicationChannel} could be created only once for each possible {@code channelId}
* value.
* Creates a new {@link ReplicationChannel} in this {@code ReplicationHub} with the given identifier.
* Identifier shouldn't be lesser than zero and greater or equal to {@code maxNumberOfChannels() - 1}. On
* a {@code ReplicationHub} instance, {@code ReplicationChannel} could be created only once for each
* possible {@code channelId} value.
*
* @param channelId the identifier of the channel. Should be equal for replicated containers
* on different nodes (servers)
* @return a new {@code ReplicationChannel} instance, in fact just incapsulating the given
* {@code channelId} for this {@code ReplicationHub}
* @throws IllegalArgumentException if the specified {@code channelId} is out of<br>
* <code>[0, {@link #maxNumberOfChannels()})</code> range
* @throws IllegalStateException if {@code ReplicationChannel} with the specified
* {@code channelId} has already been acquired on this {@code ReplicationHub}
* @param channelId the identifier of the channel. Should be equal for replicated containers on different
* nodes (servers)
* @return a new {@code ReplicationChannel} instance, in fact just incapsulating the given {@code
* channelId} for this {@code ReplicationHub}
* @throws IllegalArgumentException if the specified {@code channelId} is out of<br> <code>[0, {@link
* #maxNumberOfChannels()})</code> range
* @throws IllegalStateException if {@code ReplicationChannel} with the specified {@code channelId} has
* already been acquired on this {@code ReplicationHub}
* @see ReplicationChannel
*/
public synchronized ReplicationChannel createChannel(short channelId) {
public synchronized ReplicationChannel createChannel(int channelId) {
if (channelId < 0)
throw new IllegalArgumentException("channelId should be positive");
if (channelId >= maxNumberOfChannels())
@@ -131,41 +128,42 @@ public final class ReplicationHub extends AbstractReplication {
*/
public static final class Builder extends AbstractReplication.Builder<ReplicationHub, Builder> {
private int maxEntrySize = 1024;
private int maxNumberOfChronicles = 128;
private int maxNumberOfChannels = 128;
private Builder() {}
private Builder() {
}
/**
* Configures the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of
* {@link ChronicleHash}es, replicated via {@code ReplicationHub}s, created by this builder.
* Configures the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of {@link
* ChronicleHash}es, replicated via {@code ReplicationHub}s, created by this builder.
*
* <p>Default value is {@code 1024}.
*
* @param maxEntrySize {@link ReplicationHub#maxEntrySize()} of {@link ReplicationHub}s,
* created by this builder
* @param maxEntrySize {@link ReplicationHub#maxEntrySize()} of {@link ReplicationHub}s, created by
* this builder
* @return this builder object back, for chaining
* @see ReplicationHub#maxEntrySize()
*/
@NotNull
public Builder maxEntrySize(int maxEntrySize) {
this.maxEntrySize = maxEntrySize;
this.maxEntrySize = maxEntrySize * 64;
return this;
}
/**
* Configures the maximum number of channels could be {@linkplain #createChannel(short)}
* created for {@code ReplicationHub}s, created by this builder.
* Configures the maximum number of channels could be {@linkplain #createChannel(int)} created for
* {@code ReplicationHub}s, created by this builder.
*
* <p>Default value is {@code 128}.
*
* @param maxNumberOfChannels {@link ReplicationHub#maxNumberOfChannels()} of
* {@link ReplicationHub}s, created by this builder
* @param maxNumberOfChannels {@link ReplicationHub#maxNumberOfChannels()} of {@link ReplicationHub}s,
* created by this builder
* @return this builder object back, for chaining
* @see ReplicationHub#maxNumberOfChannels()
*/
@NotNull
public Builder maxNumberOfChannels(int maxNumberOfChannels) {
this.maxNumberOfChronicles = maxNumberOfChannels;
this.maxNumberOfChannels = maxNumberOfChannels;
return this;
}
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+6 -5
View File
@@ -53,6 +53,7 @@ final class ChannelProvider implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(ChannelProvider.class.getName());
static final Map<ReplicationHub, ChannelProvider> implMapping = new IdentityHashMap<>();
static synchronized ChannelProvider getProvider(ReplicationHub hub) throws IOException {
ChannelProvider channelProvider = implMapping.get(hub);
if (channelProvider != null)
@@ -287,7 +288,7 @@ final class ChannelProvider implements Closeable {
return new SingleThreadedDirectBitSet(bytes);
}
public ChronicleChannel createChannel(short channel) {
public ChronicleChannel createChannel(int channel) {
return new ChronicleChannel(channel);
}
@@ -306,7 +307,7 @@ final class ChannelProvider implements Closeable {
}
}
private ByteBufferBytes toBootstrapMessage(short chronicleChannel, final long lastModificationTime) {
private ByteBufferBytes toBootstrapMessage(int chronicleChannel, final long lastModificationTime) {
final ByteBufferBytes writeBuffer = new ByteBufferBytes(ByteBuffer.allocate(1 + 1 + 2 + 8));
writeBuffer.writeByte(BOOTSTRAP_MESSAGE);
writeBuffer.writeByte(localIdentifier);
@@ -316,7 +317,7 @@ final class ChannelProvider implements Closeable {
return writeBuffer;
}
private void add(short chronicleChannel,
private void add(int chronicleChannel,
Replica replica, EntryExternalizable entryExternalizable) {
channelDataLock.writeLock().lock();
try {
@@ -481,9 +482,9 @@ final class ChannelProvider implements Closeable {
public class ChronicleChannel extends Replicator implements Closeable {
private final short chronicleChannel;
private final int chronicleChannel;
private ChronicleChannel(short chronicleChannel) {
private ChronicleChannel(int chronicleChannel) {
this.chronicleChannel = chronicleChannel;
}
src/main/java/net/openhft/chronicle/map/FindMapByName.java
+58 -23
View File
@@ -16,49 +16,84 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.replication.ReplicationHub;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import java.io.File;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static net.openhft.chronicle.map.ChronicleMapBuilder.of;
/**
* @author Rob Austin.
*/
class FindMapByName {
int maxEntrySize = 10*1024;
public static final int MAP_BY_NAME_CHANNEL = 1;
private final ReplicationHub replicationHub;
private Map<String,ChronicleMapBuilder> map;
int maxEntrySize = 10 * 1024;
public FindMapByName(byte identifier) throws IOException {
private final AtomicInteger nextFreeChannel = new AtomicInteger(1);
private final Map<String, ChronicleMapBuilder> map;
final TcpTransportAndNetworkConfig tcpConfig =
TcpTransportAndNetworkConfig.of(8076)
.heartBeatInterval(1, TimeUnit.SECONDS);
replicationHub = ReplicationHub.builder()
.maxEntrySize(maxEntrySize)
.tcpTransportAndNetwork(tcpConfig)
.createWithId(identifier);
map = (Map)ChronicleMapBuilder.of(CharSequence.class, ChronicleMapBuilder.class)
/**
* @throws IOException
*/
public FindMapByName(ReplicationHub replicationHub) throws IOException {
this.replicationHub = replicationHub;
this.map = (Map) of(CharSequence.class, ChronicleMapBuilder.class)
.entrySize(300)
.instance()
.replicatedViaChannel(replicationHub.createChannel((short) 1)).create();
.replicatedViaChannel(replicationHub.createChannel((short) MAP_BY_NAME_CHANNEL)).create();
}
public ChronicleMapBuilder get(String name) throws IOException {
return map.get(name);
ChronicleMapBuilder get(String name) throws IllegalArgumentException {
ChronicleMapBuilder chronicleMapBuilder = map.get(name);
if (chronicleMapBuilder == null)
throw new IllegalArgumentException("A map of this name can not be found name=" + name);
return chronicleMapBuilder;
}
public void add(ChronicleMapBuilder builder) throws IOException {
builder.name();
public <T extends ChronicleHash> T create(String name) throws IllegalArgumentException,
IOException {
return (T) replicatedViaChannel(name).create();
}
private ChronicleHashInstanceConfig replicatedViaChannel(String name) {
return toReplicatedViaChannel(get(name));
}
/**
* similar to {@link ChronicleMapBuilder#createPersistedTo(File file) }
*
* @param name the name of the map to be created
* @param file the file which will be used as shared off heap storage
* @param <T>
* @return a new instance of ChronicleMap
* @throws IllegalArgumentException
* @throws IOException
* @see ChronicleMapBuilder#createPersistedTo(File file)
*/
public <T extends ChronicleHash> T createPersistedTo(String name, File file) throws IllegalArgumentException,
IOException {
return (T) replicatedViaChannel(name).persistedTo(file).create();
}
private ChronicleHashInstanceConfig toReplicatedViaChannel(ChronicleMapBuilder builder) {
int channelId = nextFreeChannel.incrementAndGet();
if (channelId > replicationHub.maxNumberOfChannels())
throw new IllegalStateException("There are no more free channels, you can increase the number " +
"of changes in the replicationHub by calling replicationHub.maxNumberOfChannels(..);");
return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) channelId));
}
public void add(ChronicleMapBuilder builder) {
this.map.put(builder.name(), builder);
}
}
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+28 -12
View File
@@ -19,7 +19,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.RemoteNodeValidator;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.ReplicationHub;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
import net.openhft.lang.collection.ATSDirectBitSet;
@@ -56,15 +56,33 @@ import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
*/
public class NodeDiscovery {
public static final int DEFAULT_UDP_BROADCAST_PORT = 8124;
public static final short DEFAULT_TCP_PORT = (short) 8123;
public static final int SERIALIZED_ENTRY_SIZE = 1024;
private final AddressAndPort ourAddressAndPort;
private final UdpTransportConfig udpConfig;
public static final InetAddress DEFAULT_UDP_MASK;
static {
InetAddress mask = null;
try {
mask = Inet4Address.getByName("255.255.255.255");
} catch (Exception e) {
mask = null;
}
DEFAULT_UDP_MASK = mask;
}
private final DiscoveryNodeBytesMarshallable discoveryNodeBytesMarshallable;
private final AtomicReference<NodeDiscoveryEventListener> nodeDiscoveryEventListenerAtomicReference =
new AtomicReference<NodeDiscoveryEventListener>();
private KnownNodes knownNodes;
public NodeDiscovery() throws IOException {
this((short) 8124, (short) 8123, getDefaultAddress(), Inet4Address.getByName("255.255.255.255"));
this((short) DEFAULT_UDP_BROADCAST_PORT, DEFAULT_TCP_PORT, getDefaultAddress(), DEFAULT_UDP_MASK);
}
public NodeDiscovery(short udpBroadcastPort,
@@ -79,13 +97,13 @@ public class NodeDiscovery {
(knownNodes, nodeDiscoveryEventListenerAtomicReference, ourAddressAndPort);
final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster
= new NodeDiscoveryBroadcaster(udpConfig, 1024, discoveryNodeBytesMarshallable);
= new NodeDiscoveryBroadcaster(udpConfig, SERIALIZED_ENTRY_SIZE, discoveryNodeBytesMarshallable);
discoveryNodeBytesMarshallable.setModificationNotifier(nodeDiscoveryBroadcaster);
}
public synchronized ChronicleMap<Integer, CharSequence> discoverMap() throws
public synchronized FindMapByName discoverMap() throws
IOException, InterruptedException {
final AtomicInteger ourProposedIdentifier = new AtomicInteger();
@@ -229,14 +247,12 @@ public class NodeDiscovery {
LOG.info("Using Remote identifier=" + identifier);
nodeDiscoveryEventListenerAtomicReference.set(null);
return ChronicleMapBuilder.of(Integer.class,
CharSequence.class)
.entries(20000L)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfig)
.remoteNodeValidator(remoteNodeValidator)
.createWithId(identifier))
.create();
final ReplicationHub replicationHub = ReplicationHub.builder()
.maxEntrySize(10 * 1024)
.tcpTransportAndNetwork(tcpConfig)
.createWithId(identifier);
return new FindMapByName(replicationHub);
}
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+71 -36
View File
@@ -267,7 +267,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
writeKey(key, local);
writeValue(value, local);
return readKey(sizeLocation, local);
return readValue(sizeLocation, local);
}
@@ -311,7 +311,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
writeValue(value, local);
// get the data back from the server
return readKey(sizeLocation, local);
return readValue(sizeLocation, local);
}
@@ -394,7 +394,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
writeKey((K) key, local);
// get the data back from the server
return readKey(sizeLocation, local);
return readValue(sizeLocation, local);
}
@@ -437,7 +437,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
return null;
} else
// get the data back from the server
return readKey(sizeLocation);
return readValue(sizeLocation);
}
@@ -456,7 +456,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
return null;
} else
// get the data back from the server
return readKey(sizeLocation, local);
return readValue(sizeLocation, local);
}
@@ -499,9 +499,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) {
LOG.debug("resizing buffer");
resizeBuffer(buffer.capacity() + 1024);
bytes.limit(bytes.capacity());
bytes.position(start);
resizeBuffer(buffer.capacity() + maxEntrySize, start);
} else
throw e;
@@ -562,12 +560,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
numberOfEntriesReadSoFar++;
long start = bytes.position();
// putAll if a bit more complicated than the others
// as the volume of data could cause us to have to resize our buffers
resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar);
resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar, start);
final long start = bytes.position();
final K key = e.getKey();
final Class<?> keyClass = key.getClass();
@@ -597,15 +595,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
}
private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar) {
private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) {
final long remaining = bytes.remaining();
if (remaining < maxEntrySize) {
// todo check this again as I think it may have a bug
long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar);
resizeBuffer(estimatedRequiredSize + maxEntrySize);
resizeBuffer((int) (estimatedRequiredSize + maxEntrySize), start);
}
}
@@ -622,7 +618,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
void resizeBuffer(long size) {
private void resizeBuffer(int size, long start) {
if (size < buffer.capacity())
throw new IllegalStateException("it not possible to resize the buffer smaller");
@@ -663,7 +659,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
assert buffer.capacity() == size;
assert buffer.capacity() == bytes.capacity();
assert bytes.limit() == bytes.capacity();
bytes.position(start);
}
@@ -964,6 +960,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
bytes = new ByteBufferBytes(allocateDirect(requiredSize));
}
bytes.limit(bytes.capacity());
// block until we have received all the byte in this chunk
receive(size, timeoutTime);
@@ -1054,45 +1052,82 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private void writeKey(K key) {
try {
keyValueSerializer.writeKey(key, bytes);
} catch (IllegalArgumentException e) {
resizeBuffer(bytes.capacity() + maxEntrySize);
long start = bytes.position();
for (; ; ) {
try {
keyValueSerializer.writeKey(key, bytes);
return;
} catch (IllegalArgumentException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
}
private void writeKey(K key, ThreadLocalCopies local) {
try {
keyValueSerializer.writeKey(key, bytes, local);
} catch (IllegalArgumentException e) {
resizeBuffer(bytes.capacity() + maxEntrySize);
long start = bytes.position();
for (; ; ) {
try {
keyValueSerializer.writeKey(key, bytes, local);
return;
} catch (IllegalArgumentException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
}
private V readKey(final long sizeLocation) {
private V readValue(final long sizeLocation) {
return keyValueSerializer.readValue(blockingFetch(sizeLocation), null);
}
private V readKey(final long sizeLocation, ThreadLocalCopies local) {
private V readValue(final long sizeLocation, ThreadLocalCopies local) {
return keyValueSerializer.readValue(blockingFetch(sizeLocation), local);
}
private void writeValue(V value) {
try {
keyValueSerializer.writeValue(value, bytes, null);
} catch (IllegalArgumentException e) {
resizeBuffer(bytes.capacity() + maxEntrySize);
long start = bytes.position();
for (; ; ) {
try {
keyValueSerializer.writeValue(value, bytes, null);
return;
} catch (IllegalArgumentException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
}
}
}
private void writeValue(V value, ThreadLocalCopies local) {
try {
keyValueSerializer.writeValue(value, bytes, local);
} catch (IllegalArgumentException e) {
resizeBuffer(bytes.capacity() + maxEntrySize);
long start = bytes.position();
for (; ; ) {
try {
keyValueSerializer.writeValue(value, bytes, null);
return;
} catch (IllegalArgumentException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
}
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+1 -2
View File
@@ -2022,9 +2022,8 @@ class KeyValueSerializer<K, V> {
assert writer.limit() == writer.capacity();
writer.writeBoolean(value == null);
if (value != null) {
if (value != null)
valueSerializer.writeMarshallable(value, writer, threadLocal);
}
}