From fdbee3a21f049b40d196ecd5404f77fb958a6f5b Mon Sep 17 00:00:00 2001 From: andrea Date: Tue, 18 Aug 2015 23:50:21 +0200 Subject: [PATCH] Added shutdown sequence of the disruptor used in the output part (from ProtocolProcessor down to the cable) --- ChangeLog.txt | 1 + .../org/eclipse/moquette/server/Server.java | 2 +- .../moquette/server/netty/NettyAcceptor.java | 19 +++++++++++++++++-- .../moquette/spi/impl/ProtocolProcessor.java | 19 ++++++++++++------- .../moquette/spi/impl/SimpleMessaging.java | 7 ++++--- 5 files changed, 35 insertions(+), 13 deletions(-) diff --git a/ChangeLog.txt b/ChangeLog.txt index 94e1c014..8e3fa985 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -4,6 +4,7 @@ Version 0.8: Fixed bug, does not check for zero length topic (filters) in PUBLISH, SUBSCRIBE, and UNSUBSCRIBE, issue #67 Fixed issue #69 Fixed bug #68, UNSUBSCRIBE has to fail on bad topicFilter names + Fixed bug #81, on stop of moquette broker it doesn't shutdown due to no shutdown of ringbuffer used on the back of ProtocolProcessor Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/broker/src/main/java/org/eclipse/moquette/server/Server.java b/broker/src/main/java/org/eclipse/moquette/server/Server.java index 57a7bed0..4cf80b4b 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/Server.java +++ b/broker/src/main/java/org/eclipse/moquette/server/Server.java @@ -103,8 +103,8 @@ public class Server { public void stopServer() { LOG.info("Server stopping..."); - messaging.stop(); m_acceptor.close(); + messaging.stop(); LOG.info("Server stopped"); } diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java index 985f756b..36dd03a7 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java @@ -46,6 +46,9 @@ import java.util.Properties; import javax.net.ssl.KeyManagerFactory; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; + +import io.netty.util.concurrent.EventExecutor; +import io.netty.util.concurrent.Future; import org.eclipse.moquette.commons.Constants; import org.eclipse.moquette.spi.IMessaging; import org.eclipse.moquette.parser.netty.MQTTDecoder; @@ -270,8 +273,20 @@ public class NettyAcceptor implements ServerAcceptor { if (m_bossGroup == null) { throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); } - m_workerGroup.shutdownGracefully(); - m_bossGroup.shutdownGracefully(); + Future workerWaiter = m_workerGroup.shutdownGracefully(); + Future bossWaiter = m_bossGroup.shutdownGracefully(); + + try { + workerWaiter.await(100); + } catch (InterruptedException iex) { + throw new IllegalStateException(iex); + } + + try { + bossWaiter.await(100); + } catch (InterruptedException iex) { + throw new IllegalStateException(iex); + } MessageMetrics metrics = m_metricsCollector.computeMetrics(); LOG.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote()); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index a7077088..7d448244 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -19,7 +19,6 @@ import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor; import java.nio.ByteBuffer; -import java.text.ParseException; import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -63,7 +62,7 @@ import org.slf4j.LoggerFactory; * @author andrea */ class ProtocolProcessor implements EventHandler { - + static final class WillMessage { private final String topic; private final ByteBuffer payload; @@ -111,9 +110,10 @@ class ProtocolProcessor implements EventHandler { private ExecutorService m_executor; private RingBuffer m_ringBuffer; + private Disruptor m_disruptor; ProtocolProcessor() {} - + /** * @param subscriptions the subscription store where are stored all the existing * clients subscriptions. @@ -142,12 +142,17 @@ class ProtocolProcessor implements EventHandler { //init the output ringbuffer m_executor = Executors.newFixedThreadPool(1); - Disruptor disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); - disruptor.handleEventsWith(this); - disruptor.start(); + /*Disruptor*/ m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); + m_disruptor.handleEventsWith(this); + m_disruptor.start(); // Get the ring buffer from the Disruptor to be used for publishing. - m_ringBuffer = disruptor.getRingBuffer(); + m_ringBuffer = m_disruptor.getRingBuffer(); + } + + void stop() { + m_executor.shutdown(); + m_disruptor.shutdown(); } @MQTTMessage(message = ConnectMessage.class) diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java index d75909d2..4f7ec458 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java @@ -297,14 +297,15 @@ public class SimpleMessaging implements IMessaging, EventHandler { private void processStop() { LOG.debug("processStop invoked"); + m_processor.stop(); + if (m_interceptor != null) { this.m_interceptor.stop(); + LOG.debug("interceptor firer stopped"); } m_storageService.close(); LOG.debug("subscription tree {}", subscriptions.dumpTree()); -// m_eventProcessor.halt(); -// m_executor.shutdown(); - + subscriptions = null; m_stopLatch.countDown();