Merge pull request #12 from dist-sys/refactor_v0.6.2

Refactor v0.6.2

Author: nabon

build.gradle

@@ -4,7 +4,7 @@ plugins {
 }
 
 mainClassName = 'mqttloader.Loader'
-applicationDefaultJvmArgs = ['-Djava.util.logging.config.file=logging.properties', '-Dfile.encoding=UTF-8']
+applicationDefaultJvmArgs = ['-Djava.util.logging.config.file=src/dist/logging.properties', '-Dfile.encoding=UTF-8', '--illegal-access=deny']
 
 repositories {
     mavenCentral()
@@ -20,3 +20,13 @@ dependencies {
 run {
     args '-h'.split('\\s+')
 }
+
+CreateStartScripts startScripts = project.startScripts
+startScripts.with {
+    doLast {
+        unixScript.text = unixScript.text.replaceFirst('-Djava.util.logging.config.file=src/dist/logging.properties',
+                '-Djava.util.logging.config.file=\\$APP_HOME/logging.properties')
+        windowsScript.text = windowsScript.text.replaceFirst('-Djava.util.logging.config.file=src/dist/logging.properties',
+                '-Djava.util.logging.config.file=%~dp0../logging.properties')
+    }
+}

doc/usage_en.md

@@ -47,7 +47,10 @@ For example, the following command uses a public MQTT broker provided by HiveMQ.
 `$ ./mqttloader -b tcp://broker.hivemq.com:1883 -p 1 -s 1 -m 10`
 
 ### Run on multiple machines
-You can run MQTTLoader on multiple machines, e.g., run publishers and subscribers separately on different machines.  
+You can run MQTTLoader on multiple machines.  
+Running both publishers and subscribers on a single machine may cause mutual influence, e.g., the subscribers' receiving load lowers the publishers' throughput.  
+By running publishers and subscribers separately on different machines, you can avoid such mutual influence.  
+
 For example, on a host A, you can run MQTTLoader as follows:
 
 `$ ./mqttloader -b tcp://<IP>:<PORT> -p 0 -s 1 -st 20 -n <NTP-SERVER>`
@@ -98,6 +101,9 @@ MQTTLoader starts to terminate when all of the following conditions are met.
 MQTTLoader also starts to terminate when the time specified by the parameter `-et` elapses, even if there are in-flight messages.  
 Thus, `-et` should be long sufficiently.
 
+If you want to do measurement with fixed time period, you can set the measurement time by the parameter `-et`.  
+Note that you need to set sufficiently large value to the parameter `-m`.
+
 By setting the parameter `-n`, MQTTLoader obtains the offset time from the specified NTP server and reflects it to calculate throughput and latency.  
 It might be useful for running multiple MQTTLoader on different machines.  
 
@@ -121,6 +127,8 @@ Maximum latency[ms]: 81
 Average latency[ms]: 42.23691
 ```
 For each publisher, MQTTLoader counts the number of messages sent for each second.  
+If QoS level is set to 1 or 2, counting is done when receiving PUBACK or PUBCOMP respectively.
+
 After completion, MQTTLoader collects the counted numbers from all publishers and calculates the maximum throughput, the average throughput, and the number of published messages.  
 `Throughput[msg/s]` is the list of throughputs, which are the sum of each second for all publishers.  
 Note that these calculation exclude the beginning and trailing seconds that have 0 messages.  
@@ -159,6 +167,8 @@ Thus, when running multiple MQTTLoader on different machines (e.g., publishers o
 By specifying the file name with `-tf` parameter, you can obtain throughput data like the following.
 
 ```
+Measurement start time: 2020-09-01 18:33:38.122 JST
+Measurement end time: 2020-09-01 18:33:54.104 JST
 SLOT, mqttloaderclient-pub000000, mqttloaderclient-sub000000
 0, 11955, 11218
 1, 16427, 16414
@@ -173,6 +183,8 @@ The data that used to calculate the summary data in the standard output is writt
 By specifying the file name with `-lf` parameter, you can obtain latency data like the following.
 
 ```
+Measurement start time: 2020-09-01 18:33:38.122 JST
+Measurement end time: 2020-09-01 18:33:54.104 JST
 mqttloaderclient-sub000000, mqttloaderclient-sub000001
 7, 7
 4, 4

doc/usage_jp.md

@@ -48,7 +48,10 @@ MQTTLoaderの動作を確認するだけなら、パブリックブローカを
 `$ ./mqttloader -b tcp://broker.hivemq.com:1883 -p 1 -s 1 -m 10`
 
 ### 複数台での実行
-publisherとsubscriberを別マシンで動かす、等、複数台のマシン上でMQTTLoaderを動かすこともできます。  
+複数台のマシン上でMQTTLoaderを動かすこともできます。  
+1台のマシン上でpublisherとsubscriberを動かした場合、subscriberの受信負荷によってpublisherの送信スループットが低下する等の可能性があります。  
+publisherとsubscriberを別マシンで動かすことで、負荷が相互に影響することを避けることができます。
+
 例えば、ホストA上で以下のように実行します。
 
 `$ ./mqttloader -b tcp://<IP>:<PORT> -p 0 -s 1 -st 20 -n <NTP-SERVER>`
@@ -99,6 +102,8 @@ MQTTLoaderは、以下の条件をすべて満たすと、クライアントを
 また、MQTTLoaderは、パラメータ`-et`によって指定される時間が経過すると、メッセージ送受信中であっても、終了します。  
 送受信を中断したくない場合は、`-et`は長めに設定しておくと良いでしょう。
 
+一定時間の測定を行いたい場合には、`-et`を用いて測定時間を設定し、`-m`で十分に大きな値を設定します。
+
 パラメータ`-n`を設定すると、MQTTLoaderは指定されたNTPサーバから時刻のオフセット情報（NTPサーバ時刻からのずれ）を取得し、スループットやレイテンシの計算にそれを反映します。  
 複数のMQTTLoaderを異なるマシン上で実行する場合に、利用を検討してください。
 
@@ -123,6 +128,8 @@ Average latency[ms]: 42.23691
 ```
 
 MQTTLoaderは、各publisherごとに、毎秒の送信メッセージ数をカウントします。  
+QoSレベルが1または2の場合は、それぞれ、PUBACKおよびPUBCOMPを受信したタイミングでカウントされます。
+
 全てのメッセージ送信が完了したら、MQTTLoaderは全publisherからカウントしたメッセージ数の情報を集めて集計し、最大スループット、平均スループット、送信メッセージ数を計算します。  
 `Throughput[msg/s]`の項は、スループット値の列挙です。列挙されているそれぞれの値は、各秒における全publisherの送信メッセージ数を足し合わせたものです。  
 なお、測定開始時および終了時に送信メッセージ数が0の期間がある場合は、スループットの計算からは除外されます。  
@@ -162,6 +169,8 @@ subscriberに関しても、上記と同様にして、受信メッセージの
 パラメータ`-tf`でファイル名を指定することで、以下のようなスループットの詳細データをファイルに書き出すことができます。
 
 ```
+Measurement start time: 2020-09-01 18:33:38.122 JST
+Measurement end time: 2020-09-01 18:33:44.104 JST
 SLOT, mqttloaderclient-pub000000, mqttloaderclient-sub000000
 0, 11955, 11218
 1, 16427, 16414
@@ -176,6 +185,8 @@ SLOT, mqttloaderclient-pub000000, mqttloaderclient-sub000000
 パラメータ`-lf`でファイル名を指定することで、以下のようなレイテンシの詳細データをファイルに書き出すことができます。
 
 ```
+Measurement start time: 2020-09-01 18:33:38.122 JST
+Measurement end time: 2020-09-01 18:33:44.104 JST
 mqttloaderclient-sub000000, mqttloaderclient-sub000001
 7, 7
 4, 4

logging.properties src/dist/logging.propertieslogging.properties renamed to src/dist/logging.properties

@@ -22,7 +22,9 @@
 import java.net.InetAddress;
 import java.net.SocketException;
 import java.net.UnknownHostException;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.Iterator;
 import java.util.Timer;
 import java.util.TreeMap;
@@ -52,10 +54,12 @@ public class Loader {
     private ArrayList<IClient> publishers = new ArrayList<>();
     private ArrayList<IClient> subscribers = new ArrayList<>();
     public static volatile long startTime;
+    private long endTime;
     public static volatile long offset = 0;
     public static volatile long lastRecvTime;
     public static CountDownLatch countDownLatch;
     public static Logger logger = Logger.getLogger(Loader.class.getName());
+    private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS z");
 
     public Loader(String[] args) {
         setOptions(args);
@@ -91,17 +95,27 @@ public Loader(String[] args) {
         }
 
         int execTime = Integer.valueOf(cmd.getOptionValue(Opt.EXEC_TIME.getName(), Opt.EXEC_TIME.getDefaultValue()));
+        long holdTime = startTime - Util.getTime();
+        if(holdTime > 0) execTime += (int)holdTime;
         try {
             countDownLatch.await(execTime, TimeUnit.SECONDS);
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
 
+        if(countDownLatch.getCount()>0) {
+            logger.info("Measurement timed out.");
+        } else {
+            logger.info("Measurement completed.");
+        }
+
         timer.cancel();
 
         logger.info("Terminating clients.");
         disconnectClients();
 
+        endTime = Util.getTime();
+
         logger.info("Printing results.");
         dataCleansing();
 
@@ -164,13 +178,22 @@ private void prepareClients() {
         int pubInterval = Integer.valueOf(cmd.getOptionValue(Opt.INTERVAL.getName(), Opt.INTERVAL.getDefaultValue()));
 
         for(int i=0;i<numPub;i++){
+            if(i == 0) {
+                logger.info("Publishers start to connect.");
+            }
+
             if(version==5){
                 publishers.add(new Publisher(i, broker, pubQos, retain, topic, payloadSize, numMessage, pubInterval));
             }else{
                 publishers.add(new PublisherV3(i, broker, pubQos, retain, topic, payloadSize, numMessage, pubInterval));
             }
         }
+
         for(int i=0;i<numSub;i++){
+            if(i == 0) {
+                logger.info("Subscribers start to connect.");
+            }
+
             if(version==5){
                 subscribers.add(new Subscriber(i, broker, subQos, shSub, topic));
             }else{
@@ -179,11 +202,15 @@ private void prepareClients() {
         }
     }
 
+    /**
+     * Start measurement by running publishers.
+     */
     private void startMeasurement() {
         String ntpServer = cmd.getOptionValue(Opt.NTP.getName(), Opt.NTP.getDefaultValue());
         if(ntpServer != null) {
             logger.info("Getting time information from NTP server.");
             NTPUDPClient client = new NTPUDPClient();
+            client.setDefaultTimeout(5000);
             InetAddress address = null;
             TimeInfo ti = null;
             try {
@@ -197,9 +224,14 @@ private void startMeasurement() {
             } catch (IOException e) {
                 e.printStackTrace();
             }
-            ti.computeDetails();
-            offset = ti.getOffset();
-            logger.info("Offset is "+offset+" milliseconds.");
+
+            if(ti != null) {
+                ti.computeDetails();
+                offset = ti.getOffset();
+                logger.info("Offset is "+offset+" milliseconds.");
+            } else {
+                logger.warning("Failed to get time information from NTP server.");
+            }
         }
 
         // delay: Give ScheduledExecutorService time to setup scheduling.
@@ -212,11 +244,20 @@ private void startMeasurement() {
     }
 
     private void disconnectClients() {
-        for(IClient pub: publishers){
-            pub.disconnect();
+        for(int i=0;i<publishers.size();i++){
+            if(i == 0) {
+                logger.info("Publishers start to disconnect.");
+            }
+
+            publishers.get(i).disconnect();
         }
-        for(IClient sub: subscribers){
-            sub.disconnect();
+
+        for(int i=0;i<subscribers.size();i++){
+            if(i == 0) {
+                logger.info("Subscribers start to disconnect.");
+            }
+
+            subscribers.get(i).disconnect();
         }
     }
 
@@ -366,6 +407,11 @@ private void printLatency() {
     private void thToFile(){
         StringBuilder sb = new StringBuilder();
 
+        String sTime = sdf.format(new Date(startTime));
+        String eTime = sdf.format(new Date(endTime));
+        sb.append("Measurement start time: "+sTime+"\n");
+        sb.append("Measurement end time: "+eTime+"\n");
+
         sb.append("SLOT");
         for(int i=0;i<publishers.size();i++){
             sb.append(", "+publishers.get(i).getClientId());
@@ -425,6 +471,11 @@ private void thToFile(){
     private void ltToFile(){
         StringBuilder sb = new StringBuilder();
 
+        String sTime = sdf.format(new Date(startTime));
+        String eTime = sdf.format(new Date(endTime));
+        sb.append("Measurement start time: "+sTime+"\n");
+        sb.append("Measurement end time: "+eTime+"\n");
+
         for(int i=0;i<subscribers.size();i++){
             if(i>0) sb.append(", ");
             sb.append(subscribers.get(i).getClientId());

src/main/java/mqttloader/Loader.java

@@ -27,7 +27,7 @@ public String format(LogRecord record) {
         return String.format(
                 "[%s] %s %s#%s %s%n",
                 record.getLevel().getName(),
-                new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(record.getMillis()),
+                new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS z").format(record.getMillis()),
                 record.getSourceClassName(),
                 record.getSourceMethodName(),
                 record.getMessage()

src/main/java/mqttloader/LogFormatter.java

@@ -35,6 +35,7 @@ public RecvTimeoutTask(Timer timer, int subTimeout) {
     public void run() {
         long remainingTime = subTimeout*1000 - (Util.getTime() - lastRecvTime);  // <timeout> - <elapsed time>
         if (remainingTime <= 0) {
+            Loader.logger.info("Receiving messages on subscribers timed out.");
             countDownLatch.countDown();
         } else {
             timer.schedule(new RecvTimeoutTask(timer, subTimeout), remainingTime);

src/main/java/mqttloader/RecvTimeoutTask.java

@@ -65,6 +65,7 @@ public Publisher(int clientNumber, String broker, int qos, boolean retain, Strin
             client.connect(options);
             Loader.logger.info("Publisher client is connected: "+clientId);
         } catch (MqttException e) {
+            Loader.logger.warning("Publisher client fails to connect: "+clientId);
             e.printStackTrace();
         }
     }
@@ -81,34 +82,41 @@ public void start(long delay) {
 
     @Override
     public void run() {
-        if(!client.isConnected()) {
-            Loader.countDownLatch.countDown();
-        } else {
-            if(pubInterval==0){
-                continuousRun();
-            }else{
-                periodicalRun();
-            }
+        if(pubInterval==0){
+            continuousRun();
+        }else{
+            periodicalRun();
         }
     }
 
     public void continuousRun() {
         for(int i=0;i<numMessage;i++){
             if(cancelled) {
+                Loader.logger.info("Publish task is cancelled: "+clientId);
                 break;
             }
-            publish();
+            if(client.isConnected()) {
+                publish();
+            } else {
+                Loader.logger.warning("On sending publish, client was not connected: "+clientId);
+            }
         }
 
+        Loader.logger.info("Publisher finishes to send publish: "+clientId);
         Loader.countDownLatch.countDown();
     }
 
     public void periodicalRun() {
         if(numMessage > 0) {
-            publish();
+            if(client.isConnected()) {
+                publish();
+            } else {
+                Loader.logger.warning("On sending publish, client was not connected: "+clientId);
+            }
 
             numMessage--;
             if(numMessage==0){
+                Loader.logger.info("Publisher finishes to send publish: "+clientId);
                 Loader.countDownLatch.countDown();
             }
         }
@@ -119,6 +127,7 @@ public void publish() {
         try{
             client.publish(topic, message);
         } catch(MqttException me) {
+            Loader.logger.warning("On sending publish, MqttException occurred: "+clientId);
             me.printStackTrace();
         }
 
@@ -151,10 +160,13 @@ public void disconnect() {
             e.printStackTrace();
         }
 
-        try {
-            client.disconnect();
-        } catch (MqttException e) {
-            e.printStackTrace();
+        if (client.isConnected()) {
+            try {
+                client.disconnect();
+                Loader.logger.info("Publisher client is disconnected: "+clientId);
+            } catch (MqttException e) {
+                e.printStackTrace();
+            }
         }
     }