Merge pull request #907 from tqchen/master

[DIST] Enable multiple thread  make rabit and xgboost threadsafe

Author: tqchen

.gitignore

@@ -78,5 +78,4 @@ tags
 *.iml
 *.class
 target
-
 *.swp

NEWS.md

@@ -22,6 +22,11 @@ This file records the changes in xgboost library in reverse chronological order.
   - The windows version is still blocked due to Rtools do not support ```std::thread```.
 * rabit and dmlc-core are maintained through git submodule
   - Anyone can open PR to update these dependencies now.
+* Improvements
+  - Rabit and xgboost libs are not thread-safe and use thread local PRNGs
+  - This could fix some of the previous problem which runs xgboost on multiple threads.
+* JVM Package
+  - Enable xgboost4j for java and scala
 
 ## v0.47 (2016.01.14)
 

jvm-packages/.gitignore

@@ -0,0 +1,2 @@
+tracker.py
+build.sh

jvm-packages/create_jni.sh

@@ -27,6 +27,8 @@ fi
 
 rm -f xgboost4j/src/main/resources/lib/libxgboost4j.${dl}
 mv lib/libxgboost4j.so xgboost4j/src/main/resources/lib/libxgboost4j.${dl}
+# copy python to native resources
+cp ../dmlc-core/tracker/dmlc_tracker/tracker.py xgboost4j/src/main/resources/tracker.py
 
 popd > /dev/null
 echo "complete"

jvm-packages/test_distributed.sh

@@ -1,5 +1,5 @@
 #!/bin/bash
 # Simple script to test distributed version, to be deleted later.
 cd xgboost4j-demo
-../../dmlc-core/tracker/dmlc-submit --cluster=local --num-workers=3 java -cp target/xgboost4j-demo-0.1-jar-with-dependencies.jar ml.dmlc.xgboost4j.demo.DistTrain
+java  -XX:OnError="gdb - %p" -cp target/xgboost4j-demo-0.1-jar-with-dependencies.jar ml.dmlc.xgboost4j.demo.DistTrain 4
 cd ..

jvm-packages/xgboost4j-demo/src/main/java/ml/dmlc/xgboost4j/demo/DistTrain.java

@@ -2,42 +2,79 @@
 
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 import ml.dmlc.xgboost4j.*;
 
