Class BCDServer

src/learner/bcd.h:194–274

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

  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::PREPROCESS_DATA:
        preprocessData(time, bcd);
        break;
      case BCDCall::UPDATE_MODEL:
        updateModel(time, bcd);
        break;
      case BCDCall::SAVE_MODEL:
        CHECK(bcd.has_data());
        saveModel(bcd.data());
      case BCDCall::EVALUATE_PROGRESS: {
        BCDProgress prog; evaluate(&prog);
        sys_.replyProtocalMessage(msg, prog);
        break;
      }
      default: break;
    }
  }
 protected:
  virtual void updateModel(int time, const BCDCall& call) = 0;
  virtual void evaluate(BCDProgress* prog) = 0;

  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();
    // filter tail keys
    for (int i = 0; i < grp_size; ++i, time += time_ratio_) {
      if (hit_cache) continue;
      model_.waitInMsg(kWorkerGroup, time);
      model_.finish(kWorkerGroup, time+1);
    }
    for (int i = 0; i < grp_size; ++i, time += time_ratio_) {
      // wait untill received all keys from workers
      model_.waitInMsg(kWorkerGroup, time);
      // initialize the weight
      int chl = fea_grp_[i];
      model_.clearTailFilter(chl);
      model_.value(chl).resize(model_.key(chl).size());
      model_.value(chl).setValue(bcd_conf_.init_w());
      model_.finish(kWorkerGroup, time+1);
    }
  }

  void saveModel(const DataConfig& output) {