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

Added plain publish bomber test, to measure enoding/deconding perf

Browse Source
This commit is contained in:
andsel
2016-12-26 12:13:30 +01:00
parent 686c5f4c65
commit 642fa4a4aa
8 changed files with 300 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java
+10 -7
View File
@@ -42,11 +42,11 @@ class BenchmarkPublisher {
long spentTime = stopTime - startedTime;
LOG.info("PUB: published in {} ms", spentTime);
try {
this.connection.disconnect();
} catch (MqttException mex) {
LOG.error("Disconnect error", mex);
}
// 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();
@@ -101,12 +101,10 @@ class BenchmarkPublisher {
//initialize the timer
PlatformTimer timer = PlatformTimer.detect();
IMqttActionListener pubCallback = new PublishCallback();
// println ""
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);
// print '.'
timer.sleep(pauseMicroseconds);
}
@@ -119,6 +117,11 @@ class BenchmarkPublisher {
public void waitFinish() throws InterruptedException {
m_latch.await();
try {
this.client.disconnect();
} catch (MqttException mex) {
LOG.error("Disconnect error", mex);
}
}
}
netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java
+3 -1
View File
@@ -40,6 +40,7 @@ class BenchmarkSubscriber {
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);
@@ -137,8 +138,9 @@ class BenchmarkSubscriber {
}
public void waitFinish() throws InterruptedException {
public void waitFinish() throws InterruptedException, MqttException {
this.m_latch.await();
this.client.disconnect(1000);
}
}
netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopHandler.java → netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java
+11 -2
View File
@@ -24,6 +24,7 @@ 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;
@@ -87,6 +88,10 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
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);
@@ -115,7 +120,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
LOG.debug("subscribed client to {}", msg.subscriptions());
}
static String payload2Str(ByteBuffer content) {
static String payload2Str(ByteBuffer content) {
byte[] b = new byte[content.remaining()];
content.mark();
content.get(b);
@@ -150,8 +155,12 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
// String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
// m_messaging.lostConnection(channel, clientID);
ctx.channel().close(/*false*/);
System.out.println("Processing histogram (microsecs)");
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);
// }
netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java
+1 -1
View File
@@ -133,7 +133,6 @@ public class ProtocolDecodingServer {
System.out.println("Loop server started");
startClientsTesting();
System.out.println("Started clients (sub/pub) for testing");
//Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
@@ -164,5 +163,6 @@ public class ProtocolDecodingServer {
suscriberBench.waitFinish();
publisherBench.waitFinish();
System.out.println("Started clients (sub/pub) for testing");
}
}
netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java
+107
View File
@@ -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");
}
}
netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishBomber.java
+90
View File
@@ -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();
}
}
}
netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java
+77
View File
@@ -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);
}
}
netty_parser/src/test/resources/log4j.properties
+1 -1
View File
@@ -4,7 +4,7 @@ log4j.rootLogger=INFO, stdout
log4j.logger.io.moquette=WARN
log4j.logger.io.netty.util=WARN
log4j.logger.io.moquette.parser.netty.performance=INFO
log4j.logger.io.moquette.parser.netty.performance=WARN
# stdout appender is set to be a ConsoleAppender.
log4j.appender.stdout=org.apache.log4j.ConsoleAppender