Merge pull request #891 from tqchen/master

[PYTHON] Simplify training logic, update rabit lib

Makefile

@@ -118,7 +118,7 @@ lib/libxgboost.a: $(ALL_DEP)
 
 lib/libxgboost.dll lib/libxgboost.so: $(ALL_DEP)
 	@mkdir -p $(@D)
-	$(CXX) $(CFLAGS) -shared -o $@ $(filter %.o %.a, $^) $(LDFLAGS)
+	$(CXX) $(CFLAGS) -shared -o $@ $(filter %.o %a,  $^) $(LDFLAGS)
 
 java/libxgboost4j.so: java/xgboost4j_wrapper.cpp $(ALL_DEP)
 	$(CXX) $(CFLAGS) $(JAVAINCFLAGS) -shared -o $@ $(filter %.cpp %.o %.a, $^) $(LDFLAGS)

R-package/src/Makevars

@@ -14,4 +14,5 @@ PKG_CPPFLAGS=  -I$(PKGROOT)/include -I$(PKGROOT)/dmlc-core/include -I$(PKGROOT)/
 PKG_CXXFLAGS= $(SHLIB_OPENMP_CFLAGS) $(SHLIB_PTHREAD_FLAGS)
 PKG_LIBS = $(SHLIB_OPENMP_CFLAGS) $(SHLIB_PTHREAD_FLAGS)
 OBJECTS= ./xgboost_R.o ./xgboost_custom.o ./xgboost_assert.o\
-         $(PKGROOT)/amalgamation/xgboost-all0.o $(PKGROOT)/amalgamation/dmlc-minimum0.o $(PKGROOT)/rabit/src/engine_empty.o
+         $(PKGROOT)/amalgamation/xgboost-all0.o $(PKGROOT)/amalgamation/dmlc-minimum0.o\
+         $(PKGROOT)/rabit/src/engine_empty.o $(PKGROOT)/rabit/src/c_api.o

R-package/src/Makevars.win

@@ -26,6 +26,7 @@ PKG_CPPFLAGS=  -I$(PKGROOT)/include -I$(PKGROOT)/dmlc-core/include -I$(PKGROOT)/
 PKG_CXXFLAGS= $(SHLIB_OPENMP_CFLAGS) $(SHLIB_PTHREAD_FLAGS)
 PKG_LIBS = $(SHLIB_OPENMP_CFLAGS) $(SHLIB_PTHREAD_FLAGS)
 OBJECTS= ./xgboost_R.o ./xgboost_custom.o ./xgboost_assert.o\
-         $(PKGROOT)/amalgamation/xgboost-all0.o $(PKGROOT)/amalgamation/dmlc-minimum0.o $(PKGROOT)/rabit/src/engine_empty.o
+         $(PKGROOT)/amalgamation/xgboost-all0.o $(PKGROOT)/amalgamation/dmlc-minimum0.o\
+         $(PKGROOT)/rabit/src/engine_empty.o $(PKGROOT)/rabit/src/c_api.o
 
 $(OBJECTS) : xgblib

include/xgboost/c_api.h

@@ -11,6 +11,9 @@
 #define XGB_EXTERN_C extern "C"
 #endif
 
+// XGBoost C API will include APIs in Rabit C API
+#include <rabit/c_api.h>
+
 #if defined(_MSC_VER) || defined(_WIN32)
 #define XGB_DLL XGB_EXTERN_C __declspec(dllexport)
 #else
@@ -221,6 +224,7 @@ XGB_DLL int XGBoosterFree(BoosterHandle handle);
 XGB_DLL int XGBoosterSetParam(BoosterHandle handle,
                               const char *name,
                               const char *value);
