Method loadTrainingData

src/learner/bcd.h:32–76

  virtual ~BCDScheduler() { }

  void loadTrainingData(const DataConfig& train) {
    // ask the workers to load the data
    auto load_time = tic();
    auto confs = Postmaster::partitionData(train, sys_.yp().num_workers());
    std::vector<Task> loads(confs.size());
    for (int i = 0; i < confs.size(); ++i) {
      auto bcd = loads[i].mutable_bcd();
      bcd->set_cmd(BCDCall::LOAD_DATA);
      *bcd->mutable_data() = confs[i];
    }
    int hit_cache = 0;
    port(kWorkerGroup)->submitAndWait(loads, [this, &hit_cache](){
        LoadDataReturn info; CHECK(info.ParseFromString(exec_.lastRecvReply()));
        // LL << info.DebugString();
        g_train_info_ = mergeExampleInfo(g_train_info_, info.example_info());
        hit_cache += info.hit_cache();
      });
    if (hit_cache > 0) {
      CHECK_EQ(hit_cache, loads.size()) << "clear the local caches";
      LI << "Hit local caches for the training data";
    }
    LI << "Loaded " << g_train_info_.num_ex() << " examples in "
       << toc(load_time) << " sec";


    // preprocess the training data
    for (int i = 0; i < g_train_info_.slot_size(); ++i) {
      auto info = g_train_info_.slot(i);
      CHECK(info.has_id());
      if (info.id() == 0) continue;  // it's the label
      fea_grp_.push_back(info.id());
    }
    auto preprocess_time = tic();
    Task preprocess;
    auto prep_bcd = preprocess.mutable_bcd();
    prep_bcd->set_cmd(BCDCall::PREPROCESS_DATA);
    for (auto grp : fea_grp_) prep_bcd->add_fea_grp(grp);
    prep_bcd->set_hit_cache(hit_cache > 0);
    port(kCompGroup)->submitAndWait(preprocess);
    LI << "Preprocessing is finished in " << toc(preprocess_time) << " sec";
    if (bcd_conf_.tail_feature_freq()) {
      LI << "Features with frequency <= " << bcd_conf_.tail_feature_freq()
         << " are filtered";
    }
  }

  void divideFeatureBlocks() {
    // partition feature blocks