fixed_point_filter

Author: mli

Makefile

@@ -20,7 +20,7 @@ LDFLAGS += $(THIRD_LIB) -lpthread -lrt
 PS_LIB = build/libps.a
 PS_MAIN = build/libpsmain.a
 
-all: ps app build/hello
+all: ps app
 clean:
 	rm -rf build
 

example/linear/ctr/online_l1lr.conf

@@ -34,14 +34,14 @@ async_sgd {
 algo: FTRL
 
 # The size of minibatch
-minibatch : 10000
+minibatch : 100000
 
 # The number of data passes
-num_data_pass: 10
+num_data_pass: 6
 
 # features which occurs <= *tail_feature_freq* will be filtered before
 # training. it save both memory and bandwidth.
-tail_feature_freq : 4
+tail_feature_freq : 0
 
 # It controls the countmin size. We filter the tail features by countmin, which
 # is more efficient than hash, but still is the memory bottleneck for servers. A

src/app/linear_method/async_sgd.h

@@ -38,7 +38,10 @@ struct SGDState {
 
   void update() {
     if (reporter) {
-      SGDProgress prog; prog.set_nnz(nnz);
+      SGDProgress prog;
+      prog.set_nnz(nnz);
+      prog.set_weight_sum(weight_sum); weight_sum = 0;
+      prog.set_delta_sum(delta_sum); delta_sum = 0;
       reporter->report(prog);
     }
   }
@@ -49,13 +52,18 @@ struct SGDState {
     } else if (new_weight != 0 && old_weight == 0) {
       ++ nnz;
     }
+    weight_sum += new_weight * new_weight;
+    V delta = new_weight - old_weight;
+    delta_sum += delta * delta;
   }
 
   shared_ptr<LearningRate<V>> lr;
   shared_ptr<Penalty<V>> h;
 
   int iter = 0;
   size_t nnz = 0;
+  V weight_sum = 0;
+  V delta_sum = 0;
   V max_delta = 1.0;  // maximal change of weight
   MonitorSlaver<SGDProgress>* reporter = nullptr;
 };
@@ -213,7 +221,6 @@ class AsyncSGDWorker : public ISGDCompNode, public LinearMethod {
         data.add_file(call.data().file(idx[j]));
       }
     }
-
     reader.setReader(data, sgd.minibatch(), sgd.data_buf());
     reader.setFilter(sgd.countmin_n(), sgd.countmin_k(), sgd.tail_feature_freq());
     reader.start();