+
 /*!
  * \brief update the model in one round using dtrain
  * \param handle handle
@@ -282,6 +286,7 @@ XGB_DLL int XGBoosterPredict(BoosterHandle handle,
                              unsigned ntree_limit,
                              bst_ulong *out_len,
                              const float **out_result);
+
 /*!
  * \brief load model from existing file
  * \param handle handle
@@ -353,4 +358,24 @@ XGB_DLL int XGBoosterDumpModelWithFeatures(BoosterHandle handle,
                                            bst_ulong *out_len,
                                            const char ***out_models);
 
+// --- Distributed training API----
+// NOTE: functions in rabit/c_api.h will be also available in libxgboost.so
+/*!
+ * \brief Initialize the booster from rabit checkpoint.
+ *  This is used in distributed training API.
+ * \param handle handle
+ * \param version The output version of the model.
+ * \return 0 when success, -1 when failure happens
+ */
+XGB_DLL int XGBoosterLoadRabitCheckpoint(
+    BoosterHandle handle,
+    int* version);
+
+/*!
+ * \brief Save the current checkpoint to rabit.
+ * \param handle handle
+ * \return 0 when success, -1 when failure happens
+ */
+XGB_DLL int XGBoosterSaveRabitCheckPoint(BoosterHandle handle);
+
 #endif  // XGBOOST_C_API_H_

include/xgboost/learner.h

@@ -8,7 +8,7 @@
 #ifndef XGBOOST_LEARNER_H_
 #define XGBOOST_LEARNER_H_
 
-#include <rabit.h>
+#include <rabit/rabit.h>
 #include <utility>
 #include <string>
 #include <vector>

python-package/xgboost/training.py

