| 322 | } |
| 323 | |
| 324 | virtual void preprocessData(int time, const BCDCall& call) { |
| 325 | int grp_size = call.fea_grp_size(); |
| 326 | fea_grp_.clear(); |
| 327 | for (int i = 0; i < grp_size; ++i) { |
| 328 | fea_grp_.push_back(call.fea_grp(i)); |
| 329 | } |
| 330 | bool hit_cache = call.hit_cache(); |
| 331 | int max_parallel = std::max( |
| 332 | 1, bcd_conf_.max_num_parallel_groups_in_preprocessing()); |
| 333 | // filter keys whose occurance <= bcd_conf_.tail_feature_freq() |
| 334 | std::vector<int> pull_time(grp_size); |
| 335 | for (int i = 0; i < grp_size; ++i, time += time_ratio_) { |
| 336 | if (hit_cache) continue; |
| 337 | int grp = fea_grp_[i]; |
| 338 | |
| 339 | // find all unique keys with their count in feature group i |
| 340 | SArray<Key> uniq_key; |
| 341 | SArray<uint8> key_cnt; |
| 342 | Localizer<Key, double> *localizer = new Localizer<Key, double>(); |
| 343 | |
| 344 | if (FLAGS_verbose) { |
| 345 | LI << "counting unique key [" << i + 1 << "/" << grp_size << "]"; |
| 346 | } |
| 347 | localizer->countUniqIndex(slot_reader_.index(grp), &uniq_key, &key_cnt); |
| 348 | |
| 349 | if (FLAGS_verbose) { |
| 350 | LI << "counted unique key [" << i + 1 << "/" << grp_size << "]"; |
| 351 | } |
| 352 | |
| 353 | // push to servers |
| 354 | MessagePtr count(new Message(kServerGroup, time)); |
| 355 | count->setKey(uniq_key); |
| 356 | count->addValue(key_cnt); |
| 357 | count->task.set_key_channel(grp); |
| 358 | count->addFilter(FilterConfig::KEY_CACHING); |
| 359 | auto tail = model_.set(count)->mutable_tail_filter(); |
| 360 | tail->set_insert_count(true); |
| 361 | tail->set_countmin_k(bcd_conf_.countmin_k()); |
| 362 | tail->set_countmin_n((int)(uniq_key.size()*bcd_conf_.countmin_n_ratio())); |
| 363 | CHECK_EQ(time, model_.push(count)); |
| 364 | |
| 365 | // pull filtered keys after the servers have aggregated all counts |
| 366 | MessagePtr filter(new Message(kServerGroup, time+2, time+1)); |
| 367 | filter->setKey(uniq_key); |
| 368 | filter->task.set_key_channel(grp); |
| 369 | filter->addFilter(FilterConfig::KEY_CACHING); |
| 370 | model_.set(filter)->mutable_tail_filter()->set_query_key( |
| 371 | bcd_conf_.tail_feature_freq()); |
| 372 | filter->fin_handle = [this, grp, localizer, i, grp_size]() mutable { |
| 373 | // localize the training matrix |
| 374 | if (FLAGS_verbose) { |
| 375 | LI << "started remapIndex [" << i + 1 << "/" << grp_size << "]"; |
| 376 | } |
| 377 | |
| 378 | auto X = localizer->remapIndex(grp, model_.key(grp), &slot_reader_); |
| 379 | delete localizer; |
| 380 | slot_reader_.clear(grp); |
| 381 | if (!X) return; |
nothing calls this directly
no test coverage detected