From 4e2a0dbf28e7493844b46116e098ad58acbb24aa Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 27 Dec 2016 22:28:34 +0100 Subject: [PATCH] Switched to netty 4.1.6 to include Netty's official mqtt codec to make some performance tests --- perf/pom.xml | 5 ++ .../netty/performance/BenchmarkPublisher.java | 2 +- .../NettyPublishReceiverHandler.java | 80 +++++++++++++++++++ .../ProtocolPublishDecodingServer.java | 13 +-- .../netty/performance/PublishBomber.java | 4 +- pom.xml | 2 +- 6 files changed, 97 insertions(+), 9 deletions(-) create mode 100644 perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java diff --git a/perf/pom.xml b/perf/pom.xml index 2c06cce9..43d83582 100644 --- a/perf/pom.xml +++ b/perf/pom.xml @@ -54,6 +54,11 @@ slf4j-log4j12 1.6.4 + + io.netty + netty-codec-mqtt + ${netty.version} + org.hdrhistogram diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java b/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java index 3abac23f..c974d4ff 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java @@ -91,7 +91,7 @@ class BenchmarkPublisher { } public void firePublishes() throws MqttException { - long pauseMicroseconds = (1 / this.messagesPerSecond) * 1000 * 1000; + long pauseMicroseconds = (int)((1.0 / messagesPerSecond) * 1000 * 1000); LOG.info("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds); LOG.info("PUB: publishing.."); diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java new file mode 100644 index 00000000..12a268fc --- /dev/null +++ b/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java @@ -0,0 +1,80 @@ +package io.moquette.parser.netty.performance; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +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; + + +@ChannelHandler.Sharable +class NettyPublishReceiverHandler extends ChannelInboundHandlerAdapter { + + private static final Logger LOG = LoggerFactory.getLogger(NettyPublishReceiverHandler.class); + Histogram forthNetworkTime = new Histogram(5); + + public 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()); + if (LOG.isDebugEnabled()) { + 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); + } + + + @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/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java index 76673f1a..f81f21dc 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java @@ -7,11 +7,9 @@ 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 io.netty.handler.codec.mqtt.MqttDecoder; +import io.netty.handler.codec.mqtt.MqttEncoder; import org.eclipse.paho.client.mqttv3.MqttException; -import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -44,6 +42,9 @@ public class ProtocolPublishDecodingServer { pipeline.addLast("decoder", new MQTTDecoder()); pipeline.addLast("encoder", new MQTTEncoder()); 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; @@ -87,6 +88,8 @@ public class ProtocolPublishDecodingServer { publishBombing(); publishBombing(); publishBombing(); + publishBombing(); + publishBombing(); //Bind a shutdown hook Runtime.getRuntime().addShutdownHook(new Thread() { @Override @@ -99,7 +102,7 @@ public class ProtocolPublishDecodingServer { private static void publishBombing() { LOG.info("Started publish loop"); PublishBomber heavyPublisher = new PublishBomber("localhost", 1883); - heavyPublisher.publishLoop(10000, 10000); + heavyPublisher.publishLoop(25000, 100000); heavyPublisher.disconnect(); LOG.info("Finished publish loop"); } diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java b/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java index 66478e31..014ee916 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java @@ -57,8 +57,8 @@ class PublishBomber { public void publishLoop(int messagesPerSecond, int numToSend) { - long pauseMicroseconds = (1 / messagesPerSecond) * 1000 * 1000; - LOG.info("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds); + 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(); diff --git a/pom.xml b/pom.xml index f8a72d80..1ffe9415 100644 --- a/pom.xml +++ b/pom.xml @@ -3,7 +3,7 @@ 4.0.0 UTF-8 - 4.0.33.Final + 4.1.6.Final 1.7 1.7 andsel/maven