more lbfgs, add spmv_perf

Makefile

@@ -18,7 +18,7 @@ data/localizer.o reader/batch_reader.o )
 
 DMLC_DEPS = dmlc-core/libdmlc.a
 
-all: build/difacto  build/bcd_learner_test
+all: build/difacto build/spmv_perf
 
 clean:
 	rm -rf build
@@ -46,6 +46,7 @@ dmlc-core/libdmlc.a:
 
 include tests/cpp/test.mk
 cpp-test: $(CPPTEST)
+cpp-perf: $(CPPPERF)
 
 test: cpp-test
 

src/common/kv_match.h

@@ -68,7 +68,7 @@ namespace difacto {
  * \param dst_val the destination values.
  * \param op the assignment operator (default is ASSIGN)
  * \param num_threads number of thread (default is 2)
- * \return the number of matched kv pairs
+ * \return the number of matched values
  */
 template <typename K, typename V>
 size_t KVMatch(
@@ -97,6 +97,21 @@ size_t KVMatch(
   return n;
 }
 
+/**
+ * \brief merge with various length values
+ *
+ * if src_offset is empty, fallback to the previous fixed value length version
+ *
+ * \param src_key the source keys
+ * \param src_val the source values
+ * \param src_offset the offsets for source values, can be empty
+ * \param dst_key the destination keys
+ * \param dst_val the destination values.
+ * \param dst_offset the offsets for destination values, can be empty
+ * \param op the assignment operator (default is ASSIGN)
+ * \param num_threads number of thread (default is 2)
+ * \return the number of matched kv pairs
+ */
 template <typename K, typename I, typename V>
 size_t KVMatch(
     const SArray<K>& src_key,
@@ -107,6 +122,11 @@ size_t KVMatch(
     SArray<I>* dst_offset,
     AssignOp op = ASSIGN,
     int num_threads = DEFAULT_NTHREADS) {
+  if (src_offset.empty()) {
+    if (dst_offset) dst_offset->clear();
+    return KVMatch(src_key, src_val, dst_key, dst_val, op, num_threads);
+  }
+  LOG(FATAL) << "TODO";
   return 0;
 }
 

src/lbfgs/lbfgs_learner.cc

@@ -4,7 +4,28 @@
 #include "reader/reader.h"
 namespace difacto {
 
+DMLC_REGISTER_PARAMETER(LBFGSLearnerParam);
+DMLC_REGISTER_PARAMETER(LBFGSUpdaterParam);
+
 KWArgs LBFGSLearner::Init(const KWArgs& kwargs) {
+  auto remain = Learner::Init(kwargs);
+  // init param
+  remain = param_.InitAllowUnknown(kwargs);
+  // init updater
+  std::shared_ptr<Updater> updater(new LBFGSUpdater());
+  remain = updater->Init(remain);
+  // init model store
+  model_store_ = Store::Create();
+  model_store_->set_updater(updater);
+  remain = model_store_->Init(remain);
+  // init data stores
+  data_store_ = new DataStore();
+  remain = model_store_->Init(remain);
+  tile_store_ = new TileStore(data_store_);
+  // init loss
+  loss_ = Loss::Create(param_.loss, nthreads_);
+  remain = loss_->Init(remain);
+  return remain;
 }
 
 void LBFGSLearner::RunScheduler() {
@@ -85,15 +106,24 @@ void LBFGSLearner::Process(const std::string& args, std::string* rets) {
   delete ss;
 }
 
+void LBFGSLearner::IssueJobAndWait(
+    int node_group, int job_type, const std::vector<real_t>& job_args,
+    std::vector<real_t>* job_rets) {
+
+}
+
+
 size_t LBFGSLearner::PrepareData() {
   // read train data
   Reader train(param_.data_in, param_.data_format,
                model_store_->Rank(), model_store_->NumWorkers(),
                param_.data_chunk_size);
+  size_t nrows = 0;
   tile_builder_ = new TileBuilder(tile_store_, nthreads_);
   SArray<real_t> feacnts;
   while (train.Next()) {
     auto rowblk = train.Value();
+    nrows += rowblk.size;
     tile_builder_->Add(rowblk, &feaids_, &feacnts);
     pred_.push_back(SArray<real_t>(rowblk.size));
     ++ntrain_blks_;
@@ -109,6 +139,7 @@ size_t LBFGSLearner::PrepareData() {
                param_.data_chunk_size);
     while (val.Next()) {
       auto rowblk = val.Value();
+      nrows += rowblk.size;
       tile_builder_->Add(rowblk);
       pred_.push_back(SArray<real_t>(rowblk.size));
       ++nval_blks_;
@@ -117,11 +148,12 @@ size_t LBFGSLearner::PrepareData() {
 
   // wait the previous push finished
   model_store_->Wait(t);
+  return nrows;
 }
 
 real_t LBFGSLearner::InitWorker() {
   // remove tail features
-  int filter = 0;  // TODO
+  int filter = GetUpdater()->param().tail_feature_filter;
   if (filter > 0) {
     SArray<real_t> feacnt;
     int t = model_store_->Pull(
@@ -166,11 +198,6 @@ void LBFGSLearner::LinearSearch(real_t alpha, std::vector<real_t>* status) {
   (*status)[1] = lbfgs::Inner(grads_, directions_, nthreads_);
 }
 
-// void LBFGSLearner::Evaluate(std::vector<real_t>* prog) {
-
-
-// }
-
 real_t LBFGSLearner::CalcGrad(const SArray<real_t>& w,
                               const SArray<int>& w_offset,
                               SArray<real_t>* grad) {
@@ -189,6 +216,7 @@ real_t LBFGSLearner::CalcGrad(const SArray<real_t>& w,
         data, {SArray<char>(pred_[i]), SArray<char>(pos), SArray<char>(w)}, grad);
     objv += loss_->CalcObjv(data.label, pred_[i]);
   }
+  return objv;
 }
 
 SArray<int> LBFGSLearner::GetPos(const SArray<int>& offset, const SArray<int>& colmap) {
@@ -200,4 +228,8 @@ SArray<int> LBFGSLearner::GetPos(const SArray<int>& offset, const SArray<int>& c
   return pos;
 }
 
+
+// void LBFGSLearner::Evaluate(std::vector<real_t>* prog) {
+// }
+
 }  // namespace difacto

src/lbfgs/lbfgs_learner.h

@@ -15,7 +15,12 @@ namespace difacto {
 
 class LBFGSLearner : public Learner {
  public:
-  virtual ~LBFGSLearner() { }
+  virtual ~LBFGSLearner() {
+    delete model_store_;
+    delete data_store_;
+    delete tile_store_;
+    delete loss_;
+  }
   KWArgs Init(const KWArgs& kwargs) override;
 
  protected:
@@ -26,18 +31,22 @@ class LBFGSLearner : public Learner {
   /**
    * \brief send jobs to nodes and wait them finished.
    */
-  void IssueJobAndWait(int node_group, const lbfgs::Job& job,
-                       Tracker::Monitor monitor = nullptr);
-
   void IssueJobAndWait(int node_group,
                        int job_type,
                        const std::vector<real_t>& job_args = {},
                        std::vector<real_t>* job_rets = nullptr);
 
+  /**
+   * \brief a wrapper to above
+   */
   void IssueJobAndWait(int node_group,
                        int job_type,
                        const std::vector<real_t>& job_args,
-                       real_t* job_rets);
+                       real_t* job_rets) {
+    std::vector<real_t> rets(1);
+    IssueJobAndWait(node_group, job_type, job_args, &rets);
+    *job_rets = rets[0];
+  }
 
   /**
    * \brief preprocessing the data
@@ -59,18 +68,10 @@ class LBFGSLearner : public Learner {
    */
   real_t InitWorker();
 
-  /**
-   * \brief init server
-   *
-   * load w if load_epoch is set. otherwise, initialize w
-   *
-   * @return number of model parameters
-   */
-  size_t InitServer() { }
-
   real_t CalcGrad(const SArray<real_t>& w,
                   const SArray<int>& w_offset,
                   SArray<real_t>* grad);
+
   void LinearSearch(real_t alpha, std::vector<real_t>* status);
 
   LBFGSUpdater* GetUpdater() {
@@ -79,8 +80,8 @@ class LBFGSLearner : public Learner {
   }
 
   SArray<int> GetPos(const SArray<int>& offset, const SArray<int>& colmap);
-  LBFGSLearnerParam param_;
 
+  LBFGSLearnerParam param_;
   int nthreads_ = DEFAULT_NTHREADS;
   SArray<feaid_t> feaids_;
   SArray<real_t> weights_, grads_, directions_;

src/lbfgs/lbfgs_param.h

@@ -40,6 +40,10 @@ struct LBFGSLearnerParam : public dmlc::Parameter<LBFGSLearnerParam> {
     DMLC_DECLARE_FIELD(model_in).set_default("");
     DMLC_DECLARE_FIELD(loss).set_default("fm");
     DMLC_DECLARE_FIELD(max_num_epochs).set_default(20);
+    DMLC_DECLARE_FIELD(alpha).set_default(1);
+    DMLC_DECLARE_FIELD(c1).set_default(1e-4);
+    DMLC_DECLARE_FIELD(c2).set_default(.9);
+    DMLC_DECLARE_FIELD(rho).set_default(.8);
   }
 };
 

src/lbfgs/lbfgs_twoloop.h

@@ -76,8 +76,8 @@ class Twoloop {
                      const std::vector<SArray<real_t>>& y,
                      const SArray<real_t>& grad,
                      SArray<real_t>* p) {
-    CHECK_EQ(s.size(), m_);
-    CHECK_EQ(y.size(), m_);
+    CHECK_EQ(s.size(), static_cast<size_t>(m_));
+    CHECK_EQ(y.size(), static_cast<size_t>(m_));
     size_t n = grad.size();
     p->resize(n); memset(p->data(), 0, n*sizeof(real_t));
 

src/lbfgs/lbfgs_updater.h

@@ -17,6 +17,7 @@ struct LBFGSUpdaterParam : public dmlc::Parameter<LBFGSUpdaterParam> {
     DMLC_DECLARE_FIELD(tail_feature_filter).set_default(4);
     DMLC_DECLARE_FIELD(l1).set_default(1);
     DMLC_DECLARE_FIELD(l2).set_default(.1);
+    DMLC_DECLARE_FIELD(m).set_default(10);
   }
 };
 
@@ -37,14 +38,13 @@ class LBFGSUpdater : public Updater {
 
   size_t InitWeights() {
     models_.resize(feaids_.size());
+    return models_.size();
   }
 
   void PrepareCalcDirection(real_t alpha, std::vector<real_t>* aux) {
+    if (static_cast<int>(s_.size()) > param_.m - 1) s_.erase(s_.begin());
     SArray<real_t> new_s(models_.size());
-    for (size_t i = 0; i < models_.size(); ++i) {
-      new_s[i] = alpha * models_[i];
-    }
-    if (s_.size() > param_.m - 1) s_.erase(s_.begin());
+    lbfgs::Add(alpha, models_, &new_s);
     s_.push_back(new_s);
     twoloop_.CalcIncreB(s_, y_, grads_, aux);
   }
@@ -66,11 +66,10 @@ class LBFGSUpdater : public Updater {
            SArray<real_t>* values,
            SArray<int>* offsets) override {
     if (value_type == Store::kFeaCount) {
-      values->resize(feaids.size());
-      KVMatch(feaids_, feacnts_, feaids, values);
+      KVMatch(feaids_, feacnts_, feaids, values, ASSIGN, nthreads_);
     } else if (value_type == Store::kWeight) {
-      CHECK(param_.V_dim == 0);  // TODO
-      KVMatch(feaids_, models_, feaids, values);
+      KVMatch(feaids_, models_, model_offsets_, feaids, values, offsets,
+              ASSIGN, nthreads_);
     } else {
       LOG(FATAL) << "...";
     }
@@ -81,18 +80,15 @@ class LBFGSUpdater : public Updater {
               const SArray<real_t>& values,
               const SArray<int>& offsets) override {
     if (value_type == Store::kFeaCount) {
-      feaids_ = feaids;
-      feacnts_ = values;
+      feaids_ = feaids; feacnts_ = values;
     } else if (value_type == Store::kGradient) {
       CHECK_EQ(feaids_.size(), feaids.size());
-      // update y
+      // add y = new_grad - old_grad
       if (grads_.size()) {
-        CHECK_EQ(grads_.size(), values.size());
-        SArray<real_t> new_y(grads_.size());
-        for (size_t i = 0; i < grads_.size(); ++i) {
-          new_y[i] = values[i] - grads_[i];
-        }
-        if (y_.size() > param_.m - 1) y_.erase(y_.begin());
+        if (static_cast<int>(y_.size()) == param_.m - 1) y_.erase(y_.begin());
+        SArray<real_t> new_y;
+        new_y.CopyFrom(values);
+        lbfgs::Add(-1, grads_, &new_y);
         y_.push_back(new_y);
       }
       grads_ = values;

src/lbfgs/lbfgs_utils.h

@@ -37,9 +37,9 @@ struct Job {
 /**
  * \brief return <a, b>
  */
-double Inner(const SArray<real_t>& a,
-             const SArray<real_t>& b,
-             int nthreads = DEFAULT_NTHREADS) {
+inline double Inner(const SArray<real_t>& a,
+                    const SArray<real_t>& b,
+                    int nthreads = DEFAULT_NTHREADS) {
   double res = 0;
   CHECK_EQ(a.size(), b.size());
   real_t const *ap = a.data();
@@ -52,9 +52,9 @@ double Inner(const SArray<real_t>& a,
 /**
  * \brief b += x * a
  */
-real_t Add(real_t x, const SArray<real_t>& a,
-           SArray<real_t>* b,
-           int nthreads = DEFAULT_NTHREADS) {
+inline void Add(real_t x, const SArray<real_t>& a,
+                SArray<real_t>* b,
+                int nthreads = DEFAULT_NTHREADS) {
   CHECK_EQ(a.size(), b->size());
   real_t const *ap = a.data();
   real_t *bp = b->data();

src/learner.cc

@@ -6,6 +6,7 @@
 #include "./sgd/sgd_learner.h"
 #include "./bcd/bcd_param.h"
 #include "./bcd/bcd_learner.h"
+#include "./lbfgs/lbfgs_learner.h"
 namespace difacto {
 
 DMLC_REGISTER_PARAMETER(SGDLearnerParam);
@@ -16,6 +17,8 @@ Learner* Learner::Create(const std::string& type) {
     return new SGDLearner();
   } else if (type == "bcd") {
     return new BCDLearner();
+  } else if (type == "lbfgs") {
+    return new LBFGSLearner();
   } else {
     LOG(FATAL) << "unknown learner type: " << type;
   }

tests/cpp/lbfgs_twoloop_test.cc

@@ -67,7 +67,7 @@ TEST(Twoloop, basic) {
   for (int k = 0; k < 10; ++k) {
     gen_vals(n, -1, 1, &g);
 
-    if (s.size() == m-1) {
+    if (static_cast<int>(s.size()) == m-1) {
       s.erase(s.begin());
       y.erase(y.begin());
     }