Added performance project module, to stress test specifics part of Moquette
This commit is contained in:
andsel
@@ -0,0 +1,127 @@
|
||||
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.util.concurrent.CountDownLatch;
|
||||
|
||||
class BenchmarkPublisher {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(BenchmarkPublisher.class);
|
||||
|
||||
static class PublishCallback implements IMqttActionListener {
|
||||
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
|
||||
LOG.error("PUB: Publish failed: " + exception);
|
||||
exception.printStackTrace();
|
||||
System.exit(2);
|
||||
}
|
||||
|
||||
public void onSuccess(IMqttToken asyncActionToken) {
|
||||
}
|
||||
}
|
||||
|
||||
static class ExitTopicCallback implements IMqttActionListener {
|
||||
|
||||
private final long startedTime;
|
||||
private final long numToSend;
|
||||
private final IMqttAsyncClient connection;
|
||||
private final CountDownLatch stopLatch;
|
||||
|
||||
ExitTopicCallback(long startedTime, long numToSend, IMqttAsyncClient connection, CountDownLatch stopLatch) {
|
||||
this.startedTime = startedTime;
|
||||
this.numToSend = numToSend;
|
||||
this.connection = connection;
|
||||
this.stopLatch = stopLatch;
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
|
||||
public void onFailure(IMqttToken asyncActionToken, Throwable th) {
|
||||
LOG.error("PUB: Publish failed: " + th);
|
||||
System.exit(2);
|
||||
}
|
||||
}
|
||||
|
||||
private int qos = 0;
|
||||
private boolean retain = false;
|
||||
private final int numToSend;
|
||||
private final int messagesPerSecond;
|
||||
private final String dialog_id;
|
||||
private final IMqttAsyncClient client;
|
||||
private CountDownLatch m_latch;
|
||||
|
||||
public 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() {
|
||||
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
|
||||
LOG.error("PUB: connect fail", exception);
|
||||
System.exit(2);
|
||||
}
|
||||
|
||||
public void onSuccess(IMqttToken asyncActionToken) {
|
||||
LOG.info("PUB: Successfully connected to server");
|
||||
}
|
||||
});
|
||||
|
||||
this.m_latch = new CountDownLatch(1);
|
||||
}
|
||||
|
||||
public void firePublishes() throws MqttException {
|
||||
long pauseMicroseconds = (1 / this.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();
|
||||
this.client.publish("/topic" + dialog_id, message, qos, retain, null, pubCallback);
|
||||
timer.sleep(pauseMicroseconds);
|
||||
}
|
||||
|
||||
IMqttActionListener exitCallback = new ExitTopicCallback(startTime, numToSend, this.client, m_latch);
|
||||
long nanosExit = System.nanoTime();
|
||||
byte[] exitMessage = ("Hello world!!-" + nanosExit).getBytes();
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
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;
|
||||
|
||||
class BenchmarkSubscriber {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(BenchmarkSubscriber.class);
|
||||
|
||||
class SubscriberCallback implements MqttCallback {
|
||||
|
||||
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!");
|
||||
exit = true;
|
||||
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());
|
||||
//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 '.'
|
||||
// }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
|
||||
public void deliveryComplete(IMqttDeliveryToken token) {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private int numReceived = 0;
|
||||
private long startTime = System.currentTimeMillis();
|
||||
private boolean exit = false;
|
||||
private CountDownLatch m_latch = new CountDownLatch(1);
|
||||
private boolean alreadyStarted = false;
|
||||
private Histogram histogram = new Histogram(5);
|
||||
|
||||
private final IMqttAsyncClient client;
|
||||
private final String dialog_id;
|
||||
|
||||
public 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() {
|
||||
|
||||
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
|
||||
LOG.error( "SUB: connect fail", exception);
|
||||
System.exit(2);
|
||||
}
|
||||
|
||||
public void onSuccess(IMqttToken asyncActionToken) {
|
||||
try {
|
||||
client.subscribe("/topic" + dialog_id, 1, null, new IMqttActionListener() {
|
||||
public void onFailure(IMqttToken asyncActionToken2, Throwable exception) {
|
||||
try {
|
||||
client.disconnect();
|
||||
} catch (MqttException mex) {
|
||||
LOG.error(null, mex);
|
||||
}
|
||||
}
|
||||
|
||||
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() {
|
||||
public void onFailure(IMqttToken asyncActionToken2, Throwable exception) {
|
||||
try {
|
||||
client.disconnect();
|
||||
} catch (MqttException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
m_latch.countDown();
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,188 @@
|
||||
package io.moquette.parser.netty.performance;
|
||||
|
||||
import io.moquette.parser.proto.messages.*;
|
||||
import static io.moquette.parser.proto.messages.AbstractMessage.*;
|
||||
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.ChannelInboundHandlerAdapter;
|
||||
import io.moquette.parser.proto.Utils;
|
||||
import io.netty.util.AttributeKey;
|
||||
import org.HdrHistogram.Histogram;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
/**
|
||||
* Created by andrea on 6/2/15.
|
||||
*/
|
||||
@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);
|
||||
|
||||
public LoopMQTTHandler(ProtocolDecodingServer.SharedState state) {
|
||||
this.m_state = state;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object message) {
|
||||
AbstractMessage msg = (AbstractMessage) message;
|
||||
String clientID = clientID(ctx.channel());
|
||||
|
||||
try {
|
||||
switch (msg.getMessageType()) {
|
||||
case CONNECT:
|
||||
ConnectMessage connect = (ConnectMessage) message;
|
||||
clientID = connect.getClientID();
|
||||
LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
|
||||
handleConnect(ctx, connect);
|
||||
return;
|
||||
case SUBSCRIBE:
|
||||
LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
|
||||
handleSubscribe(ctx, (SubscribeMessage) msg);
|
||||
return;
|
||||
case PUBLISH:
|
||||
LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
|
||||
handlePublish(ctx, (PublishMessage) msg);
|
||||
return;
|
||||
case DISCONNECT:
|
||||
LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
|
||||
ctx.close();
|
||||
// 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:
|
||||
PingRespMessage pingResp = new PingRespMessage();
|
||||
ctx.writeAndFlush(pingResp);
|
||||
break;
|
||||
default:
|
||||
LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
LOG.error("Bad error in processing the message", ex);
|
||||
}
|
||||
}
|
||||
|
||||
private void handlePublish(ChannelHandlerContext ctx, PublishMessage 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.getTopicName());
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("content <{}>", payload2Str(msg.getPayload()));
|
||||
}
|
||||
String decodedPayload = payload2Str(msg.getPayload());
|
||||
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
|
||||
PublishMessage pubMessage = new PublishMessage();
|
||||
pubMessage.setRetainFlag(false);
|
||||
pubMessage.setTopicName(msg.getTopicName());
|
||||
pubMessage.setQos(QOSType.MOST_ONE);
|
||||
pubMessage.setPayload(msg.getPayload());
|
||||
|
||||
m_state.getSubscriberCh().writeAndFlush(pubMessage);
|
||||
/*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.getPayload()));
|
||||
}
|
||||
|
||||
private void handleSubscribe(ChannelHandlerContext ctx, SubscribeMessage msg) {
|
||||
m_state.setForwardable(true);
|
||||
LOG.debug(" new value of flag {}, LoopHandler instance is {}", m_state.isForwardable(), this);
|
||||
SubAckMessage ackMessage = new SubAckMessage();
|
||||
ackMessage.setMessageID(msg.getMessageID());
|
||||
ctx.writeAndFlush(ackMessage);
|
||||
LOG.debug("subscribed client to {}", msg.subscriptions());
|
||||
}
|
||||
|
||||
static String payload2Str(ByteBuffer content) {
|
||||
byte[] b = new byte[content.remaining()];
|
||||
content.mark();
|
||||
content.get(b);
|
||||
content.reset();
|
||||
return new String(b);
|
||||
}
|
||||
|
||||
private void handleConnect(ChannelHandlerContext ctx, ConnectMessage msg) {
|
||||
String clientID = msg.getClientID();
|
||||
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
|
||||
ConnAckMessage koResp = new ConnAckMessage();
|
||||
koResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
|
||||
ctx.writeAndFlush(koResp);
|
||||
ctx.close();
|
||||
}
|
||||
|
||||
ConnAckMessage okResp = new ConnAckMessage();
|
||||
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
|
||||
ctx.writeAndFlush(okResp);
|
||||
}
|
||||
|
||||
@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(/*false*/);
|
||||
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();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
package io.moquette.parser.netty.performance;
|
||||
|
||||
import io.moquette.parser.commons.Constants;
|
||||
import io.moquette.parser.netty.MQTTDecoder;
|
||||
import io.moquette.parser.netty.MQTTEncoder;
|
||||
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.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;
|
||||
|
||||
/**
|
||||
* 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 {
|
||||
|
||||
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.ALL_IDLE) {
|
||||
//fire a channelInactive to trigger publish of Will
|
||||
ctx.fireChannelInactive();
|
||||
ctx.close();
|
||||
} /*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 = false;
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
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(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
|
||||
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler());
|
||||
pipeline.addLast("decoder", new MQTTDecoder());
|
||||
pipeline.addLast("encoder", new MQTTEncoder());
|
||||
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)
|
||||
.option(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();
|
||||
} catch (InterruptedException ex) {
|
||||
LOG.error(null, ex);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
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 {
|
||||
final ProtocolDecodingServer server = new ProtocolDecodingServer();
|
||||
server.init();
|
||||
|
||||
System.out.println("Loop server started");
|
||||
startClientsTesting();
|
||||
//Bind a shutdown hook
|
||||
Runtime.getRuntime().addShutdownHook(new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
server.stop();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private static void startClientsTesting() throws MqttException, InterruptedException {
|
||||
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");
|
||||
}
|
||||
}
|
||||
+107
@@ -0,0 +1,107 @@
|
||||
package io.moquette.parser.netty.performance;
|
||||
|
||||
import io.moquette.parser.netty.MQTTDecoder;
|
||||
import io.moquette.parser.netty.MQTTEncoder;
|
||||
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.timeout.IdleState;
|
||||
import io.netty.handler.timeout.IdleStateEvent;
|
||||
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;
|
||||
|
||||
/**
|
||||
* 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", new MQTTEncoder());
|
||||
pipeline.addLast("handler", new PublishReceiverHandler());
|
||||
} catch (Throwable th) {
|
||||
LOG.error("Severe error during pipeline creation", th);
|
||||
throw th;
|
||||
}
|
||||
}
|
||||
})
|
||||
.option(ChannelOption.SO_BACKLOG, 128)
|
||||
.option(ChannelOption.SO_REUSEADDR, true)
|
||||
.option(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();
|
||||
} catch (InterruptedException ex) {
|
||||
LOG.error(null, ex);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
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 {
|
||||
final ProtocolPublishDecodingServer server = new ProtocolPublishDecodingServer();
|
||||
server.init();
|
||||
|
||||
System.out.println("Loop server started");
|
||||
publishBombing();
|
||||
publishBombing();
|
||||
publishBombing();
|
||||
//Bind a shutdown hook
|
||||
Runtime.getRuntime().addShutdownHook(new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
server.stop();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private static void publishBombing() {
|
||||
LOG.info("Started publish loop");
|
||||
PublishBomber heavyPublisher = new PublishBomber("localhost", 1883);
|
||||
heavyPublisher.publishLoop(10000, 10000);
|
||||
heavyPublisher.disconnect();
|
||||
LOG.info("Finished publish loop");
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package io.moquette.parser.netty.performance;
|
||||
|
||||
|
||||
import io.moquette.parser.netty.MQTTDecoder;
|
||||
import io.moquette.parser.netty.MQTTEncoder;
|
||||
import io.moquette.parser.proto.messages.AbstractMessage;
|
||||
import io.moquette.parser.proto.messages.PublishMessage;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.*;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.SocketChannel;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import org.eclipse.jetty.toolchain.perf.PlatformTimer;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
class PublishBomber {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(PublishBomber.class);
|
||||
|
||||
private final EventLoopGroup workerGroup;
|
||||
private Channel channel;
|
||||
|
||||
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", new MQTTEncoder());
|
||||
}
|
||||
});
|
||||
|
||||
// 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(AbstractMessage msg) {
|
||||
try {
|
||||
channel.writeAndFlush(msg).await();
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public void publishLoop(int messagesPerSecond, int numToSend) {
|
||||
long pauseMicroseconds = (1 / messagesPerSecond) * 1000 * 1000;
|
||||
LOG.info("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();
|
||||
|
||||
PublishMessage pubMessage = new PublishMessage();
|
||||
pubMessage.setQos(AbstractMessage.QOSType.MOST_ONE);
|
||||
pubMessage.setTopicName("/topic");
|
||||
byte[] rawContent = ("Hello world!!-" + nanos).getBytes();
|
||||
ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(rawContent.length).put(rawContent).flip();
|
||||
pubMessage.setPayload(payload);
|
||||
sendMessage(pubMessage);
|
||||
timer.sleep(pauseMicroseconds);
|
||||
}
|
||||
LOG.info("PUB: published in {} ms", System.currentTimeMillis() - startTime);
|
||||
}
|
||||
|
||||
public void disconnect() {
|
||||
try {
|
||||
this.channel.disconnect().await();
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package io.moquette.parser.netty.performance;
|
||||
|
||||
import io.moquette.parser.proto.Utils;
|
||||
import io.moquette.parser.proto.messages.*;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.ChannelInboundHandlerAdapter;
|
||||
import io.netty.util.AttributeKey;
|
||||
import org.HdrHistogram.Histogram;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import static io.moquette.parser.proto.messages.AbstractMessage.*;
|
||||
|
||||
@ChannelHandler.Sharable
|
||||
class PublishReceiverHandler extends ChannelInboundHandlerAdapter {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(PublishReceiverHandler.class);
|
||||
Histogram forthNetworkTime = new Histogram(5);
|
||||
|
||||
public PublishReceiverHandler() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object message) {
|
||||
AbstractMessage msg = (AbstractMessage) message;
|
||||
|
||||
try {
|
||||
switch (msg.getMessageType()) {
|
||||
case PUBLISH:
|
||||
LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
|
||||
handlePublish((PublishMessage) msg);
|
||||
return;
|
||||
default:
|
||||
LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
LOG.error("Bad error in processing the message", ex);
|
||||
}
|
||||
}
|
||||
|
||||
private void handlePublish(PublishMessage msg) {
|
||||
long start = System.nanoTime();
|
||||
LOG.debug("push forward message the topic {}", msg.getTopicName());
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("content <{}>", payload2Str(msg.getPayload()));
|
||||
}
|
||||
String decodedPayload = payload2Str(msg.getPayload());
|
||||
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.getPayload()));
|
||||
}
|
||||
|
||||
static String payload2Str(ByteBuffer content) {
|
||||
byte[] b = new byte[content.remaining()];
|
||||
content.mark();
|
||||
content.get(b);
|
||||
content.reset();
|
||||
return new String(b);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
||||
LOG.info("Received channel inactive");
|
||||
ctx.channel().close();
|
||||
|
||||
System.out.println("Network time histogram (microsecs)");
|
||||
this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0);
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user