Merge pull request #31 from dist-sys/refactor_v0.8.4

Refactor v0.8.4

Author: nabon

build.gradle

@@ -4,7 +4,7 @@ plugins {
 }
 
 wrapper {
-    gradleVersion = '7.3.3'
+    gradleVersion = '7.6'
 }
 
 mainClassName = 'mqttloader.Loader'

doc/usage_en.md

@@ -1,4 +1,4 @@
-# MQTTLoader usage (v0.8.4)
+# MQTTLoader usage (v0.8.5)
 MQTTLoader is a load testing tool (client tool) for MQTT.  
 It supports both MQTT v5.0 and v3.1.1.  
 From v0.8.0, it supports TLS authentication.
@@ -13,7 +13,7 @@ Download the archive file (zip or tar) from: https://github.com/dist-sys/mqttloa
 Below is an example of downloading by using curl command.
 
 ```
-$ curl -OL https://github.com/dist-sys/mqttloader/releases/download/v0.8.4/mqttloader-0.8.4.zip
+$ curl -OL https://github.com/dist-sys/mqttloader/releases/download/v0.8.5/mqttloader-0.8.5.zip
 ```
 
 By extracting it, you can get the following files.
@@ -70,7 +70,7 @@ The following table shows the parameters which can be set in *mqttloader.conf*.
 |:-----------|:------------:|:------------|:------------|
 | broker | Yes | (none) | Broker's IP address or FQDN. <br>Ex. `broker = 127.0.0.1` |
 | broker_port | No | 1883 (non-TLS)<br>8883 (TLS) | Broker's port number. <br>Ex. `broker_port = 1883` |
-| mqtt_version | No | 5 | MQTT version. `3` for MQTT v3.1.1, and `5` for MQTT v5.0. |
+| mqtt_version | No | 5 | MQTT version. 3 for MQTT v3.1.1, and 5 for MQTT v5.0. |
 | num_publishers | No | 1 | The number of publishers. All publishers send messages to a same topic. |
 | num_subscribers | No | 1 | The number of subscribers. All subscribers are subscribe to a same topic. |
 | qos_publisher | No | 0 | QoS level of publishers. <br>Valid values are 0/1/2. |
@@ -82,10 +82,10 @@ The following table shows the parameters which can be set in *mqttloader.conf*.
 | num_messages | No | 100 | The number of messages sent by **each** publisher. |
 | ramp_up | No | 0 | Ramp-up time in seconds. <br>See **4. How to read the results** for details. |
 | ramp_down | No | 0 | Ramp-down time in seconds. <br>See **4. How to read the results** for details. |
-| interval | No | 0 | Publish interval in microseconds. |
+| interval | No | 0 | Publish interval in microseconds.<br>Regardless of num_publishers, each publisher sends messages at this interval. In addition, the gap between multiple publishers' sending is adjusted to be interval/num_publishers as far as possible. |
 | subscriber_timeout | No | 5 | Timeout for receiving messages by subscribers in seconds. |
 | exec_time | No | 60 | Maximum execution time for measurement in seconds. |
-| log_level | No | INFO | Log level. <br>Valid values are `SEVERE`/`WARNING`/`INFO`/`ALL`. |
+| log_level | No | INFO | Log level. <br>Valid values are SEVERE/WARNING/INFO/ALL. |
 | ntp | No | (none) | NTP server's IP address or FQDN. By setting this, throughput and latency are calculated based on the NTP server's time.<br>It should be set when running multiple MQTTLoader on different machines.<br>Ex. `ntp = ntp.nict.jp` |
 | output <sup>**\*1 \*2**</sup> | No | (none) | Directory path to write out measurement record. If not set, MQTTLoader runs by in-memory mode. <br>Ex. `output = /home/testuser` |
 | user_name | No | (none) | User name. Required if the broker has the configuration of password authentication. |
@@ -213,7 +213,7 @@ If QoS level is set to 1 or 2, counting is done when receiving PUBACK or PUBCOMP
 After completion, MQTTLoader calculates the maximum throughput, the average throughput, and the number of published messages.  
 `Per second throughput [msg/s]` is the time series of throughputs per second.  
 