@@ -24,7 +24,7 @@ def train(params, dtrain, num_boost_round=10, evals=(), obj=None, feval=None,
         Data to be trained.
     num_boost_round: int
         Number of boosting iterations.
-    watchlist (evals): list of pairs (DMatrix, string)
+    evals: list of pairs (DMatrix, string)
         List of items to be evaluated during training, this allows user to watch
         performance on the validation set.
     obj : function
@@ -117,48 +117,13 @@ def train(params, dtrain, num_boost_round=10, evals=(), obj=None, feval=None,
             evals_result.clear()
             evals_result.update(dict([(key, {}) for key in evals_name]))
 
-    if not early_stopping_rounds:
-        for i in range(nboost, nboost + num_boost_round):
-            bst.update(dtrain, i, obj)
-            nboost += 1
-            if len(evals) != 0:
-                bst_eval_set = bst.eval_set(evals, i, feval)
-                if isinstance(bst_eval_set, STRING_TYPES):
-                    msg = bst_eval_set
-                else:
-                    msg = bst_eval_set.decode()
-
-                if verbose_eval:
-                    if verbose_eval_every_line:
-                        if i % verbose_eval_every_line == 0 or i == num_boost_round - 1:
-                            sys.stderr.write(msg + '\n')
-                    else:
-                        sys.stderr.write(msg + '\n')
-
-                if evals_result is not None:
-                    res = re.findall("([0-9a-zA-Z@]+[-]*):-?([0-9.]+).", msg)
-                    for key in evals_name:
-                        evals_idx = evals_name.index(key)
-                        res_per_eval = len(res) // len(evals_name)
-                        for r in range(res_per_eval):
-                            res_item = res[(evals_idx*res_per_eval) + r]
-                            res_key = res_item[0]
-                            res_val = res_item[1]
-                            if res_key in evals_result[key]:
-                                evals_result[key][res_key].append(res_val)
-                            else:
-                                evals_result[key][res_key] = [res_val]
-        bst.best_iteration = (nboost - 1)
-        bst.best_ntree_limit = nboost * num_parallel_tree
-        return bst
-
-    else:
     # early stopping
+    if early_stopping_rounds is not None:
         if len(evals) < 1:
             raise ValueError('For early stopping you need at least one set in evals.')
 
         if verbose_eval:
-            sys.stderr.write("Will train until {} error hasn't decreased in {} rounds.\n".format(\
+            sys.stderr.write("Will train until {} error hasn't decreased in {} rounds.\n".format(
                 evals[-1][1], early_stopping_rounds))
 
         # is params a list of tuples? are we using multiple eval metrics?
@@ -197,7 +162,10 @@ def train(params, dtrain, num_boost_round=10, evals=(), obj=None, feval=None,
             else:
                 bst.set_param({'eta': learning_rates(i, num_boost_round)})
         bst.update(dtrain, i, obj)
+
         nboost += 1
+        # check evaluation result.
+        if len(evals) != 0:
             bst_eval_set = bst.eval_set(evals, i, feval)
 
             if isinstance(bst_eval_set, STRING_TYPES):
@@ -226,6 +194,7 @@ def train(params, dtrain, num_boost_round=10, evals=(), obj=None, feval=None,
                         else:
                             evals_result[key][res_key] = [res_val]
 
+            if early_stopping_rounds:
                 score = float(msg.rsplit(':', 1)[1])
                 if (maximize_score and score > best_score) or \
                    (not maximize_score and score < best_score):
@@ -235,11 +204,16 @@ def train(params, dtrain, num_boost_round=10, evals=(), obj=None, feval=None,
                 elif i - best_score_i >= early_stopping_rounds:
                     if verbose_eval:
                         sys.stderr.write("Stopping. Best iteration:\n{}\n\n".format(best_msg))
+                    # best iteration will be assigned in the end.
                     bst.best_score = best_score
                     bst.best_iteration = best_score_i
                     break
-        bst.best_score = best_score
+
+    if early_stopping_rounds:
+        best_score = best_score
         bst.best_iteration = best_score_i
+    else:
+        bst.best_iteration = nboost - 1
     bst.best_ntree_limit = (bst.best_iteration + 1) * num_parallel_tree
     return bst
 
@@ -486,4 +460,3 @@ def cv(params, dtrain, num_boost_round=10, nfold=3, stratified=False, folds=None
         results = np.array(results)
 
     return results
-

rabit

@@ -1 +1 @@
-Subproject commit 112d866dc92354304c0891500374fe40cdf13a50
+Subproject commit 56ec4263f9a70a315c1f153dc5897b7c1b58250c

src/c_api/c_api.cc

@@ -3,6 +3,8 @@
 #include <xgboost/data.h>
 #include <xgboost/learner.h>
 #include <xgboost/c_api.h>
+#include <xgboost/logging.h>
+#include <rabit/rabit.h>
 #include <cstdio>
 #include <vector>
 #include <string>
@@ -84,6 +86,10 @@ int XGDMatrixCreateFromFile(const char *fname,
                             int silent,
                             DMatrixHandle *out) {
   API_BEGIN();
+  if (rabit::IsDistributed()) {
+    LOG(CONSOLE) << "XGBoost distributed mode detected, "
+                 << "will split data among workers";
+  }
   *out = DMatrix::Load(
       fname, silent != 0, false);
   API_END();
@@ -526,3 +532,28 @@ int XGBoosterDumpModelWithFeatures(BoosterHandle handle,
   XGBoostDumpModelImpl(handle, featmap, with_stats, len, out_models);
   API_END();
 }
+
+int XGBoosterLoadRabitCheckpoint(BoosterHandle handle,
+                                 int* version) {
+  API_BEGIN();
+  Booster* bst = static_cast<Booster*>(handle);
+  *version = rabit::LoadCheckPoint(bst->learner());
+  if (version != 0) {
+    bst->initialized_ = true;
+  }
+  API_END();
+}
+
+int XGBoosterSaveRabitCheckPoint(BoosterHandle handle) {
+  API_BEGIN();
+  Booster* bst = static_cast<Booster*>(handle);
+  if (bst->learner()->AllowLazyCheckPoint()) {
+    rabit::LazyCheckPoint(bst->learner());
+  } else {
+    rabit::CheckPoint(bst->learner());
+  }
+  API_END();
+}
+
+// force link rabit
+static int XGBOOST_LINK_RABIT_C_API_ = RabitLinkTag();

src/common/sync.h

@@ -8,6 +8,6 @@
 #ifndef XGBOOST_COMMON_SYNC_H_
 #define XGBOOST_COMMON_SYNC_H_
 
-#include <rabit.h>
+#include <rabit/rabit.h>
 
 #endif  // XGBOOST_COMMON_SYNC_H_

tests/python/test_models.py

@@ -31,7 +31,7 @@ class TestModels(unittest.TestCase):
 
 		# learning_rates as a customized decay function
 		def eta_decay(ithround, num_boost_round):
-			return num_boost_round / ithround
+			return num_boost_round / (ithround + 1)
 		bst = xgb.train(param, dtrain, num_round, watchlist, learning_rates=eta_decay)
 		assert isinstance(bst, xgb.core.Booster)