PaddlePaddle
diff --git a/‎paddle/fluid/distributed/ps/service/ps_client.h
Copy file name to clipboard
+8Lines changed: 8 additions & 0 deletions b/‎paddle/fluid/distributed/ps/service/ps_client.h
Copy file name to clipboard
+8Lines changed: 8 additions & 0 deletions
diff --git a/‎paddle/fluid/distributed/ps/service/ps_local_client.cc
Copy file name to clipboard
+7Lines changed: 7 additions & 0 deletions b/‎paddle/fluid/distributed/ps/service/ps_local_client.cc
Copy file name to clipboard
+7Lines changed: 7 additions & 0 deletions
diff --git a/‎paddle/fluid/distributed/ps/service/ps_local_client.h
Copy file name to clipboard
+2Lines changed: 2 additions & 0 deletions b/‎paddle/fluid/distributed/ps/service/ps_local_client.h
Copy file name to clipboard
+2Lines changed: 2 additions & 0 deletions
diff --git a/‎paddle/fluid/distributed/ps/table/accessor.h
Copy file name to clipboard
+4Lines changed: 4 additions & 0 deletions b/‎paddle/fluid/distributed/ps/table/accessor.h
Copy file name to clipboard
+4Lines changed: 4 additions & 0 deletions
diff --git a/‎paddle/fluid/distributed/ps/table/ctr_dymf_accessor.cc
Copy file name to clipboard
+81-2Lines changed: 81 additions & 2 deletions b/‎paddle/fluid/distributed/ps/table/ctr_dymf_accessor.cc
Copy file name to clipboard
+81-2Lines changed: 81 additions & 2 deletions
diff --git a/‎paddle/fluid/distributed/ps/table/ctr_dymf_accessor.h
Copy file name to clipboard
+15-3Lines changed: 15 additions & 3 deletions b/‎paddle/fluid/distributed/ps/table/ctr_dymf_accessor.h
Copy file name to clipboard
+15-3Lines changed: 15 additions & 3 deletions
diff --git a/‎paddle/fluid/distributed/ps/table/ssd_sparse_table.cc
Copy file name to clipboard
+24-6Lines changed: 24 additions & 6 deletions b/‎paddle/fluid/distributed/ps/table/ssd_sparse_table.cc
Copy file name to clipboard
+24-6Lines changed: 24 additions & 6 deletions
diff --git a/‎paddle/fluid/distributed/ps/table/ssd_sparse_table.h
Copy file name to clipboard
+3Lines changed: 3 additions & 0 deletions b/‎paddle/fluid/distributed/ps/table/ssd_sparse_table.h
Copy file name to clipboard
+3Lines changed: 3 additions & 0 deletions
diff --git a/‎paddle/fluid/distributed/ps/table/table.h
Copy file name to clipboard
+2Lines changed: 2 additions & 0 deletions b/‎paddle/fluid/distributed/ps/table/table.h
Copy file name to clipboard
+2Lines changed: 2 additions & 0 deletions
@@ -359,6 +359,14 @@ class PSClient {
     return nullptr;
   }
 
+  virtual ::std::future<int32_t> SetDayId(size_t table_id, int day_id) {
+    VLOG(0) << "SetDayId Did not implement";
+    std::promise<int32_t> promise;
+    std::future<int> fut = promise.get_future();
+    promise.set_value(-1);
+    return fut;
+  }
+
  protected:
   virtual int32_t Initialize() = 0;
   PSParameter _config;
 
@@ -322,5 +322,12 @@ ::std::future<int32_t> PsLocalClient::PushSparse(size_t table_id,
   table_ptr->Push(table_context);
   return done();
 }
+
+::std::future<int32_t> PsLocalClient::SetDayId(size_t table_id, int day_id) {
+  auto* table_ptr = GetTable(table_id);
+  table_ptr->SetDayId(day_id);
+  return done();
+}
+
 }  // namespace distributed
 }  // namespace paddle