-By using the parameterse `ramp_up` and `ramp_down`, you can exclude the beginning and trailing data.  
+By using the parameters `ramp_up` and `ramp_down`, you can exclude the beginning and trailing data.  
 If you set the following parameter settings for example, the beginning one second and the trailing one second are excluded.
 
 ```
@@ -239,6 +239,7 @@ Note that if the specified directory doesn't exist, it is newly created.
 The file `mqttloader_xxxxxxxx-xxxxxx.csv` has records like the following:
 
 ```
+1599643916401359,,,
 1599643916416823,ml-EeiE-p-00001,S,
 1599643916416882,ml-EeiE-p-00000,S,
 1599643916419123,ml-EeiE-s-00000,R,3165
@@ -248,6 +249,7 @@ The file `mqttloader_xxxxxxxx-xxxxxx.csv` has records like the following:
 ```
 
 Each line, consists of comma-separeted values, indicates the following data.  
+The first line indicates the measurement start time.  
 In the case that the event type is `R`, latency data follows.
 
 ```
@@ -256,6 +258,7 @@ timestamp (Unix time in microseconds), client ID, event type (S: send, R: receiv
 
 Although MQTTLoader outputs the measurement result to the console, you can use the above .csv file for further analysis.  
 Note that the latency in the above file is in microseconds, whereas that in the console is in milliseconds with three digits after the decimal point.  
+the parameters `ramp_up` and `ramp_down` do not affect this file.  
 
 ---
 ---

doc/usage_jp.md

@@ -1,4 +1,4 @@
-# MQTTLoader 利用方法 (v0.8.4)
+# MQTTLoader 利用方法 (v0.8.5)
 MQTTLoaderは、MQTT v5.0とv3.1.1に対応した負荷テストツール（クライアントツール）です。  
 v0.8.0から、ブローカとのTLS接続にも対応しました。
 
@@ -14,7 +14,7 @@ https://github.com/dist-sys/mqttloader/releases
 以下は、Curlコマンドを使ってダウンロードする場合の例です。
 
 ```
-$ curl -OL https://github.com/dist-sys/mqttloader/releases/download/v0.8.4/mqttloader-0.8.4.zip
+$ curl -OL https://github.com/dist-sys/mqttloader/releases/download/v0.8.5/mqttloader-0.8.5.zip
 ```
 
 ダウンロードしたファイルを解凍すると、以下のディレクトリ構造が得られます。
@@ -70,27 +70,27 @@ broker_port = 1883
 |:-----------|:------------:|:------------|:------------|
 | broker | ○ | (無し) | ブローカのIPアドレスまたはFQDN。 <br>例： `broker = 127.0.0.1` |
 | broker_port | × | 1883 (non-TLS)<br>8883 (TLS) | ブローカのポート番号。 <br>例： `broker_port = 1883` |
-| mqtt_version | × | 5 | MQTTバージョン。 `3` を指定するとMQTT v3.1.1、`5` を指定するとMQTT v5.0。 |
+| mqtt_version | × | 5 | MQTTバージョン。 3 を指定するとMQTT v3.1.1、5 を指定するとMQTT v5.0。 |
 | num_publishers | × | 1 | publisher数。全publisherは同じトピックにメッセージを送信。 |
 | num_subscribers | × | 1 | subscriber数。全subscriberは同じトピックをsubscribe。 |
 | qos_publisher | × | 0 | publisherのQoSレベル。<br>設定可能な値：0/1/2 |
 | qos_subscriber | × | 0 | subscriberのQoSレベル。<br>設定可能な値：0/1/2 |
-| shared_subscription | × | false | Shared subscriptionの有効/無効を指定するフラグ。指定可能な値は `true` / `false` 。MQTT v5.0でのみ設定可。<br>有効にすると、各メッセージは全subscriberのうちいずれかひとつに届く。<br>例： `shared_subscription = true` |
-| retain | × | false | Retainの有効/無効を指定するフラグ。指定可能な値は `true` / `false` 。 |
+| shared_subscription | × | false | Shared subscriptionの有効/無効を指定するフラグ。指定可能な値は true/false 。MQTT v5.0でのみ設定可。<br>有効にすると、各メッセージは全subscriberのうちいずれかひとつに届く。<br>例： `shared_subscription = true` |
+| retain | × | false | Retainの有効/無効を指定するフラグ。指定可能な値は true/false 。 |
 | topic | × | mqttloader-test-topic | 測定で用いられるトピック名。 |
 | payload | × | 20 | publisherが送信するメッセージのペイロードサイズ。単位はbyte。設定可能な最小値は8。 |
 | num_messages | × | 100 | **各**publisherによって送信されるメッセージの数。 |
 | ramp_up | × | 0 | ランプアップ時間。単位は秒。<br>詳細は **4. 測定結果の見方** を参照。 |
 | ramp_down | × | 0 | ランプダウン時間。単位は秒。<br>詳細は **4. 測定結果の見方** を参照。 |
-| interval | × | 0 | 各publisherがメッセージを送信する間隔。単位はマイクロ秒。 |
+| interval | × | 0 | 各publisherがメッセージを送信する間隔。単位はマイクロ秒。<br>num_publishersの値とは関係なく、各publisherはそれぞれがintervalおきに送信する。また、複数publisher同士の送信タイミングは、可能な範囲でinterval/num_publishersだけずれるように調整される。 |
 | subscriber_timeout | × | 5 | subscriberの受信タイムアウト。単位は秒。 |
 | exec_time | × | 60 | 測定の実行時間上限。単位は秒。 |
-| log_level | × | INFO | ログレベル。<br>設定可能な値：`SEVERE`/`WARNING`/`INFO`/`ALL` |
+| log_level | × | INFO | ログレベル。<br>設定可能な値：SEVERE/WARNING/INFO/ALL |
 | ntp | × | (無し) | NTPサーバのIPアドレスまたはFQDN。設定すると、スループットやレイテンシの計算がNTPサーバ時刻を基準として行われる。<br>複数のMQTTLoaderを異なるマシン上で実行する場合には設定することが望ましい。<br>例：`ntp = ntp.nict.jp` |
 | output <sup>**※1※2**</sup> | × | (無し) | 測定レコードを書き出すディレクトリのパス。未指定の場合、MQTTLoaderはメモリ上でのみ動作。 <br>例： `output = /home/testuser` |
 | user_name | × | (無し) | ユーザ名（ブローカにてパスワード認証が設定されている場合に指定）。 |
 | password | × | (無し) | パスワード（ブローカにてパスワード認証が設定されている場合に指定）。 |
-| tls | × | false | TLS認証の有効/無効を指定するフラグ。指定可能な値は `true` / `false` 。 |
+| tls | × | false | TLS認証の有効/無効を指定するフラグ。指定可能な値は true/false 。 |
 | tls_rootca_cert <sup>**※1**</sup> | × | (無し) | TLSサーバ認証用のルートCA証明書（PEM形式）のパス。Java実行環境が参照する信頼済み証明書ストア（通常、Javaインストールディレクトリの `cacerts` ファイル）に登録済みのルートCAであれば、指定不要。<br>例： `tls_rootca_cert = /home/testuser/rootca.crt`  |
 | tls_client_key <sup>**※1**</sup> | × | (無し) | TLSクライアント認証用の秘密鍵（PEM形式）のパス。このパラメータを指定することで、TLSクライアント認証が有効になる。<br>例： `tls_client_key = /home/testuser/client.key` |
 | tls_client_cert_chain <sup>**※1**</sup> | × | (無し) | TLSクライアント認証用のクライアント証明書（PEM形式）のパス。中間CA証明書（PEM形式）のパスもセミコロン区切りで記載する。クライアント証明書、中間CA証明書、の順序で記載すること。ルートCA証明書は不要。（ファイル名・フォルダ名にセミコロンが含まれていると利用不可）<br>例： `tls_client_cert_chain = /home/testuser/client.crt;/home/testuser/ica.crt` |
@@ -240,6 +240,7 @@ subscriberに関しても、上記と同様にして、受信メッセージの
 このcsvファイルには、以下のようなデータが記録されます。
 
 ```
+1599643916401359,,,
 1599643916416823,ml-EeiE-p-00001,S,
 1599643916416882,ml-EeiE-p-00000,S,
 1599643916419123,ml-EeiE-s-00000,R,3165
@@ -249,14 +250,16 @@ subscriberに関しても、上記と同様にして、受信メッセージの
 ```
 
 各行は、カンマ区切りで、以下の内容となっています。  
-送受信種別が `R` の場合のみ、レイテンシも記載されます。
+先頭行には測定開始時刻のみが記載されます。  
+また、送受信種別が `R` の場合のみ、レイテンシも記載されます。
 
 ```
 タイムスタンプ（マイクロ秒単位Unix時間）, クライアントID, 送受信種別（S: 送信, R: 受信）, レイテンシ（マイクロ秒単位）
 ```
 
 MQTTLoaderは、測定結果のサマリをコンソールに出力しますが、追加の集計・分析を行いたい場合には上記のファイルを使ってください。  
 なお、コンソールに出力されるレイテンシはミリ秒単位（小数点以下3桁まで）であるのに対し、上記ファイルのレイテンシはマイクロ秒単位である点に注意してください。  
+また、パラメータ`ramp_up`と`ramp_down`はこのファイルには影響しません（すべての送受信記録が出力されます）。  
 
 ---
 ---

gradle/wrapper/gradle-wrapper.jar

@@ -1,5 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.3.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.6-bin.zip
+networkTimeout=10000
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

gradle/wrapper/gradle-wrapper.properties

@@ -1,7 +1,7 @@
 #!/bin/sh
 
 #
-# Copyright ? 2015-2021 the original authors.
+# Copyright © 2015-2021 the original authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -32,10 +32,10 @@
 #       Busybox and similar reduced shells will NOT work, because this script
 #       requires all of these POSIX shell features:
 #         * functions;
-#         * expansions á$varâ, á${var}â, á${var:-default}â, á${var+SET}â,
-#           á${var#prefix}â, á${var%suffix}â, and á$( cmd )â;
-#         * compound commands having a testable exit status, especially ácaseâ;
-#         * various built-in commands including ácommandâ, ásetâ, and áulimitâ.
+#         * expansions «$var», «${var}», «${var:-default}», «${var+SET}»,
+#           «${var#prefix}», «${var%suffix}», and «$( cmd )»;
+#         * compound commands having a testable exit status, especially «case»;
+#         * various built-in commands including «command», «set», and «ulimit».
 #
 #   Important for patching:
 #
@@ -55,7 +55,7 @@
 #       Darwin, MinGW, and NonStop.
 #
 #   (3) This script is generated from the Groovy template
-#       https://github.com/gradle/gradle/blob/master/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+#       https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
 #       within the Gradle project.
 #
 #       You can find Gradle at https://github.com/gradle/gradle/.
@@ -80,10 +80,10 @@ do
     esac
 done
 
-APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
-
-APP_NAME="Gradle"
+# This is normally unused
+# shellcheck disable=SC2034
 APP_BASE_NAME=${0##*/}
+APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
 
 # Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
 DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
@@ -143,12 +143,16 @@ fi
 if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
     case $MAX_FD in #(
       max*)
+        # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked.
+        # shellcheck disable=SC3045 
         MAX_FD=$( ulimit -H -n ) ||
             warn "Could not query maximum file descriptor limit"
     esac
     case $MAX_FD in  #(
       '' | soft) :;; #(
       *)