@@ -262,7 +269,6 @@ class AsyncSGDWorker : public ISGDCompNode, public LinearMethod {
     prog.set_num_examples_processed(
         prog.num_examples_processed() + Xw.size());
     this->reporter_.report(prog);
-    // LL << prog.objective(0) << " " << prog.auc(0);
 
     // compute the gradient
     SArray<V> grad(X->cols());
@@ -278,6 +284,13 @@ class AsyncSGDWorker : public ISGDCompNode, public LinearMethod {
     model_.clear(id);
 
     ++ processed_batch_;
+
+    // auto we = w.eigenArray();
+    // auto ge = grad.eigenArray();
+    // LL << we.minCoeff() << " " << we.maxCoeff() << " "
+    //    << w.mean() << " " << w.std() << " "
+    //    << ge.minCoeff() << " " << ge.maxCoeff() << " "
+    //    << grad.mean() << " " << grad.std();
   }
 
 private:

src/app/linear_method/loss.h

@@ -59,6 +59,7 @@ class ScalarLoss : public Loss<T> {
     if (gradient.size() != 0) CHECK_EQ(gradient.size(), X->cols());
     if (diag_hessian.size() != 0) CHECK_EQ(diag_hessian.size(), X->cols());
 
+    if (!y.size()) return;
     compute(y.eigenArray(), X, Xw.eigenArray(), gradient.eigenArray(), diag_hessian.eigenArray());
   }
 

src/app/linear_method/proto/linear.pb.h

@@ -40,7 +40,7 @@ message SGDConfig {
 
   optional int32 minibatch = 2 [default = 1000];
 
-  optional int32 data_buf = 12 [default = 100];  // in mb
+  optional int32 data_buf = 12 [default = 1000];  // in mb
   optional bool ada_grad = 5 [default = true];
 
   optional int32 max_delay = 4 [default = 0];

src/app/linear_method/proto/linear.proto

@@ -4,6 +4,7 @@
 namespace PS {
 
 class CompressingFilter : public Filter {
+ public:
   void encode(const MessagePtr& msg) {
     auto conf = find(FilterConfig::COMPRESSING, msg);
     if (!conf) return;

src/filter/compressing.h

@@ -0,0 +1,91 @@
+#pragma once
+#include "filter/filter.h"
+#include <time.h>
+namespace PS {
+
+class FixingFloatFilter : public Filter {
+ public:
+  void encode(const MessagePtr& msg) {
+    auto conf = CHECK_NOTNULL(find(FilterConfig::FIXING_FLOAT, msg))->fixed_point();
+    int n = msg->value.size();
+    CHECK_EQ(n, msg->task.value_type_size());
+    for (int i = 0; i < n; ++i) {
+      auto type = msg->task.value_type(i);
+      if (type == DataType::FLOAT) {
+        msg->value[i] = encode(SArray<float>(msg->value[i]), conf);
+      }
+      if (type == DataType::DOUBLE) {
+        msg->value[i] = encode(SArray<double>(msg->value[i]), conf);
+      }
+    }
+  }
+
+  void decode(const MessagePtr& msg) {
+    auto conf = CHECK_NOTNULL(find(FilterConfig::FIXING_FLOAT, msg))->fixed_point();
+    int n = msg->value.size();
+    CHECK_EQ(n, msg->task.value_type_size());
+    for (int i = 0; i < n; ++i) {
+      auto type = msg->task.value_type(i);
+      if (type == DataType::FLOAT) {
+        msg->value[i] = decode<float>(msg->value[i], conf);
+      }
+    }
+  }
+
+ private:
+  inline bool boolrand(int* seed) {
+    *seed = (214013 * *seed + 2531011);
+    return ((*seed >> 16) & 0x1) == 0;
+  }
+
+  template <typename V>
+  SArray<char> encode(const SArray<V>& data, const FilterConfig::FixedConfig& conf) {
+    int nbytes = conf.num_bytes();
+    CHECK_GT(nbytes, 0);
+    CHECK_LT(nbytes, 8);
+    V ratio = static_cast<V>(1<<(nbytes*4));
+    V min_v = static_cast<V>(conf.min_value());
+    V max_v = static_cast<V>(conf.max_value());
+    V bin = max_v - min_v;
+    CHECK_GT(bin, 0);
+
+    SArray<char> res(data.size() * nbytes);
+    char* res_ptr = res.data();
+    int seed = time(NULL);
+
+    for (int i = 0; i < data.size(); ++i) {
+      V proj = data[i] > max_v ? max_v : data[i] < min_v ? min_v : data[i];
+      V tmp = (proj - min_v) / bin * ratio;
+      uint64 r = static_cast<uint64>(floor(tmp)) + boolrand(&seed);
+
+      for (int j = 0; j < nbytes; ++j) {
+        *(res_ptr++) = static_cast<char>(r & 0xFF);
+        r = r >> 8;
+      }
+    }
+    return res;
+  }
+
+  template <typename V>
+  SArray<V> decode(const SArray<char>& data, const FilterConfig::FixedConfig& conf) {
+    int nbytes = conf.num_bytes();
+    V ratio = static_cast<V>(1<<(nbytes*4));
+    V min_v = static_cast<V>(conf.min_value());
+    V max_v = static_cast<V>(conf.max_value());
+    V bin = max_v - min_v;
+
+    int n = data.size() / nbytes;
+    SArray<V> res(n);
+    char* data_ptr = data.data();
+    for (int i = 0; i < n; ++i) {
+      V r = 0;
+      for (int j = 0; j < nbytes; ++j) {
+        r += static_cast<uint64>(*(data_ptr++)) << 8 * j;
+      }
+      res[i] = static_cast<V>(r) / ratio * bin + min_v;
+    }
+    return res;
+  }
+};
+
+} // namespace PS

src/filter/fixing_float.h

@@ -4,6 +4,7 @@
 namespace PS {
 
 class KeyCachingFilter : public Filter {
+ public:
   // thread safe
   void encode(const MessagePtr& msg) {
     // if (!msg->task.has_key_range()) return;