Fixing a memory leak in qpid layer. (#193)

yvgopal · web-flow · commit e7fcc0bfde41 · 2018-03-02T20:58:27.000-08:00
* Fixing a memory leak in qpid layer. When service force closes an idle connetion, it is not properly freed in qpid layer.
Includes some minor changes to make code simpler.

* Fixing a code review comment.
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/InitializableEntity.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/InitializableEntity.java
@@ -15,8 +15,8 @@
 abstract class InitializableEntity extends ClientEntity {
 
     //TODO Init and close semantics are primitive now. Fix them with support for other states like Initializing, Closing, and concurrency.
-    protected InitializableEntity(String clientId, ClientEntity parent) {
-        super(clientId, parent);
+    protected InitializableEntity(String clientId) {
+        super(clientId);
     }
 
     /**
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java
@@ -48,7 +48,7 @@ class MessageAndSessionPump extends InitializableEntity implements IMessageAndSe
     private int prefetchCount;
 
     public MessageAndSessionPump(MessagingFactory factory, String entityPath, ReceiveMode receiveMode) {
-        super(StringUtil.getShortRandomString(), null);
+        super(StringUtil.getShortRandomString());
         this.factory = factory;
         this.entityPath = entityPath;
         this.receiveMode = receiveMode;
@@ -92,7 +92,7 @@ public void registerSessionHandler(ISessionHandler handler, SessionHandlerOption
         this.sessionHandlerOptions = handlerOptions;
 
         for (int i = 0; i < handlerOptions.getMaxConcurrentSessions(); i++) {
-            this.acceptSessionsAndPumpMessage();
+            this.acceptSessionAndPumpMessages();
         }
     }
 
@@ -142,6 +142,12 @@ private void receiveAndPumpMessage() {
                             onMessageFuture = new CompletableFuture<Void>();
                             onMessageFuture.completeExceptionally(onMessageSyncEx);
                         }
+                        
+                        // Some clients are returning null from the call
+                        if(onMessageFuture == null)
+                        {
+                            onMessageFuture = CompletableFuture.completedFuture(null);
+                        }
 
                         onMessageFuture.handleAsync((v, onMessageEx) -> {
                             if (onMessageEx != null) {
@@ -203,7 +209,7 @@ private void receiveAndPumpMessage() {
         }
     }
 
-    private void acceptSessionsAndPumpMessage() {
+    private void acceptSessionAndPumpMessages() {
         if (!this.getIsClosingOrClosed()) {
             TRACE_LOGGER.debug("Accepting a session from entity '{}'", this.entityPath);
             CompletableFuture<IMessageSession> acceptSessionFuture = ClientFactory.acceptSessionFromEntityPathAsync(this.factory, this.entityPath, null, this.receiveMode);
@@ -222,7 +228,7 @@ private void acceptSessionsAndPumpMessage() {
                         // In case of any other exception, sleep and retry
                         TRACE_LOGGER.debug("AcceptSession from entity '{}' will be retried after '{}'.", this.entityPath, SLEEP_DURATION_ON_ACCEPT_SESSION_EXCEPTION);
                         Timer.schedule(() -> {
-                            MessageAndSessionPump.this.acceptSessionsAndPumpMessage();
+                            MessageAndSessionPump.this.acceptSessionAndPumpMessages();
                         }, SLEEP_DURATION_ON_ACCEPT_SESSION_EXCEPTION, TimerType.OneTimeRun);
                     }
                 } else {
@@ -293,6 +299,12 @@ private void receiveFromSessionAndPumpMessage(SessionTracker sessionTracker) {
                             onMessageFuture.completeExceptionally(onMessageSyncEx);
                         }
 
+                        // Some clients are returning null from the call
+                        if(onMessageFuture == null)
+                        {
+                            onMessageFuture = CompletableFuture.completedFuture(null);
+                        }
+                        
                         onMessageFuture.handleAsync((v, onMessageEx) -> {
                             renewCancelTimer.cancel(true);
                             if (onMessageEx != null) {
@@ -419,6 +431,12 @@ synchronized CompletableFuture<Boolean> shouldRetryOnNoMessageOrException() {
                     onCloseFuture = new CompletableFuture<Void>();
                     onCloseFuture.completeExceptionally(onCloseSyncEx);
                 }
+                
+                // Some clients are returning null from the call
+                if(onCloseFuture == null)
+                {
+                    onCloseFuture = CompletableFuture.completedFuture(null);
+                }
 
                 onCloseFuture.handleAsync((v, onCloseEx) -> {
                     renewCancelTimer.cancel(true);
@@ -441,7 +459,7 @@ synchronized CompletableFuture<Boolean> shouldRetryOnNoMessageOrException() {
                         }
 
                         this.messageAndSessionPump.openSessions.remove(this.session.getSessionId());
-                        this.messageAndSessionPump.acceptSessionsAndPumpMessage();
+                        this.messageAndSessionPump.acceptSessionAndPumpMessages();
                         return null;
                     });
                     return null;
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageReceiver.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageReceiver.java
@@ -56,7 +56,7 @@ class MessageReceiver extends InitializableEntity implements IMessageReceiver, I
     private final ConcurrentHashMap<UUID, Instant> requestResponseLockTokensToLockTimesMap;
 
     private MessageReceiver(ReceiveMode receiveMode) {
-        super(StringUtil.getShortRandomString(), null);
+        super(StringUtil.getShortRandomString());
         this.receiveMode = receiveMode;
         this.requestResponseLockTokensToLockTimesMap = new ConcurrentHashMap<>();
         if (receiveMode == ReceiveMode.PEEKLOCK) {
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java
@@ -29,7 +29,7 @@ final class MessageSender extends InitializableEntity implements IMessageSender
     private ClientSettings clientSettings;
 
     private MessageSender() {
-        super(StringUtil.getShortRandomString(), null);
+        super(StringUtil.getShortRandomString());
     }
 
     MessageSender(URI namespaceEndpointURI, String entityPath, ClientSettings clientSettings) {
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/QueueClient.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/QueueClient.java
@@ -36,7 +36,7 @@ public final class QueueClient extends InitializableEntity implements IQueueClie
     private MiscRequestResponseOperationHandler miscRequestResponseHandler;
 
     private QueueClient(ReceiveMode receiveMode, String queuePath) {
-        super(StringUtil.getShortRandomString(), null);
+        super(StringUtil.getShortRandomString());
         this.receiveMode = receiveMode;
         this.queuePath = queuePath;
         this.senderCreationLock = new Object();
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/SubscriptionClient.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/SubscriptionClient.java
@@ -38,7 +38,7 @@ public final class SubscriptionClient extends InitializableEntity implements ISu
 
 	private SubscriptionClient(ReceiveMode receiveMode, String subscriptionPath)
 	{
-		super(StringUtil.getShortRandomString(), null);		
+		super(StringUtil.getShortRandomString());		
 		this.receiveMode = receiveMode;
 		this.subscriptionPath = subscriptionPath;
 	}
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/TopicClient.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/TopicClient.java
@@ -26,7 +26,7 @@ public final class TopicClient extends InitializableEntity implements ITopicClie
     private MessageBrowser browser;
 
     private TopicClient() {
-        super(StringUtil.getShortRandomString(), null);
+        super(StringUtil.getShortRandomString());
     }
 
     public TopicClient(ConnectionStringBuilder amqpConnectionStringBuilder) throws InterruptedException, ServiceBusException {
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/ConnectionHandler.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/ConnectionHandler.java
@@ -68,12 +68,6 @@ public void onConnectionBound(Event event)
 		sasl.setMechanisms("ANONYMOUS");
 	}
 
-	@Override
-	public void onConnectionUnbound(Event event)
-	{
-		TRACE_LOGGER.debug("Connection.onConnectionUnbound: hostname:{}", event.getConnection().getHostname());
-	}
-
 	@Override
 	public void onTransportError(Event event)
 	{
@@ -89,9 +83,10 @@ public void onTransportError(Event event)
 
 		this.messagingFactory.onConnectionError(condition);
 		Connection connection = event.getConnection();
-		if (connection != null) {
-            connection.free();
-        }
+		if(connection != null)
+		{
+		    connection.free();
+		}
 	}
 
 	@Override
@@ -108,38 +103,33 @@ public void onConnectionRemoteClose(Event event)
 		final ErrorCondition error = connection.getRemoteCondition();
 		
 		TRACE_LOGGER.debug("onConnectionRemoteClose: hostname:{},errorCondition:{}", connection.getHostname(), error != null ? error.getCondition() + "," + error.getDescription() : null);
-		
-		if (connection.getRemoteState() != EndpointState.CLOSED)
+		boolean shouldFreeConnection = connection.getLocalState() == EndpointState.CLOSED;		
+		this.messagingFactory.onConnectionError(error);
+		if(shouldFreeConnection)
 		{
-			connection.close();
+		    connection.free();
 		}
-
-		this.messagingFactory.onConnectionError(error);
-		this.freeOnCloseResponse(connection);
 	}
 	
 	@Override
     public void onConnectionFinal(Event event) {
-        final Transport transport = event.getTransport();
-        if (transport != null) {
-            transport.unbind();
-            transport.free();
-        }
+	    TRACE_LOGGER.debug("onConnectionFinal: hostname:{}", event.getConnection().getHostname());
     }
 	
 	@Override
     public void onConnectionLocalClose(Event event) {
 	    Connection connection = event.getConnection();
 	    TRACE_LOGGER.debug("onConnectionLocalClose: hostname:{}", connection.getHostname());
-        this.freeOnCloseResponse(connection);
-    }
-	
-	private void freeOnCloseResponse(Connection connection) {
-        if (connection != null &&
-                connection.getLocalState() == EndpointState.CLOSED &&
-                (connection.getRemoteState() == EndpointState.CLOSED)) {
-            connection.free();
-        }
+	    if(connection.getRemoteState() == EndpointState.CLOSED)
+	    {
+	        // Service closed it first. In some such cases transport is not unbound and causing a leak.
+	        if(connection.getTransport() != null)
+	        {
+	            connection.getTransport().unbind();
+	        }
+	        
+	        connection.free();
+	    }
     }
 
 	private static SslDomain makeDomain(SslDomain.Mode mode)
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/CustomIOHandler.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/CustomIOHandler.java
@@ -24,4 +24,13 @@ public void onConnectionLocalOpen(Event event)
 		transport.setEmitFlowEventOnSend(false);
 		transport.bind(connection);
 	}
+	
+	@Override
+    public void onTransportClosed(Event event)
+    {
+	    if(event.getTransport() != null)
+	    {
+	        event.getTransport().unbind();
+	    }
+    }
 }
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/ClientEntity.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/ClientEntity.java
@@ -18,16 +18,13 @@ public abstract class ClientEntity
 {
 	private final String clientId;
 	private final Object syncClose;
-	private final ClientEntity parent;
 
 	private boolean isClosing;
 	private boolean isClosed;
 
-	protected ClientEntity(final String clientId, final ClientEntity parent)
+	protected ClientEntity(final String clientId)
 	{
 		this.clientId = clientId;
-		this.parent = parent;
-
 		this.syncClose = new Object();
 	}
 
@@ -39,21 +36,19 @@ public String getClientId()
 	}
 
 	protected boolean getIsClosed()
-	{
-		final boolean isParentClosed = this.parent != null && this.parent.getIsClosed();
+	{		
 		synchronized (this.syncClose)
 		{
-			return isParentClosed || this.isClosed;
+			return this.isClosed;
 		}
 	}
-
-	// returns true even if the Parent is (being) Closed
+	
 	protected boolean getIsClosingOrClosed()
 	{
-		final boolean isParentClosingOrClosed = this.parent != null && this.parent.getIsClosingOrClosed();
+		
 		synchronized (this.syncClose)
 		{
-			return isParentClosingOrClosed || this.isClosing || this.isClosed;
+			return this.isClosing || this.isClosed;
 		}
 	}
 
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java
@@ -116,7 +116,7 @@ private CoreMessageReceiver(final MessagingFactory factory,
 			final int prefetchCount,
 			final SettleModePair settleModePair)
 	{
-		super(name, factory);
+		super(name);
 
 		this.underlyingFactory = factory;
 		this.operationTimeout = factory.getOperationTimeout();
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java
@@ -177,7 +177,7 @@ private void closeRequestResponseLink()
 
 	private CoreMessageSender(final MessagingFactory factory, final String sendLinkName, final String senderPath)
 	{
-		super(sendLinkName, factory);
+		super(sendLinkName);
 
 		this.sendPath = senderPath;
 		this.sasTokenAudienceURI = String.format(ClientConstants.SAS_TOKEN_AUDIENCE_FORMAT, factory.getHostName(), senderPath);
@@ -595,7 +595,7 @@ private void createSendLink()
 		
 		SendLinkHandler handler = new SendLinkHandler(CoreMessageSender.this);
 		BaseHandler.setHandler(sender, handler);
-		sender.open();		
+		sender.open();
 		this.sendLink = sender;
 	}
 	
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MessagingFactory.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MessagingFactory.java
@@ -74,7 +74,7 @@ public class MessagingFactory extends ClientEntity implements IAmqpConnection
 	
 	private MessagingFactory(URI namespaceEndpointUri, ClientSettings clientSettings)
 	{
-	    super("MessagingFactory".concat(StringUtil.getShortRandomString()), null);
+	    super("MessagingFactory".concat(StringUtil.getShortRandomString()));
 	    this.namespaceEndpointUri = namespaceEndpointUri;
 	    this.clientSettings = clientSettings;
 	    
@@ -303,7 +303,7 @@ public void onConnectionError(ErrorCondition error)
 		    TRACE_LOGGER.info("Connection to host closed.");
 		    this.connetionCloseFuture.complete(null);
 			Timer.unregister(this.getClientId());
-		}
+		} 
 	}
 
 	private void onReactorError(Exception cause)
@@ -341,28 +341,25 @@ private void onReactorError(Exception cause)
 	// One of the parameters must be null
 	private void closeConnection(ErrorCondition error, Exception cause)
 	{
-	    TRACE_LOGGER.info("Closing connection to host");
 	    // Important to copy the reference of the connection as a call to getConnection might create a new connection while we are still in this method
 	    Connection currentConnection = this.connection;
 	    if(currentConnection != null)
 	    {
 	        Link[] links = this.registeredLinks.toArray(new Link[0]);
+	        this.registeredLinks.clear();
 	        
 	        TRACE_LOGGER.debug("Closing all links on the connection. Number of links '{}'", links.length);
 	        for(Link link : links)
 	        {
-	            if (link.getLocalState() != EndpointState.CLOSED && link.getRemoteState() != EndpointState.CLOSED)
-	            {
-	                link.close();
-	            }
+	            link.close();
 	        }
 	        
 	        TRACE_LOGGER.debug("Closed all links on the connection. Number of links '{}'", links.length);
 
-	        if (currentConnection.getLocalState() != EndpointState.CLOSED && currentConnection.getRemoteState() != EndpointState.CLOSED)
+	        if (currentConnection.getLocalState() != EndpointState.CLOSED)
 	        {
+	            TRACE_LOGGER.info("Closing connection to host");
 	            currentConnection.close();
-	            currentConnection.free();
 	        }
 	        
 	        for(Link link : links)
@@ -399,7 +396,7 @@ protected CompletableFuture<Void> onClose()
 		    {
 		        TRACE_LOGGER.info("Closing CBS link");
 		        cbsLinkCloseFuture = this.cbsLink.closeAsync();
-		    }		    
+		    }
 		    
 		    cbsLinkCloseFuture.thenRun(() -> this.managementLinksCache.freeAsync()).thenRun(() -> {
 		        if(this.cbsLinkCreationFuture != null && !this.cbsLinkCreationFuture.isDone())
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MiscRequestResponseOperationHandler.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MiscRequestResponseOperationHandler.java
@@ -22,7 +22,7 @@ public final class MiscRequestResponseOperationHandler extends ClientEntity
 	
 	private MiscRequestResponseOperationHandler(MessagingFactory factory, String linkName, String entityPath)
 	{
-		super(linkName, factory);
+		super(linkName);
 		
 		this.underlyingFactory = factory;
 		this.entityPath = entityPath;
diff --git a/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/RequestResponseLink.java b/azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/RequestResponseLink.java

Original file line number	Diff line number	Diff line change
`@@ -15,8 +15,8 @@`
`15`	`15`	`abstract class InitializableEntity extends ClientEntity {`
`16`	`16`
`17`	`17`	`//TODO Init and close semantics are primitive now. Fix them with support for other states like Initializing, Closing, and concurrency.`
`18`		`- protected InitializableEntity(String clientId, ClientEntity parent) {`
`19`		`- super(clientId, parent);`
	`18`	`+ protected InitializableEntity(String clientId) {`
	`19`	`+ super(clientId);`
`20`	`20`	`}`
`21`	`21`
`22`	`22`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,7 @@ final class MessageSender extends InitializableEntity implements IMessageSender`
`29`	`29`	`private ClientSettings clientSettings;`
`30`	`30`
`31`	`31`	`private MessageSender() {`
`32`		`- super(StringUtil.getShortRandomString(), null);`
	`32`	`+ super(StringUtil.getShortRandomString());`
`33`	`33`	`}`
`34`	`34`
`35`	`35`	`MessageSender(URI namespaceEndpointURI, String entityPath, ClientSettings clientSettings) {`
Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ public final class SubscriptionClient extends InitializableEntity implements ISu`
`38`	`38`
`39`	`39`	`private SubscriptionClient(ReceiveMode receiveMode, String subscriptionPath)`
`40`	`40`	`{`
`41`		`- super(StringUtil.getShortRandomString(), null);`
	`41`	`+ super(StringUtil.getShortRandomString());`
`42`	`42`	`this.receiveMode = receiveMode;`
`43`	`43`	`this.subscriptionPath = subscriptionPath;`
`44`	`44`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@ public final class TopicClient extends InitializableEntity implements ITopicClie`
`26`	`26`	`private MessageBrowser browser;`
`27`	`27`
`28`	`28`	`private TopicClient() {`
`29`		`- super(StringUtil.getShortRandomString(), null);`
	`29`	`+ super(StringUtil.getShortRandomString());`
`30`	`30`	`}`
`31`	`31`
`32`	`32`	`public TopicClient(ConnectionStringBuilder amqpConnectionStringBuilder) throws InterruptedException, ServiceBusException {`
Original file line number	Diff line number	Diff line change
`@@ -24,4 +24,13 @@ public void onConnectionLocalOpen(Event event)`
`24`	`24`	`transport.setEmitFlowEventOnSend(false);`
`25`	`25`	`transport.bind(connection);`
`26`	`26`	`}`
	`27`	`+`
	`28`	`+ @Override`
	`29`	`+ public void onTransportClosed(Event event)`
	`30`	`+ {`
	`31`	`+ if(event.getTransport() != null)`
	`32`	`+ {`
	`33`	`+ event.getTransport().unbind();`
	`34`	`+ }`
	`35`	`+ }`
`27`	`36`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,16 +18,13 @@ public abstract class ClientEntity`
`18`	`18`	`{`
`19`	`19`	`private final String clientId;`
`20`	`20`	`private final Object syncClose;`
`21`		`- private final ClientEntity parent;`
`22`	`21`
`23`	`22`	`private boolean isClosing;`
`24`	`23`	`private boolean isClosed;`
`25`	`24`
`26`		`- protected ClientEntity(final String clientId, final ClientEntity parent)`
	`25`	`+ protected ClientEntity(final String clientId)`
`27`	`26`	`{`
`28`	`27`	`this.clientId = clientId;`
`29`		`- this.parent = parent;`
`30`		`-`
`31`	`28`	`this.syncClose = new Object();`
`32`	`29`	`}`
`33`	`30`
`@@ -39,21 +36,19 @@ public String getClientId()`
`39`	`36`	`}`
`40`	`37`
`41`	`38`	`protected boolean getIsClosed()`
`42`		`- {`
`43`		`- final boolean isParentClosed = this.parent != null && this.parent.getIsClosed();`
	`39`	`+ {`
`44`	`40`	`synchronized (this.syncClose)`
`45`	`41`	`{`
`46`		`- return isParentClosed \|\| this.isClosed;`
	`42`	`+ return this.isClosed;`
`47`	`43`	`}`
`48`	`44`	`}`
`49`		`-`
`50`		`- // returns true even if the Parent is (being) Closed`
	`45`	`+`
`51`	`46`	`protected boolean getIsClosingOrClosed()`
`52`	`47`	`{`
`53`		`- final boolean isParentClosingOrClosed = this.parent != null && this.parent.getIsClosingOrClosed();`
	`48`	`+`
`54`	`49`	`synchronized (this.syncClose)`
`55`	`50`	`{`
`56`		`- return isParentClosingOrClosed \|\| this.isClosing \|\| this.isClosed;`
	`51`	`+ return this.isClosing \|\| this.isClosed;`
`57`	`52`	`}`
`58`	`53`	`}`
`59`	`54`