Adding environment variable for dotnet backend ip address

dotnet · Niharikadutta · Apr 20, 2020 · Apr 20, 2020 · May 6, 2020 · May 7, 2020
commit 8b3fd40b970eb480fb5d6677e29518242fdb9e40
diff --git a/src/csharp/Microsoft.Spark/Interop/Ipc/JvmBridge.cs b/src/csharp/Microsoft.Spark/Interop/Ipc/JvmBridge.cs
@@ -51,8 +51,9 @@ private ISocketWrapper GetConnection()
         {
             if (!_sockets.TryDequeue(out ISocketWrapper socket))
             {
+                IPEndPoint dotnetBackendIPEndpoint = SparkEnvironment.ConfigurationService.GetBackendIPEndpoint();
                 socket = SocketFactory.CreateSocket();
-                socket.Connect(IPAddress.Loopback, _portNumber);
+                socket.Connect(dotnetBackendIPEndpoint.Address, dotnetBackendIPEndpoint.Port);
             }
 
             return socket;

diff --git a/src/csharp/Microsoft.Spark/Network/DefaultSocketWrapper.cs b/src/csharp/Microsoft.Spark/Network/DefaultSocketWrapper.cs
@@ -30,15 +30,7 @@ internal sealed class DefaultSocketWrapper : ISocketWrapper
         public DefaultSocketWrapper() :
             this(new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
         {
-            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));
-            }
+            _innerSocket.Bind(new IPEndPoint(IPAddress.Any, 0));
         }
 
         /// <summary>

diff --git a/src/csharp/Microsoft.Spark/RDD.cs b/src/csharp/Microsoft.Spark/RDD.cs
@@ -6,6 +6,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Net;
+using Microsoft.Spark.Interop;
 using Microsoft.Spark.Interop.Ipc;
 using Microsoft.Spark.Network;
 using Microsoft.Spark.Utils;