+
 /**
  * Distributed training example, used to quick test distributed training.
  *
  * @author tqchen
  */
 public class DistTrain {
+  private static final Log logger = LogFactory.getLog(DistTrain.class);
+  private Map<String, String> envs = null;
+
+  private class Worker implements Runnable {
+    private final int workerId;
+
+    Worker(int workerId) {
+      this.workerId = workerId;
+    }
 
-  public static void main(String[] args) throws IOException, XGBoostError {
-    // always initialize rabit module before training.
-    Rabit.init(new HashMap<String, String>());
+    public void run() {
+      try {
+        Map<String, String> worker_env = new HashMap<String, String>(envs);
 
-    // load file from text file, also binary buffer generated by xgboost4j
-    DMatrix trainMat = new DMatrix("../../demo/data/agaricus.txt.train");
-    DMatrix testMat = new DMatrix("../../demo/data/agaricus.txt.test");
+        worker_env.put("DMLC_TASK_ID", String.valueOf(workerId));
+        // always initialize rabit module before training.
+        Rabit.init(worker_env);
 
-    HashMap<String, Object> params = new HashMap<String, Object>();
-    params.put("eta", 1.0);
-    params.put("max_depth", 2);
-    params.put("silent", 1);
-    params.put("objective", "binary:logistic");
+        // load file from text file, also binary buffer generated by xgboost4j
+        DMatrix trainMat = new DMatrix("../../demo/data/agaricus.txt.train");
+        DMatrix testMat = new DMatrix("../../demo/data/agaricus.txt.test");
 
+        HashMap<String, Object> params = new HashMap<String, Object>();
+        params.put("eta", 1.0);
+        params.put("max_depth", 2);
+        params.put("silent", 1);
+        params.put("nthread", 2);
+        params.put("objective", "binary:logistic");
 
-    HashMap<String, DMatrix> watches = new HashMap<String, DMatrix>();
-    watches.put("train", trainMat);
-    watches.put("test", testMat);
+        HashMap<String, DMatrix> watches = new HashMap<String, DMatrix>();
+        watches.put("train", trainMat);
+        watches.put("test", testMat);
 
-    //set round
-    int round = 2;
+        //set round
+        int round = 2;
 
-    //train a boost model
-    Booster booster = XGBoost.train(params, trainMat, round, watches, null, null);
+        //train a boost model
+        Booster booster = XGBoost.train(params, trainMat, round, watches, null, null);
+
+        // always shutdown rabit module after training.
+        Rabit.shutdown();
+      } catch (Exception ex){
+        logger.error(ex);
+      }
+    }
+  }
+
+  void start(int nWorkers) throws IOException, XGBoostError, InterruptedException {
+    RabitTracker tracker = new RabitTracker(nWorkers);
+    if (tracker.start()) {
+      envs = tracker.getWorkerEnvs();
+      for (int i = 0; i < nWorkers; ++i) {
+        new Thread(new Worker(i)).start();
+      }
+      tracker.waitFor();
+    }
+  }
 
-    // always shutdown rabit module after training.
-    Rabit.shutdown();
+  public static void main(String[] args) throws IOException, XGBoostError, InterruptedException {
+    new DistTrain().start(Integer.parseInt(args[0]));
   }
 }

jvm-packages/xgboost4j/src/main/java/ml/dmlc/xgboost4j/Booster.java

@@ -1,9 +1,10 @@
 package ml.dmlc.xgboost4j;
 
 import java.io.IOException;
+import java.io.Serializable;
 import java.util.Map;
 
-public interface Booster {
+public interface Booster extends Serializable {
 
   /**
    * set parameter
@@ -109,12 +110,25 @@ public interface Booster {
   float[][] predict(DMatrix data, int treeLimit, boolean predLeaf) throws XGBoostError;
 
   /**
-   * save model to modelPath
-   *
+   * save model to modelPath, the model path support depends on the path support
+   * in libxgboost. For example, if we want to save to hdfs, libxgboost need to be
+   * compiled with HDFS support.
+   * See also toByteArray
    * @param modelPath model path
    */
   void saveModel(String modelPath) throws XGBoostError;
 
+  /**
+   * Save the model as byte array representation.
+   * Write these bytes to a file will give compatible format with other xgboost bindings.
+   *
+   * If java natively support HDFS file API, use toByteArray and write the ByteArray,
+   *
+   * @return the saved byte array.
+   * @throws XGBoostError
+   */
+  byte[] toByteArray() throws XGBoostError;
+
   /**
    * Dump model into a text file.
    *

jvm-packages/xgboost4j/src/main/java/ml/dmlc/xgboost4j/JavaBoosterImpl.java

@@ -57,7 +57,6 @@ class JavaBoosterImpl implements Booster {
     setParams(params);
   }
 
-
   /**
    * load model from modelPath
    *
@@ -440,6 +439,22 @@ public Map<String, Integer> getFeatureScore(String featureMap) throws XGBoostErr
     return featureScore;
   }
 
+  /**
+   * Save the model as byte array representation.
+   * Write these bytes to a file will give compatible format with other xgboost bindings.
+   *
+   * If java natively support HDFS file API, use toByteArray and write the ByteArray,
+   *
+   * @return the saved byte array.
+   * @throws XGBoostError
+   */
+  public byte[] toByteArray() throws XGBoostError {
+    byte[][] bytes = new byte[1][];
+    JNIErrorHandle.checkCall(XgboostJNI.XGBoosterGetModelRaw(this.handle, bytes));
+    return bytes[0];
+  }
+
+
   /**
    * Load the booster model from thread-local rabit checkpoint.
    * This is only used in distributed training.
@@ -475,6 +490,27 @@ private static long[] dmatrixsToHandles(DMatrix[] dmatrixs) {
     return handles;
   }
 
+  // making Booster serializable
+  private void writeObject(java.io.ObjectOutputStream out)
+          throws IOException {
+    try {
+      out.writeObject(this.toByteArray());
+    } catch (XGBoostError ex) {
+      throw new IOException(ex.toString());
+    }
+  }
+
+  private void readObject(java.io.ObjectInputStream in)
+          throws IOException, ClassNotFoundException {
+    try {
+      this.init(null);
+      byte[] bytes = (byte[])in.readObject();
+      JNIErrorHandle.checkCall(XgboostJNI.XGBoosterLoadModelFromBuffer(this.handle, bytes));
+    } catch (XGBoostError ex) {
+      throw new IOException(ex.toString());
+    }
+  }
+
   @Override
   protected void finalize() throws Throwable {
     super.finalize();

jvm-packages/xgboost4j/src/main/java/ml/dmlc/xgboost4j/NativeLibLoader.java

@@ -21,6 +21,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+
 /**
  * class to load native library
  *
@@ -61,12 +62,32 @@ public static synchronized void initXgBoost() throws IOException {
    * three characters
    */
   private static void loadLibraryFromJar(String path) throws IOException, IllegalArgumentException{
+    String temp = createTempFileFromResource(path);
+    // Finally, load the library
+    System.load(temp);
+  }
 
+  /**
+   * Create a temp file that copies the resource from current JAR archive
+   * <p/>
+   * The file from JAR is copied into system temp file.
+   * The temporary file is deleted after exiting.
+   * Method uses String as filename because the pathname is "abstract", not system-dependent.
+   * <p/>
+   * The restrictions of {@link File#createTempFile(java.lang.String, java.lang.String)} apply to
+   * {@code path}.
+   * @param path Path to the resources in the jar
+   * @return The created temp file.
+   * @throws IOException
+   * @throws IllegalArgumentException
+   */
+  static String createTempFileFromResource(String path) throws
+          IOException, IllegalArgumentException {
+    // Obtain filename from path
     if (!path.startsWith("/")) {
       throw new IllegalArgumentException("The path has to be absolute (start with '/').");
     }
 
-    // Obtain filename from path
     String[] parts = path.split("/");
     String filename = (parts.length > 1) ? parts[parts.length - 1] : null;
 
@@ -83,7 +104,6 @@ private static void loadLibraryFromJar(String path) throws IOException, IllegalA
     if (filename == null || prefix.length() < 3) {
       throw new IllegalArgumentException("The filename has to be at least 3 characters long.");
     }
-
     // Prepare temporary file
     File temp = File.createTempFile(prefix, suffix);
     temp.deleteOnExit();
@@ -113,9 +133,7 @@ private static void loadLibraryFromJar(String path) throws IOException, IllegalA
       os.close();
       is.close();
     }
-
-    // Finally, load the library
-    System.load(temp.getAbsolutePath());
+    return temp.getAbsolutePath();
   }
 
   /**
@@ -133,8 +151,9 @@ private static void smartLoad(String libName) throws IOException {
       try {
         String libraryFromJar = nativeResourcePath + System.mapLibraryName(libName);
         loadLibraryFromJar(libraryFromJar);
-      } catch (IOException e1) {
-        throw e1;
+      } catch (IOException ioe) {
+        logger.error("failed to load library from both native path and jar");
+        throw ioe;
       }
     }
   }