Fixing high CPU and memory leak in session pump (#382)

* Fixing high CPU and memory leak in session pump caused by timed out accept session requests.

* Fixing a minor race condition when closing sender or receiver on link creation timeout.

Author: yvgopal

azure-servicebus/azure-servicebus.pom

@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.microsoft.azure</groupId>
     <artifactId>azure-servicebus</artifactId>
-    <version>1.2.15</version>
+    <version>1.2.16</version>
     <licenses>
         <license>
             <name>The MIT License (MIT)</name>

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/ReceiveLinkHandler.java

@@ -60,7 +60,7 @@ public void onLinkRemoteOpen(Event event)
 			}
 			else
 			{				
-				TRACE_LOGGER.debug("onLinkRemoteOpen: linkName:{}, remoteTarget:{}, remoteTarget:{}, action:{}", receiver.getName(), null, null, "waitingForError");
+				TRACE_LOGGER.debug("onLinkRemoteOpen: linkName:{}, remoteSource:{}, remoteTarget:{}, action:{}", receiver.getName(), null, null, "waitingForError");
 			}
 		}
 	}

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/ClientEntity.java

@@ -57,9 +57,20 @@ protected final void setClosed()
 	{
 		synchronized (this.syncClose)
 		{
+			this.isClosing = false;
 			this.isClosed = true;
 		}
 	}