@@ -262,7 +263,8 @@ public IEnumerable<T> Collect()
         {
             (int port, string secret) = CollectAndServe();
             using ISocketWrapper socket = SocketFactory.CreateSocket();
-            socket.Connect(IPAddress.Loopback, port, secret);
+            IPEndPoint dotnetBackendIPEndpoint = SparkEnvironment.ConfigurationService.GetBackendIPEndpoint();
+            socket.Connect(dotnetBackendIPEndpoint.Address, port, secret);
 
             var collector = new RDD.Collector();
             System.IO.Stream stream = socket.InputStream;

diff --git a/src/csharp/Microsoft.Spark/Services/ConfigurationService.cs b/src/csharp/Microsoft.Spark/Services/ConfigurationService.cs
@@ -4,6 +4,7 @@
 
 using System;
 using System.IO;
+using System.Net;
 using System.Runtime.InteropServices;
 
 namespace Microsoft.Spark.Services
@@ -20,7 +21,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 string DotnetBackendIPAddressEnvVarName = "DOTNET_SPARK_BACKEND_IP_ADDRESS";
         private const int DotnetBackendDebugPort = 5567;
 
         private static readonly string s_procBaseFileName = "Microsoft.Spark.Worker";
@@ -34,6 +35,29 @@ internal sealed class ConfigurationService : IConfigurationService
 
         private string _workerPath;
 
+        /// <summary>
+        /// Returns the IP Endpoint for socket communication between JVM and CLR.
+        /// </summary>
+        public IPEndPoint GetBackendIPEndpoint()
+        {
+            if (!int.TryParse(
+                Environment.GetEnvironmentVariable(DotnetBackendPortEnvVarName),
+                out int portNumber))
+            {
+                _logger.LogInfo($"'{DotnetBackendPortEnvVarName}' environment variable is not set.");
+                portNumber = DotnetBackendDebugPort;
+            }
+            string ipAddress = Environment.GetEnvironmentVariable(DotnetBackendIPAddressEnvVarName);
+            if (ipAddress == null)
+            {
+                _logger.LogInfo($"'{DotnetBackendIPAddressEnvVarName}' environment variable is not set.");
+                ipAddress = "127.0.0.1";
+            }
+            _logger.LogInfo($"Using IP address {ipAddress} and port {portNumber} for connection.");
+
+            return new IPEndPoint(IPAddress.Parse(ipAddress), portNumber);
+        }
+
         /// <summary>
         /// Returns the port number for socket communication between JVM and CLR.
         /// </summary>

diff --git a/src/csharp/Microsoft.Spark/Services/IConfigurationService.cs b/src/csharp/Microsoft.Spark/Services/IConfigurationService.cs
@@ -2,6 +2,8 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System.Net;
+
 namespace Microsoft.Spark.Services
 {
     /// <summary>
@@ -14,6 +16,11 @@ internal interface IConfigurationService
         /// </summary>
         int GetBackendPortNumber();
 
+        /// <summary>
+        /// The IP Endpoint used for communicating with the .NET backend process.
+        /// </summary>
+        IPEndPoint GetBackendIPEndpoint();
+
         /// <summary>
         /// The IP address used for communicating with the .NET backend process.
         /// </summary>

diff --git a/src/csharp/Microsoft.Spark/Sql/DataFrame.cs b/src/csharp/Microsoft.Spark/Sql/DataFrame.cs
@@ -886,9 +886,10 @@ public DataStreamWriter WriteStream() =>
         private IEnumerable<Row> GetRows(string funcName)
         {
             (int port, string secret) = GetConnectionInfo(funcName);
+            IPEndPoint dotnetBackendIPEndpoint = SparkEnvironment.ConfigurationService.GetBackendIPEndpoint();
             using (ISocketWrapper socket = SocketFactory.CreateSocket())
             {
-                socket.Connect(IPAddress.Loopback, port, secret);
+                socket.Connect(dotnetBackendIPEndpoint.Address, port, secret);
                 foreach (Row row in new RowCollector().Collect(socket))
                 {
                     yield return row;

diff --git a/...cala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala b/...cala/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, ipAddress: String): Int = {
+  def init(ipAddress: String, portNumber: Int): Int = {
     // need at least 3 threads, use 10 here for safety
     bossGroup = new NioEventLoopGroup(10)
     val workerGroup = bossGroup

diff --git a/...la/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala b/...la/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala
@@ -52,7 +52,6 @@ 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
 
@@ -91,6 +90,7 @@ object DotnetRunner extends Logging {
 
     // Time to wait for DotnetBackend to initialize in seconds.
     val backendTimeout = sys.env.getOrElse("DOTNETBACKEND_TIMEOUT", "120").toInt
+    val dotnetBackendIPAddress = sys.env.getOrElse("DOTNET_SPARK_BACKEND_IP_ADDRESS", "127.0.0.1")
 
     // Launch a DotnetBackend server for the .NET process to connect to; this will let it see our
     // Java system properties etc.
@@ -100,7 +100,7 @@ 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, dotnetBackendIPAddress)
+        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendIPAddress, dotnetBackendPortNumber)
         logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber on IP address " +
           s"$dotnetBackendIPAddress")
         initialized.release()
@@ -118,7 +118,6 @@ 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)
@@ -268,30 +267,19 @@ object DotnetRunner extends Logging {
     returnCode
   }
 
-  private def initializeSettings(args: Array[String]): (Boolean, Int, String) = {
+  private def initializeSettings(args: Array[String]): (Boolean, Int) = {
     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))
         }
       }
-    else {
-      try {
-        var addr = InetAddress.getByName(args(0))
-	      dotnetBackendIPAddress = args(0)
-      }
-      catch {
-        case e: UnknownHostException =>
-          dotnetBackendIPAddress = "localhost"
-      }
-    }
 
-    (runInDebugMode, portNumber, dotnetBackendIPAddress)
+    (runInDebugMode, portNumber)
   }
 
   private def logThrowable(throwable: Throwable): Unit =