From 0d81d29daee1d1ddc7cfa3a331c24c4bd546bf19 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 5 Dec 2014 17:54:01 +0200 Subject: [PATCH 1/2] Adding multiple clients to configurable test --- .../openhft/chronicle/map/TcpReplicator.java | 29 +- src/test/java/eg/BGChronicleTest.java | 273 +++++++++--------- .../map/example/DistributedSequenceMain.java | 32 +- 3 files changed, 188 insertions(+), 146 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 33e221791..b02c48739 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -449,6 +449,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable attached.entryReader = new TcpSocketChannelEntryReader(); attached.entryWriter = new TcpSocketChannelEntryWriter(); + attached.writeBarrier = true; key.interestOps(OP_WRITE | OP_READ); @@ -487,15 +488,16 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable channel.socket().setSoLinger(false, 0); final Attached attached = new Attached(); - channel.register(selector, OP_READ, attached); - - throttle(channel); - attached.entryReader = new TcpSocketChannelEntryReader(); attached.entryWriter = new TcpSocketChannelEntryWriter(); - attached.isServer = true; attached.entryWriter.identifierToBuffer(localIdentifier); + attached.isServer = true; + attached.writeBarrier = true; + + channel.register(selector, OP_READ, attached); + + throttle(channel); } /** @@ -622,17 +624,19 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable final long approxTime) throws InterruptedException, IOException { final SocketChannel socketChannel = (SocketChannel) key.channel(); final Attached attached = (Attached) key.attachment(); - - if (attached.entryWriter.isWorkIncomplete()) { - final boolean completed = attached.entryWriter.doWork(); + if (attached == null) throw new NullPointerException("No attached"); + TcpSocketChannelEntryWriter entryWriter = attached.entryWriter; + if (entryWriter == null) throw new NullPointerException("No entryWriter"); + if (entryWriter.isWorkIncomplete()) { + final boolean completed = entryWriter.doWork(); if (completed) - attached.entryWriter.workCompleted(); + entryWriter.workCompleted(); } else if (attached.remoteModificationIterator != null) - attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); + entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); try { - final int len = attached.entryWriter.writeBufferToSocket(socketChannel, + final int len = entryWriter.writeBufferToSocket(socketChannel, approxTime); if (len == -1) @@ -641,7 +645,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable if (len > 0) contemplateThrottleWrites(len); - if (!attached.entryWriter.hasBytesToWrite() && !attached.entryWriter.isWorkIncomplete()) + if (!entryWriter.hasBytesToWrite() && !entryWriter.isWorkIncomplete()) // TURN OP_WRITE_OFF key.interestOps(key.interestOps() & ~OP_WRITE); } catch (IOException e) { @@ -903,6 +907,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable public boolean isServer; // the frequency the remote node will send a heartbeat public boolean handShakingComplete; public long remoteHeartbeatInterval = heartBeatIntervalMillis; + public volatile boolean writeBarrier; boolean isHandShakingComplete() { return handShakingComplete; diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index df831891b..81a76a60a 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -5,86 +5,123 @@ import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import org.junit.Test; +import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; import java.util.Random; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; + +import static java.lang.System.nanoTime; +import static java.nio.ByteBuffer.wrap; public class BGChronicleTest { private static final int CHRONICLE_HEARTBEAT_SECONDS = 5; - public static final long MAX_NUMBER_OF_EATERIES = 3000000L; + public static final int CLIENTS = Integer.getInteger("clients", 10); + public static final long KEYS = Long.getLong("keys", 100000L); + public static final long MESSAGES = Long.getLong("messages", 200000L); public static final String DEFAULT_HOST = "localhost"; public static final int DEFAULT_PORT = 9050; - - Random rnd = new Random(System.currentTimeMillis()); - byte[] key = new byte[17]; - byte[] value = new byte[11]; + public static final int KEY_SIZE = Integer.getInteger("keySize", 16); + public static final int VALUE_SIZE = Integer.getInteger("valueSize", 16); + private volatile long start; public ChronicleMap startChronicleMapServer(int port) throws InterruptedException, IOException { TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(port) - .heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("serverMap") + .heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS) + .name("serverMap") .autoReconnectedUponDroppedConnection(true); - + File file = File.createTempFile("bg-test", ".deleteme"); + file.deleteOnExit(); return ChronicleMapBuilder.of(byte[].class, byte[].class) - .entries(MAX_NUMBER_OF_EATERIES) - .constantKeySizeBySample(key) - .constantValueSizeBySample(value) + .entries(KEYS) + .constantKeySizeBySample(new byte[KEY_SIZE]) + .constantValueSizeBySample(new byte[VALUE_SIZE]) .putReturnsNull(true) .removeReturnsNull(true) .replication((byte) 1, serverConfig) - .create(); + .createPersistedTo(file); } /** - * over loop-back on my mac I get around 50,000 messages per second of my local network ( LAN ), - * I get around 730 messages per second - * - * @param hostNameOrIpAddress - * @param port - * @param numberOfMessages - * @throws IOException + * over loop-back on my mac I get around 50,000 messages per second of my local network ( LAN ) */ - public void startChronicleMapClient(String hostNameOrIpAddress, final int port, final long - numberOfMessages) throws IOException { + public void startChronicleMapClient(String hostNameOrIpAddress, int port) throws IOException, InterruptedException { System.out.println("client starting"); + final InetSocketAddress remoteAddress = new InetSocketAddress(hostNameOrIpAddress, port); + start = nanoTime(); - try (ChronicleMap map = ChronicleMapBuilder.of(byte[].class, byte[].class) - .constantKeySizeBySample(key) - .constantValueSizeBySample(value) - .putReturnsNull(true) - .removeReturnsNull(true) - .statelessClient(new InetSocketAddress(hostNameOrIpAddress, port)).create()) { + ExecutorService es = Executors.newFixedThreadPool(CLIENTS); + List> futureList = new ArrayList<>(); + for (int i = 0; i < CLIENTS; i++) { + final int clientId = i; + futureList.add(es.submit(new Callable() { + @Override + public Void call() throws IOException { + try (ChronicleMap map = ChronicleMapBuilder + .of(byte[].class, byte[].class) + .constantKeySizeBySample(new byte[KEY_SIZE]) + .constantValueSizeBySample(new byte[VALUE_SIZE]) + .putReturnsNull(true) + .removeReturnsNull(true) + .statelessClient(remoteAddress) + .create()) { - System.out.println("client started"); + System.out.println(clientId + "/" + CLIENTS + " client started"); - long start = System.nanoTime(); - for (long count = 0; count < numberOfMessages; count++) { + exerciseMap(MESSAGES, map, clientId); + System.out.println(clientId + "/" + CLIENTS + " ... client finished"); + } + return null; + } + })); + } + for (Future future : futureList) { + try { + future.get(); + } catch (ExecutionException e) { + e.printStackTrace(); + } + } + es.shutdown(); + logMessagesPerSecond(MESSAGES, start); + } + + public void exerciseMap(long numberOfMessages, ChronicleMap map, int clientId) { + byte[] key = new byte[KEY_SIZE]; + byte[] value = new byte[VALUE_SIZE]; + Random rnd = new Random(); + ByteBuffer keyBB = wrap(key); + for (long count = 0; count < numberOfMessages; count += CLIENTS * 100) { + // KEYS/100 possible values after first byte. + keyBB.putInt(1, rnd.nextInt((int) (KEYS / 100))); + rnd.nextBytes(value); + for (byte b = 0; b < 100; b++) { + // 100 possible values for the first byte. + key[0] = b; if (rnd.nextBoolean()) { // puts - rnd.nextBytes(key); - rnd.nextBytes(value); + value[0] = b; map.put(key, value); } else { //gets - rnd.nextBytes(key); map.get(key); } - - if (count % 50000 == 0) { - logMessagesPerSecond(count, start); - } - } - logMessagesPerSecond(numberOfMessages, start); + if (clientId == 0 && count > 0 && count % 50000 == 0) { + logMessagesPerSecond(count, start); + } } } @@ -92,72 +129,57 @@ public class BGChronicleTest { /** * over loop-back on my mac I get around 50,000 messages per second of my local network ( LAN ), - * I get around 730 messages per second - * - * @param hostNameOrIpAddress - * @param port - * @param numberOfMessages - * @throws IOException */ - public void startReplicatingChronicleMapClient(String hostNameOrIpAddress, final int port, - final long numberOfMessages) throws - IOException { + public void startReplicatingChronicleMapClient(String hostNameOrIpAddress, final int port) + throws IOException, InterruptedException { - System.out.println("client starting"); + System.out.println("clients starting"); final InetSocketAddress serverHostPort = new InetSocketAddress(hostNameOrIpAddress, port); - TcpTransportAndNetworkConfig clientConfig = TcpTransportAndNetworkConfig.of(port + 1, serverHostPort) + final TcpTransportAndNetworkConfig clientConfig = TcpTransportAndNetworkConfig.of(port + 1, serverHostPort) .heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("clientMap") .autoReconnectedUponDroppedConnection(true); - try (ChronicleMap map = ChronicleMapBuilder.of(byte[].class, byte[].class) - .entries(MAX_NUMBER_OF_EATERIES) - .keySize(17) - .valueSize(11) - .putReturnsNull(true) - .removeReturnsNull(true) - .replication((byte) 2, clientConfig) - .create()) { + start = nanoTime(); + ExecutorService es = Executors.newFixedThreadPool(CLIENTS); + for (int i = 0; i < CLIENTS; i++) { + final int clientId = i; + es.submit(new Runnable() { + @Override + public void run() { + try (ChronicleMap map = ChronicleMapBuilder.of(byte[].class, byte[].class) + .entries(KEYS) + .constantKeySizeBySample(new byte[KEY_SIZE]) + .constantValueSizeBySample(new byte[VALUE_SIZE]) + .putReturnsNull(true) + .removeReturnsNull(true) + .replication((byte) (2 + clientId), clientConfig) + .create()) { - System.out.println("client started"); + System.out.println(clientId + "/" + CLIENTS + " client started"); - long start = System.nanoTime(); - for (long count = 0; count < numberOfMessages; count++) { - if (rnd.nextBoolean()) { - // puts - rnd.nextBytes(key); - rnd.nextBytes(value); - map.put(key, value); + exerciseMap(MESSAGES, map, clientId); + System.out.println(clientId + "/" + CLIENTS + " ... client finished"); + } - } else { - //gets - rnd.nextBytes(key); - map.get(key); } - - if (count % 50000 == 0) { - logMessagesPerSecond(count, start); - } - - } - - logMessagesPerSecond(numberOfMessages, start); - + }); } + es.awaitTermination(1, TimeUnit.HOURS); + logMessagesPerSecond(MESSAGES, start); } private void logMessagesPerSecond(double numberOfMessages, long start) { - double messagesPerSecond = numberOfMessages / (double) TimeUnit.NANOSECONDS - .toSeconds(System.nanoTime() - start); - System.out.println("messages per seconds " + messagesPerSecond); + long messagesPerSecond = (long) (numberOfMessages * 1e9 / (System.nanoTime() - start)); + System.out.printf("messages per seconds: %,d%n", messagesPerSecond); } /** * for loop back args = [] - * + *

* for server - eg.BGChronicleTest server for client - eg.BGChronicleTest client * * @param args @@ -167,60 +189,45 @@ public class BGChronicleTest { public static void main(String... args) throws IOException, InterruptedException { final BGChronicleTest o = new BGChronicleTest(); - ChronicleMap serverMap = null; - try { - - // loop-back - if (args.length == 0) { - serverMap = o.startChronicleMapServer(DEFAULT_PORT); - o.startChronicleMapClient(DEFAULT_HOST, DEFAULT_PORT, MAX_NUMBER_OF_EATERIES); - - } else { - - // server - if (args.length > 0 && args[0].equalsIgnoreCase("server")) { - int port = getPort(args, 1); - - serverMap = o.startChronicleMapServer(port); - System.out.println("server ON"); - } else - System.out.println("server if OFF"); - - // client - if (args.length == 0 || (args.length > 1 && args[0].equalsIgnoreCase("client"))) { - - // stateless replicating client - String host = getHost(args, 1); - int port = getPort(args, 2); - System.out.println("client ON...."); - o.startChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES); - - // replicating client - } else if (args.length > 1 && args[0].equalsIgnoreCase("replicatingClient")) { - String host = getHost(args, 1); - int port = getPort(args, 2); - System.out.println("client ON...."); - o.startReplicatingChronicleMapClient(host, port, MAX_NUMBER_OF_EATERIES); - - } else { - System.out.println("client if OFF"); - for (; ; ) { - Thread.sleep(1000); - } - } - - System.out.println("client if OFF"); - for (; ; ) { - Thread.sleep(1000); - } + // loop-back + if (args.length == 0) { + try (ChronicleMap serverMap = o.startChronicleMapServer(DEFAULT_PORT)) { + o.startChronicleMapClient(DEFAULT_HOST, DEFAULT_PORT); } - } finally { - if (serverMap != null) - serverMap.close(); - } + } else switch (args[0].toLowerCase()) { + case "server": { + int port = getPort(args, 1); + o.startChronicleMapServer(port); + System.out.println("server ON"); + + break; + } + + case "client": { + // stateless replicating client + String host = getHost(args, 1); + int port = getPort(args, 2); + System.out.println("client ON...."); + o.startChronicleMapClient(host, port); + break; + } + + case "replicatingClient": { + String host = getHost(args, 1); + int port = getPort(args, 2); + System.out.println("client ON...."); + o.startReplicatingChronicleMapClient(host, port); + break; + } + default: + System.err.println("Usage: java " + BGChronicleTest.class.getName() + " server {port}"); + System.err.println("\tjava " + BGChronicleTest.class.getName() + " client {hostname} {port}"); + System.err.println("\tjava " + BGChronicleTest.class.getName() + " replicatingClient {hostname} {port}"); + break; + } } private static String getHost(String[] args, int index) { diff --git a/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java b/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java index aa1b9b7ec..ba2530470 100644 --- a/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java +++ b/src/test/java/net/openhft/chronicle/map/example/DistributedSequenceMain.java @@ -25,6 +25,7 @@ import net.openhft.lang.values.LongValue; import java.io.File; import java.io.IOException; +import java.util.Arrays; public class DistributedSequenceMain { @@ -35,18 +36,47 @@ public class DistributedSequenceMain { .entries(128) .actualSegments(1) .createPersistedTo(file)) { + // throughput test. for (int t = 0; t < 5; t++) { LongValue value = DataValueClasses.newDirectReference(LongValue.class); map.acquireUsing("sequence-" + t, value); long start = System.nanoTime(); - int runs = 10000000; + int runs = 1000000; for (int i = 0; i < runs; i++) { long nextId = value.addAtomicValue(1); } double rate = (double) (System.nanoTime() - start) / runs; System.out.printf("Uncontended increment took an average of %.1f ns.%n", rate); } + + // latency test. + LongValue value = DataValueClasses.newDirectReference(LongValue.class); + map.acquireUsing("sequence-X", value); + + int runs = 1000000000, count = 0, wcount = 0; + long[] worst = new long[200000]; + long start = System.nanoTime(); + for (; count < runs; count++) { + long nextId = value.addAtomicValue(1); + long next = System.nanoTime(); + long time = next - start; + if (time > 100) { + worst[wcount++] = time; + if (wcount >= worst.length) + break; + next = System.nanoTime(); + } + start = next; + } + Arrays.sort(worst); + System.out.printf("Worst 1 in 10K, 1 in 100K, 1 in 1000K, worst is %,d / %,d / %,d / %,d %n", + worst[wcount - count / 10000], + worst[wcount - count / 100000], + worst[wcount - count / 1000000], + worst[wcount - 1] + ); } } } + From f02be23f09d8432089f9ca7387152e245ba98024 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 5 Dec 2014 17:56:07 +0200 Subject: [PATCH 2/2] Adding multiple clients to configurable test --- src/main/java/net/openhft/chronicle/map/TcpReplicator.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index b02c48739..bb8e4d6e5 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -449,7 +449,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable attached.entryReader = new TcpSocketChannelEntryReader(); attached.entryWriter = new TcpSocketChannelEntryWriter(); - attached.writeBarrier = true; key.interestOps(OP_WRITE | OP_READ); @@ -493,7 +492,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable attached.entryWriter.identifierToBuffer(localIdentifier); attached.isServer = true; - attached.writeBarrier = true; channel.register(selector, OP_READ, attached); @@ -907,7 +905,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable public boolean isServer; // the frequency the remote node will send a heartbeat public boolean handShakingComplete; public long remoteHeartbeatInterval = heartBeatIntervalMillis; - public volatile boolean writeBarrier; boolean isHandShakingComplete() { return handShakingComplete;