Divided integration and benchmark scripts

tools_scripts/benchmark/benchmarkReportGenerator.groovy

@@ -0,0 +1,65 @@
+/**
+ * Script used to reconstruct statistics
+ */
+ 
+Map populateWithTimings(reportFile, timingsMap) {
+    reportFile.eachLine {String line ->
+       if (line.startsWith("msg ID")) {
+          return
+       }
+       def tab = line.split(",")
+       int msgId = tab[0].trim() as int
+       long time = tab[1].trim() as long
+       
+       //TODO notify an alert in case of collisions
+       timingsMap[msgId] = time
+    }
+    timingsMap
+}
+
+Map calculateTransportTime(prodMap, consMap) {
+    Map transportMap = [:]
+    consMap.each {msgId, arrivalTime ->
+        long time = arrivalTime - prodMap[msgId]
+        transportMap[msgId] = time
+    }
+    transportMap
+}
+ 
+Map consumerTimings = [:]
+println "Load consumer receive timings"
+def consumerReport = new File("/home/andrea/workspace/moquette/consumer_bechmark.txt");
+populateWithTimings(consumerReport, consumerTimings)
+
+println "consumer timings loaded ${consumerTimings.size()} records"
+
+Map producerTimings = [:]
+def producerReport1 = new File("/home/andrea/workspace/moquette/producer_bechmark_Prod1.txt");
+def producerReport2 = new File("/home/andrea/workspace/moquette/producer_bechmark_Prod2.txt");
+populateWithTimings(producerReport1, producerTimings)
+println "producer 1 timings loaded ${producerTimings.size()} records"
+
+populateWithTimings(producerReport2, producerTimings)
+println "producer 2 timings loaded ${producerTimings.size()} records"
+
+Map report = calculateTransportTime(producerTimings, consumerTimings)
+File reportFile = new File("/home/andrea/workspace/moquette/report.txt")
+
+def minTime = report.min {it.value}.value / (10 ** 6)
+def maxTime = report.max {it.value}.value / (10 ** 6)
+def totalTime = 0
+
+report = report.sort()
+
+report.each {msgID, time ->
+   totalTime += time
+   reportFile <<  "$msgID, $time\n" 
+}
+
+def meanTime = (totalTime / report.size() )  / (10 ** 6)
+reportFile << "min $minTime \n max $maxTime \n mean $meanTime \n"
+
+println "created report file $reportFile"
+//reportFile.close()
+
+

tools_scripts/benchmark/benchmarker.groovy

