freq_filter at kv_vector

Author: mli

src/app/linear_method/async_sgd.h

@@ -220,11 +220,7 @@ class AsyncSGDWorker : public ISGDCompNode {
               << data.second->rows() << "-by-" << data.second->cols();
 
       // pull the weight
-      auto req = Parameter::Request(id);
-      // TODO
-      // for (int i = 0; i < conf_.pull_filter_size(); ++i) {
-      //   *req.add_filter() = conf_.pull_filter(i);
-      // }
+      auto req = Parameter::Request(id, -1, {}, sgd.pull_filter());
       model_.Pull(req, key, [this, id]() { ComputeGradient(id); });
     }
 
@@ -263,16 +259,7 @@ class AsyncSGDWorker : public ISGDCompNode {
     loss_->compute({Y, X, Xw.SMatrix()}, {grad.SMatrix()});
 
     // push the gradient
-    auto req = Parameter::Request(id);
-    // TODO
-    // for (int i = 0; i < conf_.push_filter_size(); ++i) {
-    //   // add filters
-    //   auto filter = conf_.push_filter(i);
-    //   if (filter.type() == FilterConfig::KEY_CACHING) {
-    //     filter.set_clear_cache_if_done(true);
-    //   }
-    //   *req.add_filter() = filter;
-    // }
+    auto req = Parameter::Request(id, -1, {}, conf_.async_sgd().push_filter());
     model_.Push(req, model_[id].key, grad);
     model_.clear(id);
 

src/app/linear_method/darlin.h

@@ -285,7 +285,6 @@ class DarlinWorker : public BCDWorker<Real>, public DarlinCompNode {
 
   virtual void PreprocessData(int time, Message* request) {
     BCDWorker<Real>::PreprocessData(time, request);
-    const BCDCall& call = request->task.bcd();
     // dual_ = exp(y.*(X_*w_))
     if (bcd_conf_.init_w().type() == ParamInitConfig::ZERO) {
       dual_.SetValue(1);  // an optimizatoin

src/app/linear_method/proto/linear.proto

@@ -19,8 +19,6 @@ message Config {
   optional SGDConfig async_sgd = 17;
   optional BCDConfig darlin = 15;
 
-  repeated FilterConfig push_filter = 13;
-  repeated FilterConfig pull_filter = 14;
 }
 
 extend BCDConfig {
@@ -67,6 +65,9 @@ message SGDConfig {
   // filtered feature.
   optional float countmin_n = 8 [default = 1e8];
   optional int32 countmin_k = 7 [default = 2];
+
+  repeated FilterConfig push_filter = 13;
+  repeated FilterConfig pull_filter = 14;
 }
 
 message LossConfig {

src/learner/bcd.cc

@@ -32,22 +32,22 @@ void BCDScheduler::ProcessResponse(Message* response) {
 }
 
 void BCDScheduler::Run() {
-  WaitWorkersReady();
   LoadData();
   PreprocesseData();
   DivideFeatureBlocks();
 }
 
 void BCDScheduler::LoadData() {
   // wait workers have load the data
+  WaitWorkersReady();
   auto load_time = tic();
   int n = sys_.manager().num_workers();
   while (load_data_ < n) usleep(500);
   if (hit_cache_ > 0) {
     CHECK_EQ(hit_cache_, n) << "clear the local caches";
     NOTICE("Hit local caches for the training data");
   }
-  NOTICE ("Loaded %lld examples in %g sec", g_train_info_.num_ex(), toc(load_time));
+  NOTICE ("Loaded %lu examples in %g sec", g_train_info_.num_ex(), toc(load_time));
 }
 
 void BCDScheduler::PreprocesseData() {

src/parameter/kv_layer.h

@@ -77,7 +77,7 @@ class  KVLayer : public Parameter {
   virtual void Slice(const Message& request, const std::vector<Range<Key>>& krs,
                      std::vector<Message*>* msgs);
   virtual void GetValue(Message* msg);
-  virtual void SetValue(Message* msg);
+  virtual void SetValue(const Message* msg);
  protected:
   std::mutex mu_;
   std::unordered_map<int, SArray<V>> layer_;
@@ -94,7 +94,7 @@ int KVLayer<V, Updater>::Push(const Task& task, V* data, size_t size, bool zero_
     val.CopyFrom(data, size);
   }
   Message push(task, kServerGroup);
-  Range<Key>(0, size).to(push.task.mutable_key_range());
+  Range<Key>(0, size).To(push.task.mutable_key_range());
   push.add_value(val);
   return Parameter::Push(&push);
 }
@@ -109,7 +109,7 @@ int KVLayer<V, Updater>::Pull(
     layer_[id] = SArray<V>(data, size, false);
   }
   Message pull(task, kServerGroup);
-  Range<Key>(0, size).to(pull.task.mutable_key_range());
+  Range<Key>(0, size).To(pull.task.mutable_key_range());
   if (callback) pull.callback = callback;
   return Parameter::Pull(&pull);
 }
@@ -129,14 +129,14 @@ void KVLayer<V, Updater>::Slice(
       // a tiny layer, sent it to server k
       int k = (key * 991) % n;
       if (i == k) {
-        kr.to(mut_kr);
+        kr.To(mut_kr);
       } else {
-        Range<Key>(0,0).to(mut_kr);
+        Range<Key>(0,0).To(mut_kr);
         msg->valid = false;  // invalid msg will not be sent
       }
     } else {
       // evenly parititon the data into all server nodes
-      kr.evenDivide(n, i).to(mut_kr);
+      kr.EvenDivide(n, i).To(mut_kr);
     }
   }
 
@@ -176,7 +176,7 @@ void KVLayer<V, Updater>::GetValue(Message* msg) {
 }
 
 template <typename V, class Updater>
-void KVLayer<V, Updater>::SetValue(Message* msg) {
+void KVLayer<V, Updater>::SetValue(const Message* msg) {
   // Lock l(mu_);
   CHECK_EQ(msg->value.size(), 1);
   SArray<V> recv_data(msg->value[0]);

src/parameter/kv_map.h

@@ -54,7 +54,7 @@ class KVMap : public Parameter {
   }
 
   virtual void GetValue(Message* msg);
-  virtual void SetValue(Message* msg);
+  virtual void SetValue(const Message* msg);
 
  protected:
   int k_;
@@ -75,7 +75,7 @@ void KVMap<K,V,E,S>::GetValue(Message* msg) {
 }
 
 template <typename K, typename V, typename E, typename S>
-void KVMap<K,V,E,S>::SetValue(Message* msg) {
+void KVMap<K,V,E,S>::SetValue(const Message* msg) {
   SArray<K> key(msg->key);
   size_t n = key.size();
   CHECK_EQ(msg->value.size(), 1);

src/parameter/kv_vector.h

@@ -2,6 +2,7 @@
 #include "ps.h"
 #include "parameter/parameter.h"
 #include "util/parallel_ordered_match.h"
+#include "filter/frequency_filter.h"
 namespace PS {
 // TODO doc, and filter
 /**
@@ -92,7 +93,7 @@ class KVVector : public Parameter {
     SliceKOFVMessage<K>(request, krs, msgs);
   }
   virtual void GetValue(Message* msg);
-  virtual void SetValue(Message* msg);
+  virtual void SetValue(const Message* msg);
   using Parameter::Push;
   using Parameter::Pull;
  protected:
@@ -103,13 +104,31 @@ class KVVector : public Parameter {
   std::unordered_map<int, Buffer> buffer_;  // <channel, Buffer>
 
   std::mutex mu_;  // protect the structure of data_ and buffer_
+
+  // filter tail keys
+  FreqencyFilter<Key, uint8> freq_filter_;
 };
 
 template <typename K, typename V>
-void KVVector<K,V>::SetValue(Message* msg) {
+void KVVector<K,V>::SetValue(const Message* msg) {
   // do check
   SArray<K> recv_key(msg->key);
   if (recv_key.empty()) return;
+
+  // filter request
+  if (msg->task.param().has_tail_filter() && msg->task.request()) {
+    const auto& tail = msg->task.param().tail_filter();
+    CHECK(tail.insert_count());
+    CHECK_EQ(msg->value.size(), 1);
+    SArray<uint8> count(msg->value[0]);
+    CHECK_EQ(count.size(), recv_key.size());
+    if (freq_filter_.Empty()) {
+      freq_filter_.Resize(tail.countmin_n(), tail.countmin_k());
+    }
+    freq_filter_.InsertKeys(recv_key, count);
+    return;
+  }
+
   int chl = msg->task.key_channel();
   mu_.lock();
   auto& kv = data_[chl];
@@ -167,6 +186,15 @@ void KVVector<K,V>::GetValue(Message* msg) {
   // do check
   SArray<K> recv_key(msg->key);
   if (recv_key.empty()) return;
+
+  // filter request
+  if (msg->task.param().has_tail_filter()) {
+    const auto& tail = msg->task.param().tail_filter();
+    CHECK(tail.has_freq_threshold());
+    msg->key = freq_filter_.QueryKeys(recv_key, tail.freq_threshold());
+    return;
+  }
+
   Lock l(mu_);
   auto& kv = data_[msg->task.key_channel()];
   CHECK_EQ(kv.key.size() * k_, kv.value.size());
@@ -195,9 +223,7 @@ template <typename K, typename V>
 int KVVector<K,V>::Pull(const Task& request, const SArray<K>& keys,
                         Message::Callback callback) {
   Lock l(mu_);
-  int chl = request.key_channel();
-  if (keys.empty() ) CHECK_EQ(data_.count(chl), 1) << "empty channel " << chl;
-  auto& kv = data_[chl];
+  auto& kv = data_[request.key_channel()];
   if (!keys.empty()) kv.key = keys;
   kv.value = SArray<V>(kv.key.size()*k_, 0);
 

src/parameter/parameter.h

@@ -11,8 +11,6 @@ class Parameter : public Customer {
 
   typedef std::initializer_list<int> Timestamps;
   typedef ::google::protobuf::RepeatedPtrField<FilterConfig> Filters;
-  // typedef std::initializer_list<FilterConfig> Filters;
-
   /**
    * @brief Creats a request task
    *
@@ -60,12 +58,12 @@ class Parameter : public Customer {
 
   /// @brief Set the values in "msg" into into my data strcuture, e.g..
   ///  my_val_[msg->key[0]] = msg->value(0)[0];
-  virtual void SetValue(Message* msg) = 0;
+  virtual void SetValue(const Message* msg) = 0;
 
   /// @brief the message contains the backup KV pairs sent by the master node of the key
   /// segment to its replica node. merge these pairs into my replica, say
   /// replica_[msg->sender] = ...
-  virtual void SetReplica(Message* msg) { }
+  virtual void SetReplica(const Message* msg) { }
 
   /// @brief retrieve the replica. a new server node replacing a dead server will first
   /// ask for the dead's replica node for the data

src/parameter/proto/param.proto

@@ -10,8 +10,8 @@ message ParamCall {
 
   optional TailKeyFilter tail_filter = 3;
 
-  optional bool insert_key = 5;
-  optional bool gather = 6;
+  // optional bool insert_key = 5;
+  // optional bool gather = 6;
 
   // it's a replica request
   optional bool replica = 10;
@@ -43,6 +43,6 @@ message TailKeyFilter {
   optional bool insert_count = 1;
   optional int32 freq_threshold = 2;
   optional bool query_value = 3;
-  optional int32 countmin_n = 4;
-  optional int32 countmin_k = 5;
+  optional int32 countmin_n = 4 [default = 1000000];
+  optional int32 countmin_k = 5 [default = 2];
 }

src/util/file.cc

@@ -93,7 +93,7 @@ File* File::open(const DataConfig& name,  const char* const flag) {
     auto f = new File(des, filename);
     return f;
   }
-#if USE_S3 
+#if USE_S3
   else if (s3file(filename)) {
     std::string cmd = "curl -s -X GET "+s3FileUrl(filename);
     // .gz
@@ -105,7 +105,7 @@ File* File::open(const DataConfig& name,  const char* const flag) {
     auto f = new File(des, filename);
     return f;
   }
-#endif // USE_S3 
+#endif // USE_S3
   else {
     return open(filename, flag);
   }
@@ -314,10 +314,10 @@ std::vector<std::string> s3GetFileNamesFromXml(const char* fbuf, int fsize, cons
           xmlFree(str);
         }
       }
-      xmlXPathFreeObject(xpathObj);  
+      xmlXPathFreeObject(xpathObj);
     }
-    xmlXPathFreeContext(xpathCtx); 
-    xmlFreeDoc(doc); 
+    xmlXPathFreeContext(xpathCtx);
+    xmlFreeDoc(doc);
     return files;
 }
 #endif // USE_S3
@@ -343,9 +343,7 @@ std::vector<std::string> readFilenamesInDirectory(const DataConfig& directory) {
     std::vector<std::string> files;
     string cmd = hadoopFS(directory.hdfs()) + " -ls " + dirname;
 
-    if (FLAGS_verbose) {
-      LI << "readFilenamesInDirectory hdfs ls [" << cmd << "]";
-    }
+    VLOG(1) << "readFilenamesInDirectory hdfs ls [" << cmd << "]";
 
     FILE* des = popen(cmd.c_str(), "r"); CHECK(des);
     char line[10000];
@@ -365,7 +363,7 @@ std::vector<std::string> readFilenamesInDirectory(const DataConfig& directory) {
     pclose(des);
     return files;
   }
-#if USE_S3 
+#if USE_S3
   else if (s3file(dirname)) {
     // open xml
     std::string cmd = "curl -s -X GET "+s3DirectoryUrl(dirname);
@@ -387,7 +385,7 @@ std::vector<std::string> readFilenamesInDirectory(const DataConfig& directory) {
     free (fbuf);
     return files;
   }
-#endif // USE_S3 
+#endif // USE_S3
   else {
     return readFilenamesInDirectory(dirname);
   }