[SYSTEMDS-3959] Report an error when a federated worker cannot bind its port

bin/systemds

@@ -173,6 +173,32 @@ script.
 EOF
 }
 
+# verify that $1 is a port that a server can be bound to, otherwise abort with an error.
+# $2 names the server the port is meant for.
+function checkPort {
+  local port=$1
+  local target=$2
+  local re='^[0-9]+$'
+  if ! [[ $port =~ $re ]] ; then
+    echo "error: Port '$port' for the $target is not a number"
+    printUsage
+    exit 1
+  fi
+  # drop leading zeros, so that the length check below is not fooled by e.g. 0000080505
+  local num=$port
+  while [[ ${#num} -gt 1 && $num == 0* ]] ; do num=${num#0} ; done
+  # more than 5 digits is out of range by definition, and comparing it would silently
+  # overflow the 64 bit integers of the shell for very long inputs
+  if [ ${#num} -gt 5 ] || [ "$num" -lt 1 ] || [ "$num" -gt 65535 ] ; then
+    echo "error: Port $port for the $target is out of range, expected a port in [1, 65535]"
+    printUsage
+    exit 1
+  fi
+  if [ "$num" -lt 1024 ] ; then
+    echo "warning: Port $port for the $target is a reserved system port, binding it requires elevated privileges"
+  fi
+}
+
 # print an error if no argument is supplied.
 if [ -z "$1" ] ; then
     echo "Wrong Usage. Add -help for additional parameters.";
@@ -242,11 +268,7 @@ elif echo "$1" | grep -q "WORKER"; then
     shift
   fi
   PORT=$1
-  re='^[0-9]+$'
-  if ! [[ $PORT =~ $re ]] ; then
-    echo "error: Port is not a number"
-    printUsage
-  fi
+  checkPort "$PORT" "federated worker"
   shift
 elif echo "$1" | grep -q "FEDMONITORING"; then
   FEDMONITORING=1
@@ -256,11 +278,7 @@ elif echo "$1" | grep -q "FEDMONITORING"; then
     shift
   fi
   PORT=$1
-  re='^[0-9]+$'
-  if ! [[ $PORT =~ $re ]] ; then
-    echo "error: Port is not a number"
-    printUsage
-  fi
+  checkPort "$PORT" "federated monitoring backend"
   shift
 else
   # handle optional '-f' before DML file (for consistency)

docs/site/run.md

@@ -247,7 +247,9 @@ Run in a separate terminal:
 systemds WORKER 8001
 ```
 
-This starts a worker on port `8001`.
+This starts a worker on port `8001`. The port has to be in `[1, 65535]` and free, and ports below `1024`
+are reserved system ports and require elevated privileges. Otherwise the worker reports the reason it
+could not start, e.g. `Federated worker stopped: port 8001 is already in use`.
 
 ### 4.2 Next Steps and Full Examples
 

src/main/java/org/apache/sysds/api/DMLOptions.java

@@ -41,6 +41,7 @@
 import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
 import org.apache.sysds.utils.Explain;
 import org.apache.sysds.utils.Explain.ExplainType;
+import org.apache.sysds.utils.PortUtils;
 
 /**
  * Set of DMLOptions that can be set through the command line
@@ -308,14 +309,17 @@ else if (lineageType.equalsIgnoreCase("debugger"))
 
 		if (line.hasOption("w")){
 			dmlOptions.fedWorker = true;
-			dmlOptions.fedWorkerPort = Integer.parseInt(line.getOptionValue("w"));
+			String port = line.getOptionValue("w");
+			// the argument is optional, a missing port falls back to the default federated port
+			if(port != null)
+				dmlOptions.fedWorkerPort = parsePort(port, "-w");
 		}
 
 		if (line.hasOption("fedMonitoring")) {
 			dmlOptions.fedMonitoring= true;
 			String port = line.getOptionValue("fedMonitoring");
 			if(port != null)
-				dmlOptions.fedMonitoringPort = Integer.parseInt(port);
+				dmlOptions.fedMonitoringPort = parsePort(port, "-fedMonitoring");
 			else
 				throw new org.apache.commons.cli.ParseException("No port [integer] specified for -fedMonitoring option");
 		}
@@ -401,7 +405,24 @@ else if (lineageType.equalsIgnoreCase("debugger"))
 
 		return dmlOptions;
 	}
-
+
+	/**
+	 * Parse the port of a command line option, rejecting values that no server could ever be bound to.
+	 *
+	 * @param value  the raw argument value
+	 * @param option the name of the option the value belongs to, used for the error message
+	 * @return the parsed port
+	 * @throws org.apache.commons.cli.ParseException if the value is not a port in the valid range
+	 */
+	private static int parsePort(String value, String option) throws org.apache.commons.cli.ParseException {
+		try {
+			return PortUtils.parsePort(value, option);
+		}
+		catch(IllegalArgumentException e) {
+			throw new org.apache.commons.cli.ParseException(e.getMessage());
+		}
+	}
+
 	@SuppressWarnings("static-access")
 	private static Options createCLIOptions() {
 		Options options = new Options();

src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java

@@ -30,6 +30,7 @@
 import org.apache.sysds.api.DMLScript;
 import org.apache.sysds.conf.ConfigurationManager;
 import org.apache.sysds.conf.DMLConfig;
+import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.controlprogram.caching.CacheBlock;
 import org.apache.sysds.runtime.controlprogram.federated.compression.CompressionDecoderEndStatisticsHandler;
 import org.apache.sysds.runtime.controlprogram.federated.compression.CompressionDecoderStartStatisticsHandler;
@@ -40,6 +41,7 @@
 import org.apache.sysds.runtime.lineage.LineageCacheConfig;
 import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
 import org.apache.sysds.runtime.lineage.LineageItem;
+import org.apache.sysds.utils.PortUtils;
 import org.apache.sysds.utils.stats.InfrastructureAnalyzer;
 import org.apache.sysds.utils.stats.Timing;
 
@@ -83,6 +85,14 @@ public FederatedWorker(int port, boolean debug) {
 		_port = (port == -1) ? DMLConfig.DEFAULT_FEDERATED_PORT : port;
 		_debug = debug;
 
+		// fail before any setup work if the port cannot be bound to in the first place
+		if(!PortUtils.isValidPort(_port))
+			throw new DMLRuntimeException("Failed to start federated worker: port " + _port
+				+ " is out of range, expected a port in [" + PortUtils.MIN_PORT + ", " + PortUtils.MAX_PORT + "]");
+		if(PortUtils.isPrivilegedPort(_port))
+			LOG.warn("Federated worker port " + _port + " is a reserved system port, binding it requires "
+				+ "elevated privileges and may conflict with other services.");
+
 		LineageCacheConfig.setConfig(DMLScript.LINEAGE_REUSE);
 		LineageCacheConfig.setCachePolicy(DMLScript.LINEAGE_POLICY);
 		LineageCacheConfig.setEstimator(DMLScript.LINEAGE_ESTIMATE);
@@ -121,10 +131,13 @@ private void run() {
 				e.printStackTrace();
 		}
 		catch(Exception e) {
-			// report why the worker stops, e.g., a missing certificate with ssl enabled, otherwise it exits silently
-			LOG.error("Federated worker stopped: " + e.getMessage());
+			// report why the worker stops, e.g., an occupied or reserved port, or a missing
+			// certificate with ssl enabled, otherwise it exits silently and signals success
+			final String msg = "Federated worker stopped: " + PortUtils.explainBindFailure(_port, e);
+			LOG.error(msg);
 			if(_debug)
 				e.printStackTrace();
+			throw new DMLRuntimeException(msg, e);
 		}
 		finally {
 			LOG.info("Federated Worker Shutting down.");

src/main/java/org/apache/sysds/runtime/controlprogram/federated/monitoring/FederatedMonitoringServer.java

@@ -20,6 +20,8 @@
 package org.apache.sysds.runtime.controlprogram.federated.monitoring;
 
 import org.apache.log4j.Logger;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.utils.PortUtils;
 
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
@@ -38,15 +40,24 @@
 
 public class FederatedMonitoringServer {
 	protected static Logger log = Logger.getLogger(FederatedMonitoringServer.class);
+	public static final int DEFAULT_MONITORING_PORT = 4201;
 	private final int _port;
 
 	private final boolean _debug;
 
 	public FederatedMonitoringServer(int port, boolean debug) {
-		_port = (port == -1) ? 4201 : port;
+		_port = (port == -1) ? DEFAULT_MONITORING_PORT : port;
 
 		_debug = debug;
 
+		// fail before any setup work if the port cannot be bound to in the first place
+		if(!PortUtils.isValidPort(_port))
+			throw new DMLRuntimeException("Failed to start federated monitoring backend: port " + _port
+				+ " is out of range, expected a port in [" + PortUtils.MIN_PORT + ", " + PortUtils.MAX_PORT + "]");
+		if(PortUtils.isPrivilegedPort(_port))
+			log.warn("Federated monitoring backend port " + _port + " is a reserved system port, binding it "
+				+ "requires elevated privileges and may conflict with other services.");
+
 		run();
 	}
 
@@ -88,12 +99,20 @@ protected void initChannel(Channel ch) {
 			ChannelFuture f = server.bind(_port).sync();
 			log.info("Started Federated Monitoring Backend at port: " + _port);
 			f.channel().closeFuture().sync();
-		} catch(Exception e) {
+		}
+		catch(InterruptedException e) {
 			log.info("Federated Monitoring Backend Interrupted");
-			if (_debug) {
-				log.error(e.getMessage());
+			if(_debug)
+				e.printStackTrace();
+		}
+		catch(Exception e) {
+			// report why the backend stops, e.g., an occupied or reserved port, otherwise it exits
+			// silently and signals success
+			final String msg = "Federated Monitoring Backend stopped: " + PortUtils.explainBindFailure(_port, e);
+			log.error(msg);
+			if(_debug)
 				e.printStackTrace();
-			}
+			throw new DMLRuntimeException(msg, e);
 		} finally{
 			log.info("Federated Monitoring Backend Shutting down.");
 			workerGroup.shutdownGracefully();

src/main/java/org/apache/sysds/utils/PortUtils.java

@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.utils;
+
+import java.io.IOException;
+import java.net.BindException;
+import java.net.ServerSocket;
+
+/**
+ * Helpers to validate TCP port numbers of the servers started from the command line, i.e., the federated worker (-w)
+ * and the federated monitoring backend (-fedMonitoring).
+ */
+public class PortUtils {
+	/**
+	 * The lowest port a server is allowed to bind to. Port 0 is excluded as it binds an arbitrary free port, which
+	 * peers of a federated worker cannot address.
+	 */
+	public static final int MIN_PORT = 1;
+
+	/** Highest port representable in the 16-bit TCP port field. */
+	public static final int MAX_PORT = 65535;
+
+	/** Ports up to and including this one are reserved system ports on unix-like systems. */
+	public static final int MAX_PRIVILEGED_PORT = 1023;
+
+	private PortUtils() {
+		// prevent instantiation of this utility class
+	}
+
+	/**
+	 * Check whether the given port is inside the range of ports a server can be bound to.
+	 *
+	 * @param port the port to check
+	 * @return true if the port is in [{@link #MIN_PORT}, {@link #MAX_PORT}]
+	 */
+	public static boolean isValidPort(int port) {
+		return port >= MIN_PORT && port <= MAX_PORT;
+	}
+
+	/**
+	 * Check whether the given port is a reserved system port, i.e., a port that typically requires elevated privileges
+	 * to bind to.
+	 *
+	 * @param port the port to check
+	 * @return true if the port is a privileged port
+	 */
+	public static boolean isPrivilegedPort(int port) {
+		return port >= MIN_PORT && port <= MAX_PRIVILEGED_PORT;
+	}
+
+	/**
+	 * Parse a port given as a command line argument.
+	 *
+	 * @param value  the raw argument value
+	 * @param option the name of the option the value belongs to, used for the error message
+	 * @return the parsed port
+	 * @throws IllegalArgumentException if the value is not an integer inside the valid port range
+	 */
+	public static int parsePort(String value, String option) {
+		final int port;
+		try {
+			port = Integer.parseInt(value.trim());
+		}
+		catch(NumberFormatException e) {
+			throw new IllegalArgumentException("Invalid port '" + value + "' for option " + option
+				+ ": not an integer, expected a port in [" + MIN_PORT + ", " + MAX_PORT + "]");
+		}
+		checkValidPort(port, option);
+		return port;
+	}
+
+	/**
+	 * Verify that the given port can be bound to at all, i.e., that it is inside the valid range.
+	 *
+	 * @param port   the port to check
+	 * @param option the name of the option the port originates from, used for the error message
+	 * @throws IllegalArgumentException if the port is outside the valid port range
+	 */
+	public static void checkValidPort(int port, String option) {
+		if(!isValidPort(port))
+			throw new IllegalArgumentException("Invalid port " + port + " for option " + option
+				+ ": out of range, expected a port in [" + MIN_PORT + ", " + MAX_PORT + "]");
+	}
+
+	/**
+	 * Check if a port is currently free on this machine. Only intended for error reporting, since the port can be taken
+	 * again between this check and the actual bind.
+	 *
+	 * @param port the port to check
+	 * @return true if a server socket could be opened on the port
+	 */
+	public static boolean isPortAvailable(int port) {
+		if(!isValidPort(port))
+			return false;
+		try(ServerSocket s = new ServerSocket(port)) {
+			return true;
+		}
+		catch(IOException e) {
+			return false;
+		}
+	}
+
+	/**
+	 * Translate a failure of a server bind into a message that names the likely cause, since the exceptions of the
+	 * underlying socket implementation are platform-specific and terse (e.g., 'Address already in use' without naming
+	 * the port).
+	 *
+	 * @param port the port the server tried to bind to
+	 * @param e    the exception the bind failed with
+	 * @return a human-readable explanation of the failure
+	 */
+	public static String explainBindFailure(int port, Throwable e) {
+		final String msg = (e.getMessage() != null) ? e.getMessage() : e.getClass().getSimpleName();
+		if(!isValidPort(port))
+			return "port " + port + " is out of range, expected a port in [" + MIN_PORT + ", " + MAX_PORT + "] (" + msg
+				+ ")";
+		if(e instanceof BindException) {
+			if(msg.toLowerCase().contains("permission denied"))
+				return "no permission to bind port " + port + (isPrivilegedPort(port) ? ", ports below "
+					+ (MAX_PRIVILEGED_PORT + 1) + " are reserved and require elevated privileges" : "") + " (" + msg
+					+ ")";
+			return "port " + port + " is already in use (" + msg + ")";
+		}
+		return msg;
+	}
+}