From 8e43f75f70804f3d8499bca8aa916a8d31fd6f30 Mon Sep 17 00:00:00 2001 From: andsel Date: Sat, 5 Nov 2016 18:16:59 +0100 Subject: [PATCH] Minor changes --- .../moquette/server/ConnectionDescriptor.java | 4 +- .../benchmark/conSubPubUnsub_issue204.groovy | 60 ++++++++++++------- .../benchmark/fuse_asynch_benchmarker.groovy | 4 +- 3 files changed, 41 insertions(+), 27 deletions(-) diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index e9026c6f..36c16768 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -18,8 +18,8 @@ package io.moquette.server; import io.netty.channel.Channel; /** - * Value object to maintain the information of single connection, like ClientID, IoSession, - * and other clean session fla. + * Value object to maintain the information of single connection, like ClientID, Channel, + * and clean session flag. * * * @author andrea diff --git a/tools_scripts/benchmark/conSubPubUnsub_issue204.groovy b/tools_scripts/benchmark/conSubPubUnsub_issue204.groovy index fc9f0c45..ed6e0d95 100644 --- a/tools_scripts/benchmark/conSubPubUnsub_issue204.groovy +++ b/tools_scripts/benchmark/conSubPubUnsub_issue204.groovy @@ -1,8 +1,11 @@ @GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/') -@Grab(group='org.eclipse.paho', module='mqtt-client', version='0.4.0') +//@Grab(group='org.eclipse.paho', module='mqtt-client', version='0.4.0') +@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.1.0') @Grab(group='org.eclipse.jetty.toolchain', module='jetty-perf-helper', version='1.0.5') +//@Grab(group='org.slf4j', module='slf4j-api', version='1.7.21') +//@Grab(group='org.slf4j', module='slf4j-jdk14', version='1.7.21') -import org.eclipse.paho.client.mqttv3.MqttAsyncClient +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 @@ -11,8 +14,9 @@ import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence import org.HdrHistogram.Histogram import org.eclipse.jetty.toolchain.perf.PlatformTimer -import java.nio.ByteBuffer import java.util.concurrent.Semaphore +//import groovy.util.logging.Slf4j + /** * Performance metric test for issue 204. @@ -26,9 +30,9 @@ import java.util.concurrent.Semaphore * - disconnect * [MEASURE STOP] * */ - +//@Slf4j class WorkflowTask implements MqttCallback { - private MqttAsyncClient client + private MqttClient client private int clientCounter private long startTime private long numOfIterations @@ -37,49 +41,59 @@ class WorkflowTask implements MqttCallback { private PlatformTimer timer = PlatformTimer.detect() private Semaphore disconnectReceived = new Semaphore(0) + private static String printDate() { + return new Date().format("HH:mm:ss,SSS") + " " + Thread.currentThread().getName() + } + void execute() { client.callback = this (1..numOfIterations).each { it -> startTime = System.nanoTime() - client.connect().waitForCompletion(1000) - println "connected ${it}" + println "\nAsk connect " + printDate() + client.connect()//.waitForCompletion() + println "connected ${it}, " + printDate() client.subscribe("/topic${clientCounter}", 0) - println "subscribed to /topic${clientCounter}" - //byte[] bytes = ByteBuffer.allocate(8).putLong(System.currentTimeMillis()).array() - byte[] bytes = "TestPayload".bytes + println "subscribed to /topic${clientCounter} " + printDate() + byte[] bytes = "TestPayload ${it}".bytes client.publish("/topic${clientCounter}", bytes, 0, false) - println "published to /topic${clientCounter}..wait for receive of message" + println "published ${it}, " + printDate() +// println "published to /topic${clientCounter}..wait for receive of message" disconnectReceived.acquire() - println "Acquired semaphore, value ${disconnectReceived.availablePermits()}" - client.disconnect().waitForCompletion(1000) - println "Disconnected" + println "Acquired semaphore, value ${disconnectReceived.availablePermits()} " + printDate() + println "Ask disconnect " + printDate() + client.disconnect()//.waitForCompletion() + println "Disconnected ${it}" + printDate() + "\n" + client.debug.dumpClientDebug() +// client.close() long stopTime = System.nanoTime() long spentNanoSeconds = stopTime - startTime histogram.recordValue(spentNanoSeconds) - //timer.sleep(pauseMicroseconds) +// long waitMicroseconds = pauseMicroseconds - (spentNanoSeconds / 1000) +// println "${spentNanoSeconds} spent nanos, pauseMicroseconds: ${pauseMicroseconds}" +// timer.sleep(waitMicroseconds) } println "Latency diagram (micro secs):\n" histogram.outputPercentileDistribution(System.out, 1000.0) } public void messageArrived(String topic, MqttMessage message) throws Exception { - println "Received published" + println "Received published " + printDate() if (topic != "/topic${clientCounter}".toString()) { println "Error received a publish on not matching topic ${topic}" return } - long stopTime = System.nanoTime() - long spentNanoSeconds = stopTime - startTime +// long stopTime = System.nanoTime() +// long spentNanoSeconds = stopTime - startTime disconnectReceived.release() - println "Released semaphore, value ${disconnectReceived.availablePermits()}" + println "Released semaphore, value ${disconnectReceived.availablePermits()} " + printDate() } public void connectionLost(Throwable cause) { -// println "Client ${clientCounter} lost connection" + println "Client ${clientCounter} lost connection, caused by ${cause}" } public void deliveryComplete(IMqttDeliveryToken token) { @@ -88,10 +102,10 @@ class WorkflowTask implements MqttCallback { } -MqttAsyncClient createClient(int i) { +MqttClient createClient(int i) { String tmpDir = System.getProperty("java.io.tmpdir") MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir) - new MqttAsyncClient("tcp://localhost:1883", "TestClient${i}", dataStore) + new MqttClient("tcp://localhost:1883", "TestClient${i}", dataStore) } Histogram histogram = new Histogram(5) @@ -105,7 +119,7 @@ int duration = args[1] as int long pauseMicroseconds = (1 / numOfIterationsPerSecond) * 1000 * 1000 long numOfIterations = numOfIterationsPerSecond * duration -MqttAsyncClient client = createClient(0) +MqttClient client = createClient(0) def task = new WorkflowTask(client: client, clientCounter: 0, histogram: histogram, pauseMicroseconds: pauseMicroseconds, numOfIterations: numOfIterations) diff --git a/tools_scripts/benchmark/fuse_asynch_benchmarker.groovy b/tools_scripts/benchmark/fuse_asynch_benchmarker.groovy index 03df858f..03349398 100644 --- a/tools_scripts/benchmark/fuse_asynch_benchmarker.groovy +++ b/tools_scripts/benchmark/fuse_asynch_benchmarker.groovy @@ -43,13 +43,13 @@ MQTT sub = new MQTT() //mqtt.setHost("test.mosquitto.org", 1883); sub.setHost(host, 1883) sub.setCleanSession(true) -sub.setHost(host, 1883) sub.setClientId("SubscriberClient${dialog_id}") BenchmarkSubscriber suscriberBench = new BenchmarkSubscriber(mqtt: sub, dialog_id: dialog_id) suscriberBench.connect() -BenchmarkPublisher publisherBench = new BenchmarkPublisher(mqtt: pub, numToSend: numToSend, messagesPerSecond: messagesPerSecond, dialog_id: dialog_id) +BenchmarkPublisher publisherBench = new BenchmarkPublisher(mqtt: pub, numToSend: numToSend, + messagesPerSecond: messagesPerSecond, dialog_id: dialog_id) publisherBench.connect() publisherBench.firePublishes()