Read performance test specifications from a JSON file. We lose the ability to run multiple brokers, but that wasn't very well done anyway. We can come back to it.

Author: Simon MacMullen

test/src/com/rabbitmq/examples/perf/MulticastValue.java

@@ -16,11 +16,6 @@
 
 package com.rabbitmq.examples.perf;
 
-import java.beans.IntrospectionException;
-import java.beans.Introspector;
-import java.beans.PropertyDescriptor;
-import java.lang.reflect.InvocationTargetException;
-
 class MulticastValue implements VariableValue {
     private final String name;
     private final Object value;
@@ -32,32 +27,13 @@ class MulticastValue implements VariableValue {
 
     @Override
     public void setup(MulticastParams params) {
-        setValue(params, name, value);
+        PerfUtil.setValue(params, name, value);
     }
 
     @Override
     public void teardown(MulticastParams params) {
     }
 
-    private static void setValue(Object obj, String name, Object value) {
-        try {
-            PropertyDescriptor[] props = Introspector.getBeanInfo(obj.getClass()).getPropertyDescriptors();
-            for (PropertyDescriptor prop : props) {
-                if (prop.getName().equals(name)) {
-                    prop.getWriteMethod().invoke(obj, value);
-                    return;
-                }
-            }
-            throw new RuntimeException("Could not find property " + name + " in " + obj.getClass());
-        } catch (IntrospectionException e) {
-            throw new RuntimeException(e);
-        } catch (InvocationTargetException e) {
-            throw new RuntimeException(e);
-        } catch (IllegalAccessException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
     @Override
     public String getName() {
         return name;

test/src/com/rabbitmq/examples/perf/PerfUtil.java

@@ -0,0 +1,27 @@
+package com.rabbitmq.examples.perf;
+
+import java.beans.IntrospectionException;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.lang.reflect.InvocationTargetException;
+
+public class PerfUtil {
+    public static void setValue(Object obj, Object name, Object value) {
+        try {
+            PropertyDescriptor[] props = Introspector.getBeanInfo(obj.getClass()).getPropertyDescriptors();
+            for (PropertyDescriptor prop : props) {
+                if (prop.getName().equals(name)) {
+                    prop.getWriteMethod().invoke(obj, value);
+                    return;
+                }
+            }
+            throw new RuntimeException("Could not find property " + name + " in " + obj.getClass());
+        } catch (IntrospectionException e) {
+            throw new RuntimeException(e);
+        } catch (InvocationTargetException e) {
+            throw new RuntimeException(e);
+        } catch (IllegalAccessException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

test/src/com/rabbitmq/examples/perf/PerformanceMain.java

@@ -17,46 +17,62 @@
 package com.rabbitmq.examples.perf;
 
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.tools.json.JSONReader;
 import com.rabbitmq.tools.json.JSONWriter;
 
+import java.io.FileInputStream;
 import java.io.FileWriter;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.io.PrintWriter;
-import java.util.Arrays;
+import java.io.Reader;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 public class PerformanceMain {
     private static final ConnectionFactory factory = new ConnectionFactory();
 
-    //private static final List<?> NO_FLAGS = Arrays.asList();
-    private static final List<?> PERSISTENT = Arrays.asList("persistent");
-    private static final List<?> MANDATORY = Arrays.asList("mandatory");
-    private static final List<?> IMMEDIATE = Arrays.asList("immediate");
-
     private static Map<String, Object> results = new HashMap<String, Object>();
 
     public static void main(String[] args) throws Exception {
-        runStaticBrokerTests();
-        runTests(new Scenario[]{message_size_broker_config()});
-        writeJSON();
+        String inJSON = args[0];
+        String outJSON = args[1];
+        List<Map> scenariosJSON = (List<Map>) new JSONReader().read(readFile(inJSON));
+        Scenario[] scenarios = new Scenario[scenariosJSON.size()];
+        for (int i = 0; i < scenariosJSON.size(); i++) {
+            scenarios[i] = ScenarioFactory.fromJSON(scenariosJSON.get(i), factory);
+        }
+        runStaticBrokerTests(scenarios);
+        writeJSON(outJSON);
+    }
+
+    private static String readFile(String path) throws IOException {
+        final char[] buf = new char[4096];
+        StringBuilder out = new StringBuilder();
+        Reader in = new InputStreamReader(new FileInputStream(path), "UTF-8");
+        try {
+            int chars;
+            while ((chars = in.read(buf, 0, buf.length)) > 0) {
+                out.append(buf, 0, chars);
+            }
+        } finally {
+            in.close();
+        }
+        return out.toString();
     }
 
-    private static void writeJSON() throws IOException {
-        FileWriter outFile = new FileWriter("results.js");
+    private static void writeJSON(String outJSON) throws IOException {
+        FileWriter outFile = new FileWriter(outJSON);
         PrintWriter out = new PrintWriter(outFile);
         out.println(new JSONWriter(true).write(results));
         outFile.close();
     }
 
-    private static void runStaticBrokerTests() throws Exception {
+    private static void runStaticBrokerTests(Scenario[] scenarios) throws Exception {
         Broker broker = Broker.HIPE_COARSE;
         broker.start();
-        runTests(new Scenario[]{no_ack_long(), no_consume(), no_ack(), no_ack_mandatory(), no_ack_immediate(), ack(),
-                                ack_confirm(), ack_confirm_persist(), ack_persist(), fill_drain_queue("small", 500000),
-                                fill_drain_queue("large", 2000000), consumers(), headline_publish(), headline_consume(),
-                                message_sizes_small(), message_sizes_large(), message_size_vs_producers(), rate_vs_latency()});
+        runTests(scenarios);
         broker.stop();
     }
 
@@ -68,146 +84,4 @@ private static void runTests(Scenario[] scenarios) throws Exception {
             results.put(scenario.getName(), scenario.getStats().results());
         }
     }
-
-    private static Scenario no_ack_long() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setTimeLimit(500);
-        return new SimpleScenario("no-ack-long", factory, 10000, params);
-    }
-
-    private static Scenario headline_publish() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setProducerCount(10);
-        params.setConsumerCount(0);
-        return new SimpleScenario("headline-publish", factory, params);
-    }
-
-    private static Scenario headline_consume() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setProducerCount(1);
-        params.setConsumerCount(20);
-        return new SimpleScenario("headline-consume", factory, params);
-    }
-
-    private static Scenario no_consume() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setConsumerCount(0);
-        return new SimpleScenario("no-consume", factory, params);
-    }
-
-    private static Scenario no_ack() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        return new SimpleScenario("no-ack", factory, params);
-    }
-
-    private static Scenario no_ack_mandatory() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setFlags(MANDATORY);
-        return new SimpleScenario("no-ack-mandatory", factory, params);
-    }
-
-    private static Scenario no_ack_immediate() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setFlags(IMMEDIATE);
-        return new SimpleScenario("no-ack-immediate", factory, params);
-    }
-
-    private static Scenario ack() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setAutoAck(false);
-        return new SimpleScenario("ack", factory, params);
-    }
-
-    private static Scenario ack_confirm() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setAutoAck(false);
-        params.setConfirm(10000);
-        return new SimpleScenario("ack-confirm", factory, params);
-    }
-
-    private static Scenario ack_confirm_persist() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setAutoAck(false);
-        params.setConfirm(10000);
-        params.setFlags(PERSISTENT);
-        return new SimpleScenario("ack-confirm-persist", factory, params);
-    }
-
-    private static Scenario ack_persist() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setAutoAck(false);
-        params.setFlags(PERSISTENT);
-        return new SimpleScenario("ack-persist", factory, params);
-    }
-
-    private static Scenario fill_drain_queue(String name, int count) throws IOException, InterruptedException {
-        MulticastParams fill = fill_drain_params();
-        MulticastParams drain = fill_drain_params();
-
-        fill.setConsumerCount(0);
-        fill.setProducerMsgCount(count);
-        drain.setProducerCount(0);
-        drain.setConsumerMsgCount(count);
-
-        return new SimpleScenario("fill-drain-" + name + "-queue", factory, fill, drain);
-    }
-
-    private static MulticastParams fill_drain_params() {
-        MulticastParams params = new MulticastParams();
-        params.setQueueName("test");
-        params.setExclusive(false);
-        params.setAutoDelete(true);
-        return params;
-    }
-
-    private static Scenario message_sizes_small() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        return new VaryingScenario("message-sizes-small", factory, params,
-                    var("minMsgSize", 0, 100, 200, 300, 400, 500, 600, 700, 800, 900, 1000,
-                                      1200, 1400, 1600, 1800, 2000, 3000, 4000, 5000));
-    }
-
-    private static Scenario message_sizes_large() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        return new VaryingScenario("message-sizes-large", factory, params,
-                    var("minMsgSize", 5000, 10000, 20000, 50000, 100000, 500000, 1000000));
-    }
-
-    private static Scenario consumers() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setAutoAck(false);
-        return new VaryingScenario("consumers", factory, params,
-                    var("consumerCount", 1, 2, 5, 10, 50, 100, 500),
-                    var("prefetchCount", 1, 2, 5, 10, 20, 50, 10000));
-    }
-
-    private static Scenario message_size_vs_producers() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        params.setConsumerCount(0);
-        return new VaryingScenario("message-sizes-and-producers", factory, params,
-                    var("minMsgSize", 0, 1000, 10000, 100000),
-                    var("producerCount", 1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
-    }
-
-    private static Scenario message_size_broker_config() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        return new VaryingScenario("message-sizes-and-broker-config", factory, params,
-                    var("minMsgSize", 0, 1000, 2000, 5000),
-                    new BrokerVariable(Broker.DEFAULT, Broker.HIPE, Broker.COARSE, Broker.HIPE_COARSE));
-    }
-
-    private static Scenario rate_vs_latency() throws IOException, InterruptedException {
-        MulticastParams params = params();
-        return new RateVsLatencyScenario("rate-vs-latency", factory, params);
-    }
-
-    private static MulticastParams params() {
-        MulticastParams p = new MulticastParams();
-        p.setTimeLimit(30);
-        return p;
-    }
-
-    private static Variable var(String name, Object... values) {
-        return new MulticastVariable(name, values);
-    }
 }

