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-185 fixed a number of bugs in old code and new VanillaChronicleMap and ReplicatedChronicleMap (generalized instance/bytes queries), refactoring and minor fixes, switched to bytes queries for stateless client server

Browse Source
This commit is contained in:
leventov
2014-11-29 20:39:00 +07:00
parent 9ba784c8a0
commit 258347ed71
10 changed files with 560 additions and 562 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReaderWithSize.java
+69
View File
@@ -0,0 +1,69 @@
/*
* 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.map;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.SizeMarshaller;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.Nullable;
final class ReaderWithSize<T> {
private final SizeMarshaller sizeMarshaller;
private final BytesReader<T> originalReader;
private final Provider<BytesReader<T>> readerProvider;
ReaderWithSize(SerializationBuilder<T> serializationBuilder) {
sizeMarshaller = serializationBuilder.sizeMarshaller();
originalReader = serializationBuilder.reader();
readerProvider = (Provider<BytesReader<T>>) Provider.of(originalReader.getClass());
}
public ThreadLocalCopies getCopies(ThreadLocalCopies copies) {
return readerProvider.getCopies(copies);
}
public T read(Bytes in, @Nullable ThreadLocalCopies copies) {
long size = sizeMarshaller.readSize(in);
copies = readerProvider.getCopies(copies);
BytesReader<T> reader = readerProvider.get(copies, originalReader);
return reader.read(in, size);
}
public T readNullable(Bytes in, @Nullable ThreadLocalCopies copies) {
if (in.readBoolean())
return null;
return read(in, copies);
}
public BytesReader<T> readerForLoop(@Nullable ThreadLocalCopies copies) {
copies = readerProvider.getCopies(copies);
return readerProvider.get(copies, originalReader);
}
public T readInLoop(Bytes in, BytesReader<T> reader) {
long size = sizeMarshaller.readSize(in);
return reader.read(in, size);
}
public T readNullableInLoop(Bytes in, BytesReader<T> reader) {
if (in.readBoolean())
return null;
return readInLoop(in, reader);
}
}
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+33 -20
View File
@@ -744,10 +744,13 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
Bytes keyBytes, long hash2, final long timestamp, final byte identifier) {
writeLock(null);
try {
ReadValueToBytes readValueToLazyBytes = segmentState.readValueToLazyBytes;
readValueToLazyBytes.valueSizeMarshaller(valueSizeMarshaller);
Boolean removed = (Boolean) removeWithoutLock(copies, segmentState,
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, keyBytes, keyBytes.remaining(),
keyBytesToInstance, null, null, null, hash2, true,
keyBytesToInstance, null, null, outputValueBytesToInstance,
hash2, readValueToLazyBytes, true,
timestamp, identifier, true, true);
if (!removed && LOG.isDebugEnabled()) {
LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() +
@@ -971,19 +974,19 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
@Override
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
VB, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
Object remove(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB expectedValue,
InstanceOrBytesToInstance<? super VB, V> toValue,
long hash2, boolean resultUnused) {
InstanceOrBytesToInstance<RV, V> toValue,
long hash2, ReadValue<RV> readValue, boolean resultUnused) {
writeLock(null);
try {
return removeWithoutLock(copies, segmentState,
metaKeyInterop, keyInterop, key, keySize, toKey,
getValueInterops, expectedValue, toValue,
hash2, resultUnused, currentTime(), localIdentifier, false,
hash2, readValue, resultUnused, currentTime(), localIdentifier, false,
expectedValue != null);
} finally {
if (segmentState != null)
@@ -999,16 +1002,16 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
@Override
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
VB, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
Object removeWithoutLock(
@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB expectedValue,
InstanceOrBytesToInstance<? super VB, V> toValue,
long hash2, boolean resultUnused) {
InstanceOrBytesToInstance<RV, V> toValue,
long hash2, ReadValue<RV> readValue, boolean resultUnused) {
return removeWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key, keySize,
toKey, getValueInterops, expectedValue, toValue, hash2, resultUnused,
toKey, getValueInterops, expectedValue, toValue, hash2, readValue, resultUnused,
currentTime(), localIdentifier, false, expectedValue != null);
}
@@ -1021,14 +1024,14 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
* - if resultUnused is true, null is always returned
*/
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
VB, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
Object removeWithoutLock(
@Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB expectedValue,
InstanceOrBytesToInstance<? super VB, V> toValue,
long hash2, boolean resultUnused,
InstanceOrBytesToInstance<RV, V> toValue,
long hash2, ReadValue<RV> readValue, boolean resultUnused,
long timestamp, byte identifier, boolean remote, boolean booleanResult) {
segmentStateNotNullImpliesCopiesNotNull(copies, segmentState);
assert identifier > 0;
@@ -1061,7 +1064,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
// we should ignore only remote updates
// which don't use remove, put results
// assert expectedValue != null || resultUnused;
// assert booleanResult || resultUnused;
return booleanResult ? Boolean.FALSE : null;
}
boolean isDeleted = entry.readBoolean();
@@ -1093,12 +1096,16 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
entry.writeBoolean(true);
entry.position(valuePos);
return removeEntry(copies, key, keySize, toKey, expectedValue, toValue,
resultUnused, hashLookup, entry, pos, valueSizePos, valueSize,
remote, false, booleanResult);
return removeEntry(copies, key, keySize, toKey, toValue, readValue,
resultUnused, hashLookup, entry, pos, valueSizePos,
valueSize, remote, false, booleanResult);
}
// key is not found
return booleanResult ? Boolean.FALSE : null;
if (booleanResult) {
return Boolean.FALSE;
} else {
return resultUnused ? null : readValue.readNull();
}
} finally {
if (segmentState == null && localSegmentState != null)
localSegmentState.close();
@@ -1139,8 +1146,14 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
entry.skip(keySize);
long timestampPos = entry.position();
if (shouldIgnore(entry, timestamp, identifier))
return null;
if (shouldIgnore(entry, timestamp, identifier)) {
LOG.error("Trying to replace a value for key={} on the node with id={} " +
"at time={} (current time), but the entry is updated by node " +
"with id={} at time={}. Time is not monotonic across nodes!?",
key, identifier, timestamp, entry.readByte(entry.position() - 1),
entry.readLong(entry.position() - ADDITIONAL_ENTRY_BYTES + 1));
return readValue.readNull();
}
boolean isDeleted = entry.readBoolean();
if (isDeleted)
break;
@@ -1156,7 +1169,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
return result;
}
// key is not found
return null;
return readValue.readNull();
} finally {
if (segmentState != null)
segmentState.close();
src/main/java/net/openhft/chronicle/map/Replicators.java
+3 -6
View File
@@ -47,12 +47,9 @@ final class Replicators {
final ChronicleMap chronicleMap)
throws IOException {
final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder
.keyBuilder, builder.valueBuilder);
StatelessServerConnector statelessServer = new StatelessServerConnector
(keyValueSerializer, (VanillaChronicleMap) chronicleMap,
builder.entrySize(true));
StatelessServerConnector statelessServer = new StatelessServerConnector(
builder.keyBuilder, builder.valueBuilder,
(VanillaChronicleMap) chronicleMap, builder.entrySize(true));
TcpTransportAndNetworkConfig tcpConfig = replication.tcpTransportAndNetwork();
assert tcpConfig != null;
src/main/java/net/openhft/chronicle/map/Serializer.java
-114
View File
@@ -1,114 +0,0 @@
/*
* 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.map;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
import net.openhft.chronicle.hash.serialization.internal.MetaProvider;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.model.constraints.NotNull;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
/**
* 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, ThreadLocalCopies local) throws
IllegalStateException {
final ThreadLocalCopies copies = (local == null) ? threadLocalCopies() : local;
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 ThreadLocalCopies threadLocalCopies() {
return readerProvider.getCopies(null);
}
ThreadLocalCopies writeMarshallable(O value, Bytes out, ThreadLocalCopies copies0) {
ThreadLocalCopies copies = (copies0 == null) ? threadLocalCopies() : copies0;
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) {
assert out.limit() == out.capacity();
metaValueWriter.write(valueWriter, out, value);
} else
throw new UnsupportedOperationException("");
return copies;
}
}
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+130 -98
View File
@@ -20,6 +20,7 @@ package net.openhft.chronicle.map;
import com.sun.jdi.connect.spi.ClosedConnectionException;
import net.openhft.chronicle.hash.RemoteCallTimeoutException;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.NativeBytes;
@@ -63,7 +64,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private ByteBuffer buffer;
private ByteBufferBytes bytes;
private final KeyValueSerializer<K, V> keyValueSerializer;
private final ReaderWithSize<K> keyReaderWithSize;
private final WriterWithSize<K> keyWriterWithSize;
private final ReaderWithSize<V> valueReaderWithSize;
private final WriterWithSize<V> valueWriterWithSize;
private volatile SocketChannel clientChannel;
private CloseablesManager closeables;
@@ -80,7 +84,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
static enum EventId {
HEARTBEAT,
STATEFUL_UPDATE,
LONG_SIZE, SIZE,
LONG_SIZE,
SIZE,
IS_EMPTY,
CONTAINS_KEY,
CONTAINS_VALUE,
@@ -111,8 +116,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException {
this.chronicleMapBuilder = chronicleMapBuilder;
this.config = config;
this.keyValueSerializer = new KeyValueSerializer<K, V>(chronicleMapBuilder.keyBuilder,
chronicleMapBuilder.valueBuilder);
keyReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.keyBuilder);
keyWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.keyBuilder);
valueReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.valueBuilder);
valueWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.valueBuilder);
this.putReturnsNull = chronicleMapBuilder.putReturnsNull();
this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull();
this.maxEntrySize = chronicleMapBuilder.entrySize(true);
@@ -343,25 +350,24 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
public synchronized V putIfAbsent(K key, V value) {
if (key == null)
if (key == null || value == null)
throw new NullPointerException();
final long sizeLocation = writeEventAnSkip(PUT_IF_ABSENT);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
writeKey(key, local);
writeValue(value, local);
return readValue(sizeLocation, local);
ThreadLocalCopies copies = writeKey(key);
copies = writeValue(value, copies);
return readValue(sizeLocation, copies);
}
public synchronized boolean remove(Object key, Object value) {
if (key == null)
throw new NullPointerException("key can not be null");
throw new NullPointerException();
if (value == null)
return false; // ConcurrentHashMap compat
final long sizeLocation = writeEventAnSkip(REMOVE_WITH_VALUE);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
writeKey((K) key, local);
writeValue((V) value, local);
ThreadLocalCopies copies = writeKey((K) key);
writeValue((V) value, copies);
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
@@ -369,14 +375,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
public synchronized boolean replace(K key, V oldValue, V newValue) {
if (key == null)
throw new NullPointerException("key can not be null");
if (key == null || oldValue == null || newValue == null)
throw new NullPointerException();
final long sizeLocation = writeEventAnSkip(REPLACE_WITH_OLD_AND_NEW_VALUE);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
writeKey(key, local);
writeValue(oldValue, local);
writeValue(newValue, local);
ThreadLocalCopies copies = writeKey(key);
copies = writeValue(oldValue, copies);
writeValue(newValue, copies);
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
@@ -384,16 +389,15 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
public synchronized V replace(K key, V value) {
if (key == null)
throw new NullPointerException("key can not be null");
if (key == null || value == null)
throw new NullPointerException();
final long sizeLocation = writeEventAnSkip(REPLACE);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
writeKey(key, local);
writeValue(value, local);
ThreadLocalCopies copies = writeKey(key);
copies = writeValue(value, copies);
// get the data back from the server
return readValue(sizeLocation, local);
return readValue(sizeLocation, copies);
}
@@ -449,6 +453,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
public synchronized boolean containsKey(Object key) {
if (key == null)
throw keyNotNullNPE();
final long sizeLocation = writeEventAnSkip(CONTAINS_KEY);
writeKey((K) key);
@@ -457,10 +463,16 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private static NullPointerException keyNotNullNPE() {
return new NullPointerException("key can not be null");
}
public synchronized boolean containsValue(Object value) {
if (value == null)
throw new NullPointerException();
final long sizeLocation = writeEventAnSkip(CONTAINS_VALUE);
writeValue((V) value);
writeValue((V) value, null);
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
@@ -474,11 +486,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
public synchronized V get(Object key) {
final long sizeLocation = writeEventAnSkip(GET);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
writeKey((K) key, local);
ThreadLocalCopies copies = writeKey((K) key);
// get the data back from the server
return readValue(sizeLocation, local);
return readValue(sizeLocation, copies);
}
@@ -510,67 +521,58 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
public synchronized V remove(Object key) {
if (key == null)
throw new NullPointerException("key can not be null");
throw keyNotNullNPE();
final long sizeLocation = removeReturnsNull ? writeEvent(REMOVE_WITHOUT_ACC) : writeEventAnSkip(REMOVE);
writeKey((K) key);
ThreadLocalCopies copies = writeKey((K) key);
if (removeReturnsNull) {
sendWithoutAcc(bytes, sizeLocation);
return null;
} else
// get the data back from the server
return readValue(sizeLocation);
return readValue(sizeLocation, copies);
}
public synchronized V put(K key, V value) {
if (key == null)
if (key == null || value == null)
throw new NullPointerException();
final long sizeLocation =
(putReturnsNull) ? writeEvent(PUT_WITHOUT_ACC) : writeEventAnSkip(PUT);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final long sizeLocation = (putReturnsNull) ? writeEvent(PUT_WITHOUT_ACC) : writeEventAnSkip(PUT);
writeKey(key, local);
writeValue(value, local);
ThreadLocalCopies copies = writeKey(key);
copies = writeValue(value, copies);
if (putReturnsNull) {
sendWithoutAcc(bytes, sizeLocation);
return null;
} else
// get the data back from the server
return readValue(sizeLocation, local);
return readValue(sizeLocation, copies);
}
public synchronized <R> R mapForKey(K key, @NotNull Function<? super V, R> function) {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final long sizeLocation = writeEventAnSkip(MAP_FOR_KEY);
writeKey(key, local);
writeKey(key);
writeObject(function);
final Bytes reader = blockingFetch(sizeLocation);
return (R) keyValueSerializer.readObject(reader);
return (R) reader.readObject();
}
@Override
public synchronized <R> R updateForKey(K key, @NotNull Mutator<? super V, R> mutator) {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final long sizeLocation = writeEventAnSkip(UPDATE_FOR_KEY);
writeKey(key, local);
writeKey(key);
writeObject(mutator);
final Bytes reader = blockingFetch(sizeLocation);
return (R) keyValueSerializer.readObject(reader);
return (R) reader.readObject();
}
@@ -598,9 +600,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
public synchronized void putAll(Map<? extends K, ? extends V> map) {
final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL);
final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) :
writeEventAnSkip(PUT_ALL);
writeEntries(map);
writeEntriesForPutAll(map);
if (putReturnsNull)
sendWithoutAcc(bytes, sizeLocation);
@@ -635,15 +638,24 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private synchronized void writeEntries(Map<? extends K, ? extends V> map) {
private synchronized void writeEntriesForPutAll(Map<? extends K, ? extends V> map) {
final int numberOfEntries = map.size();
int numberOfEntriesReadSoFar = 0;
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
bytes.writeStopBit(numberOfEntries);
assert bytes.limit() == bytes.capacity();
ThreadLocalCopies copies = keyWriterWithSize.getCopies(null);
Object keyWriter = keyWriterWithSize.writerForLoop(copies);
copies = valueWriterWithSize.getCopies(copies);
Object valueWriter = valueWriterWithSize.writerForLoop(copies);
for (final Map.Entry<? extends K, ? extends V> e : map.entrySet()) {
final K key = e.getKey();
final V value = e.getValue();
if (key == null || value == null)
throw new NullPointerException();
numberOfEntriesReadSoFar++;
@@ -653,18 +665,16 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// as the volume of data could cause us to have to resize our buffers
resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar, start);
final K key = e.getKey();
final Class<?> keyClass = key.getClass();
if (!kClass.isAssignableFrom(keyClass))
if (!kClass.isAssignableFrom(keyClass)) {
throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " +
"and should" +
" be of type=" + kClass);
}
writeKeyInLoop(key, keyWriter, copies);
writeKey(key, local);
final V value = e.getValue();
final Class<?> valueClass = value.getClass();
if (!vClass.isAssignableFrom(valueClass))
@@ -673,7 +683,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
"should be of type=" + vClass);
writeValue(value, local);
writeValueInLoop(value, valueWriter, copies);
final int len = (int) (bytes.position() - start);
@@ -763,9 +773,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// get the data back from the server
Bytes in = blockingFetch0(sizeLocation, transactionId, startTime);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final Collection<V> result = new ArrayList<V>();
BytesReader<V> valueReader = valueReaderWithSize.readerForLoop(null);
for (; ; ) {
final boolean hasMoreEntries = in.readBoolean();
@@ -774,7 +784,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final long size = in.readInt();
for (int i = 0; i < size; i++) {
result.add(keyValueSerializer.readValue(in, local));
result.add(valueReaderWithSize.readInLoop(in, valueReader));
}
if (!hasMoreEntries)
@@ -798,7 +808,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// get the data back from the server
Bytes in = blockingFetch0(sizeLocation, transactionId, startTime);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
BytesReader<K> keyReader = keyReaderWithSize.readerForLoop(copies);
copies = valueReaderWithSize.getCopies(copies);
BytesReader<V> valueReader = valueReaderWithSize.readerForLoop(copies);
final Map<K, V> result = new HashMap<K, V>();
for (; ; ) {
@@ -809,8 +822,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final long size = in.readInt();
for (int i = 0; i < size; i++) {
final K k = keyValueSerializer.readKey(in, local);
final V v = keyValueSerializer.readValue(in, local);
final K k = keyReaderWithSize.readInLoop(in, keyReader);
final V v = valueReaderWithSize.readInLoop(in, valueReader);
result.put(k, v);
}
@@ -846,7 +859,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// get the data back from the server
Bytes in = blockingFetch0(sizeLocation, transactionId, startTime);
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
BytesReader<K> keyReader = keyReaderWithSize.readerForLoop(null);
final Set<K> result = new HashSet<>();
for (; ; ) {
@@ -857,7 +870,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long size = in.readInt();
for (int i = 0; i < size; i++) {
result.add(keyValueSerializer.readKey(in, local));
result.add(keyReaderWithSize.readInLoop(in, keyReader));
}
if (!hasMoreEntries)
@@ -1144,18 +1157,16 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private void writeKey(K key) {
writeKey(key, null);
private ThreadLocalCopies writeKey(K key) {
return writeKey(key, null);
}
private void writeKey(K key, ThreadLocalCopies local) {
private ThreadLocalCopies writeKey(K key, ThreadLocalCopies copies) {
long start = bytes.position();
for (; ; ) {
try {
keyValueSerializer.writeKey(key, bytes, local);
return;
return keyWriterWithSize.write(bytes, key, copies);
} catch (IndexOutOfBoundsException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
} catch (IllegalArgumentException e) {
@@ -1169,43 +1180,64 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
}
private V readValue(final long sizeLocation) {
return keyValueSerializer.readValue(blockingFetch(sizeLocation), null);
}
private V readValue(final long sizeLocation, ThreadLocalCopies local) {
return keyValueSerializer.readValue(blockingFetch(sizeLocation), local);
}
private void writeValue(V value) {
writeValue(value, null);
}
private void writeValue(V value, ThreadLocalCopies local) {
private ThreadLocalCopies writeKeyInLoop(K key, Object writer, ThreadLocalCopies copies) {
long start = bytes.position();
for (; ; ) {
try {
assert bytes.position() == start;
bytes.limit(bytes.capacity());
keyValueSerializer.writeValue(value, bytes, local);
return;
return keyWriterWithSize.writeInLoop(bytes, key, writer, copies);
} catch (IndexOutOfBoundsException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
} catch (IllegalArgumentException e) {
resizeToMessage(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
}
private V readValue(final long sizeLocation, ThreadLocalCopies copies) {
return valueReaderWithSize.readNullable(blockingFetch(sizeLocation), copies);
}
private ThreadLocalCopies writeValue(V value, ThreadLocalCopies copies) {
long start = bytes.position();
for (; ; ) {
try {
assert bytes.position() == start;
bytes.limit(bytes.capacity());
return valueWriterWithSize.write(bytes, value, copies);
} catch (IllegalArgumentException e) {
resizeToMessage(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
}
private ThreadLocalCopies writeValueInLoop(V value, Object writer, ThreadLocalCopies copies) {
long start = bytes.position();
for (; ; ) {
try {
assert bytes.position() == start;
bytes.limit(bytes.capacity());
return valueWriterWithSize.writeInLoop(bytes, value, writer, copies);
} catch (IllegalArgumentException e) {
resizeToMessage(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
}
private void resizeToMessage(long start, IllegalArgumentException e) {
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+94 -210
View File
@@ -21,6 +21,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.RemoteNodeValidator;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
@@ -1344,8 +1345,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
class StatelessServerConnector<K, V> {
public static final int TRANSACTION_ID_OFFSET = SIZE_OF_SIZE + 1;
public static final boolean MAP_SUPPORTS_BYTES = Boolean.valueOf(System.getProperty
("mapSupportsBytes"));
public static final boolean MAP_SUPPORTS_BYTES = true;
private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class
.getName());
public static final StatelessChronicleMap.EventId[] VALUES
@@ -1354,15 +1354,22 @@ class StatelessServerConnector<K, V> {
public static final int HEADER_SIZE = SIZE_OF_SIZE + SIZE_OF_IS_EXCEPTION + SIZE_OF_TRANSACTION_ID;
private final KeyValueSerializer<K, V> keyValueSerializer;
private final ReaderWithSize<K> keyReaderWithSize;
private final WriterWithSize<K> keyWriterWithSize;
private final ReaderWithSize<V> valueReaderWithSize;
private final WriterWithSize<V> valueWriterWithSize;
private final VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
private double maxEntrySizeBytes;
StatelessServerConnector(@NotNull KeyValueSerializer<K, V> keyValueSerializer,
@NotNull VanillaChronicleMap<K, ?, ?, V, ?, ?> map,
int maxEntrySizeBytes) {
this.keyValueSerializer = keyValueSerializer;
StatelessServerConnector(
SerializationBuilder<K> keySerializationBuilder,
SerializationBuilder<V> valueSerializationBuilder,
@NotNull VanillaChronicleMap<K, ?, ?, V, ?, ?> map, int maxEntrySizeBytes) {
keyReaderWithSize = new ReaderWithSize<>(keySerializationBuilder);
keyWriterWithSize = new WriterWithSize<>(keySerializationBuilder);
valueReaderWithSize = new ReaderWithSize<>(valueSerializationBuilder);
valueWriterWithSize = new WriterWithSize<>(valueSerializationBuilder);
this.map = map;
this.maxEntrySizeBytes = maxEntrySizeBytes;
@@ -1465,15 +1472,11 @@ class StatelessServerConnector<K, V> {
public Work mapForKey(ByteBufferBytes reader, Bytes writer, long sizeLocation) {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K key = readKey(reader, local);
final K key = keyReaderWithSize.read(reader, null);
final Function<V, ?> function = (Function<V, ?>) reader.readObject();
try {
Object result = map.mapForKey(key, function);
keyValueSerializer.writeObject(result, writer);
writer.writeObject(result);
} catch (Throwable e) {
LOG.info("", e);
@@ -1485,15 +1488,11 @@ class StatelessServerConnector<K, V> {
}
public Work updateForKey(ByteBufferBytes reader, Bytes writer, long sizeLocation) {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K key = readKey(reader, local);
final K key = keyReaderWithSize.read(reader, null);
final Mutator<V, ?> mutator = (Mutator<V, ?>) reader.readObject();
try {
Object result = map.updateForKey(key, mutator);
keyValueSerializer.writeObject(result, writer);
writer.writeObject(result);
} catch (Throwable e) {
LOG.info("", e);
@@ -1509,9 +1508,10 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
writer.writeBoolean(map.removeWithValue(reader));
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K key = readKey(reader, local);
final V readValue = readValue(reader, local);
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
copies = valueReaderWithSize.getCopies(copies);
final K key = keyReaderWithSize.read(reader, copies);
final V readValue = valueReaderWithSize.read(reader, copies);
writer.writeBoolean(map.remove(key, readValue));
}
} catch (Throwable e) {
@@ -1526,10 +1526,11 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.replaceWithOldAndNew(reader, writer);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K key = readKey(reader, local);
final V oldValue = readValue(reader, local);
final V newValue = readValue(reader, local);
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
copies = valueReaderWithSize.getCopies(copies);
final K key = keyReaderWithSize.read(reader, copies);
final V oldValue = valueReaderWithSize.read(reader, copies);
final V newValue = valueReaderWithSize.read(reader, copies);
writer.writeBoolean(map.replace(key, oldValue, newValue));
}
} catch (Throwable e) {
@@ -1618,8 +1619,7 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
writer.writeBoolean(map.containsKey(reader));
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K k = readKey(reader, local);
final K k = keyReaderWithSize.read(reader, null);
writer.writeBoolean(map.containsKey(k));
}
} catch (Throwable e) {
@@ -1631,8 +1631,7 @@ class StatelessServerConnector<K, V> {
private Work containsValue(Bytes reader, Bytes writer, final long sizeLocation) {
// todo optimize -- eliminate
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final V v = readValue(reader, local);
final V v = valueReaderWithSize.read(reader, null);
try {
writer.writeBoolean(map.containsValue(v));
@@ -1648,9 +1647,9 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.get(reader, writer);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K k = readKey(reader, local);
writeValue(map.get(k), writer, local);
final ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
final K k = keyReaderWithSize.read(reader, copies);
valueWriterWithSize.writeNullable(writer, map.get(k), copies);
}
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
@@ -1665,9 +1664,10 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.put(reader);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K k = readKey(reader, local);
final V v = readValue(reader, local);
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
copies = valueReaderWithSize.getCopies(copies);
final K k = keyReaderWithSize.read(reader, copies);
final V v = valueReaderWithSize.read(reader, copies);
map.put(k, v);
}
@@ -1679,10 +1679,11 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.put(reader, writer);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K k = readKey(reader, local);
final V v = readValue(reader, local);
writeValue(map.put(k, v), writer, local);
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
copies = valueReaderWithSize.getCopies(copies);
final K k = keyReaderWithSize.read(reader, copies);
final V v = valueReaderWithSize.read(reader, copies);
valueWriterWithSize.writeNullable(writer, map.put(k, v), copies);
}
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
@@ -1695,8 +1696,7 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.removeKeyAsBytes(reader);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K key = readKey(reader, local);
final K key = keyReaderWithSize.read(reader, null);
map.remove(key);
}
@@ -1709,11 +1709,11 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.remove(reader, writer);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K key = readKey(reader, local);
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
final K key = keyReaderWithSize.read(reader, copies);
if (LOG.isDebugEnabled())
LOG.debug("removing entry key=" + key);
writeValue(map.remove(key), writer, local);
valueWriterWithSize.writeNullable(writer, map.remove(key), copies);
}
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
@@ -1727,8 +1727,7 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.putAll(reader);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final Map<K, V> m = readEntries(reader, local);
final Map<K, V> m = readEntries(reader);
map.putAll(m);
}
} catch (Throwable e) {
@@ -1743,8 +1742,7 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.putAll(reader);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
map.putAll(readEntries(reader, local));
map.putAll(readEntries(reader));
}
return null;
@@ -1777,32 +1775,33 @@ class StatelessServerConnector<K, V> {
}
final Iterator<V> iterator = values.iterator();
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
// this allows us to write more data than the buffer will allow
return new Work() {
@Override
public boolean doWork(Bytes writer) {
public boolean doWork(Bytes out) {
final long sizeLocation = header(out, transactionId);
final long sizeLocation = header(writer, transactionId);
ThreadLocalCopies copies = valueWriterWithSize.getCopies(null);
Object valueWriter = valueWriterWithSize.writerForLoop(copies);
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (writer.remaining() <= maxEntrySizeBytes) {
writeHeader(writer, sizeLocation, count, true);
if (out.remaining() <= maxEntrySizeBytes) {
writeHeader(out, sizeLocation, count, true);
return false;
}
count++;
writeValue(iterator.next(), writer, local);
valueWriterWithSize.writeInLoop(out, iterator.next(), valueWriter, copies);
}
writeHeader(writer, sizeLocation, count, false);
writeHeader(out, sizeLocation, count, false);
return true;
}
};
@@ -1819,33 +1818,34 @@ class StatelessServerConnector<K, V> {
return sendException(reader, writer, e);
}
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final Iterator<K> iterator = ks.iterator();
// this allows us to write more data than the buffer will allow
return new Work() {
@Override
public boolean doWork(Bytes writer) {
public boolean doWork(Bytes out) {
final long sizeLocation = header(out, transactionId);
final long sizeLocation = header(writer, transactionId);
ThreadLocalCopies copies = keyWriterWithSize.getCopies(null);
Object keyWriter = keyWriterWithSize.writerForLoop(copies);
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (writer.remaining() <= maxEntrySizeBytes) {
writeHeader(writer, sizeLocation, count, true);
if (out.remaining() <= maxEntrySizeBytes) {
writeHeader(out, sizeLocation, count, true);
return false;
}
count++;
K key = iterator.next();
writeKey(key, writer, local);
keyWriterWithSize.writeInLoop(out, key, keyWriter, copies);
}
writeHeader(writer, sizeLocation, count, false);
writeHeader(out, sizeLocation, count, false);
return true;
}
};
@@ -1864,38 +1864,41 @@ class StatelessServerConnector<K, V> {
}
final Iterator<Map.Entry<K, V>> iterator = entries.iterator();
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
// this allows us to write more data than the buffer will allow
return new Work() {
@Override
public boolean doWork(Bytes writer) {
if (writer.remaining() <= maxEntrySizeBytes)
public boolean doWork(Bytes out) {
if (out.remaining() <= maxEntrySizeBytes)
return false;
final long sizeLocation = header(writer, transactionId);
final long sizeLocation = header(out, transactionId);
ThreadLocalCopies copies = keyWriterWithSize.getCopies(null);
Object keyWriter = keyWriterWithSize.writerForLoop(copies);
copies = valueWriterWithSize.getCopies(copies);
Object valueWriter = valueWriterWithSize.writerForLoop(copies);
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (writer.remaining() <= maxEntrySizeBytes) {
if (out.remaining() <= maxEntrySizeBytes) {
writeHeader(writer, sizeLocation, count, true);
writeHeader(out, sizeLocation, count, true);
return false;
}
count++;
final Map.Entry<K, V> next = iterator.next();
writeKey(next.getKey(), writer, local);
writeValue(next.getValue(), writer, local);
keyWriterWithSize.writeInLoop(out, next.getKey(), keyWriter, copies);
valueWriterWithSize.writeInLoop(out, next.getValue(), valueWriter, copies);
}
writeHeader(writer, sizeLocation, count, false);
writeHeader(out, sizeLocation, count, false);
return true;
}
@@ -1908,10 +1911,11 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.putIfAbsent(reader, writer);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K key = readKey(reader, local);
final V v = readValue(reader, local);
writeValue(map.putIfAbsent(key, v), writer, local);
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
copies = valueReaderWithSize.getCopies(copies);
final K key = keyReaderWithSize.read(reader, copies);
final V v = valueReaderWithSize.read(reader, copies);
valueWriterWithSize.writeNullable(writer, map.putIfAbsent(key, v), copies);
}
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
@@ -1926,10 +1930,11 @@ class StatelessServerConnector<K, V> {
if (MAP_SUPPORTS_BYTES) {
map.replaceKV(reader, writer);
} else {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K k = readKey(reader, local);
final V v = readValue(reader, local);
writeValue(map.replace(k, v), writer, local);
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
copies = valueReaderWithSize.getCopies(copies);
final K k = keyReaderWithSize.read(reader, copies);
final V v = valueReaderWithSize.read(reader, copies);
valueWriterWithSize.writeNullable(writer, map.replace(k, v), copies);
}
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
@@ -1938,16 +1943,6 @@ class StatelessServerConnector<K, V> {
return null;
}
/**
* 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 writer, final ThreadLocalCopies local) {
keyValueSerializer.writeKey(key, writer, local);
}
private long reflectTransactionId(Bytes reader, Bytes writer) {
final long sizeLocation = writer.position();
writer.skip(TRANSACTION_ID_OFFSET); // isException + SIZE_OF_SIZE
@@ -1956,38 +1951,29 @@ class StatelessServerConnector<K, V> {
return sizeLocation;
}
private void writeValue(V value, final Bytes writer, ThreadLocalCopies local) {
keyValueSerializer.writeValue(value, writer, local);
}
private K readKey(Bytes reader, ThreadLocalCopies local) {
return keyValueSerializer.readKey(reader, local);
}
private V readValue(Bytes reader, ThreadLocalCopies local) {
return keyValueSerializer.readValue(reader, local);
}
private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) {
long currentPosition = out.position();
final long size = out.position() - locationOfSize;
out.writeInt(locationOfSize, (int) size); // size
out.writeBoolean(locationOfSize + SIZE_OF_SIZE, isException); // isException
out.position(currentPosition);
}
private void writeException(Bytes out, Throwable e) {
out.writeObject(e);
}
private Map<K, V> readEntries(Bytes reader, final ThreadLocalCopies local) {
private Map<K, V> readEntries(Bytes reader) {
final long numberOfEntries = reader.readStopBit();
final HashMap<K, V> result = new HashMap<K, V>();
ThreadLocalCopies copies = keyReaderWithSize.getCopies(null);
BytesReader<K> keyReader = keyReaderWithSize.readerForLoop(copies);
copies = valueReaderWithSize.getCopies(copies);
BytesReader<V> valueReader = valueReaderWithSize.readerForLoop(copies);
for (long i = 0; i < numberOfEntries; i++) {
K key = readKey(reader, local);
V value = readValue(reader, local);
K key = keyReaderWithSize.readInLoop(reader, keyReader);
V value = valueReaderWithSize.readInLoop(reader, valueReader);
result.put(key, value);
}
return result;
@@ -2040,106 +2026,4 @@ class StatelessServerConnector<K, V> {
}
class KeyValueSerializer<K, V> {
final Serializer<V> valueSerializer;
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 reader) {
if (nullCheck(reader)) return null;
return valueSerializer.readMarshallable(reader, null);
}
V readValue(Bytes reader, ThreadLocalCopies local) {
if (nullCheck(reader)) return null;
return valueSerializer.readMarshallable(reader, local);
}
<O> O readObject(Bytes reader) {
if (nullCheck(reader)) return null;
return (O) reader.readObject();
}
private boolean nullCheck(Bytes in) {
return in.readBoolean();
}
K readKey(Bytes reader) {
if (nullCheck(reader)) return null;
return keySerializer.readMarshallable(reader, null);
}
K readKey(Bytes reader, ThreadLocalCopies local) {
if (nullCheck(reader)) return null;
return keySerializer.readMarshallable(reader, local);
}
/**
* write the keysize and the key to the the {@code target} buffer
*
* @param key the key of the map
*/
void writeKey(K key, Bytes writer) {
writer.writeBoolean(key == null);
if (key != null)
keySerializer.writeMarshallable(key, writer, null);
}
void writeValue(V value, Bytes writer) {
assert writer.limit() == writer.capacity();
writer.writeBoolean(value == null);
if (value != null) {
valueSerializer.writeMarshallable(value, writer, null);
}
}
void writeKey(K key, Bytes writer, ThreadLocalCopies threadLocal) {
writer.writeBoolean(key == null);
if (key != null)
keySerializer.writeMarshallable(key, writer, threadLocal);
}
void writeValue(V value, Bytes writer, ThreadLocalCopies threadLocal) {
assert writer.limit() == writer.capacity();
writer.writeBoolean(value == null);
if (value != null)
valueSerializer.writeMarshallable(value, writer, threadLocal);
}
void writeObject(Object value, Bytes writer) {
assert writer.limit() == writer.capacity();
writer.writeBoolean(value == null);
if (value != null) {
writer.writeObject(value);
}
}
public ThreadLocalCopies threadLocalCopies() {
return valueSerializer.threadLocalCopies();
}
}
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+99 -73
View File
@@ -118,12 +118,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
transient long headerSize;
private transient Set<Map.Entry<K, V>> entrySet;
private transient SerializationBuilder<K> keyBuilder;
private transient SerializationBuilder<V> valueBuilder;
public VanillaChronicleMap(AbstractChronicleMapBuilder<K, V, ?> builder) throws IOException {
keyBuilder = builder.keyBuilder;
SerializationBuilder<K> keyBuilder = builder.keyBuilder;
kClass = keyBuilder.eClass;
keySizeMarshaller = keyBuilder.sizeMarshaller();
originalKeyReader = keyBuilder.reader();
@@ -133,7 +130,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
originalMetaKeyInterop = (MKI) keyBuilder.metaInterop();
metaKeyInteropProvider = (MetaProvider<K, KI, MKI>) keyBuilder.metaInteropProvider();
valueBuilder = builder.valueBuilder;
SerializationBuilder<V> valueBuilder = builder.valueBuilder;
vClass = valueBuilder.eClass;
isNativeValueClass = vClass.getName().endsWith("$$Native");
valueSizeMarshaller = valueBuilder.sizeMarshaller();
@@ -348,6 +345,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
ReadValueToOutputBytes readValueToOutputBytes = segmentState.readValueToOutputBytes;
readValueToOutputBytes.reuse(valueSizeMarshaller, output);
long keySize = keySizeMarshaller.readSize(entry);
entry.limit(entry.position() + keySize);
GetRemoteBytesValueInterops getRemoteBytesValueInterops =
segmentState.getRemoteBytesValueInterops;
MultiStoreBytes value = getRemoteBytesValueInterops.getValueBytes(
@@ -371,6 +369,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
ThreadLocalCopies copies = SegmentState.getCopies(null);
SegmentState segmentState = SegmentState.get(copies);
long keySize = keySizeMarshaller.readSize(entry);
entry.limit(entry.position() + keySize);
GetRemoteBytesValueInterops getRemoteBytesValueInterops =
segmentState.getRemoteBytesValueInterops;
MultiStoreBytes value = getRemoteBytesValueInterops.getValueBytes(
@@ -398,7 +397,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
public Bytes readValue(@NotNull ThreadLocalCopies copies, Bytes entry, Bytes usingBytes,
long valueSize) {
usingBytes = bytes(valueSize);
usingBytes.writeBoolean(false);
valueSizeMarshaller.writeSize(usingBytes, valueSize);
usingBytes.write(entry, entry.position(), valueSize);
return usingBytes;
@@ -418,6 +416,13 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return outputBytes;
}
@Override
public Bytes readValue(@NotNull ThreadLocalCopies copies, Bytes entry, Bytes usingBytes,
long valueSize) {
outputBytes.writeBoolean(false);
return super.readValue(copies, entry, usingBytes, valueSize);
}
@Override
public Bytes readNull() {
outputBytes.writeBoolean(true);
@@ -518,7 +523,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
lookupUsing(copies, segmentState,
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, key, keySizeMarshaller.readSize(key),
keyBytesToInstance, readValueToOutputBytes, null, outputValueBytesToInstance, false);
keyBytesToInstance, readValueToOutputBytes, null, outputValueBytesToInstance,
false);
}
private V lookupUsing(K key, V usingValue, boolean create) {
@@ -667,8 +673,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
final XStream xstream = new XStream(new JettisonMappedXmlDriver());
xstream.setMode(XStream.NO_REFERENCES);
xstream.alias("chronicle-entries", net.openhft.chronicle.map.VanillaChronicleMap.EntrySet.class);
xstream.alias("chronicle-entries", VanillaChronicleMap.EntrySet.class);
// ideally we will use java serialization if we can
lazyXStreamConverter().registerConverter(xstream);
@@ -682,8 +687,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
public synchronized void putAll(File fromFile) throws IOException {
final XStream xstream = new XStream(new JettisonMappedXmlDriver());
xstream.setMode(XStream.NO_REFERENCES);
xstream.alias("chronicle-entries", net.openhft.chronicle.map.VanillaChronicleMap.EntrySet
.class);
xstream.alias("chronicle-entries", VanillaChronicleMap.EntrySet.class);
// ideally we will use java serialization if we can
lazyXStreamConverter().registerConverter(xstream);
@@ -700,35 +704,18 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
/**
* @return a lazily created XStreamConverter
*/
private synchronized XStreamConverter lazyXStreamConverter() {
if (xStreamConverter != null)
return xStreamConverter;
final KeyValueSerializer<K, V> serializers = new KeyValueSerializer<K, V>(keyBuilder,
valueBuilder);
final ThreadLocalCopies threadLocalCopies = serializers.threadLocalCopies();
final ByteBufferBytes buffer = new ByteBufferBytes(allocateDirect((int) entrySize)
.order(ByteOrder.nativeOrder()));
xStreamConverter = new XStreamConverter(threadLocalCopies, buffer, serializers);
return xStreamConverter;
private XStreamConverter lazyXStreamConverter() {
return xStreamConverter != null ? xStreamConverter :
(xStreamConverter = new XStreamConverter());
}
private class XStreamConverter {
private final ThreadLocalCopies threadLocalCopies;
private Bytes buffer;
private final KeyValueSerializer<K, V> keyValueSerializer;
public XStreamConverter(ThreadLocalCopies threadLocalCopies,
ByteBufferBytes buffer,
final KeyValueSerializer<K, V> keyValueSerializer) {
this.threadLocalCopies = threadLocalCopies;
this.buffer = buffer;
this.keyValueSerializer = keyValueSerializer;
public XStreamConverter() {
this.buffer = new ByteBufferBytes(allocateDirect((int) entrySize)
.order(ByteOrder.nativeOrder()));
}
private void registerConverter(XStream xstream) {
@@ -827,9 +814,16 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
break;
case "chronicle-key":
return keyValueSerializer.readKey(buffer, threadLocalCopies);
long keySize = keySizeMarshaller.readSize(buffer);
ThreadLocalCopies copies = keyReaderProvider.getCopies(null);
BytesReader<K> keyReader = keyReaderProvider.get(copies, originalKeyReader);
return keyReader.read(buffer, keySize);
case "chronicle-value":
return keyValueSerializer.readValue(buffer, threadLocalCopies);
long valueSize = valueSizeMarshaller.readSize(buffer);
copies = valueReaderProvider.getCopies(null);
BytesReader<V> valueReader =
valueReaderProvider.get(copies, originalValueReader);
return valueReader.read(buffer, valueSize);
}
return null;
@@ -951,7 +945,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
};
private transient InstanceOrBytesToInstance<Bytes, V> outputValueBytesToInstance =
transient InstanceOrBytesToInstance<Bytes, V> outputValueBytesToInstance =
new InstanceOrBytesToInstance<Bytes, V>() {
@Override
public V toInstance(@NotNull ThreadLocalCopies copies, Bytes outputBytes, long size) {
@@ -1127,13 +1121,33 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key);
long keySize = metaKeyInterop.size(keyInterop, key);
return remove(copies, null, metaKeyInterop, keyInterop, key, keySize,
keyIdentity(), this, expectedValue, valueIdentity(), removeReturnsNull);
keyIdentity(), this, expectedValue, valueIdentity(), this, removeReturnsNull);
}
final void removeKeyAsBytes(Bytes key) {
ThreadLocalCopies copies = SegmentState.getCopies(null);
SegmentState segmentState = SegmentState.get(copies);
ReadValueToBytes readValueToLazyBytes = segmentState.readValueToLazyBytes;
readValueToLazyBytes.valueSizeMarshaller(valueSizeMarshaller);
long keySize = keySizeMarshaller.readSize(key);
key.limit(key.position() + keySize);
remove(null, null, DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, key, keySizeMarshaller.readSize(key),
keyBytesToInstance, null, null, null, true);
BytesBytesInterop.INSTANCE, key, keySize,
keyBytesToInstance, null, null,
outputValueBytesToInstance, readValueToLazyBytes, true);
}
final void remove(Bytes key, Bytes output) {
ThreadLocalCopies copies = SegmentState.getCopies(null);
SegmentState segmentState = SegmentState.get(copies);
ReadValueToOutputBytes readValueToOutputBytes = segmentState.readValueToOutputBytes;
readValueToOutputBytes.reuse(valueSizeMarshaller, output);
long keySize = keySizeMarshaller.readSize(key);
key.limit(key.position() + keySize);
remove(copies, segmentState,
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, key, keySize, keyBytesToInstance, null, null,
outputValueBytesToInstance, readValueToOutputBytes, false);
}
// for boolean remove(Object key, Object value);
@@ -1141,40 +1155,46 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
ThreadLocalCopies copies = SegmentState.getCopies(null);
SegmentState segmentState = SegmentState.get(copies);
long keySize = keySizeMarshaller.readSize(entry);
entry.limit(entry.position() + keySize);
GetRemoteBytesValueInterops getRemoveBytesValueInterops =
segmentState.getRemoteBytesValueInterops;
MultiStoreBytes value = getRemoveBytesValueInterops.getValueBytes(
entry, entry.position() + keySize);
getRemoveBytesValueInterops.valueSizeMarshaller(valueSizeMarshaller);
ReadValueToBytes readValueToLazyBytes = segmentState.readValueToLazyBytes;
readValueToLazyBytes.valueSizeMarshaller(valueSizeMarshaller);
return (Boolean) remove(copies, segmentState,
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, entry, keySize, keyBytesToInstance,
getRemoveBytesValueInterops, value, valueBytesToInstance, false);
getRemoveBytesValueInterops, value, valueBytesToInstance,
readValueToLazyBytes, false);
}
private <KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
VB, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
Object remove(ThreadLocalCopies copies, SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB expectedValue,
InstanceOrBytesToInstance<? super VB, V> toValue,
boolean resultUnused) {
InstanceOrBytesToInstance<RV, V> toValue,
ReadValue<RV> readValue, boolean resultUnused) {
long hash = metaKeyInterop.hash(keyInterop, key);
int segmentNum = getSegment(hash);
long segmentHash = segmentHash(hash);
Segment segment = segments[segmentNum];
return segment.remove(copies, segmentState, metaKeyInterop, keyInterop, key, keySize,
toKey, getValueInterops, expectedValue,
toValue, segmentHash, resultUnused);
toValue, segmentHash, readValue, resultUnused);
}
/**
* @throws NullPointerException if any of the arguments are null
*/
@Override
public final boolean replace(@net.openhft.lang.model.constraints.NotNull K key, @net.openhft.lang.model.constraints.NotNull V oldValue, @net.openhft.lang.model.constraints.NotNull V newValue) {
public final boolean replace(@net.openhft.lang.model.constraints.NotNull K key,
@net.openhft.lang.model.constraints.NotNull V oldValue,
@net.openhft.lang.model.constraints.NotNull V newValue) {
checkValue(oldValue);
return (Boolean) replaceIfValueIs(key, oldValue, newValue);
}
@@ -1184,7 +1204,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
*/
@Override
@SuppressWarnings("unchecked")
public final V replace(@net.openhft.lang.model.constraints.NotNull final K key, @net.openhft.lang.model.constraints.NotNull final V value) {
public final V replace(@net.openhft.lang.model.constraints.NotNull final K key,
@net.openhft.lang.model.constraints.NotNull final V value) {
return (V) replaceIfValueIs(key, null, value);
}
@@ -1236,6 +1257,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
readValueToOutputBytes.reuse(valueSizeMarshaller, output);
long keySize = keySizeMarshaller.readSize(keyAndNewValue);
keyAndNewValue.limit(keyAndNewValue.position() + keySize);
GetRemoteBytesValueInterops getRemoteBytesValueInterops =
segmentState.getRemoteBytesValueInterops;
MultiStoreBytes value = getRemoteBytesValueInterops.getValueBytes(
@@ -1269,6 +1291,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
getNewValueInterops.valueSizeMarshaller(valueSizeMarshaller);
entryAndNewValue.position(keyPosition);
entryAndNewValue.limit(keyPosition + keySize);
output.writeBoolean((Boolean) replace(copies, segmentState,
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
@@ -1337,7 +1360,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
long nextEntryPosition = entries.position() + valueSize;
entries.position(entryPosition);
put(entries);
entries.clear(); // because used as key, altering position and limit
entryPosition = nextEntryPosition;
entries.position(entryPosition);
}
}
@@ -2045,19 +2070,19 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
VB, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
Object remove(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB expectedValue,
InstanceOrBytesToInstance<? super VB, V> toValue,
long hash2, boolean resultUnused) {
InstanceOrBytesToInstance<RV, V> toValue,
long hash2, ReadValue<RV> readValue, boolean resultUnused) {
writeLock(null);
try {
return removeWithoutLock(copies, segmentState,
metaKeyInterop, keyInterop, key, keySize, toKey,
getValueInterops, expectedValue, toValue,
hash2, resultUnused);
hash2, readValue, resultUnused);
} finally {
if (segmentState != null)
segmentState.close();
@@ -2072,14 +2097,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
* returned - if resultUnused is true, null is always returned
*/
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
VB, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
Object removeWithoutLock(
@Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB expectedValue,
InstanceOrBytesToInstance<? super VB, V> toValue,
long hash2, boolean resultUnused) {
InstanceOrBytesToInstance<RV, V> toValue,
long hash2, ReadValue<RV> readValue, boolean resultUnused) {
segmentStateNotNullImpliesCopiesNotNull(copies, segmentState);
SegmentState localSegmentState = segmentState;
try {
@@ -2113,35 +2138,36 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
if (!metaValueInterop.startsWith(valueInterop, entry, expectedValue))
return Boolean.FALSE;
}
return removeEntry(copies, key, keySize, toKey, expectedValue, toValue,
resultUnused, hashLookup, entry, pos, valueSizePos, valueSize,
false, true, expectedValue != null);
return removeEntry(copies, key, keySize, toKey, toValue,
readValue, resultUnused, hashLookup, entry, pos, valueSizePos,
valueSize, false, true, expectedValue != null);
}
// key is not found
return expectedValue == null ? null : Boolean.FALSE;
if (expectedValue == null) {
return resultUnused ? null : readValue.readNull();
} else {
return Boolean.FALSE;
}
} finally {
if (segmentState == null && localSegmentState != null)
localSegmentState.close();
}
}
final <KB, VB> Object removeEntry(
final <KB, RV, VB extends RV> Object removeEntry(
ThreadLocalCopies copies,
KB key, long keySize, InstanceOrBytesToInstance<KB, K> toKey,
VB expectedValue, InstanceOrBytesToInstance<? super VB, V> toValue,
boolean resultUnused, MultiMap hashLookup, MultiStoreBytes entry, long pos,
InstanceOrBytesToInstance<RV, V> toValue,
ReadValue<RV> readValue, boolean resultUnused,
MultiMap hashLookup, MultiStoreBytes entry, long pos,
long valueSizePos, long valueSize, boolean remote, boolean removeFromMultiMap,
boolean booleanResult) {
// get the removed value, if needed
V removedValue = null;
RV removedValue = null;
if ((!booleanResult && !resultUnused) || eventListener != null) {
if (expectedValue != null)
removedValue = toValue.toInstance(copies, expectedValue, valueSize);
if (removedValue == null) {
// todo reuse some value
removedValue = readValue(copies, entry, null, valueSize);
entry.position(entry.position() - valueSize);
}
// todo reuse some value
removedValue = readValue.readValue(copies, entry, null, valueSize);
entry.position(entry.position() - valueSize);
}
// update segment state
@@ -2160,8 +2186,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
if (bytesEventListener != null)
bytesEventListener.onRemove(entry, 0L, metaDataBytes, valueSizePos);
if (eventListener != null) {
V removedValueForEventListener = removedValue != null ? removedValue :
readValue(copies, entry, null);
V removedValueForEventListener =
toValue.toInstance(copies, removedValue, valueSize);
eventListener.onRemove(toKey.toInstance(copies, key, keySize),
removedValueForEventListener);
}
@@ -2757,7 +2783,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
public void removeEntry() {
segment.removeWithoutLock(copies, segmentState,
metaKeyInterop, keyInterop, key(), keySize, map.keyIdentity(),
map, null, map.valueIdentity(), segmentHash, true);
map, null, map.valueIdentity(), segmentHash, map, true);
}
}
@@ -2793,7 +2819,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
putOnClose = false;
segment.removeWithoutLock(copies, segmentState,
metaKeyInterop, keyInterop, key(), keySize, map.keyIdentity(),
map, null, map.valueIdentity(), segmentHash, true);
map, null, map.valueIdentity(), segmentHash, map, true);
}
}
}
src/main/java/net/openhft/chronicle/map/WriterWithSize.java
+89
View File
@@ -0,0 +1,89 @@
/*
* 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.map;
import net.openhft.chronicle.hash.serialization.SizeMarshaller;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
import net.openhft.chronicle.hash.serialization.internal.MetaProvider;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.StatefulCopyable;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.Nullable;
final class WriterWithSize<T> {
private final SizeMarshaller sizeMarshaller;
private final Object originalWriter;
private final Provider writerProvider;
private final MetaBytesInterop originalMetaWriter;
private final MetaProvider metaWriterProvider;
WriterWithSize(SerializationBuilder<T> serializationBuilder) {
sizeMarshaller = serializationBuilder.sizeMarshaller();
originalWriter = serializationBuilder.interop();
writerProvider = Provider.of((Class) originalWriter.getClass());
originalMetaWriter = serializationBuilder.metaInterop();
metaWriterProvider = serializationBuilder.metaInteropProvider();
}
ThreadLocalCopies getCopies(ThreadLocalCopies copies) {
copies = writerProvider.getCopies(copies);
return metaWriterProvider.getCopies(copies);
}
public ThreadLocalCopies write(Bytes out, T t, @Nullable ThreadLocalCopies copies) {
copies = writerProvider.getCopies(copies);
Object writer = writerProvider.get(copies, originalWriter);
copies = metaWriterProvider.getCopies(copies);
MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t);
sizeMarshaller.writeSize(out, metaWriter.size(writer, t));
metaWriter.write(writer, out, t);
return copies;
}
public ThreadLocalCopies writeNullable(Bytes out, T t, @Nullable ThreadLocalCopies copies) {
out.writeBoolean(t == null);
if (t == null)
return copies;
return write(out, t, copies);
}
public Object writerForLoop(@Nullable ThreadLocalCopies copies) {
copies = writerProvider.getCopies(copies);
return writerProvider.get(copies, originalWriter);
}
public ThreadLocalCopies writeInLoop(Bytes out, T t, Object writer,
@Nullable ThreadLocalCopies copies) {
copies = metaWriterProvider.getCopies(copies);
MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t);
sizeMarshaller.writeSize(out, metaWriter.size(writer, t));
metaWriter.write(writer, out, t);
return copies;
}
public ThreadLocalCopies writeNullableInLoop(Bytes out, T t, Object writer,
@Nullable ThreadLocalCopies copies) {
out.writeBoolean(t == null);
if (t == null)
return copies;
return writeInLoop(out, t, writer, copies);
}
}
src/test/java/net/openhft/chronicle/map/SerializerTest.java
+9 -9
View File
@@ -73,23 +73,23 @@ public class SerializerTest {
ByteBufferBytes in = out.slice();
ChronicleMapBuilder cBuilder = ChronicleMapBuilder.of(Integer
.class, valueClass);
ChronicleMapBuilder cBuilder = ChronicleMapBuilder.of(Integer.class, valueClass);
OnHeapUpdatableChronicleMapBuilder builder = (OnHeapUpdatableChronicleMapBuilder) cBuilder.delegate;
builder.preMapConstruction(false);
Serializer v = new Serializer(builder.valueBuilder);
WriterWithSize valueWriterWithSize = new WriterWithSize(builder.valueBuilder);
ReaderWithSize valueReaderWithSize = new ReaderWithSize(builder.valueBuilder);
v.writeMarshallable(value, out, null);
valueWriterWithSize.write(out, value, null);
long position = out.position();
in.limit(position);
Object actual = v.readMarshallable(in, null);
Object actual = valueReaderWithSize.read(in, null);
Assert.assertEquals(actual, value);
}
@@ -107,15 +107,15 @@ public class SerializerTest {
builder.preMapConstruction(false);
{
Serializer v = new Serializer(builder.keyBuilder);
WriterWithSize keyWriterWithSize = new WriterWithSize(builder.keyBuilder);
ReaderWithSize keyReaderWithSize = new ReaderWithSize(builder.keyBuilder);
v.writeMarshallable(key, out, null);
keyWriterWithSize.write(out, key, null);
long position = out.position();
in.limit(position);
Object actual = v.readMarshallable(in, null);
Object actual = keyReaderWithSize.read(in, null);
Assert.assertEquals(actual, key);
}
src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
+34 -32
View File
@@ -27,6 +27,8 @@ import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ExecutionException;
import static org.junit.Assert.assertEquals;
/**
* @author Rob Austin.
@@ -65,7 +67,7 @@ public class StatelessClientTest {
String actual = statelessMap.mapForKey(10, ToString.INSTANCE);
Assert.assertEquals("Hello World", actual);
assertEquals("Hello World", actual);
}
}
}
@@ -88,7 +90,7 @@ public class StatelessClientTest {
String actual = statelessMap.mapForKey(11, ToString.INSTANCE);
Assert.assertEquals(null, actual);
assertEquals(null, actual);
}
}
}
@@ -118,9 +120,9 @@ public class StatelessClientTest {
Set<Map.Entry<Integer, CharSequence>> entries = statelessMap.entrySet();
Map.Entry<Integer, CharSequence> next = entries.iterator().next();
Assert.assertEquals("some value=" + next.getKey(), next.getValue());
assertEquals("some value=" + next.getKey(), next.getValue());
Assert.assertEquals(entries.size(), SIZE);
assertEquals(SIZE, entries.size());
}
}
@@ -151,7 +153,7 @@ public class StatelessClientTest {
Collection<CharSequence> values = statelessMap.values();
Assert.assertEquals(values.size(), SIZE);
assertEquals(values.size(), SIZE);
}
}
@@ -182,7 +184,7 @@ public class StatelessClientTest {
final Set<Integer> keys = statelessMap.keySet();
Assert.assertEquals(keys.size(), payload.size());
assertEquals(keys.size(), payload.size());
}
@@ -203,8 +205,8 @@ public class StatelessClientTest {
.statelessClient(new InetSocketAddress("localhost", 8056)).create()) {
Assert.assertEquals("EXAMPLE-10", statelessMap.get(10));
Assert.assertEquals(1, statelessMap.size());
assertEquals("EXAMPLE-10", statelessMap.get(10));
assertEquals(1, statelessMap.size());
}
}
}
@@ -223,8 +225,8 @@ public class StatelessClientTest {
serverMap.put(10, "EXAMPLE-10");
Assert.assertEquals("EXAMPLE-10", statelessMap.get(10));
Assert.assertEquals(1, statelessMap.size());
assertEquals("EXAMPLE-10", statelessMap.get(10));
assertEquals(1, statelessMap.size());
}
@@ -249,8 +251,8 @@ public class StatelessClientTest {
serverMap.put("hello", Collections.singletonMap("hello", "world"));
Assert.assertEquals(Collections.singletonMap("hello", "world"), statelessMap.get("hello"));
Assert.assertEquals(1, statelessMap.size());
assertEquals(Collections.singletonMap("hello", "world"), statelessMap.get("hello"));
assertEquals(1, statelessMap.size());
}
@@ -281,8 +283,8 @@ public class StatelessClientTest {
statelessMap.put("hello", data);
Assert.assertEquals(data, serverMap.get("hello"));
Assert.assertEquals(1, statelessMap.size());
assertEquals(data, serverMap.get("hello"));
assertEquals(1, statelessMap.size());
}
@@ -312,8 +314,8 @@ public class StatelessClientTest {
int value = SIZE - 10;
Assert.assertEquals("some value=" + value, statelessMap.get(value));
Assert.assertEquals(SIZE, statelessMap.size());
assertEquals("some value=" + value, statelessMap.get(value));
assertEquals(SIZE, statelessMap.size());
}
}
@@ -343,8 +345,8 @@ public class StatelessClientTest {
int value = SIZE - 10;
Assert.assertEquals("some value=" + value, statelessMap.get(value));
Assert.assertEquals(SIZE, statelessMap.size());
assertEquals("some value=" + value, statelessMap.get(value));
assertEquals(SIZE, statelessMap.size());
}
}
@@ -365,8 +367,8 @@ public class StatelessClientTest {
statelessMap.put(1, "some value");
Assert.assertEquals("some value", statelessMap.get(1));
Assert.assertEquals(1, statelessMap.size());
assertEquals("some value", statelessMap.get(1));
assertEquals(1, statelessMap.size());
}
@@ -388,13 +390,13 @@ public class StatelessClientTest {
statelessMap.put(1, "some value");
Assert.assertEquals("some value", statelessMap.get(1));
Assert.assertEquals(1, statelessMap.size());
assertEquals("some value", statelessMap.get(1));
assertEquals(1, statelessMap.size());
statelessMap.remove(1);
Assert.assertEquals(null, statelessMap.get(1));
Assert.assertEquals(0, statelessMap.size());
assertEquals(null, statelessMap.get(1));
assertEquals(0, statelessMap.size());
}
}
@@ -415,13 +417,13 @@ public class StatelessClientTest {
statelessMap.put(1, "some value");
Assert.assertEquals("some value", statelessMap.getLater(1).get());
Assert.assertEquals(1, statelessMap.size());
assertEquals("some value", statelessMap.getLater(1).get());
assertEquals(1, statelessMap.size());
statelessMap.remove(1);
Assert.assertEquals(null, statelessMap.getLater(1).get());
Assert.assertEquals(0, statelessMap.size());
assertEquals(null, statelessMap.getLater(1).get());
assertEquals(0, statelessMap.size());
}
}
@@ -441,8 +443,8 @@ public class StatelessClientTest {
.statelessClient(new InetSocketAddress("localhost", 8056)).create()) {
CharSequence oldValue = statelessMap.putLater(1, "some value").get();
Assert.assertEquals("some value", statelessMap.get(1));
Assert.assertEquals(1, statelessMap.size());
assertEquals("some value", statelessMap.get(1));
assertEquals(1, statelessMap.size());
}
}
@@ -462,8 +464,8 @@ public class StatelessClientTest {
.statelessClient(new InetSocketAddress("localhost", 8056)).create()) {
statelessMap.put(1, "some value");
statelessMap.removeLater(1).get();
Assert.assertEquals(null, statelessMap.get(1));
Assert.assertEquals(0, statelessMap.size());
assertEquals(null, statelessMap.get(1));
assertEquals(0, statelessMap.size());
}
}