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

Add tests for larger entries

Browse Source
This commit is contained in:
Peter Lawrey
2014-12-08 16:17:27 +00:00
parent cb9245361c
commit 13e9610a73
11 changed files with 201 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/serialization/internal/BytesReaders.java
+8 -7
View File
@@ -34,32 +34,33 @@ public final class BytesReaders {
* from {@link net.openhft.chronicle.hash.serialization} package.
*
* @param marshaller the actual reading implementation
* @param <E> type of the objects marshalled
* @param <E> type of the objects marshalled
* @return a {@code BytesReader} wrapping the given {@code BytesMarshaller}
*/
public static <E> BytesReader<E> fromBytesMarshaller(BytesMarshaller<E> marshaller) {
public static <E> BytesReader<E> fromBytesMarshaller(BytesMarshaller<? super E> marshaller) {
return new SimpleBytesReader<E>(marshaller);
}
private static class SimpleBytesReader<E> implements BytesReader<E> {
private static final long serialVersionUID = 0L;
private final BytesMarshaller<E> marshaller;
private final BytesMarshaller<? super E> marshaller;
public SimpleBytesReader(BytesMarshaller<E> marshaller) {
public SimpleBytesReader(BytesMarshaller<? super E> marshaller) {
this.marshaller = marshaller;
}
@Override
public E read(Bytes bytes, long size) {
return marshaller.read(bytes);
return (E) marshaller.read(bytes);
}
@Override
public E read(Bytes bytes, long size, E toReuse) {
return marshaller.read(bytes, toReuse);
return (E) marshaller.read(bytes, toReuse);
}
}
private BytesReaders() {}
private BytesReaders() {
}
}
src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java
+1 -2
View File
@@ -39,7 +39,6 @@ import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import net.openhft.lang.values.LongValue;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -985,7 +984,7 @@ abstract class AbstractChronicleMapBuilder<K, V,
private int figureBufferAllocationFactor() {
// if expected map size is about 1000, seems rather wasteful to allocate
// key and value serialization buffers each x64 of expected entry size..
return (int) Math.min(Math.max(2L, entries() >> 10),
return (int) Math.min(Math.max(4L, entries() >> 10),
VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR);
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+1 -1
View File
@@ -271,7 +271,7 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
*/
@Override
public ChronicleMapBuilder<K, V> valueMarshaller(
@NotNull BytesMarshaller<V> valueMarshaller) {
@NotNull BytesMarshaller<? super V> valueMarshaller) {
delegate.valueMarshaller(valueMarshaller);
return this;
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java
+1 -1
View File
@@ -199,7 +199,7 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
* @see #objectSerializer(ObjectSerializer)
*/
ChronicleMapBuilderI<K, V> valueMarshaller(
@NotNull BytesMarshaller<V> valueMarshaller);
@NotNull BytesMarshaller<? super V> valueMarshaller);
/**
* Configures the marshallers, used to serialize/deserialize values to/from off-heap memory in
src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java
+1 -1
View File
@@ -148,7 +148,7 @@ final class OffHeapUpdatableChronicleMapBuilder<K, V>
return super.prepareValueBytesOnAcquire(prepareValueBytes);
}
public ChronicleMapBuilderI<K, V> valueMarshaller(@NotNull BytesMarshaller<V> valueMarshaller) {
public ChronicleMapBuilderI<K, V> valueMarshaller(@NotNull BytesMarshaller<? super V> valueMarshaller) {
throw new UnsupportedOperationException("not supported for this combination of key/value type");
}
src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java
+1 -1
View File
@@ -210,7 +210,7 @@ final class OnHeapUpdatableChronicleMapBuilder<K, V>
* @see #objectSerializer(ObjectSerializer)
*/
public OnHeapUpdatableChronicleMapBuilder<K, V> valueMarshaller(
@NotNull BytesMarshaller<V> valueMarshaller) {
@NotNull BytesMarshaller<? super V> valueMarshaller) {
valueBuilder.marshaller(valueMarshaller);
return this;
}
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+7 -7
View File
@@ -231,7 +231,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
@Override
V put0(@NotNull K key, V value, boolean replaceIfPresent) {
V put1(@NotNull K key, V value, boolean replaceIfPresent) {
return put0(key, value, replaceIfPresent, localIdentifier, currentTime());
}
@@ -289,12 +289,12 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
@Override
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<RV, ? super VBI>>
RV put(ThreadLocalCopies copies, SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB value,
InstanceOrBytesToInstance<? super VB, V> toValue,
boolean replaceIfPresent, ReadValue<RV> readValue, boolean resultUnused) {
RV put2(ThreadLocalCopies copies, SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB value,
InstanceOrBytesToInstance<? super VB, V> toValue,
boolean replaceIfPresent, ReadValue<RV> readValue, boolean resultUnused) {
return put(copies, segmentState, metaKeyInterop, keyInterop, key, keySize, toKey,
getValueInterops, value, toValue, replaceIfPresent, readValue, resultUnused,
localIdentifier, currentTime());
src/main/java/net/openhft/chronicle/map/SerializationBuilder.java
+9 -6
View File
@@ -18,9 +18,6 @@
package net.openhft.chronicle.map;
import net.openhft.lang.model.*;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.hash.serialization.internal.*;
import net.openhft.lang.io.Bytes;
@@ -29,6 +26,12 @@ import net.openhft.lang.io.serialization.BytesMarshaller;
import net.openhft.lang.io.serialization.ObjectFactory;
import net.openhft.lang.io.serialization.ObjectSerializer;
import net.openhft.lang.io.serialization.impl.*;
import net.openhft.lang.model.Byteable;
import net.openhft.lang.model.DataValueGenerator;
import net.openhft.lang.model.DataValueModel;
import net.openhft.lang.model.DataValueModels;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import net.openhft.lang.values.LongValue;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -44,11 +47,11 @@ import java.util.Arrays;
import java.util.List;
import static net.openhft.chronicle.hash.serialization.SizeMarshallers.constant;
import static net.openhft.chronicle.map.Objects.hash;
import static net.openhft.chronicle.hash.serialization.SizeMarshallers.stopBit;
import static net.openhft.chronicle.map.Objects.hash;
import static net.openhft.chronicle.map.SerializationBuilder.Role.KEY;
final class SerializationBuilder<E> implements Cloneable, Serializable{
final class SerializationBuilder<E> implements Cloneable, Serializable {
private static boolean concreteClass(Class c) {
return !c.isInterface() && !Modifier.isAbstract(c.getModifiers());
@@ -194,7 +197,7 @@ final class SerializationBuilder<E> implements Cloneable, Serializable{
.<E, BytesWriter<E>>forBytesWriter(role));
}
public SerializationBuilder<E> marshaller(BytesMarshaller<E> marshaller) {
public SerializationBuilder<E> marshaller(BytesMarshaller<? super E> marshaller) {
clearDefaults();
return copyingInterop(CopyingInterop.FROM_MARSHALLER)
.reader(BytesReaders.fromBytesMarshaller(marshaller))
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+21 -21
View File
@@ -354,7 +354,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
void putBytes(ThreadLocalCopies copies, SegmentState segmentState, Bytes key, long keySize,
GetRemoteBytesValueInterops getRemoteBytesValueInterops, MultiStoreBytes value,
boolean replaceIfPresent, ReadValue<Bytes> readValue) {
put(copies, segmentState,
put2(copies, segmentState,
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, key, keySize, keyBytesToInstance,
getRemoteBytesValueInterops, value, valueBytesToInstance,
@@ -373,7 +373,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
getRemoteBytesValueInterops.valueSizeMarshaller(valueSizeMarshaller);
ReadValueToBytes readValueToLazyBytes = segmentState.readValueToLazyBytes;
readValueToLazyBytes.valueSizeMarshaller(valueSizeMarshaller);
put(copies, segmentState,
put2(copies, segmentState,
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, entry, keySize, keyBytesToInstance,
getRemoteBytesValueInterops, value, valueBytesToInstance,
@@ -453,15 +453,15 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
@Override
public final V put(K key, V value) {
return put0(key, value, true);
return put1(key, value, true);
}
@Override
public final V putIfAbsent(@net.openhft.lang.model.constraints.NotNull K key, V value) {
return put0(key, value, false);
return put1(key, value, false);
}
V put0(@net.openhft.lang.model.constraints.NotNull K key, V value, boolean replaceIfPresent) {
V put1(@net.openhft.lang.model.constraints.NotNull K key, V value, boolean replaceIfPresent) {
checkKey(key);
checkValue(value);
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
@@ -470,22 +470,22 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
MKI metaKeyInterop =
metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key);
long keySize = metaKeyInterop.size(keyInterop, key);
return put(copies, null, metaKeyInterop, keyInterop, key, keySize, keyIdentity(),
return put2(copies, null, metaKeyInterop, keyInterop, key, keySize, keyIdentity(),
this, value, valueIdentity(), replaceIfPresent, this, putReturnsNull);
}
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<RV, ? super VBI>>
RV put(ThreadLocalCopies copies, SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB value,
InstanceOrBytesToInstance<? super VB, V> toValue,
boolean replaceIfPresent, ReadValue<RV> readValue, boolean resultUnused) {
RV put2(ThreadLocalCopies copies, SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance<KB, K> toKey,
GetValueInterops<VB, VBI, MVBI> getValueInterops, VB value,
InstanceOrBytesToInstance<? super VB, V> toValue,
boolean replaceIfPresent, ReadValue<RV> readValue, boolean resultUnused) {
long hash = metaKeyInterop.hash(keyInterop, key);
int segmentNum = getSegment(hash);
long segmentHash = segmentHash(hash);
return segments[segmentNum].put(copies, segmentState,
return segments[segmentNum].put3(copies, segmentState,
metaKeyInterop, keyInterop, key, keySize, toKey,
getValueInterops, value, toValue,
segmentHash, replaceIfPresent, readValue, resultUnused);
@@ -1729,15 +1729,15 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return v;
}
private <KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<RV, ? super VBI>>
RV put(@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 value,
InstanceOrBytesToInstance<? super VB, V> toValue,
long hash2, boolean replaceIfPresent,
ReadValue<RV> readValue, boolean resultUnused) {
RV put3(@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 value,
InstanceOrBytesToInstance<? super VB, V> toValue,
long hash2, boolean replaceIfPresent,
ReadValue<RV> readValue, boolean resultUnused) {
shouldNotBeCalledFromReplicatedChronicleMap("Segment.put");
writeLock(null);
try {
src/test/java/eg/BGChronicleTest.java
+4 -4
View File
@@ -22,10 +22,10 @@ public class BGChronicleTest {
public static final int CHRONICLE_HEARTBEAT_SECONDS = 5;
public static final int READ_RATIO = Integer.getInteger("readRatio", 2);
// public static final boolean BUSY_WAIT = Boolean.parseBoolean(System.getProperty("busyWait",
// "" + (CLIENTS < Runtime.getRuntime().availableProcessors())));
public static final int REPLICAS = Integer.getInteger("replicas", 1);
public static final int CLIENTS = Integer.getInteger("clients", Math.max(1, Runtime.getRuntime().availableProcessors() - REPLICAS));
public static final boolean BUSY_WAIT = Boolean.parseBoolean(System.getProperty("busyWaiting",
"" + (CLIENTS + REPLICAS < Runtime.getRuntime().availableProcessors())));
public static final long KEYS = Long.getLong("keys", 1000_000L);
public static final long MESSAGES = Long.getLong("messages", 1000_000 + KEYS * 2);
public static final long MAX_RATE = Long.getLong("maxRate", 100000);
@@ -232,8 +232,8 @@ public class BGChronicleTest {
if (pause > 0)
Thread.sleep(pause);
// wait for residual.
// if (BUSY_WAIT)
// while (System.nanoTime() < next) ;
if (BUSY_WAIT)
while (System.nanoTime() < next) ;
// 100 possible values for the first byte.
long start;
key[0] = b;
src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java
+147
View File
@@ -0,0 +1,147 @@
/*
* 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.lang.io.DirectBytes;
import net.openhft.lang.io.DirectStore;
import net.openhft.lang.io.serialization.impl.SnappyStringMarshaller;
import org.junit.Ignore;
import org.junit.Test;
import java.beans.XMLEncoder;
import java.io.ByteArrayOutputStream;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
/**
* Created by peter on 06/12/14.
*/
public class LargeEntriesTest {
static final int ENTRIES = 640;
static final int ENTRY_SIZE = 1024;
@Test
@Ignore
public void testLargeStrings() throws ExecutionException, InterruptedException {
final ChronicleMap<String, String> map = ChronicleMapBuilder
.of(String.class, String.class)
.valueMarshaller(SnappyStringMarshaller.INSTANCE)
.actualSegments(1) // to force an error.
.entries(ENTRIES * 2)
.entrySize(ENTRY_SIZE / 4)
.putReturnsNull(true)
.create();
{
int threads = 2; //Runtime.getRuntime().availableProcessors();
ExecutorService es = Executors.newFixedThreadPool(threads);
final int block = ENTRIES / threads;
for (int i = 0; i < 3; i++) {
long start = System.currentTimeMillis();
List<Future<?>> futureList = new ArrayList<>();
for (int t = 0; t < threads; t++) {
final int finalT = t;
final int finalI = i;
futureList.add(es.submit(new Runnable() {
@Override
public void run() {
exerciseLargeStrings(map, finalI, finalT * block, finalT * block + block, ENTRY_SIZE);
}
}));
}
for (Future<?> future : futureList) {
future.get();
}
long time = System.currentTimeMillis() - start;
long operations = 3;
System.out.printf("Put/Get rate was %.1f MB/s%n", operations * ENTRIES * ENTRY_SIZE / 1e6 / (time / 1e3));
}
es.shutdown();
if (es.isTerminated())
map.close();
}
}
void exerciseLargeStrings(ChronicleMap<String, String> map, int run, int start, int finish, int entrySize) {
/*
final Thread thisThread = Thread.currentThread();
Thread monitor = new Thread(new Runnable() {
@Override
public void run() {
StringBuilder sb = new StringBuilder();
sb.append(thisThread).append("\n");
try {
while (!Thread.currentThread().isInterrupted()) {
Thread.sleep(50);
StackTraceElement[] stackTrace = thisThread.getStackTrace();
for (int i = 0; i < 4 && i < stackTrace.length; i++)
sb.append("\tat ").append(stackTrace[i]).append("\n");
sb.append("\n");
}
} catch (InterruptedException e) {
}
System.out.println(sb);
}
});
monitor.start();
*/
ByteArrayOutputStream baos = new ByteArrayOutputStream();
XMLEncoder xml = new XMLEncoder(baos);
Map<String, String> map2 = new HashMap<>();
Random rand = new Random(1);
for (int i = 0; i < entrySize / 80; i++)
map2.put("key-" + i, "value-" + rand.nextInt(1000));
xml.writeObject(map2);
xml.close();
String value = baos.toString().substring(0, entrySize);
// warmup to compression.
if (run == 0) {
DirectBytes bytes = DirectStore.allocate(entrySize).bytes();
SnappyStringMarshaller.INSTANCE.write(bytes, value);
bytes.flip();
SnappyStringMarshaller.INSTANCE.read(bytes);
bytes.release();
}
for (int i = start; i < finish; i++) {
String key = "key-" + i;
map.put(key, value);
String object = map.get(key);
assertNotNull(key, object);
assertEquals(key, entrySize, object.length());
}
// monitor.interrupt();
for (int i = start; i < finish; i++) {
// System.out.println(i);
String key = "key-" + i;
String object = map.get(key);
assertNotNull(key, object);
assertEquals(key, entrySize, object.length());
}
}
}