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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Rob Austin
2014-12-20 13:36:21 +00:00
parent 8d44587c7c f9b4ba79e5
commit 429cb2bcb2
17 changed files with 338 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+3 -46
View File
@@ -64,6 +64,7 @@ import static java.lang.Math.max;
import static java.lang.Thread.currentThread;
import static java.nio.ByteBuffer.allocateDirect;
import static net.openhft.chronicle.map.Asserts.assertNotNull;
import static net.openhft.chronicle.map.XStreamHelper.to$$Native;
import static net.openhft.lang.MemoryUnit.*;
class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
@@ -862,7 +863,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
case "chronicle-key":
if (kClass.getCanonicalName().startsWith("net.openhft.lang.values")) {
return to$$Native(reader, vClass, true);
return to$$Native(reader, vClass, true, VanillaChronicleMap.this);
} else {
long keySize = keySizeMarshaller.readSize(buffer);
ThreadLocalCopies copies = keyReaderProvider.getCopies(null);
@@ -872,7 +873,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
case "chronicle-value":
if (vClass.getCanonicalName().startsWith("net.openhft.lang.values")) {
return to$$Native(reader, vClass, false);
return to$$Native(reader, vClass, false, VanillaChronicleMap.this);
} else {
long valueSize = valueSizeMarshaller.readSize(buffer);
ThreadLocalCopies copies = valueReaderProvider.getCopies(null);
@@ -904,51 +905,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
} else return (E) unmarshallingContext.convertAnother(null, clazz);
}
private Object to$$Native(HierarchicalStreamReader reader, final Class aClass, final boolean isKey) {
final Object o = isKey ? newKeyInstance() : newValueInstance();
try {
for (Method $$nativeMethod : o.getClass().getMethods()) {
if ($$nativeMethod.getName().equals("setValue") && $$nativeMethod.getParameterTypes()
.length == 1) {
Class<?> parameterType = $$nativeMethod.getParameterTypes()[0];
String value = reader.getValue();
if (parameterType.isPrimitive()) {
// convert the primitive to their boxed type
if (parameterType == int.class)
parameterType = Integer.class;
else if (parameterType == char.class)
parameterType = Character.class;
else {
final String name = parameterType.getSimpleName();
final String properName = "java.lang." + Character.toString
(name.charAt(0)).toUpperCase()
+ name.substring(1);
parameterType = Class.forName(properName);
}
}
final Method valueOf = parameterType.getMethod("valueOf", String.class);
final Object invoke = valueOf.invoke(null, value);
$$nativeMethod.invoke(o, invoke);
return o;
}
}
} catch (Exception e) {
throw new ConversionException("class=" + aClass.getCanonicalName(), e);
}
throw new ConversionException("setValue(..) method not found in class=" + aClass
.getCanonicalName
());
}
};
xstream.registerConverter(converter);
src/main/java/net/openhft/chronicle/map/XStreamHelper.java
+75
View File
@@ -0,0 +1,75 @@
/*
* 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 com.thoughtworks.xstream.converters.ConversionException;
import com.thoughtworks.xstream.io.HierarchicalStreamReader;
import java.lang.reflect.Method;
/**
* Created by Rob on 18/12/14.
*/
class XStreamHelper {
static Object to$$Native(HierarchicalStreamReader reader, final Class aClass, final boolean isKey, ChronicleMap map) {
final Object o = isKey ? map.newKeyInstance() : map.newValueInstance();
try {
for (Method $$nativeMethod : o.getClass().getMethods()) {
if ($$nativeMethod.getName().equals("setValue") && $$nativeMethod.getParameterTypes()
.length == 1) {
Class<?> parameterType = $$nativeMethod.getParameterTypes()[0];
String value = reader.getValue();
if (parameterType.isPrimitive()) {
// convert the primitive to their boxed type
if (parameterType == int.class)
parameterType = Integer.class;
else if (parameterType == char.class)
parameterType = Character.class;
else {
final String name = parameterType.getSimpleName();
final String properName = "java.lang." + Character.toString
(name.charAt(0)).toUpperCase()
+ name.substring(1);
parameterType = Class.forName(properName);
}
}
final Method valueOf = parameterType.getMethod("valueOf", String.class);
final Object invoke = valueOf.invoke(null, value);
$$nativeMethod.invoke(o, invoke);
return o;
}
}
} catch (Exception e) {
throw new ConversionException("class=" + aClass.getCanonicalName(), e);
}
throw new ConversionException("setValue(..) method not found in class=" + aClass
.getCanonicalName
());
}
}
src/test/java/eg/TestNesterInterfaceMarshal.java
+29 -14
View File
@@ -29,10 +29,13 @@ public class TestNesterInterfaceMarshal {
public void testReplicatedMap() throws Exception {
ChronicleMap chronicleMap = null;
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS);
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig =
TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS);
try {
ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class)
.entries(5000L).entrySize(2000)
ChronicleMapBuilder builder = ChronicleMapBuilder
.of(CharSequence.class, TestInstrumentVOInterface.class)
.entries(5000L)
.keySize(10)
.replication((byte) 1, tcpTransportAndNetworkConfig);
@@ -44,20 +47,24 @@ public class TestNesterInterfaceMarshal {
ChronicleMap<CharSequence, TestInstrumentVOInterface> map = chronicleMap;
ChronicleMap chronicleMap1 = null;
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig1 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)).heartBeatInterval(1L, TimeUnit.SECONDS);
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig1 =
TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076))
.heartBeatInterval(1L, TimeUnit.SECONDS);
try {
ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class)
ChronicleMapBuilder builder = ChronicleMapBuilder
.of(CharSequence.class, TestInstrumentVOInterface.class)
.putReturnsNull(true)
.removeReturnsNull(true)
.entries(5000L).entrySize(2000)
.entries(5000L).keySize(10)
.replication((byte) 2, tcpTransportAndNetworkConfig1);
chronicleMap1 = builder.create();
} catch (Exception e) {
System.out.println("*********************Error(s) creating launcher instrument cache: " + e);
System.out.println("*********************Error(s) creating launcher " +
"instrument cache: " + e);
}
ChronicleMap<CharSequence, TestInstrumentVOInterface> replicatedMap = chronicleMap1;
@@ -107,10 +114,13 @@ public class TestNesterInterfaceMarshal {
protected ChronicleMap createReplicatedMap1() {
ChronicleMap chronicleMap = null;
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS);
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig =
TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS);
try {
ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class)
.entries(5000L).entrySize(2000)
ChronicleMapBuilder builder = ChronicleMapBuilder
.of(CharSequence.class, TestInstrumentVOInterface.class)
.entries(5000L)
.keySize(10)
.replication((byte) 1, tcpTransportAndNetworkConfig);
@@ -125,20 +135,25 @@ public class TestNesterInterfaceMarshal {
protected ChronicleMap createReplicatedMap2() {
ChronicleMap chronicleMap = null;
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)).heartBeatInterval(1L, TimeUnit.SECONDS);
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig =
TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076))
.heartBeatInterval(1L, TimeUnit.SECONDS);
try {
ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class)
ChronicleMapBuilder builder = ChronicleMapBuilder
.of(CharSequence.class, TestInstrumentVOInterface.class)
.putReturnsNull(true)
.removeReturnsNull(true)
.entries(5000L).entrySize(2000)
.entries(5000L)
.keySize(10)
.replication((byte) 2, tcpTransportAndNetworkConfig);
chronicleMap = builder.create();
} catch (Exception e) {
System.out.println("*********************Error(s) creating launcher instrument cache: " + e);
System.out.println("*********************Error(s) creating launcher " +
"instrument cache: " + e);
}
return chronicleMap;
}
src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java
-1
View File
@@ -78,7 +78,6 @@ public class CHMLatencyTestMain {
ChronicleMap<LongValue, LongValue> countersMap =
ChronicleMapBuilder.of(LongValue.class, LongValue.class)
.entries(KEYS)
.entrySize(16)
.createPersistedTo(file);
// add keys
src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java
+4 -4
View File
@@ -62,7 +62,7 @@ public class ChannelReplicationTest {
map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(2)
.entrySize(100)
.constantValueSizeBySample("EXAMPLE-1")
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
}
@@ -76,7 +76,7 @@ public class ChannelReplicationTest {
map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(2)
.entrySize(100)
.constantValueSizeBySample("EXAMPLE-1")
.instance().replicatedViaChannel(hubB.createChannel((short) 1)).create();
}
}
@@ -111,12 +111,12 @@ public class ChannelReplicationTest {
map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(2)
.entrySize(100)
.constantValueSizeBySample("EXAMPLE-1")
.instance().replicatedViaChannel(hubB.createChannel((short) 2)).create();
map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(2)
.entrySize(100)
.constantValueSizeBySample("EXAMPLE-1")
.instance().replicatedViaChannel(hubA.createChannel((short) 2)).create();
map2a.put(1, "EXAMPLE-2");
src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java
+49 -35
View File
@@ -103,17 +103,12 @@ public class ChronicleMapTest {
}
private static ChronicleMap<CharSequence, LongValue> getSharedMap(
long entries, int segments, int entrySize) throws IOException {
return getSharedMap(entries, segments, entrySize, OF_4_BYTES);
}
private static ChronicleMap<CharSequence, LongValue> getSharedMap(
long entries, int segments, int entrySize, Alignment alignment)
long entries, int segments, Alignment alignment, int keySize)
throws IOException {
return ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(entries)
.minSegments(segments)
.entrySize(entrySize)
.keySize(keySize)
.entryAndValueAlignment(alignment)
.create();
}
@@ -204,7 +199,7 @@ public class ChronicleMapTest {
int count = 300;
final ChronicleMap<Object, Object> map = ChronicleMapBuilder.of(Object.class, Object.class)
.entrySize(count)
.entries(count)
.minSegments(2).create();
for (int i = 1; i < count; i++) {
@@ -414,10 +409,13 @@ public class ChronicleMapTest {
@Test
public void testAcquireWithNullContainer() throws Exception {
ChronicleMap<CharSequence, LongValue> map = getSharedMap(10 * 1000, 128, 24);
ChronicleMap<CharSequence, LongValue> map =
ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(1000)
.entryAndValueAlignment(OF_4_BYTES)
.create();
map.acquireUsing("key", new LongValue$$Native());
assertEquals(0, map.acquireUsing("key", null).getValue());
map.close();
}
@@ -454,7 +452,11 @@ public class ChronicleMapTest {
@Test
public void testGetWithNullContainer() throws Exception {
ChronicleMap<CharSequence, LongValue> map = getSharedMap(10 * 1000, 128, 24);
ChronicleMap<CharSequence, LongValue> map =
ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(10)
.entryAndValueAlignment(OF_4_BYTES)
.create();
map.acquireUsing("key", new LongValue$$Native());
assertEquals(0, map.getUsing("key", null).getValue());
@@ -463,7 +465,11 @@ public class ChronicleMapTest {
@Test
public void testGetWithoutAcquireFirst() throws Exception {
ChronicleMap<CharSequence, LongValue> map = getSharedMap(10, 1, 24);
ChronicleMap<CharSequence, LongValue> map =
ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(10)
.entryAndValueAlignment(OF_4_BYTES)
.create();
assertNull(map.getUsing("key", DataValueClasses.newDirectReference(LongValue.class)));
map.close();
@@ -485,9 +491,9 @@ public class ChronicleMapTest {
public void testAcquireAndGet() throws IOException, ClassNotFoundException,
IllegalAccessException, InstantiationException {
int entries = 3/*00 * 1000*/;
testAcquireAndGet(getSharedMap(entries, 1, 24, OF_4_BYTES), entries);
testAcquireAndGet(getSharedMap(entries, 1, 24, NO_ALIGNMENT), entries);
testAcquireAndGet(getSharedMap(entries, 1, 24, OF_8_BYTES), entries);
testAcquireAndGet(getSharedMap(entries, 1, OF_4_BYTES, 10), entries);
testAcquireAndGet(getSharedMap(entries, 1, NO_ALIGNMENT, 10), entries);
testAcquireAndGet(getSharedMap(entries, 1, OF_8_BYTES, 10), entries);
}
public void testAcquireAndGet(ChronicleMap<CharSequence, LongValue> map, int entries)
@@ -526,9 +532,9 @@ public class ChronicleMapTest {
@Test
public void testAcquireFromMultipleThreads() throws Exception {
int entries = 1000 * 1000;
testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, NO_ALIGNMENT));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, OF_4_BYTES));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, OF_8_BYTES));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, NO_ALIGNMENT, 10));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, OF_4_BYTES, 10));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, OF_8_BYTES, 10));
}
public void testAcquireFromMultipleThreads(ChronicleMap<CharSequence, LongValue> map)
@@ -537,7 +543,7 @@ public class ChronicleMapTest {
CharSequence key = getUserCharSequence(0);
map.acquireUsing(key, new LongValue$$Native());
int iterations = 1000;
int iterations = 10000;
int noOfThreads = 10;
CyclicBarrier barrier = new CyclicBarrier(noOfThreads);
@@ -582,7 +588,7 @@ public class ChronicleMapTest {
.entries(entries)
.actualSegments(8 * 1024)
.entryAndValueAlignment(OF_8_BYTES)
.entrySize(24);
.keySize(13);
File tmpFile = File.createTempFile("testAcquirePerf", ".deleteme");
tmpFile.deleteOnExit();
@@ -603,8 +609,10 @@ public class ChronicleMapTest {
StringBuilder sb = new StringBuilder();
long next = 50 * 1000 * 1000;
// use a factor to give up to 10 digit numbers.
int factor = Math.max(1, (int) ((10 * 1000 * 1000 * 1000L - 1) / entries));
for (long j = t % independence; j < entries + independence - 1; j += independence) {
int factor = Math.max(1,
(int) ((10 * 1000 * 1000 * 1000L - 1) / entries));
for (long j = t % independence; j < entries + independence - 1;
j += independence) {
sb.setLength(0);
sb.append("us:");
sb.append(j * factor);
@@ -625,7 +633,8 @@ public class ChronicleMapTest {
future.get();
}
long time = System.currentTimeMillis() - start;
System.out.printf("Throughput %.1f M ops/sec%n", threads * entries / independence / 1000.0 / time);
System.out.printf("Throughput %.1f M ops/sec%n",
threads * entries / independence / 1000.0 / time);
}
printStatus();
File file = map.file();
@@ -654,7 +663,7 @@ public class ChronicleMapTest {
.entries(entries)
.entryAndValueAlignment(OF_8_BYTES)
.actualSegments(256)
.entrySize(24);
.keySize(13);
File tmpFile = File.createTempFile("testAcquirePerf", ".deleteme");
tmpFile.deleteOnExit();
@@ -676,8 +685,10 @@ public class ChronicleMapTest {
long next = 50 * 1000 * 1000;
Random rand = new Random();
// use a factor to give up to 10 digit numbers.
int factor = Math.max(1, (int) ((10 * 1000 * 1000 * 1000L - 1) / entries));
for (long j = t % independence; j < entries + independence - 1; j += independence) {
int factor = Math.max(1,
(int) ((10 * 1000 * 1000 * 1000L - 1) / entries));
for (long j = t % independence; j < entries + independence - 1;
j += independence) {
sb.setLength(0);
sb.append("us:");
sb.append(j * factor);
@@ -707,7 +718,8 @@ public class ChronicleMapTest {
future.get();
}
long time = System.currentTimeMillis() - start;
System.out.printf("Throughput %.1f M ops/sec%n", threads * entries / independence / 1000.0 / time);
System.out.printf("Throughput %.1f M ops/sec%n",
threads * entries / independence / 1000.0 / time);
}
printStatus();
File file = map.file();
@@ -853,14 +865,16 @@ public class ChronicleMapTest {
}
@Test
public void testPutAndRemove() throws IOException, ClassNotFoundException, IllegalAccessException, InstantiationException {
public void testPutAndRemove() throws IOException, ClassNotFoundException,
IllegalAccessException, InstantiationException {
String TMP = System.getProperty("java.io.tmpdir");
int entries = 100 * 1000;
ChronicleMap<CharSequence, CharSequence> map =
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
.entries(entries)
.minSegments(16)
.entrySize(32)
.keySize("user:".length() + 6)
.valueSize("value:".length() + 6)
.putReturnsNull(true)
.removeReturnsNull(true).create();
StringBuilder key = new StringBuilder();
@@ -1412,7 +1426,7 @@ public class ChronicleMapTest {
final ChronicleMapBuilder<CharSequence, LongValue> builder = ChronicleMapBuilder
.of(CharSequence.class, LongValue.class)
.entries(1000)
.entrySize(16);
.keySize("x".length());
final ChronicleMap<CharSequence, LongValue> map = builder.create();
@@ -1425,7 +1439,7 @@ public class ChronicleMapTest {
ChronicleMapBuilder<CharSequence, LongValue> builder = ChronicleMapBuilder
.of(CharSequence.class, LongValue.class)
.entries(1000)
.entrySize(16);
.keySize("one".length());
final ChronicleMap<CharSequence, LongValue> map = builder.create();
@@ -1483,8 +1497,7 @@ public class ChronicleMapTest {
ChronicleMapBuilder<CharSequence, String> builder = ChronicleMapBuilder
.of(CharSequence.class, String.class)
.entries(1000)
.defaultValue("")
.entrySize(16);
.defaultValue("");
try (final ChronicleMap<CharSequence, String> map = builder.create()) {
// this will add the entry
@@ -1500,7 +1513,8 @@ public class ChronicleMapTest {
try (final ChronicleMap<CharSequence, CharSequence> map = ChronicleMapBuilder
.of(CharSequence.class, CharSequence.class)
.entries(1000)
.entrySize(40)
.keySize("one".length())
.valueSize("Hello World".length())
.defaultValue("")
.create()) {
StringBuilder value = new StringBuilder();
@@ -1518,7 +1532,7 @@ public class ChronicleMapTest {
ChronicleMapBuilder<CharSequence, LongValue> builder = ChronicleMapBuilder
.of(CharSequence.class, LongValue.class)
.entries(1000)
.entrySize(40);
.keySize("one".length());
File tmpFile = File.createTempFile("testAcquireUsingLocked", ".deleteme");
tmpFile.deleteOnExit();
final ChronicleMap<CharSequence, LongValue> map = builder.createPersistedTo(tmpFile);
src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java
+28 -11
View File
@@ -34,6 +34,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
import static java.lang.Math.log10;
import static java.lang.Math.round;
import static net.openhft.chronicle.map.Alignment.OF_4_BYTES;
import static net.openhft.lang.model.DataValueClasses.newDirectReference;
import static org.junit.Assert.assertTrue;
@@ -51,12 +53,13 @@ public class EntryCountMapTest {
}
private static ChronicleMap<CharSequence, LongValue> getSharedMap(
long entries, int segments, int entrySize) throws IOException {
ChronicleMapBuilder<CharSequence, LongValue> mapBuilder = ChronicleMapBuilder.of(CharSequence
.class, LongValue.class)
long entries, int segments, int keySize) throws IOException {
ChronicleMapBuilder<CharSequence, LongValue> mapBuilder =
ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(entries)
.actualSegments(segments)
.entrySize(entrySize);
.keySize(keySize)
.maxEntryOversizeFactor(1);
return mapBuilder.createPersistedTo(getPersistenceFile());
}
@@ -182,12 +185,17 @@ public class EntryCountMapTest {
System.out.printf("Score: %.2f%n", scoreCount / score);
}
private Future<Void> testEntriesMaxSize(ExecutorService es, final int segments, final int minSize, final int maxSize, final int seed) throws IOException {
private Future<Void> testEntriesMaxSize(ExecutorService es, final int segments,
final int minSize, final int maxSize, final int seed)
throws IOException {
assert minSize <= maxSize;
Random random = new Random(seed);
int counter = minSize + random.nextInt(9999 + maxSize);
final int stride = 1 + random.nextInt(100);
final ChronicleMap<CharSequence, LongValue> map = getSharedMap(minSize, segments, 32);
int maxKeySize = "key:".length() +
(int) round(log10(moreThanMaxSize(maxSize) * stride + counter)) + 1;
final ChronicleMap<CharSequence, LongValue> map =
getSharedMap(minSize, segments, maxKeySize);
return es.submit(new Callable<Void>() {
@Override
public Void call() throws Exception {
@@ -202,14 +210,17 @@ public class EntryCountMapTest {
Random random = new Random(seed);
int counter = minSize + random.nextInt(9999 + maxSize);
int stride = 1 + random.nextInt(100);
ChronicleMap<CharSequence, LongValue> map = getSharedMap(minSize, segments, 32);
int maxKeySize = "key:".length() +
(int) round(log10(moreThanMaxSize(maxSize) * stride + counter)) + 1;
ChronicleMap<CharSequence, LongValue> map = getSharedMap(minSize, segments, maxKeySize);
testEntriesMaxSize0(segments, minSize, maxSize, counter, stride, map);
}
void testEntriesMaxSize0(int segments, int minSize, int maxSize, int counter, int stride, ChronicleMap<CharSequence, LongValue> map) {
void testEntriesMaxSize0(int segments, int minSize, int maxSize, int counter, int stride,
ChronicleMap<CharSequence, LongValue> map) {
LongValue longValue = newDirectReference(LongValue.class);
try {
for (int j = 0; j < maxSize * 14 / 10 + 300; j++) {
for (int j = 0; j < moreThanMaxSize(maxSize); j++) {
String key = "key:" + counter;
counter += stride;
// give a biased hashcode distribution.
@@ -229,14 +240,20 @@ public class EntryCountMapTest {
boolean condition = minSize <= map.size() && map.size() <= maxSize;
if (!condition) {
dumpMapStats(segments, minSize, map);
assertTrue("stride: " + stride + ", seg: " + segments + ", min: " + minSize + ", size: " + map.size(), condition);
assertTrue("stride: " + stride + ", seg: " + segments + ", min: " + minSize +
", size: " + map.size(), condition);
}
} finally {
map.close();
}
}
private void dumpMapStats(int segments, int minSize, ChronicleMap<CharSequence, LongValue> map) {
private static int moreThanMaxSize(int maxSize) {
return maxSize * 14 / 10 + 300;
}
private void dumpMapStats(int segments, int minSize,
ChronicleMap<CharSequence, LongValue> map) {
long[] a = ((VanillaChronicleMap) map).segmentSizes();
System.out.println("segs: " + segments + " min: " + minSize + " was " + map.size() + " "
+ Arrays.toString(a)
src/test/java/net/openhft/chronicle/map/MapByNameTest.java
+1 -1
View File
@@ -50,7 +50,7 @@ public class MapByNameTest {
final MapInstanceConfig<CharSequence, CharSequence> config =
(MapInstanceConfig<CharSequence, CharSequence>)
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
.entrySize(900)
.keySize("hello".length()).valueSize("world".length())
.entries(2)
.minSegments(2)
.replication((byte) 1, TcpTransportAndNetworkConfig.of(8244))
src/test/java/net/openhft/chronicle/map/OSResizesMain.java
+5 -5
View File
@@ -33,13 +33,13 @@ public class OSResizesMain {
public static void main(String[] args) throws IOException, InterruptedException {
File file = File.createTempFile("over-sized", "deleteme");
int valueSize = 1000 * 1000;
ChronicleMap<String, ByteArray> map = ChronicleMapBuilder.of(String.class, ByteArray.class)
.entrySize(50 * 1024 * 1024)
.valueSize(valueSize + 16)
.createPersistedTo(file);
byte[] chars = new byte[valueSize];
Arrays.fill(chars, (byte) '+');
ByteArray ba = new ByteArray(chars);
ChronicleMap<String, ByteArray> map = ChronicleMapBuilder.of(String.class, ByteArray.class)
.keySize("key-".length() + 4)
.constantValueSizeBySample(ba)
.createPersistedTo(file);
Arrays.fill(chars, (byte) '+');
for (int i = 0; i < 1000; i++) {
map.put("key-" + i, ba);
}
src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
+5 -7
View File
@@ -465,22 +465,20 @@ public class StatelessClientTest {
@Test(timeout = 10000)
public void testLargeEntries() throws IOException,
InterruptedException {
int keySize = 4;
int valueSize = 1000000;
int entrySize = keySize + valueSize;
char[] value = new char[valueSize];
Arrays.fill(value, 'X');
try (ChronicleMap<Integer, CharSequence> serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entrySize(entrySize)
String sampleValue = new String(value);
try (ChronicleMap<Integer, CharSequence> serverMap =
ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.constantValueSizeBySample(sampleValue)
.replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) {
try (ChronicleMap<Integer, CharSequence> statelessMap = ChronicleMapBuilder.of(Integer
.class, CharSequence.class)
.entrySize(entrySize)
.constantValueSizeBySample(sampleValue)
.statelessClient(new InetSocketAddress("localhost", 8056)).create()) {
// for (int i = 0; i < 128; i++) {
src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java
+5 -6
View File
@@ -1,6 +1,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.lang.MemoryUnit;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -24,7 +25,7 @@ public class TcpReplicationLargeEntryTest {
private ChronicleMap<Integer, CharSequence> map2;
static int s_port = 8010;
int valueSize = 1_00_000; // 1MB
int valueSize = (int) MemoryUnit.MEGABYTES.toBytes(1);
char[] valueX = new char[valueSize - 100];
char[] valueY = new char[valueSize - 100];
@@ -34,9 +35,7 @@ public class TcpReplicationLargeEntryTest {
Arrays.fill(valueX, 'X');
Arrays.fill(valueX, 'Y');
int keySize = 4;
int entrySize = keySize + valueSize;
String sampleValue = new String(valueX);
final InetSocketAddress endpoint = new InetSocketAddress("localhost", s_port + 1);
{
@@ -47,7 +46,7 @@ public class TcpReplicationLargeEntryTest {
map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entrySize(entrySize)
.constantValueSizeBySample(sampleValue)
.entries(2)
.actualSegments(1)
.replication((byte) 1, tcpConfig1)
@@ -62,7 +61,7 @@ public class TcpReplicationLargeEntryTest {
.tcpBufferSize(1024 * 64);
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entrySize(entrySize)
.constantValueSizeBySample(sampleValue)
.entries(2)
.replication((byte) 2, tcpConfig2)
.instance()
src/test/java/net/openhft/chronicle/map/TcpReplicationWithAcquireUsingLockedTest.java
+12 -8
View File
@@ -80,20 +80,22 @@ public class TcpReplicationWithAcquireUsingLockedTest {
(CharSequence.class,
TestInstrumentVOInterface
.class)
.entries(5000L).entrySize(2000)
.entries(5000L).keySize("hello".length())
.replication((byte) 1, config1).create()) {
map1a = map1;
TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076));
TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig
.of(8077, new InetSocketAddress("127.0.0.1", 8076));
try (ChronicleMap<CharSequence, TestInstrumentVOInterface> map2 = ChronicleMapBuilder.of(CharSequence.class,
try (ChronicleMap<CharSequence, TestInstrumentVOInterface> map2 =
ChronicleMapBuilder.of(CharSequence.class,
TestInstrumentVOInterface.class)
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) 2, config2)
.entries(5000L).entrySize(2000).create()) {
.entries(5000L).keySize("hello".length()).create()) {
map2a = map2;
@@ -131,20 +133,22 @@ public class TcpReplicationWithAcquireUsingLockedTest {
(CharSequence.class,
TestInstrumentVOInterface
.class)
.entries(5000L).entrySize(2000)
.entries(5000L).keySize("hello".length())
.replication((byte) 1, config1).create()) {
map1a = map1;
TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076));
TcpTransportAndNetworkConfig config2 =
TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076));
try (ChronicleMap<CharSequence, TestInstrumentVOInterface> map2 = ChronicleMapBuilder.of(CharSequence.class,
try (ChronicleMap<CharSequence, TestInstrumentVOInterface> map2 =
ChronicleMapBuilder.of(CharSequence.class,
TestInstrumentVOInterface.class)
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) 2, config2)
.entries(5000L).entrySize(2000).create()) {
.entries(5000L).keySize("hello".length()).create()) {
map2a = map2;
src/test/java/net/openhft/chronicle/map/example/LotsOfEntriesMain.java
+99
View File
@@ -0,0 +1,99 @@
package net.openhft.chronicle.map.example;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.lang.model.constraints.MaxSize;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
* Created by peter on 19/12/14.
*/
public class LotsOfEntriesMain {
public static void main(String[] args)
throws IOException, ExecutionException, InterruptedException {
workEntries(true);
workEntries(false);
}
private static void workEntries(final boolean add)
throws IOException, ExecutionException, InterruptedException {
final long entries = 100_000_000;
File file = new File("/tmp/lotsOfEntries.dat");
double averageEntrySize =
1 + // key size marshalling size
// + 2 is average oversize because we append 4-letter "-key" in a loop
(Math.log(1.024) - Math.log(0.024)) * 24 + 2 + // average key size
24 + // value size
2 // average loss due to value alignment. defaults to OF_4_BYTES because
// there are floats in values
;
final ChronicleMap<CharSequence, MyFloats> map = ChronicleMapBuilder
.of(CharSequence.class, MyFloats.class)
.entries((long) (entries *
(averageEntrySize / 8.0 +
0.5))) // account average internal fragmentation --
// because we need to round up to integral number of chunks
.entrySize(8) // "chunk mode"
.createPersistedTo(file);
int threads = Runtime.getRuntime().availableProcessors();
ExecutorService es = Executors.newFixedThreadPool(threads);
long block = (entries + threads - 1) / threads;
final long start = System.nanoTime();
List<Future<?>> futures = new ArrayList<>();
for (int t = 0; t < threads; t++) {
final long startI = t * block;
final long endI = Math.min((t + 1) * block, entries);
futures.add(es.submit(new Runnable() {
@Override
public void run() {
Random rand = new Random(startI);
StringBuilder sb = new StringBuilder();
MyFloats mf = map.newValueInstance();
if (add)
for (int i = 0; i < 6; i++)
mf.setValueAt(i, i);
for (long i = startI; i < endI; i++) {
sb.setLength(0);
int length = (int) (24 / (rand.nextFloat() + 24.0 / 1000));
sb.append(i);
while (sb.length() < length)
sb.append("-key");
try {
if (add)
map.put(sb, mf);
else
map.getUsing(sb, mf);
} catch (Exception e) {
System.out.println("map.size: "+map.size());
throw e;
}
}
}
}));
}
for (Future<?> future : futures) {
future.get();
}
long time = System.nanoTime() - start;
es.shutdown();
System.out.printf("Map.size: %,d with a throughput of %.1f million/sec to %s.%n",
map.size(), entries * 1e3 / time, add ? "add" : "get");
map.close();
}
}
interface MyFloats {
public void setValueAt(@MaxSize(6) int index, float f);
public float getValueAt(int index);
}
src/test/java/net/openhft/chronicle/map/example/ServersMapMain.java
+10 -5
View File
@@ -39,9 +39,11 @@ public class ServersMapMain {
public static void startServer() throws IOException {
File file = File.createTempFile("testServersMapMain", ".deleteme");
file.deleteOnExit();
final ChronicleMap<byte[], byte[]> serverMap = ChronicleMapBuilder.of(byte[].class, byte[].class)
final ChronicleMap<byte[], byte[]> serverMap =
ChronicleMapBuilder.of(byte[].class, byte[].class)
.putReturnsNull(true)
.entrySize(50)
.constantKeySizeBySample(new byte[8])
.constantValueSizeBySample(new byte[32])
.replication((byte) 2, TcpTransportAndNetworkConfig.of(port))
.createPersistedTo(file);
System.out.println("Server started");
@@ -59,7 +61,8 @@ public class ServersMapMain {
} else {
File file = File.createTempFile("testServersMapMain", ".deleteme");
file.deleteOnExit();
TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port));
TcpTransportAndNetworkConfig tcpConfig =
TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port));
map = ChronicleMapBuilder.of(
byte[].class, byte[].class)
@@ -83,11 +86,13 @@ public class ServersMapMain {
puts += t2 - t1;
gets += t3 - t2;
if (t2 - t1 > lastAveragePut * 100 || t3 - t2 > lastAverageGet * 100)
System.out.printf("Took put/get took %.1f/%.1f us%n", (t2 - t1) / 1e3, (t3 - t2) / 1e3);
System.out.printf("Took put/get took %.1f/%.1f us%n",
(t2 - t1) / 1e3, (t3 - t2) / 1e3);
}
lastAveragePut = puts / entries;
lastAverageGet = gets / entries;
System.out.printf("Average took put/get took %.1f/%.1f us%n", lastAveragePut / 1e3, lastAverageGet / 1e3);
System.out.printf("Average took put/get took %.1f/%.1f us%n",
lastAveragePut / 1e3, lastAverageGet / 1e3);
}
}
src/test/java/net/openhft/chronicle/map/example/StringsMapMain.java
+9 -4
View File
@@ -38,10 +38,12 @@ public class StringsMapMain {
public static void startServer() throws IOException {
File file = File.createTempFile("testServersMapMain", ".deleteme");
file.deleteOnExit();
int keyAndValueSize = (int) Math.round(Math.log10(entries));
final ChronicleMap<CharSequence, CharSequence> serverMap = ChronicleMapBuilder
.of(CharSequence.class, CharSequence.class)
.putReturnsNull(true)
.entrySize(50)
.keySize(keyAndValueSize)
.valueSize(keyAndValueSize)
.replication((byte) 2, TcpTransportAndNetworkConfig.of(port))
.createPersistedTo(file);
System.out.println("Server started");
@@ -58,7 +60,8 @@ public class StringsMapMain {
} else {
File file = File.createTempFile("testServersMapMain", ".deleteme");
file.deleteOnExit();
TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port));
TcpTransportAndNetworkConfig tcpConfig =
TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port));
map = ChronicleMapBuilder
.of(CharSequence.class, CharSequence.class)
@@ -90,11 +93,13 @@ public class StringsMapMain {
puts += t2 - t1;
gets += t3 - t2;
// if (t2 - t1 > lastAveragePut * 100 || t3 - t2 > lastAverageGet * 100)
// System.out.printf("Took put/get took %.1f/%.1f us%n", (t2 - t1) / 1e3, (t3 - t2) / 1e3);
// System.out.printf("Took put/get took %.1f/%.1f us%n",
// (t2 - t1) / 1e3, (t3 - t2) / 1e3);
}
lastAveragePut = puts / entries;
lastAverageGet = gets / entries;
System.out.printf("Average took put/get took %.1f/%.1f us%n", lastAveragePut / 1e3, lastAverageGet / 1e3);
System.out.printf("Average took put/get took %.1f/%.1f us%n",
lastAveragePut / 1e3, lastAverageGet / 1e3);
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongCASLeft.java
+3 -2
View File
@@ -50,7 +50,8 @@ public class PingPongCASLeft {
playPingPong(chm, 4, 5, true, "PingPongCASLEFT");
}
static void playPingPong(ChronicleMap<String, BondVOInterface> chm, double _coupon, double _coupon2, boolean setFirst, final String desc) {
static void playPingPong(ChronicleMap<String, BondVOInterface> chm, double _coupon,
double _coupon2, boolean setFirst, final String desc) {
BondVOInterface bond1 = newDirectReference(BondVOInterface.class);
BondVOInterface bond2 = newDirectReference(BondVOInterface.class);
BondVOInterface bond3 = newDirectReference(BondVOInterface.class);
@@ -91,6 +92,6 @@ public class PingPongCASLeft {
AffinitySupport.setThreadId();
return ChronicleMapBuilder.of(String.class, BondVOInterface.class)
.entries(16)
.entrySize(64).create();
.keySize("369604101".length()).create();
}
}
src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java
+1 -1
View File
@@ -137,7 +137,7 @@ public class ProcessInstanceLimiter implements Runnable {
ChronicleMapBuilder<String, Data> builder =
ChronicleMapBuilder.of(String.class, Data.class);
builder.entries(1000);
builder.entrySize(1024);
builder.keySize((DEFAULT_PROCESS_NAME + "#").length());
this.theSharedMap = builder.createPersistedTo(new File(sharedMapPath));
Thread t = new Thread(this, "ProcessInstanceLimiter updater");
t.setDaemon(true);