Merge branch 'master' of ssh://gerrit.onlab.us:29418/onos-next

alshabib
Commit 6e4bf91ae264128a312b1a710e2da5245709342e 6e4bf91a 2 parents 339a3d92 22e0f2f9
Showing 11 changed files with 110 additions and 70 deletions
apps/foo/pom.xml
apps/foo/src/main/java/org/onlab/onos/foo/IOLoopTestClient.java
apps/foo/src/main/java/org/onlab/onos/foo/IOLoopTestServer.java
apps/foo/src/main/java/org/onlab/onos/foo/TestIOClientCommand.java
apps/foo/src/main/java/org/onlab/onos/foo/TestIOServerCommand.java
apps/foo/src/main/java/org/onlab/onos/foo/TestMessageStream.java
core/api/src/main/java/org/onlab/onos/cluster/MastershipTerm.java
pom.xml
utils/nio/src/main/java/org/onlab/nio/IOLoop.java
utils/nio/src/test/java/org/onlab/nio/IOLoopTestClient.java
utils/nio/src/test/java/org/onlab/nio/IOLoopTestServer.java
--- a/apps/foo/pom.xml
View file @6e4bf91
+++ b/apps/foo/pom.xml
View file @6e4bf91
@@ -28,6 +28,10 @@
             <version>${project.version}</version>
         </dependency>
         <dependency>
+            <groupId>org.livetribe.slp</groupId>
+            <artifactId>livetribe-slp</artifactId>
+        </dependency>
+        <dependency>
             <groupId>org.apache.karaf.shell</groupId>
             <artifactId>org.apache.karaf.shell.console</artifactId>
         </dependency>
--- a/apps/foo/src/main/java/org/onlab/onos/foo/IOLoopTestClient.java
View file @6e4bf91
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/IOLoopTestClient.java
View file @6e4bf91
@@ -26,7 +26,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import static java.lang.String.format;
-import static java.lang.System.currentTimeMillis;
+import static java.lang.System.nanoTime;
 import static java.lang.System.out;
 import static org.onlab.onos.foo.IOLoopTestServer.PORT;
 import static org.onlab.util.Tools.delay;