@@ -203,6 +203,8 @@ class PsLocalClient : public PSClient {
     return fut;
   }
 
+  virtual ::std::future<int32_t> SetDayId(size_t table_id, int day_id);
+
  protected:
   virtual int32_t Initialize();
 
 
@@ -187,6 +187,10 @@ class ValueAccessor {
   virtual robin_hood::unordered_set<float>* GetSaveFilteredSlots() {
     return nullptr;
   }
+
+  virtual void SetDayId(int day_id) {}
+  virtual void UpdateTimeDecay(float* value, bool is_update_seen_day) {}
+
 #define DEFINE_GET_INDEX(class, field) \
   virtual int get_##field##_index() { return class ::field##_index(); }
 
 
@@ -60,6 +60,7 @@ int CtrDymfAccessor::Initialize() {
           << " embedx_dim:" << common_feature_value.embedx_dim
           << "  embedx_sgd_dim:" << common_feature_value.embedx_sgd_dim;
   InitAccessorInfo();
+  SetTimeDecayRates();
   return 0;
 }
 
@@ -77,6 +78,16 @@ void CtrDymfAccessor::InitAccessorInfo() {
       (embedx_dim + common_feature_value.embedx_sgd_dim) * sizeof(float);
 }
 
+void CtrDymfAccessor::SetTimeDecayRates() {
+  // 根据unseen_days的天数来初始化_time_decay_rates大小和对应的衰减率
+  auto delete_after_unseen_days =
+      _config.ctr_accessor_param().delete_after_unseen_days();
+  _time_decay_rates.assign(delete_after_unseen_days + 2, 0.0);
+  for (int i = 0; i <= delete_after_unseen_days + 1; ++i) {
+    _time_decay_rates[i] = pow(_show_click_decay_rate, i);
+  }
+}
+
 bool CtrDymfAccessor::Shrink(float* value) {
   auto delete_after_unseen_days =
       _config.ctr_accessor_param().delete_after_unseen_days();
@@ -110,7 +121,19 @@ bool CtrDymfAccessor::SaveCache(float* value,
 }
 
 bool CtrDymfAccessor::SaveSSD(float* value) {
-  if (common_feature_value.UnseenDays(value) > _ssd_unseenday_threshold) {
+  if (_day_id == 0) {
+    return true;
+  }
+  auto unseen_days = common_feature_value.UnseenDays(value);
+  if (unseen_days == 0) {
+    return true;
+  }
+  // for the origin load (eg. unseen_days = 0-15)
+  if (unseen_days < _config.ctr_accessor_param().delta_keep_days()) {
+    unseen_days = _day_id - unseen_days;
+  }
+  int16_t day_diff = _day_id - unseen_days;
+  if (day_diff > _ssd_unseenday_threshold) {
     return true;
   }
   return false;
@@ -207,8 +230,12 @@ void CtrDymfAccessor::UpdateStatAfterSave(float* value, int param) {
 int32_t CtrDymfAccessor::Create(float** values, size_t num) {
   for (size_t value_item = 0; value_item < num; ++value_item) {
     float* value = values[value_item];
+#ifdef PADDLE_WITH_PSLIB
     common_feature_value.UnseenDays(value) = 0;
     common_feature_value.PassId(value) = 0;
+#else
+    value[common_feature_value.UnseenDaysIndex()] = 0;
+#endif
     value[common_feature_value.DeltaScoreIndex()] = 0;
     value[common_feature_value.ShowIndex()] = 0;
     value[common_feature_value.ClickIndex()] = 0;
@@ -217,7 +244,12 @@ int32_t CtrDymfAccessor::Create(float** values, size_t num) {
     _embed_sgd_rule->InitValue(
         value + common_feature_value.EmbedWIndex(),
         value + common_feature_value.EmbedG2SumIndex(),
-        false);  // adam embed init not zero, adagrad embed init zero
+#ifdef PADDLE_WITH_GPU_GRAPH
+        false);  // adam embed init not zero, adagrad embed init zero; gpubox
+                 // set true for adam
+#else
+        true);  //  gpups use adagrad thus set false
+#endif
     _embedx_sgd_rule->InitValue(value + common_feature_value.EmbedxWIndex(),
                                 value + common_feature_value.EmbedxG2SumIndex(),
                                 false);
@@ -323,8 +355,12 @@ std::string CtrDymfAccessor::ParseToString(const float* v, int param) {
   thread_local std::ostringstream os;
   os.clear();
   os.str("");
+#ifdef PADDLE_WITH_PSLIB
   os << common_feature_value.UnseenDays(const_cast<float*>(v)) << " " << v[1]
      << " " << v[2] << " " << v[3] << " " << v[4];
+#else
+  os << v[0] << " " << v[1] << " " << v[2] << " " << v[3] << " " << v[4];
+#endif
   //    << v[5] << " " << v[6];
   for (int i = common_feature_value.EmbedG2SumIndex();
        i < common_feature_value.EmbedxG2SumIndex();
@@ -349,13 +385,55 @@ std::string CtrDymfAccessor::ParseToString(const float* v, int param) {
 
 int CtrDymfAccessor::ParseFromString(const std::string& str, float* value) {
   auto ret = paddle::string::str_to_float(str.data(), value);
+#ifdef PADDLE_WITH_PSLIB
   float unseen_day = value[common_feature_value.UnseenDaysIndex()];
   common_feature_value.UnseenDays(value) = (uint16_t)(unseen_day);
   common_feature_value.PassId(value) = 0;
+#endif
   CHECK(ret >= 7) << "expect more than 7 real:" << ret;
   return ret;
 }
 
+void CtrDymfAccessor::SetDayId(int day_id) { _day_id = day_id; }
+
+void CtrDymfAccessor::UpdateTimeDecay(float* value, bool is_update_seen_day) {
+  // 根据day_id 来进行show click 衰减和unseen_day 更新;unseen_day
+  // 为上次出现的dayid
+  if (_day_id == 0) {
+    return;
+  }
+  auto unseen_days = common_feature_value.UnseenDays(value);
+  if (unseen_days == 0) {
+    common_feature_value.UnseenDays(value) = _day_id;
+    return;
+  }
+  // for the origin load (unseenday = 0 -15)
+  if (unseen_days < _config.ctr_accessor_param().delete_after_unseen_days()) {
+    // pull
+    if (is_update_seen_day) {
+      common_feature_value.UnseenDays(value) = _day_id;
+      return;
+      // save 舍弃原始的unseenday,都变为上一天出现,保证show/click不被重复decay
+    } else {
+      common_feature_value.UnseenDays(value) = _day_id - 1;
+    }
+  }
+  int16_t day_diff = _day_id - unseen_days;
+  if (day_diff < 0) {
+    common_feature_value.UnseenDays(value) = _day_id;
+    return;
+  }
+  if (day_diff >= _config.ctr_accessor_param().delete_after_unseen_days()) {
+    return;
+  }
+  common_feature_value.Show(value) *= _time_decay_rates[day_diff];
+  common_feature_value.Click(value) *= _time_decay_rates[day_diff];
+  if (is_update_seen_day) {
+    common_feature_value.UnseenDays(value) = _day_id;
+  }
+}
+
+#ifdef PADDLE_WITH_PSLIB
 bool CtrDymfAccessor::SaveMemCache(float* value,
                                    int param,
                                    double global_cache_threshold,
@@ -367,6 +445,7 @@ bool CtrDymfAccessor::SaveMemCache(float* value,
 void CtrDymfAccessor::UpdatePassId(float* value, uint16_t pass_id) {
   common_feature_value.PassId(value) = pass_id;
 }
+#endif
 
 }  // namespace distributed
 }  // namespace paddle
@@ -45,9 +45,9 @@ class CtrDymfAccessor : public ValueAccessor {
       // float embedx_g2sum;
       std::vector<float> embedx_w;
        */
-    /* V2: support pass_id
-      uint16_t pass_id;
-      uint16_t unseen_days;
+    /* V2: support pass_id (pslib gpups)
+      uint16_t pass_id; (diff)
+      uint16_t unseen_days; (diff)
       float show;
       float click;
       float embed_w;
@@ -88,6 +88,7 @@ class CtrDymfAccessor : public ValueAccessor {
     // 根据mf_dim计算的总byte数
     int Size(int mf_dim) { return (Dim(mf_dim)) * sizeof(float); }
 
+#ifdef PADDLE_WITH_PSLIB
     uint16_t& PassId(float* val) {
       uint16_t* int16_val =
           reinterpret_cast<uint16_t*>(val + UnseenDaysIndex());
@@ -99,6 +100,9 @@ class CtrDymfAccessor : public ValueAccessor {
           reinterpret_cast<uint16_t*>(val + UnseenDaysIndex());
       return int16_val[1];
     }
+#else
+    float& UnseenDays(float* val) { return val[UnseenDaysIndex()]; }
+#endif
     float& DeltaScore(float* val) { return val[DeltaScoreIndex()]; }
     float& Show(float* val) { return val[ShowIndex()]; }
     float& Click(float* val) { return val[ClickIndex()]; }
@@ -252,15 +256,21 @@ class CtrDymfAccessor : public ValueAccessor {
     return &_save_filtered_slots;
   }
 
+  void SetDayId(int day_id) override;
+
+#ifdef PADDLE_WITH_PSLIB
   // 根据pass_id和show_threshold阈值来判断cache到ssd
   bool SaveMemCache(float* value,
                     int param,
                     double global_cache_threshold,
                     uint16_t pass_id);
   // 更新pass_id
   void UpdatePassId(float* value, uint16_t pass_id);
+#endif
+  void UpdateTimeDecay(float* value, bool is_update_seen_day);
 
  private:
+  void SetTimeDecayRates();
   // float ShowClickScore(float show, float click);
 
   // SparseValueSGDRule* _embed_sgd_rule;
@@ -269,6 +279,8 @@ class CtrDymfAccessor : public ValueAccessor {
   float _show_click_decay_rate;
   int32_t _ssd_unseenday_threshold;
   bool _show_scale = false;
+  int _day_id = 0;
+  std::vector<double> _time_decay_rates;
 
  public:  // TODO(zhaocaibei123): it should be private, but we make it public
           // for unit test
 
@@ -44,6 +44,8 @@ int32_t SSDSparseTable::Initialize() {
 
 int32_t SSDSparseTable::InitializeShard() { return 0; }
 
+void SSDSparseTable::SetDayId(int day_id) { _day_id = day_id; }
+
 int32_t SSDSparseTable::Pull(TableContext& context) {
   CHECK(context.value_type == Sparse);
   if (context.use_ptr) {
@@ -246,7 +248,11 @@ int32_t SSDSparseTable::PullSparsePtr(int shard_id,
                               sizeof(uint64_t));
                 ret = &feature_value;
               }
+
+              _value_accessor->UpdateTimeDecay(ret->data(), true);
+#ifdef PADDLE_WITH_PSLIB
               _value_accessor->UpdatePassId(ret->data(), pass_id);
+#endif
               int pull_data_idx = cur_ctx->batch_index[idx];
               pull_values[pull_data_idx] = reinterpret_cast<char*>(ret);
             }
@@ -258,7 +264,10 @@ int32_t SSDSparseTable::PullSparsePtr(int shard_id,
       } else {
         ret = itr.value_ptr();
         // int pull_data_idx = keys[i].second;
+        _value_accessor->UpdateTimeDecay(ret->data(), true);
+#ifdef PADDLE_WITH_PSLIB
         _value_accessor->UpdatePassId(ret->data(), pass_id);
+#endif
         pull_values[i] = reinterpret_cast<char*>(ret);
       }
     }
@@ -307,7 +316,10 @@ int32_t SSDSparseTable::PullSparsePtr(int shard_id,
               shard_id, reinterpret_cast<char*>(&cur_key), sizeof(uint64_t));
           ret = &feature_value;
         }
+        _value_accessor->UpdateTimeDecay(ret->data(), true);
+#ifdef PADDLE_WITH_PSLIB
         _value_accessor->UpdatePassId(ret->data(), pass_id);
+#endif
         int pull_data_idx = cur_ctx->batch_index[idx];
         pull_values[pull_data_idx] = reinterpret_cast<char*>(ret);
       }
@@ -2509,17 +2521,19 @@ int32_t SSDSparseTable::Load(const std::string& path,
   }
 
   int load_param = atoi(param.c_str());
-  size_t expect_shard_num = _sparse_table_shard_num;
-  if (file_list.size() != expect_shard_num) {
-    LOG(WARNING) << "SSDSparseTable file_size:" << file_list.size()
-                 << " not equal to expect_shard_num:" << expect_shard_num;
-    return -1;
-  }
   if (file_list.empty()) {
     LOG(WARNING) << "SSDSparseTable load file is empty, path:" << path;
     return -1;
   }
+  _value_accessor->SetDayId(_day_id);
+  VLOG(1) << " Load Set Dayid:" << _day_id;
   if (load_param > 3) {
+    size_t expect_shard_num = _sparse_table_shard_num;
+    if (file_list.size() != expect_shard_num) {
+      LOG(WARNING) << "SSDSparseTable file_size:" << file_list.size()
+                   << " not equal to expect_shard_num:" << expect_shard_num;
+      return -1;
+    }
     size_t file_start_idx = _shard_idx * _avg_local_shard_num;
     return LoadWithString(file_start_idx,
                           file_start_idx + _real_local_shard_num,
@@ -2833,7 +2847,9 @@ int32_t SSDSparseTable::LoadWithBinary(const std::string& path, int param) {
                     abort();
                   }
                   last_k = k;
+#ifdef PADDLE_WITH_PSLIB
                   _value_accessor->UpdatePassId(convert_value, 0);
+#endif
                   rocksdb::Status status = sst_writer.Put(
                       rocksdb::Slice(reinterpret_cast<char*>(&k),
                                      sizeof(uint64_t)),
@@ -2849,7 +2865,9 @@ int32_t SSDSparseTable::LoadWithBinary(const std::string& path, int param) {
                   }
                 } else {
                   auto& feature_value = shard[k];
+#ifdef PADDLE_WITH_PSLIB
                   _value_accessor->UpdatePassId(convert_value, 0);
+#endif
                   feature_value.resize(dim);
                   memcpy(const_cast<float*>(feature_value.data()),
                          convert_value,
 
@@ -128,12 +128,15 @@ class SSDSparseTable : public MemorySparseTable {
 
   int32_t CacheTable(uint16_t pass_id) override;
 
+  void SetDayId(int day_id) override;
+
  private:
   RocksDBHandler* _db;
   int64_t _cache_tk_size;
   double _local_show_threshold{0.0};
   std::vector<paddle::framework::Channel<std::string>> _fs_channel;
   std::mutex _table_mutex;
+  int _day_id = 0;
 };
 
 }  // namespace distributed
 
@@ -163,6 +163,8 @@ class Table {
   virtual void Revert() {}
   virtual void CheckSavePrePatchDone() {}
 
+  virtual void SetDayId(int day_id) {}
+
  protected:
   virtual int32_t Initialize() = 0;
   virtual int32_t InitializeAccessor();
Original file line number	Diff line number	Diff line change
`@@ -203,6 +203,8 @@ class PsLocalClient : public PSClient {`
`203`	`203`	`return fut;`
`204`	`204`	`}`
`205`	`205`
	`206`	`+ virtual ::std::future<int32_t> SetDayId(size_t table_id, int day_id);`
	`207`	`+`
`206`	`208`	`protected:`
`207`	`209`	`virtual int32_t Initialize();`
`208`	`210`