anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed perf module to be move in separate project, and disabled leak detector on bytbuf during test

Browse Source
This commit is contained in:
andsel
2018-11-11 16:32:10 +01:00
parent 8e17e99428
commit 8f7b7df124
15 changed files with 12 additions and 1476 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gradle/test.gradle
+12 -11
View File
@@ -6,8 +6,8 @@ test {
class OutputReporter implements TestListener, TestOutputListener {
private StringBuilder stdout = new StringBuilder();
private StringBuilder stderr = new StringBuilder();
private StringBuilder stdout = new StringBuilder()
private StringBuilder stderr = new StringBuilder()
@Override
void beforeSuite(TestDescriptor suite) {
@@ -22,8 +22,8 @@ class OutputReporter implements TestListener, TestOutputListener {
@Override
void beforeTest(TestDescriptor testDescriptor) {
if (!testDescriptor.isComposite()) {
stdout.setLength(0);
stderr.setLength(0);
stdout.setLength(0)
stderr.setLength(0)
}
}
@@ -50,13 +50,14 @@ class OutputReporter implements TestListener, TestOutputListener {
@Override
void onOutput(TestDescriptor testDescriptor, TestOutputEvent outputEvent) {
if (!testDescriptor.isComposite()) {
def buffer = (outputEvent.getDestination() == TestOutputEvent.Destination.StdOut) ? stdout : stderr;
buffer.append(outputEvent.getMessage());
if (outputEvent.getMessage().contains("LEAK:")) {
println stdout
println stderr
throw new RuntimeException("Netty ResourceLeakDetector");
}
def buffer = (outputEvent.getDestination() == TestOutputEvent.Destination.StdOut) ? stdout : stderr
buffer.append(outputEvent.getMessage())
// TODO re-enable this
// if (outputEvent.getMessage().contains("LEAK:")) {
// println stdout
// println stderr
// throw new RuntimeException("Netty ResourceLeakDetector")
// }
}
}
}
perf/build.gradle
-17
View File
@@ -1,17 +0,0 @@
description = 'Moquette - performance tests'
repositories {
jcenter()
mavenCentral()
maven { url 'https://oss.jfrog.org/artifactory/oss-snapshot-local' }
}
dependencies {
compile project(':moquette-broker')
compile group: 'io.netty', name: 'netty-codec-mqtt', version:"${nettyVersion}"
compile group: 'org.hdrhistogram', name: 'HdrHistogram', version:'2.1.2'
compile group: 'org.eclipse.paho', name: 'org.eclipse.paho.client.mqttv3', version:'1.2.0'
compile group: 'org.eclipse.jetty.toolchain', name: 'jetty-perf-helper', version:'1.0.5'
compile(group: 'org.mqttbee', name: 'mqtt-bee', version: 'mqtt-bee-1.0.0-20181105.094934-4')
}
perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java
-150
View File
@@ -1,150 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.parser.netty.performance;
import org.eclipse.jetty.toolchain.perf.PlatformTimer;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;
import static java.nio.charset.StandardCharsets.UTF_8;
class BenchmarkPublisher {
private static final Logger LOG = LoggerFactory.getLogger(BenchmarkPublisher.class);
static class PublishCallback implements IMqttActionListener {
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
LOG.error("PUB: Publish failed", exception);
exception.printStackTrace();
System.exit(2);
}
@Override
public void onSuccess(IMqttToken asyncActionToken) {
}
}
static class ExitTopicCallback implements IMqttActionListener {
private final long startedTime;
private final long numToSend;
private final CountDownLatch stopLatch;
ExitTopicCallback(long startedTime, long numToSend, CountDownLatch stopLatch) {
this.startedTime = startedTime;
this.numToSend = numToSend;
this.stopLatch = stopLatch;
}
@Override
public void onSuccess(IMqttToken asyncActionToken) {
long stopTime = System.currentTimeMillis();
long spentTime = stopTime - startedTime;
LOG.info("PUB: published in {} ms", spentTime);
// try {
// this.connection.disconnect();
// } catch (MqttException mex) {
// LOG.error("Disconnect error", mex);
// }
double msgPerSec = (numToSend / spentTime) * 1000;
LOG.info("PUB: speed {} msg/sec", msgPerSec);
this.stopLatch.countDown();
}
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable th) {
LOG.error("PUB: Publish failed: " + th);
System.exit(2);
}
}
private int qos;
private boolean retain;
private final int numToSend;
private final int messagesPerSecond;
private final String dialog_id;
private final IMqttAsyncClient client;
private CountDownLatch m_latch;
BenchmarkPublisher(MqttAsyncClient client, int numToSend, int messagesPerSecond, String dialog_id) {
this.client = client;
this.numToSend = numToSend;
this.messagesPerSecond = messagesPerSecond;
this.dialog_id = dialog_id;
}
public void connect() throws MqttException {
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setCleanSession(true);
this.client.connect(connectOptions, null, new IMqttActionListener() {
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
LOG.error("PUB: connect fail", exception);
System.exit(2);
}
@Override
public void onSuccess(IMqttToken asyncActionToken) {
LOG.info("PUB: Successfully connected to server");
}
});
this.m_latch = new CountDownLatch(1);
}
public void firePublishes() throws MqttException, UnsupportedEncodingException {
long pauseMicroseconds = (int) ((1.0 / messagesPerSecond) * 1000 * 1000);
LOG.info("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds);
LOG.info("PUB: publishing..");
CountDownLatch m_latch = new CountDownLatch(1);
final long startTime = System.currentTimeMillis();
//initialize the timer
PlatformTimer timer = PlatformTimer.detect();
IMqttActionListener pubCallback = new PublishCallback();
for (int i = 0; i < numToSend; i++) {
long nanos = System.nanoTime();
byte[] message = ("Hello world!!-" + nanos).getBytes(UTF_8);
this.client.publish("/topic" + dialog_id, message, qos, retain, null, pubCallback);
timer.sleep(pauseMicroseconds);
}
IMqttActionListener exitCallback = new ExitTopicCallback(startTime, numToSend, m_latch);
long nanosExit = System.nanoTime();
byte[] exitMessage = ("Hello world!!-" + nanosExit).getBytes(UTF_8);
this.client.publish("/exit" + dialog_id, exitMessage, qos, retain, null, exitCallback);
}
public void waitFinish() throws InterruptedException {
m_latch.await();
try {
this.client.disconnect();
} catch (MqttException mex) {
LOG.error("Disconnect error", mex);
}
}
}
perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java
-169
View File
@@ -1,169 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.parser.netty.performance;
import org.HdrHistogram.Histogram;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
import static java.nio.charset.StandardCharsets.UTF_8;
class BenchmarkSubscriber {
private static final Logger LOG = LoggerFactory.getLogger(BenchmarkSubscriber.class);
class SubscriberCallback implements MqttCallback {
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
//used to catch the first message published
if (!alreadyStarted) {
startTime = System.currentTimeMillis();
alreadyStarted = true;
}
//println "Received a publish on topic ${topic}"
if (topic.startsWith("/exit")) {
LOG.info("SUB: ok Exit!");
LOG.debug("Before disconnect");
// client.disconnect(1000) //wait max 1 sec
//client.disconnectForcibly()
//client.close()
LOG.debug("After disconnect");
long stopTime = System.currentTimeMillis();
long spentTime = stopTime - startTime;
LOG.info("SUB: {} received", topic);
LOG.info("SUB: subscriber disconnected, received {} messages in {} ms", numReceived, spentTime);
double msgPerSec = (numReceived / spentTime) * 1000;
LOG.info("SUB: Speed: {} msg/sec", msgPerSec);
LOG.info("SUB: Latency diagram [microsecs]");
histogram.outputPercentileDistribution(System.out, 1000.0); //nanos/1000
m_latch.countDown();
} else {
String payload = new String(message.getPayload(), UTF_8);
//long sentTime = Long.parseLong(payload.split("-")[1]);
long sentTime = Long.parseLong(payload.split("-")[1]);
long delay = System.nanoTime() - sentTime;
histogram.recordValue(delay);
// print '+'
numReceived++;
// if ((numReceived % 10000) == 0) {
// print '.'
// }
}
}
@Override
public void connectionLost(Throwable cause) {
LOG.error("Client Subscriber lost connection, caused by ", cause);
try {
client.disconnect();
} catch (MqttException mex) {
LOG.error(null, mex);
}
LOG.error("SUB: Something failed during message reception");
m_latch.countDown();
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
}
}
private int numReceived;
private long startTime = System.currentTimeMillis();
private CountDownLatch m_latch = new CountDownLatch(1);
private boolean alreadyStarted;
private Histogram histogram = new Histogram(5);
private final IMqttAsyncClient client;
private final String dialog_id;
BenchmarkSubscriber(MqttAsyncClient client, String dialog_id) {
this.client = client;
this.dialog_id = dialog_id;
}
public void connect() throws MqttException {
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setCleanSession(true);
this.client.setCallback(new SubscriberCallback());
this.client.connect(connectOptions, null, new IMqttActionListener() {
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
LOG.error("SUB: connect fail", exception);
System.exit(2);
}
@Override
public void onSuccess(IMqttToken asyncActionToken) {
try {
client.subscribe("/topic" + dialog_id, 1, null, new IMqttActionListener() {
@Override
public void onFailure(IMqttToken asyncActionToken2, Throwable exception) {
try {
client.disconnect();
} catch (MqttException mex) {
LOG.error(null, mex);
}
}
@Override
public void onSuccess(IMqttToken asyncActionToken2) {
LOG.info("SUB: subscribed to /topic{} qos: 1", dialog_id);
}
});
} catch (MqttException e) {
e.printStackTrace();
}
try {
client.subscribe("/exit" + dialog_id, 1, null, new IMqttActionListener() {
@Override
public void onFailure(IMqttToken asyncActionToken2, Throwable exception) {
try {
client.disconnect();
} catch (MqttException e) {
e.printStackTrace();
}
m_latch.countDown();
}
@Override
public void onSuccess(IMqttToken asyncActionToken2) {
LOG.info("SUB: subscribed to /exit{} qos: 1", dialog_id);
}
});
} catch (MqttException e) {
e.printStackTrace();
}
}
});
}
public void waitFinish() throws InterruptedException, MqttException {
this.m_latch.await();
this.client.disconnect(1000);
}
}
perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java
-210
View File
@@ -1,210 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.parser.netty.performance;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import org.HdrHistogram.Histogram;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static io.moquette.parser.netty.performance.NettyPublishReceiverHandler.payload2Str;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
@ChannelHandler.Sharable
class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(LoopMQTTHandler.class);
private ProtocolDecodingServer.SharedState m_state;
Histogram processingTime = new Histogram(5);
Histogram forthNetworkTime = new Histogram(5);
LoopMQTTHandler(ProtocolDecodingServer.SharedState state) {
this.m_state = state;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
MqttMessage msg = (MqttMessage) message;
MqttMessageType messageType = msg.fixedHeader().messageType();
String clientID = clientID(ctx.channel());
try {
switch (messageType) {
case CONNECT:
MqttConnectMessage connect = (MqttConnectMessage) message;
handleConnect(ctx, connect);
return;
case SUBSCRIBE:
LOG.info("Received a message of type {} from <{}>", messageType, clientID);
handleSubscribe(ctx, (MqttSubscribeMessage) msg);
return;
case PUBLISH:
LOG.info("Received a message of type {} from <{}>", messageType, clientID);
handlePublish(ctx, (MqttPublishMessage) msg);
return;
case DISCONNECT:
LOG.info("Received a message of type {} from <{}>", messageType, clientID);
ctx.close().addListener(CLOSE_ON_FAILURE);
// case PUBACK:
// NettyChannel channel;
// synchronized (m_channelMapper) {
// if (!m_channelMapper.containsKey(ctx)) {
// m_channelMapper.put(ctx, new NettyChannel(ctx));
// }
// channel = m_channelMapper.get(ctx);
// }
//
// m_messaging.handleProtocolMessage(channel, msg);
break;
case PINGREQ:
MqttFixedHeader pingHeader = new MqttFixedHeader(
MqttMessageType.PINGRESP,
false,
AT_MOST_ONCE,
false,
0);
MqttMessage pingResp = new MqttMessage(pingHeader);
ctx.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE);
break;
default:
LOG.info("Received a message of type {} from <{}>", messageType, clientID);
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
private void handlePublish(ChannelHandlerContext ctx, MqttPublishMessage msg) {
if (!m_state.isForwardable()) {
LOG.info("Subscriber not yet connected, LoopHandler instance is {}", this);
return;
}
long start = System.nanoTime();
LOG.debug("push forward message the topic {}", msg.variableHeader().topicName());
LOG.debug("content <{}>", payload2Str(msg.payload()));
String decodedPayload = payload2Str(msg.payload());
long sentTime = Long.parseLong(decodedPayload.split("-")[1]);
forthNetworkTime.recordValue(start - sentTime);
//publish always at Qos0, to don't handle PUBACK or the complete Qos2 workflow
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE,
false, 0);
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(msg.variableHeader().topicName(), 0);
MqttPublishMessage pubMessage = new MqttPublishMessage(fixedHeader, varHeader, msg.payload());
m_state.getSubscriberCh().writeAndFlush(pubMessage).addListener(CLOSE_ON_FAILURE);
/*Channel subscriberCh = m_state.getSubscriberCh();
if (subscriberCh.isWritable()) {
subscriberCh.write(pubMessage);
} else {
subscriberCh.writeAndFlush(pubMessage);
}*/
long stop = System.nanoTime();
processingTime.recordValue(stop - start);
LOG.info("Request processed in {} ns, matching {}", stop - start, payload2Str(msg.payload()));
}
private void handleSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) {
m_state.setForwardable(true);
LOG.debug(" new value of flag {}, LoopHandler instance is {}", m_state.isForwardable(), this);
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, AT_LEAST_ONCE, false, 0);
MqttSubAckPayload payload = new MqttSubAckPayload(AT_MOST_ONCE.value());
MqttSubAckMessage ackMessage = new MqttSubAckMessage(
fixedHeader,
from(msg.variableHeader().messageId()),
payload);
ctx.writeAndFlush(ackMessage).addListener(CLOSE_ON_FAILURE);
LOG.debug("subscribed client to {}", msg.payload().topicSubscriptions());
}
private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
MqttConnectPayload payload = msg.payload();
String clientID = payload.clientIdentifier();
LOG.info("Received a message of type {} from <{}>", MqttMessageType.CONNECT, clientID);
clientID(ctx.channel(), clientID);
if (clientID.toLowerCase().startsWith("sub")) {
m_state.setSubscriberCh(ctx.channel());
} else if (clientID.toLowerCase().startsWith("pub")) {
m_state.setPublisherCh(ctx.channel());
} else {
//we don't admit other names
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE,
false, 0);
MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(
MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false);
MqttConnAckMessage koResp = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
ctx.writeAndFlush(koResp).addListener(CLOSE_ON_FAILURE);
ctx.close().addListener(CLOSE_ON_FAILURE);
return;
}
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE,
false, 0);
MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(
MqttConnectReturnCode.CONNECTION_ACCEPTED, false);
MqttConnAckMessage okResp = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
ctx.writeAndFlush(okResp).addListener(CLOSE_ON_FAILURE);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
LOG.info("Received channel inactive");
// NettyChannel channel = m_channelMapper.get(ctx);
// String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
// m_messaging.lostConnection(channel, clientID);
ctx.channel().close().addListener(CLOSE_ON_FAILURE);
System.out.println("Processing time histogram (microsecs)");
this.processingTime.outputPercentileDistribution(System.out, 1000.0);
System.out.println("Network time histogram (microsecs)");
this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0);
// synchronized (m_channelMapper) {
// m_channelMapper.remove(ctx);
// }
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
if (ctx.channel().isWritable()) {
String clientID = clientID(ctx.channel());
LOG.info("Channel for client <{}> is writable again", clientID);
ctx.channel().flush();
}
ctx.fireChannelWritabilityChanged();
}
private static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf("ClientID");
private static void clientID(Channel channel, String clientID) {
channel.attr(ATTR_KEY_CLIENTID).set(clientID);
}
private static String clientID(Channel channel) {
return (String) channel.attr(ATTR_KEY_CLIENTID).get();
}
}
perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java
-94
View File
@@ -1,94 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.parser.netty.performance;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.HdrHistogram.Histogram;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
import static java.nio.charset.StandardCharsets.UTF_8;
@ChannelHandler.Sharable
class NettyPublishReceiverHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(NettyPublishReceiverHandler.class);
Histogram forthNetworkTime = new Histogram(5);
NettyPublishReceiverHandler() {
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
MqttMessage msg = (MqttMessage) message;
MqttMessageType type = msg.fixedHeader().messageType();
try {
switch (type) {
case PUBLISH:
LOG.info("Received a message of type {}", type);
handlePublish((MqttPublishMessage) msg);
return;
default:
LOG.info("Received a message of type {}", type);
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
private void handlePublish(MqttPublishMessage msg) {
long start = System.nanoTime();
LOG.debug("push forward message the topic {}", msg.variableHeader().topicName());
LOG.debug("content <{}>", payload2Str(msg.content()));
String decodedPayload = payload2Str(msg.content());
long sentTime = Long.parseLong(decodedPayload.split("-")[1]);
forthNetworkTime.recordValue(start - sentTime);
long stop = System.nanoTime();
LOG.info("Request processed in {} ns, matching {}", stop - start, decodedPayload);
}
static String payload2Str(ByteBuf content) {
byte[] rawBytes;
if (content.hasArray()) {
rawBytes = content.array();
} else {
int size = content.readableBytes();
rawBytes = new byte[size];
content.getBytes(content.readerIndex(), rawBytes);
}
return new String(rawBytes, UTF_8);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
LOG.info("Received channel inactive");
ctx.channel().close().addListener(CLOSE_ON_FAILURE);
System.out.println("Network time histogram (microsecs)");
this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0);
}
}
perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java
-187
View File
@@ -1,187 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.parser.netty.performance;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.UnsupportedEncodingException;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
/**
* Emulates a broker, but doesn't apply any protocol logic, just forward the qos0 publishes
* to the lone subscriber, it's used just to measure the protocol decoding/encoding overhead.
*
*/
public class ProtocolDecodingServer {
static class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleState e = ((IdleStateEvent) evt).state();
if (e == IdleState.READER_IDLE) {
//fire a channelInactive to trigger publish of Will
ctx.fireChannelInactive();
ctx.close().addListener(CLOSE_ON_FAILURE);
} /*else if (e.getState() == IdleState.WRITER_IDLE) {
ctx.writeAndFlush(new PingMessage());
}*/
}
}
}
private static final Logger LOG = LoggerFactory.getLogger(ProtocolDecodingServer.class);
EventLoopGroup m_bossGroup;
EventLoopGroup m_workerGroup;
static final class SharedState {
private volatile Channel subscriberCh;
private volatile Channel publisherCh;
private volatile boolean forwardPublish;
public boolean isForwardable() {
return forwardPublish;
}
void setForwardable(boolean forwardPublish) {
this.forwardPublish = forwardPublish;
}
public void setSubscriberCh(Channel subscriberCh) {
this.subscriberCh = subscriberCh;
}
public Channel getSubscriberCh() {
return subscriberCh;
}
public void setPublisherCh(Channel publisherCh) {
this.publisherCh = publisherCh;
}
public Channel getPublisherCh() {
return publisherCh;
}
}
final SharedState state = new SharedState();
void init() {
String host = "0.0.0.0";
int port = 1883;
m_bossGroup = new NioEventLoopGroup();
m_workerGroup = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(m_bossGroup, m_workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
try {
pipeline.addFirst("idleStateHandler", new IdleStateHandler(2, 0, 0));
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler());
pipeline.addLast("decoder", new MqttDecoder());
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
pipeline.addLast("handler", new LoopMQTTHandler(state));
} catch (Throwable th) {
LOG.error("Severe error during pipeline creation", th);
throw th;
}
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.option(ChannelOption.SO_REUSEADDR, true)
.childOption(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_KEEPALIVE, true);
try {
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(host, port);
LOG.info("Server binded host: {}, port: {}", host, port);
f.sync().addListener(CLOSE_ON_FAILURE);
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
@SuppressWarnings("FutureReturnValueIgnored")
public void stop() {
if (m_workerGroup == null) {
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
if (m_bossGroup == null) {
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
m_workerGroup.shutdownGracefully();
m_bossGroup.shutdownGracefully();
LOG.info("Closed boss and worker Event loops");
System.out.println("Server stopped");
}
public static void main(String[] args) throws MqttException, InterruptedException, UnsupportedEncodingException {
final ProtocolDecodingServer server = new ProtocolDecodingServer();
server.init();
System.out.println("Loop server started");
startClientsTesting();
//Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread(server::stop));
}
private static void startClientsTesting() throws MqttException, InterruptedException, UnsupportedEncodingException {
String host = "localhost";
int numToSend = 10;
int messagesPerSecond = 10000;
String dialog_id = "test1";
String tmpDir = System.getProperty("java.io.tmpdir");
MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
MqttAsyncClient pub = new MqttAsyncClient("tcp://" + host + ":1883", "PublisherClient" + dialog_id, dataStore);
MqttDefaultFilePersistence dataStoreSub = new MqttDefaultFilePersistence(tmpDir);
MqttAsyncClient sub = new MqttAsyncClient("tcp://" + host + ":1883", "SubscriberClient" + dialog_id,
dataStoreSub);
BenchmarkSubscriber suscriberBench = new BenchmarkSubscriber(sub, dialog_id);
suscriberBench.connect();
BenchmarkPublisher publisherBench = new BenchmarkPublisher(pub, numToSend, messagesPerSecond, dialog_id);
publisherBench.connect();
publisherBench.firePublishes();
suscriberBench.waitFinish();
publisherBench.waitFinish();
System.out.println("Started clients (sub/pub) for testing");
}
}
perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java
-123
View File
@@ -1,123 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.parser.netty.performance;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.UnsupportedEncodingException;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
/**
* Emulates a broker, but doesn't apply any protocol logic, just forward the qos0 publishes
* to the lone subscriber, it's used just to measure the protocol decoding/encoding overhead.
*
*/
public class ProtocolPublishDecodingServer {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolPublishDecodingServer.class);
EventLoopGroup m_bossGroup;
EventLoopGroup m_workerGroup;
void init() {
String host = "0.0.0.0";
int port = 1883;
m_bossGroup = new NioEventLoopGroup();
m_workerGroup = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(m_bossGroup, m_workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
try {
pipeline.addLast("decoder", new MqttDecoder());
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
pipeline.addLast("handler", new PublishReceiverHandler());
// pipeline.addLast("decoder", new MqttDecoder());
// pipeline.addLast("encoder", MqttEncoder.INSTANCE);
// pipeline.addLast("handler", new NettyPublishReceiverHandler());
} catch (Throwable th) {
LOG.error("Severe error during pipeline creation", th);
throw th;
}
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.option(ChannelOption.SO_REUSEADDR, true)
.childOption(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_KEEPALIVE, true);
try {
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(host, port);
LOG.info("Server binded host: {}, port: {}", host, port);
f.sync().addListener(CLOSE_ON_FAILURE);
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
@SuppressWarnings("FutureReturnValueIgnored")
public void stop() {
if (m_workerGroup == null) {
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
if (m_bossGroup == null) {
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
m_workerGroup.shutdownGracefully();
m_bossGroup.shutdownGracefully();
LOG.info("Closed boss and worker Event loops");
System.out.println("Server stopped");
}
public static void main(String[] args) throws UnsupportedEncodingException {
final ProtocolPublishDecodingServer server = new ProtocolPublishDecodingServer();
server.init();
System.out.println("Loop server started");
publishBombing();
publishBombing();
publishBombing();
publishBombing();
publishBombing();
//Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread(server::stop));
}
private static void publishBombing() throws UnsupportedEncodingException {
LOG.info("Started publish loop");
PublishBomber heavyPublisher = new PublishBomber("localhost", 1883);
heavyPublisher.publishLoop(25000, 100000);
heavyPublisher.disconnect();
LOG.info("Finished publish loop");
}
}
perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java
-108
View File
@@ -1,108 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.parser.netty.performance;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.*;
import org.apache.commons.codec.CharEncoding;
import org.eclipse.jetty.toolchain.perf.PlatformTimer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.UnsupportedEncodingException;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
class PublishBomber {
private static final Logger LOG = LoggerFactory.getLogger(PublishBomber.class);
private final EventLoopGroup workerGroup;
private Channel channel;
@SuppressWarnings("FutureReturnValueIgnored")
PublishBomber(String host, int port) {
workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true);
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder", new MqttDecoder());
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
}
});
// Start the client.
channel = b.connect(host, port).sync().channel();
} catch (Exception ex) {
LOG.error("Error received in client setup", ex);
workerGroup.shutdownGracefully();
}
}
private void sendMessage(MqttMessage msg) {
try {
channel.writeAndFlush(msg).await().addListener(CLOSE_ON_FAILURE);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void publishLoop(int messagesPerSecond, int numToSend) throws UnsupportedEncodingException {
long pauseMicroseconds = (int) ((1.0 / messagesPerSecond) * 1000 * 1000);
LOG.warn("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds);
LOG.info("PUB: publishing..");
final long startTime = System.currentTimeMillis();
//initialize the timer
PlatformTimer timer = PlatformTimer.detect();
for (int i = 0; i < numToSend; i++) {
long nanos = System.nanoTime();
byte[] rawContent = ("Hello world!!-" + nanos).getBytes(CharEncoding.UTF_8);
ByteBuf payload = Unpooled.copiedBuffer(rawContent);
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE,
false, 0);
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader("/topic", 0);
MqttPublishMessage pubMessage = new MqttPublishMessage(fixedHeader, varHeader, payload);
sendMessage(pubMessage);
timer.sleep(pauseMicroseconds);
}
LOG.info("PUB: published in {} ms", System.currentTimeMillis() - startTime);
}
public void disconnect() {
try {
this.channel.disconnect().await().addListener(CLOSE_ON_FAILURE);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java
-81
View File
@@ -1,81 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.parser.netty.performance;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.HdrHistogram.Histogram;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static io.moquette.parser.netty.performance.NettyPublishReceiverHandler.payload2Str;
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
@ChannelHandler.Sharable
class PublishReceiverHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(PublishReceiverHandler.class);
private Histogram forthNetworkTime = new Histogram(5);
PublishReceiverHandler() {
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
MqttMessage msg = (MqttMessage) message;
MqttMessageType messageType = msg.fixedHeader().messageType();
try {
switch (messageType) {
case PUBLISH:
LOG.info("Received a message of type {}", messageType);
handlePublish((MqttPublishMessage) msg);
return;
default:
LOG.info("Received a message of type {}", messageType);
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
private void handlePublish(MqttPublishMessage msg) {
long start = System.nanoTime();
LOG.debug("push forward message the topic {}", msg.variableHeader().topicName());
LOG.debug("content <{}>", payload2Str(msg.payload()));
String decodedPayload = payload2Str(msg.payload());
long sentTime = Long.parseLong(decodedPayload.split("-")[1]);
forthNetworkTime.recordValue(start - sentTime);
long stop = System.nanoTime();
LOG.info("Request processed in {} ns, matching {}", stop - start, payload2Str(msg.payload()));
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
LOG.info("Received channel inactive");
ctx.channel().close().addListener(CLOSE_ON_FAILURE);
System.out.println("Network time histogram (microsecs)");
this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0);
}
}
perf/src/main/java/io/moquette/performance/clients/mqttbee/BeeBenchmarker.java
-100
View File
@@ -1,100 +0,0 @@
package io.moquette.performance.clients.mqttbee;
import io.reactivex.Flowable;
import io.reactivex.Single;
import org.mqttbee.api.mqtt.MqttClient;
import org.mqttbee.api.mqtt.datatypes.MqttQos;
import org.mqttbee.api.mqtt.mqtt3.Mqtt3Client;
import org.mqttbee.api.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck;
import org.mqttbee.api.mqtt.mqtt3.message.publish.Mqtt3Publish;
import org.mqttbee.api.mqtt.mqtt3.message.publish.Mqtt3PublishResult;
import org.mqttbee.api.mqtt.mqtt3.message.subscribe.suback.Mqtt3SubAck;
import org.mqttbee.rx.FlowableWithSingle;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class BeeBenchmarker {
public static void main(String[] args) {
final Mqtt3Client client = MqttClient.builder()
.identifier(UUID.randomUUID().toString())
.serverHost("localhost")
.serverPort(1883)
.useMqttVersion3()
.buildReactive();
/*
* As we use the reactive API, the following line does not connect yet, but return a reactive type.
* e.g. Single is something like a lazy and reusable future.
*/
final Single<Mqtt3ConnAck> connAckSingle = client.connect().keepAlive(10, TimeUnit.SECONDS).done();
// /*
// * Same here: the following line does not subscribe yet, but return a reactive type.
// * FlowableWithSingle is a combination of the single SUBACK message and a Flowable of PUBLISH messages.
// * A Flowable is an asynchronous stream, that enables back pressure from the application over MQTT Bee to the broker.
// */
// final FlowableWithSingle<Mqtt3SubAck, Mqtt3Publish> subAckAndMatchingPublishes = client.subscribeWithStream()
// .addSubscription().topicFilter("a/b/c").qos(MqttQos.AT_LEAST_ONCE).done()
// .addSubscription().topicFilter("a/b/c/d").qos(MqttQos.EXACTLY_ONCE).done()
// .done();
//
// /*
// * The reactive types offer as many operators that will not be covered here.
// * Here we register simple callbacks to print messages when we received the CONNACK, SUBACK and matching PUBLISH messages.
// */
// final Single<Mqtt3ConnAck> connectScenario = connAckSingle
// .doOnSuccess(connAck -> System.out.println("Connected with return code " + connAck.getReturnCode()))
// .doOnError(throwable -> System.out.println("Connection failed, " + throwable.getMessage()));
//
// final Flowable<Mqtt3Publish> subscribeScenario = subAckAndMatchingPublishes
// .doOnSingle((subAck, subscription) ->
// System.out.println("Subscribed with return codes " + subAck.getReturnCodes()))
// .doOnNext(publish ->
// System.out.println("Received publish" +
// ", topic: " + publish.getTopic() +
// ", QoS: " + publish.getQos() +
// ", payload: " + new String(publish.getPayloadAsBytes())
// )
// );
//
// /*
// * Reactive types let us compose a sequence of actions
// */
// final Flowable<Mqtt3Publish> connectAndSubscribe = connectScenario.toCompletable().andThen(subscribeScenario);
//
// /*
// * By subscribing to reactive types, the sequence is executed
// */
// connectAndSubscribe.blockingSubscribe();
/*
* Fake a stream of PUBLISH messages to publish with incrementing ids
*/
final AtomicInteger counter = new AtomicInteger();
Flowable<Mqtt3Publish> messagesToPublish = Flowable.generate(emitter -> {
final int id = counter.incrementAndGet();
final Mqtt3Publish publishMessage = Mqtt3Publish.builder()
.topic("a/b/c").qos(MqttQos.AT_LEAST_ONCE).payload(("test " + id).getBytes()).build();
emitter.onNext(publishMessage);
});
/*
* Emit 1 message only every 100 milliseconds
*/
messagesToPublish = messagesToPublish.zipWith(Flowable.interval(10, TimeUnit.MILLISECONDS), (publish, aLong) -> publish);
final Single<Mqtt3ConnAck> connectScenario = connAckSingle
.doOnSuccess(connAck -> System.out.println("Connected with return code " + connAck.getReturnCode()))
.doOnError(throwable -> System.out.println("Connection failed, " + throwable.getMessage()));
final Flowable<Mqtt3PublishResult> publishScenario = client.publish(messagesToPublish)
.doOnNext(publishResult -> System.out.println("Publish acknowledged: " + new String(publishResult.getPublish().getPayloadAsBytes())));
connectScenario.toCompletable().andThen(publishScenario).blockingSubscribe();
}
}
perf/src/main/java/io/moquette/performance/clients/paho/PahoBenchmarker.java
-56
View File
@@ -1,56 +0,0 @@
package io.moquette.performance.clients.paho;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import java.nio.ByteBuffer;
public class PahoBenchmarker {
private static void publishQos0(IMqttAsyncClient client, long numMessages) throws MqttException {
System.out.println("publishing..");
long startTime = System.currentTimeMillis();
for (int i = 0; i < numMessages; i++) {
byte[] bytes = ByteBuffer.allocate(8).putLong(System.currentTimeMillis()).array();
client.publish("topic", bytes, 0, false);
}
client.publish("/exit", "Exit".getBytes(), 0, false);
long stopTime = System.currentTimeMillis();
long spentTime = stopTime -startTime;
System.out.println("published in " + spentTime + " ms");
}
public static void main(String[] args) throws MqttException, InterruptedException {
if (args.length < 2) {
System.out.println("Usage benchmarker <host> <numMessages>");
System.out.println("java io.moquette.performance.clients.paho.PahoBenchmarker localhost 1000 500 test1");
return;
}
String host = args[0];
long numMessages = Long.parseLong(args[1]);
String tmpDir = System.getProperty("java.io.tmpdir");
MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
//MemoryPersistence dataStore = new MemoryPersistence()
int rnd = (int) (Math.random() * 100);
MqttAsyncClient client = new MqttAsyncClient("tcp://" + host +":1883",
"SubscriberClient" + rnd, dataStore);
SubscriberCallback callback = new SubscriberCallback();
client.setCallback(callback);
client.connect().waitForCompletion(1000);
client.subscribe("topic", 0);
System.out.println("subscribed to topic");
client.subscribe("/exit", 0);
System.out.println("subscribed to /exit");
publishQos0(client, numMessages);
callback.waitFinish();
// Map sorted = callback.distribution.sort( { k1, k2 -> k1 <=> k2 } as Comparator)
// File csv = new File("perf.csv")
// sorted.each {delay, count -> csv << "${delay}, ${count}\r\n"}
}
}
perf/src/main/java/io/moquette/performance/clients/paho/SubscriberCallback.java
-63
View File
@@ -1,63 +0,0 @@
package io.moquette.performance.clients.paho;
import org.eclipse.paho.client.mqttv3.*;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
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 = new HashMap<>();
IMqttAsyncClient client;
void waitFinish() throws InterruptedException {
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;
System.out.println("/exit received");
System.out.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.getPayload());
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
Long value = distribution.get(delay);
if (value != null) {
distribution.put(delay, value++);
} else {
distribution.put(delay, 1L);
}
m_numReceived++;
}
}
public void deliveryComplete(IMqttDeliveryToken token) {
//To change body of implemented methods use File | Settings | File Templates.
}
}
perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java
-104
View File
@@ -1,104 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi.impl.subscriptions;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadMXBean;
import io.moquette.broker.ISubscriptionsRepository;
import io.moquette.broker.MemorySubscriptionsRepository;
import io.moquette.persistence.MemoryStorageService;
import io.moquette.spi.ISessionsStore;
import io.netty.handler.codec.mqtt.MqttQoS;
public final class PerformanceTest {
// Set CPU governor and max frequency to lowest value for this test
/* debian:
cpufreq-set -g performance -u 800MHz -c 0
cpufreq-set -g performance -u 800MHz -c 1
cpufreq-set -g performance -u 800MHz -c 2
cpufreq-set -g performance -u 800MHz -c 3
*/
public static void main(String[] args) throws IOException {
ISubscriptionsDirectory store = new CTrieSubscriptionDirectory();
MemoryStorageService memStore = new MemoryStorageService(null, null);
ISessionsStore aSessionsStore = memStore.sessionsStore();
ISubscriptionsRepository sessionsRepository = new MemorySubscriptionsRepository();
store.init(sessionsRepository);
int times = 10000;
int users = 500;
int devices = 5;
int values = 3;
int subs = 2;
int n = users * devices * values * subs;
ThreadMXBean b = ManagementFactory.getThreadMXBean();
Topic[] topics = new Topic[n];
for (int i = 0; i < n; i++) {
topics[i] = new Topic(
(i % users) + "/" + ((i / users) % devices) + "/" + ((i / (users * devices)) % values) + "/"
+ ((i / (users * devices * values)) % subs));
Subscription sub = new Subscription("CLI_ID_" + (i % users), topics[i], MqttQoS.AT_MOST_ONCE);
aSessionsStore.subscriptionStore().addNewSubscription(sub);
store.add(sub);
}
long min = Long.MAX_VALUE;
//AtomicLong min = new AtomicLong(Long.MAX_VALUE);
System.out.println("Store size: " + store.size() + " " + n);
// warm up
for (int i = 0; i < n; i++) {
store.matches(topics[i]);
}
System.out.println("press return for start");
System.in.read();
// IntStream.range(0, times)
// .parallel()
// .forEach(j -> {
for (int j = 0; j < times; j++) {
System.gc();
Thread.yield();
long time = b.getCurrentThreadCpuTime();
for (int i = 0; i < n; i++) {
store.matches(topics[i]);
}
long time2 = b.getCurrentThreadCpuTime() - time;
min = Math.min(min, time2);
// min.updateAndGet(c -> Math.min(c, time2));
System.out.println(".." + j + ": " + time2 + " min:" + min);
}
// });
System.out.println("min:" + min);
}
private PerformanceTest() {
}
}
settings.gradle
-3
View File
@@ -4,12 +4,9 @@ include ':moquette-distribution'
include ':moquette-h2-storage'
include ':moquette-mapdb-storage'
include ':moquette-embedded-test'
include ':moquette-osgi-test'
include ':moquette-performance'
project(':moquette-broker').projectDir = "$rootDir/broker" as File
project(':moquette-distribution').projectDir = "$rootDir/distribution" as File
project(':moquette-embedded-test').projectDir = "$rootDir/embedding_moquette" as File
project(':moquette-performance').projectDir = "$rootDir/perf" as File
project(':moquette-h2-storage').projectDir = "$rootDir/h2_storage" as File
project(':moquette-mapdb-storage').projectDir = "$rootDir/mapdb_storage" as File