+        # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked.
+        # shellcheck disable=SC3045 
         ulimit -n "$MAX_FD" ||
             warn "Could not set maximum file descriptor limit to $MAX_FD"
     esac
@@ -205,6 +209,12 @@ set -- \
         org.gradle.wrapper.GradleWrapperMain \
         "$@"
 
+# Stop when "xargs" is not available.
+if ! command -v xargs >/dev/null 2>&1
+then
+    die "xargs is not available"
+fi
+
 # Use "xargs" to parse quoted args.
 #
 # With -n1 it outputs one arg per line, with the quotes and backslashes removed.

gradlew

@@ -14,7 +14,7 @@
 @rem limitations under the License.
 @rem
 
-@if "%DEBUG%" == "" @echo off
+@if "%DEBUG%"=="" @echo off
 @rem ##########################################################################
 @rem
 @rem  Gradle startup script for Windows
@@ -25,7 +25,8 @@
 if "%OS%"=="Windows_NT" setlocal
 
 set DIRNAME=%~dp0
-if "%DIRNAME%" == "" set DIRNAME=.
+if "%DIRNAME%"=="" set DIRNAME=.
+@rem This is normally unused
 set APP_BASE_NAME=%~n0
 set APP_HOME=%DIRNAME%
 
@@ -40,7 +41,7 @@ if defined JAVA_HOME goto findJavaFromJavaHome
 
 set JAVA_EXE=java.exe
 %JAVA_EXE% -version >NUL 2>&1