test/src/com/rabbitmq/examples/perf/ScenarioFactory.java

@@ -0,0 +1,94 @@
+package com.rabbitmq.examples.perf;
+
+import com.rabbitmq.client.ConnectionFactory;
+
+import java.util.List;
+import java.util.Map;
+
+public class ScenarioFactory {
+    public static Scenario fromJSON(Map json, ConnectionFactory factory) {
+        String type = read("type", json, String.class);
+        String name = read("name", json, String.class);
+        Integer interval = read("interval", json, Integer.class, 1000);
+        List paramsJSON = read("params", json, List.class);
+
+        MulticastParams[] params = new MulticastParams[paramsJSON.size()];
+        for (int i = 0; i < paramsJSON.size(); i++) {
+            params[i] = paramsFromJSON((Map) paramsJSON.get(i));
+        }
+
+        if (type.equals("simple")) {
+            return new SimpleScenario(name, factory, interval, params);
+        }
+        else if (type.equals("rate-vs-latency")) {
+            return new RateVsLatencyScenario(name, factory, params[0]); // TODO
+        }
+        else if (type.equals("varying")) {
+            List variablesJSON = read("variables", json, List.class);
+            Variable[] variables = new Variable[variablesJSON.size()];
+            for (int i = 0; i < variablesJSON.size(); i++) {
+                variables[i] = variableFromJSON((Map) variablesJSON.get(i));
+            }
+
+            return new VaryingScenario(name, factory, params, variables);
+        }
+
+        throw new RuntimeException("Type " + type + " was not simple or varying.");
+    }
+
+    private static<T> T read(String key, Map map, Class<T> clazz) {
+        if (map.containsKey(key)) {
+            return read0(key, map, clazz);
+        }
+        else {
+            throw new RuntimeException("Key " + key + " not found.");
+        }
+    }
+
+    private static<T> T read(String key, Map map, Class<T> clazz, T def) {
+        if (map.containsKey(key)) {
+            return read0(key, map, clazz);
+        }
+        else {
+            return def;
+        }
+    }
+
+    private static <T> T read0(String key, Map map, Class<T> clazz) {
+        Object o = map.get(key);
+        if (clazz.isAssignableFrom(o.getClass())) {
+            return (T) o;
+        }
+        else {
+            throw new RuntimeException("Object under key " + key + " was a " + o.getClass() + ", not a " + clazz + ".");
+        }
+    }
+
+    private static MulticastParams paramsFromJSON(Map json) {
+        MulticastParams params = new MulticastParams();
+        for (Object key : json.keySet()) {
+            PerfUtil.setValue(params, hyphensToCamel((String)key), json.get(key));
+        }
+        return params;
+    }
+
+    private static Variable variableFromJSON(Map json) {
+        String type = read("type", json, String.class, "multicast");
+        String name = read("name", json, String.class);
+        Object[] values = read("values", json, List.class).toArray();
+
+        if (type.equals("multicast")) {
+            return new MulticastVariable(hyphensToCamel(name), values);
+        }
+
+        throw new RuntimeException("Type " + type + " was not multicast");
+    }
+
+    private static String hyphensToCamel(String name) {
+        String out = "";
+        for (String part : name.split("-")) {
+            out += part.substring(0, 1).toUpperCase() + part.substring(1);
+        }
+        return out.substring(0, 1).toLowerCase() + out.substring(1);
+    }
+}