Merge pull request #671 from sp193/thread-nodelay

Refactored thread stopping and starting, to not introduce 100ms delays per client

Author: Ranjan Dasgupta

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/CommsCallback.java

@@ -21,6 +21,7 @@
 import java.util.Enumeration;
 import java.util.Hashtable;
 import java.util.Vector;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
@@ -101,37 +102,33 @@ public void start(String threadName, ExecutorService executorService) {
 				completeQueue.clear();
 
 				target_state = State.RUNNING;
+				current_state = State.RUNNING;
 				if (executorService == null) {
-					new Thread(this).start();
+					callbackFuture = null;
+					callbackThread = new Thread(this);
+					callbackThread.start();
 				} else {
+					callbackThread = null;
 					callbackFuture = executorService.submit(this);
 				}
 			}
 		}
-		while (!isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
-		}			
 	}
-	
 
 	/**
 	 * Stops the callback thread. 
 	 * This call will block until stop has completed.
 	 */
 	public void stop() {
 		final String methodName = "stop";
-		
-		synchronized (lifecycle) {
-			if (callbackFuture != null) {
-				callbackFuture.cancel(true);
-			}
-		}
+
 		if (isRunning()) {
 			// @TRACE 700=stopping
 			log.fine(CLASS_NAME, methodName, "700");
 			synchronized (lifecycle) {
 				target_state = State.STOPPED;
 			}
+			// Do not allow a thread to wait for itself.
 			if (!Thread.currentThread().equals(callbackThread)) {
 				synchronized (workAvailable) {
 					// @TRACE 701=notify workAvailable and wait for run
@@ -140,9 +137,16 @@ public void stop() {
 					workAvailable.notifyAll();
 				}
 				// Wait for the thread to finish.
-				while (isRunning()) {
-					try { Thread.sleep(100); } catch (Exception e) { }
-					clientState.notifyQueueLock();
+				if (callbackFuture != null) {
+					try {
+						callbackFuture.get();
+					} catch (ExecutionException | InterruptedException e) {
+					}
+				} else {
+					try {
+						callbackThread.join();
+					} catch (InterruptedException e) {
+					}
 				}
 			}
 			// @TRACE 703=stopped
@@ -166,10 +170,6 @@ public void run() {
 		final String methodName = "run";
 		callbackThread = Thread.currentThread();
 		callbackThread.setName(threadName);
-		
-		synchronized (lifecycle) {
-			current_state = State.RUNNING;
-		}
 
 		while (isRunning()) {
 			try {
@@ -243,7 +243,6 @@ public void run() {
 		synchronized (lifecycle) {
 			current_state = State.STOPPED;
 		}
-		callbackThread = null;
 	}
 
 	private void handleActionComplete(MqttToken token)

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/CommsReceiver.java

@@ -17,6 +17,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
@@ -45,7 +46,7 @@ private enum State {STOPPED, RUNNING, STARTING, RECEIVING}
 	private final Object lifecycle = new Object();
 	private String threadName;
 	private Future<?> receiverFuture;
-	
+
 	private ClientState clientState = null;
 	private ClientComms clientComms = null;
 	private MqttInputStream in;
@@ -73,35 +74,47 @@ public void start(String threadName, ExecutorService executorService) {
 		synchronized (lifecycle) {
 			if (current_state == State.STOPPED && target_state == State.STOPPED) {
 				target_state = State.RUNNING;
+				current_state = State.RUNNING;
 				if (executorService == null) {
-					new Thread(this).start();
+					receiverFuture = null;
+					recThread = new Thread(this);
+					recThread.start();
 				} else {
+					recThread = null;
 					receiverFuture = executorService.submit(this);
 				}
 			}
 		}
-		while (!isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
-		}
 	}
 
 	/**
 	 * Stops the Receiver's thread.  This call will block.
 	 */
 	public void stop() {
 		final String methodName = "stop";
+		boolean isRunning;
+
 		synchronized (lifecycle) {
-			if (receiverFuture != null) {
-				receiverFuture.cancel(true);
-			}
 			//@TRACE 850=stopping
 			log.fine(CLASS_NAME,methodName, "850");
-			if (isRunning()) {
+			isRunning = isRunning();
+			if (isRunning) {
 				target_state = State.STOPPED;
 			}
 		}
-		while (isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
+		// This and the clause above will prevent a thread from waiting for itself.
+		if (isRunning) {
+			if (receiverFuture != null) {
+				try {
+					receiverFuture.get();
+				} catch (ExecutionException | InterruptedException e) {
+				}
+			} else {
+				try {
+					recThread.join();
+				} catch (InterruptedException e) {
+				}
+			}
 		}
 		//@TRACE 851=stopped
 		log.fine(CLASS_NAME,methodName,"851");
@@ -111,15 +124,10 @@ public void stop() {
 	 * Run loop to receive messages from the server.
 	 */
 	public void run() {
-		recThread = Thread.currentThread();
-		recThread.setName(threadName);
+		Thread.currentThread().setName(threadName);
 		final String methodName = "run";
 		MqttToken token = null;
 
-		synchronized (lifecycle) {
-			current_state = State.RUNNING;
-		}
-		
 		try {
 			State my_target;
 			synchronized (lifecycle) {
@@ -213,7 +221,6 @@ public void run() {
 			}
 		} // end try
 
-		recThread = null;
 		//@TRACE 854=<
 		log.fine(CLASS_NAME,methodName,"854");
 	}

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/CommsSender.java

@@ -17,6 +17,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
@@ -43,7 +44,7 @@ private enum State {STOPPED, RUNNING, STARTING}
 	private Thread 	sendThread		= null;
 	private String threadName;
 	private Future<?> senderFuture;
-	
+
 	private ClientState clientState = null;
 	private MqttOutputStream out;
 	private ClientComms clientComms = null;
@@ -68,56 +69,62 @@ public void start(String threadName, ExecutorService executorService) {
 		synchronized (lifecycle) {
 			if (current_state == State.STOPPED && target_state == State.STOPPED) {
 				target_state = State.RUNNING;
+				current_state = State.RUNNING;
 				if (executorService == null) {
-					new Thread(this).start();
+					senderFuture = null;
+					sendThread = new Thread(this);
+					sendThread.start();
 				} else {
+					sendThread = null;
 					senderFuture = executorService.submit(this);
 				}
 			}
 		}
-		while (!isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
-		}
 	}
 
 	/**
 	 * Stops the Sender's thread.  This call will block.
 	 */
 	public void stop() {
 		final String methodName = "stop";
-		
+		boolean isRunning;
+
 		if (!isRunning()) {
 			return;
 		}
-			
+
 		synchronized (lifecycle) {
-			if (senderFuture != null) {
-				senderFuture.cancel(true);
-			}
 			//@TRACE 800=stopping sender
 			log.fine(CLASS_NAME,methodName,"800");
-			if (isRunning()) {
+			isRunning = isRunning();
+			if (isRunning) {
 				target_state = State.STOPPED;
 				clientState.notifyQueueLock();
 			}
 		}
-		while (isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
-			clientState.notifyQueueLock();
+		// This and the clause above will prevent a thread from waiting for itself.
+		if (isRunning) {
+			if (senderFuture != null) {
+				try {
+					senderFuture.get();
+				} catch (ExecutionException | InterruptedException e) {
+				}
+			} else {
+				try {
+					sendThread.join();
+				} catch (InterruptedException e) {
+					e.printStackTrace();
+				}
+			}
 		}
 		//@TRACE 801=stopped
 		log.fine(CLASS_NAME,methodName,"801");
 	}
 
 	public void run() {
-		sendThread = Thread.currentThread();
-		sendThread.setName(threadName);
+		Thread.currentThread().setName(threadName);
 		final String methodName = "run";
 		MqttWireMessage message = null;
-		
-		synchronized (lifecycle) {
-			current_state = State.RUNNING;
-		}
 
 		try {
 			State my_target;
@@ -177,7 +184,6 @@ public void run() {
 		} finally {
 			synchronized (lifecycle) {
 				current_state = State.STOPPED;
-				sendThread = null;
 			}
 		}