-if "%ERRORLEVEL%" == "0" goto execute
+if %ERRORLEVEL% equ 0 goto execute
 
 echo.
 echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
@@ -75,13 +76,15 @@ set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
 
 :end
 @rem End local scope for the variables with windows NT shell
-if "%ERRORLEVEL%"=="0" goto mainEnd
+if %ERRORLEVEL% equ 0 goto mainEnd
 
 :fail
 rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
 rem the _cmd.exe /c_ return code!
-if  not "" == "%GRADLE_EXIT_CONSOLE%" exit 1
-exit /b 1
+set EXIT_CODE=%ERRORLEVEL%
+if %EXIT_CODE% equ 0 set EXIT_CODE=1
+if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE%
+exit /b %EXIT_CODE%
 
 :mainEnd
 if "%OS%"=="Windows_NT" endlocal

gradlew.bat

@@ -19,7 +19,7 @@
 import java.text.SimpleDateFormat;
 
 public class Constants {
-    public static final String VERSION = "0.8.4";
+    public static final String VERSION = "0.8.5";
     public static final String BROKER_PREFIX_TCP = "tcp://";
     public static final String BROKER_PREFIX_TLS = "ssl://";
     public static final String BROKER_PORT_TCP = "1883";

src/main/java/mqttloader/Constants.java

@@ -430,8 +430,11 @@ private void startMeasurement() {
         measurementStartTime = Util.getCurrentTimeWithOffset().plusMillis(delay);
         lastRecvTime = measurementStartTime;
 
-        for(AbstractClient pub: publishers){
-            ((AbstractPublisher)pub).start(delay);
+		// publish interval in micro seconds.
+		int interval = Util.getPropValueInt(Prop.INTERVAL);
+		int numPub = publishers.size();
+		for(int i=0;i<numPub;i++){
+            ((AbstractPublisher)publishers.get(i)).start(delay*Constants.MILLISECOND_IN_MICRO+interval*i/numPub);
         }
     }
 
@@ -498,6 +501,9 @@ private void calcResult() {
                 while ((str = br.readLine()) != null) {
                     StringTokenizer st = new StringTokenizer(str, ",");
                     long timestamp = Long.valueOf(st.nextToken());
+					if(!st.hasMoreTokens()) {
+						continue; // skip the first line that only has a timestamp.
+					}
                     int elapsedTime = (int)((timestamp - Util.getEpochMicros(Loader.measurementStartTime))/Constants.SECOND_IN_MICRO);
                     st.nextToken(); //client ID
                     boolean isSend = st.nextToken().equals("S") ? true : false;

src/main/java/mqttloader/Loader.java

@@ -61,6 +61,15 @@ public Recorder(File file, boolean inMemory) {
     public void run() {
         thread = Thread.currentThread();
 
+		if(!inMemory) {
+			try {
+				bw.write(Util.getEpochMicros(Loader.measurementStartTime) + ",,,");
+				bw.newLine();
+			} catch (IOException e) {
+				e.printStackTrace();
+			}
+		}
+
         Record record = null;
         while (true) {
             try {