update the port reserve of am and container based on the pre commit

Author: jiarunying-it@360.cn

src/main/java/net/qihoo/xlearning/AM/ApplicationMaster.java

@@ -111,6 +111,9 @@ public class ApplicationMaster extends CompositeService {
 
   private int outputIndex;
 
+  private int reservePortBegin = 0;
+  private int reservePortEnd = 0;
+
   /**
    * Constructor, connect to Resource Manager
    *
@@ -155,6 +158,10 @@ private ApplicationMaster() {
     tfEvaluatorContainerId = "";
     inputPath = new StringBuilder();
     outputIndex = -1;
+    this.reservePortBegin = this.conf.getInt(XLearningConfiguration.XLEARNING_RESERVE_PORT_BEGIN,
+        XLearningConfiguration.DEFAULT_XLEARNING_RESERVE_PORT_BEGIN);
+    this.reservePortEnd = this.conf.getInt(XLearningConfiguration.XLEARNING_RESERVE_PORT_END,
+        XLearningConfiguration.DEFAULT_XLEARNING_RESERVE_PORT_END);
 
     if (envs.containsKey(ApplicationConstants.Environment.CONTAINER_ID.toString())) {
       ContainerId containerId = ConverterUtils
@@ -1151,7 +1158,7 @@ private boolean run() throws IOException, NoSuchAlgorithmException {
       dmlcPsRootUri = applicationMasterHostname;
       Socket schedulerReservedSocket = new Socket();
       try {
-        schedulerReservedSocket.bind(new InetSocketAddress("127.0.0.1", 0));
+        Utilities.getReservePort(schedulerReservedSocket, InetAddress.getByName(applicationMasterHostname).getHostAddress(), reservePortBegin, reservePortEnd);
       } catch (IOException e) {
         LOG.error("Can not get available port");
       }
@@ -1256,7 +1263,7 @@ public void run() {
       dmlcTrackerUri = applicationMasterHostname;
       Socket schedulerReservedSocket = new Socket();
       try {
-        schedulerReservedSocket.bind(new InetSocketAddress("127.0.0.1", 0));
+        Utilities.getReservePort(schedulerReservedSocket, InetAddress.getByName(applicationMasterHostname).getHostAddress(), reservePortBegin, reservePortEnd);
       } catch (IOException e) {
         LOG.error("Can not get available port");
       }
@@ -1364,7 +1371,7 @@ public void run() {
       }
       Socket schedulerReservedSocket = new Socket();
       try {
-        schedulerReservedSocket.bind(new InetSocketAddress("127.0.0.1", 0));
+        Utilities.getReservePort(schedulerReservedSocket, InetAddress.getByName(applicationMasterHostname).getHostAddress(), reservePortBegin, reservePortEnd);
       } catch (IOException e) {
         LOG.error("Can not get available port");
       }

src/main/java/net/qihoo/xlearning/container/XLearningContainer.java

@@ -85,6 +85,8 @@ public class XLearningContainer {
 
   private int outputIndex;
 
+  private String localHost;
+
   private XLearningContainer() {
     this.conf = new XLearningConfiguration();
     conf.addResource(new Path(XLearningConstants.XLEARNING_JOB_CONFIGURATION));
@@ -93,6 +95,11 @@ private XLearningContainer() {
         .getenv(ApplicationConstants.Environment.CONTAINER_ID.name())));
     this.downloadRetry = conf.getInt(XLearningConfiguration.XLEARNING_DOWNLOAD_FILE_RETRY, XLearningConfiguration.DEFAULT_XLEARNING_DOWNLOAD_FILE_RETRY);
     this.envs = System.getenv();
+    if (envs.containsKey(ApplicationConstants.Environment.NM_HOST.toString())) {
+      localHost = envs.get(ApplicationConstants.Environment.NM_HOST.toString());
+    } else {
+      localHost = "127.0.0.1";
+    }
     this.xlearningAppType = envs.get(XLearningConstants.Environment.XLEARNING_APP_TYPE.toString()).toUpperCase();
     this.role = envs.get(XLearningConstants.Environment.XLEARNING_TF_ROLE.toString());
     this.index = Integer.valueOf(envs.get(XLearningConstants.Environment.XLEARNING_TF_INDEX.toString()));
@@ -183,7 +190,7 @@ private void init() {
 
     if ((("TENSORFLOW".equals(xlearningAppType) || "LIGHTLDA".equals(xlearningAppType)) && !single) || xlearningAppType.equals("DISTLIGHTGBM")) {
       try {
-        getReservePort(reservedSocket);
+        Utilities.getReservePort(reservedSocket, InetAddress.getByName(localHost).getHostAddress(), reservePortBegin, reservePortEnd);
       } catch (IOException e) {
         LOG.error("Can not get available port");
         reportFailedAndExit();
@@ -832,7 +839,7 @@ public void run() {
     if (boardEnable && this.role.equals(XLearningConstants.WORKER) && boardIndex == this.index) {
       Socket boardReservedSocket = new Socket();
       try {
-        boardReservedSocket.bind(new InetSocketAddress("127.0.0.1", 0));
+        Utilities.getReservePort(boardReservedSocket, InetAddress.getByName(localHost).getHostAddress(), reservePortBegin, reservePortEnd);
       } catch (IOException e) {
         LOG.error("Can not get available port");
         reportFailedAndExit();
@@ -969,23 +976,6 @@ private void reportSucceededAndExit() {
     System.exit(0);
   }
 
-  private void getReservePort(Socket socket) throws IOException {
-    int i = 0;
-    Random random = new Random(System.currentTimeMillis());
-    while (i < 1000) {
-      int rand = random.nextInt(reservePortEnd - reservePortBegin);
-      try {
-        socket.bind(new InetSocketAddress("127.0.0.1", reservePortBegin + rand));
-        return;
-      } catch (IOException e) {
-        try {
-          Thread.sleep(1000);
-        } catch (InterruptedException e2) {}
-      }
-    }
-    throw new IOException("couldn't allocate a unused port");
-  }
-
   public static void main(String[] args) {
     XLearningContainer container = new XLearningContainer();
     try {

src/main/java/net/qihoo/xlearning/util/Utilities.java

@@ -15,9 +15,12 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
 
 public final class Utilities {
   private static Log LOG = LogFactory.getLog(Utilities.class);
@@ -136,4 +139,21 @@ public static void addPathToEnvironment(Map<String, String> env, String userEnvK
     }
   }
 
+  public static void getReservePort(Socket socket, String localHost, int reservePortBegin, int reservePortEnd) throws IOException {
+    int i = 0;
+    Random random = new Random(System.currentTimeMillis());
+    while (i < 1000) {
+      int rand = random.nextInt(reservePortEnd - reservePortBegin);
+      try {
+        socket.bind(new InetSocketAddress(localHost, reservePortBegin + rand));
+        return;
+      } catch (IOException e) {
+        try {
+          Thread.sleep(1000);
+        } catch (InterruptedException e2) {}
+      }
+    }
+    throw new IOException("couldn't allocate a unused port");
+  }
+
 }