From 3f4a5586de4eb4907b42b10ecf0f9c2443f5ff9f Mon Sep 17 00:00:00 2001 From: andrea Date: Sun, 7 Jun 2015 17:47:50 +0200 Subject: [PATCH] Fixed mem leak due to caching of object without hasCode (NettyChannelContext --- ChangeLog.txt | 1 + .../server/netty/NettyMQTTHandler.java | 31 ++++++------ .../moquette/server/netty/NettyUtils.java | 50 +++++++++++++++++++ .../org/eclipse/moquette/spi/IMessaging.java | 2 +- .../moquette/spi/impl/ProtocolProcessor.java | 12 ++--- .../moquette/spi/impl/SimpleMessaging.java | 4 +- .../spi/impl/events/LostConnectionEvent.java | 6 +-- .../spi/impl/DummyChannelHandlerContext.java | 7 +++ .../spi/impl/ProtocolProcessorTest.java | 41 --------------- .../issue39_loopingConnecter.groovy | 34 +++++++++++++ 10 files changed, 120 insertions(+), 68 deletions(-) create mode 100644 broker/src/main/java/org/eclipse/moquette/server/netty/NettyUtils.java create mode 100644 broker/src/test/java/org/eclipse/moquette/spi/impl/DummyChannelHandlerContext.java create mode 100644 tools_scripts/integration/issue39_loopingConnecter.groovy diff --git a/ChangeLog.txt b/ChangeLog.txt index da60e2e1..e2c8e115 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -13,6 +13,7 @@ Version 0.7: Added handling of allow_anonymous clients (as in mosquitto issue #34) Modified the password_file format to use password stored in sha256 and not plain text Added ACL file for topic read|write operations. + Removed cause of leakage of Netty Sessions (fix issue #39) Version 0.6: Fix issue #42, bad password file path composition diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java index f19e2fe0..bb5a82f2 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java @@ -37,7 +37,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class); private IMessaging m_messaging; - private final Map m_channelMapper = new HashMap(); +// private final Map m_channelMapper = new HashMap<>(); @Override public void channelRead(ChannelHandlerContext ctx, Object message) { @@ -54,15 +54,15 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { case PUBREL: case DISCONNECT: case PUBACK: - NettyChannel channel; - synchronized(m_channelMapper) { - if (!m_channelMapper.containsKey(ctx)) { - m_channelMapper.put(ctx, new NettyChannel(ctx)); - } - channel = m_channelMapper.get(ctx); - } +// 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); + m_messaging.handleProtocolMessage(new NettyChannel(ctx), msg); break; case PINGREQ: PingRespMessage pingResp = new PingRespMessage(); @@ -76,13 +76,14 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - NettyChannel channel = m_channelMapper.get(ctx); - String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); - m_messaging.lostConnection(channel, clientID); +// NettyChannel channel = m_channelMapper.get(ctx); +// String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + String clientID = (String) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_CLIENTID); + m_messaging.lostConnection(clientID); ctx.close(/*false*/); - synchronized(m_channelMapper) { - m_channelMapper.remove(ctx); - } +// synchronized(m_channelMapper) { +// m_channelMapper.remove(ctx); +// } } public void setMessaging(IMessaging messaging) { diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyUtils.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyUtils.java new file mode 100644 index 00000000..ca7ab882 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyUtils.java @@ -0,0 +1,50 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.eclipse.moquette.server.netty; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.timeout.IdleStateHandler; +import io.netty.util.Attribute; +import io.netty.util.AttributeKey; + +/** + * Some Netty's channels utilities. + * + * @author andrea + */ +public class NettyUtils { + + public static Object getAttribute(ChannelHandlerContext channel, AttributeKey key) { + Attribute attr = channel.attr(key); + return attr.get(); + } + + public static void setAttribute(ChannelHandlerContext channel, AttributeKey key, Object value) { + Attribute attr = channel.attr(key); + attr.set(value); + } + + public static void setIdleTime(ChannelHandlerContext channel, int idleTime) { + if (channel.pipeline().names().contains("idleStateHandler")) { + channel.pipeline().remove("idleStateHandler"); + } + if (channel.pipeline().names().contains("idleEventHandler")) { + channel.pipeline().remove("idleEventHandler"); + } + channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime)); + channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java index 8d009db5..3bf8f637 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java @@ -30,7 +30,7 @@ public interface IMessaging { void stop(); - void lostConnection(ServerChannel session, String clientID); + void lostConnection(String clientID); void handleProtocolMessage(ServerChannel session, AbstractMessage msg); } 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 0222eb2c..01b20713 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 @@ -561,12 +561,12 @@ class ProtocolProcessor implements EventHandler { void processConnectionLost(LostConnectionEvent evt) { String clientID = evt.clientID; - if (m_clientIDs.containsKey(clientID)) { - if (!m_clientIDs.get(clientID).getSession().equals(evt.session)) { - LOG.info("Received a lost connection with client <{}> for a not matching session", clientID); - return; - } - } +// if (m_clientIDs.containsKey(clientID)) { +// if (!m_clientIDs.get(clientID).getSession().equals(evt.session)) { +// LOG.info("Received a lost connection with client <{}> for a not matching session", clientID); +// return; +// } +// } //If already removed a disconnect message was already processed for this clientID if (m_clientIDs.remove(clientID) != null) { 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 89346822..ac8c3efe 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 @@ -122,8 +122,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { } @Override - public void lostConnection(ServerChannel session, String clientID) { - disruptorPublish(new LostConnectionEvent(session, clientID)); + public void lostConnection(String clientID) { + disruptorPublish(new LostConnectionEvent(clientID)); } @Override diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/LostConnectionEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/LostConnectionEvent.java index e0dffabc..aabd9bae 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/LostConnectionEvent.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/LostConnectionEvent.java @@ -23,11 +23,11 @@ import org.eclipse.moquette.server.ServerChannel; * @author andrea */ public class LostConnectionEvent extends MessagingEvent { - public final ServerChannel session; +// public final ServerChannel session; public final String clientID; - public LostConnectionEvent(ServerChannel session, String clientID) { - this.session = session; + public LostConnectionEvent(/*ServerChannel session, */String clientID) { +// this.session = session; this.clientID = clientID; } } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/DummyChannelHandlerContext.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/DummyChannelHandlerContext.java new file mode 100644 index 00000000..ca1f24a3 --- /dev/null +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/DummyChannelHandlerContext.java @@ -0,0 +1,7 @@ +package org.eclipse.moquette.spi.impl; + +/** + * Created by andrea on 6/7/15. + */ +public class DummyChannelHandlerContext { +} diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 7326bff2..2db13398 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -634,45 +634,4 @@ public class ProtocolProcessorTest { }); assertTrue(messages.isEmpty()); } - - - /** - * Simulate a client1 (FAKE_CLIENT_ID) bound with session1, an event of connection lost - * for client2 (FAKE_CLIENT_ID, same of the client1) but from another session, verify - * the session of client1 is not closed. - * */ - @Test - public void testConnectionLostClosesTheCorrectSession() { - MockReceiverChannel channel1 = new MockReceiverChannel(); - - //init the processor - /*SubscriptionsStore subs = new SubscriptionsStore(); - subs.init(new MemoryStorageService());*/ - SubscriptionsStore subs = mock(SubscriptionsStore.class); - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator()); - - //simulate a connect from client1 FAKE_CLIENT_ID from channel1 - ConnectMessage connectMessage = new ConnectMessage(); - connectMessage.setProcotolVersion((byte) 3); - connectMessage.setClientID(FAKE_CLIENT_ID); - m_processor.processConnect(channel1, connectMessage); - assertConnectReturnCode(ConnAckMessage.CONNECTION_ACCEPTED, channel1); - //ConnAck received - - //send a connection lost event from an already disconnected client, but with same clientID (FAKE_CLIENT_ID) - //Exercise - DummyChannel channel2 = new DummyChannel(); - LostConnectionEvent lostConnectionEvent = new LostConnectionEvent(channel2, FAKE_CLIENT_ID); - m_processor.processConnectionLost(lostConnectionEvent); - - //Verify no subscriptions were deactivated for the client on session1 - verify(subs, never()).deactivate(anyString()); - } - - private void assertConnectReturnCode(byte expectedReturnCode, MockReceiverChannel receiverSession) { - AbstractMessage recvMsg = receiverSession.getMessage(); - assertTrue(recvMsg instanceof ConnAckMessage); - ConnAckMessage connAckMsg = (ConnAckMessage) recvMsg; - assertEquals(expectedReturnCode, connAckMsg.getReturnCode()); - } } diff --git a/tools_scripts/integration/issue39_loopingConnecter.groovy b/tools_scripts/integration/issue39_loopingConnecter.groovy new file mode 100644 index 00000000..5e215c77 --- /dev/null +++ b/tools_scripts/integration/issue39_loopingConnecter.groovy @@ -0,0 +1,34 @@ +@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/') +@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.0.1', ext='jar') + + +import org.eclipse.paho.client.mqttv3.MqttClient +import org.eclipse.paho.client.mqttv3.MqttMessage +import org.eclipse.paho.client.mqttv3.MqttException +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence + +/** + * A client that connect and disconnect multiple times, to check that on disconnection there isn't + * any resource leakage. + * */ + +if (args.size() != 3) { + println "usage: " + return +} + +String host = args[0] +long numSerialConnection = args[1] as long +String clientName = args[2] + +String tmpDir = System.getProperty("java.io.tmpdir") +MqttDefaultFilePersistence dataStore1 = new MqttDefaultFilePersistence(tmpDir + "/" + clientName) +MqttClient client1 = new MqttClient("tcp://${host}:1883", "ConCli${clientName}", dataStore1) + +println "start loppoing to ${numSerialConnection} conn-discon" +(1..numSerialConnection).each { + client1.connect() + client1.disconnect() + print "." +} +println "Done" \ No newline at end of file