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

Improve support for large entries

Browse Source
This commit is contained in:
Peter Lawrey
2014-12-11 07:46:05 +00:00
parent 41eef84920
commit 356e6c8181
1 changed files with 85 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java
+85 -27
View File
@@ -41,23 +41,23 @@ import static org.junit.Assert.*;
*/
public class LargeEntriesTest {
static final int ENTRIES = 6400;
static final int ENTRY_SIZE = 102400;
private static final boolean TESTING = Boolean.getBoolean("testing");
@Test
@Ignore
public void testLargeStrings() throws ExecutionException, InterruptedException, IOException {
final int ENTRIES = 250;
final int ENTRY_SIZE = 100 * 1024;
File file = File.createTempFile("largeEntries", ".deleteme");
file.deleteOnExit();
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)
.entries(ENTRIES)
.entrySize(ENTRY_SIZE / 6)
.putReturnsNull(true)
.createPersistedTo(file);
{
warmUpCompression(ENTRY_SIZE);
int threads = 4; //Runtime.getRuntime().availableProcessors();
ExecutorService es = Executors.newFixedThreadPool(threads);
final int block = ENTRIES / threads;
@@ -66,11 +66,10 @@ public class LargeEntriesTest {
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);
exerciseLargeStrings(map, finalT * block, finalT * block + block, ENTRY_SIZE);
}
}));
}
@@ -87,7 +86,70 @@ public class LargeEntriesTest {
}
}
void exerciseLargeStrings(ChronicleMap<String, String> map, int run, int start, int finish, int entrySize) {
private void warmUpCompression(int entrySize) {
String value = generateValue(entrySize);
DirectBytes bytes = DirectStore.allocate(entrySize / 6).bytes();
for (int i = 0; i < 5; i++) {
// warmup to compression.
bytes.clear();
SnappyStringMarshaller.INSTANCE.write(bytes, value);
bytes.flip();
SnappyStringMarshaller.INSTANCE.read(bytes);
}
bytes.release();
}
@Test
@Ignore
public void testLargeStringsPerf() throws ExecutionException, InterruptedException, IOException {
doLargeEntryPerf(10000, 100 * 1024);
doLargeEntryPerf(100000, 10 * 1024);
doLargeEntryPerf(10000, 100 * 1024);
doLargeEntryPerf(3000, 1024 * 1024);
}
private void doLargeEntryPerf(int ENTRIES, final int ENTRY_SIZE) throws IOException, InterruptedException, ExecutionException {
System.out.printf("Testing %,d entries of %,d KB%n", ENTRIES, ENTRY_SIZE);
File file = File.createTempFile("largeEntries", ".deleteme");
file.deleteOnExit();
final ChronicleMap<String, String> map = ChronicleMapBuilder
.of(String.class, String.class)
.valueMarshaller(SnappyStringMarshaller.INSTANCE)
.entries(ENTRIES * 2)
.entrySize(ENTRY_SIZE / 6)
.putReturnsNull(true)
.createPersistedTo(file);
{
warmUpCompression(ENTRY_SIZE);
int threads = 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;
futureList.add(es.submit(new Runnable() {
@Override
public void run() {
exerciseLargeStrings(map, 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 start, int finish, int entrySize) {
/*
final Thread thisThread = Thread.currentThread();
Thread monitor = new Thread(new Runnable() {
@@ -110,23 +172,7 @@ public class LargeEntriesTest {
});
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();
}
String value = generateValue(entrySize);
for (int i = start; i < finish; i++) {
String key = "key-" + i;
@@ -151,4 +197,16 @@ public class LargeEntriesTest {
assertEquals(key, entrySize, object.length());
}
}
private String generateValue(int entrySize) {
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();
return baos.toString().substring(0, entrySize);
}
}