Try to detect issues with reusing connections in processes

Author: rowleya

SpiNNaker-comms/src/main/java/uk/ac/manchester/spinnaker/connections/UDPConnection.java

@@ -42,11 +42,13 @@
 import java.net.SocketTimeoutException;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
+import java.util.ConcurrentModificationException;
 
 import org.slf4j.Logger;
 
 import com.google.errorprone.annotations.ForOverride;
 import com.google.errorprone.annotations.OverridingMethodsMustInvokeSuper;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
 
 import uk.ac.manchester.spinnaker.connections.model.Connection;
 import uk.ac.manchester.spinnaker.messages.sdp.SDPHeader;
@@ -102,6 +104,9 @@ public enum TrafficClass {
 
 	private int receivePacketSize = PACKET_MAX_SIZE;
 
+	@GuardedBy("this")
+	private boolean inUse = false;
+
 	/**
 	 * Main constructor, any argument of which could {@code null}.
 	 * <p>
@@ -683,7 +688,8 @@ public boolean isClosed() {
 	 * port in a NAT and/or firewall to allow incoming packets to be received.
 	 *
 	 * @param host
-	 *            The address of the SpiNNaker board to which the message should
+	 *            The address of the SpiNNaker board to which the
+	 *            message should
 	 *            be sent
 	 * @throws IOException
 	 *             If anything goes wrong
@@ -707,4 +713,16 @@ public String toString() {
 				getClass().getSimpleName().replaceAll("^.*\\.", ""),
 				localAddr(), isClosed() ? "|" : "", remoteAddr());
 	}
+
+	public synchronized void setInUse() {
+		if (inUse) {
+			throw new ConcurrentModificationException(
+					"Connection " + this + " is already in use!");
+		}
+		this.inUse = true;
+	}
+
+	public synchronized void setNotInUse() {
+		this.inUse = false;
+	}
 }

SpiNNaker-comms/src/main/java/uk/ac/manchester/spinnaker/transceiver/TxrxProcess.java

@@ -303,8 +303,13 @@ protected <Conn extends SCPConnection> TxrxProcess(
 	 * @return The pipeline instance.
 	 */
 	private RequestPipeline pipeline(SCPRequest<?> request) {
-		return requestPipelines.computeIfAbsent(
-				selector.getNextConnection(request), RequestPipeline::new);
+		var connection = selector.getNextConnection(request);
+		if (!requestPipelines.containsKey(connection)) {
+			var pipeline = new RequestPipeline(connection);
+			requestPipelines.put(connection, pipeline);
+			connection.setInUse();
+		}
+		return requestPipelines.get(connection);
 	}
 
 	/**
@@ -331,6 +336,7 @@ protected final void finishBatch()
 		for (var pipe : requestPipelines.values()) {
 			pipe.finish();
 		}
+		requestPipelines.clear();
 		if (failure != null) {
 			var hdr = failure.req.sdpHeader;
 			throw makeInstance(hdr.getDestination(), failure.exn);
@@ -799,6 +805,7 @@ void finish() throws IOException, InterruptedException {
 			}
 			log.debug("Finished called on {} with connection {}", this,
 					connection);
+			this.connection.setNotInUse();
 		}
 
 		/**