@@ -81,7 +81,7 @@ public class IOLoopTestClient {
         int wc = args.length > 1 ? Integer.parseInt(args[1]) : 6;
         int mc = args.length > 2 ? Integer.parseInt(args[2]) : 50 * 1000000;
         int ml = args.length > 3 ? Integer.parseInt(args[3]) : 128;
-        int to = args.length > 4 ? Integer.parseInt(args[4]) : 30;
+        int to = args.length > 4 ? Integer.parseInt(args[4]) : 60;
         log.info("Setting up client with {} workers sending {} {}-byte messages to {} server... ",
                  wc, mc, ml, ip);
@@ -185,7 +185,7 @@ public class IOLoopTestClient {
      */
     public void report() {
         DecimalFormat f = new DecimalFormat("#,##0");
-        out.println(format("Client: %s messages; %s bytes; %s mps; %s Mbs; %s ms latency",
+        out.println(format("Client: %s messages; %s bytes; %s mps; %s MBs; %s ns latency",
                            f.format(messages.total()), f.format(bytes.total()),
                            f.format(messages.throughput()),
                            f.format(bytes.throughput() / (1024 * msgLength)),
@@ -217,13 +217,6 @@ public class IOLoopTestClient {
             messages.add(stream.messagesIn().total());
             bytes.add(stream.bytesIn().total());
-
-//            out.println(format("Disconnected client; inbound %s mps, %s Mbps; outbound %s mps, %s Mbps",
-//                               FORMAT.format(stream.messagesIn().throughput()),
-//                               FORMAT.format(stream.bytesIn().throughput() / (1024 * msgLength)),
-//                               FORMAT.format(stream.messagesOut().throughput()),
-//                               FORMAT.format(stream.bytesOut().throughput() / (1024 * msgLength))));
-
             stream.messagesOut().reset();
             stream.bytesOut().reset();
         }
@@ -233,7 +226,7 @@ public class IOLoopTestClient {
                                        MessageStream<TestMessage> stream) {
             for (TestMessage message : messages) {
                 // TODO: summarize latency data better
-                latencyTotal += currentTimeMillis() - message.requestorTime();
+                latencyTotal += nanoTime() - message.requestorTime();
                 latencyCount++;
             }
             worker.release(messages.size());
@@ -254,7 +247,7 @@ public class IOLoopTestClient {
      */
     private class Worker implements Runnable {
-        private static final int BATCH_SIZE = 1000;
+        private static final int BATCH_SIZE = 10;
         private static final int PERMITS = 2 * BATCH_SIZE;
         private TestMessageStream stream;
@@ -297,8 +290,8 @@ public class IOLoopTestClient {
             // 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,
+                batch.add(new TestMessage(msgLength, nanoTime(), 0,
-                                          this.stream.padding(msgLength)));
+                                          stream.padding()));
             }
             acquire(size);
             stream.write(batch);
--- a/apps/foo/src/main/java/org/onlab/onos/foo/IOLoopTestServer.java
View file @6e4bf91
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/IOLoopTestServer.java
View file @6e4bf91
@@ -23,7 +23,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import static java.lang.String.format;
-import static java.lang.System.currentTimeMillis;
+import static java.lang.System.nanoTime;
 import static java.lang.System.out;
 import static org.onlab.util.Tools.delay;
 import static org.onlab.util.Tools.namedThreads;
@@ -85,11 +85,14 @@ public class IOLoopTestServer {
         IOLoopTestServer server = new IOLoopTestServer(ip, wc, ml, PORT);
         server.start();
-        // Start pruning clients.
+        // Start pruning clients and keep going until their number goes to 0.
-        while (true) {
+        int remaining = -1;
+        while (remaining == -1 || remaining > 0) {
             delay(PRUNE_FREQUENCY);
-            server.prune();
+            int r = server.prune();
+            remaining = remaining == -1 && r == 0 ? remaining : r;
         }
+        server.stop();
     }
     /**
@@ -153,7 +156,7 @@ public class IOLoopTestServer {
      */
     public void report() {
         DecimalFormat f = new DecimalFormat("#,##0");
-        out.println(format("Server: %s messages; %s bytes; %s mps; %s Mbs",
+        out.println(format("Server: %s messages; %s bytes; %s mps; %s MBs",
                            f.format(messages.total()), f.format(bytes.total()),
                            f.format(messages.throughput()),
                            f.format(bytes.throughput() / (1024 * msgLength))));
@@ -161,11 +164,15 @@ public class IOLoopTestServer {
     /**
      * Prunes the IO loops of stale message buffers.
+     *
+     * @return number of remaining IO loops among all workers.
      */
-    public void prune() {
+    public int prune() {
+        int count = 0;
         for (CustomIOLoop l : iloops) {
-            l.pruneStaleStreams();
+            count += l.pruneStaleStreams();
         }
+        return count;
     }
     // Get the next worker to which a client should be assigned
@@ -189,15 +196,8 @@ public class IOLoopTestServer {
         @Override
         protected void removeStream(MessageStream<TestMessage> stream) {
             super.removeStream(stream);
-
             messages.add(stream.messagesIn().total());
             bytes.add(stream.bytesIn().total());
-
-//            out.println(format("Disconnected server; inbound %s mps, %s Mbps; outbound %s mps, %s Mbps",
-//                               FORMAT.format(stream.messagesIn().throughput()),
-//                               FORMAT.format(stream.bytesIn().throughput() / (1024 * msgLength)),
-//                               FORMAT.format(stream.messagesOut().throughput()),
-//                               FORMAT.format(stream.bytesOut().throughput() / (1024 * msgLength))));
         }
         @Override
@@ -214,7 +214,7 @@ public class IOLoopTestServer {
             List<TestMessage> responses = Lists.newArrayListWithCapacity(messages.size());
             for (TestMessage message : messages) {
                 responses.add(new TestMessage(message.length(), message.requestorTime(),
-                                              currentTimeMillis(), message.padding()));
+                                              nanoTime(), message.padding()));
             }
             return responses;
         }
--- a/apps/foo/src/main/java/org/onlab/onos/foo/TestIOClientCommand.java
View file @6e4bf91
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/TestIOClientCommand.java
View file @6e4bf91
 package org.onlab.onos.foo;
+import org.apache.karaf.shell.commands.Argument;
 import org.apache.karaf.shell.commands.Command;
 import org.onlab.onos.cli.AbstractShellCommand;
@@ -12,12 +13,32 @@ import static org.onlab.onos.foo.IOLoopTestClient.startStandalone;
          description = "Starts the test IO loop client")
 public class TestIOClientCommand extends AbstractShellCommand {
+    @Argument(index = 0, name = "serverIp", description = "Server IP address",
+              required = false, multiValued = false)
+    String serverIp = "127.0.0.1";
+
+    @Argument(index = 1, name = "workers", description = "IO workers",
+              required = false, multiValued = false)
+    String workers = "6";
+
+    @Argument(index = 2, name = "messageCount", description = "Message count",
+              required = false, multiValued = false)
+    String messageCount = "1000000";
+
+    @Argument(index = 3, name = "messageLength", description = "Message length (bytes)",
+              required = false, multiValued = false)
+    String messageLength = "128";
+
+    @Argument(index = 4, name = "timeoutSecs", description = "Test timeout (seconds)",
+              required = false, multiValued = false)
+    String timeoutSecs = "60";
+
     @Override
     protected void execute() {
         try {
-            startStandalone(new String[]{});
+            startStandalone(new String[]{serverIp, workers, messageCount, messageLength, timeoutSecs});
         } catch (Exception e) {
-            error("Unable to start server %s", e);
+            error("Unable to start client %s", e);
         }
     }
--- a/apps/foo/src/main/java/org/onlab/onos/foo/TestIOServerCommand.java
View file @6e4bf91
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/TestIOServerCommand.java
View file @6e4bf91
 package org.onlab.onos.foo;
+import org.apache.karaf.shell.commands.Argument;
 import org.apache.karaf.shell.commands.Command;
 import org.onlab.onos.cli.AbstractShellCommand;
 import static org.onlab.onos.foo.IOLoopTestServer.startStandalone;
-
 /**
  * Starts the test IO loop server.
  */
@@ -13,10 +13,22 @@ import static org.onlab.onos.foo.IOLoopTestServer.startStandalone;
          description = "Starts the test IO loop server")
 public class TestIOServerCommand extends AbstractShellCommand {
+    @Argument(index = 0, name = "serverIp", description = "Server IP address",
+              required = false, multiValued = false)
+    String serverIp = "127.0.0.1";
+
+    @Argument(index = 1, name = "workers", description = "IO workers",
+              required = false, multiValued = false)
+    String workers = "6";
+
+    @Argument(index = 2, name = "messageLength", description = "Message length (bytes)",
+              required = false, multiValued = false)
+    String messageLength = "128";
+
     @Override
     protected void execute() {
         try {
-            startStandalone(new String[]{});
+            startStandalone(new String[]{serverIp, workers, messageLength});
         } catch (Exception e) {
             error("Unable to start server %s", e);
         }
--- a/apps/foo/src/main/java/org/onlab/onos/foo/TestMessageStream.java
View file @6e4bf91
+++ b/apps/foo/src/main/java/org/onlab/onos/foo/TestMessageStream.java
View file @6e4bf91
@@ -6,6 +6,7 @@ import org.onlab.nio.MessageStream;
 import java.nio.ByteBuffer;
 import java.nio.channels.ByteChannel;
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 /**
@@ -19,12 +20,18 @@ public class TestMessageStream extends MessageStream<TestMessage> {
     private static final int META_LENGTH = 40;
     private final int length;
+    private boolean isStrict = true;
     public TestMessageStream(int length, ByteChannel ch, IOLoop<TestMessage, ?> loop) {
         super(loop, ch, 64 * 1024, 500);
+        checkArgument(length >= META_LENGTH, "Length must be greater than header length of 40");
         this.length = length;
     }
+    void setNonStrict() {
+        isStrict = false;
+    }
+
     @Override
     protected TestMessage read(ByteBuffer rb) {
         if (rb.remaining() < length) {
@@ -32,16 +39,20 @@ public class TestMessageStream extends MessageStream<TestMessage> {
         }
         long startTag = rb.getLong();
-        checkState(startTag == START_TAG, "Incorrect message start");
+        if (isStrict) {
+            checkState(startTag == START_TAG, "Incorrect message start");
+        }
         long size = rb.getLong();
         long requestorTime = rb.getLong();
         long responderTime = rb.getLong();
-        byte[] padding = padding(length);
+        byte[] padding = padding();
         rb.get(padding);
         long endTag = rb.getLong();
-        checkState(endTag == END_TAG, "Incorrect message end");
+        if (isStrict) {
+            checkState(endTag == END_TAG, "Incorrect message end");
+        }
         return new TestMessage((int) size, requestorTime, responderTime, padding);
     }
@@ -60,7 +71,7 @@ public class TestMessageStream extends MessageStream<TestMessage> {
         wb.putLong(END_TAG);
     }
-    public byte[] padding(int msgLength) {
+    public byte[] padding() {
-        return new byte[msgLength - META_LENGTH];
+        return new byte[length - META_LENGTH];
     }
 }
--- a/core/api/src/main/java/org/onlab/onos/cluster/MastershipTerm.java
View file @6e4bf91
+++ b/core/api/src/main/java/org/onlab/onos/cluster/MastershipTerm.java
View file @6e4bf91
@@ -5,7 +5,7 @@ import java.util.Objects;
 public final class MastershipTerm {
     private final NodeId master;
-    private int termNumber;
+    private final int termNumber;
     private MastershipTerm(NodeId master, int term) {
         this.master = master;
--- a/pom.xml
View file @6e4bf91
+++ b/pom.xml
View file @6e4bf91
@@ -134,6 +134,12 @@
             </dependency>
             <dependency>
+                <groupId>org.livetribe.slp</groupId>
+                <artifactId>livetribe-slp</artifactId>
+                <version>2.2.1</version>
+            </dependency>
+
+            <dependency>
               <groupId>com.hazelcast</groupId>
               <artifactId>hazelcast</artifactId>
               <version>3.3</version>
--- a/utils/nio/src/main/java/org/onlab/nio/IOLoop.java
View file @6e4bf91
+++ b/utils/nio/src/main/java/org/onlab/nio/IOLoop.java
View file @6e4bf91
@@ -259,13 +259,16 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
     /**
      * Prunes the registered streams by discarding any stale ones.
+     *
+     * @return number of remaining streams
      */
-    public synchronized void pruneStaleStreams() {
+    public synchronized int pruneStaleStreams() {
         for (MessageStream<M> stream : streams) {
             if (stream.isStale()) {
                 stream.close();
             }
         }
+        return streams.size();
     }
 }
--- a/utils/nio/src/test/java/org/onlab/nio/IOLoopTestClient.java
View file @6e4bf91
+++ b/utils/nio/src/test/java/org/onlab/nio/IOLoopTestClient.java
View file @6e4bf91
@@ -24,7 +24,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import static java.lang.String.format;
-import static java.lang.System.currentTimeMillis;
+import static java.lang.System.nanoTime;
 import static java.lang.System.out;
 import static org.onlab.nio.IOLoopTestServer.PORT;
 import static org.onlab.util.Tools.delay;
@@ -79,7 +79,7 @@ public class IOLoopTestClient {
         int wc = args.length > 1 ? Integer.parseInt(args[1]) : 6;
         int mc = args.length > 2 ? Integer.parseInt(args[2]) : 50 * 1000000;
         int ml = args.length > 3 ? Integer.parseInt(args[3]) : 128;
-        int to = args.length > 4 ? Integer.parseInt(args[4]) : 30;
+        int to = args.length > 4 ? Integer.parseInt(args[4]) : 60;
         log.info("Setting up client with {} workers sending {} {}-byte messages to {} server... ",
                  wc, mc, ml, ip);
@@ -183,7 +183,7 @@ public class IOLoopTestClient {
      */
     public void report() {
         DecimalFormat f = new DecimalFormat("#,##0");
-        out.println(format("Client: %s messages; %s bytes; %s mps; %s Mbs; %s ms latency",
+        out.println(format("Client: %s messages; %s bytes; %s mps; %s MBs; %s ns latency",
                            f.format(messages.total()), f.format(bytes.total()),
                            f.format(messages.throughput()),
                            f.format(bytes.throughput() / (1024 * msgLength)),
@@ -212,16 +212,8 @@ public class IOLoopTestClient {
         @Override
         protected synchronized void removeStream(MessageStream<TestMessage> stream) {
             super.removeStream(stream);
-
             messages.add(stream.messagesIn().total());
             bytes.add(stream.bytesIn().total());
-
-//            out.println(format("Disconnected client; inbound %s mps, %s Mbps; outbound %s mps, %s Mbps",
-//                               FORMAT.format(stream.messagesIn().throughput()),
-//                               FORMAT.format(stream.bytesIn().throughput() / (1024 * msgLength)),
-//                               FORMAT.format(stream.messagesOut().throughput()),
-//                               FORMAT.format(stream.bytesOut().throughput() / (1024 * msgLength))));
-
             stream.messagesOut().reset();
             stream.bytesOut().reset();
         }
@@ -231,7 +223,7 @@ public class IOLoopTestClient {
                                        MessageStream<TestMessage> stream) {
             for (TestMessage message : messages) {
                 // TODO: summarize latency data better
-                latencyTotal += currentTimeMillis() - message.requestorTime();
+                latencyTotal += nanoTime() - message.requestorTime();
                 latencyCount++;
             }
             worker.release(messages.size());
@@ -252,7 +244,7 @@ public class IOLoopTestClient {
      */
     private class Worker implements Runnable {
-        private static final int BATCH_SIZE = 1000;
+        private static final int BATCH_SIZE = 50;
         private static final int PERMITS = 2 * BATCH_SIZE;
         private TestMessageStream stream;
@@ -295,8 +287,7 @@ public class IOLoopTestClient {
             // 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,
+                batch.add(new TestMessage(msgLength, nanoTime(), 0, stream.padding()));
-                                          stream.padding()));
             }
             acquire(size);
             stream.write(batch);
--- a/utils/nio/src/test/java/org/onlab/nio/IOLoopTestServer.java
View file @6e4bf91
+++ b/utils/nio/src/test/java/org/onlab/nio/IOLoopTestServer.java
View file @6e4bf91
@@ -20,7 +20,6 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import static java.lang.String.format;
-import static java.lang.System.currentTimeMillis;
 import static java.lang.System.out;
 import static org.onlab.util.Tools.delay;
 import static org.onlab.util.Tools.namedThreads;
@@ -82,11 +81,14 @@ public class IOLoopTestServer {
         IOLoopTestServer server = new IOLoopTestServer(ip, wc, ml, PORT);
         server.start();
-        // Start pruning clients.
+        // Start pruning clients and keep going until their number goes to 0.
-        while (true) {
+        int remaining = -1;
+        while (remaining == -1 || remaining > 0) {
             delay(PRUNE_FREQUENCY);
-            server.prune();
+            int r = server.prune();
+            remaining = remaining == -1 && r == 0 ? remaining : r;
         }
+        server.stop();
     }
     /**
@@ -150,7 +152,7 @@ public class IOLoopTestServer {
      */
     public void report() {
         DecimalFormat f = new DecimalFormat("#,##0");
-        out.println(format("Server: %s messages; %s bytes; %s mps; %s Mbs",
+        out.println(format("Server: %s messages; %s bytes; %s mps; %s MBs",
                            f.format(messages.total()), f.format(bytes.total()),
                            f.format(messages.throughput()),
                            f.format(bytes.throughput() / (1024 * msgLength))));
@@ -158,11 +160,15 @@ public class IOLoopTestServer {
     /**
      * Prunes the IO loops of stale message buffers.
+     *
+     * @return number of remaining IO loops among all workers.
      */
-    public void prune() {
+    public int prune() {
+        int count = 0;
         for (CustomIOLoop l : iloops) {
-            l.pruneStaleStreams();
+            count += l.pruneStaleStreams();
         }
+        return count;
     }
     // Get the next worker to which a client should be assigned
@@ -186,15 +192,8 @@ public class IOLoopTestServer {
         @Override
         protected void removeStream(MessageStream<TestMessage> stream) {
             super.removeStream(stream);
-
             messages.add(stream.messagesIn().total());
             bytes.add(stream.bytesIn().total());
-
-//            out.println(format("Disconnected server; inbound %s mps, %s Mbps; outbound %s mps, %s Mbps",
-//                               FORMAT.format(stream.messagesIn().throughput()),
-//                               FORMAT.format(stream.bytesIn().throughput() / (1024 * msgLength)),
-//                               FORMAT.format(stream.messagesOut().throughput()),
-//                               FORMAT.format(stream.bytesOut().throughput() / (1024 * msgLength))));
         }
         @Override
@@ -211,7 +210,7 @@ public class IOLoopTestServer {
             List<TestMessage> responses = Lists.newArrayListWithCapacity(messages.size());
             for (TestMessage message : messages) {
                 responses.add(new TestMessage(message.length(), message.requestorTime(),
-                                              currentTimeMillis(), message.padding()));
+                                              System.nanoTime(), message.padding()));
             }
             return responses;
         }