@@ -0,0 +1,112 @@
+//@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-snapshots/')
+//@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.0.1-SNAPSHOT')
+@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/')
+@Grab(group='org.eclipse.paho', module='mqtt-client', version='0.4.0')
+
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient
+import org.eclipse.paho.client.mqttv3.MqttMessage
+import org.eclipse.paho.client.mqttv3.MqttException
+import org.eclipse.paho.client.mqttv3.MqttCallback
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
+import java.util.concurrent.CountDownLatch
+import static groovy.json.JsonOutput.*
+import java.nio.ByteBuffer
+
+
+class SubscriberCallback implements MqttCallback {
+
+    int m_numReceived = 0
+    long m_startTime
+    boolean firstMessageReceived = false
+    
+    private CountDownLatch m_latch = new CountDownLatch(1)
+
+    //Maps delay to num of messages reached with that delay
+    protected Map<Long, Long> distribution = [:]
+    
+    void waitFinish() {
+        m_latch.await()
+    }
+
+    public void connectionLost(Throwable cause) {
+        m_latch.countDown()
+    }
+
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+        //println "Received message ${new String(message.payload)} on topic [${topic}]"
+        if (!firstMessageReceived) {
+            m_startTime = System.currentTimeMillis()
+            firstMessageReceived = true
+        }
+        if (topic == "/exit") {
+            long stopTime = System.currentTimeMillis()
+            long spentTime = stopTime - m_startTime
+            println "/exit received"
+            println "subscriber disconnected, received ${m_numReceived} messages in ${spentTime} ms (from first received to last one on topic /exit)"
+            client.disconnect()
+            m_latch.countDown()
+        } else {
+            ByteBuffer bb = ByteBuffer.wrap(message.payload)
+            long sentTime = bb.getLong()
+            long delay = System.currentTimeMillis() - sentTime
+            //println "-received in ${delay}(ms) on ${topic} with QoS ${message.qos}"
+            //distribution[delay] = distribution[delay] != null ? distribution[delay]++ : 1
+            if (distribution[delay] != null) {
+                distribution[delay]++
+            } else {
+                distribution[delay] = 1
+            }
+            m_numReceived++
+        }
+    }
+
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+}
+
+
+def publishQos0(client, long numMessages) {
+    print "publishing.."
+    long startTime = System.currentTimeMillis()
+    (1..numMessages).each {
+        byte[] bytes = ByteBuffer.allocate(8).putLong(System.currentTimeMillis()).array();
+        client.publish('topic', bytes, 0, false)
+    }
+    client.publish('/exit', 'Exit'.bytes, 0, false)
+    long stopTime = System.currentTimeMillis()
+    long spentTime = stopTime -startTime
+    println "published in ${spentTime} ms"
+}
+
+if (args.size() < 2) {
+    println "Usage benchmarker <host> <numMessages>"
+    return
+}
+
+String host = args[0]
+long numMessages = args[1] as long
+String tmpDir = System.getProperty("java.io.tmpdir")
+MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir)
+
+int rnd = (Math.random() * 100) as int
+MqttAsyncClient client = new MqttAsyncClient("tcp://${host}:1883", "SubscriberClient${rnd}", dataStore)
+def callback = new SubscriberCallback()
+client.callback = callback
+client.connect().waitForCompletion(1000);
+client.subscribe("topic", 0)
+println "subscribed to topic"
+client.subscribe("/exit", 0)
+println "subscribed to /exit"
+
+publishQos0(client, numMessages)
+
+callback.waitFinish()
+Map sorted = callback.distribution.sort( { k1, k2 -> k1 <=> k2 } as Comparator)
+//println prettyPrint(toJson(sorted))
+File csv = new File("perf.csv")
+sorted.each {delay, count -> csv << "${delay}, ${count}\r\n"}
+
+
+

tools_scripts/benchmark/fuse_asynch_publisher.groovy

@@ -0,0 +1,111 @@
+@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10')
+
+import java.net.URISyntaxException
+
+import org.fusesource.mqtt.client.BlockingConnection
+import org.fusesource.mqtt.client.CallbackConnection
+import org.fusesource.mqtt.client.MQTT
+import org.fusesource.mqtt.client.Listener
+import org.fusesource.mqtt.client.Message
+import org.fusesource.mqtt.client.QoS
+import org.fusesource.mqtt.client.Topic
+import org.fusesource.mqtt.client.Callback
+import org.fusesource.hawtbuf.Buffer
+import org.fusesource.hawtbuf.UTF8Buffer
+import java.util.concurrent.CountDownLatch
+import org.fusesource.hawtdispatch.Task
+
+if (args.size() < 2) {
+    println "Usage publisher <host> <num messages to sent>"
+    return
+}
+
+String host = args[0]
+int numToSend = args[1] as int
+MQTT mqtt = new MQTT()
+//mqtt.setHost("test.mosquitto.org", 1883);
+mqtt.setHost(host, 1883)
+mqtt.setCleanSession(true)
+mqtt.setHost(host, 1883)
+mqtt.setClientId("PublisherClient")
+
+long startTime = System.currentTimeMillis()
+CountDownLatch m_latch = new CountDownLatch(1)
+
+final CallbackConnection connection = mqtt.callbackConnection()
+//connection.resume()
+connection.connect(new Callback<Void>() {
+    public void onFailure(Throwable th) {
+        println "connect fail"
+        System.exit(2);
+    }
+
+    // Once we connect..
+    public void onSuccess(Void v) {
+
+    }
+})
+
+String topic = "/topic"
+byte[] message = 'Hello world!!'.bytes
+QoS qos = QoS.AT_MOST_ONCE
+boolean retain = false
+
+print "publishing.."
+startTime = System.currentTimeMillis()
+
+new Task() {
+    long sent = 0;
+    public void run() {
+        final Task publish = this
+
+        connection.publish(topic, message, qos, retain, new Callback<Void>() {
+            public void onSuccess(Void value) {
+                sent ++;
+
+                if( sent < numToSend ) {
+                    connection.getDispatchQueue().execute(publish);
+                } else {
+                    connection.getDispatchQueue().execute(new Runnable() {
+                        public void run() {
+                            connection.publish("/exit", message, qos, retain, new Callback<Void>() {
+                                public void onSuccess(Void value2) {
+                                    long stopTime = System.currentTimeMillis()
+                                    long spentTime = stopTime -startTime
+                                    println "published in ${spentTime} ms"
+
+                                    connection.disconnect(null)
+                                    double msgPerSec = (numToSend / spentTime) * 1000
+                                    println "speed $msgPerSec msg/sec"
+
+                                    connection.disconnect(new Callback<Void>() {
+                                        public void onSuccess(Void valueSuccess2) {
+                                            m_latch.countDown();
+                                        }
+                                        public void onFailure(Throwable th2) {
+                                            m_latch.countDown();
+                                        }
+                                    });
+                                }
+
+                                public void onFailure(Throwable th) {
+                                    println "Publish failed: " + th
+
+                                    System.exit(2);
+                                }
+                            });
+                        }
+                    });
+
+                }
+            }
+            public void onFailure(Throwable th) {
+                println "Publish failed: " + th
+
+                System.exit(2);
+            }
+        });
+    }
+}.run();
+
+m_latch.await()

