Enabling configuring IP Address for Dotnet Backend

src/csharp/Microsoft.Spark/Network/DefaultSocketWrapper.cs

@@ -6,6 +6,7 @@
 using System.IO;
 using System.Net;
 using System.Net.Sockets;
+using Microsoft.Spark.Interop;
 using Microsoft.Spark.Services;
 using Microsoft.Spark.Utils;
 
@@ -24,12 +25,20 @@ internal sealed class DefaultSocketWrapper : ISocketWrapper
         /// Default constructor that creates a new instance of DefaultSocket class which represents
         /// a traditional socket (System.Net.Socket.Socket).
         /// 
-        /// This socket is bound to Loopback with port 0.
+        /// This socket is bound to provided IP address with port 0.
         /// </summary>
         public DefaultSocketWrapper() :
             this(new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
         {
-            _innerSocket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+            string dotnetBackendIPAddress = SparkEnvironment.ConfigurationService.GetBackendIPAddress();
+            if (dotnetBackendIPAddress == "localhost" || dotnetBackendIPAddress == null)
+            {
+                _innerSocket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+            }
+            else
+            {
+                _innerSocket.Bind(new IPEndPoint(IPAddress.Parse(dotnetBackendIPAddress), 0));
+            }
         }
 
         /// <summary>

src/csharp/Microsoft.Spark/Services/ConfigurationService.cs

@@ -20,6 +20,7 @@ internal sealed class ConfigurationService : IConfigurationService
             "spark.dotnet.worker.writeBufferSize";
 
         private const string DotnetBackendPortEnvVarName = "DOTNETBACKEND_PORT";
+        private const string DotnetBackendIPAddressEnvVarName = "DOTNETBACKEND_IP_ADDRESS";
         private const int DotnetBackendDebugPort = 5567;
 
         private static readonly string s_procBaseFileName = "Microsoft.Spark.Worker";
@@ -51,6 +52,17 @@ public int GetBackendPortNumber()
             return portNumber;
         }
 
+        /// <summary>
+        /// Returns the IP address for socket communication between JVM and CLR.
+        /// </summary>
+        public string GetBackendIPAddress()
+        {
+            string ipAddress = Environment.GetEnvironmentVariable(DotnetBackendIPAddressEnvVarName);
+            _logger.LogInfo($"Using IP address {ipAddress} for connection.");
+
+            return ipAddress;
+        }
+
         /// <summary>
         /// Returns the worker executable path.
         /// </summary>

src/csharp/Microsoft.Spark/Services/IConfigurationService.cs

@@ -14,6 +14,11 @@ internal interface IConfigurationService
         /// </summary>
         int GetBackendPortNumber();
 
+        /// <summary>
+        /// The IP address used for communicating with the .NET backend process.
+        /// </summary>
+        string GetBackendIPAddress();
+
         /// <summary>
         /// The full path to the .NET worker executable.
         /// </summary>

src/scala/microsoft-spark-2.3.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala

@@ -30,7 +30,7 @@ class DotnetBackend extends Logging {
   private[this] var bootstrap: ServerBootstrap = _
   private[this] var bossGroup: EventLoopGroup = _
 
-  def init(portNumber: Int): Int = {
+  def init(portNumber: Int, ipAddress: String): Int = {
     // need at least 3 threads, use 10 here for safety
     bossGroup = new NioEventLoopGroup(10)
     val workerGroup = bossGroup
@@ -57,7 +57,7 @@ class DotnetBackend extends Logging {
       }
     })
 
-    channelFuture = bootstrap.bind(new InetSocketAddress("localhost", portNumber))
+    channelFuture = bootstrap.bind(new InetSocketAddress(ipAddress, portNumber))
     channelFuture.syncUninterruptibly()
     channelFuture.channel().localAddress().asInstanceOf[InetSocketAddress].getPort
   }

src/scala/microsoft-spark-2.3.x/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala

@@ -8,6 +8,8 @@ package org.apache.spark.deploy.dotnet
 
 import java.io.File
 import java.net.URI
+import java.lang.NumberFormatException
+import java.net.{InetAddress, URI, UnknownHostException}
 import java.nio.file.attribute.PosixFilePermissions
 import java.nio.file.{FileSystems, Files, Paths}
 import java.util.Locale
@@ -51,6 +53,7 @@ object DotnetRunner extends Logging {
     // In debug mode this runner will not launch a .NET process.
     val runInDebugMode = settings._1
     @volatile var dotnetBackendPortNumber = settings._2
+    val dotnetBackendIPAddress = settings._3
     var dotnetExecutable = ""
     var otherArgs: Array[String] = null
 
@@ -98,8 +101,9 @@ object DotnetRunner extends Logging {
       override def run() {
         // need to get back dotnetBackendPortNumber because if the value passed to init is 0
         // the port number is dynamically assigned in the backend
-        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber)
-        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber")
+        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber, dotnetBackendIPAddress)
+        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber on IP address " +
+          s"$dotnetBackendIPAddress")
         initialized.release()
         dotnetBackend.run()
       }