+	
+	protected final void setClosing()
+	{
+		synchronized (this.syncClose)
+		{
+			if (!this.isClosed) {
+				this.isClosing = true;
+			}
+		}
+	}
 
 	public final CompletableFuture<Void> closeAsync()
 	{

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java

@@ -105,10 +105,10 @@ public class CoreMessageReceiver extends ClientEntity implements IAmqpReceiver,
 	private CompletableFuture<Void> ensureLinkReopenFutureToWaitOn;
 	private final Runnable timedOutUpdateStateRequestsDaemon;
 	private final Runnable returnMesagesLoopDaemon;
-	private final ScheduledFuture<?> updateStateRequestsTimeoutChecker;
-	private final ScheduledFuture<?> returnMessagesLoopRunner;
 	private final MessagingEntityType entityType;
 	private boolean shouldRetryLinkReopenOnTransientFailure = true;
+	private ScheduledFuture<?> updateStateRequestsTimeoutChecker;
+	private ScheduledFuture<?> returnMessagesLoopRunner;
 
 	// TODO Change onReceiveComplete to handle empty deliveries. Change onError to retry updateState requests.
 	private CoreMessageReceiver(final MessagingFactory factory,
@@ -151,6 +151,12 @@ private CoreMessageReceiver(final MessagingFactory factory,
 			public void run() {
 			    try
 			    {
+			    	if (CoreMessageReceiver.this.getIsClosed())
+			    	{
+			    		CoreMessageReceiver.this.updateStateRequestsTimeoutChecker.cancel(true);
+			    		return;
+			    	}
+			    	
 			        TRACE_LOGGER.trace("Starting '{}' core message receiver's internal loop to complete timed out update state requests.", CoreMessageReceiver.this.receivePath);
 			        for(Map.Entry<String, UpdateStateWorkItem> entry : CoreMessageReceiver.this.pendingUpdateStateRequests.entrySet())
 	                {
@@ -182,6 +188,12 @@ public void run() {
             public void run() {
                 try
                 {
+                	if (CoreMessageReceiver.this.getIsClosed())
+			    	{
+			    		CoreMessageReceiver.this.returnMessagesLoopRunner.cancel(true);
+			    		return;
+			    	}
+                	
                     TRACE_LOGGER.trace("Starting '{}' core message receiver's internal loop to return messages to waiting clients.", CoreMessageReceiver.this.receivePath);
                     while(!CoreMessageReceiver.this.prefetchedMessages.isEmpty())
                     {
@@ -210,11 +222,6 @@ public void run() {
                 }
             }
         };
-		
-		// As all update state requests have the same timeout, one timer is better than having one timer per request
-		this.updateStateRequestsTimeoutChecker = Timer.schedule(timedOutUpdateStateRequestsDaemon, CoreMessageReceiver.UPDATE_STATE_REQUESTS_DAEMON_WAKE_UP_INTERVAL, TimerType.RepeatRun);
-		// Scheduling it as a separate thread that wakes up at regular very short intervals.. Doesn't wait on incoming receive requests from callers or incoming deliveries from reactor 
-		this.returnMessagesLoopRunner = Timer.schedule(returnMesagesLoopDaemon, CoreMessageReceiver.RETURN_MESSAGES_DAEMON_WAKE_UP_INTERVAL, TimerType.RepeatRun);
 	}
 
 	// Connection has to be associated with Reactor before Creating a receiver on it.
@@ -636,6 +643,11 @@ public void onOpenComplete(Exception exception)
 			if (this.linkOpen != null && !this.linkOpen.getWork().isDone())
 			{
 				AsyncUtil.completeFuture(this.linkOpen.getWork(), this);
+				
+				// As all update state requests have the same timeout, one timer is better than having one timer per request
+				this.updateStateRequestsTimeoutChecker = Timer.schedule(timedOutUpdateStateRequestsDaemon, CoreMessageReceiver.UPDATE_STATE_REQUESTS_DAEMON_WAKE_UP_INTERVAL, TimerType.RepeatRun);
+				// Scheduling it as a separate thread that wakes up at regular very short intervals.. Doesn't wait on incoming receive requests from callers or incoming deliveries from reactor 
+				this.returnMessagesLoopRunner = Timer.schedule(returnMesagesLoopDaemon, CoreMessageReceiver.RETURN_MESSAGES_DAEMON_WAKE_UP_INTERVAL, TimerType.RepeatRun);
 			}
 
 			if(this.receiveLinkReopenFuture != null && !this.receiveLinkReopenFuture.isDone())
@@ -924,14 +936,15 @@ public void run()
 					{
 						if (!linkOpen.getWork().isDone())
 						{
-						    CoreMessageReceiver.this.closeInternals(false);
-						    CoreMessageReceiver.this.setClosed();
-						    
 							Exception operationTimedout = new TimeoutException(
 									String.format(Locale.US, "%s operation on ReceiveLink(%s) to path(%s) timed out at %s.", "Open", CoreMessageReceiver.this.receiveLink.getName(), CoreMessageReceiver.this.receivePath, ZonedDateTime.now()),
 									CoreMessageReceiver.this.lastKnownLinkError);
 							TRACE_LOGGER.warn(operationTimedout.getMessage());
 							ExceptionUtil.completeExceptionally(linkOpen.getWork(), operationTimedout, CoreMessageReceiver.this, true);
+							
+							CoreMessageReceiver.this.setClosing();
+						    CoreMessageReceiver.this.closeInternals(false);
+						    CoreMessageReceiver.this.setClosed();
 						}
 					}
 				}

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java

@@ -675,14 +675,15 @@ public void run()
 					{
 						if (!CoreMessageSender.this.linkFirstOpen.isDone())
 						{
-						    CoreMessageSender.this.closeInternals(false);
-						    CoreMessageSender.this.setClosed();
-						    
 							Exception operationTimedout = new TimeoutException(
 									String.format(Locale.US, "Open operation on SendLink(%s) on Entity(%s) timed out at %s.",	CoreMessageSender.this.sendLink.getName(), CoreMessageSender.this.getSendPath(), ZonedDateTime.now().toString()),
 									CoreMessageSender.this.lastKnownErrorReportedAt.isAfter(Instant.now().minusSeconds(ClientConstants.SERVER_BUSY_BASE_SLEEP_TIME_IN_SECS)) ? CoreMessageSender.this.lastKnownLinkError : null);
 							TRACE_LOGGER.warn(operationTimedout.getMessage());
 							ExceptionUtil.completeExceptionally(CoreMessageSender.this.linkFirstOpen, operationTimedout, CoreMessageSender.this, true);
+							
+							CoreMessageSender.this.setClosing();
+						    CoreMessageSender.this.closeInternals(false);
+						    CoreMessageSender.this.setClosed();
 						}
 					}
 				}

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/Util.java

@@ -385,7 +385,7 @@ static int encodeMessageToCustomArray(Message message, byte[] encodedBytes, int
 	// Pass little less than client timeout to the server so client doesn't time out before server times out
 	static Duration adjustServerTimeout(Duration clientTimeout)
 	{
-		return clientTimeout.minusMillis(100);
+		return clientTimeout.minusMillis(200);
 	}
 
 	// This is not super stable for some reason

pom.xml

@@ -13,7 +13,7 @@
 		<proton-j-version>0.31.0</proton-j-version>
 	  	<junit-version>4.12</junit-version>
 		<slf4j-version>1.7.0</slf4j-version>
-		<client-current-version>1.2.15</client-current-version>		
+		<client-current-version>1.2.16</client-current-version>		
 	</properties>
 
 	<modules>