tools_scripts/benchmark/fuse_asynch_subscriber.groovy

@@ -0,0 +1,117 @@
+@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10')
+
+import org.fusesource.mqtt.client.BlockingConnection
+import org.fusesource.mqtt.client.CallbackConnection
+import org.fusesource.mqtt.client.MQTT
+import org.fusesource.mqtt.client.Listener
+import org.fusesource.mqtt.client.Message
+import org.fusesource.mqtt.client.QoS
+import org.fusesource.mqtt.client.Topic
+import org.fusesource.mqtt.client.Callback
+import org.fusesource.hawtbuf.Buffer
+import org.fusesource.hawtbuf.UTF8Buffer
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+if (args.size() != 1) {
+    println "Usage subscriber <host>"
+    return
+}
+
+String host = args[0]
+MQTT mqtt = new MQTT()
+//mqtt.setHost("test.mosquitto.org", 1883);
+mqtt.setHost(host, 1883)
+mqtt.setCleanSession(true)
+mqtt.setHost(host, 1883)
+mqtt.setClientId("SubscriberClient")
+
+int numReceived = 0
+long startTime = System.currentTimeMillis()
+boolean exit = false
+CountDownLatch m_latch = new CountDownLatch(1)
+
+final CallbackConnection connection = mqtt.callbackConnection()
+connection.listener(new Listener() {
+
+    public void onDisconnected() {
+        println "Listener received disconnection"
+    }
+    public void onConnected() {
+        println "Listener received a connected to ${host}"
+        startTime = System.currentTimeMillis()
+    }
+
+    public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {
+        // You can now process a received message from a topic.
+        // Once process execute the ack runnable.
+        ack.run()
+        //println "Received a publish on topic ${topic}"
+        if (topic.toString() == "/exit") {
+            println "Ok Exit!"
+            exit = true
+            connection.disconnect(null)
+            long stopTime = System.currentTimeMillis()
+            long spentTime = stopTime - startTime
+            println "/exit received"
+            println "subscriber disconnected, received ${numReceived} messages in ${spentTime} ms"
+            double msgPerSec = (numReceived / spentTime) * 1000
+            println "speed $msgPerSec msg/sec"
+            m_latch.countDown()
+        } else {
+            numReceived++
+//        if ((numReceived % 10000) == 0) {
+//            print '.'
+//        }
+        }
+
+    }
+    public void onFailure(Throwable value) {
+        connection.disconnect(null)
+        println "Some thing failed $value"
+        m_latch.countDown()
+    }
+})
+
+//Topic[] topics = [new Topic("/topic", QoS.AT_MOST_ONCE), new Topic("/exit", QoS.AT_MOST_ONCE)]
+//byte[] qoses = connection.subscribe(topics)
+//println "subscribed to /topic qos: ${qoses[0]}"
+//println "subscribed to /exit qos: ${qoses[1]}"
+
+connection.connect(new Callback<Void>() {
+    public void onFailure(Throwable value) {
+        System.exit(2);
+    }
+
+    // Once we connect..
+    public void onSuccess(Void v) {
+        // Subscribe to /topic
+        Topic[] topics = [new Topic("/topic", QoS.AT_LEAST_ONCE)]
+        connection.subscribe(topics, new Callback<byte[]>() {
+            public void onSuccess(byte[] qoses) {
+                // The result of the subcribe request.
+                println "subscribed to /topic ${qoses}"
+            }
+            public void onFailure(Throwable value) {
+                connection.disconnect(null); // subscribe failed.
+            }
+        });
+
+
+        // Subscribe to /topic
+        Topic[] exitTopic = [new Topic("/exit", QoS.AT_LEAST_ONCE)]
+        connection.subscribe(exitTopic, new Callback<byte[]>() {
+            public void onSuccess(byte[] qoses) {
+                // The result of the subcribe request.
+                println "subscribed to /exit ${qoses}"
+            }
+            public void onFailure(Throwable value) {
+                connection.disconnect(null); // subscribe failed.
+                m_latch.countDown()
+            }
+        });
+    }
+})
+
+
+m_latch.await()