@@ -115,6 +119,7 @@ object DotnetRunner extends Logging {
           val builder = new ProcessBuilder(processParameters)
           val env = builder.environment()
           env.put("DOTNETBACKEND_PORT", dotnetBackendPortNumber.toString)
+          env.put("DOTNETBACKEND_IP_ADDRESS", dotnetBackendIPAddress)
 
           for ((key, value) <- Utils.getSystemProperties if key.startsWith("spark.")) {
             env.put(key, value)
@@ -264,19 +269,30 @@ object DotnetRunner extends Logging {
     returnCode
   }
 
-  private def initializeSettings(args: Array[String]): (Boolean, Int) = {
+  private def initializeSettings(args: Array[String]): (Boolean, Int, String) = {
     val runInDebugMode = (args.length == 1 || args.length == 2) && args(0).equalsIgnoreCase(
       "debug")
     var portNumber = 0
+    var dotnetBackendIPAddress = "localhost"
     if (runInDebugMode) {
       if (args.length == 1) {
         portNumber = DEBUG_PORT
       } else if (args.length == 2) {
-        portNumber = Integer.parseInt(args(1))
+          portNumber = Integer.parseInt(args(1))
+        }
+      }
+    else {
+      try {
+        var addr = InetAddress.getByName(args(0))
+	      dotnetBackendIPAddress = args(0)
+      }
+      catch {
+        case e: UnknownHostException =>
+          dotnetBackendIPAddress = "localhost"
       }
     }
 
-    (runInDebugMode, portNumber)
+    (runInDebugMode, portNumber, dotnetBackendIPAddress)
   }
 
   private def logThrowable(throwable: Throwable): Unit =

src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala

@@ -30,7 +30,7 @@ class DotnetBackend extends Logging {
   private[this] var bootstrap: ServerBootstrap = _
   private[this] var bossGroup: EventLoopGroup = _
 
-  def init(portNumber: Int): Int = {
+  def init(portNumber: Int, ipAddress: String): Int = {
     // need at least 3 threads, use 10 here for safety
     bossGroup = new NioEventLoopGroup(10)
     val workerGroup = bossGroup
@@ -57,7 +57,7 @@ class DotnetBackend extends Logging {
       }
     })
 
-    channelFuture = bootstrap.bind(new InetSocketAddress("localhost", portNumber))
+    channelFuture = bootstrap.bind(new InetSocketAddress(ipAddress, portNumber))
     channelFuture.syncUninterruptibly()
     channelFuture.channel().localAddress().asInstanceOf[InetSocketAddress].getPort
   }

src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala

@@ -7,7 +7,8 @@
 package org.apache.spark.deploy.dotnet
 
 import java.io.File
-import java.net.URI
+import java.lang.NumberFormatException
+import java.net.{InetAddress, URI, UnknownHostException}
 import java.nio.file.attribute.PosixFilePermissions
 import java.nio.file.{FileSystems, Files, Paths}
 import java.util.Locale
@@ -51,6 +52,7 @@ object DotnetRunner extends Logging {
     // In debug mode this runner will not launch a .NET process.
     val runInDebugMode = settings._1
     @volatile var dotnetBackendPortNumber = settings._2
+    val dotnetBackendIPAddress = settings._3
     var dotnetExecutable = ""
     var otherArgs: Array[String] = null
 
@@ -98,8 +100,9 @@ object DotnetRunner extends Logging {
       override def run() {
         // need to get back dotnetBackendPortNumber because if the value passed to init is 0
         // the port number is dynamically assigned in the backend
-        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber)
-        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber")
+        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber, dotnetBackendIPAddress)
+        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber on IP address " +
+          s"$dotnetBackendIPAddress")
         initialized.release()
         dotnetBackend.run()
       }
@@ -115,6 +118,7 @@ object DotnetRunner extends Logging {
           val builder = new ProcessBuilder(processParameters)
           val env = builder.environment()
           env.put("DOTNETBACKEND_PORT", dotnetBackendPortNumber.toString)
+          env.put("DOTNETBACKEND_IP_ADDRESS", dotnetBackendIPAddress)
 
           for ((key, value) <- Utils.getSystemProperties if key.startsWith("spark.")) {
             env.put(key, value)
@@ -264,19 +268,30 @@ object DotnetRunner extends Logging {
     returnCode
   }
 
-  private def initializeSettings(args: Array[String]): (Boolean, Int) = {
+  private def initializeSettings(args: Array[String]): (Boolean, Int, String) = {
     val runInDebugMode = (args.length == 1 || args.length == 2) && args(0).equalsIgnoreCase(
       "debug")
     var portNumber = 0
+    var dotnetBackendIPAddress = "localhost"
     if (runInDebugMode) {
       if (args.length == 1) {
         portNumber = DEBUG_PORT
       } else if (args.length == 2) {
-        portNumber = Integer.parseInt(args(1))
+          portNumber = Integer.parseInt(args(1))
+        }
+      }
+    else {
+      try {
+        var addr = InetAddress.getByName(args(0))
+	      dotnetBackendIPAddress = args(0)
+      }
+      catch {
+        case e: UnknownHostException =>
+          dotnetBackendIPAddress = "localhost"
       }
     }
 
-    (runInDebugMode, portNumber)
+    (runInDebugMode, portNumber, dotnetBackendIPAddress)
   }
 
   private def logThrowable(throwable: Throwable): Unit =

src/scala/microsoft-spark-3.0.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala

@@ -30,7 +30,7 @@ class DotnetBackend extends Logging {
   private[this] var bootstrap: ServerBootstrap = _
   private[this] var bossGroup: EventLoopGroup = _
 
-  def init(portNumber: Int): Int = {
+  def init(portNumber: Int, ipAddress: String): Int = {
     // need at least 3 threads, use 10 here for safety
     bossGroup = new NioEventLoopGroup(10)
     val workerGroup = bossGroup
@@ -57,7 +57,7 @@ class DotnetBackend extends Logging {
       }
     })
 
-    channelFuture = bootstrap.bind(new InetSocketAddress("localhost", portNumber))
+    channelFuture = bootstrap.bind(new InetSocketAddress(ipAddress, portNumber))
     channelFuture.syncUninterruptibly()
     channelFuture.channel().localAddress().asInstanceOf[InetSocketAddress].getPort
   }

src/scala/microsoft-spark-3.0.x/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala

@@ -8,6 +8,8 @@ package org.apache.spark.deploy.dotnet
 
 import java.io.File
 import java.net.URI
+import java.lang.NumberFormatException
+import java.net.{InetAddress, URI, UnknownHostException}
 import java.nio.file.attribute.PosixFilePermissions
 import java.nio.file.{FileSystems, Files, Paths}
 import java.util.Locale
@@ -51,6 +53,7 @@ object DotnetRunner extends Logging {
     // In debug mode this runner will not launch a .NET process.
     val runInDebugMode = settings._1
     @volatile var dotnetBackendPortNumber = settings._2
+    val dotnetBackendIPAddress = settings._3
     var dotnetExecutable = ""
     var otherArgs: Array[String] = null
 
@@ -98,8 +101,9 @@ object DotnetRunner extends Logging {
       override def run() {
         // need to get back dotnetBackendPortNumber because if the value passed to init is 0
         // the port number is dynamically assigned in the backend
-        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber)
-        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber")
+        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber, dotnetBackendIPAddress)
+        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber on IP address " +
+          s"$dotnetBackendIPAddress")
         initialized.release()
         dotnetBackend.run()
       }
@@ -115,6 +119,7 @@ object DotnetRunner extends Logging {
           val builder = new ProcessBuilder(processParameters)
           val env = builder.environment()
           env.put("DOTNETBACKEND_PORT", dotnetBackendPortNumber.toString)
+          env.put("DOTNETBACKEND_IP_ADDRESS", dotnetBackendIPAddress)
 
           for ((key, value) <- Utils.getSystemProperties if key.startsWith("spark.")) {
             env.put(key, value)
@@ -264,19 +269,30 @@ object DotnetRunner extends Logging {
     returnCode
   }
 
-  private def initializeSettings(args: Array[String]): (Boolean, Int) = {
+  private def initializeSettings(args: Array[String]): (Boolean, Int, String) = {
     val runInDebugMode = (args.length == 1 || args.length == 2) && args(0).equalsIgnoreCase(
       "debug")
     var portNumber = 0
+    var dotnetBackendIPAddress = "localhost"
     if (runInDebugMode) {
       if (args.length == 1) {
         portNumber = DEBUG_PORT
       } else if (args.length == 2) {
-        portNumber = Integer.parseInt(args(1))
+          portNumber = Integer.parseInt(args(1))
+        }
+      }
+    else {
+      try {
+        var addr = InetAddress.getByName(args(0))
+	      dotnetBackendIPAddress = args(0)
+      }
+      catch {
+        case e: UnknownHostException =>
+          dotnetBackendIPAddress = "localhost"
       }
     }
 
-    (runInDebugMode, portNumber)
+    (runInDebugMode, portNumber, dotnetBackendIPAddress)
   }
 
   private def logThrowable(throwable: Throwable): Unit =