From 642fa4a4aad91afc35c9b2a6b45867edf81065c6 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 26 Dec 2016 12:13:30 +0100 Subject: [PATCH] Added plain publish bomber test, to measure enoding/deconding perf --- .../netty/performance/BenchmarkPublisher.java | 17 +-- .../performance/BenchmarkSubscriber.java | 4 +- ...{LoopHandler.java => LoopMQTTHandler.java} | 13 ++- .../performance/ProtocolDecodingServer.java | 2 +- .../ProtocolPublishDecodingServer.java | 107 ++++++++++++++++++ .../netty/performance/PublishBomber.java | 90 +++++++++++++++ .../performance/PublishReceiverHandler.java | 77 +++++++++++++ .../src/test/resources/log4j.properties | 2 +- 8 files changed, 300 insertions(+), 12 deletions(-) rename netty_parser/src/test/java/io/moquette/parser/netty/performance/{LoopHandler.java => LoopMQTTHandler.java} (93%) create mode 100644 netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java create mode 100644 netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishBomber.java create mode 100644 netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java index 8f8f4bb6..3abac23f 100644 --- a/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java +++ b/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java @@ -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); + } } } diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java index f1d77050..8b9eb14c 100644 --- a/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java +++ b/netty_parser/src/test/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java @@ -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); } } \ No newline at end of file diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopHandler.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java similarity index 93% rename from netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopHandler.java rename to netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java index b80bcbc0..8aa64098 100644 --- a/netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopHandler.java +++ b/netty_parser/src/test/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java @@ -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); // } diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java index e16ee53c..90d6a43c 100644 --- a/netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java +++ b/netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java @@ -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"); } } diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java new file mode 100644 index 00000000..76673f1a --- /dev/null +++ b/netty_parser/src/test/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java @@ -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() { + @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"); + } + +} diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishBomber.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishBomber.java new file mode 100644 index 00000000..66478e31 --- /dev/null +++ b/netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishBomber.java @@ -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() { + @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(); + } + } +} diff --git a/netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java b/netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java new file mode 100644 index 00000000..270328e7 --- /dev/null +++ b/netty_parser/src/test/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java @@ -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); + } + +} \ No newline at end of file diff --git a/netty_parser/src/test/resources/log4j.properties b/netty_parser/src/test/resources/log4j.properties index e7e1498f..2ee59103 100644 --- a/netty_parser/src/test/resources/log4j.properties +++ b/netty_parser/src/test/resources/log4j.properties @@ -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