tools_scripts/benchmark/fuse_publisher.groovy

@@ -0,0 +1,35 @@
+@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10')
+
+import java.net.URISyntaxException
+
+import org.fusesource.mqtt.client.BlockingConnection
+import org.fusesource.mqtt.client.MQTT
+import org.fusesource.mqtt.client.QoS
+
+if (args.size() < 2) {
+    println "Usage publisher <host> <num messages to sent>"
+    return
+}
+
+String host = args[0]
+int numToSend = args[1] as int
+
+MQTT mqtt = new MQTT()
+//mqtt.setHost("test.mosquitto.org", 1883);
+mqtt.setHost(host, 1883)
+BlockingConnection connection = mqtt.blockingConnection()
+connection.connect()
+
+print "publishing.."
+long startTime = System.currentTimeMillis()
+(1..numToSend).each {
+    connection.publish("/topic", 'Hello world!!'.bytes, QoS.AT_MOST_ONCE, false)
+}
+connection.publish('/exit', 'Exit'.bytes, QoS.AT_MOST_ONCE, false)
+long stopTime = System.currentTimeMillis()
+long spentTime = stopTime -startTime
+println "published in ${spentTime} ms"
+connection.disconnect()
+double msgPerSec = (numToSend / spentTime) * 1000
+println "speed $msgPerSec msg/sec"
+println "disconnected"

tools_scripts/benchmark/fuse_subscriber.groovy

@@ -0,0 +1,58 @@
+@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10')
+
+import java.net.URISyntaxException
+
+import org.fusesource.mqtt.client.BlockingConnection
+import org.fusesource.mqtt.client.MQTT
+import org.fusesource.mqtt.client.Message
+import org.fusesource.mqtt.client.QoS
+import org.fusesource.mqtt.client.Topic
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+if (args.size() != 1) {
+    println "Usage subscriber <host>"
+    return
+}
+
+String host = args[0]
+MQTT mqtt = new MQTT()
+//mqtt.setHost("test.mosquitto.org", 1883);
+mqtt.setHost(host, 1883)
+mqtt.setCleanSession(true)
+mqtt.setHost(host, 1883)
+
+mqtt.setClientId("SubscriberClient")
+BlockingConnection connection = mqtt.blockingConnection()
+connection.connect()
+println "Connected to ${host}"
+
+Topic[] topics = [new Topic("/topic", QoS.AT_MOST_ONCE), new Topic("/exit", QoS.AT_MOST_ONCE)]
+byte[] qoses = connection.subscribe(topics)
+println "subscribed to /topic qos: ${qoses[0]}"
+println "subscribed to /exit qos: ${qoses[1]}"
+
+int numReceived = 0
+long startTime = System.currentTimeMillis()
+boolean exit = false
+
+while(!exit) {
+    Message message = connection.receive()
+    if (message.topic == "/exit") {
+        exit = true
+    } else {
+        numReceived++
+//        if ((numReceived % 10000) == 0) {
+//            print '.'
+//        }
+    }
+    message.ack()
+}
+connection.disconnect()
+long stopTime = System.currentTimeMillis()
+long spentTime = stopTime - startTime
+println "/exit received"
+println "subscriber disconnected, received ${numReceived} messages in ${spentTime} ms"
+double msgPerSec = (numReceived / spentTime) * 1000
+println "speed $msgPerSec msg/sec"
+

