Working on IO loop tests.
diff --git a/utils/nio/src/test/java/org/onlab/nio/IOLoopTestClient.java b/utils/nio/src/test/java/org/onlab/nio/IOLoopTestClient.java
index 9942740..6f6bd6d 100644
--- a/utils/nio/src/test/java/org/onlab/nio/IOLoopTestClient.java
+++ b/utils/nio/src/test/java/org/onlab/nio/IOLoopTestClient.java
@@ -1,5 +1,6 @@
package org.onlab.nio;
+import com.google.common.collect.Lists;
import org.onlab.util.Counter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -23,6 +24,7 @@
import java.util.concurrent.TimeoutException;
import static java.lang.String.format;
+import static java.lang.System.currentTimeMillis;
import static java.lang.System.out;
import static org.onlab.nio.IOLoopTestServer.PORT;
import static org.onlab.util.Tools.delay;
@@ -46,15 +48,18 @@
Counter messages;
Counter bytes;
+ long latencyTotal = 0;
+ long latencyCount = 0;
+
/**
* Main entry point to launch the client.
*
* @param args command-line arguments
- * @throws IOException if unable to connect to server
- * @throws InterruptedException if latch wait gets interrupted
- * @throws ExecutionException if wait gets interrupted
- * @throws TimeoutException if timeout occurred while waiting for completion
+ * @throws java.io.IOException if unable to connect to server
+ * @throws InterruptedException if latch wait gets interrupted
+ * @throws java.util.concurrent.ExecutionException if wait gets interrupted
+ * @throws java.util.concurrent.TimeoutException if timeout occurred while waiting for completion
*/
public static void main(String[] args)
throws IOException, InterruptedException, ExecutionException, TimeoutException {
@@ -95,7 +100,7 @@
* @param mc message count to send per client
* @param ml message length in bytes
* @param port socket port
- * @throws IOException if unable to create IO loops
+ * @throws java.io.IOException if unable to create IO loops
*/
public IOLoopTestClient(InetAddress ip, int wc, int mc, int ml, int port) throws IOException {
this.ip = ip;
@@ -113,7 +118,7 @@
/**
* Starts the client workers.
*
- * @throws IOException if unable to open connection
+ * @throws java.io.IOException if unable to open connection
*/
public void start() throws IOException {
messages = new Counter();
@@ -141,7 +146,7 @@
* channel with the given IO loop.
*
* @param loop loop with which the channel should be registered
- * @throws IOException if the socket could not be open or connected
+ * @throws java.io.IOException if the socket could not be open or connected
*/
private void openConnection(CustomIOLoop loop) throws IOException {
SocketAddress sa = new InetSocketAddress(ip, port);
@@ -156,15 +161,17 @@
* Waits for the client workers to complete.
*
* @param secs timeout in seconds
- * @throws ExecutionException if execution failed
- * @throws InterruptedException if interrupt occurred while waiting
- * @throws TimeoutException if timeout occurred
+ * @throws java.util.concurrent.ExecutionException if execution failed
+ * @throws InterruptedException if interrupt occurred while waiting
+ * @throws java.util.concurrent.TimeoutException if timeout occurred
*/
public void await(int secs) throws InterruptedException,
ExecutionException, TimeoutException {
for (CustomIOLoop l : iloops) {
if (l.worker.task != null) {
l.worker.task.get(secs, TimeUnit.SECONDS);
+ latencyTotal += l.latencyTotal;
+ latencyCount += l.latencyCount;
}
}
messages.freeze();
@@ -176,10 +183,11 @@
*/
public void report() {
DecimalFormat f = new DecimalFormat("#,##0");
- out.println(format("Client: %s messages; %s bytes; %s mps; %s Mbs",
+ out.println(format("Client: %s messages; %s bytes; %s mps; %s Mbs; %s ms latency",
f.format(messages.total()), f.format(bytes.total()),
f.format(messages.throughput()),
- f.format(bytes.throughput() / (1024 * msgLength))));
+ f.format(bytes.throughput() / (1024 * msgLength)),
+ f.format(latencyTotal / latencyCount)));
}
@@ -187,6 +195,9 @@
private class CustomIOLoop extends IOLoop<TestMessage, TestMessageStream> {
Worker worker = new Worker();
+ long latencyTotal = 0;
+ long latencyCount = 0;
+
public CustomIOLoop() throws IOException {
super(500);
@@ -217,7 +228,12 @@
@Override
protected void processMessages(List<TestMessage> messages,
- MessageStream<TestMessage> b) {
+ MessageStream<TestMessage> stream) {
+ for (TestMessage message : messages) {
+ // TODO: summarize latency data better
+ latencyTotal += currentTimeMillis() - message.requestorTime();
+ latencyCount++;
+ }
worker.release(messages.size());
}
@@ -239,15 +255,15 @@
private static final int BATCH_SIZE = 1000;
private static final int PERMITS = 2 * BATCH_SIZE;
- private TestMessageStream b;
+ private TestMessageStream stream;
private FutureTask<Worker> task;
// Stuff to throttle pump
private final Semaphore semaphore = new Semaphore(PERMITS);
private int msgWritten;
- void pump(TestMessageStream b) {
- this.b = b;
+ void pump(TestMessageStream stream) {
+ this.stream = stream;
task = new FutureTask<>(this, this);
wpool.execute(task);
}
@@ -257,18 +273,15 @@
try {
log.info("Worker started...");
- List<TestMessage> batch = new ArrayList<>();
- for (int i = 0; i < BATCH_SIZE; i++) {
- batch.add(new TestMessage(msgLength));
- }
-
while (msgWritten < msgCount) {
- msgWritten += writeBatch(b, batch);
+ int size = Math.min(BATCH_SIZE, msgCount - msgWritten);
+ writeBatch(size);
+ msgWritten += size;
}
// Now try to get all the permits back before sending poison pill
semaphore.acquireUninterruptibly(PERMITS);
- b.close();
+ stream.close();
log.info("Worker done...");
@@ -278,18 +291,15 @@
}
- private int writeBatch(TestMessageStream b, List<TestMessage> batch)
- throws IOException {
- int count = Math.min(BATCH_SIZE, msgCount - msgWritten);
- acquire(count);
- if (count == BATCH_SIZE) {
- b.write(batch);
- } else {
- for (int i = 0; i < count; i++) {
- b.write(batch.get(i));
- }
+ private void writeBatch(int size) throws IOException {
+ // Build a batch of messages
+ List<TestMessage> batch = Lists.newArrayListWithCapacity(size);
+ for (int i = 0; i < size; i++) {
+ batch.add(new TestMessage(msgLength, currentTimeMillis(), 0,
+ stream.padding()));
}
- return count;
+ acquire(size);
+ stream.write(batch);
}