Method AsyncFeatureStoreMgr

serving/processor/storage/feature_store_mgr.cc:147–191

} // namespace

AsyncFeatureStoreMgr::AsyncFeatureStoreMgr(ModelConfig* config, WorkFn fn) :
    stop_(false),
    thread_num_(config->read_thread_num),
    update_thread_num_(config->update_thread_num),
    active_thread_index_(0),
    active_update_thread_index_(0) {
  if (thread_num_ < 1 || thread_num_ > MANAGER_MAX_THREAD_NUM) {
    LOG(FATAL) << "Invalid IO thread num, required [1, 96], get "
               << thread_num_;
  }

  if (update_thread_num_ < 1 ||
      update_thread_num_ > MANAGER_MAX_UPDATE_THREAD_NUM) {
    LOG(FATAL) << "Invalid IO thread num, required [1, 16], get "
               << update_thread_num_;
  }

  task_queues_.resize(thread_num_);
  threads_.resize(thread_num_);
  mutex_.resize(thread_num_);
  cv_.resize(thread_num_);
  store_.resize(thread_num_);
  for (int i = 0; i < thread_num_; ++i) {
    mutex_[i] = new std::mutex();
    cv_[i] = new std::condition_variable();
    ready_[i] = false;
    sleeping_[i] = false;
    store_[i] = CreateFeatureStore(config);
    threads_[i].reset(new std::thread(!fn? &ThreadRun : fn, this, i, false));
  }

  update_task_queues_.resize(update_thread_num_);
  update_threads_.resize(update_thread_num_);
  update_mutex_.resize(update_thread_num_);
  update_cv_.resize(update_thread_num_);
  update_store_.resize(update_thread_num_);
  for (int i = 0; i < update_thread_num_; ++i) {
    update_mutex_[i] = new std::mutex();
    update_cv_[i] = new std::condition_variable();
    update_ready_[i] = false;
    update_sleeping_[i] = false;
    update_store_[i] = CreateFeatureStore(config);
    update_threads_[i].reset(new std::thread(!fn ? &ThreadRun : fn, this, i, true));
  }
}

AsyncFeatureStoreMgr::~AsyncFeatureStoreMgr() {
  // stop all IO threads