tools_scripts/benchmark/publisher_client.groovy

@@ -0,0 +1,42 @@
+@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/')
+@Grab(group='org.eclipse.paho', module='mqtt-client', version='0.4.0')
+
+import org.eclipse.paho.client.mqttv3.MqttClient
+import org.eclipse.paho.client.mqttv3.MqttMessage
+import org.eclipse.paho.client.mqttv3.MqttException
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions
+
+if (args.size() < 2) {
+    println "Usage publisher <host> <num messages to sent>"
+    return
+}
+
+String host = args[0]
+int numToSend = args[1] as int
+
+String tmpDir = System.getProperty("java.io.tmpdir")
+MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir)
+
+MqttClient client = new MqttClient("tcp://${host}:1883", "PublisherClient", dataStore)
+//Only for Apollo ---
+//MqttConnectOptions connOptions = new MqttConnectOptions()
+//connOptions.userName ='admin'
+//connOptions.password = 'password' as char[]
+//Only for Apollo ---
+//client.connect(connOptions)
+client.connect(connOptions)
+MqttMessage message = new MqttMessage('Hello world!!'.bytes)
+message.setQos(0)
+print "publishing..${numToSend} messages, "
+long startTime = System.currentTimeMillis()
+//(1..1000000).each {
+(1..numToSend).each {
+    client.publish('topic', 'Hello world!!'.bytes, 0, false)
+}
+client.publish('/exit', 'Exit'.bytes, 0, false)
+long stopTime = System.currentTimeMillis()
+long spentTime = stopTime -startTime
+println "published in ${spentTime} ms"
+client.disconnect()
+println "disconnected"

tools_scripts/benchmark/subscriber_client.groovy

@@ -0,0 +1,81 @@
+@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/')
+@Grab(group='org.eclipse.paho', module='mqtt-client', version='0.4.0')
+
+import org.eclipse.paho.client.mqttv3.MqttClient
+import org.eclipse.paho.client.mqttv3.MqttMessage
+import org.eclipse.paho.client.mqttv3.MqttException
+import org.eclipse.paho.client.mqttv3.MqttCallback
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+
+class SubscriberCallback implements MqttCallback {
+
+    int m_numReceived = 0
+    long m_startTime
+    boolean firstMessageReceived = false
+    
+    private CountDownLatch m_latch = new CountDownLatch(1)
+    
+    void waitFinish() {
+        m_latch.await()
+    }
+
+    public void connectionLost(Throwable cause) {
+        m_latch.countDown()
+    }
+
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+        //println "Received message ${new String(message.payload)} on topic [${topic}]"
+        if (!firstMessageReceived) {
+            m_startTime = System.currentTimeMillis()
+            firstMessageReceived = true
+        }
+        if (topic == "/exit") {
+            long stopTime = System.currentTimeMillis()
+            long spentTime = stopTime - m_startTime
+            println "/exit received"
+            println "subscriber disconnected, received ${m_numReceived} messages in ${spentTime} ms"
+            client.disconnect()
+            m_latch.countDown()
+        } else {
+            //println "-received ${message} on ${topic} with QoS ${message.qos}"
+            m_numReceived++
+        }
+    }
+
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+}
+
+if (args.size() < 2) {
+    println "Usage subscriber <host> <QoS>"
+    return
+}
+
+String host = args[0]
+int qos = args[1] as int
+String tmpDir = System.getProperty("java.io.tmpdir")
+MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir)
+
+int rnd = (Math.random() * 100) as int
+MqttClient client = new MqttClient("tcp://${host}:1883", "SubscriberClient${rnd}", dataStore)
+def callback = new SubscriberCallback()
+client.callback = callback
+//Only for Apollo ---
+//MqttConnectOptions connOptions = new MqttConnectOptions()
+//connOptions.userName ='admin'
+//connOptions.password = 'password' as char[]
+//client.connect(connOptions)
+//Only for Apollo ---
+client.connect()
+client.subscribe("topic", 0)
+println "subscribed to topic"
+client.subscribe("/exit", 0)
+println "subscribed to /exit"
+
+callback.waitFinish()