Class BCDWorker

src/learner/bcd.h:277–518

template <typename V>
class BCDWorker : public App, public BCDCommon {
 public:
  BCDWorker(const string& name, const BCDConfig& conf)
      : App(name), BCDCommon(conf), model_(name+"_model", name) { }
  virtual ~BCDWorker() { }

  void process(const MessagePtr& msg) {
    CHECK(msg->task.has_bcd());
    auto bcd = msg->task.bcd();
    int time  = msg->task.time() * time_ratio_;
    switch (bcd.cmd()) {
      case BCDCall::LOAD_DATA: {
        LoadDataReturn ret;
        int hit_cache = 0;
        CHECK(bcd.has_data());
        loadData(bcd.data(), ret.mutable_example_info(), &hit_cache);
        ret.set_hit_cache(hit_cache);
        sys_.replyProtocalMessage(msg, ret);
        break;
      }
      case BCDCall::PREPROCESS_DATA:
        preprocessData(time, bcd);
        break;
      case BCDCall::UPDATE_MODEL:
        computeGradient(time, bcd, msg);
        msg->finished = false;  //
        break;
      case BCDCall::EVALUATE_PROGRESS: {
        BCDProgress prog; evaluate(&prog);
        sys_.replyProtocalMessage(msg, prog);
        break;
      }
      default: break;
    }
  }
 protected:
  virtual void computeGradient(int time, const BCDCall& bcd, MessagePtr msg) = 0;
  virtual void evaluate(BCDProgress* prog) = 0;

  void loadData(const DataConfig& data, ExampleInfo* info, int *hit_cache) {
    *hit_cache = dataCache("train", true);
    if (!(*hit_cache)) {
      slot_reader_.init(data, bcd_conf_.local_cache());
      slot_reader_.read(info);
    }
  }

  virtual void preprocessData(int time, const BCDCall& call) {
    int grp_size = call.fea_grp_size();
    fea_grp_.clear();
    for (int i = 0; i < grp_size; ++i) {
      fea_grp_.push_back(call.fea_grp(i));
    }
    bool hit_cache = call.hit_cache();
    int max_parallel = std::max(
        1, bcd_conf_.max_num_parallel_groups_in_preprocessing());
    // filter keys whose occurance <= bcd_conf_.tail_feature_freq()
    std::vector<int> pull_time(grp_size);