diff --git a/bin/systemds b/bin/systemds index b9dcfa6fbcb..9b03369ae3a 100755 --- a/bin/systemds +++ b/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) diff --git a/docs/site/run.md b/docs/site/run.md index 29ed1818769..4fc9fd0b6d1 100644 --- a/docs/site/run.md +++ b/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 diff --git a/src/main/java/org/apache/sysds/api/DMLOptions.java b/src/main/java/org/apache/sysds/api/DMLOptions.java index 10c41e3d0a8..fc73422ae2d 100644 --- a/src/main/java/org/apache/sysds/api/DMLOptions.java +++ b/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(); diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java index 682cc8e3fff..895b1007213 100644 --- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java +++ b/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."); diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/monitoring/FederatedMonitoringServer.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/monitoring/FederatedMonitoringServer.java index d1a482a689a..090a7ed73f8 100644 --- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/monitoring/FederatedMonitoringServer.java +++ b/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(); diff --git a/src/main/java/org/apache/sysds/utils/PortUtils.java b/src/main/java/org/apache/sysds/utils/PortUtils.java new file mode 100644 index 00000000000..000c5b72f9e --- /dev/null +++ b/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; + } +} diff --git a/src/test/java/org/apache/sysds/test/component/federated/FederatedWorkerPortTest.java b/src/test/java/org/apache/sysds/test/component/federated/FederatedWorkerPortTest.java new file mode 100644 index 00000000000..ae8b477bad3 --- /dev/null +++ b/src/test/java/org/apache/sysds/test/component/federated/FederatedWorkerPortTest.java @@ -0,0 +1,168 @@ +/* + * 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.test.component.federated; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.net.ServerSocket; + +import org.apache.commons.cli.ParseException; +import org.apache.sysds.api.DMLOptions; +import org.apache.sysds.conf.DMLConfig; +import org.apache.sysds.runtime.DMLRuntimeException; +import org.apache.sysds.runtime.controlprogram.federated.FederatedWorker; +import org.apache.sysds.utils.PortUtils; +import org.junit.Test; + +/** + * Tests that a federated worker started with a port it cannot bind to reports the reason instead of terminating + * silently, covering out of range, reserved and already occupied ports. + */ +public class FederatedWorkerPortTest { + + @Test + public void validPortRange() { + assertFalse(PortUtils.isValidPort(-1)); + assertFalse(PortUtils.isValidPort(0)); + assertTrue(PortUtils.isValidPort(1)); + assertTrue(PortUtils.isValidPort(DMLConfig.DEFAULT_FEDERATED_PORT)); + assertTrue(PortUtils.isValidPort(65535)); + assertFalse(PortUtils.isValidPort(65536)); + assertFalse(PortUtils.isValidPort(80505)); + } + + @Test + public void privilegedPortRange() { + assertTrue(PortUtils.isPrivilegedPort(80)); + assertTrue(PortUtils.isPrivilegedPort(1023)); + assertFalse(PortUtils.isPrivilegedPort(1024)); + assertFalse(PortUtils.isPrivilegedPort(DMLConfig.DEFAULT_FEDERATED_PORT)); + } + + @Test + public void parseValidPort() { + assertEquals(4040, PortUtils.parsePort("4040", "-w")); + assertEquals(4040, PortUtils.parsePort(" 4040 ", "-w")); + } + + @Test + public void parseOutOfRangePort() { + try { + PortUtils.parsePort("80505", "-w"); + fail("expected an exception for a port outside the valid range"); + } + catch(IllegalArgumentException e) { + assertTrue(e.getMessage(), e.getMessage().contains("out of range")); + } + } + + @Test + public void parseNonNumericPort() { + try { + PortUtils.parsePort("notAPort", "-w"); + fail("expected an exception for a non numeric port"); + } + catch(IllegalArgumentException e) { + assertTrue(e.getMessage(), e.getMessage().contains("not an integer")); + } + } + + @Test + public void cliRejectsOutOfRangePort() { + assertParseFails(new String[] {"-w", "80505"}, "out of range"); + } + + @Test + public void cliRejectsNegativePort() { + // -1 is the internal marker for 'use the default port' and must not be accepted from outside + assertParseFails(new String[] {"-w", "-1"}, "out of range"); + } + + @Test + public void cliRejectsPortZero() { + // port 0 binds an arbitrary free port, which no coordinator could address + assertParseFails(new String[] {"-w", "0"}, "out of range"); + } + + @Test + public void cliRejectsNonNumericPort() { + assertParseFails(new String[] {"-w", "notAPort"}, "not an integer"); + } + + @Test + public void cliRejectsOutOfRangeMonitoringPort() { + assertParseFails(new String[] {"-fedMonitoring", "80505"}, "out of range"); + } + + @Test + public void cliAcceptsValidPort() throws ParseException { + DMLOptions opts = DMLOptions.parseCLArguments(new String[] {"-w", "8001"}); + assertTrue(opts.fedWorker); + assertEquals(8001, opts.fedWorkerPort); + } + + @Test + public void cliDefaultsWithoutPort() throws ParseException { + // the port argument is optional, a missing one falls back to the default federated port + DMLOptions opts = DMLOptions.parseCLArguments(new String[] {"-w"}); + assertTrue(opts.fedWorker); + assertEquals(-1, opts.fedWorkerPort); + } + + @Test + public void workerRejectsOutOfRangePort() { + try { + new FederatedWorker(80505, false); + fail("expected the federated worker to reject a port outside the valid range"); + } + catch(DMLRuntimeException e) { + assertTrue(e.getMessage(), e.getMessage().contains("out of range")); + } + } + + @Test + public void workerReportsOccupiedPort() throws IOException { + try(ServerSocket occupied = new ServerSocket(0)) { + final int port = occupied.getLocalPort(); + try { + new FederatedWorker(port, false); + fail("expected the federated worker to reject the already occupied port " + port); + } + catch(DMLRuntimeException e) { + assertTrue(e.getMessage(), e.getMessage().contains("already in use")); + assertTrue(e.getMessage(), e.getMessage().contains(String.valueOf(port))); + } + } + } + + private static void assertParseFails(String[] args, String expectedMessagePart) { + try { + DMLOptions.parseCLArguments(args); + fail("expected a parse exception for arguments " + String.join(" ", args)); + } + catch(ParseException e) { + assertTrue(e.getMessage(), e.getMessage().contains(expectedMessagePart)); + } + } +}